B002 ModelicaSystemDoE split

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.
+    """

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',

tests/test_ModelicaSystemDoE.py → 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