Co-Simulation Components

Contents

Co-Simulation Components#

FMU Component#

FMI 2.0 Co-Simulation FMU wrapper.

This module exposes FMUComponent, a CoSimComponent implementation backed by an FMI 2.0 co-simulation FMU via fmpy. The component derives input/output ports from the FMU model description, handles units for REAL signals, applies parameter starts during initialization, and supports direct feedthrough evaluation.

syssimx.components.fmu._require_fmpy()[source]#
Return type:

None

class syssimx.components.fmu.FMUComponent[source]#

Bases: CoSimComponent

FMU co-simulation wrapper implementing the CoSimComponent interface.

Supports FMI 2.0 co-simulation FMUs, typed input/output ports (REAL, INT, BOOL, STRING), and unit-aware REAL ports via pint.

state: dict[str, Any]#
_vrs_in_bytes: dict[str, int]#
_vrs_out_bytes: dict[str, int]#
__init__(name, fmu_path, group=None)[source]#

Create the FMU component and derive port specifications.

Parameters:
  • name (str) – Component name.

  • fmu_path (str) – Path to the FMU file.

  • group (str | None) – Optional component group for organization.

Raises:
_md: ModelDescription#
_md_vars: dict[str, ScalarVariable]#
_instance: FMU2Slave | None#
_unzipdir: str | None#
direct_feedthrough: dict[str, set[str]]#
_vrs_in_real: dict[str, int]#
_vrs_in_int: dict[str, int]#
_vrs_in_bool: dict[str, int]#
_vrs_in_str: dict[str, int]#
_vrs_out_real: dict[str, int]#
_vrs_out_int: dict[str, int]#
_vrs_out_bool: dict[str, int]#
_vrs_out_str: dict[str, int]#
get_default_parameters()[source]#

Return default parameter values from the FMU model description.

Returns:

Dict of parameter names to default values.

Return type:

dict[str, Any]

_build_port_specs()[source]#

Build input/output PortSpec objects from the model description.

Return type:

None

_build_value_reference_map()[source]#

Cache value references for fast typed access.

Return type:

None

_analyze_model_structure()[source]#

Populate model structure dependencies for outputs/derivatives/initials.

Return type:

None

_detect_direct_feedthrough()[source]#

Detect direct feedthrough dependencies for FMU outputs.

Return type:

dict[str, set[str]]

_require_instance()[source]#

Return the FMU instance or raise if not initialized.

Return type:

None

_initialize_component(t0)[source]#

Initialize and instantiate the FMU for simulation.

  1. Extracts the FMU archive.

  2. Creates the FMU2Slave instance.

  3. Sets up the experiment with the start time.

  4. Enters initialization mode.

  5. Applies parameter start values.

  6. Applies input start values from PortStates.

  7. Exits initialization mode.

Parameters:

t0 (float) – Simulation start time.

Return type:

None

reinitialize_instance(t0)[source]#

Recreate the FMU instance and enter initialized state at t0.

This is intentionally different from FMI reset(). Some FMUs do not support reliable rollback through fmi2Reset after stepping, but can be restored by creating a fresh instance and applying reconstructed initial conditions through parameters.

Parameters:

t0 (float)

Return type:

None

_apply_parameters_starts()[source]#

Apply parameter start values to the FMU instance.

Return type:

None

_apply_input_starts()[source]#

Push initial input values from PortStates into the FMU.

Return type:

None

set_parameters(**parameters)[source]#

Set parameter values with validation and unit handling.

Parameters:

**parameters – Parameter name/value pairs.

Raises:
  • KeyError – If a parameter name is unknown.

  • TypeError – If a value does not match the parameter type or unit.

Return type:

None

get_parameters(*names)[source]#

Return parameter values, optionally limited to specific names.

Parameters:

*names (str) – Optional parameter names to fetch.

Raises:

KeyError – If a requested parameter name is unknown.

Return type:

dict[str, Any]

_abc_impl = <_abc._abc_data object>#
set_inputs(signals, t=None)[source]#

Set input port values and push them to the FMU.

Parameters:
  • signals (dict[str, Any]) – Mapping of input names to values.

  • t (float | None) – Optional timestamp to record with inputs.

Raises:
  • KeyError – If an input name is unknown.

  • TypeError – If a value type does not match the port type.

  • ValueError – If a REAL input is missing a value after unit conversion.

Return type:

None

get_outputs()[source]#

Return current output values as a dict of name to value.

Return type:

dict[str, Any]

_update_output_states(t=None, event_names=None)[source]#

Refresh output PortStates from the FMU instance.

Parameters:
  • t (float | None) – Optional timestamp to record with outputs.

  • event_names (list[str] | None) – Reserved for event-based updates (unused).

Return type:

None

_do_step_internal(t, dt)[source]#

Perform a single time step in the FMU.

Parameters:
  • t (float) – Current communication point.

  • dt (float) – Communication step size.

Return type:

None

set_state(state, t)[source]#

Restore FMU state from a serialized variable dictionary.

Parameters:
  • state (dict[str, Any]) – Mapping of variable names to stored attributes/values.

  • t (float) – Time to reinitialize the FMU at.

Return type:

None

get_state()[source]#

Return FMU variables as a serialized dictionary.

The dictionary includes variables that are not fixed and not local, along with their units and current values pulled from the FMU instance.

Returns:

Dict mapping variable names to unit/value attributes.

Return type:

dict[str, dict[str, Any]]

evaluate_outputs(inputs, t=None)[source]#

Evaluate outputs for a given input set without advancing time.

Parameters:
  • inputs (dict[str, Any]) – Input values to apply.

  • t (float | None) – Optional time to use for evaluation.

Returns:

Mapping of output names to raw (unitless) values.

Return type:

dict[str, Any]

reset()[source]#

Reset the component and release the FMU instance.

After reset, the component can be reinitialized via initialize().

Return type:

None

soft_reset(t0=0.0)[source]#

Reset the FMU to initial state without releasing the instance.

Uses FMI reset() to return to the instantiated state, then reinitializes with parameter and input starts.

Parameters:

t0 (float) – New start time for the simulation.

Return type:

None

_validate_parameter(name, value)[source]#

Validate and normalize a parameter value.

Parameters:
  • name (str) – Parameter name.

  • value (Any) – Proposed parameter value.

Returns:

Validated value, converted to the expected type/unit.

Raises:

TypeError – If the value is incompatible with the parameter type or unit.

Return type:

Quantity | float | int | bool | str

syssimx.components.fmu._port_type_from_var(var)[source]#

Determine PortType from ScalarVariable type.

Parameters:

var (ScalarVariable) – The model variable to determine the port type for.

Returns:

The corresponding PortType.

Return type:

PortType

Raises:

NotImplementedError – If the variable type is unsupported.

syssimx.components.fmu._convert_start_value(start, port_type, unit)[source]#

Convert start value string to appropriate type based on PortType and unit.

Parameters:
  • start (str | None) – The start value as a string.

  • port_type (PortType) – The type of the port.

  • unit (str | None) – The unit of the port, if applicable.

Returns:

The converted start value in the appropriate type.

Return type:

Union[float, int, bool, str, None]

syssimx.components.fmu._in_vr_map_for_type(comp, pt)[source]#

Return the input value-reference map for a port type.

Parameters:
Return type:

dict[str, int]

FEM Component#

NGSolve transient structural-mechanics co-simulation wrapper.

FEMComponent is the concrete FEM backend for SysSimX. It wraps an NGSolve finite-element model that is advanced in time with a constant-average-acceleration (trapezoidal) Newmark scheme and exposes it through the CoSimComponent interface. The class owns the parts that are common to any transient structural NGSolve model:

  • the Newmark state grid functions (u, v, a) and their previous-step buffers (u_old, v_old, a_old) and the velocity/acceleration update,

  • snapshot / restore of that state for hybrid event localization and rollback,

  • multidim grid-function history recording (gated by _record_history),

  • reset of the Newmark state,

  • a micro-stepped do_step loop built from overridable hooks.

A concrete model (e.g. the controlled FEM pendulum) subclasses this and supplies the mesh, finite-element spaces, the variational form, and the physics-specific hooks. The base is deliberately backend-specific (NGSolve) and problem-specific (transient structural dynamics); a non-transient NGSolve model can override the step hooks to opt out of the Newmark update.

Requires the ngsolve optional dependency (pip install syssimx[fem]).

class syssimx.components.fem.FEMComponent[source]#

Bases: CoSimComponent

Concrete NGSolve transient structural-mechanics component.

Subclasses implement _initialize_component() (build mesh, spaces, forms, initial state) and _solve_step() (the nonlinear solve of one sub-step), and may override the optional step hooks below.

__init__(name, label=None, group=None)[source]#

Initialize a new co-simulation component.

Parameters:
  • name (str) – Unique identifier for this component. Must be unique within the containing System. Used for connection definitions and graph construction.

  • label (str | None) – Human-readable display name for visualization and logging. Defaults to name if not provided.

  • group (str | None) – Optional category for grouping related components (e.g., ‘sensors’, ‘actuators’, ‘controllers’). Used for filtering and organization in large systems.

Example

>>> comp = MyComponent("pid_1", label="PID Controller", group="controllers")
>>> comp.name
'pid_1'
>>> comp.label
'PID Controller'
_init_newmark_state(fes)[source]#

Allocate the six Newmark state grid functions over fes.

Parameters:

fes (Any)

Return type:

None

_shift_newmark_state()[source]#

Copy the current state into the previous-step buffers.

Return type:

None

_advance_newmark()[source]#

Update velocity and acceleration from the freshly solved displacement.

Constant-average-acceleration (trapezoidal) Newmark update:

v = 2/τ (u - u_old) - v_old a = 2/τ (v - v_old) - a_old

Return type:

None

_register_history_field(history_attr, source_vec_fn)[source]#

Register a history grid function to receive a frame each sub-step.

Parameters:
  • history_attr (str) – Name of the GridFunction(..., multidim=0) attribute on this component that accumulates frames. Resolved by name at record time so it survives reallocation in reset.

  • source_vec_fn (Callable[[], Any]) – Callable returning the source vector to append.

Return type:

None

_record_history_frame()[source]#

Append one frame to every registered history field.

Skipped when _record_history is False (e.g. during hybrid trial steps) so the history reflects accepted simulation time only.

Return type:

None

snapshot_state()[source]#

Capture the complete Newmark state (current and previous step).

Return type:

dict[str, Any]

restore_state(snapshot, t)[source]#

Restore the complete Newmark state from a snapshot.

Parameters:
Return type:

None

_extra_snapshot()[source]#

Subclass hook: extra fields to add to the snapshot (e.g. contact gap).

Return type:

dict[str, Any]

_restore_extra(snapshot)[source]#

Subclass hook: restore extra fields captured by _extra_snapshot.

Parameters:

snapshot (dict[str, Any])

Return type:

None

do_step(t, dt)[source]#

Advance from t to t + dt via internal micro-stepping.

Overrides the single-shot CoSimComponent.do_step: outputs and history are updated within each accepted sub-step so that recorded results and internal event hints are sub-step accurate.

Parameters:
Return type:

None

_do_step_internal(t, dt)[source]#

Subclass hook for time step computation.

Override this method to implement the component’s simulation logic. This is where state integration, solver calls, and internal computations happen.

Parameters:
  • t (float) – Current simulation time at the start of the step (seconds).

  • dt (float) – Time step size to advance (seconds).

Return type:

None

Example

Simple Euler integration:

def _do_step_internal(self, t: float, dt: float) -> None:
    u = self.inputs['u'].get()
    self._state += self._derivative(self._state, u) * dt

Note

  • Do NOT update self.t here; do_step() handles that

  • Do NOT call _update_output_states(); done by do_step()

  • For event detection, call report_internal_event() if micro-stepping detects a zero-crossing

See also

do_step(): Public interface that calls this method report_internal_event(): Report events from micro-stepping

_effective_substep(dt)[source]#

Return the nominal internal sub-step for a macro step dt.

Parameters:

dt (float)

Return type:

float

_pre_solve(t_current, effective_dt)[source]#

Hook before the nonlinear solve (e.g. contact update, adaptive τ).

Parameters:
Return type:

None

abstractmethod _solve_step()[source]#

Solve the nonlinear variational form for the current sub-step.

Return type:

None

_post_solve(t_current)[source]#

Hook after the Newmark update (e.g. event detection, stress fields).

Parameters:

t_current (float)

Return type:

None

_after_substep(t_current)[source]#

Hook after outputs are recorded (e.g. redraw, monitoring update).

Parameters:

t_current (float)

Return type:

None

reset()[source]#

Zero the Newmark state grid functions.

Return type:

None

load_mesh_from_file(path)[source]#

Optional: load a mesh from file (GMSH, VTK, XDMF, …).

Parameters:

path (Path)

Return type:

None

export_results(path, format='vtk')[source]#

Optional: export the time series to file.

Parameters:
Return type:

None

_abc_impl = <_abc._abc_data object>#

OpenSim Component#

class syssimx.components.opensim.OpenSimComponent[source]#

Bases: CoSimComponent

Wrapper around an OpenSim model to implement the CoSimComponent interface.

parameters: dict[str, Any] = {}#
__init__(name, osim_model_path, group=None)[source]#

Constructor for the OpenSimComponent class.

Parameters:
  • name (str)

  • osim_model_path (str)

  • group (str | None)

model: opensim.Model#
state: opensim.State#
manager: opensim.Manager#
_finalize_model(t0)[source]#

Complete OpenSim initialization after model is built/loaded. Call this at the end of subclass _initialize_component().

Parameters:

t0 (float)

Return type:

None

realize()[source]#

Realizes the stage: Position, Velocity, Dynamics, Acceleration

Return type:

None

reset()[source]#

Reset the component to a clean state before (before initialization).

Return type:

None

get_coordinate_value(coord_name)[source]#

Get current value of named coordinate (generalized position).

Parameters:

coord_name (str)

Return type:

float

get_coordinate_speed(coord_name)[source]#

Get current speed of named coordinate (generalized velocity).

Parameters:

coord_name (str)

Return type:

float

get_coordinate_acceleration(coord_name)[source]#

Get current acceleration of named coordinate.

Parameters:

coord_name (str)

Return type:

float

_abc_impl = <_abc._abc_data object>#
set_coordinate_value(coord_name, value)[source]#

Set value of named coordinate.

Parameters:
Return type:

None

set_coordinate_speed(coord_name, value)[source]#

Set speed of named coordinate.

Parameters:
Return type:

None

_find_coordinate(name)[source]#

Find coordinate by name in model.

Parameters:

name (str)

Return type:

opensim.Coordinate