diff --git a/OMPython/ModelicaSystem.py b/OMPython/ModelicaSystem.py index 570c7b43..5531e477 100644 --- a/OMPython/ModelicaSystem.py +++ b/OMPython/ModelicaSystem.py @@ -468,12 +468,12 @@ def set_command_line_options(self, command_line_option: str): """ Set the provided command line option via OMC setCommandLineOptions(). """ - exp = f'setCommandLineOptions("{command_line_option}")' - self.sendExpression(exp) + expr = f'setCommandLineOptions("{command_line_option}")' + self.sendExpression(expr=expr) def _loadFile(self, fileName: OMCPath): # load file - self.sendExpression(f'loadFile("{fileName.as_posix()}")') + self.sendExpression(expr=f'loadFile("{fileName.as_posix()}")') # for loading file/package, loading model and building model def _loadLibrary(self, libraries: list): @@ -491,7 +491,7 @@ def _loadLibrary(self, libraries: list): expr_load_lib = f"loadModel({element[0]})" else: expr_load_lib = f'loadModel({element[0]}, {{"{element[1]}"}})' - self.sendExpression(expr_load_lib) + self.sendExpression(expr=expr_load_lib) else: raise ModelicaSystemError("loadLibrary() failed, Unknown type detected: " f"{element} is of type {type(element)}, " @@ -514,8 +514,8 @@ def setWorkDirectory(self, work_directory: Optional[str | os.PathLike] = None) - raise IOError(f"{workdir} could not be created") logger.info("Define work dir as %s", workdir) - exp = f'cd("{workdir.as_posix()}")' - self.sendExpression(exp) + expr = f'cd("{workdir.as_posix()}")' + self.sendExpression(expr=expr) # set the class variable _work_dir ... self._work_dir = workdir @@ -561,7 +561,7 @@ def buildModel(self, variableFilter: Optional[str] = None): def sendExpression(self, expr: str, parsed: bool = True) -> Any: try: - retval = self._session.sendExpression(command=expr, parsed=parsed) + retval = self._session.sendExpression(expr=expr, parsed=parsed) except OMCSessionException as ex: raise ModelicaSystemError(f"Error executing {repr(expr)}: {ex}") from ex @@ -577,16 +577,16 @@ def _requestApi( properties: Optional[str] = None, ) -> Any: if entity is not None and properties is not None: - exp = f'{apiName}({entity}, {properties})' + expr = f'{apiName}({entity}, {properties})' elif entity is not None and properties is None: if apiName in ("loadFile", "importFMU"): - exp = f'{apiName}("{entity}")' + expr = f'{apiName}("{entity}")' else: - exp = f'{apiName}({entity})' + expr = f'{apiName}({entity})' else: - exp = f'{apiName}()' + expr = f'{apiName}()' - return self.sendExpression(exp) + return self.sendExpression(expr=expr) def _xmlparse(self, xml_file: OMCPath): if not xml_file.is_file(): @@ -1275,8 +1275,8 @@ def getSolutions( # get absolute path result_file = result_file.absolute() - result_vars = self.sendExpression(f'readSimulationResultVars("{result_file.as_posix()}")') - self.sendExpression("closeSimulationResultFile()") + result_vars = self.sendExpression(expr=f'readSimulationResultVars("{result_file.as_posix()}")') + self.sendExpression(expr="closeSimulationResultFile()") if varList is None: return result_vars @@ -1293,9 +1293,9 @@ def getSolutions( if var not in result_vars: raise ModelicaSystemError(f"Requested data {repr(var)} does not exist") variables = ",".join(var_list_checked) - res = self.sendExpression(f'readSimulationResult("{result_file.as_posix()}",{{{variables}}})') + res = self.sendExpression(expr=f'readSimulationResult("{result_file.as_posix()}",{{{variables}}})') np_res = np.array(res) - self.sendExpression("closeSimulationResultFile()") + self.sendExpression(expr="closeSimulationResultFile()") return np_res @staticmethod @@ -1395,7 +1395,7 @@ def _set_method_helper( "structural, final, protected, evaluated or has a non-constant binding. " "Use sendExpression(...) and rebuild the model using buildModel() API; " "command to set the parameter before rebuilding the model: " - "sendExpression(\"setParameterValue(" + "sendExpression(expr=\"setParameterValue(" f"{self._model_name}, {key}, {val if val is not None else ''}" ")\").") @@ -2054,16 +2054,16 @@ def prepare(self) -> int: pk_value = pc_structure[idx_structure] if isinstance(pk_value, str): pk_value_str = self.get_session().escape_str(pk_value) - expression = f"setParameterValue({self._model_name}, {pk_structure}, \"{pk_value_str}\")" + expr = f"setParameterValue({self._model_name}, {pk_structure}, \"{pk_value_str}\")" elif isinstance(pk_value, bool): pk_value_bool_str = "true" if pk_value else "false" - expression = f"setParameterValue({self._model_name}, {pk_structure}, {pk_value_bool_str});" + expr = f"setParameterValue({self._model_name}, {pk_structure}, {pk_value_bool_str});" else: - expression = f"setParameterValue({self._model_name}, {pk_structure}, {pk_value})" - res = self._mod.sendExpression(expression) + expr = f"setParameterValue({self._model_name}, {pk_structure}, {pk_value})" + res = self._mod.sendExpression(expr=expr) if not res: raise ModelicaSystemError(f"Cannot set structural parameter {self._model_name}.{pk_structure} " - f"to {pk_value} using {repr(expression)}") + f"to {pk_value} using {repr(expr)}") self._mod.buildModel() diff --git a/OMPython/OMCSession.py b/OMPython/OMCSession.py index 1e2a5383..3a72778f 100644 --- a/OMPython/OMCSession.py +++ b/OMPython/OMCSession.py @@ -280,13 +280,13 @@ def is_file(self, *, follow_symlinks=True) -> bool: """ Check if the path is a regular file. """ - return self._session.sendExpression(f'regularFileExists("{self.as_posix()}")') + return self._session.sendExpression(expr=f'regularFileExists("{self.as_posix()}")') def is_dir(self, *, follow_symlinks=True) -> bool: """ Check if the path is a directory. """ - return self._session.sendExpression(f'directoryExists("{self.as_posix()}")') + return self._session.sendExpression(expr=f'directoryExists("{self.as_posix()}")') def is_absolute(self): """ @@ -304,7 +304,7 @@ def read_text(self, encoding=None, errors=None, newline=None) -> str: The additional arguments `encoding`, `errors` and `newline` are only defined for compatibility with Path() definition. """ - return self._session.sendExpression(f'readFile("{self.as_posix()}")') + return self._session.sendExpression(expr=f'readFile("{self.as_posix()}")') def write_text(self, data: str, encoding=None, errors=None, newline=None): """ @@ -317,7 +317,7 @@ def write_text(self, data: str, encoding=None, errors=None, newline=None): raise TypeError(f"data must be str, not {data.__class__.__name__}") data_omc = self._session.escape_str(data) - self._session.sendExpression(f'writeFile("{self.as_posix()}", "{data_omc}", false);') + self._session.sendExpression(expr=f'writeFile("{self.as_posix()}", "{data_omc}", false);') return len(data) @@ -330,20 +330,20 @@ def mkdir(self, mode=0o777, parents=False, exist_ok=False): if self.is_dir() and not exist_ok: raise FileExistsError(f"Directory {self.as_posix()} already exists!") - return self._session.sendExpression(f'mkdir("{self.as_posix()}")') + return self._session.sendExpression(expr=f'mkdir("{self.as_posix()}")') def cwd(self): """ Returns the current working directory as an OMCPath object. """ - cwd_str = self._session.sendExpression('cd()') + cwd_str = self._session.sendExpression(expr='cd()') return OMCPath(cwd_str, session=self._session) def unlink(self, missing_ok: bool = False) -> None: """ Unlink (delete) the file or directory represented by this path. """ - res = self._session.sendExpression(f'deleteFile("{self.as_posix()}")') + res = self._session.sendExpression(expr=f'deleteFile("{self.as_posix()}")') if not res and not missing_ok: raise FileNotFoundError(f"Cannot delete file {self.as_posix()} - it does not exists!") @@ -373,12 +373,12 @@ def _omc_resolve(self, pathstr: str) -> str: Internal function to resolve the path of the OMCPath object using OMC functions *WITHOUT* changing the cwd within OMC. """ - expression = ('omcpath_cwd := cd(); ' - f'omcpath_check := cd("{pathstr}"); ' # check requested pathstring - 'cd(omcpath_cwd)') + expr = ('omcpath_cwd := cd(); ' + f'omcpath_check := cd("{pathstr}"); ' # check requested pathstring + 'cd(omcpath_cwd)') try: - result = self._session.sendExpression(command=expression, parsed=False) + result = self._session.sendExpression(expr=expr, parsed=False) result_parts = result.split('\n') pathstr_resolved = result_parts[1] pathstr_resolved = pathstr_resolved[1:-1] # remove quotes @@ -407,7 +407,7 @@ def size(self) -> int: if not self.is_file(): raise OMCSessionException(f"Path {self.as_posix()} is not a file!") - res = self._session.sendExpression(f'stat("{self.as_posix()}")') + res = self._session.sendExpression(expr=f'stat("{self.as_posix()}")') if res[0]: return int(res[1]) @@ -582,7 +582,7 @@ def sendExpression(self, command: str, parsed: bool = True) -> Any: 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(command=command, parsed=parsed) + return self.omc_process.sendExpression(expr=command, parsed=parsed) class PostInitCaller(type): @@ -701,7 +701,7 @@ def __post_init__(self) -> None: def __del__(self): if isinstance(self._omc_zmq, zmq.Socket): try: - self.sendExpression("quit()") + self.sendExpression(expr="quit()") except OMCSessionException as exc: logger.warning(f"Exception on sending 'quit()' to OMC: {exc}! Continue nevertheless ...") finally: @@ -759,7 +759,7 @@ def omcpath_tempdir(self, tempdir_base: Optional[OMCPath] = None) -> OMCPath: if sys.version_info < (3, 12): tempdir_str = tempfile.gettempdir() else: - tempdir_str = self.sendExpression("getTempDirectoryPath()") + tempdir_str = self.sendExpression(expr="getTempDirectoryPath()") tempdir_base = self.omcpath(tempdir_str) tempdir: Optional[OMCPath] = None @@ -825,7 +825,7 @@ def execute(self, command: str): return self.sendExpression(command, parsed=False) - def sendExpression(self, command: str, parsed: bool = True) -> Any: + def sendExpression(self, expr: str, parsed: bool = True) -> Any: """ Send an expression to the OMC server and return the result. @@ -842,12 +842,12 @@ def sendExpression(self, command: str, parsed: bool = True) -> Any: if self._omc_zmq is None: raise OMCSessionException("No OMC running. Please create a new instance of OMCSession!") - logger.debug("sendExpression(%r, parsed=%r)", command, parsed) + logger.debug("sendExpression(expr='%r', parsed=%r)", str(expr), parsed) attempts = 0 while True: try: - self._omc_zmq.send_string(str(command), flags=zmq.NOBLOCK) + self._omc_zmq.send_string(str(expr), flags=zmq.NOBLOCK) break except zmq.error.Again: pass @@ -861,7 +861,7 @@ def sendExpression(self, command: str, parsed: bool = True) -> Any: raise OMCSessionException(f"No connection with OMC (timeout={timeout}). " f"Log-file says: \n{log_content}") time.sleep(timeout / 50.0) - if command == "quit()": + if expr == "quit()": self._omc_zmq.close() self._omc_zmq = None return None @@ -871,13 +871,13 @@ def sendExpression(self, command: str, parsed: bool = True) -> Any: if result.startswith('Error occurred building AST'): raise OMCSessionException(f"OMC error: {result}") - if command == "getErrorString()": + if expr == "getErrorString()": # no error handling if 'getErrorString()' is called if parsed: logger.warning("Result of 'getErrorString()' cannot be parsed!") return result - if command == "getMessagesStringInternal()": + if expr == "getMessagesStringInternal()": # no error handling if 'getMessagesStringInternal()' is called if parsed: logger.warning("Result of 'getMessagesStringInternal()' cannot be parsed!") @@ -931,7 +931,7 @@ def sendExpression(self, command: str, parsed: bool = True) -> Any: log_level = log_raw[0][8] log_id = log_raw[0][9] - msg_short = (f"[OMC log for 'sendExpression({command}, {parsed})']: " + msg_short = (f"[OMC log for 'sendExpression(expr={expr}, parsed={parsed})']: " f"[{log_kind}:{log_level}:{log_id}] {log_message}") # response according to the used log level @@ -953,7 +953,7 @@ def sendExpression(self, command: str, parsed: bool = True) -> Any: msg_long_list.append(msg_long) if has_error: msg_long_str = '\n'.join(f"{idx:02d}: {msg}" for idx, msg in enumerate(msg_long_list)) - raise OMCSessionException(f"OMC error occurred for 'sendExpression({command}, {parsed}):\n" + raise OMCSessionException(f"OMC error occurred for 'sendExpression(expr={expr}, parsed={parsed}):\n" f"{msg_long_str}") if parsed is False: