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, progress=None)[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.

  • progress (Callable[[float, float], None] | None) – Optional callback invoked after every completed macro step with (t, tf), where t is the simulation time reached. Use it to drive progress bars or live plots, e.g. progress=lambda t, tf: bar.update(...).

Returns:

A SimulationResult with the recorded component histories, the event log, and run metadata (time span, macro step, wall time, algorithm).

Return type:

SimulationResult

Example

>>> system.initialize(t0=0.0)
>>> result = system.run(t0=0.0, tf=10.0, dt=0.001)
>>> df = result.to_dataframe(component="Pendulum")

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

describe()[source]#

Return a human-readable report of the system’s structure.

Summarizes the components, connections, execution order (generations), detected algebraic loops, direct-feedthrough relations, and the configured master algorithm. The structural metadata is computed during initialize(); calling describe() before initialization reports on whatever has been assembled so far.

Returns:

A multi-line report string, ready for print().

Return type:

str

Example

>>> system.initialize(t0=0.0)
>>> print(system.describe())
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

Simulation Results#

Simulation result container for SysSimX runs.

SimulationResult wraps the per-component output histories recorded during a run together with run metadata (time span, macro step, wall time, algorithm) and provides convenient post-processing accessors: pandas DataFrame conversion (tidy long format or per-component wide format), CSV export, and the recorded event log.

A result is returned by syssimx.system.system.System.run() and can also be built from an already-run system via SimulationResult.from_system().

class syssimx.system.results.SimulationResult[source]#

Bases: object

Container for the outputs and metadata of one simulation run.

system_name#

Name of the simulated system.

t0#

Start time of the run in seconds.

tf#

End time of the run in seconds.

dt#

Macro communication step size in seconds.

wall_time#

Wall-clock duration of the run in seconds.

algorithm#

Name of the master algorithm used.

histories#

Mapping component name -> (time_array, {port: values}) exactly as returned by CoSimComponent.get_history_arrays().

events#

Recorded event occurrences, as stored by the system history.

__init__(system_name, t0, tf, dt, wall_time, algorithm, histories, events=None)[source]#
Parameters:
classmethod from_system(system, t0, tf, dt, wall_time=nan)[source]#

Build a result from an already-run system.

Parameters:

  • system (System) – The system whose component histories to capture.

  • t0 (float) – Start time of the run.

  • tf (float) – End time of the run.

  • dt (float) – Macro step size used.

  • wall_time (float) – Optional wall-clock duration in seconds.

Return type:

SimulationResult

property component_names: list[str]#

Names of all components with recorded history.

to_dataframe(component=None)[source]#

Convert recorded histories to a pandas DataFrame.

Parameters:

component (str | None) – When given, return a wide DataFrame for that single component: one time column plus one column per output port. Without it, return a tidy long-format DataFrame with columns component, port, time, value covering every component (components may have different time grids, which the long format represents without padding).

Returns:

The requested DataFrame. Values that are pint quantities are reduced to their magnitudes.

Return type:

DataFrame

to_csv(path, component=None)[source]#

Write the result to a CSV file and return the written path.

Parameters:
Return type:

Path

syssimx.system.results._magnitudes(values)[source]#

Reduce an array (possibly of pint quantities) to plain magnitudes.

Parameters:

values (Any)

Return type:

ndarray

Declarative System Descriptions#

Declarative system descriptions: build a System from YAML/JSON.

A system description is a mapping with the following shape (YAML shown; JSON with the same structure is equally supported):

system:
  name: Quickstart
components:
  - name: Source
    class: "my_package.components:LinearSource"   # "module:ClassName"
    args: {a: 1.0, b: 0.0}                        # constructor kwargs
    parameters: {gain: 2.0}                       # set_parameters(...) after construction
connections:
  - src: Source.y            # "Component.port" (split at the first dot) ...
    dst: Integrator.u
  - src: {component: A, port: out}                # ... or explicit mapping form
    dst: {component: B, port: in}
event_connections:                                # optional
  - src: Plant.wall_hit
    dst: Controller.reset
algorithm:                                        # optional (default gauss_seidel)
  type: gauss_seidel        # jacobi | gauss_seidel | hybrid | ijcsa
run:                                              # optional defaults for run_from_config/CLI
  t0: 0.0
  tf: 5.0
  dt: 0.1

Component classes are resolved by import path, so any installed CoSimComponent subclass — including user-defined ones — can be used without registration. Use build_system() to assemble a System and run_from_config() to assemble, initialize, and run it in one call.

exception syssimx.system.loader.ConfigError[source]#

Bases: ValueError

Raised when a system description is invalid.

syssimx.system.loader.load_config(path)[source]#

Load a system description from a YAML or JSON file.

The format is chosen by file extension: .json is parsed as JSON, everything else (.yaml/.yml) as YAML.

Parameters:

path (str | Path) – Path to the description file.

Returns:

The parsed configuration mapping.

Raises:

ConfigError – If the file does not contain a mapping at top level.

Return type:

dict[str, Any]

syssimx.system.loader.build_system(config)[source]#

Assemble a System from a description.

Parameters:

config (dict[str, Any] | str | Path) – A parsed configuration mapping, or a path to a YAML/JSON file (which is loaded via load_config()).

Returns:

The assembled (not yet initialized) system.

Raises:

ConfigError – On any invalid or missing configuration entry.

Return type:

System

syssimx.system.loader.run_from_config(config, t0=None, tf=None, dt=None, progress=None)[source]#

Assemble, initialize, and run a system from a description.

Run settings are taken from the description’s run section; the t0/tf/dt arguments override individual entries.

Parameters:

  • config (dict[str, Any] | str | Path) – Configuration mapping or path to a YAML/JSON file.

  • t0 (float | None) – Override for the start time.

  • tf (float | None) – Override for the end time.

  • dt (float | None) – Override for the macro step size.

  • progress (Any) – Optional progress callback forwarded to System.run.

Returns:

The SimulationResult of the run.

Raises:

ConfigError – If any of t0/tf/dt is neither configured nor provided as an argument.

Return type:

SimulationResult

syssimx.system.loader._build_component(entry)[source]#

Instantiate one component from its description entry.

Parameters:

entry (dict[str, Any])

Return type:

CoSimComponent

syssimx.system.loader._resolve_class(spec)[source]#

Resolve a "module.path:ClassName" import spec to a class.

Parameters:

spec (str)

Return type:

type

syssimx.system.loader._parse_endpoint(entry, key)[source]#

Parse a connection endpoint: "Comp.port" or {component, port}.

Parameters:
Return type:

tuple[str, str]

syssimx.system.loader._require_component(system, comp_name, entry)[source]#

Raise a helpful error when a connection references an unknown component.

Parameters:
Return type:

None

Graph Analysis#

Graph construction and execution ordering helpers for System.

This module builds annotated connection graphs for a System, detects algebraic loops based on zero-delay direct-feedthrough dependencies, and computes a parallelizable execution order. It also provides utilities for identifying delayed producers and collecting zero-delay interface unknowns.

The helpers in this module mutate the following System fields:

  • graph: full connection graph with edge port annotations.

  • _dag: zero-delay direct-feedthrough dependency graph (directed acyclic graph).

  • algebraic_loops: list of strongly connected components in _dag.

  • _scc_index: component name to SCC id mapping for the _dag condensation.

  • _incoming_by_dst: incoming connection list keyed by destination component.

  • _input_sources: mapping of destination ports to their driving connections.

  • execution_order: list of component generations for execution.

  • execution_idx: component name to generation index mapping.

Typical usage:

build_graphs(system) compute_execution_order(system) interface_inputs, driver_map = collect_global_interface_unknowns(system)

syssimx.system.graph.collect_active_outputs(system)[source]#

Collect output ports that are used by outgoing connections.

Parameters:

system (System) – System whose connections are inspected.

Returns:

Mapping of component name to the set of output port names that are referenced by connections.

Return type:

dict[str, set[str]]

syssimx.system.graph.build_graphs(system)[source]#

Build annotated connection graphs and detect algebraic loops.

Populates full-connection and zero-delay dependency graphs, then computes algebraic loops (SCCs) on the zero-delay dependency graph.

Parameters:

system (System) – System to populate with graph structures.

Raises:

RuntimeError – If multiple connections drive the same input port.

Return type:

None

syssimx.system.graph.compute_execution_order(system)[source]#

Compute a parallelizable execution order from zero-delay dependencies.

Condenses the zero-delay graph into SCCs, orders those SCCs in topological generations, and then expands each generation back to component names.

Parameters:

system (System) – System to populate with execution ordering data.

Return type:

None

syssimx.system.graph.is_delayed_producer(system, name)[source]#

Check whether a component is a delayed producer.

A delayed producer has no zero-delay incident edges in the _dag, but still feeds into at least one component that participates in zero-delay structure. These are often actuator-like components whose outputs influence the closed loop only through downstream state.

Parameters:

  • system (System) – System containing the graph data.

  • name (str) – Component name to evaluate.

Returns:

True if the component is considered a delayed producer.

Return type:

bool

syssimx.system.graph.move_delayed_producers_to_last_generation(system)[source]#

Move delayed producers to a final execution generation.

Removes delayed producers from their current generations and appends them as a final, sorted generation in the execution order.

Parameters:

system (System) – System whose execution order is modified.

Return type:

None

syssimx.system.graph.collect_global_interface_unknowns(system)[source]#

Collect interface inputs that participate in zero-delay couplings.

Parameters:

system (System) – System containing the zero-delay dependency graph.

Returns:

  • interface_inputs: list of (dst_comp_name, dst_port_name).

  • driver_map: mapping (dst_comp, dst_port) -> (src_comp, src_port).

Return type:

A tuple containing