[ModelicaSystemCmd] use OMCPath for file system interactions

syntron · syntron · commit a4c8378a2556 · 2025-07-12T01:06:13.000+02:00
diff --git a/OMPython/ModelicaSystem.py b/OMPython/ModelicaSystem.py
@@ -39,15 +39,13 @@
 import numpy as np
 import os
 import pathlib
-import platform
-import re
 import subprocess
 import textwrap
 from typing import Optional, Any
 import warnings
 import xml.etree.ElementTree as ET
 
-from OMPython.OMCSession import OMCSessionException, OMCSessionZMQ, OMCProcessLocal, OMCPath
+from OMPython.OMCSession import OMCSessionException, OMCSessionZMQ, OMCProcessLocal, OMCPath, OMCSessionRunData
 
 # define logger using the current module name as ID
 logger = logging.getLogger(__name__)
@@ -113,7 +111,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: OMCPath,
+            modelname: str,
+            timeout: Optional[float] = None,
+    ) -> None:
+        self._session = session
         self._runpath = pathlib.Path(runpath).resolve().absolute()
         self._model_name = modelname
         self._timeout = timeout
@@ -174,27 +179,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:
+        """
+        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}")
@@ -203,40 +193,49 @@ def get_cmd(self) -> list:
 
         return cmdl
 
-    def run(self) -> int:
-        """Run the requested simulation.
-
-        Returns
-        -------
-            Subprocess return code (0 on success).
+    def run_def(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())
+        return omc_run_data
 
-        if platform.system() == "Windows":
-            path_dll = ""
+    @staticmethod
+    def run_cmd(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.
+        """
 
-            # 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))
+        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"]
 
-            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
+        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=self._runpath,
-                                    timeout=self._timeout, check=True)
+            cmdres = subprocess.run(
+                cmdl,
+                capture_output=True,
+                text=True,
+                env=my_env,
+                cwd=cmd_run_data.cmd_path,
+                timeout=cmd_run_data.cmd_timeout,
+                check=True,
+            )
             stdout = cmdres.stdout.strip()
             stderr = cmdres.stderr.strip()
             returncode = cmdres.returncode
@@ -252,6 +251,17 @@ def run(self) -> int:
 
         return returncode
 
+    def run(self) -> int:
+        """Run the requested simulation.
+
+        Returns
+        -------
+            Subprocess return code (0 on success).
+        """
+        cmd_run_data = self.run_def()
+        cmd_run_data = self._session.omc_run_data_update(cmd_run_data, session=self._session)
+        return self.run_cmd(cmd_run_data=cmd_run_data)
+
     @staticmethod
     def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, str]]]:
         """
@@ -952,7 +962,8 @@ def simulate_cmd(
         """
 
         om_cmd = ModelicaSystemCmd(
-            runpath=pathlib.Path(self.getWorkDirectory()),
+            session=self._getconn,
+            runpath=self.getWorkDirectory(),
             modelname=self._model_name,
             timeout=timeout,
         )
@@ -1562,7 +1573,8 @@ def linearize(self, lintime: Optional[float] = None, simflags: Optional[str] = N
             )
 
         om_cmd = ModelicaSystemCmd(
-            runpath=pathlib.Path(self.getWorkDirectory()),
+            session=self._getconn,
+            runpath=self.getWorkDirectory(),
             modelname=self._model_name,
             timeout=timeout,
         )
diff --git a/OMPython/OMCSession.py b/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
@@ -431,6 +434,43 @@ def size(self) -> int:
     OMCPath = OMCPathReal
 
 
+@dataclasses.dataclass
+class OMCSessionRunData:
+    """
+    Data class to store the command line data for running a model executable in the OMC environment.
+    """
+    # cmd_path is based on the selected OMCProcess definition
+    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 is based on the selected OMCProcess definition
+    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
+
+    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__(
@@ -521,15 +561,26 @@ def omcpath_tempdir(self) -> OMCPath:
 
         return tempdir
 
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData, session: OMCSessionZMQ) -> 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, session=session)
+
     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)
 
@@ -624,7 +675,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,
@@ -706,6 +757,15 @@ def _get_portfile_path(self) -> Optional[pathlib.Path]:
 
         return portfile_path
 
+    @abc.abstractmethod
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData, session: OMCSessionZMQ) -> OMCSessionRunData:
+        """
+        Modify data 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):
 
@@ -716,6 +776,9 @@ def __init__(
         super().__init__()
         self._omc_port = omc_port
 
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData, session: OMCSessionZMQ) -> OMCSessionRunData:
+        raise OMCSessionException("OMCProcessPort does not support omc_run_data_update()!")
+
 
 class OMCProcessLocal(OMCProcess):
 
@@ -797,6 +860,39 @@ def _omc_port_get(self) -> str:
 
         return port
 
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData, session: OMCSessionZMQ) -> OMCSessionRunData:
+        omc_run_data_copy = dataclasses.replace(omc_run_data)
+
+        cmd_path = session.omcpath(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:
+            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()
+
+        return omc_run_data_copy
+
 
 class OMCProcessDockerHelper(OMCProcess):
 
@@ -903,6 +999,9 @@ def get_docker_container_id(self) -> str:
 
         return self._dockerCid
 
+    def omc_run_data_update(self, omc_run_data: OMCSessionRunData, session: OMCSessionZMQ) -> OMCSessionRunData:
+        raise OMCSessionException("OMCProcessDocker(Container) does not support omc_run_data_update()!")
+
 
 class OMCProcessDocker(OMCProcessDockerHelper):
 
@@ -1208,3 +1307,6 @@ 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, session: OMCSessionZMQ) -> OMCSessionRunData:
+        raise OMCSessionException("OMCProcessWSL does not support omc_run_data_update()!")
diff --git a/tests/test_FMIExport.py b/tests/test_FMIExport.py
@@ -1,12 +1,14 @@
 import OMPython
 import shutil
 import os
