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_message_filter_registration(broker):
fFed, ffedinfo = AddFederate(broker, "zmq", 1, 1, "filter")
mFed, mfedinfo = AddFederate(broker, "zmq", 1, 1, "message")
h.helicsFederateRegisterGlobalEndpoint(mFed, "port1", "")
h.helicsFederateRegisterGlobalEndpoint(mFed, "port2", None)
f1 = h.helicsFederateRegisterFilter (fFed, h.helics_filtertype_custom, "filter1")
f2 = h.helicsFederateRegisterFilter (fFed, h.helics_filtertype_custom, "filter2")
h.helicsFederateRegisterEndpoint (fFed, "fout", "")
h.helicsFederateRegisterFilter (fFed, h.helics_filtertype_custom, "filter0/fout")
h.helicsFederateEnterExecutingModeAsync(fFed)
h.helicsFederateEnterExecutingMode(mFed)
h.helicsFederateEnterExecutingModeComplete(fFed)
filter_name = h.helicsFilterGetName(f1)
assert filter_name == "filter/filter1"
filter_name = h.helicsFilterGetName(f2)
assert filter_name == "filter/filter2"
# filter_target = h.helicsFilterGetTarget(f2)
def test_message_federate_send(mFed):
epid1 = h.helicsFederateRegisterEndpoint(mFed, "ep1", None)
epid2 = h.helicsFederateRegisterGlobalEndpoint(mFed, "ep2", "random")
h.helicsFederateSetTimeProperty(mFed, h.helics_property_time_delta, 1.0)
h.helicsFederateEnterExecutingMode(mFed)
data = "random-data"
h.helicsEndpointSendEventRaw(epid1, "ep2", data, 1.0)
granted_time = h.helicsFederateRequestTime(mFed, 2.0)
assert granted_time == 1.0
res = h.helicsFederateHasMessage(mFed)
assert res == 1
res = h.helicsEndpointHasMessage(epid1)
assert res == 0
def test_message_filter_function(broker):
fFed, ffedinfo = AddFederate(broker, "zmq", 1, 1, "filter")
mFed, mfedinfo = AddFederate(broker, "zmq", 1, 1, "message")
p1 = h.helicsFederateRegisterGlobalEndpoint(mFed, "port1", "")
p2 = h.helicsFederateRegisterGlobalEndpoint(mFed, "port2", "random")
f1 = h.helicsFederateRegisterGlobalFilter(fFed, h.helics_filtertype_custom, "filter1")
h.helicsFilterAddSourceTarget(f1, "port1")
f2 = h.helicsFederateRegisterGlobalFilter(fFed, h.helics_filtertype_delay, "filter2")
h.helicsFilterAddSourceTarget(f2, "port1")
h.helicsFederateRegisterEndpoint(fFed,'fout','');
f3 = h.helicsFederateRegisterFilter(fFed, h.helics_filtertype_random_delay, 'filter3');
h.helicsFilterAddSourceTarget(f3,'filter/fout');
h.helicsFilterSet(f2, "delay", 2.5)
h.helicsFederateEnterExecutingModeAsync(fFed)
h.helicsFederateEnterExecutingMode(mFed)
h.helicsFederateEnterExecutingModeComplete(fFed)
state = h.helicsFederateGetState(fFed)
assert state == 2
def test_message_filter_registration(broker):
fFed, ffedinfo = AddFederate(broker, "zmq", 1, 1, "filter")
mFed, mfedinfo = AddFederate(broker, "zmq", 1, 1, "message")
h.helicsFederateRegisterGlobalEndpoint(mFed, "port1", "")
h.helicsFederateRegisterGlobalEndpoint(mFed, "port2", None)
f1 = h.helicsFederateRegisterFilter (fFed, h.helics_filtertype_custom, "filter1")
f2 = h.helicsFederateRegisterFilter (fFed, h.helics_filtertype_custom, "filter2")
h.helicsFederateRegisterEndpoint (fFed, "fout", "")
h.helicsFederateRegisterFilter (fFed, h.helics_filtertype_custom, "filter0/fout")
h.helicsFederateEnterExecutingModeAsync(fFed)
h.helicsFederateEnterExecutingMode(mFed)
h.helicsFederateEnterExecutingModeComplete(fFed)
filter_name = h.helicsFilterGetName(f1)
assert filter_name == "filter/filter1"
filter_name = h.helicsFilterGetName(f2)
assert filter_name == "filter/filter2"
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:
stop_at_time, value_to_send = get_input(grantedtime)
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:
print(">>>>>>>> Requesting time = {}".format(stop_at_time))