CoFmuPy

https://irt-saint-exupery.github.io/CoFmuPy/
Admin
CoFmuPy is a Python library designed for rapid prototyping of digital twins through the...
Tokens:64,014
Snippets:502
Trust Score:9.7
Update:1 month ago
Show doc for...
Tokens:
Raw
### Instantiate Coordinator and load configuration

Source: https://irt-saint-exupery.github.io/CoFmuPy/getting_started

Creates a Coordinator instance and loads the simulation setup from example1/config.json. This is the first step to start a simulation.

```python
# Import the main CoFmuPy component
from cofmupy import Coordinator

# Create an instance of the Coordinator
coordinator = Coordinator()

# Load the simulation setup from a JSON configuration file
coordinator.start("example1/config.json")
```

--------------------------------

### start(conf_path, fixed_point_init=False, fixed_point_kwargs=None)

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Starts the coordinator with the given configuration file. It parses the configuration, starts the graph engine and master, and loads stream handlers and data storages.

```APIDOC
## start(conf_path, fixed_point_init=False, fixed_point_kwargs=None)

### Description
Start the coordinator with the given configuration file.

### Method
start

### Parameters
- **conf_path** (str) - Required - Path to the configuration file.
- **fixed_point_init** (bool) - Optional - Whether to use the fixed-point initialization method. Default: False.
- **fixed_point_kwargs** (dict) - Optional - Keyword arguments for the fixed point initialization method if fixed_point is set to True. Defaults to None, in which case the default values are used "solver": "fsolve", "time_step": minimum_default_step_size, and "xtol": 1e-5.

### Request Example
No request body required.

### Response
No return value documented.
```

--------------------------------

### Output of starting the Coordinator

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/plot_00_get_started

Shows the console output when starting the Coordinator with the example configuration. It includes warnings about ignored properties ('supplier', 'unit') and a message 'target is not a list' and 'Skipping Fixed Point Initialization'.

```text
Unknown property is ignore : supplier
Unknown property is ignore : supplier
Unknown property is ignore : unit
target is not a list
Skipping Fixed Point Initialization

```

--------------------------------

### Instantiate and start the Coordinator

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Initialize the Coordinator and start the simulation components using a JSON configuration file.

```python
from cofmupy import Coordinator

# Instantiate the Coordinator
my_coordinator = Coordinator()

# Start the Coordinator (and all its components) from a JSON configuration file
my_coordinator.start(conf_path="my_config_file.json")
```

--------------------------------

### Start the simulation system with coordinator.start

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/plot_01_fmu_proxy

Starts the coordinator, loading FMUs and proxies and preparing the simulation. The output shows warnings about unknown properties and a message that the target is not a list, followed by a resistor section and a skipped fixed point initialization.

```python
coordinator.start(config_path)

```

```text
Unknown property is ignore : supplier
Unknown property is ignore : supplier
Unknown property is ignore : unit
target is not a list
============ Resistor =============
Skipping Fixed Point Initialization

```

--------------------------------

### start(conf_path)

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Starts the Coordinator and all its components from a JSON configuration file. This method initializes the master, stream handlers, and storage handler based on the provided configuration.

```APIDOC
## start(conf_path)

### Description
Starts the Coordinator and all its components from a JSON configuration file. This method initializes the master, stream handlers, and storage handler based on the provided configuration.

### Method
start

### Signature
`start(conf_path: str)`

### Parameters
- **conf_path** (str) - Required - Path to the JSON configuration file.

### Example
```python
my_coordinator.start(conf_path="my_config_file.json")
```

```

--------------------------------

### Start the Coordinator with configuration file

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/plot_00_get_started

Loads the configuration file via the Coordinator's start method. This initializes the Master, data stream handlers, and graph engine. The output shows warnings about ignored properties and a note about skipping fixed point initialization.

```python
coordinator.start(config_path)

```

--------------------------------

### Log message when Kafka consumer thread starts

Source: https://irt-saint-exupery.github.io/CoFmuPy/user_guide/advanced/kafka_thread_manager

Example log output shown when the consumer thread starts. The thread name is included in the message.

```text
Kafka consumer started consuming in thread '<thread_name>'
```

--------------------------------

### Start the Simulation System

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/plot_01_fmu_proxy.ipynb

Loads the FMUs and proxies defined in the configuration file and prepares the simulation environment.

```python
coordinator.start(config_path)
```

--------------------------------

### Configure and start CoFmuPy Coordinator

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/academic_examples/plot_00_bouncing_ball.ipynb

