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

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

modelName: Optional[str] = None,

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

commandLineOptions: Optional[str] = None,

variableFilter: Optional[str] = None,

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

omhome: Optional[str] = None,

session: Optional[OMCSessionBase] = 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")

self.quantitiesList = []

self.paramlist = {}

self.inputlist = {}

self.outputlist = {}

self.continuouslist = {}

self.simulateOptions = {}

self.overridevariables = {}

self.simoptionsoverride = {}

self.linearOptions = {'startTime': 0.0, 'stopTime': 1.0, 'stepSize': 0.002, 'tolerance': 1e-8}

self.optimizeOptions = {'startTime': 0.0, 'stopTime': 1.0, 'numberOfIntervals': 500, 'stepSize': 0.002,

'tolerance': 1e-8}

self.linearinputs = [] # linearization input list

self.linearoutputs = [] # linearization output list

self.linearstates = [] # linearization states list

self.tempdir = ""

if session is not None:

if not isinstance(session, OMCSessionZMQ):

raise ModelicaSystemError("Invalid session data provided!")

self.getconn = session

else:

self.getconn = OMCSessionZMQ(omhome=omhome)

# set commandLineOptions if provided by users

self.setCommandLineOptions(commandLineOptions=commandLineOptions)

if lmodel is None:

lmodel = []

if not isinstance(lmodel, list):

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

self.xmlFile = None

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

self.modelName = modelName # Model class name

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

self.inputFlag = False # for model with input quantity

self.simulationFlag = False # if the model is simulated?

self.outputFlag = False

self.csvFile = '' # for storing inputs condition

self.resultfile = "" # for storing result file

self.variableFilter = variableFilter

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

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

# set default command Line Options for linearization as

# linearize() will use the simulation executable and runtime

# flag -l to perform linearization

self.setCommandLineOptions("--linearizationDumpLanguage=python")

self.setCommandLineOptions("--generateSymbolicLinearization")

self.setTempDirectory(customBuildDirectory)

if self.fileName is not None:

self.loadLibrary(lmodel=self.lmodel)

self.loadFile(fileName=self.fileName)

# allow directly loading models from MSL without fileName

elif fileName is None and modelName is not None:

self.loadLibrary(lmodel=self.lmodel)

self.buildModel(variableFilter)

def setCommandLineOptions(self, commandLineOptions: str):

# set commandLineOptions if provided by users

if commandLineOptions is None:

return

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(apiCall, 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 setTempDirectory(self, customBuildDirectory):

# create a unique temp directory for each session and build the model in that directory

if customBuildDirectory is not None:

if not os.path.exists(customBuildDirectory):

raise IOError(customBuildDirectory, " does not exist")

self.tempdir = customBuildDirectory

else:

self.tempdir = tempfile.mkdtemp()

if not os.path.exists(self.tempdir):

raise IOError(self.tempdir, " cannot be created")

logger.info("Define tempdir as %s", self.tempdir)

exp = f'cd("{pathlib.Path(self.tempdir).absolute().as_posix()}")'

self.sendExpression(exp)

def getWorkDirectory(self):

return self.tempdir

def _run_cmd(self, cmd: list, timeout: Optional[int] = None):

logger.debug("Run OM command %s in %s", cmd, self.tempdir)

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

dllPath = ""

# set the process environment from the generated .bat file in windows which should have all the dependencies

batFilePath = pathlib.Path(self.tempdir) / f"{self.modelName}.bat"

if not batFilePath.exists():

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

with open(batFilePath, 'r') as file:

for line in file:

match = re.match(r"^SET PATH=([^%]*)", line, re.IGNORECASE)

if match:

dllPath = match.group(1).strip(';') # Remove any trailing semicolons

my_env = os.environ.copy()

my_env["PATH"] = dllPath + 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(cmd, capture_output=True, text=True, env=my_env, cwd=self.tempdir,

timeout=timeout)

stdout = cmdres.stdout.strip()

stderr = cmdres.stderr.strip()

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

if cmdres.returncode != 0:

raise ModelicaSystemError(f"Error running command {cmd}: return code = {cmdres.returncode}")

if stderr:

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

except subprocess.TimeoutExpired:

raise ModelicaSystemError(f"Timeout running command {repr(cmd)}")

except Exception as ex:

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

def buildModel(self, variableFilter=None):

if variableFilter is not None:

self.variableFilter = variableFilter

if self.variableFilter is not None:

varFilter = f'variableFilter="{self.variableFilter}"'

else:

varFilter = 'variableFilter=".*"'

logger.debug("varFilter=%s", varFilter)

buildModelResult = self.requestApi("buildModel", self.modelName, properties=varFilter)

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

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

self.xmlparse()

def sendExpression(self, expr, parsed=True):

logger.debug("sendExpression(%r, %r)", expr, parsed)

return self.getconn.sendExpression(expr, parsed)

# request to OMC

def requestApi(self, apiName, entity=None, properties=None): # 2

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

if not self.xmlFile.exists():

ModelicaSystemError(f"XML file not generated: {self.xmlFile}")

tree = ET.parse(self.xmlFile)

rootCQ = tree.getroot()

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

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

"solver", "outputFormat"):

self.simulateOptions[key] = attr.get(key)

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

scalar = {}

for key in ("name", "description", "variability", "causality", "alias"):

scalar[key] = sv.get(key)

scalar["changeable"] = sv.get('isValueChangeable')

scalar["aliasvariable"] = sv.get('aliasVariable')

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

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

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

self.paramlist[scalar["name"]] = self.overridevariables[scalar["name"]]

else:

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

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

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

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

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

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

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

self.quantitiesList.append(scalar)

def getQuantities(self, names=None): # 3

"""

if names is None:

return self.quantitiesList

elif isinstance(names, str):

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

elif isinstance(names, list):

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

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

def getContinuous(self, names=None): # 4

"""

if not self.simulationFlag:

if names is None:

return self.continuouslist

elif isinstance(names, str):

return [self.continuouslist.get(names, "NotExist")]

elif isinstance(names, list):

return [self.continuouslist.get(x, "NotExist") for x in names]

else:

if names is None:

for i in self.continuouslist:

try:

value = self.getSolutions(i)

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

except Exception as ex:

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

return self.continuouslist

elif isinstance(names, str):

if names in self.continuouslist:

value = self.getSolutions(names)

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

return [self.continuouslist.get(names)]

else:

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

elif isinstance(names, list):

valuelist = []

for i in names:

if i in self.continuouslist:

value = self.getSolutions(i)

self.continuouslist[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.paramlist

elif isinstance(names, str):

return [self.paramlist.get(names, "NotExist")]

elif isinstance(names, list):

return [self.paramlist.get(x, "NotExist") for x in names]

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

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

"""Get input values.

if names is None:

return self.inputlist

elif isinstance(names, str):

return [self.inputlist.get(names, "NotExist")]

elif isinstance(names, list):

return [self.inputlist.get(x, "NotExist") for x in names]

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

def getOutputs(self, names: Optional[str | list[str]] = None): # 7

"""Get output values.

if not self.simulationFlag:

if names is None:

return self.outputlist

elif isinstance(names, str):

return [self.outputlist.get(names, "NotExist")]

else:

return [self.outputlist.get(x, "NotExist") for x in names]

else:

if names is None:

for i in self.outputlist:

value = self.getSolutions(i)

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

return self.outputlist

elif isinstance(names, str):

if names in self.outputlist:

value = self.getSolutions(names)

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

return [self.outputlist.get(names)]

else:

return names, " is not Output"

elif isinstance(names, list):

valuelist = []

for i in names:

if i in self.outputlist:

value = self.getSolutions(i)

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

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

else:

return i, "is not Output"

return valuelist

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

def getSimulationOptions(self, names=None): # 8

"""

if names is None:

return self.simulateOptions

elif isinstance(names, str):

return [self.simulateOptions.get(names, "NotExist")]

elif isinstance(names, list):

return [self.simulateOptions.get(x, "NotExist") for x in names]

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

def getLinearizationOptions(self, names=None): # 9

"""

if names is None:

return self.linearOptions

elif isinstance(names, str):

return [self.linearOptions.get(names, "NotExist")]

elif isinstance(names, list):

return [self.linearOptions.get(x, "NotExist") for x in names]

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

def getOptimizationOptions(self, names=None): # 10

"""

if names is None:

return self.optimizeOptions

elif isinstance(names, str):

return [self.optimizeOptions.get(names, "NotExist")]

elif isinstance(names, list):

return [self.optimizeOptions.get(x, "NotExist") for x in names]

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

def get_exe_file(self) -> pathlib.Path:

"""Get path to model executable."""

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

return pathlib.Path(self.tempdir) / f"{self.modelName}.exe"

else:

return pathlib.Path(self.tempdir) / self.modelName

def simulate(self, resultfile=None, simflags=None, timeout: Optional[int] = None): # 11

"""

if resultfile is None:

# default result file generated by OM

self.resultfile = (pathlib.Path(self.tempdir) / f"{self.modelName}_res.mat").as_posix()

elif os.path.exists(resultfile):

self.resultfile = resultfile

else:

self.resultfile = (pathlib.Path(self.tempdir) / resultfile).as_posix()

# always define the resultfile to use

resultfileflag = " -r=" + self.resultfile

# allow runtime simulation flags from user input

if simflags is None:

simflags = ""

else:

simflags = " " + simflags

overrideFile = pathlib.Path(self.tempdir) / f"{self.modelName}_override.txt"

if self.overridevariables or self.simoptionsoverride:

tmpdict = self.overridevariables.copy()

tmpdict.update(self.simoptionsoverride)

# write to override file

with open(overrideFile, "w") as file:

for key, value in tmpdict.items():

file.write(f"{key}={value}\n")

override = " -overrideFile=" + overrideFile.as_posix()

else:

override = ""

if self.inputFlag: # if model has input quantities

for i in self.inputlist:

val = self.inputlist[i]

if val is None:

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

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

self.inputlist[i] = [(float(self.simulateOptions["startTime"]), 0.0),

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

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

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

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

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

self.csvFile = self.createCSVData() # create csv file

csvinput = " -csvInput=" + self.csvFile.as_posix()

else:

csvinput = ""

exe_file = self.get_exe_file()

if not exe_file.exists():

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

cmd = exe_file.as_posix() + override + csvinput + resultfileflag + simflags

cmd = [s for s in cmd.split(' ') if s]

self._run_cmd(cmd=cmd, timeout=timeout)

self.simulationFlag = True

# to extract simulation results

def getSolutions(self, varList=None, resultfile=None): # 12

"""

if resultfile is None:

resFile = self.resultfile

else:

resFile = resultfile

# check for result file exits

if not os.path.exists(resFile):

raise ModelicaSystemError(f"Result file does not exist {resFile}")

resultVars = self.sendExpression(f'readSimulationResultVars("{resFile}")')

self.sendExpression("closeSimulationResultFile()")

if varList is None:

return resultVars

elif isinstance(varList, str):

if varList not in resultVars and varList != "time":

raise ModelicaSystemError(f"Requested data {repr(varList)} does not exist")

res = self.sendExpression(f'readSimulationResult("{resFile}", {{{varList}}})')

npRes = np.array(res)

self.sendExpression("closeSimulationResultFile()")

return npRes

elif isinstance(varList, list):

for var in varList:

if var == "time":

continue

if var not in resultVars:

raise ModelicaSystemError(f"Requested data {repr(var)} does not exist")

variables = ",".join(varList)

res = self.sendExpression(f'readSimulationResult("{resFile}",{{{variables}}})')

npRes = np.array(res)

self.sendExpression("closeSimulationResultFile()")

return npRes

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

@staticmethod

def _strip_space(name):

if isinstance(name, str):

return name.replace(" ", "")

elif isinstance(name, list):

return [x.replace(" ", "") for x in name]

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

def setMethodHelper(self, args1, args2, args3, args4=None):

"""

def apply_single(args1):

args1 = self._strip_space(args1)

value = args1.split("=")

if value[0] in args2:

if args3 == "parameter" and self.isParameterChangeable(value[0], value[1]):

args2[value[0]] = value[1]

if args4 is not None:

args4[value[0]] = value[1]

elif args3 != "parameter":

args2[value[0]] = value[1]

if args4 is not None:

args4[value[0]] = value[1]

return True

else:

raise ModelicaSystemError("Unhandled case in setMethodHelper.apply_single() - "

f"{repr(value[0])} is not a {repr(args3)} variable")

result = []

if isinstance(args1, str):

result = [apply_single(args1)]

elif isinstance(args1, list):

result = []

args1 = self._strip_space(args1)

for var in args1:

result.append(apply_single(var))

return all(result)

def setContinuous(self, cvals): # 13

"""

return self.setMethodHelper(cvals, self.continuouslist, "continuous", self.overridevariables)

def setParameters(self, pvals): # 14

"""

return self.setMethodHelper(pvals, self.paramlist, "parameter", self.overridevariables)

def isParameterChangeable(self, name, value):

q = self.getQuantities(name)

if q[0]["changeable"] == "false":

logger.verbose(f"setParameters() failed : It is not possible to set the following signal {repr(name)}. "

"It seems to be structural, final, protected or evaluated or has a non-constant binding, "

f"use sendExpression(\"setParameterValue({self.modelName}, {name}, {value})\", "

"parsed=False) and rebuild the model using buildModel() API")

return False

return True

def setSimulationOptions(self, simOptions): # 16

"""

return self.setMethodHelper(simOptions, self.simulateOptions, "simulation-option", self.simoptionsoverride)

def setLinearizationOptions(self, linearizationOptions): # 18

"""

return self.setMethodHelper(linearizationOptions, self.linearOptions, "Linearization-option", None)

def setOptimizationOptions(self, optimizationOptions): # 17

"""

return self.setMethodHelper(optimizationOptions, self.optimizeOptions, "optimization-option", None)

def setInputs(self, name): # 15

"""

if isinstance(name, str):

name = self._strip_space(name)

value = name.split("=")

if value[0] in self.inputlist:

tmpvalue = eval(value[1])

if isinstance(tmpvalue, int) or isinstance(tmpvalue, float):

self.inputlist[value[0]] = [(float(self.simulateOptions["startTime"]), float(value[1])),

(float(self.simulateOptions["stopTime"]), float(value[1]))]

elif isinstance(tmpvalue, list):

self.checkValidInputs(tmpvalue)

self.inputlist[value[0]] = tmpvalue

self.inputFlag = True

else:

raise ModelicaSystemError(f"{value[0]} is not an input")

elif isinstance(name, list):

name = self._strip_space(name)

for var in name:

value = var.split("=")

if value[0] in self.inputlist:

tmpvalue = eval(value[1])

if isinstance(tmpvalue, int) or isinstance(tmpvalue, float):

self.inputlist[value[0]] = [(float(self.simulateOptions["startTime"]), float(value[1])),

(float(self.simulateOptions["stopTime"]), float(value[1]))]

elif isinstance(tmpvalue, list):

self.checkValidInputs(tmpvalue)

self.inputlist[value[0]] = tmpvalue

self.inputFlag = True

else:

raise ModelicaSystemError(f"{value[0]} is not an input!")

def checkValidInputs(self, name):

if name != sorted(name, key=lambda x: x[0]):

raise ModelicaSystemError('Time value should be in increasing order')

for l in name:

if isinstance(l, tuple):

# if l[0] < float(self.simValuesList[0]):

if l[0] < float(self.simulateOptions["startTime"]):

ModelicaSystemError('Input time value is less than simulation startTime')

if len(l) != 2:

ModelicaSystemError(f'Value for {l} is in incorrect format!')

else:

ModelicaSystemError('Error!!! Value must be in tuple format')

def createCSVData(self) -> pathlib.Path:

start_time: float = float(self.simulateOptions["startTime"])

stop_time: float = float(self.simulateOptions["stopTime"])

# Replace None inputs with a default constant zero signal

inputs: dict[str, list[tuple[float, float]]] = {}

for input_name, input_signal in self.inputlist.items():

if input_signal is None:

inputs[input_name] = [(start_time, 0.0), (stop_time, 0.0)]

else:

inputs[input_name] = input_signal

# Collect all unique timestamps across all input signals

all_times = np.array(

sorted({t for signal in inputs.values() for t, _ in signal}),

dtype=float

)

# Interpolate missing values

interpolated_inputs: dict[str, np.ndarray] = {}

for signal_name, signal_values in inputs.items():

signal = np.array(signal_values)

interpolated_inputs[signal_name] = np.interp(

all_times,

signal[:, 0], # times

signal[:, 1] # values

)

# Write CSV file

input_names = list(interpolated_inputs.keys())

header = ['time'] + input_names + ['end']

csv_rows = [header]

for i, t in enumerate(all_times):

row = [

t, # time

*(interpolated_inputs[name][i] for name in input_names), # input values

0 # trailing 'end' column

]

csv_rows.append(row)

csvFile = pathlib.Path(self.tempdir) / f'{self.modelName}.csv'

with open(csvFile, "w", newline="") as f:

writer = csv.writer(f)

writer.writerows(csv_rows)

return csvFile

# to convert Modelica model to FMU

def convertMo2Fmu(self, version="2.0", fmuType="me_cs", fileNamePrefix="<default>", includeResources=True): # 19

"""

if fileNamePrefix == "<default>":

fileNamePrefix = self.modelName

if includeResources:

includeResourcesStr = "true"

else:

includeResourcesStr = "false"

properties = f'version="{version}", fmuType="{fmuType}", fileNamePrefix="{fileNamePrefix}", includeResources={includeResourcesStr}'

fmu = self.requestApi('buildModelFMU', self.modelName, properties)

# report proper error message

if not os.path.exists(fmu):

raise ModelicaSystemError(f"Missing FMU file: {fmu}")

return fmu

# to convert FMU to Modelica model

def convertFmu2Mo(self, fmuName): # 20

"""

fileName = self.requestApi('importFMU', fmuName)

# report proper error message

if not os.path.exists(fileName):

raise ModelicaSystemError(f"Missing file {fileName}")