Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def AddFederate(broker, core_type="zmq", count=1, deltat=1.0, name_prefix="fed"):
# Create Federate Info object that describes the federate properties #
fedinfo = h.helicsCreateFederateInfo()
# Set Federate name #
h.helicsFederateInfoSetCoreName(fedinfo, name_prefix)
# Set core type from string #
h.helicsFederateInfoSetCoreTypeFromString(fedinfo, "zmq")
# Federate init string #
fedinitstring = "--broker=mainbroker --federates={}".format(count)
h.helicsFederateInfoSetCoreInitString(fedinfo, fedinitstring)
# Set the message interval (timedelta) for federate. Note th#
# HELICS minimum message time interval is 1 ns and by default
# it uses a time delta of 1 second. What is provided to the
# setTimedelta routine is a multiplier for the default timedelta.
# Set one second message interval #
h.helicsFederateInfoSetTimeProperty(fedinfo, h.helics_property_time_delta, deltat)
h.helicsFederateInfoSetIntegerProperty(fedinfo, h.helics_property_int_log_level, 1)
mFed = h.helicsCreateMessageFederate(name_prefix, fedinfo)
def test_value_federate_runFederateTimeoutTest(helicsBroker):
fedinitstring = "--broker=mainbroker --federates=1"
deltat = 0.01
# Create Federate Info object that describes the federate properties #
fedinfo = h.helicsCreateFederateInfo()
# Set Federate name #
h.helicsFederateInfoSetCoreName(fedinfo, "TestA Core")
# Set core type from string #
h.helicsFederateInfoSetCoreTypeFromString(fedinfo, "zmq")
# Federate init string #
h.helicsFederateInfoSetCoreInitString(fedinfo, fedinitstring)
# Set the message interval (timedelta) for federate. Note th#
# HELICS minimum message time interval is 1 ns and by default
# it uses a time delta of 1 second. What is provided to the
# setTimedelta routine is a multiplier for the default timedelta.
# Set one second message interval #
h.helicsFederateInfoSetTimeProperty(fedinfo, h.helics_property_time_delta, deltat)
h.helicsFederateInfoSetIntegerProperty(fedinfo, h.helics_property_int_log_level, 1)
vFed = h.helicsCreateValueFederate("TestA Federate", fedinfo)
# Create broker #
broker = h.helicsCreateBroker("zmq", "", initstring)
isconnected = h.helicsBrokerIsConnected(broker)
if isconnected == 1:
pass
# Create Federate Info object that describes the federate properties #
fedinfo = h.helicsCreateFederateInfo()
# Set Federate name #
h.helicsFederateInfoSetCoreName(fedinfo, "TestA Core")
# Set core type from string #
h.helicsFederateInfoSetCoreTypeFromString(fedinfo, "zmq")
# Federate init string #
h.helicsFederateInfoSetCoreInitString(fedinfo, fedinitstring)
# Set the message interval (timedelta) for federate. Note th#
# HELICS minimum message time interval is 1 ns and by default
# it uses a time delta of 1 second. What is provided to the
# setTimedelta routine is a multiplier for the default timedelta.
# Set one second message interval #
h.helicsFederateInfoSetTimeProperty(fedinfo, h.helics_property_time_delta, deltat)
h.helicsFederateInfoSetIntegerProperty(fedinfo, h.helics_property_int_log_level, 1)
vFed = h.helicsCreateValueFederate("TestA Federate", fedinfo)
def create_value_federate(deltat=1.0, fedinitstring="--federates=1"):
fedinfo = h.helicsFederateInfoCreate()
status = h.helicsFederateInfoSetFederateName(fedinfo, "TestB Federate")
assert status == 0
status = h.helicsFederateInfoSetCoreTypeFromString(fedinfo, "zmq")
assert status == 0
status = h.helicsFederateInfoSetCoreInitString(fedinfo, fedinitstring)
assert status == 0
status = h.helicsFederateInfoSetTimeDelta(fedinfo, deltat)
assert status == 0
status = h.helicsFederateInfoSetLoggingLevel(fedinfo, 1)
assert status == 0
fed = h.helicsCreateCombinationFederate(fedinfo)
return fed
def create_value_federate(broker, deltat=1.0, fedinitstring="--broker=mainbroker --federates=1"):
fedinfo = h.helicsFederateInfoCreate()
status = h.helicsFederateInfoSetFederateName(fedinfo, "TestA Federate")
assert status == 0
status = h.helicsFederateInfoSetCoreTypeFromString(fedinfo, "zmq")
assert status == 0
status = h.helicsFederateInfoSetCoreInitString(fedinfo, fedinitstring)
assert status == 0
status = h.helicsFederateInfoSetTimeDelta(fedinfo, deltat)
assert status == 0
status = h.helicsFederateInfoSetLoggingLevel(fedinfo, 1)
assert status == 0
fed = h.helicsCreateCombinationFederate(fedinfo)
return fed
def create_federate(deltat=1.0, fedinitstring="--federates=1"):
fedinfo = h.helicsFederateInfoCreate()
status = h.helicsFederateInfoSetFederateName(fedinfo, "Combination Federate")
assert status == 0
status = h.helicsFederateInfoSetCoreTypeFromString(fedinfo, "zmq")
assert status == 0
status = h.helicsFederateInfoSetCoreInitString(fedinfo, fedinitstring)
assert status == 0
status = h.helicsFederateInfoSetTimeDelta(fedinfo, deltat)
assert status == 0
status = h.helicsFederateInfoSetLoggingLevel(fedinfo, 1)
assert status == 0
fed = h.helicsCreateCombinationFederate(fedinfo)
return fed
print("Checking if Broker is connected")
isconnected = h.helicsBrokerIsConnected(broker)
print("Checked if Broker is connected")
if isconnected == 1:
print("Broker created and connected")
# Create Federate Info object that describes the federate properties #
fedinfo = h.helicsFederateInfoCreate()
# Set Federate name #
status = h.helicsFederateInfoSetFederateName(fedinfo, "TestA Federate")
# Set core type from string #
status = h.helicsFederateInfoSetCoreTypeFromString(fedinfo, "zmq")
# Federate init string #
status = h.helicsFederateInfoSetCoreInitString(fedinfo, fedinitstring)
# Set the message interval (timedelta) for federate. Note th#
# HELICS minimum message time interval is 1 ns and by default
# it uses a time delta of 1 second. What is provided to the
# setTimedelta routine is a multiplier for the default timedelta.
# Set one second message interval #
status = h.helicsFederateInfoSetTimeDelta(fedinfo, deltat)
status = h.helicsFederateInfoSetLoggingLevel(fedinfo, 1)
# Create value federate #
vfed = h.helicsCreateValueFederate(fedinfo)