Merge branch 'ModelicaSystemCmd_use_OMCPath' into MP_merge

Author: syntron

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
@@ -187,27 +192,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}")
@@ -216,40 +206,51 @@ 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()
+        omc_run_data_updated = self._session.omc_run_data_update(omc_run_data, session=self._session)
 
-        logger.debug("Run OM command %s in %s", repr(cmdl), self._runpath.as_posix())
+        return omc_run_data_updated
 
-        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_cwd_local,
+                timeout=cmd_run_data.cmd_timeout,
+                check=True,
+            )
             stdout = cmdres.stdout.strip()
             stderr = cmdres.stderr.strip()
             returncode = cmdres.returncode
@@ -265,6 +266,16 @@ 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()
+        return self.run_cmd(cmd_run_data=cmd_run_data)
+
     @staticmethod
     def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, str]]]:
         """
@@ -379,8 +390,6 @@ def __init__(
         self._linearized_states: list[str] = []  # linearization states list
 
         if omc_process is not None:
-            if not isinstance(omc_process, OMCProcessLocal):
-                raise ModelicaSystemError("Invalid (local) omc process definition provided!")
             self._getconn = OMCSessionZMQ(omc_process=omc_process)
         else:
             self._getconn = OMCSessionZMQ(omhome=omhome)
@@ -970,7 +979,8 @@ def simulate_cmd(
         """
 
         om_cmd = ModelicaSystemCmd(
-            runpath=pathlib.Path(self.getWorkDirectory()),
+            session=self._getconn,
+            runpath=self.getWorkDirectory(),
             modelname=self._model_name,
             timeout=timeout,
         )
@@ -1581,7 +1591,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,
         )