cosim_toolbox.sims package

CoSim Toolbox Simulations Package

This module provides a unified API for reading and writing co-simulation federates

class cosim_toolbox.sims.Collect(*values)

Bases: Enum

MAYBE = 'maybe'

NO = 'no'

YES = 'yes'

class cosim_toolbox.sims.DockerRunner

Bases: object

Collection of static methods used in building and running the docker-compose.yaml for running a new service or simulator.

static define_sh(scenario_name: str, use_meta_db: str = 'mongo', use_data_db: str = 'postgres') → None

Create the run shell file from the provided scenario

Parameters:

scenario_name (str) – Name of the scenario
use_meta_db (str)
use_data_db (str)

static define_yaml(scenario_name: str, use_meta_db: str = 'mongo', use_data_db: str = 'postgres') → None

Create the docker-compose.yaml from the provided scenario

Parameters:

scenario_name (str) – Name of the scenario
use_meta_db (str)
use_data_db (str)

static run_remote_yaml(scenario_name: str, path: str = '/run/python/test_federation') → None

Runs the docker-compose.yaml on a remote compute node: TODO: Test, ‘path’ have not been set in examples at this point

Parameters:

scenario_name (str) – Name of the scenario run by this docker-compose.yaml
path (str, optional) – Path to docker-compose-yaml on remote hose. Defaults to “/run/python/test_federation”.

static run_yaml(scenario_name: str) → None

Runs the provided scenario by calling the appropriate docker-compose.yaml

Parameters:: scenario_name (str) – Name of the scenario run by this docker-compose.yaml

class cosim_toolbox.sims.Federate(fed_name: str = '', *, debug: bool = True)

Bases: object

This class definition is intended to be a reasonable, generic class for Python-based federates in HELICS. It outlines the typical federate operational procedure in the “_main_” function; users that don’t need anything fancy will probably be able to call those few functions and get a working federate.

This class gets its configuration from the metadata database following the standard CST definition of the “federations” document.

To be overly clear, this class is intended to be sub-classed and overloaded to allow users to customize it as necessary. If nothing else, the “update_model” method will always need updating to perform the particular calculations federate is responsible for. There are several other methods that are likely candidates for subclassing based on the particular needs of the federate or the larger federation: “enter_initializing_mode” “enter_executing_mode” “time_request”

All of these have the simplest version of these HELICS operations but there are more complex versions that HELICS supports that allow for things like iterations and asynchronous or non-blocking operations (further details can be found in the HELICS documentation).

The existing methods to pull in values from the HELICS federation and push values out are likely to be sufficient for most federates but, again these can be overloaded in a subclass if, for example, the number of HELICS inputs and publications is very large and only a few are used during a given time-step.

hfed: The HELICS federate object used to access the HELICS interfaces federate_name (str): The federate name

federate

Dictionary with all configuration information, including but not limited to the HELICS JSON config string

Type:: dict

federate_type

The federate type. Must be “value”, “message”, or “combo”

Type:: str

config

Valid HELICS config JSON string

Type:: dict

granted_time

The last time granted to this federate

Type:: float

period

The size of the simulated time step takes when requesting the next time

Type:: float

scenario_name

The scenario name

Type:: str

scenario

Dictionary with all scenario configuration information

Type:: dict

federation_name

The federation name

Type:: str

federation

Dictionary with all federate configuration information

Type:: dict

calculate_next_requested_time() → float

Determines the next simulated time to request from HELICS

Many federates run at very regular time steps and thus the calculation of the requested time is trivial. In some cases, though, the requested time may be more dynamic and this method provides a place for users to overload the default calculation method if they need something more complex.

Returns:

Calculated time for the next HELICS time: request

Return type:

self.next_requested_time

create_federate(scenario_name: str, use_meta_db: str = 'mongo', use_data_db: str = 'postgres') → None

Create CST and HELICS federates

Creates and defines both the instance of this class,(the Co-Simulation federate) and the HELICS federate object (self.hfed). Any initialization that cannot take place on instantiation of the federate object should be done here. In this case, initializing any class attribute values that come from the metadata database have to take place after connecting to said database.

Parameters:

scenario_name (str) – Name of scenario used to store configuration information in the dbConfigs
use_meta_db (str, optional) – Whether to use a metadata database (e.g., MongoDB). Defaults to mongo.
use_data_db (str, optional) – Whether to use a timeseries database (e.g., PostgreSQL). Defaults to postgres.

Raises:

NameError – Scenario name is undefined (None)

Returns:

None

create_helics_fed() → None

Creates the HELICS federate object

Using the HELICS configuration document from the dbConfigs, this method creates the HELICS federate. HELICS has distinct APIs for the creation of a federate based on its type and thus, the type of federate needs to be defined as an instance attribute to enable the correct API to be called.

Raises:: ValueError – Invalid value for self.federate_type
Returns:: None

