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)
# assert filter_target == "port2"
h.helicsFederateFinalize(mFed)
h.helicsFederateFinalize(fFed)
FreeFederate(fFed, ffedinfo)
FreeFederate(mFed, mfedinfo)
def test_value_federate_runFederateTestNamedPoint(vFed):
defaultValue = "start of a longer string in place of the shorter one and now this should be very long"
defVal = 5.3
#testValue1 = "inside of the functional relationship of helics"
testValue1 = "short string"
testVal1 = 45.7823
testValue2 = "I am a string"
testVal2 = 0.0
pubid = h.helicsFederateRegisterGlobalPublication(vFed, "pub1", h.helics_data_type_named_point, "")
subid = h.helicsFederateRegisterSubscription(vFed, "pub1", "")
h.helicsInputSetDefaultNamedPoint(subid, defaultValue, defVal)
h.helicsFederateEnterExecutingMode(vFed)
# publish string1 at time=0.0;
h.helicsPublicationPublishNamedPoint(pubid, testValue1, testVal1)
# double val;
val = h.helicsInputGetNamedPoint(subid)
# assert value == defaultValue
assert val == [defaultValue, defVal]
grantedtime = h.helicsFederateRequestTime(vFed, 1.0)
assert grantedtime == 0.01
# get the value
val2 = h.helicsInputGetNamedPoint(subid)
# make sure the string is what we expect
def test_value_federate_runFederateTestComplex(vFed):
rDefaultValue = 1.0
iDefaultValue = 1.0
rTestValue = 2.0
iTestValue = 2.0
pubid = h.helicsFederateRegisterGlobalPublication (vFed, "pub1", h.helics_data_type_complex, "")
subid = h.helicsFederateRegisterSubscription (vFed, "pub1", "")
h.helicsInputSetDefaultComplex(subid, rDefaultValue, iDefaultValue)
h.helicsFederateEnterExecutingMode (vFed)
# publish string1 at time=0.0;
h.helicsPublicationPublishComplex(pubid, rTestValue, iTestValue)
value1, value2 = h.helicsInputGetComplex(subid)
assert value1 == rDefaultValue
assert value2 == iDefaultValue
grantedtime = h.helicsFederateRequestTime (vFed, 1.0)
assert grantedtime == 0.01
value1, value2 = h.helicsInputGetComplex(subid)
assert value1 == rTestValue
assert value2 == iTestValue
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
data = "hello world"
filt_key = h.helicsFilterGetName(f1);
assert filt_key == 'filter1';
filt_key = h.helicsFilterGetName(f2);
assert filt_key == 'filter2';
h.helicsEndpointSendMessageRaw(p1, "port2", data)
h.helicsFederateRequestTimeAsync(mFed, 1.0)
grantedtime = h.helicsFederateRequestTime(fFed, 1.0)
assert grantedtime == 1.0
grantedtime = h.helicsFederateRequestTimeComplete(mFed)
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 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 == ''
sub_units = h.helicsInputGetUnits(subid3)
assert sub_units == 'V'
def test_message_federate_endpoint_registration(mFed):
epid1 = h.helicsFederateRegisterEndpoint(mFed, "ep1", None)
epid2 = h.helicsFederateRegisterGlobalEndpoint(mFed, "ep2", "random")
h.helicsFederateEnterExecutingMode(mFed)
endpoint_name = h.helicsEndpointGetName(epid1)
assert endpoint_name == "TestA Federate/ep1"
endpoint_name = h.helicsEndpointGetName(epid2)
assert endpoint_name == "ep2"
endpoint_name = h.helicsEndpointGetType(epid1)
assert endpoint_name == ""
endpoint_name = h.helicsEndpointGetType(epid2)
assert endpoint_name == "random"
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
res = h.helicsEndpointHasMessage(epid2)
assert res == 1
###################### Reference to Publications and Subscription form index #############################
pubid = {}
subid = {}
for i in range(0,pubkeys_count):
pubid["m{}".format(i)] = h.helicsFederateGetPublicationByIndex(fed, i)
pubtype = h.helicsPublicationGetType(pubid["m{}".format(i)])
print(pubtype)
for i in range(0,subkeys_count):
subid["m{}".format(i)] = h.helicsFederateGetInputByIndex(fed, i)
status = h.helicsInputSetDefaultComplex(subid["m{}".format(i)], 0, 0)
sub_key = h.helicsSubscriptionGetKey(subid["m{}".format(i)])
print( 'Registered Subscription ---> {}'.format(sub_key))
###################### Entereing Execution Mode ##########################################################
status = h.helicsFederateEnterInitializingMode(fed)
status = h.helicsFederateEnterExecutingMode(fed)
#Pypower Processing (inputs)
hours = 24
total_inteval = int(60 * 60 * hours)
grantedtime = -1
pf_interval = 5*60 # in seconds (minimim_resolution)
acopf_interval = 15*60 # in seconds (minimim_resolution)
random.seed(0)
peak_demand = []
ppc = []
case_format = case118()
peak_demand = case_format['bus'][:,2][:].copy()
ppc = case_format.copy()
###################### creating fixed load profiles for each bus based on PF interval #############################
###################### Reference to Publications and Subscription form index #############################
pubid = {}
subid = {}
for i in range(0,pubkeys_count):
pubid["m{}".format(i)] = h.helicsFederateGetPublicationByIndex(fed, i)
pubtype = h.helicsPublicationGetType(pubid["m{}".format(i)])
print(pubtype)
for i in range(0,subkeys_count):
subid["m{}".format(i)] = h.helicsFederateGetInputByIndex(fed, i)
status = h.helicsInputSetDefaultComplex(subid["m{}".format(i)], 0, 0)
sub_key = h.helicsSubscriptionGetKey(subid["m{}".format(i)])
print( 'Registered Subscription ---> {}'.format(sub_key))
###################### Entereing Execution Mode ##########################################################
status = h.helicsFederateEnterInitializingMode(fed)
status = h.helicsFederateEnterExecutingMode(fed)
#Pypower Processing (inputs)
hours = 24
total_inteval = int(60 * 60 * hours)
grantedtime = -1
pf_interval = 5*60 # in seconds (minimim_resolution)
acopf_interval = 15*60 # in seconds (minimim_resolution)
random.seed(0)
peak_demand = []
ppc = []
case_format = case118()
peak_demand = case_format['bus'][:,2][:].copy()
ppc = case_format.copy()
###################### creating fixed load profiles for each bus based on PF interval #############################