Co-simulation Interfaces¶

The HELICS Interface¶

Hierarchical Engine for Large-scale Infrastructure Co-Simulation (HELICS) provides an open-source, general-purpose, modular, highly-scalable co-simulation framework that runs cross-platform (Linux, Windows, and Mac OS X). It is not a modeling tool by itself, but rather an integration tool that enables multiple existing simulation tools (known as “federates”) to exchange data during runtime and stay synchronized in time such that together they act as one large simulation, or “federation”.

The HELICS interface for PyDSS is built to reduce the complexity of setting up large-scale co-simulation scenarios. The user defines publications and subscriptions to exchange data with external federates.

A minimal HELICS example is availble in the examples folder (top directory of the repository). Enabling the HELICS interface requires user to define additional parammeters in the scenario TOML file.

Interface Overview¶

The HELICS interface can be enabled and configured using the simulation.toml file. The following attributes can be configured:

pydantic model pydss.simulation_input_models.HelicsModel[source]¶

Defines the user inputs for HELICS.

Show JSON schema

{
   "title": "InputsBaseModel",
   "description": "Defines the user inputs for HELICS.",
   "type": "object",
   "properties": {
      "Co-simulation Mode": {
         "default": false,
         "description": "Set to true to enable the HELICS interface for co-simulation.",
         "title": "co_simulation_mode",
         "type": "boolean"
      },
      "Iterative Mode": {
         "default": false,
         "description": "Iterative mode",
         "title": "iterative_mode",
         "type": "boolean"
      },
      "Error tolerance": {
         "default": 0.0001,
         "description": "Error tolerance",
         "title": "error_tolerance",
         "type": "number"
      },
      "Max co-iterations": {
         "default": 15,
         "description": "Max number of co-simulation iterations",
         "title": "max_co_iterations",
         "type": "integer"
      },
      "Broker": {
         "default": "mainbroker",
         "description": "Broker name",
         "title": "broker",
         "type": "string"
      },
      "Broker port": {
         "default": 0,
         "description": "Broker port",
         "title": "broker_port",
         "type": "integer"
      },
      "Federate name": {
         "default": "pydss",
         "description": "Name of the federate",
         "title": "federate_name",
         "type": "string"
      },
      "Time delta": {
         "default": 0.01,
         "description": "The property controlling the minimum time delta for a federate.",
         "title": "time_delta",
         "type": "number"
      },
      "Core type": {
         "default": "zmq",
         "description": "Core type to be use for communication",
         "title": "core_type",
         "type": "string"
      },
      "Uninterruptible": {
         "default": true,
         "description": "Can the federate be interrupted",
         "title": "uninterruptible",
         "type": "boolean"
      },
      "Helics logging level": {
         "default": 5,
         "description": "Logging level for the federate. Refer to HELICS documentation.",
         "title": "logging_level",
         "type": "integer"
      }
   },
   "additionalProperties": false
}

Config:

title: str = InputsBaseModel
str_strip_whitespace: bool = True
validate_assignment: bool = True
validate_default: bool = True
extra: str = forbid
use_enum_values: bool = False
populate_by_name: bool = True

Fields:

Validators:

field broker: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='mainbroker', alias='Broker', alias_priority=2, title='broker', description='Broker name')] = 'mainbroker' (alias 'Broker')¶: Broker name

field broker_port: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=0, alias='Broker port', alias_priority=2, title='broker_port', description='Broker port')] = 0 (alias 'Broker port')¶: Broker port

field co_simulation_mode: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=False, alias='Co-simulation Mode', alias_priority=2, title='co_simulation_mode', description='Set to true to enable the HELICS interface for co-simulation.')] = False (alias 'Co-simulation Mode')¶: Set to true to enable the HELICS interface for co-simulation.

field core_type: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='zmq', alias='Core type', alias_priority=2, title='core_type', description='Core type to be use for communication')] = 'zmq' (alias 'Core type')¶: Core type to be use for communication

field error_tolerance: Annotated[float, FieldInfo(annotation=NoneType, required=False, default=0.0001, alias='Error tolerance', alias_priority=2, title='error_tolerance', description='Error tolerance')] = 0.0001 (alias 'Error tolerance')¶: Error tolerance

field federate_name: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='pydss', alias='Federate name', alias_priority=2, title='federate_name', description='Name of the federate')] = 'pydss' (alias 'Federate name')¶: Name of the federate

field iterative_mode: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=False, alias='Iterative Mode', alias_priority=2, title='iterative_mode', description='Iterative mode')] = False (alias 'Iterative Mode')¶: Iterative mode

field logging_level: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=5, alias='Helics logging level', alias_priority=2, title='logging_level', description='Logging level for the federate. Refer to HELICS documentation.')] = 5 (alias 'Helics logging level')¶

Logging level for the federate. Refer to HELICS documentation.

Validated by:

check_logging_level

field max_co_iterations: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=15, alias='Max co-iterations', alias_priority=2, title='max_co_iterations', description='Max number of co-simulation iterations')] = 15 (alias 'Max co-iterations')¶

Max number of co-simulation iterations

Validated by:

check_max_co_iterations

field time_delta: Annotated[float, FieldInfo(annotation=NoneType, required=False, default=0.01, alias='Time delta', alias_priority=2, title='time_delta', description='The property controlling the minimum time delta for a federate.')] = 0.01 (alias 'Time delta')¶: The property controlling the minimum time delta for a federate.

field uninterruptible: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=True, alias='Uninterruptible', alias_priority=2, title='uninterruptible', description='Can the federate be interrupted')] = True (alias 'Uninterruptible')¶: Can the federate be interrupted

validator check_logging_level » logging_level[source]¶

validator check_max_co_iterations » max_co_iterations[source]¶

co_simulation_mode: Set to true to enable the HELICS interface (default: false).
federate_name: Required to identify a federate in a co-simulation with many federates.
Additional settings for convergence, timing, and iteration can be configured here.

For more information on these values, refer to the HELICS documentation.

Once the HELICS co-simulation interface has been enabled, the next step is to set up publications and subscriptions for data exchange with external federates. PyDSS enables zero-code setup of these modules. Each scenario can have its own publication and subscription definitions, managed by files in the ExportLists directory.

Publication tags (names) follow this convention:

<federate name>.<object type>.<object name>.<object property>

examples,

federate1.Circuit.70008.TotalPower
federate1.Load.load_1.VoltageMagAng

where federate name is defined in the project’s simulation.toml file.

Setting up Publications¶

Publications (data sent to external federates) can be configured using Exports.toml. This file is also used to define export variables for a simulation scenario. By setting the publish attribute to true, PyDSS automatically sets up a HELICS publication. The file supports multiple filtering options including regex to publish only what is needed.

examples:

The following example sets up publications for all PV system powers in the model. Setting publish to false will still write the data to the HDF5 store, but will not publish it on the HELICS interface.

[[PVSystems]]
property = "Powers"
sample_interval = 1
publish = true
store_values_type = "all"

There are two options to filter and publish a subset of elements. You can use the name_regexes attribute to filter elements matching regex expressions, or use the names attribute to explicitly list elements.

Filtering using regex expressions:

[[PVSystems]]
property = "Powers"
name_regexes = [".*pvgnem.*"]
sample_interval = 1
publish = true
store_values_type = "all"

Filtering using explicit element names:

[[PVSystems]]
property = "Powers"
sample_interval = 1
names = ["PVSystems.pv1", "PVSystems.pv2"]
publish = true
store_values_type = "all"

Setting up Subscriptions¶

Subscriptions (data received from external federates) are configured using Subscriptions.toml in the ExportLists directory for a given scenario. Valid subscriptions should conform to the following model:

pydantic model pydss.helics_interface.Subscription[source]¶

Show JSON schema

{
   "title": "Subscription",
   "type": "object",
   "properties": {
      "model": {
         "title": "Model",
         "type": "string"
      },
      "property": {
         "title": "Property",
         "type": "string"
      },
      "id": {
         "title": "Id",
         "type": "string"
      },
      "unit": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Unit"
      },
      "subscribe": {
         "default": true,
         "title": "Subscribe",
         "type": "boolean"
      },
      "data_type": {
         "$ref": "#/$defs/DataType"
      },
      "multiplier": {
         "default": 1.0,
         "title": "Multiplier",
         "type": "number"
      },
      "object": {
         "default": null,
         "title": "Object"
      },
      "states": {
         "default": [
            0.0,
            0.0,
            0.0,
            0.0,
            0.0
         ],
         "items": {
            "anyOf": [
               {
                  "type": "number"
               },
               {
                  "type": "integer"
               },
               {
                  "type": "boolean"
               }
            ]
         },
         "title": "States",
         "type": "array"
      },
      "sub": {
         "default": null,
         "title": "Sub"
      }
   },
   "$defs": {
      "DataType": {
         "enum": [
            "double",
            "vector",
            "string",
            "boolean",
            "integer"
         ],
         "title": "DataType",
         "type": "string"
      }
   },
   "required": [
      "model",
      "property",
      "id",
      "data_type"
   ]
}

Config:

arbitrary_types_allowed: bool = True

Fields:

field data_type: DataType [Required]¶

field id: str [Required]¶

field model: str [Required]¶

field multiplier: float = 1.0¶

field object: Any = None¶

field property: str [Required]¶

field states: List[float | int | bool] = [0.0, 0.0, 0.0, 0.0, 0.0]¶

field sub: Any = None¶

field subscribe: bool = True¶

field unit: str | None = None¶

When setting up subscriptions, note that the subscription tag is generated by the external federate and must be known before configuration. In the example below, values received from subscription tag test.load1.power are used to update the kw property of load Load.mpx000635970. The multiplier property can scale values before they update the model.

example

[[subscriptions]]
model = "Load.mpx000635970"
property = "kw"
id = "test.load1.power"
unit = "kW"
subscribe = true
data_type = "double"
multiplier = 1

A complete example is available in examples/external_interfaces/.

The Socket Interface¶

The socket interface is implemented as a PyDSS controller (pydss.pyControllers.Controllers.SocketController.SocketController). It is well suited for situations where an existing external controller needs to be integrated into the simulation environment — for example, integrating a controller for thermostatically controlled loads implemented in Modelica or Python. This allows integration without modifying the external controller.

The socket interface is also useful for hardware-in-the-loop simulations, integrating the simulation engine with actual hardware. Interfaces for Modbus-TCP and DNP3 communications have been developed and tested with PyDSS. A minimal socket example is provided in examples/external_interfaces/pydss_project. The socket scenario defines the socket controller in its pyControllerList folder. The controller configuration specifies the element to control, the socket connection details, and the data to exchange:

["Load.mpx000635970"]
IP = "127.0.0.1"
Port = 5001
Encoding = false
Buffer = 1024
Index = "Even,Even"
Inputs = "VoltagesMagAng,Powers"
Outputs = "kW"

Finally, the minimal example below shows how to retrive data from the sockets and return new values for parameters defined in the definations file.

# first of all import the socket library
import socket
import struct

# next create a socket object
sockets = []
for i in range(2):
        s = socket.socket()
        s.bind(('127.0.0.1', 5001 + i))
        s.listen(5)
        sockets.append(s)
while True:
        # Establish connection with client.
        conns = []
        for s in sockets:
                c, addr = s.accept()
                conns.append(c)
        while True:
                for c in conns: #Reading data from all ports
                        Data = c.recv(1024)
                        if Data: #Creating a list of doubles from the recieved byte stream
                                numDoubles = int(len(Data) / 8)
                                tag = str(numDoubles) + 'd'
                                Data = list(struct.unpack(tag, Data))
                for c , v in zip(conns, [5, 3]): #Writing data to all ports
                        values = [v]
                        c.sendall(struct.pack('%sd' % len(values), *values))