From 159ef933bd1958879b52e0f003c9a696cff528af Mon Sep 17 00:00:00 2001 From: Tesshub Date: Thu, 6 Nov 2025 16:47:19 +0100 Subject: [PATCH 1/2] =?UTF-8?q?=E2=9C=A8property=20defauts=20added=20in=20?= =?UTF-8?q?init?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- corrai/fmu.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/corrai/fmu.py b/corrai/fmu.py index 1c994a2..4d98b06 100644 --- a/corrai/fmu.py +++ b/corrai/fmu.py @@ -265,6 +265,7 @@ def __init__( simulation_dir: Path = None, output_list: list[str] = None, boundary_table_name: str | None = None, + default_properties: dict[str, float | int | str] = None, ): super().__init__(is_dynamic=True) fmu_path = Path(fmu_path) if isinstance(fmu_path, str) else fmu_path @@ -291,6 +292,7 @@ def __init__( UserWarning, stacklevel=2, ) + self.default_properties = default_properties or {} def get_property_values( self, property_list: str | tuple[str, ...] | list[str] @@ -432,7 +434,10 @@ def simulate( """ property_dict = dict(property_dict or {}) - + merged_properties = self.default_properties.copy() + if property_dict: + merged_properties.update(property_dict) + property_dict = merged_properties if property_dict and debug_param: print(property_dict) From 2d9415923becdd369dc9ec1f176d4f5850b63ec4 Mon Sep 17 00:00:00 2001 From: BaptisteDE Date: Fri, 14 Nov 2025 14:47:55 +0100 Subject: [PATCH 2/2] =?UTF-8?q?=E2=9C=A8=20get=20properties=20values=20et?= =?UTF-8?q?=20and=20set=20property=20values=20in=20ModelicaFmuModel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- corrai/fmu.py | 87 +++++++++++++++++++++++++++++++---------------- tests/test_fmu.py | 6 ++-- 2 files changed, 61 insertions(+), 32 deletions(-) diff --git a/corrai/fmu.py b/corrai/fmu.py index 4a926e6..adcd29e 100644 --- a/corrai/fmu.py +++ b/corrai/fmu.py @@ -2,16 +2,37 @@ import shutil import tempfile import warnings -from pathlib import Path from contextlib import contextmanager +from pathlib import Path import fmpy -from fmpy import simulate_fmu import pandas as pd +from fmpy import simulate_fmu from sklearn.pipeline import Pipeline from corrai.base.model import Model +MODEL_VARIABLES_TYPES_MAP = { + "Real": float, + "Float32": float, + "Float64": float, + "Integer": int, + "Int8": int, + "UInt8": int, + "Int16": int, + "UInt16": int, + "Int32": int, + "UInt32": int, + "Int64": int, + "UInt64": int, + "Boolean": bool, + "String": str, + # Not yet implemented + # ('Enumeration'), + # ('Binary'), + # ('Clock'), +} + DEFAULT_SIMULATION_OPTIONS = { "startTime": 0, "stopTime": 24 * 3600, @@ -271,7 +292,6 @@ def __init__( simulation_dir: Path = None, output_list: list[str] = None, boundary_table_name: str | None = None, - default_properties: dict[str, float | int | str] = None, ): super().__init__(is_dynamic=True) fmu_path = Path(fmu_path) if isinstance(fmu_path, str) else fmu_path @@ -298,7 +318,26 @@ def __init__( UserWarning, stacklevel=2, ) - self.default_properties = default_properties or {} + self.property_dict = self._get_init_property_dict() + + def _get_init_property_dict(self): + model_description = fmpy.read_model_description(self.fmu_path.as_posix()) + fmu_param_dict = { + var.name: var + for var in model_description.modelVariables + if var.causality == "parameter" + } + prop_dict = {} + for name, var in fmu_param_dict.items(): + try: + prop_dict[name] = MODEL_VARIABLES_TYPES_MAP[var.type](var.start) + except TypeError: + prop_dict[name] = var.start + + except KeyError: + pass + + return prop_dict def get_property_values( self, property_list: str | tuple[str, ...] | list[str] @@ -325,22 +364,13 @@ def get_property_values( >>> model.get_property_values(["x.k", "y.k"]) ['2.0', '2.0'] """ - if isinstance(property_list, str): - property_list = (property_list,) + property_list = ( + [property_list] if isinstance(property_list, str) else property_list + ) + return [self.property_dict[prop] for prop in property_list] - model_description = fmpy.read_model_description(self.fmu_path.as_posix()) - variable_map = {var.name: var for var in model_description.modelVariables} - values = [] - for prop in property_list: - if prop in variable_map: - try: - val = float(variable_map[prop].start) - except (ValueError, TypeError): - val = variable_map[prop].start - values.append(val) - else: - values.append(None) - return values + def set_property_values(self, property_dict: dict): + self.property_dict.update(property_dict.items()) def simulate( self, @@ -442,13 +472,12 @@ def simulate( 2.0 3.0 """ - property_dict = dict(property_dict or {}) - merged_properties = self.default_properties.copy() - if property_dict: - merged_properties.update(property_dict) - property_dict = merged_properties + + simu_property = self.property_dict.copy() + simu_property.update(dict(property_dict or {})) + if property_dict and debug_param: - print(property_dict) + print(simu_property) simulation_options = { **DEFAULT_SIMULATION_OPTIONS, @@ -468,8 +497,8 @@ def simulate( ) boundary_df = None - if property_dict: - boundary_df = property_dict.pop("boundary", boundary_df) + if simu_property: + boundary_df = simu_property.pop("boundary", boundary_df) if simulation_options: sim_boundary = simulation_options.pop("boundary", boundary_df) @@ -505,7 +534,7 @@ def simulate( local_boundary, ): if local_boundary is not None and self.boundary_table_name: - property_dict[f"{self.boundary_table_name}.fileName"] = ( + simu_property[f"{self.boundary_table_name}.fileName"] = ( local_boundary.as_posix() ) @@ -515,7 +544,7 @@ def simulate( stop_time=stop_sec, step_size=step_sec, relative_tolerance=simulation_options["tolerance"], - start_values=property_dict, + start_values=simu_property, output=self.output_list, solver=simulation_options["solver"], output_interval=output_int_sec, diff --git a/tests/test_fmu.py b/tests/test_fmu.py index 07ef0d3..e45ae71 100644 --- a/tests/test_fmu.py +++ b/tests/test_fmu.py @@ -141,9 +141,6 @@ def test_get_property_values(self): fmu_path=PACKAGE_DIR / "rosen.fmu", output_list=["res.showNumber"], ) - values = simu.get_property_values(("res.showNumber",)) - assert isinstance(values, list) - assert len(values) == 1 vals = simu.get_property_values("x.k") assert vals == [2.0] @@ -154,6 +151,9 @@ def test_get_property_values(self): vals = simu.get_property_values(["x.k", "y.k"]) assert vals == [2.0, 2.0] + simu.set_property_values({"x.k": 3.0}) + assert simu.get_property_values("x.k") == [3.0] + def test_simulate_parallel(self): simu = ModelicaFmuModel( fmu_path=PACKAGE_DIR / "rosen.fmu",