Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
132 changes: 71 additions & 61 deletions OMPython/OMCSession.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
Loading