simulation_model

Modules

simulation_model module

class SimulationModel(id, dir_conf=None)[source]

Bases: object

A simulation model for building digital twins.

This class manages component collections, connections between components, cycle removal for feedback control loops, and topological sorting to determine optimal execution order for simulation.

Parameters:
  • id (str) – Unique identifier for the model.

  • dir_conf (Optional[List[str]]) – List of directories to store model-related files.

Mathematical Formulation:

The simulation model preparation process involves two main steps: cycle removal and topological sorting to create an executable simulation sequence.

Component Dependency Graph:

The simulation model can be represented as a directed multigraph \(G = (V, E, \iota)\) comprising:

\[V = \{c_1, c_2, ..., c_n\}\]
\[E = \{e_1, e_2, e_3, ...\}\]
\[\iota: E \rightarrow V \times V\]
where:
  • \(V\) is the set of vertices (components)

  • \(E\) is the set of edge identifiers (connections between components)

  • \(\iota\) is the incidence function mapping edges to vertex pairs

  • Each edge \(e_i \in E\) with \(\iota(e_i) = (c_j, c_k)\) indicates that component \(c_j\) provides input to component \(c_k\)

  • Multiple edges can map to the same vertex pair (multigraph): \(\iota(e_i) = \iota(e_j) = (c_p, c_q)\)

Optimized Cycle Removal Process:

To find the execution order of the components (i.e. the topologically sorted order), we need to remove cycles from the dependency graph first. Such cycles can arise in the simulation model due to different reasons, e.g. modeling of feedback control loops or mutual dependencies between components (model a requires an input from model b and model b requires an input from model a).

The optimized cycle removal process uses a greedy algorithm that minimizes the total number of edges removed:

1. Cycle Detection: Identify the set of simple cycles \(\mathcal{C} = \{C_1, C_2, ..., C_m\}\) in the graph where we can write one simple cycle as a sequence of edges \(C = ((c_1, c_2), (c_2, c_3), ..., (c_k, c_1))\), i.e. the cycle starts and ends at the same component and can’t visit any other component more than once.

  1. Edge Participation Analysis: For each edge \(e \in E\), count its participation in cycles:

    \[p(e) = |\{C \in \mathcal{C} \; | \; e \in C\}|\]

    This gives the number of cycles that edge \(e\) participates in.

  2. Greedy Edge Selection: Select the edge that participates in the maximum number of cycles:

    \[e^* = \underset{e \in E}{\operatorname{argmax}} \; p(e)\]
  3. Iterative Removal: Remove the selected edge and repeat until no cycles remain:

    \[E_{k+1} = E_k \setminus \{e^*_k\}\]

    where \(e^*_k\) is the optimal edge selected at iteration \(k\).

The process terminates when \(G_{final} = (V, E_{final})\) is acyclic:

\[E_{acyclic} = E_{final}, \quad \mathcal{C}(G_{final}) = \emptyset\]

All removed edges become required initialization connections:

\[E_{init} = E \setminus E_{acyclic}\]

This means, for all \((c_i, c_j) \in E_{init}\), \(c_j\) must have initial values provided.

Topological Sorting Process:

After cycle removal, we need to find a topological ordering of the acyclic graph \(G_{acyclic} = (V, E_{acyclic})\). A topological ordering is a linear arrangement of vertices \(L\) such that for every directed edge \((c_i, c_j) \in E_{acyclic}\), component \(c_i\) appears before component \(c_j\) in the ordering. In practical terms, this means when executing component \(c_j\), all components \(c_i\) that provides inputs to \(c_j\) must have already been executed.

The goal is to determine an execution sequence:

\[L = (c_1, c_2, ..., c_n)\]

And a priority level for each component:

\[P = (p_1, p_2, ..., p_n)\]

where:

  • Each \(L_p\) contains components that can execute at priority level \(p\)

  • Components with the same priority level can execute in parallel (no dependencies between them)

All of the above prepares the model for simulation and is done when the load() method is called.

id

Unique identifier for the model.

Type:

str

components

Dictionary of all components in the model.

Type:

dict

_execution_order

Ordered list of component groups for execution.

Type:

list

_flat_execution_order

Flattened list of components in execution order.

Type:

list

_components_no_cycles

Copy of components with cycles removed.

Type:

dict

_required_initialization_connections

Connections that require initial values.

Type:

list

See also

twin4build.simulator.simulator.Simulator: Handles simulation execution using the prepared execution order

References

The methodology is based on: “An Ontology-based Innovative Energy Modeling Framework for Scalable and Adaptable Building Digital Twins” by Bjørnskov & Jradi. This class implements the optimized cycle removal and topological sorting procedures.

Examples

Basic model setup and preparation:

>>> model = SimulationModel(id="building_model")
>>> # Create components
>>> schedule = tb.ScheduleSystem(id="schedule")
>>> damper = tb.DamperTorchSystem(id="damper")
>>> # Add components to model
>>> model.add_component(schedule)
>>> model.add_component(damper)
>>> # Connect schedule output to damper input
>>> model.add_connection(schedule, damper, "scheduleValue", "damperPosition")
>>> # Apply optimized cycle removal and topological sorting during model loading
>>> model.load()
>>> # Model is now ready for simulation with Simulator class
>>> # Execution order and cycle-free structure are prepared with minimal edge removal
add_component(component, components=None)[source]

Add a component to the model.

Parameters:

component (core.System) – The component to add.

Raises:

AssertionError – If the component is not an instance of core.System.

Return type:

None

add_connection(sender_component, receiver_component, outputPort, inputPort, components=None)[source]

Add a connection between two components in the system.

Parameters:
  • sender_component (core.System) – The component sending the connection.

  • receiver_component (core.System) – The component receiving the connection.

  • outputPort (str) – Name of the sender property.

  • inputPort (str) – Name of the receiver property.

Raises:
  • AssertionError – If property names are invalid for the components.

  • AssertionError – If a connection already exists.

Return type:

None

cache(start_time=None, end_time=None, step_size=None, simulator=None)[source]

Cache data and create folder structure for time series data.

Parameters:
  • start_time (Optional[datetime.datetime]) – Start time for caching.

  • end_time (Optional[datetime.datetime]) – End time for caching.

  • step_size (Optional[int]) – Time step size for caching.

Return type:

None

check_for_for_missing_initial_values()[source]

Check for missing initial values in components.

Raises:

Exception – If any component is missing an initial value.

Return type:

None

count_components()[source]
Return type:

int

count_connections()[source]
Return type:

int

get_component_by_class(dict_, class_, filter=None)[source]

Get components of a specific class from a dictionary.

Parameters:
  • dict (Dict) – The dictionary to search.

  • class (Type) – The class to filter by.

  • filter (Optional[Callable]) – Additional filter function.

Returns:

List of components matching the class and filter.

Return type:

List

get_dir(folder_list=[], filename=None)[source]

Get the directory path for storing model-related files.

Parameters:
  • folder_list (List[str]) – List of folder names to create.

  • filename (Optional[str]) – Name of the file to create.

Returns:

The full path to the directory or file, and a boolean indicating if the file exists.

Return type:

Tuple[str, bool]

get_object_properties(object_)[source]

Get all properties of an object.

Parameters:

object (Any) – The object to get properties from.

Returns:

A dictionary of object properties.

Return type:

Dict

get_simple_cycles(components)[source]

Get the simple cycles in the system graph.

Parameters:

components (Dict) – Dictionary of components.

Returns:

List of simple cycles.

Return type:

List[List[core.System]]

get_simple_graph(components)[source]

Get a simple graph representation of the system graph. This is a simplified version of the system graph that drops information about edge labels (Connection and ConnectionPoint pairs).

Returns:

The simple graph representation.

Return type:

Dict

initialize(start_time, end_time, step_size, simulator)[source]

Initialize the model for simulation.

Parameters:
  • start_time (datetime.datetime) – Start time for the simulation.

  • end_time (datetime.datetime) – End time for the simulation.

  • step_size (int) – Time step size for the simulation.

  • simulator (core.Simulator) – Simulator instance.