+import pathlib
 
 
 def test_CauerLowPassAnalog():
     mod = OMPython.ModelicaSystem(modelName="Modelica.Electrical.Analog.Examples.CauerLowPassAnalog",
                                   lmodel=["Modelica"])
-    tmp = mod.getWorkDirectory()
+    # TODO [OMCPath]: need to work using OMCPath
+    tmp = pathlib.Path(mod.getWorkDirectory())
     try:
         fmu = mod.convertMo2Fmu(fileNamePrefix="CauerLowPassAnalog")
         assert os.path.exists(fmu)
@@ -16,7 +18,8 @@ def test_CauerLowPassAnalog():
 
 def test_DrumBoiler():
     mod = OMPython.ModelicaSystem(modelName="Modelica.Fluid.Examples.DrumBoiler.DrumBoiler", lmodel=["Modelica"])
-    tmp = mod.getWorkDirectory()
+    # TODO [OMCPath]: need to work using OMCPath
+    tmp = pathlib.Path(mod.getWorkDirectory())
     try:
         fmu = mod.convertMo2Fmu(fileNamePrefix="DrumBoiler")
         assert os.path.exists(fmu)
diff --git a/tests/test_ModelicaSystem.py b/tests/test_ModelicaSystem.py
@@ -105,6 +105,7 @@ def test_customBuildDirectory(tmp_path, model_firstorder):
     tmpdir = tmp_path / "tmpdir1"
     tmpdir.mkdir()
     m = OMPython.ModelicaSystem(filePath, "M", customBuildDirectory=tmpdir)
+    # TODO [OMCPath]: need to work using OMCPath
     assert pathlib.Path(m.getWorkDirectory()).resolve() == tmpdir.resolve()
     result_file = tmpdir / "a.mat"
     assert not result_file.exists()
diff --git a/tests/test_ModelicaSystemCmd.py b/tests/test_ModelicaSystemCmd.py
