"""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].

"""A compiled model executable."""

def __init__(

self,

session: OMCSessionZMQ,

runpath: OMCPath,

modelname: Optional[str] = None,

timeout: Optional[float] = None,

) -> None:

if modelname is None:

raise ModelicaSystemError("Missing model name!")

self._session = session

self._runpath = runpath

self._model_name = modelname

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(

okey: str,

oval: str | bool | numbers.Number,

) -> str:

"""

# check oval for any string representations of numbers (or bool) and convert these to Python representations

if isinstance(oval, str):

try:

oval_evaluated = ast.literal_eval(oval)

if isinstance(oval_evaluated, (numbers.Number, bool)):

oval = oval_evaluated

except (ValueError, SyntaxError):

pass

if isinstance(oval, str):

oval_str = oval.strip()

elif isinstance(oval, bool):

oval_str = 'true' if oval else 'false'

elif isinstance(oval, numbers.Number):

oval_str = str(oval)

else:

raise ModelicaSystemError(f"Invalid value for override key {okey}: {type(oval)}")

return f"{okey}={oval_str}"

if not isinstance(key, str):

raise ModelicaSystemError(f"Invalid argument key: {repr(key)} (type: {type(key)})")

key = key.strip()

if isinstance(val, dict):

if key != 'override':

raise ModelicaSystemError("Dictionary input only possible for key 'override'!")

for okey, oval in val.items():

if not isinstance(okey, str):

raise ModelicaSystemError("Invalid key for argument 'override': "

f"{repr(okey)} (type: {type(okey)})")

if not isinstance(oval, (str, bool, numbers.Number, type(None))):

raise ModelicaSystemError(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(okey=okey, oval=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 ModelicaSystemError(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) -> OMCSessionRunData:

"""

# 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()

omc_run_data = OMCSessionRunData(

cmd_path=self._runpath.as_posix(),

cmd_model_name=self._model_name,

cmd_args=self.get_cmd_args(),

cmd_result_path=result_file,

cmd_timeout=self._timeout,

)

omc_run_data_updated = self._session.omc_run_data_update(

omc_run_data=omc_run_data,

)

return omc_run_data_updated

@staticmethod

def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, Any] | numbers.Number]]:

"""

warnings.warn("The argument 'simflags' is depreciated and will be removed in future versions; "

"please use 'simargs' instead", 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 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:])

override_dict = {}

for item in override.split(','):

kv = item.split('=')

if not 0 < len(kv) < 3:

raise ModelicaSystemError(f"Invalid value for '-override': {override}")

if kv[0]:

try:

override_dict[kv[0]] = kv[1]

except (KeyError, IndexError) as ex:

raise ModelicaSystemError(f"Invalid value for '-override': {override}") from ex

simargs[parts[0]] = override_dict

"""

def __init__(

self,

commandLineOptions: Optional[list[str]] = None,

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

omhome: Optional[str] = None,

omc_process: Optional[OMCProcess] = 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]]] = {}

# _outputs values are str before simulate(), but they can be

# np.float64 after simulate().

self._outputs: dict[str, Any] = {}

# same for _continuous

self._continuous: dict[str, Any] = {}

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 | float] = {

'startTime': 0.0,

'stopTime': 1.0,

'stepSize': 0.002,

'tolerance': 1e-8,

}

self._optimization_options = self._linearization_options | {

'numberOfIntervals': 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

if omc_process is not None:

self._session = OMCSessionZMQ(omc_process=omc_process)

else:

self._session = OMCSessionZMQ(omhome=omhome)

# set commandLineOptions using default values or the user defined list

if commandLineOptions is None:

# set default command line options to improve the performance of linearization and to avoid recompilation if

# the simulation executable is reused in linearize() via the runtime flag '-l'

commandLineOptions = [

"--linearizationDumpLanguage=python",

"--generateSymbolicLinearization",

]

for opt in commandLineOptions:

self.setCommandLineOptions(commandLineOptions=opt)

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

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

self._work_dir: OMCPath = self.setWorkDirectory(customBuildDirectory)

self._model_name: Optional[str] = None

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

self._file_name: Optional[OMCPath] = None

self._variable_filter: Optional[str] = None

def model(

self,

name: Optional[str] = None,

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

libraries: Optional[list[str | tuple[str, str]]] = None,

variable_filter: Optional[str] = None,

build: bool = True,

) -> None:

"""Load and build a Modelica model.

if self._model_name is not None:

raise ModelicaSystemError("Can not reuse this instance of ModelicaSystem "

f"defined for {repr(self._model_name)}!")

if name is None or not isinstance(name, str):

raise ModelicaSystemError("A model name must be provided!")

if libraries is None:

libraries = []

if not isinstance(libraries, list):

raise ModelicaSystemError(f"Invalid input type for libraries: {type(libraries)} - list expected!")

# set variables

self._model_name = name # Model class name

self._libraries = libraries # may be needed if model is derived from other model

self._variable_filter = variable_filter

if self._libraries:

self._loadLibrary(libraries=self._libraries)

self._file_name = None

if file is not None:

file_path = pathlib.Path(file)

# special handling for OMCProcessLocal - consider a relative path

if isinstance(self._session.omc_process, OMCProcessLocal) and not file_path.is_absolute():

file_path = pathlib.Path.cwd() / file_path

if not file_path.is_file():

raise IOError(f"Model file {file_path} does not exist!")

self._file_name = self.getWorkDirectory() / file_path.name

if (isinstance(self._session.omc_process, OMCProcessLocal)

and file_path.as_posix() == self._file_name.as_posix()):

pass

elif self._file_name.is_file():

raise IOError(f"Simulation model file {self._file_name} exist - not overwriting!")

else:

content = file_path.read_text(encoding='utf-8')

self._file_name.write_text(content)

if self._file_name is not None:

self._loadFile(fileName=self._file_name)

if build:

self.buildModel(variable_filter)

def session(self) -> OMCSessionZMQ:

"""

return self._session

def setCommandLineOptions(self, commandLineOptions: str):

"""

exp = f'setCommandLineOptions("{commandLineOptions}")'

self.sendExpression(exp)

def _loadFile(self, fileName: OMCPath):

# load file

self.sendExpression(f'loadFile("{fileName.as_posix()}")')

# for loading file/package, loading model and building model

def _loadLibrary(self, libraries: list):

# load Modelica standard libraries or Modelica files if needed

for element in libraries:

if element is not None:

if isinstance(element, str):

if element.endswith(".mo"):

apiCall = "loadFile"

else:

apiCall = "loadModel"

self._requestApi(apiName=apiCall, entity=element)

elif isinstance(element, tuple):

if not element[1]:

expr_load_lib = f"loadModel({element[0]})"

else:

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')

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

"""

if customBuildDirectory is not None:

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

if not workdir.is_dir():

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

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)

exp = f'cd("{workdir.as_posix()}")'

self.sendExpression(exp)

# set the class variable _work_dir ...

self._work_dir = workdir

# ... and also return the defined path

return workdir

def getWorkDirectory(self) -> OMCPath:

"""

return self._work_dir

def buildModel(self, variableFilter: Optional[str] = None):

filter_def: Optional[str] = None

if variableFilter is not None:

filter_def = variableFilter

elif self._variable_filter is not None:

filter_def = self._variable_filter

if filter_def is not None:

var_filter = f'variableFilter="{filter_def}"'

else:

var_filter = 'variableFilter=".*"'

buildModelResult = self._requestApi(apiName="buildModel", entity=self._model_name, properties=var_filter)

logger.debug("OM model build result: %s", buildModelResult)

# check if the executable exists ...

om_cmd = ModelicaSystemCmd(

session=self._session,

runpath=self.getWorkDirectory(),

modelname=self._model_name,

timeout=5.0,

)

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

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

cmd_definition = om_cmd.definition()

returncode = self._session.run_model_executable(cmd_run_data=cmd_definition)

if returncode != 0:

raise ModelicaSystemError("Model executable not working!")

xml_file = self._session.omcpath(buildModelResult[0]).parent / buildModelResult[1]

self._xmlparse(xml_file=xml_file)

def sendExpression(self, expr: str, parsed: bool = True) -> Any:

try:

retval = self._session.sendExpression(expr, parsed)

except OMCSessionException 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)}")

