System

System#

Connection#

Connection definitions for co-simulation components.

This module defines data classes for representing connections between components in a co-simulation environment.

Classes:

  • Connection: Standard data connection between two components.

  • EventConnection: Event-based connection for event handling.

class syssimx.system.connection.Connection[source]

Bases: object

Defines a connection between two components in a co-simulation environment.

src_comp

Name of the source component.

Type:

str

src_port

Name of the output port on the source component.

Type:

str

dst_comp

Name of the destination component.

Type:

str

dst_port

Name of the input port on the destination component.

Type:

str

src_comp: str
src_port: str
dst_comp: str
dst_port: str
key()[source]

Unique key for duplicate detection.

Return type:

tuple[str, str, str, str]

__init__(src_comp, src_port, dst_comp, dst_port)
Parameters:

  • src_comp (str)

  • src_port (str)

  • dst_comp (str)

  • dst_port (str)

Return type:

None

class syssimx.system.connection.EventConnection[source]

Bases: object

Defines an event connection between a source (event producer) and target (event listener).

When added to a System via add_event_connection(): - The system automatically subscribes the target to the event - No manual Event object creation or subscribe_event() call required

src_comp

Name of the component that produces the event (has event indicator)

Type:

str

src_port

Name of the event (must match event indicator name on source)

Type:

str

dst_comp

Name of the component that handles the event

Type:

str

dst_port

Name of the event input port on the target (for visualization)

Type:

str

src_comp: str
src_port: str
dst_comp: str
dst_port: str
property event_name: str

Name of the event (same as src_port).

property target_comp: str

Name of the target component.

key()[source]

Unique key for duplicate detection.

Return type:

tuple[str, str, str, str]

__init__(src_comp, src_port, dst_comp, dst_port)
Parameters:

  • src_comp (str)

  • src_port (str)

  • dst_comp (str)

  • dst_port (str)

Return type:

None

System Class#

Co-simulation system orchestration and management.

This module provides the System class, which is the central orchestrator for heterogeneous co-simulation. It manages components, connections, execution order computation, and delegates simulation stepping to pluggable algorithms.

Key Responsibilities:

  • Component Management: Register and organize CoSimComponent instances

  • Connection Validation: Type and unit compatibility checking between ports

  • Graph Analysis: Build dependency graphs and detect algebraic loops

  • Execution Ordering: Compute parallelizable generations using topological sort

  • Algorithm Delegation: Step simulation using Jacobi, Gauss-Seidel, or Hybrid algorithms

  • Event Routing: Dispatch events between components via event connections

  • History Aggregation: Collect time-series data from all components

Typical Usage:

Building and running a co-simulation:

from syssimx import System, Connection
from syssimx.components import FMUComponent

# Create system
system = System(name="ControlledPendulum")

# Add components
pendulum = FMUComponent("Pendulum", fmu_path="Pendulum.fmu")
controller = FMUComponent("PID", fmu_path="Controller.fmu")
system.add_component(pendulum)
system.add_component(controller)

# Define connections
system.add_connection(Connection(
    src_comp="Pendulum", src_port="angle",
    dst_comp="PID", dst_port="measurement"
))
system.add_connection(Connection(
    src_comp="PID", src_port="control",
    dst_comp="Pendulum", dst_port="torque"
))

# Initialize and run
system.initialize(t0=0.0)
system.run(t0=0.0, tf=10.0, dt=0.001)

# Retrieve results
history = system.get_history()

See also

Connection: Signal connections between component ports EventConnection: Event routing between components syssimx.system.algorithms: Stepping algorithm implementations syssimx.system.graph: Graph analysis utilities

class syssimx.system.system.System[source]#

Bases: object

Central orchestrator for heterogeneous co-simulation systems.

The System class manages a collection of interconnected CoSimComponent instances, validates their connections, computes execution order, and coordinates simulation stepping through a pluggable algorithm.

name#

Identifier for this system.

Type:

str

components#

Registered components by name.

Type:

dict[str, CoSimComponent]

connections#

Signal connections between components.

Type:

list[Connection]

event_connections#

Event routing connections.

Type:

list[EventConnection]

graph#

Complete connection graph (including delayed).

Type:

nx.MultiDiGraph

groups#

Components organized by group.

Type:

dict[str, list[CoSimComponent]]

execution_order#

Topologically sorted generations. Each generation contains components that can execute in parallel.

Type:

list[list[str]]

algebraic_loops#

Detected algebraic loops (SCCs > 1).

Type:

list[list[str]]

algorithm#

Stepping algorithm (Gauss-Seidel, Jacobi, Hybrid).

Type:

Algorithm

history#

Aggregated time-series data from all components.

Type:

SystemHistory

is_initialized#

Whether initialize() has been called.

Type:

bool

t#

Current simulation time.

Type:

float

Example

Creating a simple two-component system:

>>>     system = System("Feedback")
>>>     system.add_component(plant)
>>>     system.add_component(controller)
>>>     system.add_connection(Connection(
>>>         src_comp="plant", src_port="y",
>>>         dst_comp="controller", dst_port="measurement"
>>>     ))
>>>     system.initialize(t0=0.0)
>>>     system.run(t0=0.0, tf=10.0, dt=0.01)

