"""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, runpath: pathlib.Path, modelname: str, timeout: Optional[float] = None) -> None:

self._runpath = pathlib.Path(runpath).resolve().absolute()

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_exe(self) -> pathlib.Path:

"""Get the path to the compiled model executable."""

if platform.system() == "Windows":

path_exe = self._runpath / f"{self._model_name}.exe"

else:

path_exe = self._runpath / self._model_name

if not path_exe.exists():

raise ModelicaSystemError(f"Application file path not found: {path_exe}")

return path_exe

def get_cmd(self) -> list:

"""Get a list with the path to the executable and all command line args.

path_exe = self.get_exe()

cmdl = [path_exe.as_posix()]

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 run(self) -> int:

"""Run the requested simulation.

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._model_name}.bat"

if not path_bat.exists():

raise ModelicaSystemError("Batch file (*.bat) does not exist " + str(path_bat))

with open(file=path_bat, mode='r', encoding='utf-8') as fh:

for line in fh:

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, check=True)

stdout = cmdres.stdout.strip()

stderr = cmdres.stderr.strip()

returncode = cmdres.returncode

logger.debug("OM output for command %s:\n%s", repr(cmdl), stdout)

if stderr:

raise ModelicaSystemError(f"Error running command {repr(cmdl)}: {stderr}")

except subprocess.TimeoutExpired as ex:

raise ModelicaSystemError(f"Timeout running command {repr(cmdl)}") from ex

except subprocess.CalledProcessError as ex:

raise ModelicaSystemError(f"Error running command {repr(cmdl)}") from ex

return returncode

@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,

fileName: Optional[str | os.PathLike | pathlib.Path] = None,

modelName: Optional[str] = None,

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

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

variableFilter: Optional[str] = None,

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

omhome: Optional[str] = None,

omc_process: Optional[OMCProcessLocal] = None,

build: bool = True,

) -> None:

"""Initialize, load and build a model.

if fileName is None and modelName is None and not lmodel: # all None

raise ModelicaSystemError("Cannot create ModelicaSystem object without any arguments")

if modelName is None:

raise ModelicaSystemError("A modelname must be provided (argument modelName)!")

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

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

self._inputs: dict[str, list | None] = {}

# _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 = {'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:

if not isinstance(omc_process, OMCProcessLocal):

raise ModelicaSystemError("Invalid (local) omc process definition provided!")

self._getconn = OMCSessionZMQ(omc_process=omc_process)

else:

self._getconn = 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)

if lmodel is None:

lmodel = []

if not isinstance(lmodel, list):

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

self._lmodel = lmodel # may be needed if model is derived from other model

self._model_name = modelName # Model class name

self._file_name = pathlib.Path(fileName).resolve() if fileName is not None else None # Model file/package name

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

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

self._variable_filter = variableFilter

if self._file_name is not None and not self._file_name.is_file(): # if file does not exist

raise IOError(f"{self._file_name} does not exist!")

self._work_dir: pathlib.Path = self.setWorkDirectory(customBuildDirectory)

if self._file_name is not None:

self._loadLibrary(lmodel=self._lmodel)

self._loadFile(fileName=self._file_name)

# allow directly loading models from MSL without fileName

elif fileName is None and modelName is not None:

self._loadLibrary(lmodel=self._lmodel)

if build:

self.buildModel(variableFilter)

def setCommandLineOptions(self, commandLineOptions: str):

"""

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

self.sendExpression(exp)

def _loadFile(self, fileName: pathlib.Path):

# load file

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

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

def _loadLibrary(self, lmodel: list):

# load Modelica standard libraries or Modelica files if needed

for element in lmodel:

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) -> pathlib.Path:

"""

if customBuildDirectory is not None:

workdir = pathlib.Path(customBuildDirectory).absolute()

if not workdir.is_dir():

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

else:

workdir = pathlib.Path(tempfile.mkdtemp()).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) -> pathlib.Path:

"""

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)

xml_file = pathlib.Path(buildModelResult[0]).parent / buildModelResult[1]

self._xmlparse(xml_file=xml_file)

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

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)}: {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: pathlib.Path):

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

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

else:

if names is None:

for i in self._continuous:

try:

value = self.getSolutions(i)

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

except (OMCSessionException, ModelicaSystemError) as ex:

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

else:

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

if isinstance(names, list):

valuelist = []

for i in names:

if i in self._continuous:

value = self.getSolutions(i)

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

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

else:

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.

if names is None:

return self._params

elif isinstance(names, str):

return [self._params[names]]

elif 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 | list: # 6

"""Get values of input signals.

if names is None:

return self._inputs

elif isinstance(names, str):

return [self._inputs[names]]

elif 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): # 7

"""Get values of output signals.

if not self._simulated:

if names is None:

return self._outputs

elif isinstance(names, str):

return [self._outputs[names]]

else:

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

else:

if names is None:

for i in self._outputs:

value = self.getSolutions(i)

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

return self._outputs

elif isinstance(names, str):

if names in self._outputs:

value = self.getSolutions(names)

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

return [self._outputs[names]]

else:

raise KeyError(names)

elif isinstance(names, list):

valuelist = []

for i in names:

if i in self._outputs:

value = self.getSolutions(i)

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

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

else:

raise KeyError(i)

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

elif isinstance(names, str):

return [self._simulate_options[names]]

elif 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 | list:

"""Get simulation options used for linearization.

if names is None:

return self._linearization_options

elif isinstance(names, str):

return [self._linearization_options[names]]

elif 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 | list:

"""Get simulation options used for optimization.

if names is None:

return self._optimization_options

elif isinstance(names, str):

return [self._optimization_options[names]]

elif isinstance(names, list):

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

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

def simulate_cmd(

self,