From 4739a00582d4e6c0a005e4af3f256d4fb21fce96 Mon Sep 17 00:00:00 2001 From: syntron Date: Sat, 24 Jan 2026 21:23:10 +0100 Subject: [PATCH] (B002) split ModelicaSystemDoE [ModelicaSystem] split ModelicaSystemDoE into ModelicaDoEABC and ModelicaDoE [ModelicaSystem] rename ModelicaSystemDoE => ModelicaDoEOMC * add compatibility variable for ModelicaSystemDoE [test_ModelicaDoEOMC] rename from ModelicaSystemDoE and update [ModelicaSystem] update ModelicaDoEABC to use ModelicaSystemABC [ModelicaSystem] define doe_get_solutions() as separate method --- OMPython/ModelicaSystem.py | 234 ++++++++++++------ OMPython/__init__.py | 7 + ...icaSystemDoE.py => test_ModelicaDoEOMC.py} | 20 +- 3 files changed, 176 insertions(+), 85 deletions(-) rename tests/{test_ModelicaSystemDoE.py => test_ModelicaDoEOMC.py} (88%) diff --git a/OMPython/ModelicaSystem.py b/OMPython/ModelicaSystem.py index 383377a7..e44b37d4 100644 --- a/OMPython/ModelicaSystem.py +++ b/OMPython/ModelicaSystem.py @@ -15,7 +15,7 @@ import re import textwrap import threading -from typing import Any, cast, Optional +from typing import Any, cast, Optional, Tuple import warnings import xml.etree.ElementTree as ET @@ -2112,9 +2112,9 @@ class ModelicaSystem(ModelicaSystemOMC): """ -class ModelicaSystemDoE: +class ModelicaDoEABC(metaclass=abc.ABCMeta): """ - Class to run DoEs based on a (Open)Modelica model using ModelicaSystem + Base class to run DoEs based on a (Open)Modelica model using ModelicaSystem Example ------- @@ -2187,7 +2187,7 @@ def run_doe(): def __init__( self, # ModelicaSystem definition to use - mod: ModelicaSystemOMC, + mod: ModelicaSystemABC, # simulation specific input # TODO: add more settings (simulation options, input options, ...) simargs: Optional[dict[str, Optional[str | dict[str, str] | numbers.Number]]] = None, @@ -2200,7 +2200,7 @@ def __init__( ModelicaSystem.simulate(). Additionally, the path to store the result files is needed (= resultpath) as well as a list of parameters to vary for the Doe (= parameters). All possible combinations are considered. """ - if not isinstance(mod, ModelicaSystemOMC): + if not isinstance(mod, ModelicaSystemABC): raise ModelicaSystemError("Missing definition of ModelicaSystem!") self._mod = mod @@ -2256,30 +2256,11 @@ def prepare(self) -> int: param_non_structural_combinations = list(itertools.product(*param_non_structure.values())) for idx_pc_structure, pc_structure in enumerate(param_structure_combinations): - - build_dir = self._resultpath / f"DOE_{idx_pc_structure:09d}" - build_dir.mkdir() - self._mod.setWorkDirectory(work_directory=build_dir) - - sim_param_structure = {} - for idx_structure, pk_structure in enumerate(param_structure.keys()): - sim_param_structure[pk_structure] = pc_structure[idx_structure] - - pk_value = pc_structure[idx_structure] - if isinstance(pk_value, str): - pk_value_str = self.get_session().escape_str(pk_value) - expr = f"setParameterValue({self._model_name}, {pk_structure}, \"{pk_value_str}\")" - elif isinstance(pk_value, bool): - pk_value_bool_str = "true" if pk_value else "false" - expr = f"setParameterValue({self._model_name}, {pk_structure}, {pk_value_bool_str});" - else: - expr = f"setParameterValue({self._model_name}, {pk_structure}, {pk_value})" - res = self._mod.sendExpression(expr=expr) - if not res: - raise ModelicaSystemError(f"Cannot set structural parameter {self._model_name}.{pk_structure} " - f"to {pk_value} using {repr(expr)}") - - self._mod.buildModel() + sim_param_structure = self._prepare_structure_parameters( + idx_pc_structure=idx_pc_structure, + pc_structure=pc_structure, + param_structure=param_structure, + ) for idx_non_structural, pk_non_structural in enumerate(param_non_structural_combinations): sim_param_non_structural = {} @@ -2324,6 +2305,17 @@ def prepare(self) -> int: return len(doe_sim) + @abc.abstractmethod + def _prepare_structure_parameters( + self, + idx_pc_structure: int, + pc_structure: Tuple, + param_structure: dict[str, list[str] | list[int] | list[float]], + ) -> dict[str, str | int | float]: + """ + Handle structural parameters. This should be implemented by the derived class + """ + def get_doe_definition(self) -> Optional[dict[str, dict[str, Any]]]: """ Get the defined DoE as a dict, where each key is the result filename and the value is a dict of simulation @@ -2435,65 +2427,157 @@ def worker(worker_id, task_queue): return doe_def_total == doe_def_done + +class ModelicaDoEOMC(ModelicaDoEABC): + """ + Class to run DoEs based on a (Open)Modelica model using ModelicaSystemOMC + + The example is the same as defined for ModelicaDoEABC + """ + + def __init__( + self, + # ModelicaSystem definition to use + mod: ModelicaSystemOMC, + # simulation specific input + # TODO: add more settings (simulation options, input options, ...) + simargs: Optional[dict[str, Optional[str | dict[str, str] | numbers.Number]]] = None, + # DoE specific inputs + resultpath: Optional[str | os.PathLike] = None, + parameters: Optional[dict[str, list[str] | list[int] | list[float]]] = None, + ) -> None: + + if not isinstance(mod, ModelicaSystemOMC): + raise ModelicaSystemError(f"Invalid definition for mod: {type(mod)} - expect ModelicaSystemOMC!") + + super().__init__( + mod=mod, + simargs=simargs, + resultpath=resultpath, + parameters=parameters, + ) + + def _prepare_structure_parameters( + self, + idx_pc_structure: int, + pc_structure: Tuple, + param_structure: dict[str, list[str] | list[int] | list[float]], + ) -> dict[str, str | int | float]: + build_dir = self._resultpath / f"DOE_{idx_pc_structure:09d}" + build_dir.mkdir() + self._mod.setWorkDirectory(work_directory=build_dir) + + # need to repeat this check to make the linters happy + if not isinstance(self._mod, ModelicaSystemOMC): + raise ModelicaSystemError(f"Invalid definition for mod: {type(self._mod)} - expect ModelicaSystemOMC!") + + sim_param_structure = {} + for idx_structure, pk_structure in enumerate(param_structure.keys()): + sim_param_structure[pk_structure] = pc_structure[idx_structure] + + pk_value = pc_structure[idx_structure] + if isinstance(pk_value, str): + pk_value_str = self.get_session().escape_str(pk_value) + expr = f"setParameterValue({self._model_name}, {pk_structure}, \"{pk_value_str}\")" + elif isinstance(pk_value, bool): + pk_value_bool_str = "true" if pk_value else "false" + expr = f"setParameterValue({self._model_name}, {pk_structure}, {pk_value_bool_str});" + else: + expr = f"setParameterValue({self._model_name}, {pk_structure}, {pk_value})" + res = self._mod.sendExpression(expr=expr) + if not res: + raise ModelicaSystemError(f"Cannot set structural parameter {self._model_name}.{pk_structure} " + f"to {pk_value} using {repr(expr)}") + + self._mod.buildModel() + + return sim_param_structure + def get_doe_solutions( self, var_list: Optional[list] = None, ) -> Optional[tuple[str] | dict[str, dict[str, np.ndarray]]]: """ - Get all solutions of the DoE run. The following return values are possible: + Wrapper for doe_get_solutions() + """ + if not isinstance(self._mod, ModelicaSystemOMC): + raise ModelicaSystemError(f"Invalid definition for mod: {type(self._mod)} - expect ModelicaSystemOMC!") - * A list of variables if val_list == None + return doe_get_solutions( + msomc=self._mod, + resultpath=self._resultpath, + doe_def=self.get_doe_definition(), + var_list=var_list, + ) - * The Solutions as dict[str, pd.DataFrame] if a value list (== val_list) is defined. - The following code snippet can be used to convert the solution data for each run to a pandas dataframe: +def doe_get_solutions( + msomc: ModelicaSystemOMC, + resultpath: OMCPath, + doe_def: Optional[dict] = None, + var_list: Optional[list] = None, +) -> Optional[tuple[str] | dict[str, dict[str, np.ndarray]]]: + """ + Get all solutions of the DoE run. The following return values are possible: - ``` - import pandas as pd + * A list of variables if val_list == None - doe_sol = doe_mod.get_doe_solutions() - for key in doe_sol: - data = doe_sol[key]['data'] - if data: - doe_sol[key]['df'] = pd.DataFrame.from_dict(data=data) - else: - doe_sol[key]['df'] = None - ``` + * The Solutions as dict[str, pd.DataFrame] if a value list (== val_list) is defined. - """ - if not isinstance(self._doe_def, dict): - return None + The following code snippet can be used to convert the solution data for each run to a pandas dataframe: - if len(self._doe_def) == 0: - raise ModelicaSystemError("No result files available - all simulations did fail?") + ``` + import pandas as pd - sol_dict: dict[str, dict[str, Any]] = {} - for resultfilename in self._doe_def: - resultfile = self._resultpath / resultfilename + doe_sol = doe_mod.get_doe_solutions() + for key in doe_sol: + data = doe_sol[key]['data'] + if data: + doe_sol[key]['df'] = pd.DataFrame.from_dict(data=data) + else: + doe_sol[key]['df'] = None + ``` - sol_dict[resultfilename] = {} + """ + if not isinstance(doe_def, dict): + return None - if not self._doe_def[resultfilename][self.DICT_RESULT_AVAILABLE]: - msg = f"No result file available for {resultfilename}" - logger.warning(msg) - sol_dict[resultfilename]['msg'] = msg - sol_dict[resultfilename]['data'] = {} - continue + if len(doe_def) == 0: + raise ModelicaSystemError("No result files available - all simulations did fail?") - if var_list is None: - var_list_row = list(self._mod.getSolutions(resultfile=resultfile)) - else: - var_list_row = var_list - - try: - sol = self._mod.getSolutions(varList=var_list_row, resultfile=resultfile) - sol_data = {var: sol[idx] for idx, var in enumerate(var_list_row)} - sol_dict[resultfilename]['msg'] = 'Simulation available' - sol_dict[resultfilename]['data'] = sol_data - except ModelicaSystemError as ex: - msg = f"Error reading solution for {resultfilename}: {ex}" - logger.warning(msg) - sol_dict[resultfilename]['msg'] = msg - sol_dict[resultfilename]['data'] = {} - - return sol_dict + sol_dict: dict[str, dict[str, Any]] = {} + for resultfilename in doe_def: + resultfile = resultpath / resultfilename + + sol_dict[resultfilename] = {} + + if not doe_def[resultfilename][ModelicaDoEABC.DICT_RESULT_AVAILABLE]: + msg = f"No result file available for {resultfilename}" + logger.warning(msg) + sol_dict[resultfilename]['msg'] = msg + sol_dict[resultfilename]['data'] = {} + continue + + if var_list is None: + var_list_row = list(msomc.getSolutions(resultfile=resultfile)) + else: + var_list_row = var_list + + try: + sol = msomc.getSolutions(varList=var_list_row, resultfile=resultfile) + sol_data = {var: sol[idx] for idx, var in enumerate(var_list_row)} + sol_dict[resultfilename]['msg'] = 'Simulation available' + sol_dict[resultfilename]['data'] = sol_data + except ModelicaSystemError as ex: + msg = f"Error reading solution for {resultfilename}: {ex}" + logger.warning(msg) + sol_dict[resultfilename]['msg'] = msg + sol_dict[resultfilename]['data'] = {} + + return sol_dict + + +class ModelicaSystemDoE(ModelicaDoEOMC): + """ + Compatibility class. + """ diff --git a/OMPython/__init__.py b/OMPython/__init__.py index 1f086293..9f4408d5 100644 --- a/OMPython/__init__.py +++ b/OMPython/__init__.py @@ -17,7 +17,10 @@ ModelicaSystemOMC, ModelExecutionCmd, ModelicaSystemDoE, + ModelicaDoEOMC, ModelicaSystemError, + + doe_get_solutions, ) from OMPython.OMCSession import ( OMCPath, @@ -47,11 +50,15 @@ 'ModelicaSystemOMC', 'ModelExecutionCmd', 'ModelicaSystemDoE', + 'ModelicaDoEOMC', 'ModelicaSystemError', 'OMCPath', 'OMCSession', + + 'doe_get_solutions', + 'OMCSessionCmd', 'OMCSessionDocker', 'OMCSessionDockerContainer', diff --git a/tests/test_ModelicaSystemDoE.py b/tests/test_ModelicaDoEOMC.py similarity index 88% rename from tests/test_ModelicaSystemDoE.py rename to tests/test_ModelicaDoEOMC.py index 8b1d1a09..143932fc 100644 --- a/tests/test_ModelicaSystemDoE.py +++ b/tests/test_ModelicaDoEOMC.py @@ -51,7 +51,7 @@ def param_doe() -> dict[str, list]: return param -def test_ModelicaSystemDoE_local(tmp_path, model_doe, param_doe): +def test_ModelicaDoEOMC_local(tmp_path, model_doe, param_doe): tmpdir = tmp_path / 'DoE' tmpdir.mkdir(exist_ok=True) @@ -61,19 +61,19 @@ def test_ModelicaSystemDoE_local(tmp_path, model_doe, param_doe): model_name="M", ) - doe_mod = OMPython.ModelicaSystemDoE( + doe_mod = OMPython.ModelicaDoEOMC( mod=mod, parameters=param_doe, resultpath=tmpdir, simargs={"override": {'stopTime': '1.0'}}, ) - _run_ModelicaSystemDoe(doe_mod=doe_mod) + _run_ModelicaDoEOMC(doe_mod=doe_mod) @skip_on_windows @skip_python_older_312 -def test_ModelicaSystemDoE_docker(tmp_path, model_doe, param_doe): +def test_ModelicaDoEOMC_docker(tmp_path, model_doe, param_doe): omcs = OMPython.OMCSessionDocker(docker="openmodelica/openmodelica:v1.25.0-minimal") omversion = omcs.sendExpression("getVersion()") assert isinstance(omversion, str) and omversion.startswith("OpenModelica") @@ -86,18 +86,18 @@ def test_ModelicaSystemDoE_docker(tmp_path, model_doe, param_doe): model_name="M", ) - doe_mod = OMPython.ModelicaSystemDoE( + doe_mod = OMPython.ModelicaDoEOMC( mod=mod, parameters=param_doe, simargs={"override": {'stopTime': '1.0'}}, ) - _run_ModelicaSystemDoe(doe_mod=doe_mod) + _run_ModelicaDoEOMC(doe_mod=doe_mod) @pytest.mark.skip(reason="Not able to run WSL on github") @skip_python_older_312 -def test_ModelicaSystemDoE_WSL(tmp_path, model_doe, param_doe): +def test_ModelicaDoEOMC_WSL(tmp_path, model_doe, param_doe): omcs = OMPython.OMCSessionWSL() omversion = omcs.sendExpression("getVersion()") assert isinstance(omversion, str) and omversion.startswith("OpenModelica") @@ -110,16 +110,16 @@ def test_ModelicaSystemDoE_WSL(tmp_path, model_doe, param_doe): model_name="M", ) - doe_mod = OMPython.ModelicaSystemDoE( + doe_mod = OMPython.ModelicaDoEOMC( mod=mod, parameters=param_doe, simargs={"override": {'stopTime': '1.0'}}, ) - _run_ModelicaSystemDoe(doe_mod=doe_mod) + _run_ModelicaDoEOMC(doe_mod=doe_mod) -def _run_ModelicaSystemDoe(doe_mod): +def _run_ModelicaDoEOMC(doe_mod): doe_count = doe_mod.prepare() assert doe_count == 16