-
Notifications
You must be signed in to change notification settings - Fork 139
Description
Hello,
I built an FMU using PythonFMU to run a Cantera simulation. I'm now trying to build a dataset to train an ML model, but when I try to change the tunable parameters in a similar fashion as parameter_variation.py, the simulation will only run with the values initialized in the PythonFMU init function. I've confirmed that the parameters are registered as tunable and have tried changing them with setReal, apply_starting_values, and by passing it a start_values dictionary but all of them give the same results. The only way that I've been able to change the parameters is by editing their initializations in the PythonFMU definition file. Code is below and FMU is attached. Thanks for the help and let me know if you have any questions.
Here is the code for the pythonFMU:
`from pythonfmu import Fmi2Slave, Fmi2Causality,Fmi2Variability, Real
import cantera as ct
class HexaneCombuster(Fmi2Slave):
def __init__(self, **kwargs):
super().__init__(**kwargs)
#initialize and register parameters
self.reactor_temperature = 925
self.reactor_pressure = 1.046138*ct.one_atm
self.hexane_inlet_concentrations = 0.050
self.residence_time = 2
self.reactor_volume = 30.5*(1e-2) ** 3
#register parameters
#parameters = ['reactor_temperature', 'reactor_pressure', 'hexane_inlet_concentrations','residence_time','reactor_volume']
#for parameter in parameters:
# self.register_variable(Real(parameter,variability=Fmi2Variability.tunable, causality=Fmi2Causality.parameter))
self.register_variable(Real('reactor_temperature',variability=Fmi2Variability.tunable, causality=Fmi2Causality.parameter, start=925))
self.register_variable(Real('reactor_pressure',variability=Fmi2Variability.tunable, causality=Fmi2Causality.parameter, start = 1.046138*ct.one_atm))
self.register_variable(Real('hexane_inlet_concentrations',variability=Fmi2Variability.tunable, causality=Fmi2Causality.parameter, start = 0.050))
self.register_variable(Real('residence_time',variability=Fmi2Variability.tunable, causality=Fmi2Causality.parameter, start = 2))
self.register_variable(Real('reactor_volume',variability=Fmi2Variability.tunable, causality=Fmi2Causality.parameter, start = 30.5*(1e-2)**3))
#initialize cantera stuff
self.gas = ct.Solution("example_data/n-hexane-NUIG-2015.yaml")
inlet_concentrations = {"NC7H16": self.hexane_inlet_concentrations, "O2": 0.0275, "HE":(0.968-self.hexane_inlet_concentrations)}
self.gas.TPX = self.reactor_temperature, self.reactor_pressure, inlet_concentrations
self.fuel_air_mixture_tank = ct.Reservoir(self.gas)
self.exhaust = ct.Reservoir(self.gas)
self.stirred_reactor = ct.IdealGasMoleReactor(self.gas, energy="on", volume=self.reactor_volume)
self.mass_flow_controller = ct.MassFlowController(upstream=self.fuel_air_mixture_tank, downstream=self.stirred_reactor, mdot=self.stirred_reactor.mass/self.residence_time)
self.pressure_regulator = ct.PressureController(upstream=self.stirred_reactor, downstream=self.exhaust, primary=self.mass_flow_controller,K=1e-3)
self.reactor_network = ct.ReactorNet([self.stirred_reactor])
#initialize and register outputs
self._update_outputs()
outputs = [
'temperature',
'pressure',
'density',
'mean_mol_weight',
'HE_mass_frac',
'NC7H16_mass_frac',
'H2_mass_frac',
'O2_mass_frac',
'H2O_mass_frac',
'CO_mass_frac',
'CO2_mass_frac',
'CH4_mass_frac',
'C2H4_mass_frac',
'C3H6_mass_frac',
]
for output in outputs:
self.register_variable(Real(output,causality=Fmi2Causality.output))
def _get_mass_fraction(self,key):
ans = 0
try:
ans = self.reactor_network.reactors[0].thermo.mass_fraction_dict()[key]
except KeyError:
ans = 0
return ans
def _update_outputs(self):
self.temperature = self.reactor_network.reactors[0].thermo.T
self.pressure = self.reactor_network.reactors[0].thermo.P
self.density = self.reactor_network.reactors[0].thermo.density
self.mean_mol_weight = self.reactor_network.reactors[0].thermo.mean_molecular_weight
self.HE_mass_frac = self._get_mass_fraction('NC7H16')
self.NC7H16_mass_frac = self._get_mass_fraction('NC7H16')
self.H2_mass_frac = self._get_mass_fraction('H2')
self.O2_mass_frac = self._get_mass_fraction('O2')
self.H2O_mass_frac = self._get_mass_fraction('H2O')
self.CO_mass_frac = self._get_mass_fraction('CO')
self.CO2_mass_frac = self._get_mass_fraction('CO2')
self.CH4_mass_frac = self._get_mass_fraction('CH4')
self.C2H4_mass_frac = self._get_mass_fraction('C2H4')
self.C3H6_mass_frac = self._get_mass_fraction('C3H6')
def do_step(self, current_time, step_size):
#do step
self.reactor_network.advance(current_time+step_size)
#update actual outputs
self._update_outputs()
return True
if name == "main":
testObj = HexaneCombuster(instance_name="test1")
for t in range(10):
testObj.do_step(t,1)
print(testObj.temperature)`
And here is the code for running it in fmPy:
`import fmpy
#from fmpy.simulation import apply_start_values
from fmpy.fmi2 import FMU2Slave
import numpy as np
import cantera as ct
#import pandas as pd
test_parameters = [0.05, 106000, 3.05, 2, 925]
def Sim_run(this_round_input_parameters):
n_steps = 1001
t = np.linspace(0, 1, n_steps)
input_names = ['hexane_inlet_concentration', 'reactor_pressure', 'reactor_volume', 'residence_time', 'reactor_temperature']
fmu_filename = 'HexaneCombuster.fmu'
# Define structured dtype for the FMU input
dtype = [('time', np.float64)] + [(name, np.float64) for name in input_names]
# Initialize input array
input_values = np.zeros(n_steps, dtype=dtype)
# Fill time column
input_values['time'] = t
# Fill each variable column with constant input values
for i, name in enumerate(input_names):
input_values[name] = this_round_input_parameters[i]
# Define start values (if needed)
start_values_dict = {'hexane_inlet_concentrations':this_round_input_parameters[0], 'reactor_pressure':this_round_input_parameters[1], 'reactor_volume':this_round_input_parameters[2], 'residence_time':this_round_input_parameters[3], 'reactor_temperature':this_round_input_parameters[4]}
#start_values_dict = {}
model_description = fmpy.read_model_description(fmu_filename)
for v in model_description.modelVariables:
if v.causality == 'parameter':
print(v.name, v.variability, v.initial)
unzipdir = fmpy.extract(fmu_filename)
fmu = FMU2Slave(
guid = model_description.guid,
unzipDirectory = unzipdir,
modelIdentifier = model_description.coSimulation.modelIdentifier,
instanceName = 'instance1'
)
fmu.instantiate()
fmu.setupExperiment(startTime=0)
#apply_start_values(fmu, model_description,start_values_dict)
fmu.enterInitializationMode()
fmu.exitInitializationMode()
time = 0.0
step_size = 1e-3
stop_time = 1
results = []
in_vrs = [0,1,2,3,4]
result_vrs = [5,6,7,8,9,10,11,12,13,14,15,16,17,18]
fmu.setReal(vr=in_vrs, value=this_round_input_parameters)
while time < stop_time:
fmu.doStep(currentCommunicationPoint=time, communicationStepSize=step_size)
time += step_size
if time > 0.02:
#print(fmu.getReal(in_vrs))
result = fmu.getReal(result_vrs)
results.append(result)
# Run simulation
#output = fmpy.simulate_fmu(
# fmu_filename,
# start_time=0.0,
# stop_time=1.0,
# start_values=start_values_dict,
# input=input_values
#)
return list(results[-1])[1:]
data = [[test_parameters, Sim_run(test_parameters)],[[0.1, 212000, 1.5, 3, 1100],Sim_run([0.1, 212000, 1.5, 3, 1100])]]
print(data)`