[ModelExecution*] create classes to handle model execution

* rename ModelicaSystemCmd => ModelExecutionCmd
* rename OMCSessionRunData => ModelExecutionData
* create class ModelExecutionException
* move some code:
  * OMCSession.omc_run_data_update() => merge into ModelExecutionCmd.define()
  * OMCSession.run_model_executable() => ModelExecutionData.run()

Author: syntron

OMPython/ModelicaSystem.py

@@ -22,8 +22,10 @@
 import re
 
 from OMPython.OMCSession import (
+    ModelExecutionData,
+    ModelExecutionException,
+
     OMCSessionException,
-    OMCSessionRunData,
     OMCSession,
     OMCSessionLocal,
     OMCPath,
@@ -35,7 +37,7 @@
 
 class ModelicaSystemError(Exception):
     """
-    Exception used in ModelicaSystem and ModelicaSystemCmd classes.
+    Exception used in ModelicaSystem classes.
     """
 
 
@@ -90,7 +92,7 @@ def __getitem__(self, index: int):
         return {0: self.A, 1: self.B, 2: self.C, 3: self.D}[index]
 
 
-class ModelicaSystemCmd:
+class ModelExecutionCmd:
     """
     All information about a compiled model executable. This should include data about all structured parameters, i.e.
     parameters which need a recompilation of the model. All non-structured parameters can be easily changed without
@@ -99,16 +101,22 @@ class ModelicaSystemCmd:
 
     def __init__(
             self,
-            session: OMCSession,
-            runpath: OMCPath,
-            modelname: Optional[str] = None,
+            runpath: os.PathLike,
+            cmd_prefix: list[str],
+            cmd_local: bool = False,
+            cmd_windows: bool = False,
+            timeout: float = 10.0,
+            model_name: Optional[str] = None,
     ) -> None:
-        if modelname is None:
-            raise ModelicaSystemError("Missing model name!")
+        if model_name is None:
+            raise ModelExecutionException("Missing model name!")
 
-        self._session = session
-        self._runpath = runpath
-        self._model_name = modelname
+        self._cmd_local = cmd_local
+        self._cmd_windows = cmd_windows
+        self._cmd_prefix = cmd_prefix
+        self._runpath = pathlib.PurePosixPath(runpath)
+        self._model_name = model_name
+        self._timeout = timeout
 
         # dictionaries of command line arguments for the model executable
         self._args: dict[str, str | None] = {}
@@ -153,26 +161,26 @@ def override2str(
             elif isinstance(oval, numbers.Number):
                 oval_str = str(oval)
             else:
-                raise ModelicaSystemError(f"Invalid value for override key {okey}: {type(oval)}")
+                raise ModelExecutionException(f"Invalid value for override key {okey}: {type(oval)}")
 
             return f"{okey}={oval_str}"
 
         if not isinstance(key, str):
-            raise ModelicaSystemError(f"Invalid argument key: {repr(key)} (type: {type(key)})")
+            raise ModelExecutionException(f"Invalid argument key: {repr(key)} (type: {type(key)})")
         key = key.strip()
 
         if isinstance(val, dict):
             if key != 'override':
-                raise ModelicaSystemError("Dictionary input only possible for key 'override'!")
+                raise ModelExecutionException("Dictionary input only possible for key 'override'!")
 
             for okey, oval in val.items():
                 if not isinstance(okey, str):
-                    raise ModelicaSystemError("Invalid key for argument 'override': "
-                                              f"{repr(okey)} (type: {type(okey)})")
+                    raise ModelExecutionException("Invalid key for argument 'override': "
+                                                  f"{repr(okey)} (type: {type(okey)})")
 
                 if not isinstance(oval, (str, bool, numbers.Number, type(None))):
-                    raise ModelicaSystemError(f"Invalid input for 'override'.{repr(okey)}: "
-                                              f"{repr(oval)} (type: {type(oval)})")
+                    raise ModelExecutionException(f"Invalid input for 'override'.{repr(okey)}: "
+                                                  f"{repr(oval)} (type: {type(oval)})")
 
                 if okey in self._arg_override:
                     if oval is None:
@@ -194,7 +202,7 @@ def override2str(
         elif isinstance(val, numbers.Number):
             argval = str(val)
         else:
-            raise ModelicaSystemError(f"Invalid argument value for {repr(key)}: {repr(val)} (type: {type(val)})")
+            raise ModelExecutionException(f"Invalid argument value for {repr(key)}: {repr(val)} (type: {type(val)})")
 
         if key in self._args:
             logger.warning(f"Override model executable argument: {repr(key)} = {repr(argval)} "
@@ -234,7 +242,7 @@ def get_cmd_args(self) -> list[str]:
 
         return cmdl
 
-    def definition(self) -> OMCSessionRunData:
+    def definition(self) -> ModelExecutionData:
         """
         Define all needed data to run the model executable. The data is stored in an OMCSessionRunData object.
         """
@@ -243,18 +251,50 @@ def definition(self) -> OMCSessionRunData:
         if not isinstance(result_file, str):
             result_file = (self._runpath / f"{self._model_name}.mat").as_posix()
 
-        omc_run_data = OMCSessionRunData(
-            cmd_path=self._runpath.as_posix(),
+        # as this is the local implementation, pathlib.Path can be used
+        cmd_path = self._runpath
+
+        cmd_library_path = None
+        if self._cmd_local and self._cmd_windows:
+            cmd_library_path = ""
+
+            # set the process environment from the generated .bat file in windows which should have all the dependencies
+            # for this pathlib.PurePosixPath() must be converted to a pathlib.Path() object, i.e. WindowsPath
+            path_bat = pathlib.Path(cmd_path) / f"{self._model_name}.bat"
+            if not path_bat.is_file():
+                raise ModelExecutionException("Batch file (*.bat) does not exist " + str(path_bat))
+
+            content = path_bat.read_text(encoding='utf-8')
+            for line in content.splitlines():
+                match = re.match(pattern=r"^SET PATH=([^%]*)", string=line, flags=re.IGNORECASE)
+                if match:
+                    cmd_library_path = match.group(1).strip(';')  # Remove any trailing semicolons
+            my_env = os.environ.copy()
+            my_env["PATH"] = cmd_library_path + os.pathsep + my_env["PATH"]
+
+            cmd_model_executable = cmd_path / f"{self._model_name}.exe"
+        else:
+            # for Linux the paths to the needed libraries should be included in the executable (using rpath)
+            cmd_model_executable = cmd_path / self._model_name
+
+        # define local(!) working directory
+        cmd_cwd_local = None
+        if self._cmd_local:
+            cmd_cwd_local = cmd_path.as_posix()
+
+        omc_run_data = ModelExecutionData(
+            cmd_path=cmd_path.as_posix(),
             cmd_model_name=self._model_name,
             cmd_args=self.get_cmd_args(),
-            cmd_result_path=result_file,
+            cmd_result_file=result_file,
+            cmd_prefix=self._cmd_prefix,
+            cmd_library_path=cmd_library_path,
+            cmd_model_executable=cmd_model_executable.as_posix(),
+            cmd_cwd_local=cmd_cwd_local,
+            cmd_timeout=self._timeout,
         )
 
-        omc_run_data_updated = self._session.omc_run_data_update(
-            omc_run_data=omc_run_data,
-        )
-
-        return omc_run_data_updated
+        return omc_run_data
 
     @staticmethod
     def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, Any] | numbers.Number]]:
@@ -263,15 +303,19 @@ def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, Any] | n
 
         The return data can be used as input for self.args_set().
         """
-        warnings.warn("The argument 'simflags' is depreciated and will be removed in future versions; "
-                      "please use 'simargs' instead", DeprecationWarning, stacklevel=2)
+        warnings.warn(
+            message="The argument 'simflags' is depreciated and will be removed in future versions; "
+                    "please use 'simargs' instead",
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
 
         simargs: dict[str, Optional[str | dict[str, Any] | numbers.Number]] = {}
 
         args = [s for s in simflags.split(' ') if s]
         for arg in args:
             if arg[0] != '-':
-                raise ModelicaSystemError(f"Invalid simulation flag: {arg}")
+                raise ModelExecutionException(f"Invalid simulation flag: {arg}")
             arg = arg[1:]
             parts = arg.split('=')
             if len(parts) == 1:
@@ -283,12 +327,12 @@ def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, Any] | n
                 for item in override.split(','):
                     kv = item.split('=')
                     if not 0 < len(kv) < 3:
-                        raise ModelicaSystemError(f"Invalid value for '-override': {override}")
+                        raise ModelExecutionException(f"Invalid value for '-override': {override}")
                     if kv[0]:
                         try:
                             override_dict[kv[0]] = kv[1]
                         except (KeyError, IndexError) as ex:
-                            raise ModelicaSystemError(f"Invalid value for '-override': {override}") from ex
+                            raise ModelExecutionException(f"Invalid value for '-override': {override}") from ex
 
                 simargs[parts[0]] = override_dict
 
@@ -542,15 +586,17 @@ def buildModel(self, variableFilter: Optional[str] = None):
         logger.debug("OM model build result: %s", build_model_result)
 
         # check if the executable exists ...
-        om_cmd = ModelicaSystemCmd(
-            session=self._session,
+        om_cmd = ModelExecutionCmd(
             runpath=self.getWorkDirectory(),
-            modelname=self._model_name,
+            cmd_local=self._session.model_execution_local,
+            cmd_windows=self._session.model_execution_windows,
+            cmd_prefix=self._session.model_execution_prefix(cwd=self.getWorkDirectory()),
+            model_name=self._model_name,
         )
         # ... by running it - output help for command help
         om_cmd.arg_set(key="help", val="help")
         cmd_definition = om_cmd.definition()
-        returncode = self._session.run_model_executable(cmd_run_data=cmd_definition)
+        returncode = cmd_definition.run()
         if returncode != 0:
             raise ModelicaSystemError("Model executable not working!")
 
@@ -1035,7 +1081,7 @@ def simulate_cmd(
             result_file: OMCPath,
             simflags: Optional[str] = None,
             simargs: Optional[dict[str, Optional[str | dict[str, Any] | numbers.Number]]] = None,
-    ) -> ModelicaSystemCmd:
+    ) -> ModelExecutionCmd:
         """
         This method prepares the simulates model according to the simulation options. It returns an instance of
         ModelicaSystemCmd which can be used to run the simulation.
@@ -1057,10 +1103,12 @@ def simulate_cmd(
             An instance if ModelicaSystemCmd to run the requested simulation.
         """
 
-        om_cmd = ModelicaSystemCmd(
-            session=self._session,
+        om_cmd = ModelExecutionCmd(
             runpath=self.getWorkDirectory(),
-            modelname=self._model_name,
+            cmd_local=self._session.model_execution_local,
+            cmd_windows=self._session.model_execution_windows,
+            cmd_prefix=self._session.model_execution_prefix(cwd=self.getWorkDirectory()),
+            model_name=self._model_name,
         )
 
         # always define the result file to use
@@ -1166,7 +1214,7 @@ def simulate(
             self._result_file.unlink()
         # ... run simulation ...
         cmd_definition = om_cmd.definition()
-        returncode = self._session.run_model_executable(cmd_run_data=cmd_definition)
+        returncode = cmd_definition.run()
         # and check returncode *AND* resultfile
         if returncode != 0 and self._result_file.is_file():
             # check for an empty (=> 0B) result file which indicates a crash of the model executable
@@ -1769,10 +1817,12 @@ def linearize(
                 "use ModelicaSystem() to build the model first"
             )
 
-        om_cmd = ModelicaSystemCmd(
-            session=self._session,
+        om_cmd = ModelExecutionCmd(
             runpath=self.getWorkDirectory(),
-            modelname=self._model_name,
+            cmd_local=self._session.model_execution_local,
+            cmd_windows=self._session.model_execution_windows,
+            cmd_prefix=self._session.model_execution_prefix(cwd=self.getWorkDirectory()),
+            model_name=self._model_name,
         )
 
         # See comment in simulate_cmd regarding override file and OM version
@@ -1819,7 +1869,7 @@ def linearize(
         linear_file.unlink(missing_ok=True)
 
         cmd_definition = om_cmd.definition()
-        returncode = self._session.run_model_executable(cmd_run_data=cmd_definition)
+        returncode = cmd_definition.run()
         if returncode != 0:
             raise ModelicaSystemError(f"Linearize failed with return code: {returncode}")
         if not linear_file.is_file():
@@ -2005,7 +2055,7 @@ def __init__(
             self._parameters = {}
 
         self._doe_def: Optional[dict[str, dict[str, Any]]] = None
-        self._doe_cmd: Optional[dict[str, OMCSessionRunData]] = None
+        self._doe_cmd: Optional[dict[str, ModelExecutionData]] = None
 
     def get_session(self) -> OMCSession:
         """
@@ -2124,7 +2174,7 @@ def get_doe_definition(self) -> Optional[dict[str, dict[str, Any]]]:
         """
         return self._doe_def
 
-    def get_doe_command(self) -> Optional[dict[str, OMCSessionRunData]]:
+    def get_doe_command(self) -> Optional[dict[str, ModelExecutionData]]:
         """
         Get the definitions of simulations commands to run for this DoE.
         """
@@ -2170,13 +2220,13 @@ def worker(worker_id, task_queue):
                 if cmd_definition is None:
                     raise ModelicaSystemError("Missing simulation definition!")
 
-                resultfile = cmd_definition.cmd_result_path
+                resultfile = cmd_definition.cmd_result_file
                 resultpath = self.get_session().omcpath(resultfile)
 
                 logger.info(f"[Worker {worker_id}] Performing task: {resultpath.name}")
 
                 try:
-                    returncode = self.get_session().run_model_executable(cmd_run_data=cmd_definition)
+                    returncode = cmd_definition.run()
                     logger.info(f"[Worker {worker_id}] Simulation {resultpath.name} "
                                 f"finished with return code: {returncode}")
                 except ModelicaSystemError as ex: