"""Modelica model linearization results.

n: int

m: int

p: int

A: list

B: list

C: list

D: list

x0: list[float]

u0: list[float]

stateVars: list[str]

inputVars: list[str]

outputVars: list[str]

def __iter__(self):

"""Allow unpacking A, B, C, D = result."""

yield self.A

yield self.B

yield self.C

yield self.D

def __getitem__(self, index: int):

"""Allow accessing A, B, C, D via result[0] through result[3].

"""

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:

"""

def override2str(

orkey: str,

orval: str | bool | numbers.Number,

) -> str:

"""

# 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]]:

"""

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:

"""

for arg in args:

self.arg_set(key=arg, val=args[arg])

def get_cmd_args(self) -> list[str]:

"""

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:

"""

# 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]]:

"""

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

"""

def __init__(

self,

session: OMSessionABC,

work_directory: Optional[str | os.PathLike] = None,

) -> None:

"""Create a ModelicaSystem instance. To define the model use model() or convertFmu2Mo().

self._quantities: list[dict[str, Any]] = []

self._params: dict[str, str] = {} # even numerical values are stored as str

self._inputs: dict[str, list[tuple[float, float]]] = {}

self._outputs: dict[str, np.float64] = {} # numpy.float64 as it allows to define None values

self._continuous: dict[str, np.float64] = {} # numpy.float64 as it allows to define None values

self._simulate_options: dict[str, str] = {}

self._override_variables: dict[str, str] = {}

self._simulate_options_override: dict[str, str] = {}

self._linearization_options: dict[str, str] = {

'startTime': str(0.0),

'stopTime': str(1.0),

'stepSize': str(0.002),

'tolerance': str(1e-8),

}

self._optimization_options = self._linearization_options | {

'numberOfIntervals': str(500),

}

self._linearized_inputs: list[str] = [] # linearization input list

self._linearized_outputs: list[str] = [] # linearization output list

self._linearized_states: list[str] = [] # linearization states list

self._simulated = False # True if the model has already been simulated

self._result_file: Optional[OMPathABC] = None # for storing result file

self._model_name: Optional[str] = None

self._libraries: Optional[list[str | tuple[str, str]]] = None

self._file_name: Optional[OMPathABC] = None

self._variable_filter: Optional[str] = None

self._session = session

# get OpenModelica version

version_str = self._session.get_version()

self._version = self._parse_om_version(version=version_str)

self._work_dir: OMPathABC = self.setWorkDirectory(work_directory)

def get_session(self) -> OMSessionABC:

"""

return self._session

def get_model_name(self) -> str:

"""

if not isinstance(self._model_name, str):

raise ModelicaSystemError("No model name defined!")

return self._model_name

def setWorkDirectory(self, work_directory: Optional[str | os.PathLike] = None) -> OMPathABC:

"""

if work_directory is not None:

workdir = self._session.omcpath(work_directory).absolute()

if not workdir.is_dir():

raise IOError(f"Provided work directory does not exists: {work_directory}!")

else:

workdir = self._session.omcpath_tempdir().absolute()

if not workdir.is_dir():

raise IOError(f"{workdir} could not be created")

logger.info("Define work dir as %s", workdir)

self._session.set_workdir(workdir=workdir)

# set the class variable _work_dir ...

self._work_dir = workdir

# ... and also return the defined path

return workdir

def getWorkDirectory(self) -> OMPathABC:

"""

return self._work_dir

def check_model_executable(self):

"""

# check if the executable exists ...

om_cmd = ModelExecutionCmd(

runpath=self.getWorkDirectory(),

cmd_local=self._session.model_execution_local,

cmd_windows=self._session.model_execution_windows,

cmd_prefix=self._session.model_execution_prefix(cwd=self.getWorkDirectory()),

timeout=self._session.set_timeout(),

model_name=self._model_name,

)

# ... by running it - output help for command help

om_cmd.arg_set(key="help", val="help")

cmd_definition = om_cmd.definition()

returncode = cmd_definition.run()

if returncode != 0:

raise ModelicaSystemError("Model executable not working!")

def _xmlparse(self, xml_file: OMPathABC):

if not xml_file.is_file():

raise ModelicaSystemError(f"XML file not generated: {xml_file}")

xml_content = xml_file.read_text()

tree = ET.ElementTree(ET.fromstring(xml_content))

root = tree.getroot()

if root is None:

raise ModelicaSystemError(f"Cannot read XML file: {xml_file}")

for attr in root.iter('DefaultExperiment'):

for key in ("startTime", "stopTime", "stepSize", "tolerance",

"solver", "outputFormat"):

self._simulate_options[key] = str(attr.get(key))

for sv in root.iter('ScalarVariable'):

translations = {

"alias": "alias",

"aliasvariable": "aliasVariable",

"causality": "causality",

"changeable": "isValueChangeable",

"description": "description",

"name": "name",

"variability": "variability",

}

scalar: dict[str, Any] = {}

for key_dst, key_src in translations.items():

val = sv.get(key_src)

scalar[key_dst] = None if val is None else str(val)

ch = list(sv)

for att in ch:

scalar["start"] = att.get('start')

scalar["min"] = att.get('min')

scalar["max"] = att.get('max')

scalar["unit"] = att.get('unit')

# save parameters in the corresponding class variables

if scalar["variability"] == "parameter":

if scalar["name"] in self._override_variables:

self._params[scalar["name"]] = self._override_variables[scalar["name"]]

else:

self._params[scalar["name"]] = scalar["start"]

if scalar["variability"] == "continuous":

self._continuous[scalar["name"]] = np.float64(scalar["start"])

if scalar["causality"] == "input":

self._inputs[scalar["name"]] = scalar["start"]

if scalar["causality"] == "output":

self._outputs[scalar["name"]] = np.float64(scalar["start"])

self._quantities.append(scalar)

def getQuantities(self, names: Optional[str | list[str]] = None) -> list[dict]:

"""

if names is None:

return self._quantities

if isinstance(names, str):

r = [x for x in self._quantities if x["name"] == names]

if r == []:

raise KeyError(names)

return r

if isinstance(names, list):

return [x for y in names for x in self._quantities if x["name"] == y]

raise ModelicaSystemError("Unhandled input for getQuantities()")

def getContinuousInitial(

self,

names: Optional[str | list[str]] = None,

) -> dict[str, np.float64] | list[np.float64]:

"""

if names is None:

return self._continuous

if isinstance(names, str):

return [self._continuous[names]]

if isinstance(names, list):

return [self._continuous[x] for x in names]

raise ModelicaSystemError("Unhandled input for getContinousInitial()")

def getParameters(

self,

names: Optional[str | list[str]] = None,

) -> dict[str, str] | list[str]:

"""Get parameter values.

if names is None:

return self._params

if isinstance(names, str):

return [self._params[names]]

if isinstance(names, list):

return [self._params[x] for x in names]

raise ModelicaSystemError("Unhandled input for getParameters()")

def getInputs(

self,

names: Optional[str | list[str]] = None,

) -> dict[str, list[tuple[float, float]]] | list[list[tuple[float, float]]]:

"""Get values of input signals.

if names is None:

return self._inputs

if isinstance(names, str):

return [self._inputs[names]]

if isinstance(names, list):

return [self._inputs[x] for x in names]

raise ModelicaSystemError("Unhandled input for getInputs()")

def getOutputsInitial(

self,

names: Optional[str | list[str]] = None,

) -> dict[str, np.float64] | list[np.float64]:

"""

if names is None:

return self._outputs

if isinstance(names, str):

return [self._outputs[names]]

if isinstance(names, list):

return [self._outputs[x] for x in names]

raise ModelicaSystemError("Unhandled input for getOutputsInitial()")

def getSimulationOptions(

self,

names: Optional[str | list[str]] = None,

) -> dict[str, str] | list[str]:

"""Get simulation options such as stopTime and tolerance.

if names is None:

return self._simulate_options

if isinstance(names, str):

return [self._simulate_options[names]]

if isinstance(names, list):

return [self._simulate_options[x] for x in names]

raise ModelicaSystemError("Unhandled input for getSimulationOptions()")

def getLinearizationOptions(

self,

names: Optional[str | list[str]] = None,

) -> dict[str, str] | list[str]:

"""Get simulation options used for linearization.

if names is None:

return self._linearization_options

if isinstance(names, str):

return [self._linearization_options[names]]

if isinstance(names, list):

return [self._linearization_options[x] for x in names]

raise ModelicaSystemError("Unhandled input for getLinearizationOptions()")

def getOptimizationOptions(

self,

names: Optional[str | list[str]] = None,

) -> dict[str, str] | list[str]:

"""Get simulation options used for optimization.

if names is None:

return self._optimization_options

if isinstance(names, str):

return [self._optimization_options[names]]

if isinstance(names, list):

return [self._optimization_options[x] for x in names]

raise ModelicaSystemError("Unhandled input for getOptimizationOptions()")

@staticmethod

def _parse_om_version(version: str) -> tuple[int, int, int]:

"""

match = re.search(pattern=r"v?(\d+)\.(\d+)\.(\d+)", string=version)

if not match:

raise ValueError(f"Version not found in: {version}")

major, minor, patch = map(int, match.groups())

return major, minor, patch

def _process_override_data(

self,

om_cmd: ModelExecutionCmd,

override_file: OMPathABC,

override_var: dict[str, str],

override_sim: dict[str, str],

) -> None:

"""

if len(override_var) == 0 and len(override_sim) == 0:

return

override_content = ""

if override_var:

override_content += "\n".join([f"{key}={value}" for key, value in override_var.items()]) + "\n"

# simulation options are not read from override file from version >= 1.26.0,

# pass them to simulation executable directly as individual arguments

# see https://github.com/OpenModelica/OpenModelica/pull/14813

if override_sim:

if self._version >= (1, 26, 0):

for key, opt_value in override_sim.items():

if key == "solver":

k = "s"

else:

k = key

om_cmd.arg_set(key=k, val=str(opt_value))

else:

override_content += "\n".join([f"{key}={value}" for key, value in override_sim.items()]) + "\n"

if override_content:

override_file.write_text(override_content)

om_cmd.arg_set(key="overrideFile", val=override_file.as_posix())

def simulate_cmd(

self,

result_file: OMPathABC,

simflags: Optional[str] = None,

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

) -> ModelExecutionCmd:

"""

om_cmd = ModelExecutionCmd(

runpath=self.getWorkDirectory(),

cmd_local=self._session.model_execution_local,

cmd_windows=self._session.model_execution_windows,

cmd_prefix=self._session.model_execution_prefix(cwd=self.getWorkDirectory()),

timeout=self._session.set_timeout(),

model_name=self._model_name,

)

# always define the result file to use

om_cmd.arg_set(key="r", val=result_file.as_posix())

# allow runtime simulation flags from user input

if simflags is not None:

om_cmd.args_set(args=om_cmd.parse_simflags(simflags=simflags))

if simargs:

om_cmd.args_set(args=simargs)

self._process_override_data(

om_cmd=om_cmd,

override_file=result_file.parent / f"{result_file.stem}_override.txt",

override_var=self._override_variables,

override_sim=self._simulate_options_override,

)

if self._inputs: # if model has input quantities

for key, val in self._inputs.items():

if val is None:

val = [(float(self._simulate_options["startTime"]), 0.0),

(float(self._simulate_options["stopTime"]), 0.0)]

self._inputs[key] = val

if float(self._simulate_options["startTime"]) != val[0][0]:

raise ModelicaSystemError(f"startTime not matched for Input {key}!")

if float(self._simulate_options["stopTime"]) != val[-1][0]:

raise ModelicaSystemError(f"stopTime not matched for Input {key}!")

# csvfile is based on name used for result file

csvfile = result_file.parent / f"{result_file.stem}.csv"

# write csv file and store the name

csvfile = self._createCSVData(csvfile=csvfile)

om_cmd.arg_set(key="csvInput", val=csvfile.as_posix())

return om_cmd

def simulate(

self,

resultfile: Optional[str | os.PathLike] = None,

simflags: Optional[str] = None,

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

) -> None:

"""Simulate the model according to simulation options.

if resultfile is None:

# default result file generated by OM

self._result_file = self.getWorkDirectory() / f"{self._model_name}_res.mat"

elif isinstance(resultfile, OMPathABC):

self._result_file = resultfile

else:

self._result_file = self._session.omcpath(resultfile)

if not self._result_file.is_absolute():

self._result_file = self.getWorkDirectory() / resultfile

if not isinstance(self._result_file, OMPathABC):

raise ModelicaSystemError(f"Invalid result file path: {self._result_file} - must be an OMCPath object!")

om_cmd = self.simulate_cmd(

result_file=self._result_file,

simflags=simflags,

simargs=simargs,

)

# delete resultfile ...

if self._result_file.is_file():

self._result_file.unlink()

# ... run simulation ...

cmd_definition = om_cmd.definition()

returncode = cmd_definition.run()

# and check returncode *AND* resultfile

if returncode != 0 and self._result_file.is_file():

# check for an empty (=> 0B) result file which indicates a crash of the model executable

# see: https://github.com/OpenModelica/OMPython/issues/261