Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_value_federate_single_transfer(vFed):
pubid = h.helicsFederateRegisterGlobalPublication (vFed, "pub1", h.helics_data_type_string, "");
subid = h.helicsFederateRegisterSubscription (vFed, "pub1", "");
h.helicsFederateEnterExecutingMode(vFed)
h.helicsPublicationPublishString(pubid, "string1")
grantedtime = h.helicsFederateRequestTime(vFed, 1.0)
assert grantedtime == 0.01
s = h.helicsInputGetString(subid)
assert s == "string1"
def test_value_federate_runFederateTestInteger(vFed):
defaultValue = 1
testValue = 2
pubid = h.helicsFederateRegisterGlobalPublication (vFed, "pub1", h.helics_data_type_int, "")
subid = h.helicsFederateRegisterSubscription (vFed, "pub1", "")
h.helicsInputSetDefaultInteger(subid, defaultValue)
h.helicsFederateEnterExecutingMode (vFed)
h.helicsPublicationPublishInteger(pubid, testValue)
value = h.helicsInputGetInteger(subid)
assert value == defaultValue
grantedtime = h.helicsFederateRequestTime(vFed, 1.0)
assert grantedtime == 0.01
value = h.helicsInputGetInteger(subid)
assert value == testValue
h.helicsPublicationPublishInteger(pubid, testValue + 1)
def test_value_federate_subscription_and_publication_registration(vFed):
pubid3 = h.helicsFederateRegisterTypePublication(vFed, "pub3", "double", "V")
subid1 = h.helicsFederateRegisterSubscription(vFed, "sub1", "")
subid2 = h.helicsFederateRegisterSubscription(vFed, "sub2", "")
subid3 = h.helicsFederateRegisterSubscription(vFed, "sub3", "V")
h.helicsFederateEnterExecutingMode(vFed)
publication_type = h.helicsPublicationGetType(pubid3)
assert publication_type == 'double'
sub_key = h.helicsSubscriptionGetKey(subid1)
assert sub_key == 'sub1'
sub_type = h.helicsInputGetType(subid1)
assert sub_type == ''
sub_key = h.helicsSubscriptionGetKey(subid2)
assert sub_key == 'sub2'
sub_key = h.helicsSubscriptionGetKey(subid3)
assert sub_key == 'sub3'
sub_type = h.helicsInputGetType(subid3)
assert sub_type == ''
def test_value_federate_runFederateTestBool(vFed):
defaultValue = True
testValue1 = True
testValue2 = False
# register the publications
pubid = h.helicsFederateRegisterGlobalPublication(vFed, "pub1", h.helics_data_type_boolean, "")
subid = h.helicsFederateRegisterSubscription(vFed, "pub1", "")
h.helicsInputSetDefaultBoolean(subid, h.helics_true if defaultValue else h.helics_false)
h.helicsFederateEnterExecutingMode(vFed)
# publish string1 at time=0.0;
h.helicsPublicationPublishBoolean(pubid, h.helics_true if testValue1 else h.helics_false)
val = h.helicsInputGetBoolean(subid)
assert val == h.helics_true if defaultValue else h.helics_false
grantedtime = h.helicsFederateRequestTime (vFed, 1.0)
assert grantedtime == 0.01
# get the value
val = h.helicsInputGetBoolean(subid)
def test_value_federate_subscription_and_publication_registration(vFed):
pubid3 = h.helicsFederateRegisterTypePublication(vFed, "pub3", "double", "V")
subid1 = h.helicsFederateRegisterSubscription(vFed, "sub1", "")
subid2 = h.helicsFederateRegisterSubscription(vFed, "sub2", "")
subid3 = h.helicsFederateRegisterSubscription(vFed, "sub3", "V")
h.helicsFederateEnterExecutingMode(vFed)
publication_type = h.helicsPublicationGetType(pubid3)
assert publication_type == 'double'
sub_key = h.helicsSubscriptionGetKey(subid1)
assert sub_key == 'sub1'
sub_type = h.helicsInputGetType(subid1)
assert sub_type == ''
sub_key = h.helicsSubscriptionGetKey(subid2)
assert sub_key == 'sub2'
sub_key = h.helicsSubscriptionGetKey(subid3)
assert sub_key == 'sub3'
sub_type = h.helicsInputGetType(subid3)
def test_value_federate_runFederateTestVectorD(vFed):
defaultValue = [0, 1, 2]
testValue = [3, 4, 5]
pubid = h.helicsFederateRegisterGlobalPublication (vFed, "pub1", h.helics_data_type_vector, "")
subid = h.helicsFederateRegisterSubscription (vFed, "pub1", "")
h.helicsInputSetDefaultVector(subid, defaultValue)
h.helicsFederateEnterExecutingMode(vFed)
h.helicsPublicationPublishVector(pubid, testValue)
value = h.helicsInputGetVector(subid)
assert value == [0, 1, 2]
grantedtime = h.helicsFederateRequestTime(vFed, 1.0)
assert grantedtime == 0.01
value = h.helicsInputGetVector(subid)
assert value == [3, 4, 5]
def main(delay=None):
broker = create_broker()
fed = create_value_federate(broker)
pubid = h.helicsFederateRegisterGlobalTypePublication(fed, "federate1-to-federate2", h.HELICS_DATA_TYPE_STRING, "")
subid = h.helicsFederateRegisterSubscription(fed, "federate2-to-federate1", "double", "")
epid = h.helicsFederateRegisterGlobalEndpoint(fed, "endpoint1", "")
if delay is not None:
fid = h.helicsFederateRegisterSourceFilter(fed, h.helics_delay_filter, "endpoint2", "filter-name")
h.helicsSubscriptionSetDefaultDouble(subid, 0)
print("Entering execution mode")
h.helicsFederateEnterExecutionMode(fed)
if delay is not None:
h.helicsFilterSet(fid, "delay", 2.0)
grantedtime = -1
while True:
try:
def main():
fed = create_value_federate()
pubid = h.helicsFederateRegisterGlobalTypePublication(fed, "federate2-to-federate1", h.HELICS_DATA_TYPE_STRING, "")
subid = h.helicsFederateRegisterSubscription (fed, "federate1-to-federate2", "double", "")
epid = h.helicsFederateRegisterGlobalEndpoint(fed, "endpoint2", "")
# print("Setting default value")
h.helicsSubscriptionSetDefaultDouble(subid, 0)
print("Entering execution mode")
h.helicsFederateEnterExecutionMode(fed)
grantedtime = -1
while True:
try:
stop_at_time, value_to_send = get_input(grantedtime)
except KeyboardInterrupt:
print("")
break
while grantedtime < stop_at_time: