Skip to content
Merged
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
156 changes: 136 additions & 20 deletions src/OMSimulatorPython/fmu.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,12 @@
from OMSimulator.connector import Connector
from OMSimulator.unit import Unit
from OMSimulator.variable import Variable, SignalType
from OMSimulator.values import Values
from OMSimulator import namespace, utils
from OMSimulator.capi import Capi, Status
from OMSimulator.cref import CRef
from OMSimulator.enumeration import Enumeration
import warnings

logger = logging.getLogger(__name__)

Expand All @@ -61,6 +63,49 @@
'ma':6
}

## default initial attribute table for FMI 2.0, based on variability and causality
initialDefaultTableFmi2 = {
"fixed": {
"input": "unknown",
"output": "unknown",
"parameter": "exact",
"calculatedParameter": "calculated",
"local": "calculated",
"independent": "unknown",
},
"tunable": {
"input": "unknown",
"output": "unknown",
"parameter": "exact",
"calculatedParameter": "calculated",
"local": "calculated",
"independent": "unknown",
},
"constant": {
"input": "unknown",
"output": "exact",
"parameter": "unknown",
"calculatedParameter": "exact",
"local": "exact",
"independent": "unknown",
},
"discrete": {
"input": "unknown",
"output": "calculated",
"parameter": "unknown",
"calculatedParameter": "calculated",
"local": "calculated",
"independent": "unknown",
},
"continuous": {
"input": "unknown",
"output": "calculated",
"parameter": "unknown",
"calculatedParameter": "calculated",
"local": "calculated",
"independent": "unknown",
},
}
class FMU:
def __init__(self, fmu_path: Union[str, Path], instanceName: str = None):
'''Initialize the FMU by loading modelDescription.xml from the FMU archive.'''
Expand Down Expand Up @@ -93,7 +138,9 @@ def __init__(self, fmu_path: Union[str, Path], instanceName: str = None):
self.appliedExperiment = {}
self.apiCall = []
self.instanceName = instanceName
self.fmuInstantitated = False
self.value = Values()
self.fmuInstantiated = False
self.fmuInitialized = False
self.mode = None
# override the FMU's declared defaultExperiment; must be set before instantiate()
# since they need to reach the FMU's fmi2SetupExperiment call, not just the
Expand Down Expand Up @@ -263,6 +310,9 @@ def _get(attr, default):
'stepSize': _get('stepSize', 1e-3),
}

def _getInitialAttribute(self, variability, causality):
return initialDefaultTableFmi2.get(variability, {}).get(causality)

def _parse_variables_fmi2(self, model_description):
'''Parses variables from the ModelVariables section of modelDescription.xml'''
scalar_variables = model_description.xpath('//ModelVariables/ScalarVariable')
Expand All @@ -272,7 +322,7 @@ def _parse_variables_fmi2(self, model_description):
value_reference = scalar_var.get('valueReference')
causality = scalar_var.get('causality', 'local')
variability = scalar_var.get('variability', 'continuous')

initial = scalar_var.get('initial') if scalar_var.get('initial') is not None else self._getInitialAttribute(variability, causality)
var_type = None
unit = None
start = None
Expand All @@ -293,7 +343,7 @@ def _parse_variables_fmi2(self, model_description):
self._states.append(self._variables[derivative_index - 1])

# Create and store the variable
variable = Variable(name, description, value_reference, causality, variability, var_type, unit, start, declaredType)
variable = Variable(name, description, value_reference, causality, variability, initial, var_type, unit, start, declaredType)

# Assign unit definitions if applicable
if unit:
Expand Down Expand Up @@ -327,7 +377,7 @@ def _parse_variables_fmi3(self, model_description):
value_reference = element.get("valueReference")
causality = element.get("causality", "local")
variability = element.get("variability", "continuous")
initial = element.get("initial")
initial = element.get('initial') if element.get('initial') is not None else self._getInitialAttribute(variability, causality)
start = element.get("start")
derivative_index = int(element.get('derivative', '-1'))
declaredType = element.get('declaredType')
Expand All @@ -346,7 +396,7 @@ def _parse_variables_fmi3(self, model_description):
# print(f"var_name: {name}, var_type: {var_type}, causality: {causality}, variability: {variability}, initial: {initial}, start: {start}")

# Create and store the variable
variable = Variable(name, description, value_reference, causality, variability, var_type, unit, start, declaredType)
variable = Variable(name, description, value_reference, causality, variability, initial, var_type, unit, start, declaredType)

# Assign unit definitions if applicable
if unit:
Expand Down Expand Up @@ -418,6 +468,12 @@ def makeConnectors(self):
connectors.append(connector)
return connectors

def getVariableByName(self, name: str) -> Variable | None:
for var in self.variables:
if str(var.name) == name:
return var
return None

def varExist(self, cref: str) -> bool:
return any(var.name == cref for var in self.variables)

Expand Down Expand Up @@ -506,6 +562,30 @@ def splitModelName(self):
else:
return parts[-1]

def applyStartValues(self):
"""
Apply start values to the FMU's variables. set before instantiation. This method should be called after the FMU is loaded and before it is instantiated.
"""
for key, (value, type, _, _) in self.value.start_values.items():
mappedCrefs = f"{self.instanceName}.root.{self.instanceName}.{key}"
#Determine the variable type
type, status = Capi.getVariableType(mappedCrefs)
if status != Status.ok:
raise RuntimeError(f"Failed to get variable type for {key}: {status}")

match SignalType(type):
case SignalType.Real: # oms_signal_type_real
return self._setReal(mappedCrefs, value)
case SignalType.Integer: # oms_signal_type_integer
return self._setInteger(mappedCrefs, value)
case SignalType.Boolean: # oms_signal_type_boolean
return self._setBoolean(mappedCrefs, value)
case SignalType.String: # oms_signal_type_string
return self._setString(mappedCrefs, value)
case SignalType.Enumeration: # oms_signal_type_enumeration
return self._setInteger(mappedCrefs, value) # Treat enumeration as integer
case _:
raise TypeError(f"Unsupported type: {type}")

def instantiate(self):
'''Instantiate the FMU for simulation.
Expand Down Expand Up @@ -588,45 +668,46 @@ def instantiate(self):
status = Capi.setVariableStepSize(self.instanceName, 1e-6, 1e-12, self.appliedExperiment['stepSize'])
if status != Status.ok:
raise RuntimeError(f"Failed to set variable step size: {status}")

## apply start values before instantiation, so that initialization can use them before entering fmi2_enterInitializationMode
self.applyStartValues()
status = Capi.instantiate(self.instanceName)
if status != Status.ok:
raise RuntimeError(f"Failed to instantiate model: {status}")
self.fmuInstantitated = True
self.fmuInstantiated = True

self.apiCall.append(f'oms.instantiate("{self.instanceName}")')

def setStartTime(self, startTime: float):
if self.fmuInstantitated is False:
if self.fmuInstantiated is False:
raise RuntimeError("FMU must be instantiated before setting start time")
status = Capi.setStartTime(self.instanceName, startTime)
if status != Status.ok:
raise RuntimeError(f"Failed to set start time: {status}")

def setStopTime(self, stopTime: float):
if self.fmuInstantitated is False:
if self.fmuInstantiated is False:
raise RuntimeError("FMU must be instantiated before setting stop time")
status = Capi.setStopTime(self.instanceName, stopTime)
if status != Status.ok:
raise RuntimeError(f"Failed to set stop time: {status}")

def setTolerance(self, tolerance: float):
if self.fmuInstantitated is False:
if self.fmuInstantiated is False:
raise RuntimeError("FMU must be instantiated before setting tolerance")
status = Capi.setTolerance(self.instanceName, tolerance)
if status != Status.ok:
raise RuntimeError(f"Failed to set tolerance: {status}")

def setStepSize(self, stepSize: float):
if self.fmuInstantitated is False:
if self.fmuInstantiated is False:
raise RuntimeError("FMU must be instantiated before setting variable step size")
status = Capi.setVariableStepSize(self.instanceName, 1e-6, 1e-12, stepSize)
if status != Status.ok:
raise RuntimeError(f"Failed to set variable step size: {status}")

def setSolver(self, method: str):
'''Set the ODE solver ('euler' or 'cvode'). Only applies to model-exchange FMUs.'''
if self.fmuInstantitated is False:
if self.fmuInstantiated is False:
raise RuntimeError("FMU must be instantiated before setting the solver")
if method not in _FMU_SOLVER:
raise ValueError(f"Invalid solver '{method}': expected one of {sorted(_FMU_SOLVER)}")
Expand All @@ -635,7 +716,7 @@ def setSolver(self, method: str):
raise RuntimeError(f"Failed to set solver: {status}")

def getValue(self, cref: str):
if self.fmuInstantitated is False:
if self.fmuInstantiated is False:
raise RuntimeError("FMU must be instantiated before getting values")

'''Get the value of a variable by its name.'''
Expand Down Expand Up @@ -687,15 +768,49 @@ def _getString(self, cref: CRef):
raise RuntimeError(f"Failed to get value for {cref}: {status}")
return value

def setValue(self, cref: str, value):
def checkfmiRequirements(self, cref: CRef):
"""Checks if a component variable meets the FMI requirements."""
variable = self.getVariableByName(str(cref))

if variable and self.fmuInstantiated and not self.fmuInitialized:
allowed = (variable.isInput() or variable.isExact())
if not allowed:
warnings.warn(
f"Cannot set variable '{cref}' after the FMU "
f"is instantiated. Only variables with "
f"causality='input' or initial='exact' can "
f"be set at this stage. Variable '{cref}' has "
f"causality='{variable.causality.name}' and "
f"initial='{variable.initial}'",RuntimeWarning)
return False

if variable and self.fmuInitialized:
allowed = (variable.isInput()) or (variable.isParameter() and variable.isTunable())
if not allowed:
warnings.warn(
f"Cannot set variable '{cref}' after the FMU "
f"is initialized. Only variables with "
f"causality='input' or "
f"causality='parameter' and variability='tunable' can "
f"be set at this stage. Variable '{cref}' has "
f"causality='{variable.causality.name}' and "
f"variability='{variable.variability}'",RuntimeWarning)
return False

return True

def setValue(self, cref: str, value, unit=None, description = None):
"""Sets a value for a specific CRef in the model."""

if self.fmuInstantitated is False:
raise RuntimeError("FMU must be instantiated before setting values")

if not self.varExist(CRef(cref)):
raise KeyError(f"Variable '{cref}' does not exist in the FMU")

if self.fmuInstantiated is False:
return self.value.setValue(cref, value, unit, description)

if not self.checkfmiRequirements(CRef(cref)):
return # Skip setting the value if FMI requirements are not met

mappedCrefs = f"{self.instanceName}.root.{self.instanceName}.{cref}"

# Determine the variable type
Expand Down Expand Up @@ -742,23 +857,24 @@ def _setString(self, mapped_cref: str, value: str):
raise RuntimeError(f"Failed to set value for {mapped_cref}: {status}")

def setResultFile(self, filename: str):
if self.fmuInstantitated is False:
if self.fmuInstantiated is False:
raise RuntimeError("FMU must be instantiated before setting result file")

status = Capi.setResultFile(self.instanceName, filename)
if status != Status.ok:
raise RuntimeError(f"Failed to setResultFile {filename}: {status}")

def initialize(self):
if self.fmuInstantitated is False:
if self.fmuInstantiated is False:
raise RuntimeError("FMU must be instantiated before initialization")

status = Capi.initialize(self.instanceName)
if status != Status.ok:
raise RuntimeError(f"Failed to initialize model: {status}")
self.fmuInitialized = True

def simulate(self):
if self.fmuInstantitated is False:
if self.fmuInstantiated is False:
raise RuntimeError("FMU must be instantiated before simulation")

status = Capi.simulate(self.instanceName)
Expand Down
Loading