diff --git a/OMPython/ModelicaSystem.py b/OMPython/ModelicaSystem.py index ae4a5108..61c920b8 100644 --- a/OMPython/ModelicaSystem.py +++ b/OMPython/ModelicaSystem.py @@ -38,15 +38,12 @@ import numbers import numpy as np import os -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, OMCSessionRunData, OMCSessionZMQ, OMCProcessLocal, OMCPath # define logger using the current module name as ID logger = logging.getLogger(__name__) @@ -112,7 +109,14 @@ def __getitem__(self, index: int): class ModelicaSystemCmd: """A compiled model executable.""" - def __init__(self, runpath: OMCPath, 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 = runpath self._model_name = modelname self._timeout = timeout @@ -227,27 +231,12 @@ def args_set( for arg in args: self.arg_set(key=arg, val=args[arg]) - def get_exe(self) -> OMCPath: - """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 sorted(self._args): if self._args[key] is None: cmdl.append(f"-{key}") @@ -256,54 +245,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=omc_run_data) - return returncode + return omc_run_data_updated @staticmethod def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, Any] | numbers.Number]]: @@ -1031,6 +992,7 @@ def simulate_cmd( """ om_cmd = ModelicaSystemCmd( + session=self._getconn, runpath=self.getWorkDirectory(), modelname=self._model_name, timeout=timeout, @@ -1127,7 +1089,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 @@ -1679,6 +1642,7 @@ def linearize( ) om_cmd = ModelicaSystemCmd( + session=self._getconn, runpath=self.getWorkDirectory(), modelname=self._model_name, timeout=timeout, @@ -1716,7 +1680,8 @@ def linearize( linear_file = self.getWorkDirectory() / "linearized_model.py" linear_file.unlink(missing_ok=True) - 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}") if not linear_file.is_file(): diff --git a/OMPython/OMCSession.py b/OMPython/OMCSession.py index 47e2fd74..11716989 100644 --- 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 @@ -456,6 +459,48 @@ class OMCPathCompatibilityWindows(pathlib.WindowsPath, OMCPathCompatibility): OMCPath = OMCPathReal +@dataclasses.dataclass +class OMCSessionRunData: + # TODO: rename OMCExcecutableModelData + """ + 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,6 +601,53 @@ 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) @@ -660,7 +752,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, @@ -748,6 +840,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): """ @@ -761,6 +862,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): """ @@ -845,6 +952,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): """ @@ -960,6 +1109,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): """ @@ -1274,3 +1429,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()!") diff --git a/OMPython/__init__.py b/OMPython/__init__.py index 1da0a0a3..6144f1c2 100644 --- a/OMPython/__init__.py +++ b/OMPython/__init__.py @@ -37,7 +37,7 @@ """ from OMPython.ModelicaSystem import LinearizationResult, ModelicaSystem, ModelicaSystemCmd, ModelicaSystemError -from OMPython.OMCSession import (OMCSessionCmd, OMCSessionException, OMCSessionZMQ, +from OMPython.OMCSession import (OMCSessionCmd, OMCSessionException, OMCSessionRunData, OMCSessionZMQ, OMCProcessPort, OMCProcessLocal, OMCProcessDocker, OMCProcessDockerContainer, OMCProcessWSL) @@ -50,6 +50,7 @@ 'OMCSessionCmd', 'OMCSessionException', + 'OMCSessionRunData', 'OMCSessionZMQ', 'OMCProcessPort', 'OMCProcessLocal', diff --git a/tests/test_ModelicaSystemCmd.py b/tests/test_ModelicaSystemCmd.py index 3544a1bd..844bd8d4 100644 --- a/tests/test_ModelicaSystemCmd.py +++ b/tests/test_ModelicaSystemCmd.py @@ -18,7 +18,11 @@ def model_firstorder(tmp_path): @pytest.fixture def mscmd_firstorder(model_firstorder): mod = OMPython.ModelicaSystem(fileName=model_firstorder.as_posix(), modelName="M") - mscmd = OMPython.ModelicaSystemCmd(runpath=mod.getWorkDirectory(), modelname=mod._model_name) + mscmd = OMPython.ModelicaSystemCmd( + session=mod._getconn, + runpath=mod.getWorkDirectory(), + modelname=mod._model_name, + ) return mscmd @@ -32,8 +36,7 @@ def test_simflags(mscmd_firstorder): with pytest.deprecated_call(): mscmd.args_set(args=mscmd.parse_simflags(simflags="-noEventEmit -noRestart -override=a=1,x=3")) - assert mscmd.get_cmd() == [ - mscmd.get_exe().as_posix(), + assert mscmd.get_cmd_args() == [ '-noEventEmit', '-noRestart', '-override=a=1,b=2,x=3', @@ -43,8 +46,7 @@ def test_simflags(mscmd_firstorder): "override": {'b': None}, }) - assert mscmd.get_cmd() == [ - mscmd.get_exe().as_posix(), + assert mscmd.get_cmd_args() == [ '-noEventEmit', '-noRestart', '-override=a=1,x=3',