Rewrite __init__.py

OMPython/OMCSession.py

@@ -34,11 +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
@@ -456,6 +459,47 @@ class OMCPathCompatibilityWindows(pathlib.WindowsPath, OMCPathCompatibility):
     OMCPath = OMCPathReal
 
 
+@dataclasses.dataclass
+class OMCSessionRunData:
+    """
+    Data class to store the command line data for running a model executable in the OMC environment.
+
+    All data should be defined for the environment, where OMC is running (local, docker or WSL)
+    """
+    # cmd_path is the expected working directory
+    cmd_path: str
+    cmd_model_name: str
+    # command line arguments for the model executable
+    cmd_args: list[str]
+    # result file with the simulation output
+    cmd_result_path: str
+
+    # command prefix data (as list of strings); needed for docker or WSL
+    cmd_prefix: Optional[list[str]] = None
+    # cmd_model_executable is build out of cmd_path and cmd_model_name; this is mainly needed on Windows (add *.exe)
+    cmd_model_executable: Optional[str] = None
+    # additional library search path; this is mainly needed if OMCProcessLocal is run on Windows
+    cmd_library_path: Optional[str] = None
+    # command timeout
+    cmd_timeout: Optional[float] = 10.0
+
+    # working directory to be used on the *local* system
+    cmd_cwd_local: Optional[str] = None
+
+    def get_cmd(self) -> list[str]:
+        """
+        Get the command line to run the model executable in the environment defined by the OMCProcess definition.
+        """
+
+        if self.cmd_model_executable is None:
+            raise OMCSessionException("No model file defined for the model executable!")
+
+        cmdl = [] if self.cmd_prefix is None else self.cmd_prefix
+        cmdl += [self.cmd_model_executable] + self.cmd_args
+
+        return cmdl
+
+
 class OMCSessionZMQ:
 
     def __init__(
@@ -556,15 +600,65 @@ def omcpath_tempdir(self, tempdir_base: Optional[OMCPath] = None) -> OMCPath:
 
         return tempdir
 
+    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)
 
         return self.sendExpression(command, parsed=False)
 
     def sendExpression(self, command: str, parsed: bool = True) -> Any:
+        """
+        Send an expression to the OMC server and return the result.
+        """
         if self.omc_zmq is None:
-            raise OMCSessionException("No OMC running. Create a new instance of OMCSessionZMQ!")
+            raise OMCSessionException("No OMC running. Create a new instance of OMCProcess!")
 
         logger.debug("sendExpression(%r, parsed=%r)", command, parsed)
 
@@ -659,7 +753,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,
@@ -747,6 +841,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):
     """
@@ -760,6 +863,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):
     """
@@ -769,7 +878,7 @@ class OMCProcessLocal(OMCProcess):
     def __init__(
             self,
             timeout: float = 10.00,
-            omhome: Optional[str] = None,
+            omhome: Optional[str | os.PathLike] = None,
     ) -> None:
 
         super().__init__(timeout=timeout)