Creates a CoFmuPy configuration dictionary with a single FMU and no connections, starts the coordinator, and retrieves the default value of the coefficient of restitution 'e'.

```python
from cofmupy import Coordinator

# We define a simple CoFmuPy config dictionary with a single FMU and no connection
config = {
    "fmus": [
        {"id": "bouncing_ball", "path": "../../../resources/fmus/BouncingBall.fmu"}
    ],
    "connections": []
}

# We create a Coordinator object and start it with the config dictionary
coordinator = Coordinator()
coordinator.start(config)

# Let's check the default value of the coefficient of restitution `e` in the FMU.
# A variable in CoFmuPy is identified by a tuple (fmu_id, variable_name).
e = coordinator.get_variable(("bouncing_ball", "e"))
print("Default value of coefficient of restitution 'e' (from the FMU):", e)
```

--------------------------------

### Coordinator()

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Instantiates the Coordinator object, which is the main entry point for the CoFmuPy library. It initializes internal components but does not start them until the start method is called.

```APIDOC
## Coordinator()

### Description
Instantiates the Coordinator object, which is the main entry point for the CoFmuPy library. It initializes internal components but does not start them until the start method is called.

### Method
Constructor

### Signature
`Coordinator()`

### Parameters
None

### Example
```python
from cofmupy import Coordinator

my_coordinator = Coordinator()
```

```

--------------------------------

### Install CoFmuPy with pip

Source: https://irt-saint-exupery.github.io/CoFmuPy/getting_started

Install the CoFmuPy package using pip. Run this command in your terminal before using the framework.

```bash
pip install cofmupy
```

--------------------------------

### Install Dependencies

Source: https://irt-saint-exupery.github.io/CoFmuPy/contributing

Commands to upgrade pip, install project requirements, and install the package in editable mode.

```bash
pip install -U pip
pip install -r requirements.txt -r requirements_dev.txt
pip install -e .
```

--------------------------------

### Start coordinator and get variable value

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/academic_examples/plot_01_dual_mass_oscillator

Creates a Coordinator object, starts it with the configuration, and retrieves the value of the 'm1' variable from the 'mass1' FMU. The output shows 'Skipping Fixed Point Initialization' and the value [1.0].

```python
from cofmupy import Coordinator

# We create a Coordinator object and start it with the config dictionary
coordinator = Coordinator()
coordinator.start(config)

# Let's check the default value of the mass of Mass1 in the FMU.
# A variable in CoFmuPy is identified by a tuple (fmu_id, variable_name).
mass_val = coordinator.get_variable(("mass1", "m1"))
print("Value of mass1 'm1' (from the FMU):", mass_val)

```

```text
Skipping Fixed Point Initialization
Value of mass1 'm1' (from the FMU): [1.0]

```

--------------------------------

### Example JSON configuration file for cofmupy

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/plot_00_get_started

Shows the content of the JSON configuration file used in this example. It defines two FMUs (source and resistor), two connections (source output to resistor input, and CSV file for resistance), and simulation settings (cosim_method 'jacobi' and edge_sep ' -> ').

```json
{
  "fmus": [
    {
      "id": "source",
      "path": "example1/source.fmu",
      "supplier": "IRT Saint Exupéry"
    },
    {
      "id": "resistor",
      "path": "example1/resistor.fmu",
      "initialization": {
        "R": 0.5
      },
      "supplier": "IRT Saint Exupéry"
    }
  ],
  "connections": [
    {
      "source": {
        "id": "source",
        "variable": "V",
        "unit": "V"
      },
      "target": {
        "id": "resistor",
        "variable": "V",
        "unit": "V"
      }
    },
    {
      "source": {
        "type": "csv",
        "path": "example1/resistor.csv",
        "variable": "R",
        "unit": "Ohm",
        "interpolation": "previous"
      },
      "target": {
        "id": "resistor",
        "variable": "R",
        "unit": "Ohm"
      }
    }
  ],
  "cosim_method": "jacobi",
  "edge_sep": " -> "
}

```

--------------------------------

### start_master method

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Starts the master algorithm with a configuration dictionary. Supports optional fixed-point initialization with custom keyword arguments.

```python
def start_master(
    self, config: dict, fixed_point_init=False, fixed_point_kwargs=None
):
    """
    Start the master algorithm with the given configuration.

    Args:
        config (dict): configuration for the master algorithm containing the FMUs,
            connections, sequence order, and loop method.
        fixed_point_init (bool): whether to use the fixed-point initialization method.
        fixed_point_kwargs (dict): keyword arguments for the fixed point initialization
            method if fixed_point is set to True. Defaults to None, in which
            case the default values are used "solver": "fsolve",
            "time_step": minimum_default_step_size, and "xtol": 1e-5.
    """
    self.master = DefaultMaster(
        fmu_config_list=config["fmus"],
        connections=config["connections"],
        sequence_order=config["sequence_order"],
        cosim_method=config["cosim_method"],
        iterative=config["iterative"],
        fixed_point=fixed_point_init,
        fixed_point_kwargs=fixed_point_kwargs,
    )
    self.master.init_simulation(input_dict={})
```

--------------------------------

### Setup Virtual Environment on Windows

Source: https://irt-saint-exupery.github.io/CoFmuPy/contributing

Commands to create and activate a virtual environment using venv on Windows.

```bash
python3 -m venv env_cofmupy
env_cofmupy/Scripts/activate.bat
```

--------------------------------

### Initialize Git LFS

Source: https://irt-saint-exupery.github.io/CoFmuPy/contributing

Commands to install and checkout Git LFS files after cloning the repository.

```bash
git lfs install
git lfs checkout
```

--------------------------------

### Setup Virtual Environment on Linux

Source: https://irt-saint-exupery.github.io/CoFmuPy/contributing

Commands to create and activate a virtual environment using venv on Linux.

```bash
python3 -m venv env_cofmupy
source env_cofmupy/bin/activate
```

--------------------------------

### Download example resources for CoFmuPy simulation

Source: https://irt-saint-exupery.github.io/CoFmuPy/getting_started

Downloads and extracts the example1.zip file containing FMUs and configuration for the voltage source and resistor scenario. Uses urllib and zipfile, then removes the zip file.

```python
import os
import urllib.request
import zipfile

url = "https://share-is.pf.irt-saintexupery.com/s/39zaG9HkQWnePbi/download"
resources_path = "example1.zip"

urllib.request.urlretrieve(url, resources_path)
with zipfile.ZipFile(resources_path, "r") as zip_ref:
    zip_ref.extractall(".")
os.remove(resources_path)

print("Resources unzipped in example1 folder!")
```

--------------------------------

### FMU description example

Source: https://irt-saint-exupery.github.io/CoFmuPy/user_guide/configuration_file

Example of an FMU definition with id, name, path, stepsize, initialization, and supplier attributes. The initialization values can be overridden by connections.

```json
{
    "id": "load",
    "name": "variable load",
    "path": "path/to/load.fmu",
    "stepsize": "0.2 sec",
    "initialization": {"V_in": 800, "R": 2.5},
    "supplier": "My company"
}
```

--------------------------------

### start_master(config: dict, fixed_point_init=False, fixed_point_kwargs=None)

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Starts the master algorithm with the given configuration. The configuration must contain the FMUs, connections, sequence order, and loop method. Optional fixed-point initialization can be enabled.

```APIDOC
## start_master(config: dict, fixed_point_init=False, fixed_point_kwargs=None)

### Description
Start the master algorithm with the given configuration.

### Method
start_master

### Parameters
- **config** (dict) - Required - Configuration for the master algorithm containing the FMUs, connections, sequence order, and loop method.
- **fixed_point_init** (bool) - Optional - Whether to use the fixed-point initialization method. Default: False.
- **fixed_point_kwargs** (dict) - Optional - Keyword arguments for the fixed point initialization method if fixed_point is set to True. Defaults to None, in which case the default values are used "solver": "fsolve", "time_step": minimum_default_step_size, and "xtol": 1e-5.

### Source Code
```python
def start_master(
    self, config: dict, fixed_point_init=False, fixed_point_kwargs=None
):
    """
    Start the master algorithm with the given configuration.

    Args:
        config (dict): configuration for the master algorithm containing the FMUs,
            connections, sequence order, and loop method.
        fixed_point_init (bool): whether to use the fixed-point initialization method.
        fixed_point_kwargs (dict): keyword arguments for the fixed point initialization
            method if fixed_point is set to True. Defaults to None, in which
            case the default values are used "solver": "fsolve",
            "time_step": minimum_default_step_size, and "xtol": 1e-5.
    """
    self.master = DefaultMaster(
        fmu_config_list=config["fmus"],
        connections=config["connections"],
        sequence_order=config["sequence_order"],
        cosim_method=config["cosim_method"],
        iterative=config["iterative"],
        fixed_point=fixed_point_init,
        fixed_point_kwargs=fixed_point_kwargs,
    )
    self.master.init_simulation(input_dict={})
```
```

--------------------------------

