diff --git a/OMPython/ModelicaSystem.py b/OMPython/ModelicaSystem.py index fbaa5876..6888ba43 100644 --- a/OMPython/ModelicaSystem.py +++ b/OMPython/ModelicaSystem.py @@ -34,6 +34,7 @@ import csv import logging +import numbers import os import platform import re @@ -45,8 +46,9 @@ import pathlib from dataclasses import dataclass from typing import Optional +import warnings -from OMPython.OMCSession import OMCSessionBase, OMCSessionZMQ +from OMPython.OMCSession import OMCSessionZMQ, OMCSessionException # define logger using the current module name as ID logger = logging.getLogger(__name__) @@ -107,6 +109,205 @@ def __getitem__(self, index: int): return {0: self.A, 1: self.B, 2: self.C, 3: self.D}[index] +class ModelicaSystemCmd: + """ + Execute a simulation by running the comiled model. + """ + + def __init__(self, runpath: pathlib.Path, modelname: str, timeout: Optional[int] = None) -> None: + """ + Initialisation + + Parameters + ---------- + runpath : pathlib.Path + modelname : str + timeout : Optional[int], None + """ + self._runpath = pathlib.Path(runpath).resolve().absolute() + self._modelname = modelname + self._timeout = timeout + self._args = {} + + def arg_set(self, key: str, val: str | dict = None) -> None: + """ + Set one argument for the executeable model. + + Parameters + ---------- + key : str + val : str, None + """ + if not isinstance(key, str): + raise ModelicaSystemError(f"Invalid argument key: {repr(key)} (type: {type(key)})") + key = key.strip() + if val is None: + argval = None + elif isinstance(val, str): + argval = val.strip() + elif isinstance(val, numbers.Number): + argval = str(val) + elif key == 'override' and isinstance(val, dict): + argval = self._args['override'] if 'override' in self._args else {} + for overwrite_key in val: + if not isinstance(overwrite_key, str) or not isinstance(val[overwrite_key], (str, numbers.Number)): + raise ModelicaSystemError("Invalid argument for 'override': " + f"{repr(overwrite_key)} = {repr(val[overwrite_key])}") + argval[overwrite_key] = val[overwrite_key] + else: + raise ModelicaSystemError(f"Invalid argument value for {repr(key)}: {repr(val)} (type: {type(val)})") + + if key in self._args: + logger.warning(f"Overwrite model executable argument: {repr(key)} = {repr(argval)} " + f"(was: {repr(self._args[key])})") + self._args[key] = argval + + def args_set(self, args: dict) -> None: + """ + Define arguments for the model executable. + + Parameters + ---------- + args : dict + """ + for arg in args: + self.arg_set(key=arg, val=args[arg]) + + def get_exe(self) -> pathlib.Path: + """ + Get the path to the executable / complied model. + + Returns + ------- + pathlib.Path + """ + if platform.system() == "Windows": + path_exe = self._runpath / f"{self._modelname}.exe" + else: + path_exe = self._runpath / self._modelname + + if not path_exe.exists(): + raise ModelicaSystemError(f"Application file path not found: {path_exe}") + + return path_exe + + def get_cmd(self) -> list: + """ + Run the requested simulation + + Returns + ------- + list + """ + + path_exe = self.get_exe() + + cmdl = [path_exe.as_posix()] + for key in self._args: + if self._args[key] is None: + cmdl.append(f"-{key}") + elif key == 'override' and isinstance(self._args[key], dict): + valstr = ','.join([f"{valkey}={str(self._args[key][valkey])}" for valkey in self._args[key]]) + cmdl.append(f"-{key}={valstr}") + else: + cmdl.append(f"-{key}={self._args[key]}") + + return cmdl + + def run(self) -> int: + """ + Run the requested simulation + + Returns + ------- + int + """ + + cmdl: list = self.get_cmd() + + logger.debug("Run OM command %s in %s", repr(cmdl), self._runpath.as_posix()) + + if platform.system() == "Windows": + path_dll = "" + + # set the process environment from the generated .bat file in windows which should have all the dependencies + path_bat = self._runpath / f"{self._modelname}.bat" + if not path_bat.exists(): + ModelicaSystemError("Batch file (*.bat) does not exist " + str(path_bat)) + + with open(path_bat, 'r') as file: + for line in file: + match = re.match(r"^SET PATH=([^%]*)", line, re.IGNORECASE) + if match: + path_dll = match.group(1).strip(';') # Remove any trailing semicolons + my_env = os.environ.copy() + my_env["PATH"] = path_dll + os.pathsep + my_env["PATH"] + else: + # TODO: how to handle path to resources of external libraries for any system not Windows? + my_env = None + + try: + cmdres = subprocess.run(cmdl, capture_output=True, text=True, env=my_env, cwd=self._runpath, + timeout=self._timeout) + stdout = cmdres.stdout.strip() + stderr = cmdres.stderr.strip() + returncode = cmdres.returncode + + logger.debug("OM output for command %s:\n%s", cmdl, stdout) + + if stderr: + raise ModelicaSystemError(f"Error running command {cmdl}: {stderr}") + except subprocess.TimeoutExpired: + raise ModelicaSystemError(f"Timeout running command {repr(cmdl)}") + except Exception as ex: + raise ModelicaSystemError(f"Error running command {cmdl}") from ex + + return returncode + + @staticmethod + def parse_simflags(simflags: str) -> dict: + """ + Parse a simflag definition; this is depreciated! + + Parameters + ---------- + simflags : str + + Returns + ------- + dict + """ + # add old style simulation arguments + warnings.warn("The argument 'simflags' is depreciated and will be removed in future versions; " + "please use 'simargs' instead", DeprecationWarning, stacklevel=2) + + simargs = {} + + args = [s for s in simflags.split(' ') if s] + for arg in args: + if arg[0] != '-': + raise ModelicaSystemError(f"Invalid simulation flag: {arg}") + arg = arg[1:] + parts = arg.split('=') + if len(parts) == 1: + simargs[parts[0]] = None + elif parts[0] == 'override': + override = '='.join(parts[1:]) + + simargs[parts[0]] = {} + for item in override.split(','): + kv = item.split('=') + if not (0 < len(kv) < 3): + raise ModelicaSystemError(f"Invalide value for '-override': {override}") + if kv[0]: + try: + simargs[parts[0]][kv[0]] = kv[1] + except (KeyError, IndexError) as ex: + raise ModelicaSystemError(f"Invalide value for '-override': {override}") from ex + + return simargs + + class ModelicaSystem: def __init__( self, @@ -115,11 +316,9 @@ def __init__( lmodel: Optional[list[str | tuple[str, str]]] = None, commandLineOptions: Optional[str] = None, variableFilter: Optional[str] = None, - customBuildDirectory: Optional[str | os.PathLike] = None, - verbose: bool = True, - raiseerrors: bool = False, + customBuildDirectory: Optional[str | os.PathLike | pathlib.Path] = None, omhome: Optional[str] = None, - session: Optional[OMCSessionBase] = None + session: Optional[OMCSessionZMQ] = None ): """Initialize, load and build a model. @@ -144,9 +343,6 @@ def __init__( customBuildDirectory: Path to a directory to be used for temporary files like the model executable. If left unspecified, a tmp directory will be created. - verbose: If True, enable verbose logging. - raiseerrors: If True, raise exceptions instead of just logging - OpenModelica errors. omhome: OPENMODELICAHOME value to be used when creating the OMC session. session: OMC session to be used. If unspecified, a new session @@ -158,7 +354,7 @@ def __init__( mod = ModelicaSystem("ModelicaModel.mo", "modelName", [("Modelica","3.2.3"), "PowerSystems"]) """ if fileName is None and modelName is None and not lmodel: # all None - raise Exception("Cannot create ModelicaSystem object without any arguments") + raise ModelicaSystemError("Cannot create ModelicaSystem object without any arguments") self.quantitiesList = [] self.paramlist = {} @@ -174,25 +370,23 @@ def __init__( self.linearinputs = [] # linearization input list self.linearoutputs = [] # linearization output list self.linearstates = [] # linearization states list - self.tempdir = "" - - self._verbose = verbose if session is not None: + if not isinstance(session, OMCSessionZMQ): + raise ModelicaSystemError("Invalid session data provided!") self.getconn = session else: self.getconn = OMCSessionZMQ(omhome=omhome) - # needed for properly deleting the session - self._omc_log_file = self.getconn._omc_log_file - self._omc_process = self.getconn._omc_process - # set commandLineOptions if provided by users self.setCommandLineOptions(commandLineOptions=commandLineOptions) if lmodel is None: lmodel = [] + if not isinstance(lmodel, list): + raise ModelicaSystemError(f"Invalid input type for lmodel: {type(lmodel)} - list expected!") + self.xmlFile = None self.lmodel = lmodel # may be needed if model is derived from other model self.modelName = modelName # Model class name @@ -201,13 +395,11 @@ def __init__( self.simulationFlag = False # if the model is simulated? self.outputFlag = False self.csvFile = '' # for storing inputs condition - self.resultfile = "" # for storing result file + self.resultfile = None # for storing result file self.variableFilter = variableFilter - self._raiseerrors = raiseerrors - - if fileName is not None and not self.fileName.is_file(): # if file does not exist - raise IOError(f"File Error: {self.fileName} does not exist!!!") + if self.fileName is not None and not self.fileName.is_file(): # if file does not exist + raise IOError(f"{self.fileName} does not exist!") # set default command Line Options for linearization as # linearize() will use the simulation executable and runtime @@ -215,15 +407,15 @@ def __init__( self.setCommandLineOptions("--linearizationDumpLanguage=python") self.setCommandLineOptions("--generateSymbolicLinearization") - self.setTempDirectory(customBuildDirectory) + self.tempdir = self.setTempDirectory(customBuildDirectory) - if fileName is not None: - self.loadLibrary() - self.loadFile() + if self.fileName is not None: + self.loadLibrary(lmodel=self.lmodel) + self.loadFile(fileName=self.fileName) # allow directly loading models from MSL without fileName - if fileName is None and modelName is not None: - self.loadLibrary() + elif fileName is None and modelName is not None: + self.loadLibrary(lmodel=self.lmodel) self.buildModel(variableFilter) @@ -232,110 +424,55 @@ def setCommandLineOptions(self, commandLineOptions: str): if commandLineOptions is None: return exp = f'setCommandLineOptions("{commandLineOptions}")' - if not self.sendExpression(exp): - self._check_error() + self.sendExpression(exp) - def loadFile(self): + def loadFile(self, fileName: pathlib.Path): # load file - loadMsg = self.sendExpression(f'loadFile("{self.fileName.as_posix()}")') - # Show notification or warnings to the user when verbose=True OR if some error occurred i.e., not result - if self._verbose or not loadMsg: - self._check_error() + self.sendExpression(f'loadFile("{fileName.as_posix()}")') # for loading file/package, loading model and building model - def loadLibrary(self): + def loadLibrary(self, lmodel: list): # load Modelica standard libraries or Modelica files if needed - for element in self.lmodel: + for element in lmodel: if element is not None: if isinstance(element, str): if element.endswith(".mo"): apiCall = "loadFile" else: apiCall = "loadModel" - result = self.requestApi(apiCall, element) + self.requestApi(apiCall, element) elif isinstance(element, tuple): if not element[1]: - libname = f"loadModel({element[0]})" + expr_load_lib = f"loadModel({element[0]})" else: - libname = f'loadModel({element[0]}, {{"{element[1]}"}})' - result = self.sendExpression(libname) + expr_load_lib = f'loadModel({element[0]}, {{"{element[1]}"}})' + self.sendExpression(expr_load_lib) else: raise ModelicaSystemError("loadLibrary() failed, Unknown type detected: " f"{element} is of type {type(element)}, " "The following patterns are supported:\n" '1)["Modelica"]\n' '2)[("Modelica","3.2.3"), "PowerSystems"]\n') - # Show notification or warnings to the user when verbose=True OR if some error occurred i.e., not result - if self._verbose or not result: - self._check_error() - def setTempDirectory(self, customBuildDirectory): + def setTempDirectory(self, customBuildDirectory) -> pathlib.Path: # create a unique temp directory for each session and build the model in that directory if customBuildDirectory is not None: if not os.path.exists(customBuildDirectory): raise IOError(customBuildDirectory, " does not exist") - self.tempdir = customBuildDirectory + tempdir = pathlib.Path(customBuildDirectory) else: - self.tempdir = tempfile.mkdtemp() - if not os.path.exists(self.tempdir): - raise IOError(self.tempdir, " cannot be created") + tempdir = pathlib.Path(tempfile.mkdtemp()) + if not tempdir.is_dir(): + raise IOError(tempdir, " cannot be created") - logger.info("Define tempdir as %s", self.tempdir) - exp = f'cd("{pathlib.Path(self.tempdir).as_posix()}")' + logger.info("Define tempdir as %s", tempdir) + exp = f'cd("{tempdir.absolute().as_posix()}")' self.sendExpression(exp) - def getWorkDirectory(self): - return self.tempdir - - def _run_cmd(self, cmd: list): - logger.debug("Run OM command %s in %s", cmd, self.tempdir) - - if platform.system() == "Windows": - dllPath = "" - - # set the process environment from the generated .bat file in windows which should have all the dependencies - batFilePath = pathlib.Path(self.tempdir) / f"{self.modelName}.bat" - if not batFilePath.exists(): - ModelicaSystemError("Batch file (*.bat) does not exist " + batFilePath) + return tempdir - with open(batFilePath, 'r') as file: - for line in file: - match = re.match(r"^SET PATH=([^%]*)", line, re.IGNORECASE) - if match: - dllPath = match.group(1).strip(';') # Remove any trailing semicolons - my_env = os.environ.copy() - my_env["PATH"] = dllPath + os.pathsep + my_env["PATH"] - else: - # TODO: how to handle path to resources of external libraries for any system not Windows? - my_env = None - - try: - p = subprocess.Popen(cmd, env=my_env, stdout=subprocess.PIPE, - stderr=subprocess.PIPE, cwd=self.tempdir) - stdout, stderr = p.communicate() - - stdout = stdout.decode('ascii').strip() - stderr = stderr.decode('ascii').strip() - if stderr: - raise ModelicaSystemError(f"Error running command {cmd}: {stderr}") - if self._verbose and stdout: - logger.info("OM output for command %s:\n%s", cmd, stdout) - p.wait() - p.terminate() - except Exception as e: - raise ModelicaSystemError(f"Exception {type(e)} running command {cmd}: {e}") - - def _check_error(self): - errstr = self.sendExpression("getErrorString()") - if not errstr: - return - self._raise_error(errstr=errstr) - - def _raise_error(self, errstr: str): - if self._raiseerrors: - raise ModelicaSystemError(f"OM error: {errstr}") - else: - logger.error(errstr) + def getWorkDirectory(self) -> pathlib.Path: + return self.tempdir def buildModel(self, variableFilter=None): if variableFilter is not None: @@ -347,16 +484,20 @@ def buildModel(self, variableFilter=None): varFilter = 'variableFilter=".*"' logger.debug("varFilter=%s", varFilter) buildModelResult = self.requestApi("buildModel", self.modelName, properties=varFilter) - if self._verbose: - logger.info("OM model build result: %s", buildModelResult) - self._check_error() + logger.debug("OM model build result: %s", buildModelResult) self.xmlFile = pathlib.Path(buildModelResult[0]).parent / buildModelResult[1] self.xmlparse() def sendExpression(self, expr, parsed=True): - logger.debug("sendExpression(%r, %r)", expr, parsed) - return self.getconn.sendExpression(expr, parsed) + try: + retval = self.getconn.sendExpression(expr, parsed) + except OMCSessionException as ex: + raise ModelicaSystemError(f"Error executing {repr(expr)}") from ex + + logger.debug(f"Result of executing {repr(expr)}: {repr(retval)}") + + return retval # request to OMC def requestApi(self, apiName, entity=None, properties=None): # 2 @@ -369,17 +510,12 @@ def requestApi(self, apiName, entity=None, properties=None): # 2 exp = f'{apiName}({entity})' else: exp = f'{apiName}()' - try: - res = self.sendExpression(exp) - except Exception as e: - self._raise_error(errstr=f"Exception {type(e)} raised: {e}") - res = None - return res + + return self.sendExpression(exp) def xmlparse(self): if not self.xmlFile.exists(): - self._raise_error(errstr=f"XML file not generated: {self.xmlFile}") - return + ModelicaSystemError(f"XML file not generated: {self.xmlFile}") tree = ET.parse(self.xmlFile) rootCQ = tree.getroot() @@ -395,19 +531,11 @@ def xmlparse(self): scalar["changeable"] = sv.get('isValueChangeable') scalar["aliasvariable"] = sv.get('aliasVariable') ch = list(sv) - start = None - min = None - max = None - unit = None for att in ch: - start = att.get('start') - min = att.get('min') - max = att.get('max') - unit = att.get('unit') - scalar["start"] = start - scalar["min"] = min - scalar["max"] = max - scalar["unit"] = unit + scalar["start"] = att.get('start') + scalar["min"] = att.get('min') + scalar["max"] = att.get('max') + scalar["unit"] = att.get('unit') if scalar["variability"] == "parameter": if scalar["name"] in self.overridevariables: @@ -438,6 +566,8 @@ def getQuantities(self, names=None): # 3 elif isinstance(names, list): return [x for y in names for x in self.quantitiesList if x["name"] == y] + raise ModelicaSystemError("Unhandled input for getQuantities()") + def getContinuous(self, names=None): # 4 """ This method returns dict. The key is continuous names and value is corresponding continuous value. @@ -459,8 +589,8 @@ def getContinuous(self, names=None): # 4 try: value = self.getSolutions(i) self.continuouslist[i] = value[0][-1] - except Exception: - raise ModelicaSystemError(f"OM error: {i} could not be computed") + except OMCSessionException as ex: + raise ModelicaSystemError(f"{i} could not be computed") from ex return self.continuouslist elif isinstance(names, str): @@ -469,7 +599,7 @@ def getContinuous(self, names=None): # 4 self.continuouslist[names] = value[0][-1] return [self.continuouslist.get(names)] else: - raise ModelicaSystemError(f"OM error: {names} is not continuous") + raise ModelicaSystemError(f"{names} is not continuous") elif isinstance(names, list): valuelist = [] @@ -479,9 +609,11 @@ def getContinuous(self, names=None): # 4 self.continuouslist[i] = value[0][-1] valuelist.append(value[0][-1]) else: - raise ModelicaSystemError(f"OM error: {i} is not continuous") + raise ModelicaSystemError(f"{i} is not continuous") return valuelist + raise ModelicaSystemError("Unhandled input for getContinous()") + def getParameters(self, names: Optional[str | list[str]] = None) -> dict[str, str] | list[str]: # 5 """Get parameter values. @@ -509,7 +641,9 @@ def getParameters(self, names: Optional[str | list[str]] = None) -> dict[str, st elif isinstance(names, str): return [self.paramlist.get(names, "NotExist")] elif isinstance(names, list): - return ([self.paramlist.get(x, "NotExist") for x in names]) + return [self.paramlist.get(x, "NotExist") for x in names] + + raise ModelicaSystemError("Unhandled input for getParameters()") def getInputs(self, names: Optional[str | list[str]] = None) -> dict | list: # 6 """Get input values. @@ -543,7 +677,9 @@ def getInputs(self, names: Optional[str | list[str]] = None) -> dict | list: # elif isinstance(names, str): return [self.inputlist.get(names, "NotExist")] elif isinstance(names, list): - return ([self.inputlist.get(x, "NotExist") for x in names]) + return [self.inputlist.get(x, "NotExist") for x in names] + + raise ModelicaSystemError("Unhandled input for getInputs()") def getOutputs(self, names: Optional[str | list[str]] = None): # 7 """Get output values. @@ -588,7 +724,7 @@ def getOutputs(self, names: Optional[str | list[str]] = None): # 7 elif isinstance(names, str): return [self.outputlist.get(names, "NotExist")] else: - return ([self.outputlist.get(x, "NotExist") for x in names]) + return [self.outputlist.get(x, "NotExist") for x in names] else: if names is None: for i in self.outputlist: @@ -601,7 +737,7 @@ def getOutputs(self, names: Optional[str | list[str]] = None): # 7 self.outputlist[names] = value[0][-1] return [self.outputlist.get(names)] else: - return (names, " is not Output") + return names, " is not Output" elif isinstance(names, list): valuelist = [] for i in names: @@ -610,9 +746,11 @@ def getOutputs(self, names: Optional[str | list[str]] = None): # 7 self.outputlist[i] = value[0][-1] valuelist.append(value[0][-1]) else: - return (i, "is not Output") + return i, "is not Output" return valuelist + raise ModelicaSystemError("Unhandled input for getOutputs()") + def getSimulationOptions(self, names=None): # 8 """ This method returns dict. The key is simulation option names and value is corresponding simulation option value. @@ -627,7 +765,9 @@ def getSimulationOptions(self, names=None): # 8 elif isinstance(names, str): return [self.simulateOptions.get(names, "NotExist")] elif isinstance(names, list): - return ([self.simulateOptions.get(x, "NotExist") for x in names]) + return [self.simulateOptions.get(x, "NotExist") for x in names] + + raise ModelicaSystemError("Unhandled input for getSimulationOptions()") def getLinearizationOptions(self, names=None): # 9 """ @@ -643,7 +783,9 @@ def getLinearizationOptions(self, names=None): # 9 elif isinstance(names, str): return [self.linearOptions.get(names, "NotExist")] elif isinstance(names, list): - return ([self.linearOptions.get(x, "NotExist") for x in names]) + return [self.linearOptions.get(x, "NotExist") for x in names] + + raise ModelicaSystemError("Unhandled input for getLinearizationOptions()") def getOptimizationOptions(self, names=None): # 10 """ @@ -657,40 +799,42 @@ def getOptimizationOptions(self, names=None): # 10 elif isinstance(names, str): return [self.optimizeOptions.get(names, "NotExist")] elif isinstance(names, list): - return ([self.optimizeOptions.get(x, "NotExist") for x in names]) + return [self.optimizeOptions.get(x, "NotExist") for x in names] - def get_exe_file(self) -> pathlib.Path: - """Get path to model executable.""" - if platform.system() == "Windows": - return pathlib.Path(self.tempdir) / f"{self.modelName}.exe" - else: - return pathlib.Path(self.tempdir) / self.modelName + raise ModelicaSystemError("Unhandled input for getOptimizationOptions()") - def simulate(self, resultfile=None, simflags=None): # 11 + def simulate(self, resultfile: Optional[str] = None, simflags: Optional[str] = None, + simargs: Optional[dict[str, str | None]] = None, + timeout: Optional[int] = None): # 11 """ This method simulates model according to the simulation options. usage >>> simulate() >>> simulate(resultfile="a.mat") >>> simulate(simflags="-noEventEmit -noRestart -override=e=0.3,g=10") # set runtime simulation flags + >>> simulate(simargs={"noEventEmit": None, "noRestart": None, "override": "e=0.3,g=10"}) # using simargs """ + + om_cmd = ModelicaSystemCmd(runpath=self.tempdir, modelname=self.modelName, timeout=timeout) + if resultfile is None: - r = "" - self.resultfile = (pathlib.Path(self.tempdir) / f"{self.modelName}_res.mat").as_posix() + # default result file generated by OM + self.resultfile = self.tempdir / f"{self.modelName}_res.mat" + elif os.path.exists(resultfile): + self.resultfile = pathlib.Path(resultfile) else: - if os.path.exists(resultfile): - self.resultfile = resultfile - else: - self.resultfile = (pathlib.Path(self.tempdir) / resultfile).as_posix() - r = " -r=" + self.resultfile + self.resultfile = self.tempdir / resultfile + # always define the resultfile to use + om_cmd.arg_set(key="r", val=self.resultfile.as_posix()) # allow runtime simulation flags from user input - if simflags is None: - simflags = "" - else: - simflags = " " + simflags + if simflags is not None: + om_cmd.args_set(args=om_cmd.parse_simflags(simflags=simflags)) + + if simargs: + om_cmd.args_set(args=simargs) - overrideFile = pathlib.Path(self.tempdir) / f"{self.modelName}_override.txt" + overrideFile = self.tempdir / f"{self.modelName}_override.txt" if self.overridevariables or self.simoptionsoverride: tmpdict = self.overridevariables.copy() tmpdict.update(self.simoptionsoverride) @@ -698,9 +842,8 @@ def simulate(self, resultfile=None, simflags=None): # 11 with open(overrideFile, "w") as file: for key, value in tmpdict.items(): file.write(f"{key}={value}\n") - override = " -overrideFile=" + overrideFile.as_posix() - else: - override = "" + + om_cmd.arg_set(key="overrideFile", val=overrideFile.as_posix()) if self.inputFlag: # if model has input quantities for i in self.inputlist: @@ -711,25 +854,22 @@ def simulate(self, resultfile=None, simflags=None): # 11 self.inputlist[i] = [(float(self.simulateOptions["startTime"]), 0.0), (float(self.simulateOptions["stopTime"]), 0.0)] if float(self.simulateOptions["startTime"]) != val[0][0]: - errstr = f"!!! startTime not matched for Input {i}" - self._raise_error(errstr=errstr) - return + raise ModelicaSystemError(f"startTime not matched for Input {i}!") if float(self.simulateOptions["stopTime"]) != val[-1][0]: - errstr = f"!!! stopTime not matched for Input {i}" - self._raise_error(errstr=errstr) - return - self.createCSVData() # create csv file - csvinput = " -csvInput=" + self.csvFile - else: - csvinput = "" + raise ModelicaSystemError(f"stopTime not matched for Input {i}!") + self.csvFile = self.createCSVData() # create csv file + + om_cmd.arg_set(key="csvInput", val=self.csvFile.as_posix()) - exe_file = self.get_exe_file() - if not exe_file.exists(): - raise Exception(f"Error: Application file path not found: {exe_file}") + # delete resultfile ... + if self.resultfile.is_file(): + self.resultfile.unlink() + # ... run simulation ... + returncode = om_cmd.run() + # and check returncode *AND* resultfile + if returncode != 0 and self.resultfile.is_file(): + logger.warning(f"Return code = {returncode} but result file exists!") - cmd = exe_file.as_posix() + override + csvinput + r + simflags - cmd = cmd.split(" ") - self._run_cmd(cmd=cmd) self.simulationFlag = True # to extract simulation results @@ -746,47 +886,47 @@ def getSolutions(self, varList=None, resultfile=None): # 12 >>> getSolutions(["Name1","Name2"],resultfile=""c:/a.mat"") """ if resultfile is None: - resFile = self.resultfile + resFile = self.resultfile.as_posix() else: resFile = resultfile # check for result file exits if not os.path.exists(resFile): - errstr = f"Error: Result file does not exist {resFile}" - self._raise_error(errstr=errstr) - return + raise ModelicaSystemError(f"Result file does not exist {resFile}") resultVars = self.sendExpression(f'readSimulationResultVars("{resFile}")') self.sendExpression("closeSimulationResultFile()") if varList is None: return resultVars elif isinstance(varList, str): if varList not in resultVars and varList != "time": - self._raise_error(errstr=f'!!! {varList} does not exist') - return + raise ModelicaSystemError(f"Requested data {repr(varList)} does not exist") res = self.sendExpression(f'readSimulationResult("{resFile}", {{{varList}}})') npRes = np.array(res) self.sendExpression("closeSimulationResultFile()") return npRes elif isinstance(varList, list): - # varList, = varList - for v in varList: - if v == "time": + for var in varList: + if var == "time": continue - if v not in resultVars: - self._raise_error(errstr=f'!!! {v} does not exist') - return + if var not in resultVars: + raise ModelicaSystemError(f"Requested data {repr(var)} does not exist") variables = ",".join(varList) res = self.sendExpression(f'readSimulationResult("{resFile}",{{{variables}}})') npRes = np.array(res) self.sendExpression("closeSimulationResultFile()") return npRes - def strip_space(self, name): + raise ModelicaSystemError("Unhandled input for getSolutions()") + + @staticmethod + def _strip_space(name): if isinstance(name, str): return name.replace(" ", "") elif isinstance(name, list): return [x.replace(" ", "") for x in name] + raise ModelicaSystemError("Unhandled input for strip_space()") + def setMethodHelper(self, args1, args2, args3, args4=None): """ Helper function for setParameter(),setContinuous(),setSimulationOptions(),setLinearizationOption(),setOptimizationOption() @@ -796,7 +936,7 @@ def setMethodHelper(self, args1, args2, args3, args4=None): args4 - dict() which stores the new override variables list, """ def apply_single(args1): - args1 = self.strip_space(args1) + args1 = self._strip_space(args1) value = args1.split("=") if value[0] in args2: if args3 == "parameter" and self.isParameterChangeable(value[0], value[1]): @@ -811,7 +951,8 @@ def apply_single(args1): return True else: - self._raise_error(errstr=f'"{value[0]}" is not a {args3} variable') + raise ModelicaSystemError("Unhandled case in setMethodHelper.apply_single() - " + f"{repr(value[0])} is not a {repr(args3)} variable") result = [] if isinstance(args1, str): @@ -819,7 +960,7 @@ def apply_single(args1): elif isinstance(args1, list): result = [] - args1 = self.strip_space(args1) + args1 = self._strip_space(args1) for var in args1: result.append(apply_single(var)) @@ -847,13 +988,11 @@ def setParameters(self, pvals): # 14 def isParameterChangeable(self, name, value): q = self.getQuantities(name) - if (q[0]["changeable"] == "false"): - if self._verbose: - logger.info("setParameters() failed : It is not possible to set " - f'the following signal "{name}", It seems to be structural, final, ' - "protected or evaluated or has a non-constant binding, use sendExpression(" - f"setParameterValue({self.modelName}, {name}, {value}), " - "parsed=false) and rebuild the model using buildModel() API") + if q[0]["changeable"] == "false": + logger.verbose(f"setParameters() failed : It is not possible to set the following signal {repr(name)}. " + "It seems to be structural, final, protected or evaluated or has a non-constant binding, " + f"use sendExpression(\"setParameterValue({self.modelName}, {name}, {value})\", " + "parsed=False) and rebuild the model using buildModel() API") return False return True @@ -896,7 +1035,7 @@ def setInputs(self, name): # 15 >>> setInputs(["Name1=value1","Name2=value2"]) """ if isinstance(name, str): - name = self.strip_space(name) + name = self._strip_space(name) value = name.split("=") if value[0] in self.inputlist: tmpvalue = eval(value[1]) @@ -908,24 +1047,22 @@ def setInputs(self, name): # 15 self.inputlist[value[0]] = tmpvalue self.inputFlag = True else: - errstr = value[0] + " is not an input" - self._raise_error(errstr=errstr) + raise ModelicaSystemError(f"{value[0]} is not an input") elif isinstance(name, list): - name = self.strip_space(name) + name = self._strip_space(name) for var in name: value = var.split("=") if value[0] in self.inputlist: tmpvalue = eval(value[1]) - if (isinstance(tmpvalue, int) or isinstance(tmpvalue, float)): + if isinstance(tmpvalue, int) or isinstance(tmpvalue, float): self.inputlist[value[0]] = [(float(self.simulateOptions["startTime"]), float(value[1])), (float(self.simulateOptions["stopTime"]), float(value[1]))] - elif (isinstance(tmpvalue, list)): + elif isinstance(tmpvalue, list): self.checkValidInputs(tmpvalue) self.inputlist[value[0]] = tmpvalue self.inputFlag = True else: - errstr = value[0] + " is not an input" - self._raise_error(errstr=errstr) + raise ModelicaSystemError(f"{value[0]} is not an input!") def checkValidInputs(self, name): if name != sorted(name, key=lambda x: x[0]): @@ -940,7 +1077,7 @@ def checkValidInputs(self, name): else: ModelicaSystemError('Error!!! Value must be in tuple format') - def createCSVData(self) -> None: + def createCSVData(self) -> pathlib.Path: start_time: float = float(self.simulateOptions["startTime"]) stop_time: float = float(self.simulateOptions["stopTime"]) @@ -981,12 +1118,14 @@ def createCSVData(self) -> None: ] csv_rows.append(row) - self.csvFile: str = (pathlib.Path(self.tempdir) / f'{self.modelName}.csv').as_posix() + csvFile = self.tempdir / f'{self.modelName}.csv' - with open(self.csvFile, "w", newline="") as f: + with open(csvFile, "w", newline="") as f: writer = csv.writer(f) writer.writerows(csv_rows) + return csvFile + # to convert Modelica model to FMU def convertMo2Fmu(self, version="2.0", fmuType="me_cs", fileNamePrefix="", includeResources=True): # 19 """ @@ -1009,7 +1148,7 @@ def convertMo2Fmu(self, version="2.0", fmuType="me_cs", fileNamePrefix=" LinearizationResult: + def linearize(self, lintime: Optional[float] = None, simflags: Optional[str] = None, + simargs: Optional[dict[str, str | None]] = None, + timeout: Optional[int] = None) -> LinearizationResult: """Linearize the model according to linearOptions. Args: lintime: Override linearOptions["stopTime"] value. simflags: A string of extra command line flags for the model - binary. + binary. - depreciated in favor of simargs + simargs: A dict with command line flags and possible options; example: "simargs={'csvInput': 'a.csv'}" + timeout: Possible timeout for the execution of OM. Returns: A LinearizationResult object is returned. This allows several @@ -1068,7 +1210,9 @@ def linearize(self, lintime: Optional[float] = None, simflags: Optional[str] = N raise IOError("Linearization cannot be performed as the model is not build, " "use ModelicaSystem() to build the model first") - overrideLinearFile = pathlib.Path(self.tempdir) / f'{self.modelName}_override_linear.txt' + om_cmd = ModelicaSystemCmd(runpath=self.tempdir, modelname=self.modelName, timeout=timeout) + + overrideLinearFile = self.tempdir / f'{self.modelName}_override_linear.txt' with open(overrideLinearFile, "w") as file: for key, value in self.overridevariables.items(): @@ -1076,8 +1220,7 @@ def linearize(self, lintime: Optional[float] = None, simflags: Optional[str] = N for key, value in self.linearOptions.items(): file.write(f"{key}={value}\n") - override = " -overrideFile=" + overrideLinearFile.as_posix() - logger.debug(f"overwrite = {override}") + om_cmd.arg_set(key="overrideFile", val=overrideLinearFile.as_posix()) if self.inputFlag: nameVal = self.getInputs() @@ -1087,38 +1230,29 @@ def linearize(self, lintime: Optional[float] = None, simflags: Optional[str] = N for l in tupleList: if l[0] < float(self.simulateOptions["startTime"]): raise ModelicaSystemError('Input time value is less than simulation startTime') - self.createCSVData() - csvinput = " -csvInput=" + self.csvFile - else: - csvinput = "" + self.csvFile = self.createCSVData() + om_cmd.arg_set(key="csvInput", val=self.csvFile.as_posix()) - # prepare the linearization runtime command - exe_file = self.get_exe_file() + om_cmd.arg_set(key="l", val=str(lintime or self.linearOptions["stopTime"])) - linruntime = f' -l={lintime or self.linearOptions["stopTime"]}' + # allow runtime simulation flags from user input + if simflags is not None: + om_cmd.args_set(args=om_cmd.parse_simflags(simflags=simflags)) - if simflags is None: - simflags = "" - else: - simflags = " " + simflags + if simargs: + om_cmd.args_set(args=simargs) - if not exe_file.exists(): - raise Exception(f"Error: Application file path not found: {exe_file}") - else: - cmd = exe_file.as_posix() + linruntime + override + csvinput + simflags - cmd = cmd.split(' ') - self._run_cmd(cmd=cmd) + self.simulationFlag = om_cmd.run() # code to get the matrix and linear inputs, outputs and states - linearFile = pathlib.Path(self.tempdir) / "linearized_model.py" + linearFile = self.tempdir / "linearized_model.py" # support older openmodelica versions before OpenModelica v1.16.2 where linearize() generates "linear_modelname.mo" file if not linearFile.exists(): linearFile = pathlib.Path(f'linear_{self.modelName}.py') if not linearFile.exists(): - errormsg = self.sendExpression("getErrorString()") - raise ModelicaSystemError(f"Linearization failed: {linearFile} not found: {errormsg}") + raise ModelicaSystemError(f"Linearization failed: {linearFile} not found!") # this function is called from the generated python code linearized_model.py at runtime, # to improve the performance by directly reading the matrices A, B, C and D from the julia code and avoid building the linearized modelica model @@ -1133,8 +1267,8 @@ def linearize(self, lintime: Optional[float] = None, simflags: Optional[str] = N self.linearstates = stateVars return LinearizationResult(n, m, p, A, B, C, D, x0, u0, stateVars, inputVars, outputVars) - except ModuleNotFoundError: - raise Exception("ModuleNotFoundError: No module named 'linearized_model'") + except ModuleNotFoundError as ex: + raise ModelicaSystemError("No module named 'linearized_model'") from ex def getLinearInputs(self): """ diff --git a/OMPython/OMCSession.py b/OMPython/OMCSession.py index 63fbba00..90e60dae 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 @@ -45,21 +46,22 @@ import sys import tempfile import time +from typing import Optional import uuid import pyparsing import zmq import warnings # TODO: replace this with the new parser -from OMPython import OMTypedParser -from OMPython import OMParser +from OMPython.OMTypedParser import parseString as om_parser_typed +from OMPython.OMParser import om_parser_basic # define logger using the current module name as ID logger = logging.getLogger(__name__) -class DummyPopen(): +class DummyPopen: def __init__(self, pid): self.pid = pid self.process = psutil.Process(pid) @@ -75,154 +77,158 @@ def wait(self, timeout): return self.process.wait(timeout=timeout) -class OMCSessionBase(metaclass=abc.ABCMeta): +class OMCSessionException(Exception): + pass + - def __init__(self, readonly=False): +class OMCSessionCmd: + + 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 clearOMParserResult(self): - OMParser.result = {} - - def execute(self, command): - warnings.warn("This function is depreciated and will be removed in future versions; " - "please use sendExpression() instead", DeprecationWarning, stacklevel=1) + def sendExpression(self, command, parsed=True): + return self._session.sendExpression(command=command, parsed=parsed) - return self.sendExpression(command, parsed=False) + def _ask(self, question: str, opt: Optional[list[str]] = None, parsed: Optional[bool] = True): - @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 + if opt is None: + expression = question + elif isinstance(opt, list): + expression = f"{question}({','.join(opt)})" + else: + raise OMCSessionException(f"Invalid definition of options for {repr(question)}: {repr(opt)}") - def ask(self, question, opt=None, parsed=True): - p = (question, opt, parsed) + p = (expression, parsed) if self._readonly and question != 'getErrorString': # can use cache if readonly if p in self._omc_cache: return self._omc_cache[p] - if opt: - expression = f'{question}({opt})' - else: - expression = question - - logger.debug('OMC ask: %s - parsed: %s', expression, parsed) - try: - res = self.sendExpression(expression, parsed=parsed) - except Exception: - 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 return res + def _ask_with_fallback(self, question: str, opt: Optional[list[str]] = None): + """ + Version of ask() with OMTypedParser as fallback for OMParser which is used in ask() => sendExpression() if + parsed is set to True. + """ + try: + # FIXME: OMPython exception UnboundLocalError exception for 'Modelica.Fluid.Machines.ControlledPump' + return self._ask(question=question, opt=opt) + except pyparsing.ParseException as ex: + logger.warning('OMTypedParser error: %s', ex.msg) + result = self._ask(question=question, opt=opt, parsed=False) + try: + answer = om_parser_basic(result) + return answer[2:] + except (TypeError, UnboundLocalError) as ex: + logger.warning('OMParser error: %s', ex) + return result + # TODO: Open Modelica Compiler API functions. Would be nice to generate these. def loadFile(self, filename): - return self.ask('loadFile', f'"{filename}"') + return self._ask(question='loadFile', opt=[f'"{filename}"']) def loadModel(self, className): - return self.ask('loadModel', className) + return self._ask(question='loadModel', opt=[className]) def isModel(self, className): - return self.ask('isModel', className) + return self._ask(question='isModel', opt=[className]) def isPackage(self, className): - return self.ask('isPackage', className) + return self._ask(question='isPackage', opt=[className]) def isPrimitive(self, className): - return self.ask('isPrimitive', className) + return self._ask(question='isPrimitive', opt=[className]) def isConnector(self, className): - return self.ask('isConnector', className) + return self._ask(question='isConnector', opt=[className]) def isRecord(self, className): - return self.ask('isRecord', className) + return self._ask(question='isRecord', opt=[className]) def isBlock(self, className): - return self.ask('isBlock', className) + return self._ask(question='isBlock', opt=[className]) def isType(self, className): - return self.ask('isType', className) + return self._ask(question='isType', opt=[className]) def isFunction(self, className): - return self.ask('isFunction', className) + return self._ask(question='isFunction', opt=[className]) def isClass(self, className): - return self.ask('isClass', className) + return self._ask(question='isClass', opt=[className]) def isParameter(self, className): - return self.ask('isParameter', className) + return self._ask(question='isParameter', opt=[className]) def isConstant(self, className): - return self.ask('isConstant', className) + return self._ask(question='isConstant', opt=[className]) def isProtected(self, className): - return self.ask('isProtected', className) + return self._ask(question='isProtected', opt=[className]) def getPackages(self, className="AllLoadedClasses"): - return self.ask('getPackages', className) + return self._ask(question='getPackages', opt=[className]) def getClassRestriction(self, className): - return self.ask('getClassRestriction', className) + return self._ask(question='getClassRestriction', opt=[className]) def getDerivedClassModifierNames(self, className): - return self.ask('getDerivedClassModifierNames', className) + return self._ask(question='getDerivedClassModifierNames', opt=[className]) def getDerivedClassModifierValue(self, className, modifierName): - return self.ask('getDerivedClassModifierValue', f'{className}, {modifierName}') + return self._ask(question='getDerivedClassModifierValue', opt=[className, modifierName]) def typeNameStrings(self, className): - return self.ask('typeNameStrings', className) + return self._ask(question='typeNameStrings', opt=[className]) def getComponents(self, className): - return self.ask('getComponents', className) + return self._ask(question='getComponents', opt=[className]) def getClassComment(self, className): try: - return self.ask('getClassComment', className) + 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.message) + logger.warning('OMTypedParser error: %s', ex.msg) return 'No description available' def getNthComponent(self, className, comp_id): """ returns with (type, name, description) """ - return self.ask('getNthComponent', f'{className}, {comp_id}') + return self._ask(question='getNthComponent', opt=[className, comp_id]) def getNthComponentAnnotation(self, className, comp_id): - return self.ask('getNthComponentAnnotation', f'{className}, {comp_id}') + return self._ask(question='getNthComponentAnnotation', opt=[className, comp_id]) def getImportCount(self, className): - return self.ask('getImportCount', className) + return self._ask(question='getImportCount', opt=[className]) def getNthImport(self, className, importNumber): # [Path, id, kind] - return self.ask('getNthImport', f'{className}, {importNumber}') + return self._ask(question='getNthImport', opt=[className, importNumber]) def getInheritanceCount(self, className): - return self.ask('getInheritanceCount', className) + return self._ask(question='getInheritanceCount', opt=[className]) def getNthInheritedClass(self, className, inheritanceDepth): - return self.ask('getNthInheritedClass', f'{className}, {inheritanceDepth}') + return self._ask(question='getNthInheritedClass', opt=[className, inheritanceDepth]) def getParameterNames(self, className): try: - return self.ask('getParameterNames', className) + return self._ask(question='getParameterNames', opt=[className]) except KeyError as ex: logger.warning('OMPython error: %s', ex) # FIXME: OMC returns with a different structure for empty parameter set @@ -230,53 +236,29 @@ def getParameterNames(self, className): def getParameterValue(self, className, parameterName): try: - return self.ask('getParameterValue', f'{className}, {parameterName}') + return self._ask(question='getParameterValue', opt=[className, parameterName]) except pyparsing.ParseException as ex: - logger.warning('OMTypedParser error: %s', ex.message) + logger.warning('OMTypedParser error: %s', ex.msg) return "" def getComponentModifierNames(self, className, componentName): - return self.ask('getComponentModifierNames', f'{className}, {componentName}') + return self._ask(question='getComponentModifierNames', opt=[className, componentName]) def getComponentModifierValue(self, className, componentName): - try: - # FIXME: OMPython exception UnboundLocalError exception for 'Modelica.Fluid.Machines.ControlledPump' - return self.ask('getComponentModifierValue', f'{className}, {componentName}') - except pyparsing.ParseException as ex: - logger.warning('OMTypedParser error: %s', ex.message) - result = self.ask('getComponentModifierValue', f'{className}, {componentName}', parsed=False) - try: - answer = OMParser.check_for_values(result) - OMParser.result = {} - return answer[2:] - except (TypeError, UnboundLocalError) as ex: - logger.warning('OMParser error: %s', ex) - return result + return self._ask_with_fallback(question='getComponentModifierValue', opt=[className, componentName]) def getExtendsModifierNames(self, className, componentName): - return self.ask('getExtendsModifierNames', f'{className}, {componentName}') + return self._ask(question='getExtendsModifierNames', opt=[className, componentName]) def getExtendsModifierValue(self, className, extendsName, modifierName): - try: - # FIXME: OMPython exception UnboundLocalError exception for 'Modelica.Fluid.Machines.ControlledPump' - return self.ask('getExtendsModifierValue', f'{className}, {extendsName}, {modifierName}') - except pyparsing.ParseException as ex: - logger.warning('OMTypedParser error: %s', ex.message) - result = self.ask('getExtendsModifierValue', f'{className}, {extendsName}, {modifierName}', parsed=False) - try: - answer = OMParser.check_for_values(result) - OMParser.result = {} - return answer[2:] - except (TypeError, UnboundLocalError) as ex: - logger.warning('OMParser error: %s', ex) - return result + return self._ask_with_fallback(question='getExtendsModifierValue', opt=[className, extendsName, modifierName]) def getNthComponentModification(self, className, comp_id): # FIXME: OMPython exception Results KeyError exception # get {$Code(....)} field # \{\$Code\((\S*\s*)*\)\} - value = self.ask('getNthComponentModification', f'{className}, {comp_id}', parsed=False) + value = self._ask(question='getNthComponentModification', opt=[className, comp_id], parsed=False) value = value.replace("{$Code(", "") return value[:-3] # return self.re_Code.findall(value) @@ -292,28 +274,24 @@ 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( - 'getClassNames', - (f'{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()}' - ) + 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 -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 @@ -323,7 +301,7 @@ def __init__(self, readonly=False, timeout=10.00, self._serverIPAddress = "127.0.0.1" self._interactivePort = None # FIXME: this code is not well written... need to be refactored - self._temp_dir = tempfile.gettempdir() + self._temp_dir = pathlib.Path(tempfile.gettempdir()) # generate a random string for this session self._random_string = uuid.uuid4().hex # omc log file @@ -348,7 +326,7 @@ def __init__(self, readonly=False, timeout=10.00, self._dockerNetwork = dockerNetwork self._create_omc_log_file("port") self._timeout = timeout - self._port_file = os.path.join("/tmp" if docker else self._temp_dir, self._port_file).replace("\\", "/") + self._port_file = ((pathlib.Path("/tmp") if docker else self._temp_dir) / self._port_file).as_posix() self._interactivePort = port # set omc executable path and args self._set_omc_command([ @@ -364,14 +342,15 @@ def __init__(self, readonly=False, timeout=10.00, def __del__(self): try: self.sendExpression("quit()") - except Exception: + except OMCSessionException: pass self._omc_log_file.close() try: self._omc_process.wait(timeout=2.0) - except Exception: + except subprocess.TimeoutExpired: if self._omc_process: - logger.warning("OMC did not exit after being sent the quit() command; killing the process with pid=%s", self._omc_process.pid) + logger.warning("OMC did not exit after being sent the quit() command; " + "killing the process with pid=%s", self._omc_process.pid) self._omc_process.kill() self._omc_process.wait() @@ -381,7 +360,7 @@ def _create_omc_log_file(self, suffix): else: log_filename = f"openmodelica.{self._currentUser}.{suffix}.{self._random_string}.log" # this file must be closed in the destructor - self._omc_log_file = open(pathlib.Path(self._temp_dir) / log_filename, "w+") + self._omc_log_file = open(self._temp_dir / log_filename, "w+") def _start_omc_process(self, timeout): if sys.platform == 'win32': @@ -401,18 +380,19 @@ def _start_omc_process(self, timeout): try: with open(self._dockerCidFile, "r") as fin: self._dockerCid = fin.read().strip() - except Exception: + except IOError: pass if self._dockerCid: break time.sleep(timeout / 40.0) try: os.remove(self._dockerCidFile) - except Exception: + except FileNotFoundError: pass if self._dockerCid is None: logger.error("Docker did not start. Log-file says:\n%s" % (open(self._omc_log_file.name).read())) - raise Exception("Docker did not start (timeout=%f might be too short especially if you did not docker pull the image before this command)." % timeout) + raise OMCSessionException("Docker did not start (timeout=%f might be too short especially if you did " + "not docker pull the image before this command)." % timeout) dockerTop = None if self._docker or self._dockerContainer: @@ -429,17 +409,16 @@ def _start_omc_process(self, timeout): try: self._omc_process = DummyPopen(int(columns[1])) except psutil.NoSuchProcess: - raise Exception( - f"Could not find PID {dockerTop} - is this a docker instance spawned without --pid=host?\n" - f"Log-file says:\n{open(self._omc_log_file.name).read()}") + raise OMCSessionException( + f"Could not find PID {dockerTop} - is this a docker instance spawned " + f"without --pid=host?\nLog-file says:\n{open(self._omc_log_file.name).read()}") break if self._omc_process is not None: break time.sleep(timeout / 40.0) if self._omc_process is None: - - raise Exception("Docker top did not contain omc process %s:\n%s\nLog-file says:\n%s" - % (self._random_string, dockerTop, open(self._omc_log_file.name).read())) + raise OMCSessionException("Docker top did not contain omc process %s:\n%s\nLog-file says:\n%s" + % (self._random_string, dockerTop, open(self._omc_log_file.name).read())) return self._omc_process def _getuid(self): @@ -460,7 +439,9 @@ def _set_omc_command(self, omc_path_and_args_list): if (self._docker or self._dockerContainer) and sys.platform == "win32": extraFlags = ["-d=zmqDangerousAcceptConnectionsFromAnywhere"] if not self._interactivePort: - raise Exception("docker on Windows requires knowing which port to connect to. For dockerContainer=..., the container needs to have already manually exposed this port when it was started (-p 127.0.0.1:n:n) or you get an error later.") + raise OMCSessionException("docker on Windows requires knowing which port to connect to. For " + "dockerContainer=..., the container needs to have already manually exposed " + "this port when it was started (-p 127.0.0.1:n:n) or you get an error later.") else: extraFlags = [] if self._docker: @@ -473,7 +454,7 @@ def _set_omc_command(self, omc_path_and_args_list): dockerNetworkStr = [] extraFlags = ["-d=zmqDangerousAcceptConnectionsFromAnywhere"] else: - raise Exception('dockerNetwork was set to %s, but only \"host\" or \"separate\" is allowed') + raise OMCSessionException('dockerNetwork was set to %s, but only \"host\" or \"separate\" is allowed') self._dockerCidFile = self._omc_log_file.name + ".docker.cid" omcCommand = ["docker", "run", "--cidfile", self._dockerCidFile, "--rm", "--env", "USER=%s" % self._currentUser, "--user", str(self._getuid())] + self._dockerExtraArgs + dockerNetworkStr + [self._docker, self._dockerOpenModelicaPath] elif self._dockerContainer: @@ -503,7 +484,7 @@ def _get_omhome(self, omhome: str = None): if path_to_omc is not None: return pathlib.Path(path_to_omc).parents[1] - raise ValueError("Cannot find OpenModelica executable, please install from openmodelica.org") + raise OMCSessionException("Cannot find OpenModelica executable, please install from openmodelica.org") def _get_omc_path(self) -> pathlib.Path: return self.omhome / "bin" / "omc" @@ -516,9 +497,10 @@ def _connect_to_omc(self, timeout): while True: if self._dockerCid: try: - self._port = subprocess.check_output(["docker", "exec", self._dockerCid, "cat", self._port_file], stderr=subprocess.DEVNULL).decode().strip() + self._port = subprocess.check_output(["docker", "exec", self._dockerCid, "cat", self._port_file], + stderr=subprocess.DEVNULL).decode().strip() break - except Exception: + except subprocess.CalledProcessError: pass else: if os.path.isfile(self._port_file): @@ -533,7 +515,8 @@ def _connect_to_omc(self, timeout): name = self._omc_log_file.name self._omc_log_file.close() logger.error("OMC Server did not start. Please start it! Log-file says:\n%s" % open(name).read()) - raise Exception(f"OMC Server did not start (timeout={timeout}). Could not open file {self._port_file}") + raise OMCSessionException(f"OMC Server did not start (timeout={timeout}). " + "Could not open file {self._port_file}") time.sleep(timeout / 80.0) self._port = self._port.replace("0.0.0.0", self._serverIPAddress) @@ -546,10 +529,18 @@ 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 Exception("Process Exited, No connection with OMC. Create a new instance of OMCSessionZMQ") + raise OMCSessionException("Process Exited, No connection with OMC. Create a new instance of OMCSessionZMQ!") + + logger.debug("sendExpression(%r, parsed=%r)", command, parsed) attempts = 0 while True: @@ -563,7 +554,7 @@ def sendExpression(self, command, parsed=True): self._omc_log_file.seek(0) log = self._omc_log_file.read() self._omc_log_file.close() - raise Exception(f"No connection with OMC (timeout={self._timeout}). Log-file says: \n{log}") + 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() @@ -571,8 +562,19 @@ def sendExpression(self, command, parsed=True): return None else: result = self._omc.recv_string() + + # allways check for error + self._omc.send_string("getErrorString()", flags=zmq.NOBLOCK) + error_raw = self._omc.recv_string() + error_str = om_parser_typed(error_raw) + if error_str: + if "Error" in error_str: + raise OMCSessionException(f"OM Error for 'sendExpression({command}, {parsed})': {error_str}") + else: + logger.warning(f"[OM]: {error_str}") + if parsed is True: - answer = OMTypedParser.parseString(result) + answer = om_parser_typed(result) return answer else: return result diff --git a/OMPython/OMParser.py b/OMPython/OMParser.py index f1708947..c9bb1796 100644 --- a/OMPython/OMParser.py +++ b/OMPython/OMParser.py @@ -892,3 +892,15 @@ def check_for_values(string): check_for_values(next_set) return result + + +# TODO: hack to be able to use one entry point wich also resets the (global) variable results +# this should be checked such that the content of this file can be used as class with correct handling of +# variable usage +def om_parser_basic(string: str): + result_return = check_for_values(string=string) + + global result + result = {} + + return result_return diff --git a/OMPython/__init__.py b/OMPython/__init__.py index 29eaca99..53368a03 100644 --- a/OMPython/__init__.py +++ b/OMPython/__init__.py @@ -36,15 +36,17 @@ CONDITIONS OF OSMC-PL. """ -from OMPython.OMCSession import OMCSessionBase, OMCSessionZMQ -from OMPython.ModelicaSystem import ModelicaSystem, ModelicaSystemError, LinearizationResult +from OMPython.OMCSession import OMCSessionCmd, OMCSessionZMQ, OMCSessionException +from OMPython.ModelicaSystem import ModelicaSystem, ModelicaSystemCmd, ModelicaSystemError, LinearizationResult # global names imported if import 'from OMPython import *' is used __all__ = [ 'ModelicaSystem', + 'ModelicaSystemCmd', 'ModelicaSystemError', 'LinearizationResult', + 'OMCSessionException', 'OMCSessionZMQ', - 'OMCSessionBase', + 'OMCSessionCmd', ] diff --git a/tests/test_FMIExport.py b/tests/test_FMIExport.py index 15e94227..0d6d0ff9 100644 --- a/tests/test_FMIExport.py +++ b/tests/test_FMIExport.py @@ -14,7 +14,8 @@ def __del__(self): def testCauerLowPassAnalog(self): print("testing Cauer") - mod = OMPython.ModelicaSystem(modelName="Modelica.Electrical.Analog.Examples.CauerLowPassAnalog", lmodel="Modelica") + mod = OMPython.ModelicaSystem(modelName="Modelica.Electrical.Analog.Examples.CauerLowPassAnalog", + lmodel=["Modelica"]) self.tmp = mod.getWorkDirectory() fmu = mod.convertMo2Fmu(fileNamePrefix="CauerLowPassAnalog") @@ -22,7 +23,7 @@ def testCauerLowPassAnalog(self): def testDrumBoiler(self): print("testing DrumBoiler") - mod = OMPython.ModelicaSystem(modelName="Modelica.Fluid.Examples.DrumBoiler.DrumBoiler", lmodel="Modelica") + mod = OMPython.ModelicaSystem(modelName="Modelica.Fluid.Examples.DrumBoiler.DrumBoiler", lmodel=["Modelica"]) self.tmp = mod.getWorkDirectory() fmu = mod.convertMo2Fmu(fileNamePrefix="DrumBoiler") diff --git a/tests/test_ModelicaSystem.py b/tests/test_ModelicaSystem.py index 5704c950..ac1d0f50 100644 --- a/tests/test_ModelicaSystem.py +++ b/tests/test_ModelicaSystem.py @@ -35,7 +35,7 @@ def worker(): def test_setParameters(self): omc = OMPython.OMCSessionZMQ() model_path = omc.sendExpression("getInstallationDirectoryPath()") + "/share/doc/omc/testmodels/" - mod = OMPython.ModelicaSystem(model_path + "BouncingBall.mo", "BouncingBall", raiseerrors=True) + mod = OMPython.ModelicaSystem(model_path + "BouncingBall.mo", "BouncingBall") # method 1 mod.setParameters("e=1.234") @@ -59,7 +59,7 @@ def test_setParameters(self): def test_setSimulationOptions(self): omc = OMPython.OMCSessionZMQ() model_path = omc.sendExpression("getInstallationDirectoryPath()") + "/share/doc/omc/testmodels/" - mod = OMPython.ModelicaSystem(model_path + "BouncingBall.mo", "BouncingBall", raiseerrors=True) + mod = OMPython.ModelicaSystem(model_path + "BouncingBall.mo", "BouncingBall") # method 1 mod.setSimulationOptions("stopTime=1.234") @@ -89,7 +89,7 @@ def test_relative_path(self): model_relative = str(model_file) assert "/" not in model_relative - mod = OMPython.ModelicaSystem(model_relative, "M", raiseerrors=True) + mod = OMPython.ModelicaSystem(model_relative, "M") assert float(mod.getParameters("a")[0]) == -1 finally: # clean up the temporary file @@ -99,9 +99,8 @@ def test_customBuildDirectory(self): filePath = (self.tmp / "M.mo").as_posix() tmpdir = self.tmp / "tmpdir1" tmpdir.mkdir() - m = OMPython.ModelicaSystem(filePath, "M", raiseerrors=True, - customBuildDirectory=tmpdir) - assert pathlib.Path(m.getWorkDirectory()).resolve() == tmpdir.resolve() + m = OMPython.ModelicaSystem(filePath, "M", customBuildDirectory=tmpdir) + assert m.getWorkDirectory().resolve() == tmpdir.resolve() result_file = tmpdir / "a.mat" assert not result_file.exists() m.simulate(resultfile="a.mat") @@ -109,7 +108,7 @@ def test_customBuildDirectory(self): def test_getSolutions(self): filePath = (self.tmp / "M.mo").as_posix() - mod = OMPython.ModelicaSystem(filePath, "M", raiseerrors=True) + mod = OMPython.ModelicaSystem(filePath, "M") x0 = 1 a = -1 tau = -1 / a @@ -145,7 +144,7 @@ def test_getters(self): y = der(x); end M_getters; """) - mod = OMPython.ModelicaSystem(model_file.as_posix(), "M_getters", raiseerrors=True) + mod = OMPython.ModelicaSystem(model_file.as_posix(), "M_getters") q = mod.getQuantities() assert isinstance(q, list) @@ -323,7 +322,7 @@ def test_simulate_inputs(self): y = x; end M_input; """) - mod = OMPython.ModelicaSystem(model_file.as_posix(), "M_input", raiseerrors=True) + mod = OMPython.ModelicaSystem(model_file.as_posix(), "M_input") mod.setSimulationOptions("stopTime=1.0") diff --git a/tests/test_ModelicaSystemCmd.py b/tests/test_ModelicaSystemCmd.py new file mode 100644 index 00000000..6257a2a6 --- /dev/null +++ b/tests/test_ModelicaSystemCmd.py @@ -0,0 +1,42 @@ +import OMPython +import pathlib +import shutil +import tempfile +import unittest + + +import logging +logger = logging.getLogger(__name__) +logging.basicConfig(level=logging.DEBUG) + + +class ModelicaSystemCmdTester(unittest.TestCase): + def __init__(self, *args, **kwargs): + super(ModelicaSystemCmdTester, self).__init__(*args, **kwargs) + self.tmp = pathlib.Path(tempfile.mkdtemp(prefix='tmpOMPython.tests')) + self.model = self.tmp / "M.mo" + with open(self.model, "w") as fout: + fout.write("""model M + Real x(start = 1, fixed = true); + parameter Real a = -1; +equation + der(x) = x*a; +end M; + """) + self.mod = OMPython.ModelicaSystem(self.model.as_posix(), "M") + + def __del__(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_simflags(self): + mscmd = OMPython.ModelicaSystemCmd(runpath=self.mod.tempdir, modelname=self.mod.modelName) + mscmd.args_set(args={"noEventEmit": None, "noRestart": None, "override": {'b': 2}}) + mscmd.args_set(args=mscmd.parse_simflags(simflags="-noEventEmit -noRestart -override=a=1,x=3")) + + logger.info(mscmd.get_cmd()) + + assert mscmd.get_cmd() == [mscmd.get_exe().as_posix(), '-noEventEmit', '-noRestart', '-override=b=2,a=1,x=3'] + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_OMSessionCmd.py b/tests/test_OMSessionCmd.py new file mode 100644 index 00000000..96fd667a --- /dev/null +++ b/tests/test_OMSessionCmd.py @@ -0,0 +1,22 @@ +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') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_linearization.py b/tests/test_linearization.py index 2cc49fed..bf759fde 100644 --- a/tests/test_linearization.py +++ b/tests/test_linearization.py @@ -61,7 +61,7 @@ def test_getters(self): y2 = phi + u1; end Pendulum; """) - mod = OMPython.ModelicaSystem(model_file.as_posix(), "Pendulum", ["Modelica"], raiseerrors=True) + mod = OMPython.ModelicaSystem(model_file.as_posix(), "Pendulum", ["Modelica"]) d = mod.getLinearizationOptions() assert isinstance(d, dict) diff --git a/tests/test_optimization.py b/tests/test_optimization.py index ae93d0b8..283df062 100644 --- a/tests/test_optimization.py +++ b/tests/test_optimization.py @@ -45,8 +45,7 @@ def test_example(self): end BangBang2021; """) - mod = OMPython.ModelicaSystem(model_file.as_posix(), "BangBang2021", - raiseerrors=True) + mod = OMPython.ModelicaSystem(model_file.as_posix(), "BangBang2021") mod.setOptimizationOptions(["numberOfIntervals=16", "stopTime=1", "stepSize=0.001", "tolerance=1e-8"])