oedisi.componentframework package

Submodules

oedisi.componentframework.basic_component module

Generate basic component from description JSON.

class oedisi.componentframework.basic_component.ComponentDescription(*, directory: str, execute_function: str, static_inputs: list[~oedisi.componentframework.system_configuration.AnnotatedType], dynamic_inputs: list[~oedisi.componentframework.system_configuration.AnnotatedType], dynamic_outputs: list[~oedisi.componentframework.system_configuration.AnnotatedType], capabilities: ~oedisi.componentframework.system_configuration.ComponentCapabilities = <factory>)

Bases: BaseModel

Component description for simple ComponentType.

Parameters:
capabilities: ComponentCapabilities
directory: str

Where code is stored relative to where this is run

dynamic_inputs: list[AnnotatedType]

List of input types. Typically subscriptions.

dynamic_outputs: list[AnnotatedType]

List of output types. Typically publications.

execute_function: str

Command to execute component

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

static_inputs: list[AnnotatedType]

List of types for the parameter

oedisi.componentframework.basic_component.basic_component(comp_desc: ComponentDescription, type_checker)

Create a new component type from component definition data.

Parameters:

  • comp_desc (ComponentDescription) – Simplified component representation usually from a JSON file

  • type_checker (function taking the type and value and returning a boolean)

Returns:

ComponentType from the description

Return type:

BasicComponent(system_configuration.ComponentType)

oedisi.componentframework.basic_component.component_from_json(filepath, type_checker)

Load component description from JSON file and create component type.

Parameters:

  • filepath (str | Path)

  • type_checker (function taking the type and value and returning a boolean)

Return type:

BasicComponent class

oedisi.componentframework.mock_component module

Mock component and federate for testing HELICS simulations.

MockComponent and MockFederate allow you to instantiate a mock component with a specified set of inputs and outputs. The parameters dictionary should contain a list under “inputs” and “outputs”. During implementation, the value pi is passed around every second until t=100.

MockComponent defines the ComponentType for a simple testing component.

MockFederate defines the corresponding implementation.

class oedisi.componentframework.mock_component.MockComponent(base_config: HELICSFederateConfig, parameters: dict[str, dict[str, str]], directory: str, host: str | None = None, port: int | None = None, comp_type: str | None = None)

Bases: ComponentType

Mock component for testing HELICS-based simulations.

Provides a configurable mock component with dynamic inputs and outputs for use in testing and validation scenarios.

property dynamic_inputs

Dynamic input ports.

property dynamic_outputs

Dynamic output ports.

property execute_function

Path to mock component execution script.

generate_helics_config(outputs)

Generate HELICS configuration file for the mock component.

Parameters:

outputs (dict[str, str]) – Mapping of output port names to HELICS data types.

generate_input_mapping(links)

Generate input mapping file for subscriptions.

Parameters:

links (dict) – Mapping of local input port names to HELICS subscription keys.

process_parameters(parameters)

Process and configure component parameters.

Parameters:

parameters (dict[str, dict[str, str]]) – Configuration dictionary with “inputs” and “outputs” keys.

class oedisi.componentframework.mock_component.MockFederate

Bases: object

Mock HELICS federate for testing simulations.

Loads configuration and subscriptions from files, then publishes test values during simulation.

run()

Execute simulation, publishing values for 100 seconds.

oedisi.componentframework.mock_component.destroy_federate(fed)

Disconnect and free a HELICS federate.

oedisi.componentframework.mock_component.get_default_value(date_type: HelicsDataType)

Return pi value for mock publications.

oedisi.componentframework.system_configuration module

Define common types and methods for configuration.

The final method generate_runner_config brings together a WiringDiagram and a dictionary of ComponentTypes with optional comparison using a compatability checking function.

The ComponentType defines the common interface that component configuration must implement. By default, this can be instantiated using basic_component.py.

The WiringDiagram configuration contains a list of components and the links between them.

class oedisi.componentframework.system_configuration.AnnotatedType(*, type: str, description: str | None = None, unit: str | None = None, port_id: str | None = None)

Bases: BaseModel

Represent the type on component static input, dynamic input, and dynamic output.

Currently not checked in any type checker.

description: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

port_id: str | None

Manually set port id

property port_name

Input or output name which defaults to type name or port_id.

type: str

Name referencing oedisi Pydantic type or empty string

unit: str | None
class oedisi.componentframework.system_configuration.Component(*, name: str, type: str, host: str | None = None, container_port: int | None = None, image: str = '', parameters: dict[str, Any], helics_config_override: SharedFederateConfig | None = None)

Bases: BaseModel

A component configuration in WiringDiagram.

container_port: int | None

Port used in Docker Compose and Kubernetes

helics_config_override: SharedFederateConfig | None
host: str | None

Hostname used in Docker Compose and Kubernetes

image: str

Image used in Docker Compose and Kubernetes

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

Name or identifier of this component instance

parameters: dict[str, Any]

Configuration passed onto each component.

port(port_name: str) → Port

Create Port object for connecting this component at a port name.

type: str

Type of component referencing component type dictionary

classmethod validate_image(image: str, info: ValidationInfo) → str

Add latest tag to name if not specified.

class oedisi.componentframework.system_configuration.ComponentCapabilities(*, version: str = '1.0', broker_config: bool = False)

Bases: BaseModel

Component capability declarations for build-time validation.

Parameters:

  • version (str) – Capabilities schema version.

  • broker_config (bool) – Whether this component supports receiving federate_config in static_inputs.json. If True, the component can be used with WiringDiagram.shared_helics_config.

broker_config: bool
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

version: str
class oedisi.componentframework.system_configuration.ComponentStruct(*, component: Component, links: list[Link])

Bases: BaseModel

Component with its associated links for multi-container configuration.

component: Component

All links with the component as target

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class oedisi.componentframework.system_configuration.ComponentType(base_config: HELICSFederateConfig, parameters: dict[str, dict[str, str]], directory: str, host: str | None = None, port: int | None = None, comp_type: str | None = None)

Bases: ABC

Abstract type for component configuration.

The components define the main restrictions on how components can be configured. In the simplest case, the basic_copmonent function constructs a type from a ComponentDescription which just write files. There are no restrictions, so for example, one possibility is for the component type to interact with a web service.

First, the class is initialized for each component using the name, parameters, and target directory. The federate name in HELICS should be the name initialized here.

Then the dynamic_inputs and dynamic_outputs are used to check types and verify the links used between components. These can depend on the initialziation parameters. The dynamic_outputs should be initialized under the prefix name/.

Next, generate_input_mapping is then called with a mapping of the variable names to the HELICS subscription keys. The individual federate should then use these names to subscribe at the right location. This can also be used for endpoint targets less often.

Finally, the execute_function property defines the command to run the component.

abstract property dynamic_inputs: dict[str, AnnotatedType]

Dictionary of dynamic input port names to types.

abstract property dynamic_outputs: dict[str, AnnotatedType]

Dictionary of dynamic output port names to types.

abstract property execute_function: str

Command to execute the component federate.

abstract generate_input_mapping(links: dict[str, str])

Generate input mapping from link target ports to HELICS subscription keys.

class oedisi.componentframework.system_configuration.Federate(*, directory: str, hostname: str = 'localhost', name: str, exec: str)

Bases: BaseModel

Federate configuration for HELICS CLI runner.

directory: str

Directory in which the execution should take place

exec: str

Command to start component

hostname: str

Hostname for Docker Compose and Kubernetes

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

Name of component (used for logs)

Bases: BaseModel

Connection between component ports in wiring diagram.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

source: str

Name of component that is publishing

source_port: str

Publication/dynamic output name

target: str

Name of component that is subscribing

target_port: str

Dynamic input name mapped to source_port