Return type:

None

load(rdf_file=None, fcn=None, verbose=False, validate_model=True, force_config_overwrite=False)[source]

Load and set up the model for simulation.

Parameters:
  • rdf_file (Optional[str]) – Path to a serialized model.

  • fcn (Optional[Callable]) – Custom function to be applied during model loading.

  • verbose (bool) – Whether to print verbose output during loading.

  • validate_model (bool) – Whether to perform model validation.

  • force_config_overwrite (bool) – If True, all parameters are read from the config file. If False, only the parameters that are None are read from the config file. If you want to use the fcn function

  • parameters (to set the)

  • overwritten. (you should set force_config_overwrite to False to avoid it being)

Return type:

None

load_estimation_result(filename=None, result=None)[source]

Load a chain log from a file or dictionary.

Parameters:
  • filename (Optional[str]) – The filename to load the chain log from.

  • result (Optional[Dict]) – The chain log dictionary to load.

Raises:

AssertionError – If invalid arguments are provided.

Return type:

None

make_pickable()[source]

Make the model instance pickable by removing unpickable references.

This method prepares the Model instance for use with multiprocessing, e.g. in the Estimator class.

Return type:

None

remove_component(component, components=None)[source]

Remove a component from the model.

Parameters:

component (core.System) – The component to remove.

Return type:

None

remove_connection(sender_component, receiver_component, outputPort, inputPort, components=None)[source]

Remove a connection between two components in the system.

Parameters:
  • sender_component (core.System) – The component sending the connection.

  • receiver_component (core.System) – The component receiving the connection.

  • sender_property_name (str) – Name of the sender property.

  • receiver_property_name (str) – Name of the receiver property.

Raises:

ValueError – If the specified connection does not exist.

Return type:

None

reset()[source]

Reset the model to its initial state.

Return type:

None

reset_torch_tensors()[source]

Reset all torch.Tensor objects in the model to remove TensorWrapper references.

This method iterates through all components and their attributes to find torch.Tensor objects that might contain TensorWrapper (which causes pickling issues). It creates new tensors with the same values but without gradient tracking.

This is particularly useful when switching from AD (automatic differentiation) to FD (finite difference) methods in the Estimator, as AD methods create gradient-tracking tensors that cannot be pickled for multiprocessing.

Return type:

None

restore_parameters(keep_values=True)[source]
Return type:

None

serialize()[source]

Serialize the simulation model.

set_custom_initial_dict(_custom_initial_dict)[source]

Set custom initial values for components.

Parameters:

_custom_initial_dict (Dict[str, Dict[str, Any]]) – Dictionary of custom initial values.

Raises:

AssertionError – If unknown component IDs are provided.

Return type:

None

set_initial_values(dict_)[source]

Set initial values for all components in the model.

Return type:

None

set_parameters_from_array(values, components, parameter_names, normalized=None, overwrite=False, save_original=False)[source]

Set parameters for components from an array.

Parameters:
  • values (List[Any]) – List of parameter values.

  • component_list (List[core.System]) – List of components to set parameters for.

  • attr_list (List[str]) – List of attribute names corresponding to the parameters.

Raises:

AssertionError – If a component doesn’t have the specified attribute.

Return type:

None

set_parameters_from_config(d, component)[source]

Recursively set parameters from a dictionary.

set_save_simulation_result(flag=True, c=None)[source]
validate()[source]

Validate the model by checking IDs and connections.

Return type:

None

validate_components()[source]

Validate the parameters of all components in the model.

Raises:

AssertionError – If any component has invalid parameters.

Return type:

None

validate_connections()[source]

Validate the connections between components in the model.

Raises:

AssertionError – If any required connections are missing.

Return type:

None

validate_ids()[source]

Validate the IDs of all components in the model.

Raises:

AssertionError – If any component has an invalid ID.

Return type:

None

visualize(query=None, literals=True)[source]

Visualize the simulation model.

Return type:

None

property components: dict
property dir_conf: List[str]
property execution_order: List[str]
property flat_execution_order: List[str]