property current_time

destroy_federate() → None

Removes HELICS federate from federation

As part of ending a HELICS co-simulation it is good housekeeping to formally destroy the model federate. Doing so informs the rest of the federation that it is no longer a part of the co-simulation and they should proceed without it (if applicable). Generally this is done when the co-simulation is complete and all federates end execution at more or less the same wall-clock time.

enter_executing_mode() → None

Moves the Federate to executing mode

Similar to initializing mode, there are a few different ways of handling HELICS executing mode and what is implemented here is the simplest. If you need something more complex or specific, overload or redefine this method.

Returns:: None

enter_initialization() → None

Moves federate to HELICS initializing mode

There are a few stages to a federate in HELICS with initializing mode being the first after the Federate is created. Entering initializing mode is a global synchronous event for all federates and provides an opportunity to do some fancy things around dynamic configuration of the Federate. What is implemented here is the simplest, most vanilla means of entering initializing mode. If you need something more complex, overload or redefine this method.

Returns:: None

get_data_from_federation() → None

Collects inputs from federation and stores them

This method is an automated way of getting data the rest of the federation has sent out. Directly accessing the value and message interfaces via the HELICS federate (hfed object) provides a much richer set of metadata associated with these interfaces. The implementation here is vanilla and is expected to be sufficient for many use cases.

Returns:: None

get_helics_config() → None

Sets instance attributes to enable HELICS config query of dbConfigs

HELICS configuration information is generally stored in the dbConfigs and is copied into the self.federation attribute. This method pulls out a few keys configuration parameters from that attribute to make them more easily accessible.

Returns:: None

on_enter_executing_mode() → None

Functionality executed after entering executing mode

By default, no functionality is implemented

Returns:: None

on_enter_initialization_mode() → None

Functionality executed after entering initialization

By default, no functionality is implemented

Returns:: None

on_start() → None

Functionality executed prior to entering initializing

By default, no functionality is implemented

Returns:: None

request_time(requested_time: float) → float

Requests next simulated time from HELICS

HELICS provides a variety of means of requesting time. The most common is a simple hfed.request_time(float) which is a blocking call. There are others that make the time request but allow users to continue working on something else while they wait for HELICS to get back to them with the granted time. This method is here just to allow users to redefine or overload and re-implement how they want to do time requests.

Parameters:: requested_time – Simulated time this federate needs to request
Returns:: Simulated time granted by HELICS
Return type:: self.granted_time

reset_data_to_federation() → None

Sets all values in dictionary of values being sent out via publications and endpoints in the data_to_federation dictionary to “None”.

Any values in these dictionaries set to None do not result in a new output via HELICS. This method wipes out all data so that only entries added to the dictionary after calling this method will be published, preventing duplicate publication of data that has not changed and does not need to be re-sent. This also helps manage the data being logged in the time-series database.

Returns:: None

run(scenario_name: str, use_meta_db: str = 'mongo', use_data_db: str = 'postgres') → None

Runs the HELICS federate class

Parameters:

scenario_name (str) – Name of scenario used to store configuration information in the dbConfigs
use_meta_db (str, optional) – Whether to use a metadata database (e.g., MongoDB). Defaults to mongo.
use_data_db (str, optional) – Whether to use a timeseries database (e.g., PostgreSQL). Defaults to postgres.

run_cosim_loop() → None

Runs the generic HELICS co-sim loop

This HELICS co-sim loop runs until it the simulated time reaches self.stop_time. self.enter_initialization_mode() and self.enter_executing_mode(), and self. simulate_next_step have been implemented and should be overloaded/redefined as necessary to fit the needs of a given federate and/or federation.

Returns:: None

send_data_to_federation(reset=False) → None

Sends specified outputs to rest of HELICS federation

This method provides an easy way for users to send out any data to the rest of the federation. Users pass in a dict structured the same as the “data_from_federation” with sub-dicts for publications and endpoints and keys inside those dicts for the name of the pub or endpoint. The value for the keys is slightly different, though:

pubs: value is the data to send endpoints: value is a dictionary as follows:
{
    "destination": <target endpoint name, may be an empty string>
    "payload": <data to send>
}

Since endpoints can send multiple messages, each message needs its own entry in the pub_data.

Parameters:: reset (bool, optional) – When set erases published value which prevents re-publication of the value until manually set to a non-None value. Any entry in this dictionary that is None is not sent out via HELICS. Defaults to False.
Returns:: None

set_metadata() → None

Sets instance attributes to enable HELICS config query of metadata data store.

HELICS configuration information is generally stored in the metadataDB and is copied into the self.federation attribute. This method pulls out a few keys configuration parameters from that attribute to make them more easily accessible.

Returns:: None

simulate_next_step() → None

Advances the Federate to its next simulated time

This method is the core of the main co-simulation loop where the time request is made and once granted, data from the rest of the federation is collected and used to update the internal model before sending out new data for the rest of the federation to use.

Returns:: None

update_internal_model() → None

Perform federate specific calculations to bring model up to date

After receiving inputs from the rest of the federation, each federate updates its internal model, generally using the new inputs to perform the necessary calculations. This aligns the Federate state with that of the rest of the federation

This is entirely user-defined code and is intended to be defined by sub-classing and/or overloading.

Returns:: None

property use_timescale: bool

Sets the timescale flag

Returns:: state of timescale flag
Return type:: bool

write_to_logger(name: str, key: str, value: Any, table: str = None, message_time: float = None, receiving_federate: str = None, receiving_endpoint: str = None)

Populates a TSRecord object with the output of a publication or endpoint and adds it to the timeseries data manager queue of data to write.

Parameters:

name (str) – Name of the federate sending the data
key (str) – Name of the key associated with the data (pub or
name) (endpoint)
value (Any) – Value being sent
table (str, optional) – DEPRECATED. Defaults to None.
message_time (float, optional) – Simulation ordinal time when data is being sent. Defaults to None.
receiving_federate (str, optional) – Name of federate being sent the message (endpoint only). Defaults to None.
receiving_endpoint (str, optional) – Name of endpoint receiving the message (endpoint only). Defaults to None.

class cosim_toolbox.sims.FederateConfig(name: str, **kwargs)

Bases: object

config(_n: str, _v: any) → dict

Adds key specified by first parameter with value specified by the second parameter to the federate config (“_fed_cnfg”) attribute of this object

Parameters:

_n (str) – Key under which new attribute will be added
_v (any) – Value added to dictionary

Returns:

Dictionary to which the new value was added.

Return type:

dict

define_io()

docker(address: int = 0)

find_input_group(key: str) → None | tuple

find_output_group(key: str) → None | tuple

unique() → str

class cosim_toolbox.sims.FederateLogger(fed_name: str = 'logger', **kwargs)

Bases: Federate

get_helics_config() → None

Sets a few class attributes related to HELICS configuration.

Also determines which publications need to be pushed into the time-series database.

Overload of Federate method

update_internal_model() → None: Takes latest published values or sent messages (endpoints) and pushes them back into the time-series database using the base class methods.

class cosim_toolbox.sims.FederationConfig(scenario_name: str, analysis_name: str, federation_name: str, docker: bool = False, use_meta_db: str = 'mongo', use_data_db: str = 'postgres')

Bases: object

add_federate_config(fed: FederateConfig)

add_group(name: str, data_type: str, key_format: dict, **kwargs)

static add_group_subs(pub_group: HelicsPubGroup, sub_group: HelicsSubGroup, des_format: dict)

add_pub_sub(from_fed: HelicsMsg, to_fed: HelicsMsg, v_name: str, v_type: str, v_unit: str)

add_subs(from_fed: str, to_fed_list: list, info: bool = False)

check_pubs() → dict

check_subs() → dict

define_io()

del_federate_config(name: str)

get_federation_document() → dict

Builds and returns the complete federation configuration document.

This method iterates through all registered federates, generates their individual HELICS JSON configurations, and compiles them into a single dictionary that defines the entire federation.

Returns:

The federation configuration document, ready to be: serialized (e.g., to a JSON file or database).

Return type:

dict

get_scenario_document(start_time: str, stop_time: str) → dict

Builds and returns the scenario configuration document.

Parameters:

start_time (str) – The simulation start time (e.g., “2023-12-07T15:31:27”).
stop_time (str) – The simulation stop time (e.g., “2023-12-08T15:31:27”).

Returns:

The scenario configuration document, ready to be: serialized.

Return type:

dict

write_config(start, stop)

write_file_helics_configs(): This method iterates through all registered federates, generates their individual HELICS JSON configurations, and write them on disk.

class cosim_toolbox.sims.HelicsMsg(name: str, **kwargs)

Bases: object

Provides a data structure for building up the HELICS configuration definitions for publications and subscriptions.

collect(collect: Collect) → None

Todo

Parameters:: collect (Collect) – _description_

config(_n: str, _v: any) → dict

Adds key specified by first parameter with value specified by the second parameter to the config (“_cnfg”) attribute of this object

Parameters:

_n (str) – Key under which new attribute will be added
_v (any) – Value added to dictionary

Returns:

Dictionary to which the new value was added.

Return type:

dict

end_point(diction: dict, _c: Collect = None) → None

endpt(_k: str, _d: list | str, _g: bool = None, _c: Collect = None) → None

get_pubs()

get_subs()

publication(diction: dict, _c: Collect = None) → None

pubs(_k: str, _t: str, _o: str, _p: str, _g: bool = True, _c: Collect = None) → None

