From 35e7cae94cdb92caa225d8faf56e14faaff54cc6 Mon Sep 17 00:00:00 2001 From: syntron Date: Mon, 9 Feb 2026 21:36:03 +0100 Subject: [PATCH 1/2] [OM(C)SessionABC] small fixes * comments * prepare cmd_prefix handling within OMSession * fix timeout handling --- OMPython/OMCSession.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/OMPython/OMCSession.py b/OMPython/OMCSession.py index 84293746..406e1e76 100644 --- a/OMPython/OMCSession.py +++ b/OMPython/OMCSession.py @@ -809,12 +809,20 @@ def __init__( # store variables self._timeout = timeout + # command prefix (to be used for docker or WSL) + self._cmd_prefix: list[str] = [] def __post_init__(self) -> None: """ Post initialisation method. """ + def get_cmd_prefix(self) -> list[str]: + """ + Get session definition used for this instance of OMPath. + """ + return self._cmd_prefix.copy() + @staticmethod def escape_str(value: str) -> str: """ @@ -843,7 +851,7 @@ def set_workdir(self, workdir: OMPathABC) -> None: @abc.abstractmethod def omcpath(self, *path) -> OMPathABC: """ - Create an OMPathBase object based on the given path segments and the current class. + Create an OMPathABC object based on the given path segments and the current class. """ @abc.abstractmethod @@ -907,13 +915,12 @@ def __init__( """ Initialisation for OMCSession """ + super().__init__(timeout=timeout) # some helper data self.model_execution_windows = platform.system() == "Windows" self.model_execution_local = False - # store variables - self._timeout = timeout # generate a random string for this instance of OMC self._random_string = uuid.uuid4().hex # get a temporary directory @@ -990,6 +997,7 @@ def __del__(self): self._omc_process.kill() self._omc_process.wait() finally: + self._omc_process = None def _timeout_loop( From e2fe29eb9cc73e6df1fe1c5101ee4c9648b0fc08 Mon Sep 17 00:00:00 2001 From: syntron Date: Mon, 9 Feb 2026 21:41:37 +0100 Subject: [PATCH 2/2] (D002) move OMCSessionZMQ [__init__] define OMSessionABC in the public interface [OMCSessionZMQ] move class definition such that it can be derived from OMSessionABC * needed for the compatibility layer --- OMPython/OMCSession.py | 132 ++++++++++++++++++++++------------------- 1 file changed, 71 insertions(+), 61 deletions(-) diff --git a/OMPython/OMCSession.py b/OMPython/OMCSession.py index 406e1e76..dd3b2858 100644 --- a/OMPython/OMCSession.py +++ b/OMPython/OMCSession.py @@ -691,67 +691,6 @@ def run(self) -> int: return returncode -class OMCSessionZMQ: - """ - This class is a compatibility layer for the new schema using OMCSession* classes. - """ - - def __init__( - self, - timeout: float = 10.00, - omhome: Optional[str] = None, - omc_process: Optional[OMCSessionABC] = None, - ) -> None: - """ - Initialisation for OMCSessionZMQ - """ - warnings.warn(message="The class OMCSessionZMQ is depreciated and will be removed in future versions; " - "please use OMCProcess* classes instead!", - category=DeprecationWarning, - stacklevel=2) - - if omc_process is None: - omc_process = OMCSessionLocal(omhome=omhome, timeout=timeout) - elif not isinstance(omc_process, OMCSessionABC): - raise OMCSessionException("Invalid definition of the OMC process!") - self.omc_process = omc_process - - def __del__(self): - del self.omc_process - - @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 OMCSessionABC.escape_str(value=value) - - def omcpath(self, *path) -> OMPathABC: - """ - Create an OMCPath object based on the given path segments and the current OMC process definition. - """ - return self.omc_process.omcpath(*path) - - def omcpath_tempdir(self, tempdir_base: Optional[OMPathABC] = None) -> OMPathABC: - """ - Get a temporary directory using OMC. It is our own implementation as non-local usage relies on OMC to run all - filesystem related access. - """ - return self.omc_process.omcpath_tempdir(tempdir_base=tempdir_base) - - def execute(self, command: str): - return self.omc_process.execute(command=command) - - def sendExpression(self, command: str, parsed: bool = True) -> Any: - """ - Send an expression to the OMC server and return the result. - - The complete error handling of the OMC result is done within this method using '"getMessagesStringInternal()'. - Caller should only check for OMCSessionException. - """ - return self.omc_process.sendExpression(expr=command, parsed=parsed) - - class PostInitCaller(type): """ Metaclass definition to define a new function __post_init__() which is called after all __init__() functions where @@ -1387,6 +1326,77 @@ def _omc_port_get(self) -> str: return port +class OMCSessionZMQ(OMSessionABC): + """ + This class is a compatibility layer for the new schema using OMCSession* classes. + """ + + def __init__( + self, + timeout: float = 10.00, + omhome: Optional[str] = None, + omc_process: Optional[OMCSessionABC] = None, + ) -> None: + """ + Initialisation for OMCSessionZMQ + """ + warnings.warn(message="The class OMCSessionZMQ is depreciated and will be removed in future versions; " + "please use OMCProcess* classes instead!", + category=DeprecationWarning, + stacklevel=2) + + if omc_process is None: + omc_process = OMCSessionLocal(omhome=omhome, timeout=timeout) + elif not isinstance(omc_process, OMCSessionABC): + raise OMCSessionException("Invalid definition of the OMC process!") + self.omc_process = omc_process + + def __del__(self): + if hasattr(self, 'omc_process'): + del self.omc_process + + @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 OMCSessionABC.escape_str(value=value) + + def omcpath(self, *path) -> OMPathABC: + """ + Create an OMCPath object based on the given path segments and the current OMC process definition. + """ + return self.omc_process.omcpath(*path) + + def omcpath_tempdir(self, tempdir_base: Optional[OMPathABC] = None) -> OMPathABC: + """ + Get a temporary directory using OMC. It is our own implementation as non-local usage relies on OMC to run all + filesystem related access. + """ + return self.omc_process.omcpath_tempdir(tempdir_base=tempdir_base) + + def execute(self, command: str): + return self.omc_process.execute(command=command) + + def sendExpression(self, command: str, parsed: bool = True) -> Any: + """ + Send an expression to the OMC server and return the result. + + The complete error handling of the OMC result is done within this method using '"getMessagesStringInternal()'. + Caller should only check for OMCSessionException. + """ + return self.omc_process.sendExpression(expr=command, parsed=parsed) + + def get_version(self) -> str: + return self.omc_process.get_version() + + def model_execution_prefix(self, cwd: Optional[OMPathABC] = None) -> list[str]: + return self.omc_process.model_execution_prefix(cwd=cwd) + + def set_workdir(self, workdir: OMPathABC) -> None: + return self.omc_process.set_workdir(workdir=workdir) + + class OMCSessionDockerABC(OMCSessionABC, metaclass=abc.ABCMeta): """ Base class for OMCSession implementations which run the OMC server in a Docker container.