### 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 '' ``` -------------------------------- ### 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) ``` -------------------------------- ### 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 . ``` -------------------------------- ### 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 ``` -------------------------------- ### 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:: 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": " -> " } ```