### Configure and run CoFmuPy simulation

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/academic_examples/plot_00_bouncing_ball.ipynb

Sets up the FMU configuration with initial values for e and h, starts the coordinator, runs the simulation, and plots the results.

```python
config = {
    "fmus": [
        {
            "id": "bouncing_ball",
            "path": "../../../resources/fmus/BouncingBall.fmu",
            "initialization": {"e": 0.5, "h": 5},
        }
    ],
    "connections": []
}

coordinator = Coordinator()
coordinator.start(config)

coordinator.run_simulation(step_size, end_time)

results = coordinator.get_results()
plot_height(results)
```

--------------------------------

### Setup Virtual Environment with Conda

Source: https://irt-saint-exupery.github.io/CoFmuPy/contributing

Commands to create and activate a virtual environment using conda.

```bash
conda create -n env_cofmupy python=3.10
conda activate env_cofmupy
```

--------------------------------

### Download and unzip resources for the FMU and proxy example

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/plot_01_fmu_proxy

Downloads the example1.zip archive from the shared repository, extracts it to the current directory, and removes the zip file. Prints a confirmation message after successful extraction.

```python
import os
import urllib.request
import zipfile

url = "https://share-is.pf.irt-saintexupery.com/s/HSNSeteJPoJjyXx/download"
resources_path = "example1.zip"

urllib.request.urlretrieve(url, resources_path)
with zipfile.ZipFile(resources_path, "r") as zip_ref:
    zip_ref.extractall(".")
os.remove(resources_path)

print("Resources unzipped in example1 folder!")
```

```text
Resources unzipped in example1 folder!
```

--------------------------------

### Enable fixed-point initialization with fixed_point_init=True

Source: https://irt-saint-exupery.github.io/CoFmuPy/user_guide/cosimulation_initialization

Starts the simulation coordinator with fixed-point initialization enabled, ensuring a dynamically valid initial state. When disabled, user-provided or default initial values are used without verification, which may be faster but can lead to inconsistencies.

```python
coordinator.start(config_path, fixed_point_init=True)
```

--------------------------------

### Fmu2Handler.reset

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/wrappers

Resets the FMU to its initial state and sets up the experiment. Calls setupExperiment with start time 0.0, enters initialization mode, and exits initialization mode.

```APIDOC
## Fmu2Handler.reset

### Description
Resets the FMU to its initial state and sets up the experiment. Calls setupExperiment with start time 0.0, enters initialization mode, and exits initialization mode.

### Method
reset()

### Parameters
None

### Request Example
```python
handler.reset()
```
```

--------------------------------

### get_results method

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Retrieves the simulation results as a dictionary. Requires the coordinator to be initialized via start().

```python
def get_results(self) -> dict:
    """
    Get the results from the simulation.

    Returns:
        dict: dataframe with the results.
    """
    if self.master is None:
        raise RuntimeError("Coordinator not initialized. Call start() first.")
    return self.master.get_results()
```

--------------------------------

### start_graph_engine(config: dict)

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Starts the graph engine with the given configuration. The configuration must contain the FMUs, symbolic nodes, connections, and edge separation.

```APIDOC
## start_graph_engine(config: dict)

### Description
Start the graph engine with the given configuration.

### Method
start_graph_engine

### Parameters
- **config** (dict) - Required - Configuration for the graph engine containing the FMUs, connections, and edge separation.

### Source Code
```python
def start_graph_engine(self, config: dict):
    """
    Start the graph engine with the given configuration.

    Args:
        config (dict): configuration for the graph engine containing the FMUs,
            connections, and edge separation.
    """
    self.graph_engine = GraphEngine(
        config["fmus"],
        config["symbolic_nodes"],
        config["connections"],
        config["edge_sep"],
    )
```
```

--------------------------------

### Visualize the connection graph

Source: https://irt-saint-exupery.github.io/CoFmuPy/getting_started

Displays a graphical representation of the simulated system using the coordinator's graph engine. Useful for verifying the setup.

```python
# Display a graphical representation of the simulated system
graph = coordinator.graph_engine.plot_graph()
graph.show()
```

--------------------------------

### Customize fixed-point solver with fixed_point_kwargs

Source: https://irt-saint-exupery.github.io/CoFmuPy/user_guide/cosimulation_initialization

Passes a dictionary of solver options to customize the fixed-point solver. The example sets solver to 'fsolve', time_step to 5e-8, tolerance to 1e-5, and max_iterations to 100. These options are passed along with fixed_point_init=True.

