diff --git a/OMPython/ModelicaSystem.py b/OMPython/ModelicaSystem.py index 1ee8b993..fa0d2ba1 100644 --- a/OMPython/ModelicaSystem.py +++ b/OMPython/ModelicaSystem.py @@ -46,7 +46,7 @@ from dataclasses import dataclass from typing import Optional -from OMPython.OMCSession import OMCSessionBase, OMCSessionZMQ +from OMPython.OMCSession import OMCSessionZMQ # define logger using the current module name as ID logger = logging.getLogger(__name__) @@ -117,7 +117,7 @@ def __init__( variableFilter: Optional[str] = None, customBuildDirectory: Optional[str | os.PathLike] = None, omhome: Optional[str] = None, - session: Optional[OMCSessionBase] = None, + session: Optional[OMCSessionZMQ] = None, build: Optional[bool] = True ): """Initialize, load and build a model. diff --git a/OMPython/OMCSession.py b/OMPython/OMCSession.py index ae3adced..98ae8461 100644 --- a/OMPython/OMCSession.py +++ b/OMPython/OMCSession.py @@ -3,6 +3,8 @@ Definition of an OMC session. """ +from __future__ import annotations + __license__ = """ This file is part of OpenModelica. @@ -33,7 +35,6 @@ """ import shutil -import abc import getpass import logging import json @@ -80,33 +81,15 @@ class OMCSessionException(Exception): pass -class OMCSessionBase(metaclass=abc.ABCMeta): +class OMCSessionCmd: - def __init__(self, readonly=False): + def __init__(self, session: OMCSessionZMQ, readonly: Optional[bool] = False): + if not isinstance(session, OMCSessionZMQ): + raise OMCSessionException("Invalid session definition!") + self._session = session self._readonly = readonly self._omc_cache = {} - 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) - - @abc.abstractmethod - def sendExpression(self, command, parsed=True): - """ - Sends an expression to the OpenModelica. The return type is parsed as if the - expression was part of the typed OpenModelica API (see ModelicaBuiltin.mo). - * Integer and Real are returned as Python numbers - * Strings, enumerations, and typenames are returned as Python strings - * Arrays, tuples, and MetaModelica lists are returned as tuples - * Records are returned as dicts (the name of the record is lost) - * Booleans are returned as True or False - * NONE() is returned as None - * SOME(value) is returned as value - """ - pass - def _ask(self, question: str, opt: Optional[list[str]] = None, parsed: Optional[bool] = True): if opt is None: @@ -114,7 +97,7 @@ def _ask(self, question: str, opt: Optional[list[str]] = None, parsed: Optional[ elif isinstance(opt, list): expression = f"{question}({','.join([str(x) for x in opt])})" else: - raise Exception(f"Invalid definition of options for {repr(question)}: {repr(opt)}") + raise OMCSessionException(f"Invalid definition of options for {repr(question)}: {repr(opt)}") p = (expression, parsed) @@ -126,10 +109,9 @@ def _ask(self, question: str, opt: Optional[list[str]] = None, parsed: Optional[ logger.debug('OMC ask: %s (parsed=%s)', expression, parsed) try: - res = self.sendExpression(expression, parsed=parsed) - except OMCSessionException: - logger.error("OMC failed: %s, %s, parsed=%s", question, opt, parsed) - raise + res = self._session.sendExpression(expression, parsed=parsed) + except OMCSessionException as ex: + raise OMCSessionException("OMC _ask() failed: %s (parsed=%s)", expression, parsed) from ex # save response self._omc_cache[p] = res @@ -201,9 +183,11 @@ def getClassComment(self, className): try: return self._ask(question='getClassComment', opt=[className]) except pyparsing.ParseException as ex: - logger.warning("Method 'getClassComment' failed for %s", className) - logger.warning('OMTypedParser error: %s', ex.msg) + logger.warning("Method 'getClassComment(%s)' failed; OMTypedParser error: %s", + className, ex.msg) return 'No description available' + except OMCSessionException: + raise def getNthComponent(self, className, comp_id): """ returns with (type, name, description) """ @@ -232,13 +216,18 @@ def getParameterNames(self, className): logger.warning('OMPython error: %s', ex) # FIXME: OMC returns with a different structure for empty parameter set return [] + except OMCSessionException: + raise def getParameterValue(self, className, parameterName): try: return self._ask(question='getParameterValue', opt=[className, parameterName]) except pyparsing.ParseException as ex: - logger.warning('OMTypedParser error: %s', ex.msg) + logger.warning("Method 'getParameterValue(%s, %s)' failed; OMTypedParser error: %s", + className, parameterName, ex.msg) return "" + except OMCSessionException: + raise def getComponentModifierNames(self, className, componentName): return self._ask(question='getComponentModifierNames', opt=[className, componentName]) @@ -273,26 +262,22 @@ def getNthComponentModification(self, className, comp_id): # end getClassNames; def getClassNames(self, className=None, recursive=False, qualified=False, sort=False, builtin=False, showProtected=False): - value = self._ask(question='getClassNames', - opt=[className] if className else [] + [f'recursive={str(recursive).lower()}', - f'qualified={str(qualified).lower()}', - f'sort={str(sort).lower()}', - f'builtin={str(builtin).lower()}', - f'showProtected={str(showProtected).lower()}'] - ) - return value + opt = [className] if className else [] + [f'recursive={str(recursive).lower()}', + f'qualified={str(qualified).lower()}', + f'sort={str(sort).lower()}', + f'builtin={str(builtin).lower()}', + f'showProtected={str(showProtected).lower()}'] + return self._ask(question='getClassNames', opt=opt) -class OMCSessionZMQ(OMCSessionBase): +class OMCSessionZMQ: - def __init__(self, readonly=False, timeout=10.00, + def __init__(self, timeout=10.00, docker=None, dockerContainer=None, dockerExtraArgs=None, dockerOpenModelicaPath="omc", dockerNetwork=None, port=None, omhome: str = None): if dockerExtraArgs is None: dockerExtraArgs = [] - super().__init__(readonly=readonly) - self.omhome = self._get_omhome(omhome=omhome) self._omc_process = None @@ -530,11 +515,20 @@ def _connect_to_omc(self, timeout): self._omc.setsockopt(zmq.IMMEDIATE, True) # Queue messages only to completed connections self._omc.connect(self._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!") + attempts = 0 while True: try: diff --git a/OMPython/__init__.py b/OMPython/__init__.py index eee36acc..0d4ab686 100644 --- a/OMPython/__init__.py +++ b/OMPython/__init__.py @@ -36,7 +36,7 @@ CONDITIONS OF OSMC-PL. """ -from OMPython.OMCSession import OMCSessionBase, OMCSessionZMQ, OMCSessionException +from OMPython.OMCSession import OMCSessionCmd, OMCSessionZMQ, OMCSessionException from OMPython.ModelicaSystem import ModelicaSystem, ModelicaSystemError, LinearizationResult # global names imported if import 'from OMPython import *' is used @@ -47,5 +47,5 @@ 'OMCSessionException', 'OMCSessionZMQ', - 'OMCSessionBase', + 'OMCSessionCmd', ] diff --git a/tests/test_OMSessionCmd.py b/tests/test_OMSessionCmd.py new file mode 100644 index 00000000..5e369636 --- /dev/null +++ b/tests/test_OMSessionCmd.py @@ -0,0 +1,24 @@ +import OMPython +import unittest + + +class OMCSessionCmdTester(unittest.TestCase): + def __init__(self, *args, **kwargs): + super(OMCSessionCmdTester, self).__init__(*args, **kwargs) + + def test_isPackage(self): + omczmq = OMPython.OMCSessionZMQ() + omccmd = OMPython.OMCSessionCmd(session=omczmq) + assert not omccmd.isPackage('Modelica') + + def test_isPackage2(self): + mod = OMPython.ModelicaSystem(modelName="Modelica.Electrical.Analog.Examples.CauerLowPassAnalog", + lmodel=["Modelica"]) + omccmd = OMPython.OMCSessionCmd(session=mod.getconn) + assert omccmd.isPackage('Modelica') + + # TODO: add more checks ... + + +if __name__ == '__main__': + unittest.main()