See also

Connection: Signal connection specification Algorithm: Base class for stepping algorithms

__init__(name)[source]#

Initialize a new co-simulation system.

Parameters:

name (str) – Identifier for this system. Used in logging and history tracking.

Example

>>> system = System("ControlledPendulum")
>>> system.name
'ControlledPendulum'
set_algorithm(algorithm)[source]#

Set the co-simulation stepping algorithm.

The algorithm determines how components are stepped during simulation. Available algorithms include:

  • GaussSeidelAlgorithm: Sequential stepping (default)

  • JacobiAlgorithm: Parallel stepping with delayed inputs

  • HybridAlgorithm: Event-driven with bisection localization

Parameters:

algorithm (Algorithm) – An instance implementing the Algorithm interface.

Raises:

TypeError – If algorithm doesn’t implement Algorithm.

Return type:

None

Example

>>> from syssimx.system.algorithms import JacobiAlgorithm
>>> system.set_algorithm(JacobiAlgorithm())

Note

If components with event indicators are detected during initialize(), the algorithm is automatically upgraded to HybridAlgorithm.

add_component(component)[source]#

Register a component with the system.

Components must be added before connections can reference them. Each component’s history is automatically registered with the system’s history tracker.

Parameters:

component (CoSimComponent) – A CoSimComponent instance to add. Its name attribute must be unique within this system.

Raises:

  • RuntimeError – If called after initialize().

  • ValueError – If a component with the same name already exists.

Example

>>> pendulum = FMUComponent("Pendulum", fmu_path="model.fmu")
>>> system.add_component(pendulum)
>>> "Pendulum" in system.components
True

Note

If the component has a group attribute set, it will be added to system.groups[group] for visual organization.

_validate_connection(c)[source]#
Parameters:

c (Connection)

Return type:

None

add_connection(connection)[source]#

Add a signal connection between two components.

Validates port existence, type compatibility, and unit compatibility before registering the connection.

Parameters:

connection (Connection) – A Connection specifying source and destination component/port pairs.

Raises:

  • ValueError – If source or destination component not in system, or if connection is a duplicate.

  • KeyError – If specified ports don’t exist on the components.

  • TypeError – If port types or units are incompatible.

Return type:

None

Example

>>> system.add_connection(Connection(
...     src_comp="Sensor", src_port="value",
...     dst_comp="Controller", dst_port="measurement"
... ))

See also

Connection: Connection specification class add_event_connection(): For event-based connections

_validate_event_connection(connection)[source]#

Validate an event connection before registration.

Performs the following checks:

  1. Source component exists in the system

  2. Target component exists in the system

  3. Event indicator is registered on the source component

  4. No duplicate connections exist

Parameters:

connection (EventConnection) – The EventConnection to validate.

Raises:

  • ValueError – If source or target component not in system, or if connection is a duplicate.

  • KeyError – If the event indicator is not registered on the source component.

Return type:

None

Note

Event subscription on the target is handled automatically by add_event_connection(), not during validation.

add_event_connection(connection)[source]#

Add an event connection with automatic target subscription.

Registers an event connection and automatically subscribes the target component to receive the event. This eliminates the need for manual Event object creation and subscribe_event() calls.

Parameters:

connection (EventConnection) – EventConnection specifying the source component’s event indicator and the target component to notify.

Raises:

  • ValueError – If source/target not in system or duplicate connection.

  • KeyError – If event indicator not registered on source component.

Return type:

None

Example

>>> # Ball emits 'bounce' event, floor handles it
>>> system.add_event_connection(EventConnection(
...     src_comp="Ball", event_name="bounce",
...     dst_comp="Floor"
... ))

Note

The target component must implement _handle_events_internal() to respond to the event when it fires.

See also

EventConnection: Event connection specification dispatch_event(): Manual event dispatching

get_event_targets(source_comp, event_name)[source]#

Get component names that receive a specific event.

Parameters:

  • source_comp (str) – Name of the component emitting the event.

  • event_name (str) – Name of the event indicator.

Returns:

List of component names subscribed to this event. Returns empty list if no subscribers.

Return type:

list[str]

dispatch_event(event, t, notify=True)[source]#

Dispatch an event to all subscribed components.

Resolves event listeners based on registered event connections and optionally notifies them by calling their handle_event() method.

Parameters:

  • event (Event) – The Event object containing source and event name.

  • t (float) – Time at which the event occurred.

  • notify (bool) – If True, calls handle_event() on each target. If False, only returns the target list without notifying.

Returns:

List of component names that are subscribed to this event.

Raises:

ValueError – If the event source is not in the system.

Return type:

list[str]

Example

>>> event = Event(name="threshold", source="Sensor")
>>> targets = system.dispatch_event(event, t=1.5)
>>> print(targets)
['Controller', 'Logger']
build_graphs()[source]#

Build dependency graphs from registered connections.

Constructs two graph representations:

  1. self.graph: Complete MultiDiGraph with all connections, including delayed connections (for visualization)

  2. self._dag: DiGraph with only zero-delay, direct-feedthrough connections (for execution ordering)

Also detects algebraic loops as strongly connected components (SCCs) with more than one node on the direct-feedthrough graph.

Note

Called automatically during initialize(). The detected algebraic loops are stored in self.algebraic_loops.

See also

syssimx.system.graph: Graph construction utilities

Return type:

None

compute_execution_order()[source]#

Compute parallelizable execution order from the dependency graph.

Performs a topological sort on the zero-delay dependency DAG to produce generations of components. Components within the same generation have no dependencies on each other and can execute in parallel.

The result is stored in self.execution_order as a list of lists, where each inner list contains component names in one generation.

Example

After calling:

system.execution_order = [
    ['Source', 'Reference'],      # Generation 0
    ['Controller'],               # Generation 1
    ['Plant'],                    # Generation 2
    ['Sensor']                    # Generation 3
]

See also

syssimx.system.graph: Execution order computation

Return type:

None

classify_components()[source]#

Classify components by their hybrid simulation capabilities.

Categorizes all components based on whether they have event indicators, event subscriptions, or neither. This classification is used to determine if hybrid simulation is needed.

Returns:

  • "event_sources": Components with event indicators (require rollback support for bisection)

  • "event_listeners": Components subscribed to events (will receive event notifications)

  • "continuous_only": Components without any hybrid capabilities (pure continuous dynamics)

Return type:

Dictionary with the following keys

Note

A component can be both an event source and an event listener. The categorization is stored in instance attributes self.event_sources, self.event_listeners, and self.continuous_only.

initialize(t0)[source]#

Initialize the system and all components at start time.

Performs the complete initialization sequence:

  1. Classifies components and auto-selects HybridAlgorithm if event sources are detected

  2. Creates port state objects for all components so that feedthrough detection and input propagation can work

  3. Detects direct feedthrough for pure-Python components via perturbation (FMU components already have this from their model description)

  4. Builds dependency graphs and computes execution order; algebraic loops are identified from zero-delay edges

  5. Iterates over generations in execution order:

    1. Propagates initial input values from upstream outputs into the generation’s input port states

    2. Initializes components (FMUs apply stored port values via _apply_input_starts during init mode)

    3. Solves algebraic loops within the generation

    4. Performs a zero-step (dt=0) to establish consistent initial outputs for downstream generations

This ordering ensures that each generation receives consistent initial values from already-initialized upstream components before entering FMU initialization mode.

Parameters:

t0 (float) – Initial simulation time in seconds.

Return type:

None

Example

>>> system.add_component(plant)
>>> system.add_component(controller)
>>> system.add_connection(connection)
>>> system.initialize(t0=0.0)
>>> system.is_initialized
True

Note

Must be called after all components and connections are added, but before run(). Calling initialize() locks the system against further component additions.

_set_inputs_for_generation(gen, t)[source]#

Set input values for all components in a generation.

For each component in the generation, retrieves values from connected source ports and sets them as inputs. This propagates signal values through the connection graph.

If a component is not yet initialized (e.g. during the generation-based initialization sequence), values are written directly to the PortState objects so that they are available when the component’s initialize() is called later. For already-initialized components the regular set_inputs() path is used, which also pushes values into the underlying solver (e.g. an FMU instance).

Parameters:

  • gen (list[str]) – List of component names in the current generation.

  • t (float) – Current simulation time for timestamping inputs.

Return type:

None

Note

Only non-None source values are propagated. Components with no incoming connections or all-None sources receive no updates.

run(t0, tf, dt)[source]#

Run the simulation from start time to end time.

Advances the simulation by repeatedly calling the algorithm’s step() method with the specified time step size.

Parameters:

  • t0 (float) – Start time in seconds (should match initialize(t0)).

  • tf (float) – End time in seconds.

  • dt (float) – Fixed time step size in seconds. The final step may be smaller to land exactly on tf.

Example

>>> system.initialize(t0=0.0)
>>> system.run(t0=0.0, tf=10.0, dt=0.001)
>>> # Simulation complete, retrieve history
>>> history = system.get_history()

Note

The algorithm may take sub-steps during event localization (Hybrid algorithm) or iteration (IJCSA for algebraic loops). The dt parameter controls the macro step size.

get_history()[source]#

Retrieve time-series history from all components.

Collects the recorded output history from every component in the system, plus any event history records.

Returns:

  • Keys are component names, values are tuples of (time_array, values_dict) from get_history_arrays()

  • Special key "Events" contains event occurrence records

Return type:

Dictionary with the following structure

Example

>>> history = system.get_history()
>>> t, values = history["Pendulum"]
>>> plt.plot(t, values["angle"])
>>> # Access events
>>> events = history["Events"]

See also

CoSimComponent.get_history_arrays(): Component history format

reset()[source]#

Reset the system and all components to uninitialized state.

Clears all component states, histories, and internal flags, allowing the system to be re-initialized from scratch.

Example

>>> system.initialize(t0=0.0)
>>> system.run(t0=0.0, tf=10.0, dt=0.01)
>>> system.reset()
>>> system.is_initialized
False
Return type:

None