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_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'
sub_type = h.helicsInputGetType(subid2)
assert sub_type == ''
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'
sub_type = h.helicsInputGetType(subid2)
assert sub_type == ''
federate_name = h.helicsFederateGetName(fed)[-1]
print(" Federate {} has been registered".format(federate_name))
pubkeys_count = h.helicsFederateGetPublicationCount(fed)
subkeys_count = h.helicsFederateGetInputCount(fed)
print(subkeys_count)
###################### 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 = []
endpoint_count = h.helicsFederateGetEndpointCount(fed)
subkeys_count = h.helicsFederateGetInputCount(fed)
print(subkeys_count)
print(endpoint_count)
###################### Reference to Publications and Subscription form index #############################
endid = {}
subid = {}
for i in range(0,endpoint_count):
endid["m{}".format(i)] = h.helicsFederateGetEndpointByIndex(fed, i)
end_name = h.helicsEndpointGetName(endid["m{}".format(i)])
print( 'Registered Endpoint ---> {}'.format(end_name))
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))
print( '###############################################################################################')
print( '######################## Entering Execution Mode ##########################################')
###################### Entering Execution Mode ##########################################################
h.helicsFederateEnterExecutingMode(fed)
hours = 24
total_inteval = int(60 * 60 * hours)
grantedtime = -1
update_interval = 5*60
feeder_limit_upper = 4 * (1000*1000)
feeder_limit_lower = 2.7 * (1000*1000)
k = 0
data ={};
time_sim = []; feeder_real_power = []; feeder_imag_power = []
endpoint_count = h.helicsFederateGetEndpointCount(fed)
subkeys_count = h.helicsFederateGetInputCount(fed)
print(subkeys_count)
print(endpoint_count)
###################### Reference to Publications and Subscription form index #############################
endid = {}
subid = {}
for i in range(0,endpoint_count):
endid["m{}".format(i)] = h.helicsFederateGetEndpointByIndex(fed, i)
end_name = h.helicsEndpointGetName(endid["m{}".format(i)])
logger.info( 'Registered Endpoint ---> {}'.format(end_name))
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)])
logger.info( 'Registered Subscription ---> {}'.format(sub_key))
print( '###############################################################################################')
print( '######################## Entering Execution Mode ##########################################')
###################### Entering Execution Mode ##########################################################
h.helicsFederateEnterExecutingMode(fed)
hours = 24
total_inteval = int(60 * 60 * hours)
grantedtime = -1
update_interval = 5*60
feeder_limit_upper = 4 * (1000*1000)
feeder_limit_lower = 2.7 * (1000*1000)
k = 0
data ={};
time_sim = []; feeder_real_power = []; feeder_imag_power = []
federate_name = h.helicsFederateGetName(fed)[-1]
print(" Federate {} has been registered".format(federate_name))
pubkeys_count = h.helicsFederateGetPublicationCount(fed)
subkeys_count = h.helicsFederateGetInputCount(fed)
print(subkeys_count)
###################### 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 = []
federate_name = h.helicsFederateGetName(fed)[-1]
print(" Federate {} has been registered".format(federate_name))
pubkeys_count = h.helicsFederateGetPublicationCount(fed)
subkeys_count = h.helicsFederateGetInputCount(fed)
print(subkeys_count)
###################### 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 = []