[OMCSessionRunData] use class to move run of model executable to OMSessionZMQ

Author: syntron

OMPython/ModelicaSystem.py

@@ -40,16 +40,13 @@
 import numpy as np
 import os
 import pathlib
-import platform
-import re
-import subprocess
 import tempfile
 import textwrap
 from typing import Optional, Any
 import warnings
 import xml.etree.ElementTree as ET
 
-from OMPython.OMCSession import OMCSessionException, OMCSessionZMQ, OMCProcessLocal
+from OMPython.OMCSession import OMCSessionException, OMCSessionRunData, OMCSessionZMQ, OMCProcessLocal
 
 # define logger using the current module name as ID
 logger = logging.getLogger(__name__)
@@ -115,7 +112,14 @@ def __getitem__(self, index: int):
 class ModelicaSystemCmd:
     """A compiled model executable."""
 
-    def __init__(self, runpath: pathlib.Path, modelname: str, timeout: Optional[float] = None) -> None:
+    def __init__(
+            self,
+            session: OMCSessionZMQ,
+            runpath: pathlib.Path,
+            modelname: str,
+            timeout: Optional[float] = None,
+    ) -> None:
+        self._session = session
         self._runpath = pathlib.Path(runpath).resolve().absolute()
         self._model_name = modelname
         self._timeout = timeout
@@ -176,27 +180,12 @@ def args_set(self, args: dict[str, Optional[str | dict[str, str]]]) -> None:
         for arg in args:
             self.arg_set(key=arg, val=args[arg])
 
-    def get_exe(self) -> pathlib.Path:
-        """Get the path to the compiled model executable."""
-        if platform.system() == "Windows":
-            path_exe = self._runpath / f"{self._model_name}.exe"
-        else:
-            path_exe = self._runpath / self._model_name
-
-        if not path_exe.exists():
-            raise ModelicaSystemError(f"Application file path not found: {path_exe}")
-
-        return path_exe
-
-    def get_cmd(self) -> list:
-        """Get a list with the path to the executable and all command line args.
-
-        This can later be used as an argument for subprocess.run().
+    def get_cmd_args(self) -> list[str]:
+        """
+        Get a list with the command arguments for the model executable.
         """
 
-        path_exe = self.get_exe()
-
-        cmdl = [path_exe.as_posix()]
+        cmdl = []
         for key in self._args:
             if self._args[key] is None:
                 cmdl.append(f"-{key}")
@@ -205,54 +194,26 @@ def get_cmd(self) -> list:
 
         return cmdl
 
-    def run(self) -> int:
-        """Run the requested simulation.
-
-        Returns
-        -------
-            Subprocess return code (0 on success).
+    def definition(self) -> OMCSessionRunData:
         """
+        Define all needed data to run the model executable. The data is stored in an OMCSessionRunData object.
+        """
+        # ensure that a result filename is provided
+        result_file = self.arg_get('r')
+        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(),
+            cmd_model_name=self._model_name,
+            cmd_args=self.get_cmd_args(),
+            cmd_result_path=result_file,
+            cmd_timeout=self._timeout,
+        )
 
-        cmdl: list = self.get_cmd()
-
-        logger.debug("Run OM command %s in %s", repr(cmdl), self._runpath.as_posix())
-
-        if platform.system() == "Windows":
-            path_dll = ""
-
-            # set the process environment from the generated .bat file in windows which should have all the dependencies
-            path_bat = self._runpath / f"{self._model_name}.bat"
-            if not path_bat.exists():
-                raise ModelicaSystemError("Batch file (*.bat) does not exist " + str(path_bat))
-
-            with open(file=path_bat, mode='r', encoding='utf-8') as fh:
-                for line in fh:
-                    match = re.match(r"^SET PATH=([^%]*)", line, re.IGNORECASE)
-                    if match:
-                        path_dll = match.group(1).strip(';')  # Remove any trailing semicolons
-            my_env = os.environ.copy()
-            my_env["PATH"] = path_dll + os.pathsep + my_env["PATH"]
-        else:
-            # TODO: how to handle path to resources of external libraries for any system not Windows?
-            my_env = None
-
-        try:
-            cmdres = subprocess.run(cmdl, capture_output=True, text=True, env=my_env, cwd=self._runpath,
-                                    timeout=self._timeout, check=True)
-            stdout = cmdres.stdout.strip()
-            stderr = cmdres.stderr.strip()
-            returncode = cmdres.returncode
-
-            logger.debug("OM output for command %s:\n%s", repr(cmdl), stdout)
-
-            if stderr:
-                raise ModelicaSystemError(f"Error running command {repr(cmdl)}: {stderr}")
-        except subprocess.TimeoutExpired as ex:
-            raise ModelicaSystemError(f"Timeout running command {repr(cmdl)}") from ex
-        except subprocess.CalledProcessError as ex:
-            raise ModelicaSystemError(f"Error running command {repr(cmdl)}") from ex
+        omc_run_data_updated = self._session.omc_run_data_update(omc_run_data)
 
-        return returncode
+        return omc_run_data_updated
 
     @staticmethod
     def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, str]]]:
@@ -943,7 +904,12 @@ def simulate_cmd(
             An instance if ModelicaSystemCmd to run the requested simulation.
         """
 
-        om_cmd = ModelicaSystemCmd(runpath=self._tempdir, modelname=self._model_name, timeout=timeout)
+        om_cmd = ModelicaSystemCmd(
+            session=self._getconn,
+            runpath=self.getWorkDirectory(),
+            modelname=self._model_name,
+            timeout=timeout,
+        )
 
         # always define the result file to use
         om_cmd.arg_set(key="r", val=result_file.as_posix())
@@ -1033,7 +999,8 @@ def simulate(
         if self._result_file.is_file():
             self._result_file.unlink()
         # ... run simulation ...
-        returncode = om_cmd.run()
+        cmd_definition = om_cmd.definition()
+        returncode = self._getconn.run_model_executable(cmd_run_data=cmd_definition)
         # 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
@@ -1452,7 +1419,12 @@ def load_module_from_path(module_name, file_path):
                 "use ModelicaSystem() to build the model first"
             )
 
-        om_cmd = ModelicaSystemCmd(runpath=self._tempdir, modelname=self._model_name, timeout=timeout)
+        om_cmd = ModelicaSystemCmd(
+            session=self._getconn,
+            runpath=self._tempdir,
+            modelname=self._model_name,
+            timeout=timeout,
+        )
 
         overrideLinearFile = self._tempdir / f'{self._model_name}_override_linear.txt'
 
@@ -1484,7 +1456,8 @@ def load_module_from_path(module_name, file_path):
         if simargs:
             om_cmd.args_set(args=simargs)
 
-        returncode = om_cmd.run()
+        cmd_definition = om_cmd.definition()
+        returncode = self._getconn.run_model_executable(cmd_run_data=cmd_definition)
         if returncode != 0:
             raise ModelicaSystemError(f"Linearize failed with return code: {returncode}")
 

OMPython/OMCSession.py

@@ -34,12 +34,14 @@
  CONDITIONS OF OSMC-PL.
 """
 
+import abc
 import dataclasses
 import io
 import json
 import logging
 import os
 import pathlib
+import platform
 import psutil
 import pyparsing
 import re
@@ -365,6 +367,53 @@ def __del__(self):
 
         self.omc_zmq = None
 
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData) -> OMCSessionRunData:
+        """
+        Modify data based on the selected OMCProcess implementation.
+
+        Needs to be implemented in the subclasses.
+        """
+        return self.omc_process.omc_run_data_update(omc_run_data=omc_run_data)
+
+    @staticmethod
+    def run_model_executable(cmd_run_data: OMCSessionRunData) -> int:
+        """
+        Run the command defined in cmd_run_data. This class is defined as static method such that there is no need to
+        keep instances of over classes around.
+        """
+
+        my_env = os.environ.copy()
+        if isinstance(cmd_run_data.cmd_library_path, str):
+            my_env["PATH"] = cmd_run_data.cmd_library_path + os.pathsep + my_env["PATH"]
+
+        cmdl = cmd_run_data.get_cmd()
+
+        logger.debug("Run OM command %s in %s", repr(cmdl), cmd_run_data.cmd_path)
+        try:
+            cmdres = subprocess.run(
+                cmdl,
+                capture_output=True,
+                text=True,
+                env=my_env,
+                cwd=cmd_run_data.cmd_cwd_local,
+                timeout=cmd_run_data.cmd_timeout,
+                check=True,
+            )
+            stdout = cmdres.stdout.strip()
+            stderr = cmdres.stderr.strip()
+            returncode = cmdres.returncode
+
+            logger.debug("OM output for command %s:\n%s", repr(cmdl), stdout)
+
+            if stderr:
+                raise OMCSessionException(f"Error running model executable {repr(cmdl)}: {stderr}")
+        except subprocess.TimeoutExpired as ex:
+            raise OMCSessionException(f"Timeout running model executable {repr(cmdl)}") from ex
+        except subprocess.CalledProcessError as ex:
+            raise OMCSessionException(f"Error running model executable {repr(cmdl)}") from ex
+
+        return returncode
+
     def execute(self, command: str):
         warnings.warn("This function is depreciated and will be removed in future versions; "
                       "please use sendExpression() instead", DeprecationWarning, stacklevel=2)
