From 35e7cae94cdb92caa225d8faf56e14faaff54cc6 Mon Sep 17 00:00:00 2001 From: syntron Date: Mon, 9 Feb 2026 21:36:03 +0100 Subject: [PATCH 1/3] [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/3] (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. From 06b237a7faa7afca30bc1ea6fade30619025652c Mon Sep 17 00:00:00 2001 From: syntron Date: Mon, 9 Feb 2026 21:27:05 +0100 Subject: [PATCH 3/3] (D003) improve OMCPath [OMPathABC] improve definition * add get_session() * fix return values [(_)OMCPath] improve definition * check return value from OMC * define return value for methods --- OMPython/OMCSession.py | 60 +++++++++++++++++++++++++++--------------- 1 file changed, 39 insertions(+), 21 deletions(-) diff --git a/OMPython/OMCSession.py b/OMPython/OMCSession.py index dd3b2858..79f8d16b 100644 --- a/OMPython/OMCSession.py +++ b/OMPython/OMCSession.py @@ -307,7 +307,13 @@ def __init__(self, *path, session: OMSessionABC) -> None: super().__init__(*path) self._session = session - def with_segments(self, *pathsegments): + def get_session(self) -> OMSessionABC: + """ + Get session definition used for this instance of OMPath. + """ + return self._session + + def with_segments(self, *pathsegments) -> OMPathABC: """ Create a new OMCPath object with the given path segments. @@ -328,7 +334,7 @@ def is_dir(self) -> bool: """ @abc.abstractmethod - def is_absolute(self): + def is_absolute(self) -> bool: """ Check if the path is an absolute path. """ @@ -340,13 +346,13 @@ def read_text(self) -> str: """ @abc.abstractmethod - def write_text(self, data: str): + def write_text(self, data: str) -> int: """ Write text data to the file represented by this path. """ @abc.abstractmethod - def mkdir(self, parents: bool = True, exist_ok: bool = False): + def mkdir(self, parents: bool = True, exist_ok: bool = False) -> None: """ Create a directory at the path represented by this class. @@ -356,7 +362,7 @@ def mkdir(self, parents: bool = True, exist_ok: bool = False): """ @abc.abstractmethod - def cwd(self): + def cwd(self) -> OMPathABC: """ Returns the current working directory as an OMPathABC object. """ @@ -368,12 +374,12 @@ def unlink(self, missing_ok: bool = False) -> None: """ @abc.abstractmethod - def resolve(self, strict: bool = False): + def resolve(self, strict: bool = False) -> OMPathABC: """ Resolve the path to an absolute path. """ - def absolute(self): + def absolute(self) -> OMPathABC: """ Resolve the path to an absolute path. Just a wrapper for resolve(). """ @@ -401,29 +407,38 @@ def is_file(self) -> bool: """ Check if the path is a regular file. """ - return self._session.sendExpression(expr=f'regularFileExists("{self.as_posix()}")') + retval = self.get_session().sendExpression(expr=f'regularFileExists("{self.as_posix()}")') + if not isinstance(retval, bool): + raise OMCSessionException(f"Invalid return value for is_file(): {retval} - expect bool") + return retval def is_dir(self) -> bool: """ Check if the path is a directory. """ - return self._session.sendExpression(expr=f'directoryExists("{self.as_posix()}")') + retval = self.get_session().sendExpression(expr=f'directoryExists("{self.as_posix()}")') + if not isinstance(retval, bool): + raise OMCSessionException(f"Invalid return value for is_dir(): {retval} - expect bool") + return retval - def is_absolute(self): + def is_absolute(self) -> bool: """ - Check if the path is an absolute path. + Check if the path is an absolute path. Special handling to differentiate Windows and Posix definitions. """ if isinstance(self._session, OMCSessionLocal) and platform.system() == 'Windows': return pathlib.PureWindowsPath(self.as_posix()).is_absolute() - return super().is_absolute() + return pathlib.PurePosixPath(self.as_posix()).is_absolute() def read_text(self) -> str: """ Read the content of the file represented by this path as text. """ - return self._session.sendExpression(expr=f'readFile("{self.as_posix()}")') + retval = self.get_session().sendExpression(expr=f'readFile("{self.as_posix()}")') + if not isinstance(retval, str): + raise OMCSessionException(f"Invalid return value for read_text(): {retval} - expect str") + return retval - def write_text(self, data: str): + def write_text(self, data: str) -> int: """ Write text data to the file represented by this path. """ @@ -435,7 +450,7 @@ def write_text(self, data: str): return len(data) - def mkdir(self, parents: bool = True, exist_ok: bool = False): + def mkdir(self, parents: bool = True, exist_ok: bool = False) -> None: """ Create a directory at the path represented by this class. @@ -446,14 +461,15 @@ def mkdir(self, parents: bool = True, exist_ok: bool = False): if self.is_dir() and not exist_ok: raise FileExistsError(f"Directory {self.as_posix()} already exists!") - return self._session.sendExpression(expr=f'mkdir("{self.as_posix()}")') + if not self._session.sendExpression(expr=f'mkdir("{self.as_posix()}")'): + raise OMCSessionException(f"Error on directory creation for {self.as_posix()}!") - def cwd(self): + def cwd(self) -> OMPathABC: """ Returns the current working directory as an OMPathABC object. """ cwd_str = self._session.sendExpression(expr='cd()') - return OMCPath(cwd_str, session=self._session) + return type(self)(cwd_str, session=self._session) def unlink(self, missing_ok: bool = False) -> None: """ @@ -463,7 +479,7 @@ def unlink(self, missing_ok: bool = False) -> None: if not res and not missing_ok: raise FileNotFoundError(f"Cannot delete file {self.as_posix()} - it does not exists!") - def resolve(self, strict: bool = False): + def resolve(self, strict: bool = False) -> OMPathABC: """ Resolve the path to an absolute path. This is done based on available OMC functions. """ @@ -494,8 +510,10 @@ def _omc_resolve(self, pathstr: str) -> str: 'cd(omcpath_cwd)') try: - result = self._session.sendExpression(expr=expr, parsed=False) - result_parts = result.split('\n') + retval = self.get_session().sendExpression(expr=expr, parsed=False) + if not isinstance(retval, str): + raise OMCSessionException(f"Invalid return value for _omc_resolve(): {retval} - expect str") + result_parts = retval.split('\n') pathstr_resolved = result_parts[1] pathstr_resolved = pathstr_resolved[1:-1] # remove quotes except OMCSessionException as ex: