From 4808209d3a62049d890bc92777c90984bc870104 Mon Sep 17 00:00:00 2001 From: syntron Date: Mon, 26 Jan 2026 21:45:46 +0100 Subject: [PATCH] (B004) define OMCSession:OMSessionABC [OMCSession] update OMCSession* to use OMSessionABC as baseline and further cleanup [ModelicaSystem] shortcut to use OMCSession = OMSessionABC for now [ModelicaSystem] fix usage of OMCSession; replace by OMSessionABC fix usage of OMCSession [OMSessionABC] fix OMCPath; rename to OMPathABC --- OMPython/ModelicaSystem.py | 11 ++-- OMPython/OMCSession.py | 119 ++++++++++++++++++++++++++++++++----- OMPython/__init__.py | 4 +- tests/test_OMCPath.py | 2 +- 4 files changed, 114 insertions(+), 22 deletions(-) diff --git a/OMPython/ModelicaSystem.py b/OMPython/ModelicaSystem.py index f2b3adf0..acdd41da 100644 --- a/OMPython/ModelicaSystem.py +++ b/OMPython/ModelicaSystem.py @@ -26,10 +26,11 @@ ModelExecutionException, OMCSessionException, - OMCSession, OMCSessionLocal, OMPathABC, + + OMSessionABC, ) # define logger using the current module name as ID @@ -347,7 +348,7 @@ class ModelicaSystemABC(metaclass=abc.ABCMeta): def __init__( self, - session: OMCSession, + session: OMSessionABC, work_directory: Optional[str | os.PathLike] = None, ) -> None: """Create a ModelicaSystem instance. To define the model use model() or convertFmu2Mo(). @@ -397,7 +398,7 @@ def __init__( self._file_name: Optional[OMPathABC] = None self._variable_filter: Optional[str] = None - def get_session(self) -> OMCSession: + def get_session(self) -> OMSessionABC: """ Return the OMC session used for this class. """ @@ -1502,7 +1503,7 @@ def __init__( command_line_options: Optional[list[str]] = None, work_directory: Optional[str | os.PathLike] = None, omhome: Optional[str] = None, - session: Optional[OMCSession] = None, + session: Optional[OMSessionABC] = None, ) -> None: """Create a ModelicaSystem instance. To define the model use model() or convertFmu2Mo(). @@ -2225,7 +2226,7 @@ def __init__( self._doe_def: Optional[dict[str, dict[str, Any]]] = None self._doe_cmd: Optional[dict[str, ModelExecutionData]] = None - def get_session(self) -> OMCSession: + def get_session(self) -> OMSessionABC: """ Return the OMC session used for this class. """ diff --git a/OMPython/OMCSession.py b/OMPython/OMCSession.py index 242febf0..91115061 100644 --- a/OMPython/OMCSession.py +++ b/OMPython/OMCSession.py @@ -70,8 +70,8 @@ class OMCSessionCmd: Implementation of Open Modelica Compiler API functions. Depreciated! """ - def __init__(self, session: OMCSession, readonly: bool = False): - if not isinstance(session, OMCSession): + def __init__(self, session: OMSessionABC, readonly: bool = False): + if not isinstance(session, OMSessionABC): raise OMCSessionException("Invalid OMC process definition!") self._session = session self._readonly = readonly @@ -301,7 +301,7 @@ class OMPathABC(pathlib.PurePosixPath, metaclass=abc.ABCMeta): limited compared to standard pathlib.Path objects. """ - def __init__(self, *path, session: OMCSession) -> None: + def __init__(self, *path, session: OMSessionABC) -> None: super().__init__(*path) self._session = session @@ -610,7 +610,7 @@ def __init__( self, timeout: float = 10.00, omhome: Optional[str] = None, - omc_process: Optional[OMCSession] = None, + omc_process: Optional[OMCSessionABC] = None, ) -> None: """ Initialisation for OMCSessionZMQ @@ -622,7 +622,7 @@ def __init__( if omc_process is None: omc_process = OMCSessionLocal(omhome=omhome, timeout=timeout) - elif not isinstance(omc_process, OMCSession): + elif not isinstance(omc_process, OMCSessionABC): raise OMCSessionException("Invalid definition of the OMC process!") self.omc_process = omc_process @@ -634,7 +634,7 @@ def escape_str(value: str) -> str: """ Escape a string such that it can be used as string within OMC expressions, i.e. escape all double quotes. """ - return OMCSession.escape_str(value=value) + return OMCSessionABC.escape_str(value=value) def omcpath(self, *path) -> OMPathABC: """ @@ -689,7 +689,7 @@ def __call__(cls, *args, **kwargs): return obj -class OMCSessionMeta(abc.ABCMeta, PostInitCaller): +class OMSessionMeta(abc.ABCMeta, PostInitCaller): """ Helper class to get a combined metaclass of ABCMeta and PostInitCaller. @@ -698,7 +698,98 @@ class OMCSessionMeta(abc.ABCMeta, PostInitCaller): """ -class OMCSession(metaclass=OMCSessionMeta): +class OMSessionABC(metaclass=OMSessionMeta): + """ + This class implements the basic structure a OMPython session definition needs. It provides the structure for an + implementation using OMC as backend (via ZMQ) or a dummy implementation which just runs a model executable. + """ + + def __init__( + self, + timeout: float = 10.00, + **kwargs, + ) -> None: + """ + Initialisation for OMSessionBase + """ + + # some helper data + self.model_execution_windows = platform.system() == "Windows" + self.model_execution_local = False + + # store variables + self._timeout = timeout + + def __post_init__(self) -> None: + """ + Post initialisation method. + """ + + @staticmethod + def escape_str(value: str) -> str: + """ + Escape a string such that it can be used as string within OMC expressions, i.e. escape all double quotes. + """ + return value.replace("\\", "\\\\").replace('"', '\\"') + + @abc.abstractmethod + def model_execution_prefix(self, cwd: Optional[OMPathABC] = None) -> list[str]: + """ + Helper function which returns a command prefix. + """ + + @abc.abstractmethod + def get_version(self) -> str: + """ + Get the OM version. + """ + + @abc.abstractmethod + def set_workdir(self, workdir: OMPathABC) -> None: + """ + Set the workdir for this session. + """ + + @abc.abstractmethod + def omcpath(self, *path) -> OMPathABC: + """ + Create an OMPathBase object based on the given path segments and the current class. + """ + + @abc.abstractmethod + def omcpath_tempdir(self, tempdir_base: Optional[OMPathABC] = None) -> OMPathABC: + """ + Get a temporary directory based on the specific definition for this session. + """ + + @staticmethod + def _tempdir(tempdir_base: OMPathABC) -> OMPathABC: + names = [str(uuid.uuid4()) for _ in range(100)] + + tempdir: Optional[OMPathABC] = None + for name in names: + # create a unique temporary directory name + tempdir = tempdir_base / name + + if tempdir.exists(): + continue + + tempdir.mkdir(parents=True, exist_ok=False) + break + + if tempdir is None or not tempdir.is_dir(): + raise FileNotFoundError(f"Cannot create a temporary directory in {tempdir_base}!") + + return tempdir + + @abc.abstractmethod + def sendExpression(self, expr: str, parsed: bool = True) -> Any: + """ + Function needed to send expressions to the OMC server via ZMQ. + """ + + +class OMCSessionABC(OMSessionABC, metaclass=abc.ABCMeta): """ Base class for an OMC session started via ZMQ. This class contains common functionality for all variants of an OMC session definition. @@ -1104,7 +1195,7 @@ def _get_portfile_path(self) -> Optional[pathlib.Path]: return portfile_path -class OMCSessionPort(OMCSession): +class OMCSessionPort(OMCSessionABC): """ OMCSession implementation which uses a port to connect to an already running OMC server. """ @@ -1117,7 +1208,7 @@ def __init__( self._omc_port = omc_port -class OMCSessionLocal(OMCSession): +class OMCSessionLocal(OMCSessionABC): """ OMCSession implementation which runs the OMC server locally on the machine (Linux / Windows). """ @@ -1198,7 +1289,7 @@ def _omc_port_get(self) -> str: return port -class OMCSessionDockerHelper(OMCSession): +class OMCSessionDockerABC(OMCSessionABC, metaclass=abc.ABCMeta): """ Base class for OMCSession implementations which run the OMC server in a Docker container. """ @@ -1326,7 +1417,7 @@ def model_execution_prefix(self, cwd: Optional[OMPathABC] = None) -> list[str]: return docker_cmd -class OMCSessionDocker(OMCSessionDockerHelper): +class OMCSessionDocker(OMCSessionDockerABC): """ OMC process running in a Docker container. """ @@ -1468,7 +1559,7 @@ def _docker_omc_start(self) -> Tuple[subprocess.Popen, DockerPopen, str]: return omc_process, docker_process, docker_cid -class OMCSessionDockerContainer(OMCSessionDockerHelper): +class OMCSessionDockerContainer(OMCSessionDockerABC): """ OMC process running in a Docker container (by container ID). """ @@ -1561,7 +1652,7 @@ def _docker_omc_start(self) -> Tuple[subprocess.Popen, DockerPopen]: return omc_process, docker_process -class OMCSessionWSL(OMCSession): +class OMCSessionWSL(OMCSessionABC): """ OMC process running in Windows Subsystem for Linux (WSL). """ diff --git a/OMPython/__init__.py b/OMPython/__init__.py index ae47e747..b04db846 100644 --- a/OMPython/__init__.py +++ b/OMPython/__init__.py @@ -26,7 +26,7 @@ OMPathABC, OMCPath, - OMCSession, + OMCSessionABC, ModelExecutionData, ModelExecutionException, @@ -58,7 +58,7 @@ 'OMPathABC', 'OMCPath', - 'OMCSession', + 'OMCSessionABC', 'doe_get_solutions', diff --git a/tests/test_OMCPath.py b/tests/test_OMCPath.py index f4a32eae..df01b86a 100644 --- a/tests/test_OMCPath.py +++ b/tests/test_OMCPath.py @@ -49,7 +49,7 @@ def test_OMCPath_OMCProcessWSL(): del omcs -def _run_OMCPath_checks(omcs: OMPython.OMCSession): +def _run_OMCPath_checks(omcs: OMPython.OMCSessionABC): p1 = omcs.omcpath_tempdir() p2 = p1 / 'test' p2.mkdir()