@@ -468,7 +517,7 @@ def sendExpression(self, command: str, parsed: bool = True) -> Any:
                 raise OMCSessionException("Cannot parse OMC result") from ex
 
 
-class OMCProcess:
+class OMCProcess(metaclass=abc.ABCMeta):
 
     def __init__(
             self,
@@ -550,6 +599,15 @@ def _get_portfile_path(self) -> Optional[pathlib.Path]:
 
         return portfile_path
 
+    @abc.abstractmethod
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData) -> OMCSessionRunData:
+        """
+        Update the OMCSessionRunData object based on the selected OMCProcess implementation.
+
+        Needs to be implemented in the subclasses.
+        """
+        raise NotImplementedError("This method must be implemented in subclasses!")
+
 
 class OMCProcessPort(OMCProcess):
 
@@ -560,6 +618,12 @@ def __init__(
         super().__init__()
         self._omc_port = omc_port
 
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData) -> OMCSessionRunData:
+        """
+        Update the OMCSessionRunData object based on the selected OMCProcess implementation.
+        """
+        raise OMCSessionException("OMCProcessPort does not support omc_run_data_update()!")
+
 
 class OMCProcessLocal(OMCProcess):
 
@@ -641,6 +705,48 @@ def _omc_port_get(self) -> str:
 
         return port
 
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData) -> OMCSessionRunData:
+        """
+        Update the OMCSessionRunData object based on the selected OMCProcess implementation.
+        """
+        # create a copy of the data
+        omc_run_data_copy = dataclasses.replace(omc_run_data)
+
+        # as this is the local implementation, pathlib.Path can be used
+        cmd_path = pathlib.Path(omc_run_data_copy.cmd_path)
+
+        if platform.system() == "Windows":
+            path_dll = ""
+
+            # set the process environment from the generated .bat file in windows which should have all the dependencies
+            path_bat = cmd_path / f"{omc_run_data.cmd_model_name}.bat"
+            if not path_bat.is_file():
+                raise OMCSessionException("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(r"^SET PATH=([^%]*)", line, re.IGNORECASE)
+                if match:
+                    path_dll = match.group(1).strip(';')  # Remove any trailing semicolons
+            my_env = os.environ.copy()
+            my_env["PATH"] = path_dll + os.pathsep + my_env["PATH"]
+
+            omc_run_data_copy.cmd_library_path = path_dll
+
+            cmd_model_executable = cmd_path / f"{omc_run_data_copy.cmd_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 / omc_run_data_copy.cmd_model_name
+
+        if not cmd_model_executable.is_file():
+            raise OMCSessionException(f"Application file path not found: {cmd_model_executable}")
+        omc_run_data_copy.cmd_model_executable = cmd_model_executable.as_posix()
+
+        # define local(!) working directory
+        omc_run_data_copy.cmd_cwd_local = omc_run_data.cmd_path
+
+        return omc_run_data_copy
+
 
 class OMCProcessDockerHelper(OMCProcess):
 
@@ -747,6 +853,12 @@ def get_docker_container_id(self) -> str:
 
         return self._dockerCid
 
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData) -> OMCSessionRunData:
+        """
+        Update the OMCSessionRunData object based on the selected OMCProcess implementation.
+        """
+        raise OMCSessionException("OMCProcessDocker* does not support omc_run_data_update()!")
+
 
 class OMCProcessDocker(OMCProcessDockerHelper):
 
@@ -1052,3 +1164,9 @@ def _omc_port_get(self) -> str:
                     f"pid={self._omc_process.pid if isinstance(self._omc_process, subprocess.Popen) else '?'}")
 
         return port
+
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData) -> OMCSessionRunData:
+        """
+        Update the OMCSessionRunData object based on the selected OMCProcess implementation.
+        """
+        raise OMCSessionException("OMCProcessWSL does not support omc_run_data_update()!")