Coverage for backend/ahuora-builder/src/ahuora_builder/custom/duty_heat_exchanger.py: 72%
157 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from pyomo.core.base.component import Component
2from pyomo.environ import (
3 Block,
4 Constraint,
5 Reference,
6 check_optimal_termination,
7 units as pyunits,
8 value,
9)
11from idaes.core import (
12 ControlVolume0DBlock,
13 EnergyBalanceType,
14 MaterialBalanceType,
15 MomentumBalanceType,
16 UnitModelBlockData,
17 declare_process_block_class,
18 useDefault,
19)
20from idaes.core.initialization import SingleControlVolumeUnitInitializer
21from idaes.core.solvers import get_solver
22from idaes.core.util.config import is_physical_parameter_block
23from idaes.core.util.exceptions import InitializationError
24from idaes.core.util.tables import create_stream_table_dataframe
25import idaes.logger as idaeslog
26from idaes.models.unit_models.heat_exchanger import add_hx_references, hx_process_config
27from pyomo.common.config import Bool, ConfigBlock, ConfigValue, In
30class DutyHeatExchangerInitializer(SingleControlVolumeUnitInitializer):
31 """Initialize a duty-specified two-stream heat exchanger."""
33 def initialization_routine(
34 self,
35 model: Block,
36 plugin_initializer_args: dict | None = None,
37 copy_inlet_state: bool = False,
38 duty=1000 * pyunits.W,
39 ):
40 return super(SingleControlVolumeUnitInitializer, self).initialization_routine(
41 model=model,
42 plugin_initializer_args=plugin_initializer_args,
43 copy_inlet_state=copy_inlet_state,
44 duty=duty,
45 )
47 def initialize_main_model(
48 self,
49 model: Block,
50 copy_inlet_state: bool = False,
51 duty=1000 * pyunits.W,
52 ):
53 init_log = idaeslog.getInitLogger(
54 model.name, self.get_output_level(), tag="unit"
55 )
56 solve_log = idaeslog.getSolveLogger(
57 model.name, self.get_output_level(), tag="unit"
58 )
59 solver = self._get_solver()
61 self.initialize_control_volume(model.hot_side, copy_inlet_state)
62 init_log.info_high("Initialization Step 1a (hot side) Complete.")
64 self.initialize_control_volume(model.cold_side, copy_inlet_state)
65 init_log.info_high("Initialization Step 1b (cold side) Complete.")
67 model.energy_balance_constraint.deactivate()
68 cold_was_fixed = model.cold_side.heat[model.flowsheet().time.first()].fixed
69 if not cold_was_fixed:
70 model.cold_side.heat.fix(duty)
71 for i in model.hot_side.heat:
72 model.hot_side.heat[i].set_value(
73 -_heat_value(
74 model.cold_side.heat[i], duty, model.hot_side.heat.get_units()
75 )
76 )
78 with idaeslog.solver_log(solve_log, idaeslog.DEBUG) as slc:
79 res = solver.solve(model, tee=slc.tee)
80 init_log.info_high("Initialization Step 2 {}.".format(idaeslog.condition(res)))
82 if not cold_was_fixed:
83 model.cold_side.heat.unfix()
84 model.energy_balance_constraint.activate()
86 with idaeslog.solver_log(solve_log, idaeslog.DEBUG) as slc:
87 res = solver.solve(model, tee=slc.tee)
88 init_log.info("Initialization Completed, {}".format(idaeslog.condition(res)))
90 return res
93@declare_process_block_class("DutyHeatExchanger")
94class DutyHeatExchangerData(UnitModelBlockData):
95 """Two-stream heat exchanger with a user-specified signed heat duty."""
97 default_initializer = DutyHeatExchangerInitializer
99 CONFIG = UnitModelBlockData.CONFIG(implicit=True)
100 _SideCONFIG = ConfigBlock()
101 _SideCONFIG.declare(
102 "has_phase_equilibrium",
103 ConfigValue(default=False, domain=Bool),
104 )
105 _SideCONFIG.declare(
106 "material_balance_type",
107 ConfigValue(default=MaterialBalanceType.useDefault, domain=In(MaterialBalanceType)),
108 )
109 _SideCONFIG.declare(
110 "energy_balance_type",
111 ConfigValue(default=EnergyBalanceType.useDefault, domain=In(EnergyBalanceType)),
112 )
113 _SideCONFIG.declare(
114 "momentum_balance_type",
115 ConfigValue(default=MomentumBalanceType.pressureTotal, domain=In(MomentumBalanceType)),
116 )
117 _SideCONFIG.declare(
118 "has_pressure_change",
119 ConfigValue(default=False, domain=Bool),
120 )
121 _SideCONFIG.declare(
122 "property_package",
123 ConfigValue(default=useDefault, domain=is_physical_parameter_block),
124 )
125 _SideCONFIG.declare("property_package_args", ConfigBlock(implicit=True))
127 CONFIG.declare("hot_side", _SideCONFIG(doc="Hot fluid config arguments"))
128 CONFIG.declare("cold_side", _SideCONFIG(doc="Cold fluid config arguments"))
129 CONFIG.declare("hot_side_name", ConfigValue(default=None, domain=str))
130 CONFIG.declare("cold_side_name", ConfigValue(default=None, domain=str))
132 def build(self):
133 super().build()
134 hx_process_config(self)
136 self.hot_side = ControlVolume0DBlock(
137 dynamic=self.config.dynamic,
138 has_holdup=self.config.has_holdup,
139 property_package=self.config.hot_side.property_package,
140 property_package_args=self.config.hot_side.property_package_args,
141 )
142 self.hot_side.add_state_blocks(
143 has_phase_equilibrium=self.config.hot_side.has_phase_equilibrium
144 )
145 self.hot_side.add_material_balances(
146 balance_type=self.config.hot_side.material_balance_type,
147 has_phase_equilibrium=self.config.hot_side.has_phase_equilibrium,
148 )
149 self.hot_side.add_energy_balances(
150 balance_type=self.config.hot_side.energy_balance_type,
151 has_heat_transfer=True,
152 )
153 self.hot_side.add_momentum_balances(
154 balance_type=self.config.hot_side.momentum_balance_type,
155 has_pressure_change=self.config.hot_side.has_pressure_change,
156 )
158 self.cold_side = ControlVolume0DBlock(
159 dynamic=self.config.dynamic,
160 has_holdup=self.config.has_holdup,
161 property_package=self.config.cold_side.property_package,
162 property_package_args=self.config.cold_side.property_package_args,
163 )
164 self.cold_side.add_state_blocks(
165 has_phase_equilibrium=self.config.cold_side.has_phase_equilibrium
166 )
167 self.cold_side.add_material_balances(
168 balance_type=self.config.cold_side.material_balance_type,
169 has_phase_equilibrium=self.config.cold_side.has_phase_equilibrium,
170 )
171 self.cold_side.add_energy_balances(
172 balance_type=self.config.cold_side.energy_balance_type,
173 has_heat_transfer=True,
174 )
175 self.cold_side.add_momentum_balances(
176 balance_type=self.config.cold_side.momentum_balance_type,
177 has_pressure_change=self.config.cold_side.has_pressure_change,
178 )
180 self.add_inlet_port(name="hot_side_inlet", block=self.hot_side)
181 self.add_outlet_port(name="hot_side_outlet", block=self.hot_side)
182 self.add_inlet_port(name="cold_side_inlet", block=self.cold_side)
183 self.add_outlet_port(name="cold_side_outlet", block=self.cold_side)
185 self.heat_duty = Reference(self.cold_side.heat[:])
186 add_hx_references(self)
188 hunits = self.hot_side.config.property_package.get_metadata().get_derived_units
190 def rule_energy_balance(blk, t):
191 return blk.hot_side.heat[t] == -pyunits.convert(
192 blk.cold_side.heat[t], to_units=hunits("power")
193 )
195 self.energy_balance_constraint = Constraint(
196 self.flowsheet().time, rule=rule_energy_balance
197 )
199 def initialize_build(
200 self,
201 hot_side_state_args=None,
202 cold_side_state_args=None,
203 outlvl=idaeslog.NOTSET,
204 solver=None,
205 optarg=None,
206 duty=None,
207 ):
208 init_log = idaeslog.getInitLogger(self.name, outlvl, tag="unit")
209 solve_log = idaeslog.getSolveLogger(self.name, outlvl, tag="unit")
210 opt = get_solver(solver, optarg)
212 hot_side_flags = self.hot_side.initialize(
213 outlvl=outlvl, optarg=optarg, solver=solver, state_args=hot_side_state_args
214 )
215 init_log.info_high("Initialization Step 1a (hot side) Complete.")
217 cold_side_flags = self.cold_side.initialize(
218 outlvl=outlvl, optarg=optarg, solver=solver, state_args=cold_side_state_args
219 )
220 init_log.info_high("Initialization Step 1b (cold side) Complete.")
222 self.energy_balance_constraint.deactivate()
223 cold_was_fixed = self.cold_side.heat[self.flowsheet().time.first()].fixed
224 if not cold_was_fixed: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true
225 self.cold_side.heat.fix(self._initialization_duty(duty, self.cold_side.heat.get_units()))
226 for i in self.hot_side.heat:
227 self.hot_side.heat[i].set_value(
228 -_heat_value(
229 self.cold_side.heat[i],
230 self._initialization_duty(duty, self.hot_side.heat.get_units()),
231 self.hot_side.heat.get_units(),
232 )
233 )
235 with idaeslog.solver_log(solve_log, idaeslog.DEBUG) as slc:
236 res = opt.solve(self, tee=slc.tee)
237 init_log.info_high("Initialization Step 2 {}.".format(idaeslog.condition(res)))
239 if not cold_was_fixed: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 self.cold_side.heat.unfix()
241 self.energy_balance_constraint.activate()
243 with idaeslog.solver_log(solve_log, idaeslog.DEBUG) as slc:
244 res = opt.solve(self, tee=slc.tee)
245 init_log.info_high("Initialization Step 3 {}.".format(idaeslog.condition(res)))
247 self.hot_side.release_state(hot_side_flags, outlvl=outlvl)
248 self.cold_side.release_state(cold_side_flags, outlvl=outlvl)
250 if not check_optimal_termination(res): 250 ↛ 251line 250 didn't jump to line 251 because the condition on line 250 was never true
251 raise InitializationError(
252 f"{self.name} failed to initialize successfully. Please check "
253 "the output logs for more information."
254 )
256 return res
258 @staticmethod
259 def _initialization_duty(duty, to_units):
260 if duty is None: 260 ↛ 262line 260 didn't jump to line 262 because the condition on line 260 was always true
261 return 1000 if to_units is None else pyunits.convert_value(1000, from_units=pyunits.W, to_units=to_units)
262 return pyunits.convert_value(duty[0], from_units=duty[1], to_units=to_units)
264 def diagnose(self) -> list[tuple[Component, str]]:
265 problems = []
266 hot_flow_mass_var = self.hot_side.properties_in[0].flow_mass
267 cold_flow_mass_var = self.cold_side.properties_in[0].flow_mass
268 hot_flow_mass = value(hot_flow_mass_var, exception=False)
269 cold_flow_mass = value(cold_flow_mass_var, exception=False)
271 if hot_flow_mass is None:
272 problems.append(
273 (hot_flow_mass_var, "Hot-side inlet mass flow rate is not available")
274 )
275 elif hot_flow_mass <= 0:
276 problems.append(
277 (hot_flow_mass_var, "Hot-side inlet mass flow rate must be positive")
278 )
280 if cold_flow_mass is None:
281 problems.append(
282 (cold_flow_mass_var, "Cold-side inlet mass flow rate is not available")
283 )
284 elif cold_flow_mass <= 0:
285 problems.append(
286 (cold_flow_mass_var, "Cold-side inlet mass flow rate must be positive")
287 )
289 if problems:
290 return problems
292 mass_flow_difference = hot_flow_mass / cold_flow_mass
293 if mass_flow_difference > 100: 293 ↛ 300line 293 didn't jump to line 300 because the condition on line 293 was always true
294 problems.append(
295 (
296 self.hot_side.properties_in[0].flow_mass,
297 f"Mass flow rate on hot side is {mass_flow_difference:.2f} times higher than cold side. This may cause convergence issues. Consider adjusting the model or providing better initial guesses.",
298 )
299 )
300 elif mass_flow_difference < 0.01:
301 problems.append(
302 (
303 self.cold_side.properties_in[0].flow_mass,
304 f"Mass flow rate on cold side is {1 / mass_flow_difference:.2f} times higher than hot side. This may cause convergence issues. Consider adjusting the model or providing better initial guesses.",
305 )
306 )
308 hot_in = value(self.hot_side.properties_in[0].temperature) or 0
309 cold_in = value(self.cold_side.properties_in[0].temperature) or 0
310 hot_out = value(self.hot_side.properties_out[0].temperature) or 0
311 cold_out = value(self.cold_side.properties_out[0].temperature) or 0
313 if cold_out > hot_in: 313 ↛ 320line 313 didn't jump to line 320 because the condition on line 313 was always true
314 problems.append(
315 (
316 self.heat_duty[0],
317 f"Cold outlet temperature is above hot inlet temperature ({cold_out - hot_in:.2f} K crossing). Reduce the heat load or check the stream conditions.",
318 )
319 )
320 if hot_out < cold_in: 320 ↛ 328line 320 didn't jump to line 328 because the condition on line 320 was always true
321 problems.append(
322 (
323 self.heat_duty[0],
324 f"Hot outlet temperature is below cold inlet temperature ({cold_in - hot_out:.2f} K crossing). Reduce the heat load or check the stream conditions.",
325 )
326 )
328 return problems
330 def _get_stream_table_contents(self, time_point=0):
331 return create_stream_table_dataframe(
332 {
333 "Hot Inlet": self.hot_side_inlet,
334 "Hot Outlet": self.hot_side_outlet,
335 "Cold Inlet": self.cold_side_inlet,
336 "Cold Outlet": self.cold_side_outlet,
337 },
338 time_point=time_point,
339 )
341 @staticmethod
342 def ahuora_metadata():
343 from ahuora_unit_ops.json_config import JsonUnitOpConfig
345 return JsonUnitOpConfig.model_validate(
346 {
347 "key": "duty_heat_exchanger",
348 "objectType": "duty_heat_exchanger",
349 "enumMember": "DutyHeatExchanger",
350 "displayType": "Duty Heat Exchanger",
351 "displayName": "Duty Heat Exchanger",
352 "categoryPath": ["chemical", "heating_and_cooling"],
353 "ports": {
354 "coldInlet": _stream_port("Cold Inlet", "inlet", "CS"),
355 "hotInlet": _stream_port("Hot Inlet", "inlet", "HS"),
356 "coldOutlet": _stream_port("Cold Outlet", "outlet", "CS"),
357 "hotOutlet": _stream_port("Hot Outlet", "outlet", "HS"),
358 },
359 "propertyPackagePorts": {
360 "Cold Side": ["coldInlet", "coldOutlet"],
361 "Hot Side": ["hotInlet", "hotOutlet"],
362 },
363 "graphicObject": {"kind": "unitop_graphic"},
364 "indexSets": [],
365 "properties": {
366 "heat_duty": {
367 "propertySetGroup": "default",
368 "displayName": "Heat Load",
369 "indexSets": None,
370 "sumToOne": False,
371 "value": None,
372 "unit": None,
373 "unitType": "heatflow",
374 "description": None,
375 "type": "numeric",
376 "many": False,
377 "default": 1,
378 "options": {},
379 "hasTimeIndex": True,
380 }
381 },
382 "propertySetGroups": {
383 "default": {
384 "type": "stateVars",
385 "displayName": "Properties",
386 "stateVars": ["heat_duty"],
387 "toggle": None,
388 }
389 },
390 "keyProperties": ["heat_duty"],
391 "splitterFractionName": None,
392 "idaesAdapter": {
393 "constructor": "ahuora_builder.custom.duty_heat_exchanger.DutyHeatExchanger",
394 "args": {
395 "hot_side": {
396 "kind": "dict",
397 "args": _side_adapter_args("Hot Side"),
398 },
399 "cold_side": {
400 "kind": "dict",
401 "args": _side_adapter_args("Cold Side"),
402 },
403 "dynamic": {"kind": "constant", "value": False},
404 },
405 "ports": {
406 "kind": "port_map",
407 "mapping": {
408 "cold_side_inlet": {"port": "coldInlet", "inlet": True},
409 "hot_side_inlet": {"port": "hotInlet", "inlet": True},
410 "cold_side_outlet": {"port": "coldOutlet", "inlet": False},
411 "hot_side_outlet": {"port": "hotOutlet", "inlet": False},
412 },
413 },
414 "properties": None,
415 },
416 "frontend": {
417 "showInPanel": True,
418 "variant": {
419 "familyKey": "heat_exchanger",
420 "label": "Duty Heat Exchanger",
421 "selectorLabel": "Heat Exchanger Type",
422 "default": False,
423 "order": 30,
424 "preservePorts": True,
425 "preserveGraphic": True,
426 },
427 },
428 }
429 )
432def _stream_port(display_name: str, port_type: str, stream_name: str) -> dict:
433 return {
434 "displayName": display_name,
435 "type": port_type,
436 "streamType": "stream",
437 "many": False,
438 "default": 1,
439 "minimum": 1,
440 "makeStream": True,
441 "streamOffset": 1,
442 "streamName": stream_name,
443 }
446def _side_adapter_args(label: str) -> dict:
447 return {
448 "property_package": {"kind": "property_package", "label": label},
449 "has_pressure_change": {"kind": "constant", "value": False},
450 }
453def _heat_value(heat_var, fallback, to_units=None):
454 if heat_var.value is None: 454 ↛ 455line 454 didn't jump to line 455 because the condition on line 454 was never true
455 return fallback
456 from_units = heat_var.get_units()
457 if from_units is None or to_units is None: 457 ↛ 458line 457 didn't jump to line 458 because the condition on line 457 was never true
458 return heat_var.value
459 return pyunits.convert_value(
460 heat_var.value, from_units=from_units, to_units=to_units
461 )