@@ -782,7 +891,7 @@ def __init__(
         self._omc_port = self._omc_port_get()
 
     @staticmethod
-    def _omc_home_get(omhome: Optional[str] = None) -> pathlib.Path:
+    def _omc_home_get(omhome: Optional[str | os.PathLike] = None) -> pathlib.Path:
         # use the provided path
         if omhome is not None:
             return pathlib.Path(omhome)
@@ -844,6 +953,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):
     """
@@ -854,7 +1005,7 @@ def __init__(
             self,
             timeout: float = 10.00,
             dockerExtraArgs: Optional[list] = None,
-            dockerOpenModelicaPath: str = "omc",
+            dockerOpenModelicaPath: str | os.PathLike = "omc",
             dockerNetwork: Optional[str] = None,
             port: Optional[int] = None,
     ) -> None:
@@ -864,7 +1015,7 @@ def __init__(
             dockerExtraArgs = []
 
         self._dockerExtraArgs = dockerExtraArgs
-        self._dockerOpenModelicaPath = dockerOpenModelicaPath
+        self._dockerOpenModelicaPath = pathlib.PurePosixPath(dockerOpenModelicaPath)
         self._dockerNetwork = dockerNetwork
 
         self._interactivePort = port
@@ -959,6 +1110,28 @@ 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.
+        """
+        omc_run_data_copy = dataclasses.replace(omc_run_data)
+
+        omc_run_data_copy.cmd_prefix = (
+                [
+                    "docker", "exec",
+                    "--user", str(self._getuid()),
+                    "--workdir", omc_run_data_copy.cmd_path,
+                ]
+                + self._dockerExtraArgs
+                + [self._dockerCid]
+        )
+
+        cmd_path = pathlib.PurePosixPath(omc_run_data_copy.cmd_path)
+        cmd_model_executable = cmd_path / omc_run_data_copy.cmd_model_name
+        omc_run_data_copy.cmd_model_executable = cmd_model_executable.as_posix()
+
+        return omc_run_data_copy
+
 
 class OMCProcessDocker(OMCProcessDockerHelper):
     """
@@ -970,7 +1143,7 @@ def __init__(
             timeout: float = 10.00,
             docker: Optional[str] = None,
             dockerExtraArgs: Optional[list] = None,
-            dockerOpenModelicaPath: str = "omc",
+            dockerOpenModelicaPath: str | os.PathLike = "omc",
             dockerNetwork: Optional[str] = None,
             port: Optional[int] = None,
     ) -> None:
@@ -1053,7 +1226,7 @@ def _docker_omc_cmd(
                        ]
                        + self._dockerExtraArgs
                        + dockerNetworkStr
-                       + [self._docker, self._dockerOpenModelicaPath]
+                       + [self._docker, self._dockerOpenModelicaPath.as_posix()]
                        + omc_path_and_args_list
                        + extraFlags)
 
@@ -1113,7 +1286,7 @@ def __init__(
             timeout: float = 10.00,
             dockerContainer: Optional[str] = None,
             dockerExtraArgs: Optional[list] = None,
-            dockerOpenModelicaPath: str = "omc",
+            dockerOpenModelicaPath: str | os.PathLike = "omc",
             dockerNetwork: Optional[str] = None,
             port: Optional[int] = None,
     ) -> None:
@@ -1165,7 +1338,7 @@ def _docker_omc_cmd(self, omc_path_and_args_list) -> list:
                            "--user", str(self._getuid()),
                        ]
                        + self._dockerExtraArgs
-                       + [self._dockerCid, self._dockerOpenModelicaPath]
+                       + [self._dockerCid, self._dockerOpenModelicaPath.as_posix()]
                        + omc_path_and_args_list
                        + extraFlags)
 
@@ -1211,25 +1384,33 @@ def __init__(
 
         super().__init__(timeout=timeout)
 
-        # get wsl base command
-        self._wsl_cmd = ['wsl']
-        if isinstance(wsl_distribution, str):
-            self._wsl_cmd += ['--distribution', wsl_distribution]
-        if isinstance(wsl_user, str):
-            self._wsl_cmd += ['--user', wsl_user]
-        self._wsl_cmd += ['--']
-
         # where to find OpenModelica
         self._wsl_omc = wsl_omc
+        # store WSL distribution and user
+        self._wsl_distribution = wsl_distribution
+        self._wsl_user = wsl_user
         # start up omc executable, which is waiting for the ZMQ connection
         self._omc_process = self._omc_process_get()
         # connect to the running omc instance using ZMQ
         self._omc_port = self._omc_port_get()
 
+    def _wsl_cmd(self, wsl_cwd: Optional[str] = None) -> list[str]:
+        # get wsl base command
+        wsl_cmd = ['wsl']
+        if isinstance(self._wsl_distribution, str):
+            wsl_cmd += ['--distribution', self._wsl_distribution]
+        if isinstance(self._wsl_user, str):
+            wsl_cmd += ['--user', self._wsl_user]
+        if isinstance(wsl_cwd, str):
+            wsl_cmd += ['--cd', wsl_cwd]
+        wsl_cmd += ['--']
+
+        return wsl_cmd
+
     def _omc_process_get(self) -> subprocess.Popen:
         my_env = os.environ.copy()
 
-        omc_command = self._wsl_cmd + [
+        omc_command = self._wsl_cmd() + [
             self._wsl_omc,
             "--locale=C",
             "--interactive=zmq",
@@ -1252,7 +1433,7 @@ def _omc_port_get(self) -> str:
                 omc_portfile_path = self._get_portfile_path()
                 if omc_portfile_path is not None:
                     output = subprocess.check_output(
-                        args=self._wsl_cmd + ["cat", omc_portfile_path.as_posix()],
+                        args=self._wsl_cmd() + ["cat", omc_portfile_path.as_posix()],
                         stderr=subprocess.DEVNULL,
                     )
                     port = output.decode().strip()
@@ -1273,3 +1454,17 @@ 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.
+        """
+        omc_run_data_copy = dataclasses.replace(omc_run_data)
+
+        omc_run_data_copy.cmd_prefix = self._wsl_cmd(wsl_cwd=omc_run_data.cmd_path)
+
+        cmd_path = pathlib.PurePosixPath(omc_run_data_copy.cmd_path)
+        cmd_model_executable = cmd_path / omc_run_data_copy.cmd_model_name
+        omc_run_data_copy.cmd_model_executable = cmd_model_executable.as_posix()
+
+        return omc_run_data_copy