Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
261 changes: 9 additions & 252 deletions OMPython/ModelicaSystem.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,13 @@

import numpy as np

from OMPython.OMCSession import (
from OMPython.model_execution import (
ModelExecutionCmd,
ModelExecutionData,
ModelExecutionException,

OMCSessionException,
)
from OMPython.OMCSession import (
OMSessionException,
OMCSessionLocal,

OMPathABC,
Expand Down Expand Up @@ -95,253 +97,6 @@ def __getitem__(self, index: int):
return {0: self.A, 1: self.B, 2: self.C, 3: self.D}[index]


class ModelExecutionCmd:
"""
All information about a compiled model executable. This should include data about all structured parameters, i.e.
parameters which need a recompilation of the model. All non-structured parameters can be easily changed without
the need for recompilation.
"""

def __init__(
self,
runpath: os.PathLike,
cmd_prefix: list[str],
cmd_local: bool = False,
cmd_windows: bool = False,
timeout: float = 10.0,
model_name: Optional[str] = None,
) -> None:
if model_name is None:
raise ModelExecutionException("Missing model name!")

self._cmd_local = cmd_local
self._cmd_windows = cmd_windows
self._cmd_prefix = cmd_prefix
self._runpath = pathlib.PurePosixPath(runpath)
self._model_name = model_name
self._timeout = timeout

# dictionaries of command line arguments for the model executable
self._args: dict[str, str | None] = {}
# 'override' argument needs special handling, as it is a dict on its own saved as dict elements following the
# structure: 'key' => 'key=value'
self._arg_override: dict[str, str] = {}

def arg_set(
self,
key: str,
val: Optional[str | dict[str, Any] | numbers.Number] = None,
) -> None:
"""
Set one argument for the executable model.

Args:
key: identifier / argument name to be used for the call of the model executable.
val: value for the given key; None for no value and for key == 'override' a dictionary can be used which
indicates variables to override
"""

def override2str(
orkey: str,
orval: str | bool | numbers.Number,
) -> str:
"""
Convert a value for 'override' to a string taking into account differences between Modelica and Python.
"""
# check oval for any string representations of numbers (or bool) and convert these to Python representations
if isinstance(orval, str):
try:
val_evaluated = ast.literal_eval(orval)
if isinstance(val_evaluated, (numbers.Number, bool)):
orval = val_evaluated
except (ValueError, SyntaxError):
pass

if isinstance(orval, str):
val_str = orval.strip()
elif isinstance(orval, bool):
val_str = 'true' if orval else 'false'
elif isinstance(orval, numbers.Number):
val_str = str(orval)
else:
raise ModelExecutionException(f"Invalid value for override key {orkey}: {type(orval)}")

return f"{orkey}={val_str}"

if not isinstance(key, str):
raise ModelExecutionException(f"Invalid argument key: {repr(key)} (type: {type(key)})")
key = key.strip()

if isinstance(val, dict):
if key != 'override':
raise ModelExecutionException("Dictionary input only possible for key 'override'!")

for okey, oval in val.items():
if not isinstance(okey, str):
raise ModelExecutionException("Invalid key for argument 'override': "
f"{repr(okey)} (type: {type(okey)})")

if not isinstance(oval, (str, bool, numbers.Number, type(None))):
raise ModelExecutionException(f"Invalid input for 'override'.{repr(okey)}: "
f"{repr(oval)} (type: {type(oval)})")

if okey in self._arg_override:
if oval is None:
logger.info(f"Remove model executable override argument: {repr(self._arg_override[okey])}")
del self._arg_override[okey]
continue

logger.info(f"Update model executable override argument: {repr(okey)} = {repr(oval)} "
f"(was: {repr(self._arg_override[okey])})")

if oval is not None:
self._arg_override[okey] = override2str(orkey=okey, orval=oval)

argval = ','.join(sorted(self._arg_override.values()))
elif val is None:
argval = None
elif isinstance(val, str):
argval = val.strip()
elif isinstance(val, numbers.Number):
argval = str(val)
else:
raise ModelExecutionException(f"Invalid argument value for {repr(key)}: {repr(val)} (type: {type(val)})")

if key in self._args:
logger.warning(f"Override model executable argument: {repr(key)} = {repr(argval)} "
f"(was: {repr(self._args[key])})")
self._args[key] = argval

def arg_get(self, key: str) -> Optional[str | dict[str, str | bool | numbers.Number]]:
"""
Return the value for the given key
"""
if key in self._args:
return self._args[key]

return None

def args_set(
self,
args: dict[str, Optional[str | dict[str, Any] | numbers.Number]],
) -> None:
"""
Define arguments for the model executable.
"""
for arg in args:
self.arg_set(key=arg, val=args[arg])

def get_cmd_args(self) -> list[str]:
"""
Get a list with the command arguments for the model executable.
"""

cmdl = []
for key in sorted(self._args):
if self._args[key] is None:
cmdl.append(f"-{key}")
else:
cmdl.append(f"-{key}={self._args[key]}")

return cmdl

def definition(self) -> ModelExecutionData:
"""
Define all needed data to run the model executable. The data is stored in an OMCSessionRunData object.
"""
# ensure that a result filename is provided
result_file = self.arg_get('r')
if not isinstance(result_file, str):
result_file = (self._runpath / f"{self._model_name}.mat").as_posix()

# as this is the local implementation, pathlib.Path can be used
cmd_path = self._runpath

cmd_library_path = None
if self._cmd_local and self._cmd_windows:
cmd_library_path = ""

# set the process environment from the generated .bat file in windows which should have all the dependencies
# for this pathlib.PurePosixPath() must be converted to a pathlib.Path() object, i.e. WindowsPath
path_bat = pathlib.Path(cmd_path) / f"{self._model_name}.bat"
if not path_bat.is_file():
raise ModelExecutionException("Batch file (*.bat) does not exist " + str(path_bat))

content = path_bat.read_text(encoding='utf-8')
for line in content.splitlines():
match = re.match(pattern=r"^SET PATH=([^%]*)", string=line, flags=re.IGNORECASE)
if match:
cmd_library_path = match.group(1).strip(';') # Remove any trailing semicolons
my_env = os.environ.copy()
my_env["PATH"] = cmd_library_path + os.pathsep + my_env["PATH"]

cmd_model_executable = cmd_path / f"{self._model_name}.exe"
else:
# for Linux the paths to the needed libraries should be included in the executable (using rpath)
cmd_model_executable = cmd_path / self._model_name

# define local(!) working directory
cmd_cwd_local = None
if self._cmd_local:
cmd_cwd_local = cmd_path.as_posix()

omc_run_data = ModelExecutionData(
cmd_path=cmd_path.as_posix(),
cmd_model_name=self._model_name,
cmd_args=self.get_cmd_args(),
cmd_result_file=result_file,
cmd_prefix=self._cmd_prefix,
cmd_library_path=cmd_library_path,
cmd_model_executable=cmd_model_executable.as_posix(),
cmd_cwd_local=cmd_cwd_local,
cmd_timeout=self._timeout,
)

return omc_run_data

@staticmethod
def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, Any] | numbers.Number]]:
"""
Parse a simflag definition; this is deprecated!

The return data can be used as input for self.args_set().
"""
warnings.warn(
message="The argument 'simflags' is depreciated and will be removed in future versions; "
"please use 'simargs' instead",
category=DeprecationWarning,
stacklevel=2,
)

simargs: dict[str, Optional[str | dict[str, Any] | numbers.Number]] = {}

args = [s for s in simflags.split(' ') if s]
for arg in args:
if arg[0] != '-':
raise ModelExecutionException(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:])

override_dict = {}
for item in override.split(','):
kv = item.split('=')
if not 0 < len(kv) < 3:
raise ModelExecutionException(f"Invalid value for '-override': {override}")
if kv[0]:
try:
override_dict[kv[0]] = kv[1]
except (KeyError, IndexError) as ex:
raise ModelExecutionException(f"Invalid value for '-override': {override}") from ex

simargs[parts[0]] = override_dict

return simargs


class ModelicaSystemABC(metaclass=abc.ABCMeta):
"""
Base class to simulate a Modelica models.
Expand Down Expand Up @@ -1688,7 +1443,7 @@ def sendExpression(self, expr: str, parsed: bool = True) -> Any:
"""
try:
retval = self._session.sendExpression(expr=expr, parsed=parsed)
except OMCSessionException as ex:
except OMSessionException as ex:
raise ModelicaSystemError(f"Error executing {repr(expr)}: {ex}") from ex

logger.debug(f"Result of executing {repr(expr)}: {textwrap.shorten(repr(retval), width=100)}")
Expand Down Expand Up @@ -2829,7 +2584,9 @@ def _prepare_structure_parameters(


class ModelicaSystemCmd(ModelExecutionCmd):
# TODO: docstring
"""
Compatibility class; in the new version it is renamed as MOdelExecutionCmd.
"""

def __init__(
self,
Expand Down
Loading
Loading