Defines a HELICS publication definition and adds it to the “_pubs” attribute of this object. This API supports the definition of the publication “info” field which is used by GridLAB-D to link the publication to the GridLAB-D object. This API does not include support of the publication “unit” field.

Parameters:

_k (str) – HELICS key (name) of publication
_t (str) – HELICS data type of publication
_o (str) – HELICS “info” object name
_p (str) – HELICS “info” object property associate with the name
_g (bool, optional) – Indicates whether publication is global in the HELICS namespace. Defaults to True.
_c (Collect, optional) – Collect object used by the logger. Defaults to None.

pubs_e(_k: str, _t: str, _u: str, _g: bool = None, _c: Collect = None) → None

Defines a HELICS publication definition and adds it to the “_pubs” attribute of this object. Includes support for the publication “unit” field.

Parameters:

_k (str) – HELICS key (name) of publication
_t (str) – HELICS data type of publication
_u (str) – HELICS unit of publication
_g (bool, optional) – Indicates whether publication is global in the HELICS namespace. Defaults to True.
_c (Collect, optional) – Collect object used by the logger. Defaults to None.

pubs_n(_k: str, _t: str, _g: bool = True, _c: Collect = None) → None

Defines a HELICS publication definition and adds it to the “_pubs” attribute of this object. Does not include support for the “info” field used by GridLAB-D for HELICS configuration. Does not include support of the publication “unit” field.

Parameters:

_k (str) – HELICS key (name) of publication
_t (str) – HELICS data type of publication
_g (bool, optional) – Indicates whether publication is global in the HELICS namespace. Defaults to True.
_c (Collect, optional) – Collect object used by the logger. Defaults to None.

subs(_k: str, _t: str, _o: str, _p: str) → None

Defines a HELICS subscription definition and adds it to the “_subs” attribute of this object. This API supports the definition of the subscription “info” field which is used by GridLAB-D to link the publication to the GridLAB-D object. This does not include support of the subscription “unit” field.

Parameters:

_k (str) – HELICS key of subscription indicating which publication this subscription is linked to.
_t (str) – HELICS data type of subscription
_o (str) – HELICS “info” object name
_p (str) – HELICS “info” object property associate with the name

subs_e(_k: str, _t: str, _u: str, _r: bool = None) → None

Defines a HELICS subscription definition and adds it to the “_subs” attribute of this object. This API supports the EnergyPlus to link the subscription to the EnergyPlus object. This supports the subscription “connection_required” flag.

Parameters:

_k (str) – HELICS key of subscription indicating which publication this subscription is linked to.
_t (str) – HELICS data type of subscription
_u (str) – unit name
_r (bool, optional) – HELICS “required” flag. Setting this flag will cause HELICS to throw an error if the HELICS subscription does not connect to the publication indicated by the “key” field. Defaults to None.

subs_n(_k, _t) → None

Defines a HELICS subscription definition and adds it to the “_subs” attribute of this object. This API does not support the subscription “info”, “required”, or “type” field.

Parameters:

_k (str) – HELICS key of subscription indicating which publication this subscription is linked to.
_t (str) – HELICS data type of subscription

subscribe_from_published(h_msg: object, varfilter: str)

subscription(diction: dict) → None

static verify(diction: dict, name: str, value: any)

write_file(_fn: str) → None

Adds publications and subscriptions to the objects “_cnfg” (configuration) attribute and writes it to the specified file.

Parameters:: _fn (str) – File name (including path) to which configuration will be written.

write_json() → dict

Adds publications and subscriptions to the objects “_cnfg” (configuration) attribute and returns it as a dictionary.

Returns:: Configuration dict after adding publications and subscriptions
Return type:: dict

class cosim_toolbox.sims.HelicsPubGroup(name: str, data_type: str, key_format: dict, **kwargs): Bases: HelicsFormatter

class cosim_toolbox.sims.HelicsSubGroup(name: str, data_type: str, key_format: dict = None, **kwargs): Bases: HelicsFormatter

Submodules

cosim_toolbox.sims.bench_profile.bench_profile(func)

class cosim_toolbox.sims.dockerRunner.DockerRunner

Bases: object

Collection of static methods used in building and running the docker-compose.yaml for running a new service or simulator.

static define_sh(scenario_name: str, use_meta_db: str = 'mongo', use_data_db: str = 'postgres') → None

Create the run shell file from the provided scenario

Parameters:

scenario_name (str) – Name of the scenario
use_meta_db (str)
use_data_db (str)

static define_yaml(scenario_name: str, use_meta_db: str = 'mongo', use_data_db: str = 'postgres') → None

Create the docker-compose.yaml from the provided scenario

Parameters:

scenario_name (str) – Name of the scenario
use_meta_db (str)
use_data_db (str)

static run_remote_yaml(scenario_name: str, path: str = '/run/python/test_federation') → None

Runs the docker-compose.yaml on a remote compute node: TODO: Test, ‘path’ have not been set in examples at this point

Parameters:

scenario_name (str) – Name of the scenario run by this docker-compose.yaml
path (str, optional) – Path to docker-compose-yaml on remote hose. Defaults to “/run/python/test_federation”.

static run_yaml(scenario_name: str) → None

Runs the provided scenario by calling the appropriate docker-compose.yaml

Parameters:: scenario_name (str) – Name of the scenario run by this docker-compose.yaml

Created on 12/14/2023

Federate class that defines the basic operations of Python-based federates in CoSim Toolbox (CST).

@author: Trevor Hardy trevor.hardy@pnnl.gov

class cosim_toolbox.sims.federate.Federate(fed_name: str = '', *, debug: bool = True)

Bases: object

This class definition is intended to be a reasonable, generic class for Python-based federates in HELICS. It outlines the typical federate operational procedure in the “_main_” function; users that don’t need anything fancy will probably be able to call those few functions and get a working federate.

This class gets its configuration from the metadata database following the standard CST definition of the “federations” document.

To be overly clear, this class is intended to be sub-classed and overloaded to allow users to customize it as necessary. If nothing else, the “update_model” method will always need updating to perform the particular calculations federate is responsible for. There are several other methods that are likely candidates for subclassing based on the particular needs of the federate or the larger federation: “enter_initializing_mode” “enter_executing_mode” “time_request”

All of these have the simplest version of these HELICS operations but there are more complex versions that HELICS supports that allow for things like iterations and asynchronous or non-blocking operations (further details can be found in the HELICS documentation).

The existing methods to pull in values from the HELICS federation and push values out are likely to be sufficient for most federates but, again these can be overloaded in a subclass if, for example, the number of HELICS inputs and publications is very large and only a few are used during a given time-step.

hfed: The HELICS federate object used to access the HELICS interfaces federate_name (str): The federate name

federate

Dictionary with all configuration information, including but not limited to the HELICS JSON config string

Type:: dict

federate_type

The federate type. Must be “value”, “message”, or “combo”

Type:: str

config

Valid HELICS config JSON string

Type:: dict

granted_time

The last time granted to this federate

Type:: float

period

The size of the simulated time step takes when requesting the next time

Type:: float

scenario_name

The scenario name

Type:: str

scenario

Dictionary with all scenario configuration information

Type:: dict

federation_name

The federation name

Type:: str

federation

Dictionary with all federate configuration information

Type:: dict

calculate_next_requested_time() → float

Determines the next simulated time to request from HELICS

Many federates run at very regular time steps and thus the calculation of the requested time is trivial. In some cases, though, the requested time may be more dynamic and this method provides a place for users to overload the default calculation method if they need something more complex.

Returns:

Calculated time for the next HELICS time: request

Return type:

self.next_requested_time

create_federate(scenario_name: str, use_meta_db: str = 'mongo', use_data_db: str = 'postgres') → None

Create CST and HELICS federates

Creates and defines both the instance of this class,(the Co-Simulation federate) and the HELICS federate object (self.hfed). Any initialization that cannot take place on instantiation of the federate object should be done here. In this case, initializing any class attribute values that come from the metadata database have to take place after connecting to said database.

Parameters:

scenario_name (str) – Name of scenario used to store configuration information in the dbConfigs
use_meta_db (str, optional) – Whether to use a metadata database (e.g., MongoDB). Defaults to mongo.
use_data_db (str, optional) – Whether to use a timeseries database (e.g., PostgreSQL). Defaults to postgres.

Raises:

NameError – Scenario name is undefined (None)

Returns:

None

create_helics_fed() → None

Creates the HELICS federate object

Using the HELICS configuration document from the dbConfigs, this method creates the HELICS federate. HELICS has distinct APIs for the creation of a federate based on its type and thus, the type of federate needs to be defined as an instance attribute to enable the correct API to be called.

Raises:: ValueError – Invalid value for self.federate_type
Returns:: None

property current_time

destroy_federate() → None

Removes HELICS federate from federation

As part of ending a HELICS co-simulation it is good housekeeping to formally destroy the model federate. Doing so informs the rest of the federation that it is no longer a part of the co-simulation and they should proceed without it (if applicable). Generally this is done when the co-simulation is complete and all federates end execution at more or less the same wall-clock time.

enter_executing_mode() → None

Moves the Federate to executing mode

Similar to initializing mode, there are a few different ways of handling HELICS executing mode and what is implemented here is the simplest. If you need something more complex or specific, overload or redefine this method.

Returns:: None

enter_initialization() → None

Moves federate to HELICS initializing mode

There are a few stages to a federate in HELICS with initializing mode being the first after the Federate is created. Entering initializing mode is a global synchronous event for all federates and provides an opportunity to do some fancy things around dynamic configuration of the Federate. What is implemented here is the simplest, most vanilla means of entering initializing mode. If you need something more complex, overload or redefine this method.

Returns:: None

get_data_from_federation() → None

Collects inputs from federation and stores them

This method is an automated way of getting data the rest of the federation has sent out. Directly accessing the value and message interfaces via the HELICS federate (hfed object) provides a much richer set of metadata associated with these interfaces. The implementation here is vanilla and is expected to be sufficient for many use cases.

Returns:: None

get_helics_config() → None

Sets instance attributes to enable HELICS config query of dbConfigs

HELICS configuration information is generally stored in the dbConfigs and is copied into the self.federation attribute. This method pulls out a few keys configuration parameters from that attribute to make them more easily accessible.

Returns:: None

on_enter_executing_mode() → None

Functionality executed after entering executing mode

By default, no functionality is implemented

Returns:: None

on_enter_initialization_mode() → None

Functionality executed after entering initialization

By default, no functionality is implemented

Returns:: None

on_start() → None

Functionality executed prior to entering initializing

By default, no functionality is implemented

Returns:: None

request_time(requested_time: float) → float

Requests next simulated time from HELICS

HELICS provides a variety of means of requesting time. The most common is a simple hfed.request_time(float) which is a blocking call. There are others that make the time request but allow users to continue working on something else while they wait for HELICS to get back to them with the granted time. This method is here just to allow users to redefine or overload and re-implement how they want to do time requests.

Parameters:: requested_time – Simulated time this federate needs to request
Returns:: Simulated time granted by HELICS
Return type:: self.granted_time

reset_data_to_federation() → None

Sets all values in dictionary of values being sent out via publications and endpoints in the data_to_federation dictionary to “None”.

Any values in these dictionaries set to None do not result in a new output via HELICS. This method wipes out all data so that only entries added to the dictionary after calling this method will be published, preventing duplicate publication of data that has not changed and does not need to be re-sent. This also helps manage the data being logged in the time-series database.

Returns:: None

run(scenario_name: str, use_meta_db: str = 'mongo', use_data_db: str = 'postgres') → None

Runs the HELICS federate class

Parameters:

scenario_name (str) – Name of scenario used to store configuration information in the dbConfigs
use_meta_db (str, optional) – Whether to use a metadata database (e.g., MongoDB). Defaults to mongo.
use_data_db (str, optional) – Whether to use a timeseries database (e.g., PostgreSQL). Defaults to postgres.

run_cosim_loop() → None

Runs the generic HELICS co-sim loop

This HELICS co-sim loop runs until it the simulated time reaches self.stop_time. self.enter_initialization_mode() and self.enter_executing_mode(), and self. simulate_next_step have been implemented and should be overloaded/redefined as necessary to fit the needs of a given federate and/or federation.

Returns:: None

send_data_to_federation(reset=False) → None

Sends specified outputs to rest of HELICS federation

This method provides an easy way for users to send out any data to the rest of the federation. Users pass in a dict structured the same as the “data_from_federation” with sub-dicts for publications and endpoints and keys inside those dicts for the name of the pub or endpoint. The value for the keys is slightly different, though:

pubs: value is the data to send endpoints: value is a dictionary as follows:
{
    "destination": <target endpoint name, may be an empty string>
    "payload": <data to send>
}

Since endpoints can send multiple messages, each message needs its own entry in the pub_data.

Parameters:: reset (bool, optional) – When set erases published value which prevents re-publication of the value until manually set to a non-None value. Any entry in this dictionary that is None is not sent out via HELICS. Defaults to False.
Returns:: None

set_metadata() → None

Sets instance attributes to enable HELICS config query of metadata data store.

HELICS configuration information is generally stored in the metadataDB and is copied into the self.federation attribute. This method pulls out a few keys configuration parameters from that attribute to make them more easily accessible.

Returns:: None

simulate_next_step() → None

Advances the Federate to its next simulated time

This method is the core of the main co-simulation loop where the time request is made and once granted, data from the rest of the federation is collected and used to update the internal model before sending out new data for the rest of the federation to use.

Returns:: None

update_internal_model() → None

Perform federate specific calculations to bring model up to date

After receiving inputs from the rest of the federation, each federate updates its internal model, generally using the new inputs to perform the necessary calculations. This aligns the Federate state with that of the rest of the federation

This is entirely user-defined code and is intended to be defined by sub-classing and/or overloading.

Returns:: None

property use_timescale: bool

Sets the timescale flag

Returns:: state of timescale flag
Return type:: bool

write_to_logger(name: str, key: str, value: Any, table: str = None, message_time: float = None, receiving_federate: str = None, receiving_endpoint: str = None)

Populates a TSRecord object with the output of a publication or endpoint and adds it to the timeseries data manager queue of data to write.

Parameters:

name (str) – Name of the federate sending the data
key (str) – Name of the key associated with the data (pub or
name) (endpoint)
value (Any) – Value being sent
table (str, optional) – DEPRECATED. Defaults to None.
message_time (float, optional) – Simulation ordinal time when data is being sent. Defaults to None.
receiving_federate (str, optional) – Name of federate being sent the message (endpoint only). Defaults to None.
receiving_endpoint (str, optional) – Name of endpoint receiving the message (endpoint only). Defaults to None.

Created on 12/14/2023

Data logger class that defines the basic operations of Python-based logger federate in Cosim Toolbox. This is instantiated to create a Federate that collects data from the federates sent via HELICS and pushes it into the time-series database. All the HELICS functionality is contained in the Federate class.

@author: mitch.pelton@pnnl.gov

class cosim_toolbox.sims.federateLogger.FederateLogger(fed_name: str = 'logger', **kwargs)

Bases: Federate

get_helics_config() → None

Sets a few class attributes related to HELICS configuration.

Also determines which publications need to be pushed into the time-series database.

Overload of Federate method

update_internal_model() → None: Takes latest published values or sent messages (endpoints) and pushes them back into the time-series database using the base class methods.

cosim_toolbox.sims.federateLogger.main(federate_name: str, scenario_name: str, use_meta_db: str = 'mongo', use_data_db: str = 'postgres') → None

Created on 30 June 2025

Defines the Federation class which is used to programmatically define a federation of federate class modules the pubs and subs of those federates using helicsMsg class and write it out to a federation configuration JSON.

@author: mitch.pelton@pnnl.gov

class cosim_toolbox.sims.federation.FederateConfig(name: str, **kwargs)

Bases: object

config(_n: str, _v: any) → dict

Adds key specified by first parameter with value specified by the second parameter to the federate config (“_fed_cnfg”) attribute of this object

Parameters:

_n (str) – Key under which new attribute will be added
_v (any) – Value added to dictionary

Returns:

Dictionary to which the new value was added.

Return type:

dict

define_io()

docker(address: int = 0)

find_input_group(key: str) → None | tuple

find_output_group(key: str) → None | tuple

unique() → str

class cosim_toolbox.sims.federation.FederationConfig(scenario_name: str, analysis_name: str, federation_name: str, docker: bool = False, use_meta_db: str = 'mongo', use_data_db: str = 'postgres')

Bases: object

add_federate_config(fed: FederateConfig)

add_group(name: str, data_type: str, key_format: dict, **kwargs)

static add_group_subs(pub_group: HelicsPubGroup, sub_group: HelicsSubGroup, des_format: dict)

add_pub_sub(from_fed: HelicsMsg, to_fed: HelicsMsg, v_name: str, v_type: str, v_unit: str)

add_subs(from_fed: str, to_fed_list: list, info: bool = False)

check_pubs() → dict

check_subs() → dict

define_io()

del_federate_config(name: str)

get_federation_document() → dict

Builds and returns the complete federation configuration document.

This method iterates through all registered federates, generates their individual HELICS JSON configurations, and compiles them into a single dictionary that defines the entire federation.

Returns:

The federation configuration document, ready to be: serialized (e.g., to a JSON file or database).

Return type:

dict

get_scenario_document(start_time: str, stop_time: str) → dict

Builds and returns the scenario configuration document.

Parameters:

start_time (str) – The simulation start time (e.g., “2023-12-07T15:31:27”).
stop_time (str) – The simulation stop time (e.g., “2023-12-08T15:31:27”).

Returns:

The scenario configuration document, ready to be: serialized.

Return type:

dict

write_config(start, stop)

write_file_helics_configs(): This method iterates through all registered federates, generates their individual HELICS JSON configurations, and write them on disk.

Created on 20 Dec 2023

Defines the HelicsMsg class which is used to programmatically define pubs and subs of a HELICS class and write it out to a HELICS federation configuration JSON.

@author: mitch.pelton@pnnl.gov

class cosim_toolbox.sims.helicsConfig.Collect(*values)

Bases: Enum

MAYBE = 'maybe'

NO = 'no'

YES = 'yes'

class cosim_toolbox.sims.helicsConfig.Group(*values)

Bases: Enum

EPT = 'ept'

INP = 'inp'

PUB = 'pub'

SUB = 'sub'

class cosim_toolbox.sims.helicsConfig.HelicsEndPtGroup(name: str, key_format: dict, destination: str, des_format: dict, **kwargs): Bases: HelicsEndpointFormatter

class cosim_toolbox.sims.helicsConfig.HelicsEndpointFormatter(name: str, key_format: dict, destination: str, des_format: dict, group: Group)

Bases: object

format_endpoints()

class cosim_toolbox.sims.helicsConfig.HelicsFormatter(name: str, key_format: dict, group: Group)

Bases: object

format_variables()

class cosim_toolbox.sims.helicsConfig.HelicsMsg(name: str, **kwargs)

