[OMCSessionZMQ] split class - new: OMCProcess(Dummy|Local)

Author: syntron

OMPython/OMCSession.py

@@ -271,6 +271,148 @@ def getClassNames(self, className=None, recursive=False, qualified=False, sort=F
 
 class OMCSessionZMQ:
 
+    def __init__(self,
+                 timeout: float = 10.00,
+                 omhome: Optional[str] = None,
+                 omc_process: Optional[OMCProcess] = None):
+
+        if omc_process is None:
+            omc_process = OMCProcessLocal(omhome=omhome, timeout=timeout)
+        elif not isinstance(omc_process, OMCProcess):
+            raise OMCSessionException("Invalid definition of the OMC process!")
+        self._omc_process = omc_process
+
+        # variables to store compiled re expressions use in self.sendExpression()
+        self._re_log_entries = None
+        self._re_log_raw = None
+
+        # Create the ZeroMQ socket and connect to OMC server
+        context = zmq.Context.instance()
+        omc = context.socket(zmq.REQ)
+        omc.setsockopt(zmq.LINGER, 0)  # Dismisses pending messages if closed
+        omc.setsockopt(zmq.IMMEDIATE, True)  # Queue messages only to completed connections
+        omc.connect(self._omc_process.get_port())
+
+        self._omc = omc
+
+    def __del__(self):
+        if self._omc:
+            try:
+                self.sendExpression("quit()")
+            except OMCSessionException:
+                pass
+
+            del self._omc
+
+        self._omc = None
+
+    def execute(self, command):
+        warnings.warn("This function is depreciated and will be removed in future versions; "
+                      "please use sendExpression() instead", DeprecationWarning, stacklevel=1)
+
+        return self.sendExpression(command, parsed=False)
+
+    def sendExpression(self, command, parsed=True):
+        p = self._omc_process.poll()  # check if process is running
+        if p is not None:
+            raise OMCSessionException("Process Exited, No connection with OMC. Create a new instance of OMCSessionZMQ!")
+
+        if self._omc is None:
+            raise OMCSessionException("No OMC running. Create a new instance of OMCSessionZMQ!")
+
+        logger.debug("sendExpression(%r, parsed=%r)", command, parsed)
+
+        attempts = 0
+        while True:
+            try:
+                self._omc.send_string(str(command), flags=zmq.NOBLOCK)
+                break
+            except zmq.error.Again:
+                pass
+            attempts += 1
+            if attempts >= 50:
+                self._omc_filehandle_log.seek(0)
+                log = self._omc_filehandle_log.read()
+                self._omc_filehandle_log.close()
+                raise OMCSessionException(f"No connection with OMC (timeout={self._timeout}). Log-file says: \n{log}")
+            time.sleep(self._timeout / 50.0)
+        if command == "quit()":
+            self._omc.close()
+            self._omc = None
+            return None
+        else:
+            result = self._omc.recv_string()
+
+            if command == "getErrorString()":
+                # no error handling if 'getErrorString()' is called
+                pass
+            elif command == "getMessagesStringInternal()":
+                # no error handling if 'getMessagesStringInternal()' is called; parsing NOT possible!
+                if parsed:
+                    logger.warning("Result of 'getMessagesStringInternal()' cannot be parsed - set parsed to False!")
+                    parsed = False
+            else:
+                # allways check for error
+                self._omc.send_string('getMessagesStringInternal()', flags=zmq.NOBLOCK)
+                error_raw = self._omc.recv_string()
+                # run error handling only if there is something to check
+                if error_raw != "{}\n":
+                    if not self._re_log_entries:
+                        self._re_log_entries = re.compile(pattern=r'record OpenModelica\.Scripting\.ErrorMessage'
+                                                                  '(.*?)'
+                                                                  r'end OpenModelica\.Scripting\.ErrorMessage;',
+                                                          flags=re.MULTILINE | re.DOTALL)
+                    if not self._re_log_raw:
+                        self._re_log_raw = re.compile(
+                            pattern=r"\s+message = \"(.*?)\",\n"  # message
+                                    r"\s+kind = .OpenModelica.Scripting.ErrorKind.(.*?),\n"  # kind
+                                    r"\s+level = .OpenModelica.Scripting.ErrorLevel.(.*?),\n"  # level
+                                    r"\s+id = (.*?)"  # id
+                                    "(,\n|\n)",  # end marker
+                            flags=re.MULTILINE | re.DOTALL)
+
+                    # extract all ErrorMessage records
+                    log_entries = self._re_log_entries.findall(string=error_raw)
+                    for log_entry in reversed(log_entries):
+                        log_raw = self._re_log_raw.findall(string=log_entry)
+                        if len(log_raw) != 1 or len(log_raw[0]) != 5:
+                            logger.warning("Invalid ErrorMessage record returned by 'getMessagesStringInternal()':"
+                                           f" {repr(log_entry)}!")
+
+                        log_message = log_raw[0][0].encode().decode('unicode_escape')
+                        log_kind = log_raw[0][1]
+                        log_level = log_raw[0][2]
+                        log_id = log_raw[0][3]
+
+                        msg = (f"[OMC log for 'sendExpression({command}, {parsed})']: "
+                               f"[{log_kind}:{log_level}:{log_id}] {log_message}")
+
+                        # response according to the used log level
+                        # see: https://build.openmodelica.org/Documentation/OpenModelica.Scripting.ErrorLevel.html
+                        if log_level == 'error':
+                            raise OMCSessionException(msg)
+                        elif log_level == 'warning':
+                            logger.warning(msg)
+                        elif log_level == 'notification':
+                            logger.info(msg)
+                        else:  # internal
+                            logger.debug(msg)
+
+            if parsed is True:
+                try:
+                    return om_parser_typed(result)
+                except pyparsing.ParseException as ex:
+                    logger.warning('OMTypedParser error: %s. Returning the basic parser result.', ex.msg)
+                    try:
+                        return om_parser_basic(result)
+                    except (TypeError, UnboundLocalError) as ex:
+                        raise OMCSessionException("Cannot parse OMC result") from ex
+            else:
+                return result
+
+
+class OMCProcess:
+
     def __init__(self,
                  timeout: float = 10.00,
                  omhome: Optional[str] = None):
@@ -279,9 +421,47 @@ def __init__(self,
         self._omhome = self._omc_home_get(omhome=omhome)
         self._timeout = timeout
 
-        # variables to store compiled re expressions use in self.sendExpression()
-        self._re_log_entries = None
-        self._re_log_raw = None
+        self._omc_port: Optional[str] = None
+
+    @staticmethod
+    def _omc_home_get(omhome: Optional[str] = None):
+        # use the provided path
+        if omhome is not None:
+            return pathlib.Path(omhome)
+
+        # check the environment variable
+        omhome = os.environ.get('OPENMODELICAHOME')
+        if omhome is not None:
+            return pathlib.Path(omhome)
+
+        # Get the path to the OMC executable, if not installed this will be None
+        path_to_omc = shutil.which("omc")
+        if path_to_omc is not None:
+            return pathlib.Path(path_to_omc).parents[1]
+
+        raise OMCSessionException("Cannot find OpenModelica executable, please install from openmodelica.org")
+
+    def get_port(self) -> str:
+        if not isinstance(self._omc_port, str):
+            raise OMCSessionException(f"Invalid port to connect to OMC process: {self._omc_port}")
+        return self._omc_port
+
+
+class OMCProcessDummy(OMCProcess):
+
+    def __init__(self,
+                 omc_port: str):
+        super().__init__()
+        self._omc_port = omc_port
+
+
+class OMCProcessLocal(OMCProcess):
+
+    def __init__(self,
+                 timeout: float = 10.00,
+                 omhome: Optional[str] = None):
+
+        super().__init__(timeout=timeout, omhome=omhome)
 
         # generate a random string for this session
         self._random_string = uuid.uuid4().hex
@@ -319,21 +499,7 @@ def __init__(self,
         # connect to the running omc instance using ZMQ
         self._omc_port = self._omc_port_get(timeout)
 
-        # Create the ZeroMQ socket and connect to OMC server
-        context = zmq.Context.instance()
-        omc = context.socket(zmq.REQ)
-        omc.setsockopt(zmq.LINGER, 0)  # Dismisses pending messages if closed
-        omc.setsockopt(zmq.IMMEDIATE, True)  # Queue messages only to completed connections
-        omc.connect(self._omc_port)
-
-        self._omc = omc
-
     def __del__(self):
-        try:
-            self.sendExpression("quit()")
-        except OMCSessionException:
-            pass
-
         if self._omc_filehandle_log is not None:
             self._omc_filehandle_log.close()
 
@@ -379,24 +545,6 @@ def _omc_command_get(self, omc_path_and_args_list) -> list:
 
         return omc_command
 
-    @staticmethod
-    def _omc_home_get(omhome: Optional[str] = None):
-        # use the provided path
-        if omhome is not None:
-            return pathlib.Path(omhome)
-
-        # check the environment variable
-        omhome = os.environ.get('OPENMODELICAHOME')
-        if omhome is not None:
-            return pathlib.Path(omhome)
-
-        # Get the path to the OMC executable, if not installed this will be None
-        path_to_omc = shutil.which("omc")
-        if path_to_omc is not None:
-            return pathlib.Path(path_to_omc).parents[1]
-
-        raise OMCSessionException("Cannot find OpenModelica executable, please install from openmodelica.org")
-
     def _omc_path_get(self) -> pathlib.Path:
         return self._omhome / "bin" / "omc"
 
@@ -438,110 +586,6 @@ def _omc_port_get_main(self) -> Optional[str]:
 
         return port
 
-    def execute(self, command):
-        warnings.warn("This function is depreciated and will be removed in future versions; "
-                      "please use sendExpression() instead", DeprecationWarning, stacklevel=1)
-
-        return self.sendExpression(command, parsed=False)
-
-    def sendExpression(self, command, parsed=True):
-        p = self._omc_process.poll()  # check if process is running
-        if p is not None:
-            raise OMCSessionException("Process Exited, No connection with OMC. Create a new instance of OMCSessionZMQ!")
-
-        if self._omc is None:
-            raise OMCSessionException("No OMC running. Create a new instance of OMCSessionZMQ!")
-
-        logger.debug("sendExpression(%r, parsed=%r)", command, parsed)
-
-        attempts = 0
-        while True:
-            try:
-                self._omc.send_string(str(command), flags=zmq.NOBLOCK)
-                break
-            except zmq.error.Again:
-                pass
-            attempts += 1
-            if attempts >= 50:
-                self._omc_filehandle_log.seek(0)
-                log = self._omc_filehandle_log.read()
-                self._omc_filehandle_log.close()
-                raise OMCSessionException(f"No connection with OMC (timeout={self._timeout}). Log-file says: \n{log}")
-            time.sleep(self._timeout / 50.0)
-        if command == "quit()":
-            self._omc.close()
-            self._omc = None
-            return None
-        else:
-            result = self._omc.recv_string()
-
-            if command == "getErrorString()":
-                # no error handling if 'getErrorString()' is called
-                pass
-            elif command == "getMessagesStringInternal()":
-                # no error handling if 'getMessagesStringInternal()' is called; parsing NOT possible!
-                if parsed:
-                    logger.warning("Result of 'getMessagesStringInternal()' cannot be parsed - set parsed to False!")
-                    parsed = False
-            else:
-                # allways check for error
-                self._omc.send_string('getMessagesStringInternal()', flags=zmq.NOBLOCK)
-                error_raw = self._omc.recv_string()
-                # run error handling only if there is something to check
-                if error_raw != "{}\n":
-                    if not self._re_log_entries:
-                        self._re_log_entries = re.compile(pattern=r'record OpenModelica\.Scripting\.ErrorMessage'
-                                                                  '(.*?)'
-                                                                  r'end OpenModelica\.Scripting\.ErrorMessage;',
-                                                          flags=re.MULTILINE | re.DOTALL)
-                    if not self._re_log_raw:
-                        self._re_log_raw = re.compile(
-                            pattern=r"\s+message = \"(.*?)\",\n"  # message
-                                    r"\s+kind = .OpenModelica.Scripting.ErrorKind.(.*?),\n"  # kind
-                                    r"\s+level = .OpenModelica.Scripting.ErrorLevel.(.*?),\n"  # level
-                                    r"\s+id = (.*?)"  # id
-                                    "(,\n|\n)",  # end marker
-                            flags=re.MULTILINE | re.DOTALL)
-
-                    # extract all ErrorMessage records
-                    log_entries = self._re_log_entries.findall(string=error_raw)
-                    for log_entry in reversed(log_entries):
-                        log_raw = self._re_log_raw.findall(string=log_entry)
-                        if len(log_raw) != 1 or len(log_raw[0]) != 5:
-                            logger.warning("Invalid ErrorMessage record returned by 'getMessagesStringInternal()':"
-                                           f" {repr(log_entry)}!")
-
-                        log_message = log_raw[0][0].encode().decode('unicode_escape')
-                        log_kind = log_raw[0][1]
-                        log_level = log_raw[0][2]
-                        log_id = log_raw[0][3]
-
-                        msg = (f"[OMC log for 'sendExpression({command}, {parsed})']: "
-                               f"[{log_kind}:{log_level}:{log_id}] {log_message}")
-
-                        # response according to the used log level
-                        # see: https://build.openmodelica.org/Documentation/OpenModelica.Scripting.ErrorLevel.html
-                        if log_level == 'error':
-                            raise OMCSessionException(msg)
-                        elif log_level == 'warning':
-                            logger.warning(msg)
-                        elif log_level == 'notification':
-                            logger.info(msg)
-                        else:  # internal
-                            logger.debug(msg)
-
-            if parsed is True:
-                try:
-                    return om_parser_typed(result)
-                except pyparsing.ParseException as ex:
-                    logger.warning('OMTypedParser error: %s. Returning the basic parser result.', ex.msg)
-                    try:
-                        return om_parser_basic(result)
-                    except (TypeError, UnboundLocalError) as ex:
-                        raise OMCSessionException("Cannot parse OMC result") from ex
-            else:
-                return result
-
 
 # noinspection PyPep8Naming
 class OMCSessionZMQDocker(OMCSessionZMQ):