```python
# Customize fixed point solver
fixed_point_kwargs = {
    "solver": "fsolve",
    "time_step": 5e-8,
    "tolerance": 1e-5,
    "max_iterations": 100,
}

coordinator.start(config_path, fixed_point_init=True, fixed_point_kwargs=fixed_point_kwargs)
```

--------------------------------

### Start the Kafka consumer thread

Source: https://irt-saint-exupery.github.io/CoFmuPy/user_guide/advanced/kafka_thread_manager

Starts the background daemon thread that runs the consuming loop. It polls messages from Kafka and calls the callback for each message. Logs a message when the thread starts.

```python
manager.start()
```

--------------------------------

### Kafka message format example

Source: https://irt-saint-exupery.github.io/CoFmuPy/user_guide/data_sources

Example of Kafka messages containing time and value pairs in dictionary format.

```json
{"t": 0, "R": 100},
{"t": 0.1, "R": 200}

```

--------------------------------

### Serve Documentation

Source: https://irt-saint-exupery.github.io/CoFmuPy/contributing

Command to generate and serve the documentation locally using MkDocs.

```bash
mkdocs serve
```

--------------------------------

### cofmupy-extract-fmu example with model.fmu

Source: https://irt-saint-exupery.github.io/CoFmuPy/user_guide/helpers

Example invocation of cofmupy-extract-fmu on an FMU file named model.fmu in the current directory. The command extracts and displays all FMU metadata in a well-structured table.

```bash
cofmupy-extract-fmu model.fmu

```

--------------------------------

### CSV data source file example

Source: https://irt-saint-exupery.github.io/CoFmuPy/user_guide/data_sources

Example CSV file with a header containing time column 't' and variable columns. Used as input for the CSV data handler.

```csv
t,R,V_out,Iload
0,100,5,0
0.1,100,5,0.8
0.5,100,10,0.8
0.7,200,5,2.5

```

--------------------------------

### Start Coordinator and read FMU variable

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/academic_examples/plot_01_dual_mass_oscillator.ipynb

Creates a Coordinator object, starts it with the configuration dictionary, and reads the default value of the mass variable 'm1' from the FMU identified as 'mass1'. The variable is accessed using a tuple (fmu_id, variable_name).

```python
from cofmupy import Coordinator

# We create a Coordinator object and start it with the config dictionary
coordinator = Coordinator()
coordinator.start(config)

# Let's check the default value of the mass of Mass1 in the FMU.
# A variable in CoFmuPy is identified by a tuple (fmu_id, variable_name).
mass_val = coordinator.get_variable(("mass1", "m1"))
print("Value of mass1 'm1' (from the FMU):", mass_val)
```

--------------------------------

### GraphEngine.__init__

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/graph_engine

Initializes the GraphEngine instance, sets up the FMU and connection lists, and constructs the graph.

```APIDOC
## GraphEngine.__init__

### Description
Initializes the GraphEngine instance, sets up the FMU and connection lists, and constructs the graph.

### Method
Constructor

### Signature
`__init__(self, fmu_list: List[Dict], symbolic_nodes: List[Dict], conn_list: List[Dict], edge_sep: str = " -> ") -> None`

### Parameters
- **fmu_list** (List[Dict]) - Required - List of FMU data (each FMU represented as a dictionary).
- **symbolic_nodes** (List[Dict]) - Required - List of symbolic nodes (each node represented as a dictionary).
- **conn_list** (List[Dict]) - Required - List of connections (each connection represented as a dictionary).
- **edge_sep** (str) - Optional - Separator used for edge labels. Default is " -> ".

### Returns
None
```

--------------------------------

### ConfigParser.__init__

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/config_parser

Initializes the ConfigParser with a file path or dictionary, loads the configuration, applies defaults, validates, and builds configurations for graph, master, storages, and handlers.

```APIDOC
## ConfigParser.__init__

### Description
Initializes the ConfigParser instance with the provided file path or dictionary. It loads the configuration, applies defaults, performs validation, and builds the configuration dictionaries for graph engine, master solver, data storages, and stream handlers.

### Method
Constructor

### Signature
`__init__(self, file_path: Union[str, Dict]) -> None`

### Parameters
- **file_path** (Union[str, Dict]) - Required - Path to the configuration file or a dictionary containing the configuration.

### Attributes Set
- `file_path` - The provided file path or dictionary.
- `_config_object` - The parsed configuration object.
- `graph_config` - Configuration for the graph engine (dict).
- `master_config` - Configuration for the master solver (dict).
- `data_storages` - Storage settings for external data (dict).
- `stream_handlers` - Handlers for external data streams (dict).
- `error_in_config` - Boolean indicating whether errors exist in the configuration.

### Behavior
1. Loads the configuration using `_load_config(file_check=True)`.
2. Applies defaults and performs validation via `_validate_configuration()`.
3. Builds storage, master, handlers, and graph configurations via `_build_storage_config()`, `_build_master_config()`, `_build_handlers_config()`, and `_build_graph_config()`.
```

--------------------------------

### get_variable(name)

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/master

Gets the value of the given tuple fmu/variable.

```APIDOC
## get_variable(name)

### Description
Get the value of the given tuple fmu/variable.

### Method
get_variable(name)

### Parameters
- **name** (tuple) - Required - variable name as (fmu_id, var_name).

### Returns
- **list** (list) - value of the variable, as a list.

### Source Code
```python
def get_variable(self, name: tuple[str, str]):
    """
    Get the value of the given tuple fmu/variable.

    Args:
        name (tuple): variable name as (fmu_id, var_name).

    Returns:
        list: value of the variable, as a list.
    """
    fmu_id, var_name = name
    return self.fmu_handlers[fmu_id].get_variable(var_name)
```
```

--------------------------------

### Retrieve simulation results

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Get the simulation results as a Pandas dataframe.

```python
results_df = my_coordinator.get_results()
```

--------------------------------

### FmuProxyHandler.__init__

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/wrappers

Initializes the FmuProxyHandler with a given proxy model file path. Loads the proxy instance and sets up simulation attributes.

```APIDOC
## FmuProxyHandler.__init__

### Description
Initialize the wrapper class with a given proxy model file path.

### Method
Constructor

### Endpoint
FmuProxyHandler(path)

### Parameters
#### Path Parameters
- **path** (str) - Required - The file path to the proxy model. This should be in the format .py or .py::.

### Attributes
- **_proxy** - The proxy instance loaded from the file.
- **fmu** - Alias for the proxy instance for compatibility with FmuXHandler.
- **_time** (float) - The current simulation time, initialized to 0.0.
- **description** (ProxyModelDescription) - A model description object compatible with FmuXHandler helpers, containing metadata about the proxy model.
- **default_step_size** (Optional[float]) - The default step size for the simulation, extracted from the proxy model description.
- **var_name2attr** (Dict[str, ProxyVarAttr]) - A mapping of variable names to their attributes for quick lookup.
- **output_var_names** - A list of output variable names extracted from the proxy model.

### Raises
- Any exceptions raised by `load_proxy_class_from_file` or other operations during initialization.

### Methods
- **reset()** - Resets the internal simulation time to 0.0.
- **get_variable(name: str) -> list** - Retrieves the value of a variable by name as a single-element list.
- **set_variables(input_dict: Dict[str, Any])** - Sets multiple variables based on the provided dictionary of variable names and values.
- **step(current_time: float, step_size: float, input_dict: Dict[str, Any]) -> Dict[str, Any]** - Advances the simulation by a given step size, applying inputs before stepping and returning the output variables.
- **get_state() -> Dict[str, Any]** - Retrieves the current state of the simulation, including time and variable values, in a JSON-serializable format.
- **set_state(state: Dict[str, Any])** - Restores the simulation state from a given dictionary.
```

--------------------------------

### get_variable method

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/wrappers

Gets the variable matching the given name, returning its value as a list.

```python
def get_variable(self, name: str) -> list:
    """
    Gets the variable matching the given name.

    Args:
        name (str): The name of the variable to get

    Returns:
        list: The value of the variable, as a list.
    """
    var = self.var_name2attr[name]
    return getattr(self.fmu, f"get{var.type}")([var.valueReference])
```

--------------------------------

### get_results() -> dict

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Get the results from the simulation. Returns a dictionary containing the results dataframe.

```APIDOC
## get_results() -> dict

### Description
Get the results from the simulation.

### Method
Function call

### Parameters
None

### Returns
- **dict** - dataframe with the results.

### Errors
- Raises RuntimeError if the coordinator is not initialized (call start() first).

### Example
```python
results = coordinator.get_results()
```
```

--------------------------------

### Read and print the JSON configuration file

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/plot_01_fmu_proxy

Opens the configuration file at example1/config_with_fmu_proxy.json and prints its content. The file defines the FMU source, the Python proxy resistor, and the connections between them.

```python
config_path = "example1/config_with_fmu_proxy.json"
with open(config_path, "r") as f:
    print(f.read())

```

--------------------------------

### start_graph_engine method

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Initializes the graph engine using the provided configuration dictionary containing FMUs, connections, and edge separation settings.

```python
def start_graph_engine(self, config: dict):
    """
    Start the graph engine with the given configuration.

    Args:
        config (dict): configuration for the graph engine containing the FMUs,
            connections, and edge separation.
    """
    self.graph_engine = GraphEngine(
        config["fmus"],
        config["symbolic_nodes"],
        config["connections"],
        config["edge_sep"],
    )
```

--------------------------------

### init_simulation(input_dict=None)

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/master

Initializes the simulation environment and FMUs. This method sets up necessary dictionaries and initializes FMUs with either a fixed point algorithm or values from the input dictionary.

```APIDOC
## init_simulation(input_dict=None)

### Description
Initializes the simulation environment and FMUs. This method sets up the necessary dictionaries for the simulation and initializes the FMUs with either a fixed point algorithm or values provided in the input dictionary.

### Method
Function call

### Parameters
- **input_dict** (dict) - Optional - A dictionary containing input values for the simulation. Defaults to None.

### Steps
1. Sets the current simulation time to 0.
2. If fixed_point is True, calls the _fixed_point_init() method.
3. Otherwise, sets the inputs using the provided input_dict and initializes each FMU with these values.

**Note**: The FMUs are reset after setting the initial values.
```

--------------------------------

### get_variable(name)

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/wrappers

Gets the value of the variable matching the given name. Returns the value as a list.

```APIDOC
## get_variable(name)

### Description
Gets the variable matching the given name.

### Method
get_variable(name)

### Parameters
- **name** (str) - Required - The name of the variable to get

### Returns
- **list** - The value of the variable, as a list.

### Example
```python
value = fmu.get_variable("variable_name")
```
```

--------------------------------

### Read JSON configuration file content

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/plot_00_get_started

Opens and prints the JSON configuration file 'example1/config_with_csv.json' to display its content. This is the first step before loading it with the Coordinator.

```python
config_path = "example1/config_with_csv.json"
with open(config_path, "r") as f:
    print(f.read())

```

--------------------------------

### get_variable_type method

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Retrieves the type of a specified variable. Requires the coordinator to be initialized via start() before use.

```python
def get_variable_type(self, name: tuple) -> str:
    """
    Get the type of the given variable.

    Args:
        name (tuple): variable name as (fmu_id, var_name).

    Returns:
        str: type of the variable.
    """
    if self.master is None:
        raise RuntimeError("Coordinator not initialized. Call start() first.")

    return self.master.get_variable_type(name)
```

--------------------------------

### Download and Prepare Simulation Resources

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/plot_01_fmu_proxy.ipynb

Downloads the required FMU and Python proxy files from a remote repository and extracts them locally.

```python
import os
import urllib.request
import zipfile

url = "https://share-is.pf.irt-saintexupery.com/s/HSNSeteJPoJjyXx/download"
resources_path = "example1.zip"

urllib.request.urlretrieve(url, resources_path)
with zipfile.ZipFile(resources_path, "r") as zip_ref:
    zip_ref.extractall(".")
os.remove(resources_path)

print("Resources unzipped in example1 folder!")
```

--------------------------------

### KafkaDataStreamHandler Configuration

Source: https://irt-saint-exupery.github.io/CoFmuPy/user_guide/advanced/stream_handler_module

Example configuration dictionary for the KafkaDataStreamHandler, defining the topic, bootstrap servers, and interpolation settings.

```json
{
    "type": "kafka",
    "config": {
        "topic": "digital-twin",
        "bootstrap_servers": "localhost:9092",
        "group_id": "my-consumer-group",
        "interpolation": "linear",
        "timeout": 2, # timeout for consuming messages
        "kafka_backend_conf":"path_to/backend_config.json"
    }
}
```

--------------------------------

### FmuHandlerFactory.__init__

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/wrappers

Initializes the FmuHandlerFactory with a given path and reads the model description. Returns the appropriate FMU handler based on the FMI version.

```APIDOC
## FmuHandlerFactory.__init__

### Description
Initializes the FmuHandlerFactory with the given path and reads the model description.

### Method
Constructor

### Endpoint
FmuHandlerFactory(path)

### Parameters
#### Path Parameters
- **path** (str) - Required - The file path to the FMU.

### Returns
- **FmuXHandler** - The appropriate FMU handler based on the FMI version.

### Raises
- **ValueError** - If the FMI version is not recognized.

### Source Code
```python
def __init__(self, path: str) -> "FmuXHandler":
    """
    Initializes the FmuHandlerFactory with the given path and reads the model
    description.

    Args:
        path (str): The file path to the FMU.

    Returns:
        FmuXHandler: The appropriate FMU handler based on the FMI version.

    Raises:
        ValueError: If the FMI version is not recognized.
    """
    self.path = path

    # case of .py file or .py::<class_name>
    if self.path.endswith(".py") or "::" in self.path:
        self.description = ProxyModelDescription()
    else:
        self.description = read_model_description(path)