Bases: object

Provides a data structure for building up the HELICS configuration definitions for publications and subscriptions.

collect(collect: Collect) → None

Todo

Parameters:: collect (Collect) – _description_

config(_n: str, _v: any) → dict

Adds key specified by first parameter with value specified by the second parameter to the config (“_cnfg”) attribute of this object

Parameters:

_n (str) – Key under which new attribute will be added
_v (any) – Value added to dictionary

Returns:

Dictionary to which the new value was added.

Return type:

dict

end_point(diction: dict, _c: Collect = None) → None

endpt(_k: str, _d: list | str, _g: bool = None, _c: Collect = None) → None

get_pubs()

get_subs()

publication(diction: dict, _c: Collect = None) → None

pubs(_k: str, _t: str, _o: str, _p: str, _g: bool = True, _c: Collect = None) → None

Defines a HELICS publication definition and adds it to the “_pubs” attribute of this object. This API supports the definition of the publication “info” field which is used by GridLAB-D to link the publication to the GridLAB-D object. This API does not include support of the publication “unit” field.

Parameters:

_k (str) – HELICS key (name) of publication
_t (str) – HELICS data type of publication
_o (str) – HELICS “info” object name
_p (str) – HELICS “info” object property associate with the name
_g (bool, optional) – Indicates whether publication is global in the HELICS namespace. Defaults to True.
_c (Collect, optional) – Collect object used by the logger. Defaults to None.

pubs_e(_k: str, _t: str, _u: str, _g: bool = None, _c: Collect = None) → None

Defines a HELICS publication definition and adds it to the “_pubs” attribute of this object. Includes support for the publication “unit” field.

Parameters:

_k (str) – HELICS key (name) of publication
_t (str) – HELICS data type of publication
_u (str) – HELICS unit of publication
_g (bool, optional) – Indicates whether publication is global in the HELICS namespace. Defaults to True.
_c (Collect, optional) – Collect object used by the logger. Defaults to None.

pubs_n(_k: str, _t: str, _g: bool = True, _c: Collect = None) → None

Defines a HELICS publication definition and adds it to the “_pubs” attribute of this object. Does not include support for the “info” field used by GridLAB-D for HELICS configuration. Does not include support of the publication “unit” field.

Parameters:

_k (str) – HELICS key (name) of publication
_t (str) – HELICS data type of publication
_g (bool, optional) – Indicates whether publication is global in the HELICS namespace. Defaults to True.
_c (Collect, optional) – Collect object used by the logger. Defaults to None.

subs(_k: str, _t: str, _o: str, _p: str) → None

Defines a HELICS subscription definition and adds it to the “_subs” attribute of this object. This API supports the definition of the subscription “info” field which is used by GridLAB-D to link the publication to the GridLAB-D object. This does not include support of the subscription “unit” field.

Parameters:

_k (str) – HELICS key of subscription indicating which publication this subscription is linked to.
_t (str) – HELICS data type of subscription
_o (str) – HELICS “info” object name
_p (str) – HELICS “info” object property associate with the name

subs_e(_k: str, _t: str, _u: str, _r: bool = None) → None

Defines a HELICS subscription definition and adds it to the “_subs” attribute of this object. This API supports the EnergyPlus to link the subscription to the EnergyPlus object. This supports the subscription “connection_required” flag.

Parameters:

_k (str) – HELICS key of subscription indicating which publication this subscription is linked to.
_t (str) – HELICS data type of subscription
_u (str) – unit name
_r (bool, optional) – HELICS “required” flag. Setting this flag will cause HELICS to throw an error if the HELICS subscription does not connect to the publication indicated by the “key” field. Defaults to None.

subs_n(_k, _t) → None

Defines a HELICS subscription definition and adds it to the “_subs” attribute of this object. This API does not support the subscription “info”, “required”, or “type” field.

Parameters:

_k (str) – HELICS key of subscription indicating which publication this subscription is linked to.
_t (str) – HELICS data type of subscription

subscribe_from_published(h_msg: object, varfilter: str)

subscription(diction: dict) → None

static verify(diction: dict, name: str, value: any)

write_file(_fn: str) → None

Adds publications and subscriptions to the objects “_cnfg” (configuration) attribute and writes it to the specified file.

Parameters:: _fn (str) – File name (including path) to which configuration will be written.

write_json() → dict

Adds publications and subscriptions to the objects “_cnfg” (configuration) attribute and returns it as a dictionary.

Returns:: Configuration dict after adding publications and subscriptions
Return type:: dict

class cosim_toolbox.sims.helicsConfig.HelicsPubGroup(name: str, data_type: str, key_format: dict, **kwargs): Bases: HelicsFormatter

class cosim_toolbox.sims.helicsConfig.HelicsSubGroup(name: str, data_type: str, key_format: dict = None, **kwargs): Bases: HelicsFormatter