return retval

# request to OMC

def _requestApi(

self,

apiName: str,

entity: Optional[str] = None,

properties: Optional[str] = None,

) -> Any:

if entity is not None and properties is not None:

exp = f'{apiName}({entity}, {properties})'

elif entity is not None and properties is None:

if apiName in ("loadFile", "importFMU"):

exp = f'{apiName}("{entity}")'

else:

exp = f'{apiName}({entity})'

else:

exp = f'{apiName}()'

return self.sendExpression(exp)

def _xmlparse(self, xml_file: OMCPath):

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))

rootCQ = tree.getroot()

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

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

"solver", "outputFormat"):

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

for sv in rootCQ.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"]] = scalar["start"]

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

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

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

self._outputs[scalar["name"]] = 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 getContinuous(

self,

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

) -> dict[str, str | numbers.Real] | list[str | numbers.Real]:

"""Get values of continuous signals.

if not self._simulated:

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]

if names is None:

for name in self._continuous:

try:

value = self.getSolutions(name)

self._continuous[name] = value[0][-1]

except (OMCSessionException, ModelicaSystemError) as ex:

raise ModelicaSystemError(f"{name} could not be computed") from ex

return self._continuous

if isinstance(names, str):

if names in self._continuous:

value = self.getSolutions(names)

self._continuous[names] = value[0][-1]

return [self._continuous[names]]

raise ModelicaSystemError(f"{names} is not continuous")

if isinstance(names, list):

valuelist = []

for name in names:

if name in self._continuous:

value = self.getSolutions(name)

self._continuous[name] = value[0][-1]

valuelist.append(value[0][-1])

else:

raise ModelicaSystemError(f"{name} 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]:

"""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 getOutputs(

self,

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

) -> dict[str, str | numbers.Real] | list[str | numbers.Real]:

"""Get values of output signals.

if not self._simulated:

if names is None:

return self._outputs

if isinstance(names, str):

return [self._outputs[names]]

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

if names is None:

for name in self._outputs:

value = self.getSolutions(name)

self._outputs[name] = value[0][-1]

return self._outputs

if isinstance(names, str):

if names in self._outputs:

value = self.getSolutions(names)

self._outputs[names] = value[0][-1]

return [self._outputs[names]]

raise KeyError(names)

if isinstance(names, list):

valuelist = []

for name in names:

if name in self._outputs:

value = self.getSolutions(name)

self._outputs[name] = value[0][-1]

valuelist.append(value[0][-1])

else:

raise KeyError(name)

return valuelist

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

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 | float] | list[str | float]:

"""Get simulation options used for linearization.