class oedisi.componentframework.system_configuration.Port(*, name: str, port_name: str)

Bases: BaseModel

Port identifier for creating links between components.

connect(port: Port) → Link

Create link from this port to target port.

Parameters:

  • self (Port) – Source port

  • port (Port) – Target port

Return type:

Link connecting self and port.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

Component name

port_name: str

Dynamic input/output name

class oedisi.componentframework.system_configuration.RunnerConfig(*, name: str, federates: list[Federate])

Bases: BaseModel

HELICS running config for the full simulation.

Examples

Save to JSON

>>> with open(filename, "w") as f:
...    json.dump(config.model_dump(mode="json"), f)

Run Simulation

$ helics run –path=filename

federates: list[Federate]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str
class oedisi.componentframework.system_configuration.WiringDiagram(*, name: str, components: list[Component], links: list[Link], shared_helics_config: SharedFederateConfig | None = None)

Bases: BaseModel

Cosimulation configuration. This may end up wrapped in another interface.

Parameters:
add_component(c: Component) → None

Add component to wiring diagram.

Add link to wiring diagram.

classmethod check_component_names(components)

Check that the components all have unique names.

Validate that link source and target components exist.

clean_model(target_directory='.') → None

Remove component directories, log files, and stray broker processes.

Parameters:

target_directory (str = ".") – Build folder with component directories and logs.

components: list[Component]

Component configuration and types

classmethod empty(name='unnamed')

Create empty wiring diagram with no components or links.

Create mapping from component names to their incoming links.

List of links {source, source_port, target, target_port} between components

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

Name of the simulation

shared_helics_config: SharedFederateConfig | None
oedisi.componentframework.system_configuration.generate_runner_config(wiring_diagram: ~oedisi.componentframework.system_configuration.WiringDiagram, component_types: dict[str, type[~oedisi.componentframework.system_configuration.ComponentType]], compatibility_checker=<function _bad_compatability_checker>, target_directory='.')

Create HELICS run configuration from wiring diagram and component types.

Parameters:

  • wiring_diagram (WiringDiagram) – Configuration describing components, parameters, and links between them

  • component_types (Dict[str, Type[ComponentType]]) – Dictionary for the wiring diagram component types to Python component types

  • compatibility_checker (function of two types to a bool) – Each link uses the compatability_checker to ensure the link types are compatible.

Returns:

Configuration which can be used to run the cosimulation

Return type:

RunnerConfig

Raises:

Can raise any exception from component type initialization –

oedisi.componentframework.system_configuration.initialize_federates(wiring_diagram: WiringDiagram, component_types: dict[str, type[ComponentType]], compatability_checker, target_directory='.') → list[Federate]

Initialize all the federates.

Extracts config and sends it to each Component in an initalization step, then finds all dynamic inputs and outputs and sends input mappings.

Parameters:

  • wiring_diagram

  • component_types (dictionary of component type names to ComponentType class)

  • compatibility_checker (function from types to bool) – Check if source type is compatible with target_type

  • target_directory (str | Path = ".") – Directory where all components should be initialized.

Return type:

List of Federate run configuration

Raises:

ComponentType classes may return errors on configuration. –

oedisi.componentframework.wiring_diagram_utils module

Wiring Diagram utilities.

Wiring diagrams can be hard to manage in their final list based form. Some utilities plot, and future additions include nested wiring diagram composition and a programmatic interface.

oedisi.componentframework.wiring_diagram_utils.get_graph(wiring_diagram: WiringDiagram)

Get networkx graph representation of wiring_diagram.

oedisi.componentframework.wiring_diagram_utils.plot_graph_bokeh(wiring_diagram: WiringDiagram)

Plot wiring diagram using bokeh for interactive visualization.

oedisi.componentframework.wiring_diagram_utils.plot_graph_matplotlib(wiring_diagram: WiringDiagram)

Plot wiring diagram using matplotlib with spring layout.

Module contents