```
```

--------------------------------

### Initialize FMU handlers, step size, and dictionaries

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/master

Initialization sequence for the master: loads FMUs, checks connections, computes the smallest default step size, sets sequence order, and initializes input/output/result dictionaries. If no step sizes are found, defaults to 1.0. If fixed_point is enabled without kwargs, sets default solver to 'fsolve' with xtol 1e-5.

```python
# Load FMUs into dict of FMU Handlers
self.fmu_handlers = self._load_fmus()

# Check if the names of the variables match between the connection dict and
# the FMUs
self._check_connections()

default_step_sizes = []
for fmu in self.fmu_handlers.values():
    default_step_sizes.append(fmu.default_step_size)

## find the smaller of all step sizes
# remove None from default_step_sizes
default_step_sizes = [x for x in default_step_sizes if x is not None]
if len(default_step_sizes) == 0:
    self.default_step_size = 1.0
else:
    self.default_step_size = np.min(default_step_sizes)

# Sequence order of execution as a List of FMU IDs. Extracted by config
# parser module
# Sequence order of execution as a List of FMU IDs. Extracted by config parser
self.sequence_order = sequence_order
if self.sequence_order is None:
    self.sequence_order = [d[self.__keys["id"]] for d in self.fmu_config_list]

# init current_time to None to check if init_simulation() has been called
self.current_time = None
# Init output and input dictionaries for FMUs to maintain state between steps
# Initialize arrays for inputs and outputs
self._input_dict = {
    fmu_id: np.zeros(len(fmu.get_input_names()))
    for fmu_id, fmu in self.fmu_handlers.items()
}

self._output_dict = {
    fmu_id: np.zeros(len(fmu.get_output_names()))
    for fmu_id, fmu in self.fmu_handlers.items()
}
# Results dictionary to store the output values for each step
self._results = defaultdict(list)

self.fixed_point = fixed_point
self.fixed_point_kwargs = fixed_point_kwargs

if fixed_point and fixed_point_kwargs is None:
    self.fixed_point_kwargs = {
        "solver": "fsolve",
        "time_step": self.default_step_size,
        "xtol": 1e-5,
    }
```

--------------------------------

### get_variable_names() -> list

Source: https://irt-saint-exupery.github.io/CoFmuPy/api/coordinator

Get the names of all variables in the system. Returns a list of variable names as (fmu_id, var_name) tuples.

```APIDOC
## get_variable_names() -> list

### Description
Get the names of all variables in the system.

### Method
Function call

### Parameters
None

### Returns
- **list** - list of variable names as (fmu_id, var_name) tuples.

### Errors
- Raises RuntimeError if the coordinator is not initialized (call start() first).

### Example
```python
names = coordinator.get_variable_names()
```
```

--------------------------------

### Example JSON configuration file content

Source: https://irt-saint-exupery.github.io/CoFmuPy/generated/gallery/plot_01_fmu_proxy

The printed content of the configuration file. It lists the FMU source and the Python proxy resistor, their initialization parameters, the connections (including a CSV-based connection for the resistor value), the co-simulation method (jacobi), and the edge separator.

```json
{
    "fmus": [
        {
            "id": "source",
            "path": "example1/source.fmu",
            "supplier": "IRT Saint Exupéry"
        },
        {
            "id": "resistor",
            "path": "example1/resistor.py::Resistor",
            "initialization": {
                "R": 0.5
            },
            "supplier": "IRT Saint Exupéry"
        }
    ],
    "connections": [
        {
            "source": {
                "id": "source",
                "variable": "V",
                "unit": "V"
            },
            "target": {
                "id": "resistor",
                "variable": "V",
                "unit": "V"
            }
        },
        {
            "source": {
                "type": "csv",
                "path": "example1/resistor.csv",
                "variable": "R",
                "unit": "Ohm",
                "interpolation": "previous"
            },
            "target": {
                "id": "resistor",
                "variable": "R",
                "unit": "Ohm"
            }
        }
    ],
    "cosim_method": "jacobi",
    "edge_sep": " -> "
}

```
Tokens:
Raw