[OMCSessionZMQ] move file to Compat_v400

Author: syntron

OMPython/Compat_v400/__init__.py

@@ -4,7 +4,9 @@
 """
 
 from OMPython.Compat_v400.omc_session_cmd import OMCSessionCmd
+from OMPython.Compat_v400.omc_session_zmq import OMCSessionZMQ
 
 __all__ = [
     'OMCSessionCmd',
+    'OMCSessionZMQ',
 ]

OMPython/Compat_v400/omc_session_zmq.py

@@ -0,0 +1,97 @@
+# -*- coding: utf-8 -*-
+"""
+Definition of class OMSessionZMQ - compatibility class
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Optional
+import warnings
+
+from OMPython.Base import (
+    OMPathABC,
+
+    OMSessionABC,
+    OMSessionException,
+)
+from OMPython.OMC import (
+    OMCSessionABC,
+    OMCSessionLocal,
+)
+
+# define logger using the current module name as ID
+logger = logging.getLogger(__name__)
+
+
+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 OMSessionException("Invalid definition of the OMC process!")
+        self.omc_process = omc_process
+
+        super().__init__(timeout=timeout)
+
+    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)

OMPython/OMCSession.py

@@ -6,17 +6,9 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Optional
-import warnings
 
-from OMPython.Base import (
-    OMSessionException,
-    OMSessionABC,
-    OMPathABC,
-)
 from OMPython.OMC.omc_session import DockerPopen
 from OMPython.OMC import (
-    OMCSessionABC,
     OMCSessionDocker,
     OMCSessionDockerContainer,
     OMCSessionLocal,
@@ -28,79 +20,6 @@
 logger = logging.getLogger(__name__)
 
 
-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 OMSessionException("Invalid definition of the OMC process!")
-        self.omc_process = omc_process
-
-        super().__init__(timeout=timeout)
-
-    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)
-
-
 DummyPopen = DockerPopen
 OMCProcessLocal = OMCSessionLocal
 OMCProcessPort = OMCSessionPort

OMPython/__init__.py

@@ -22,6 +22,7 @@
 )
 from OMPython.Compat_v400 import (
     OMCSessionCmd,
+    OMCSessionZMQ,
 )
 from OMPython.OMC import (
     OMCPath,
@@ -56,9 +57,6 @@
     ModelicaSystemCmd,
 )
 from OMPython.OMCSession import (
-
-    OMCSessionZMQ,
-
     OMCProcessLocal,
     OMCProcessPort,
     OMCProcessDocker,
@@ -80,6 +78,7 @@
     'OMCPath',
 
     'OMCSessionCmd',
+    'OMCSessionZMQ',
 
     'OMCSessionABC',
     'OMCSessionDocker',
@@ -106,8 +105,6 @@
     'OMPathRunnerBash',
     'OMPathRunnerLocal',
 
-    'OMCSessionZMQ',
-
     'OMCProcessLocal',
     'OMCProcessPort',
     'OMCProcessDocker',