Coverage for backend/idaes_service/solver/custom/thermal_utility_systems/desuperheater.py: 87%
167 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-11-06 23:27 +0000
1# Pyomo core
2from pyomo.environ import (
3 Constraint,
4 Expression,
5 NonNegativeReals,
6 Suffix,
7 Var,
8 value,
9 units as UNIT,
10)
11from pyomo.core.base.reference import Reference
12from pyomo.common.config import ConfigBlock, ConfigValue, Bool
14# IDAES core
15from idaes.core import (
16 declare_process_block_class,
17 UnitModelBlockData,
18 useDefault,
19 StateBlock,
20)
21from idaes.core.util import scaling
22from idaes.core.util.config import is_physical_parameter_block
23from idaes.core.util.tables import create_stream_table_dataframe
24from idaes.core.solvers import get_solver
25from idaes.core.initialization import ModularInitializerBase
27# Other
28from property_packages.build_package import build_package
30# Logger
31import idaes.logger as idaeslog
33# Typing
34from typing import List
37__author__ = "Ahuora Centre for Smart Energy Systems, University of Waikato, New Zealand"
39# Set up logger
40_log = idaeslog.getLogger(__name__)
42class DesuperheaterInitializer(ModularInitializerBase):
43 """Initializer for ``Desuperheater``.
45 Parameters
46 ----------
47 blk : Desuperheater
48 The unit model block to initialize.
49 solver : optional
50 A Pyomo/IDAES solver instance. Defaults to :func:`idaes.core.solvers.get_solver`.
51 solver_options : dict, optional
52 Options to set on the solver, e.g. tolerances.
53 outlvl : int, optional
54 IDAES log level (e.g. :data:`idaes.logger.WARNING`).
56 Returns
57 -------
58 pyomo.opt.results.results_.SolverResults
59 Result from the final solve.
60 """
62 def initialize(self, blk, **kwargs):
63 # --- Solver setup
64 solver = kwargs.get("solver", None) or get_solver()
65 solver_options = kwargs.get("solver_options", {})
66 for k, v in solver_options.items(): 66 ↛ 67line 66 didn't jump to line 67 because the loop on line 66 never started
67 solver.options[k] = v
69 outlvl = kwargs.get("outlvl", idaeslog.WARNING)
70 log = idaeslog.getLogger(__name__)
72 # --- Time index
73 t0 = blk.flowsheet().time.first()
75 # --- 1) Initialize inlet state blocks
76 inlet_blocks = list(blk.inlet_blocks)
78 inlet_steam, inlet_water = inlet_blocks
79 inlet_water[t0].pressure.set_value(
80 inlet_steam[t0].pressure
81 )
82 inlet_water[t0].enth_mol.set_value(
83 blk.water.htpx(
84 blk.bfw_temperature[t0],
85 inlet_water[t0].pressure,
86 )
87 )
88 for sb in inlet_blocks:
89 if hasattr(sb, "initialize"): 89 ↛ 88line 89 didn't jump to line 88 because the condition on line 89 was always true
90 sb.initialize(outlvl=outlvl)
92 # --- 2) Seed satuarate inlet-related state
93 sb = blk._int_sat_vap_state
94 sb[t0].pressure.set_value(
95 inlet_steam[t0].pressure
96 )
97 sb[t0].enth_mol.set_value(
98 inlet_steam[t0].enth_mol_sat_phase["Vap"]
99 )
100 if hasattr(sb, "initialize"): 100 ↛ 104line 100 didn't jump to line 104 because the condition on line 100 was always true
101 sb.initialize(outlvl=outlvl)
103 # --- 3) Aggregate inlet info for seeding mixed state block
104 ms = blk.outlet_state
105 ms[t0].pressure.set_value(
106 inlet_steam[t0].pressure
107 )
108 if value(blk.deltaT_superheat[t0]) > 0:
109 ms[t0].enth_mol.set_value(
110 blk.water.htpx(
111 T=blk._int_sat_vap_state[t0].temperature + blk.deltaT_superheat[t0],
112 p=inlet_steam[t0].pressure,
113 )
114 )
115 else:
116 ms[t0].enth_mol.set_value(
117 blk._int_sat_vap_state[t0].enth_mol
118 )
120 if value(inlet_steam[t0].enth_mol) > value(ms[t0].enth_mol) > value(inlet_water[t0].enth_mol): 120 ↛ 125line 120 didn't jump to line 125 because the condition on line 120 was always true
121 ms[t0].flow_mol.set_value(
122 inlet_steam[t0].flow_mol * (inlet_steam[t0].enth_mol - ms[t0].enth_mol) / (ms[t0].enth_mol - inlet_water[t0].enth_mol)
123 )
124 else:
125 ms[t0].flow_mol.set_value(
126 inlet_steam[t0].flow_mol
127 )
128 if hasattr(ms, "initialize"): 128 ↛ 132line 128 didn't jump to line 132 because the condition on line 128 was always true
129 ms.initialize(outlvl=outlvl)
131 # --- 4) Solve
132 res = solver.solve(blk, tee=False)
133 log.info(f"Desuperheater init status: {res.solver.termination_condition}")
134 return res
136def _make_config_block(config):
137 """Declare configuration options for the Desuperheater unit.
139 Declares property package references and integer counts for inlets and outlets.
141 Args:
142 config (ConfigBlock): The mutable configuration block to populate.
143 """
144 config.declare(
145 "property_package",
146 ConfigValue(
147 default=useDefault,
148 domain=is_physical_parameter_block,
149 description="Property package to use for control volume",
150 ),
151 )
152 config.declare(
153 "property_package_args",
154 ConfigBlock(
155 implicit=True,
156 description="Arguments to use for constructing property packages",
157 ),
158 )
160@declare_process_block_class("Desuperheater")
161class DesuperheaterData(UnitModelBlockData):
162 """Desuperheater unit operation.
164 The Desuperheater injects a small water flow into a superheated steam flow to reduce superheat.
165 Desuperheating is common before using steam. Heat loss and pressure loss may be defined.
166 An intermediate saturated state is used for a key reference point.
168 Key features:
169 - Material, energy, and momentum balances around the desuperheater
170 - User-specified target amount of superheat at the exit of the desuperheater
171 - Optional heat and pressure losses.
173 Attributes:
174 inlet_list (list[str]): Names for inlet ports.
175 outlet_list (list[str]): Names for outlet ports (incl. condensate/ and vent).
176 inlet_blocks (list): StateBlocks for all inlets.
177 outlet_blocks (list): StateBlocks for all outlets.
178 _int_sat_vap_state: Intermediate saturated vapour StateBlock.
180 State variables:
181 deltaT_superheat (Var): Target degree of superheat in the steam at the exit of the process.
182 bfw_temperature (Var): Temperature of the inlet boiler feed water flow for desuperheating (degC).
183 heat_loss (Var): Heat loss from the header (W).
184 pressure_loss (Var): Pressure drop from inlet minimum to mixed state (Pa).
185 """
187 default_initializer=DesuperheaterInitializer
188 CONFIG = UnitModelBlockData.CONFIG()
189 _make_config_block(CONFIG)
191 def build(self) -> None:
192 """Build the unit model structure (ports, states, constraints)."""
193 # 1. Inherit standard UnitModelBlockData properties and functions
194 super().build()
196 # 2. Validate input parameters are valid
197 self._validate_model_config()
199 # 3. Create lists of ports with state blocks to add
200 self.inlet_list = self._create_inlet_port_name_list()
201 self.outlet_list = self._create_outlet_port_name_list()
203 # 4. Declare ports, state blocks and state property bounds
204 self.inlet_blocks = self._add_ports_with_state_blocks(
205 stream_list=self.inlet_list,
206 is_inlet=True,
207 has_phase_equilibrium=False,
208 is_defined_state=True,
209 )
210 self.outlet_blocks = self._add_ports_with_state_blocks(
211 stream_list=self.outlet_list,
212 is_inlet=False,
213 has_phase_equilibrium=False,
214 is_defined_state=False
215 )
216 self._internal_blocks = self._add_internal_state_blocks()
217 self._add_bounds_to_state_properties()
219 # 4. Declare references, variables and expressions for external and internal use
220 self._create_references()
221 self._create_variables()
222 self._create_expressions()
224 # 5. Set balance equations
225 self._add_material_balances()
226 self._add_energy_balances()
227 self._add_momentum_balances()
228 self._add_additional_constraints()
230 # 6. Other
231 self.scaling_factor = Suffix(direction=Suffix.EXPORT)
233 # ------------------------------------------------------------------
234 # Helpers & construction utilities
235 # ------------------------------------------------------------------
236 def _validate_model_config(self) -> bool:
237 """Validate configuration for inlet and outlet counts.
239 Raises:
240 ValueError: If ``property_package is None``.
241 """
242 if self.config.property_package is None: 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true
243 raise ValueError("Desuperheater: Property package not defined.")
244 return True
246 def _create_inlet_port_name_list(self) -> List[str]:
247 """Build ordered inlet port names.
249 Returns:
250 list[str]: Names
251 """
253 return (
254 [
255 "inlet_steam", "inlet_water"
256 ]
257 )
259 def _create_outlet_port_name_list(self) -> List[str]:
260 """Build ordered outlet port names.
262 Returns:
263 list[str]: Names
264 """
265 return [
266 "outlet",
267 ]
269 def _add_ports_with_state_blocks(self,
270 stream_list: List[str],
271 is_inlet: List[str],
272 has_phase_equilibrium: bool=False,
273 is_defined_state: bool=None,
274 ) -> List[StateBlock]:
275 """Construct StateBlocks and expose them as ports.
277 Creates a StateBlock per named stream and attaches a corresponding inlet or
278 outlet Port. Inlet blocks are defined states; outlet blocks are calculated states.
280 Args:
281 stream_list (list[str]): Port/StateBlock base names to create.
282 is_inlet (bool): If True, create inlet ports with ``defined_state=True``;
283 otherwise create outlet ports with ``defined_state=False``.
284 has_phase_equilibrium (bool)
286 Returns:
287 list: The created StateBlocks, in the same order as ``stream_list``.
288 """
289 # Create empty list to hold StateBlocks for return
290 state_block_ls = []
292 # Setup StateBlock argument dict
293 tmp_dict = dict(**self.config.property_package_args)
294 tmp_dict["has_phase_equilibrium"] = has_phase_equilibrium
295 if is_defined_state == None: 295 ↛ 296line 295 didn't jump to line 296 because the condition on line 295 was never true
296 tmp_dict["defined_state"] = True if is_inlet else False
297 else:
298 tmp_dict["defined_state"] = is_defined_state
300 # Create an instance of StateBlock for all streams
301 for s in stream_list:
302 sb = self.config.property_package.build_state_block(
303 self.flowsheet().time, doc=f"Thermophysical properties at {s}", **tmp_dict
304 )
305 setattr(
306 self, s + "_state",
307 sb
308 )
309 state_block_ls.append(sb)
310 add_fn = self.add_inlet_port if is_inlet else self.add_outlet_port
311 add_fn(
312 name=s,
313 block=sb,
314 )
316 return state_block_ls
318 def _add_internal_state_blocks(self) -> List[StateBlock]:
319 """Create the intermediate StateBlock(s)."""
320 # The _int_sat_vap_state:
321 # - Has phase equilibrium enabled.
322 # - Is not a defined state (solved from balances).
323 tmp_dict = dict(**self.config.property_package_args)
324 tmp_dict["has_phase_equilibrium"] = True
325 tmp_dict["defined_state"] = False
326 self._int_sat_vap_state = self.config.property_package.build_state_block(
327 self.flowsheet().time,
328 doc="Thermophysical properties internal saturated vapour state.",
329 **tmp_dict
330 )
331 self._int_sat_vap_state[:].flow_mol.fix(1)
333 return [
334 self._int_sat_vap_state,
335 ]
337 def _add_bounds_to_state_properties(self) -> None:
338 """Add lower and/or upper bounds to state properties.
340 - Set nonnegativity lower bounds on all inlet/intermediate/outlet flows.
341 """
342 for sb in (self.inlet_blocks + self.outlet_blocks):
343 for t in sb:
344 sb[t].flow_mol.setlb(0.0)
346 def _create_references(self) -> None:
347 """Create convenient References.
349 Creates references to _int_mixed_inlet_state properties:
350 - ``bfw_flow_mass``
351 - ``bfw_flow_mol``
352 """
353 self.bfw_flow_mass = Reference(
354 self.inlet_water_state[:].flow_mass
355 )
356 self.bfw_flow_mol = Reference(
357 self.inlet_water_state[:].flow_mol
358 )
359 self.inlet_water_state[:].flow_mol.unfix()
360 self.water = build_package("helmholtz", ["water"], ["Liq"])
362 def _create_variables(self) -> None:
363 """Declare decision/parameter variables for the unit.
365 Creates:
366 - ``heat_loss``
367 - ``pressure_loss``
368 - ``bfw_temperature``
369 - ``deltaT_superheat``
370 """
371 # Get units consistent with the property package
372 units_meta = self.config.property_package.get_metadata()
374 # User defined: Heat and pressure losses
375 self.heat_loss = Var(
376 self.flowsheet().time,
377 domain=NonNegativeReals,
378 doc="Heat loss. Default: 0 kW.",
379 units=units_meta.get_derived_units("power")
380 )
381 self.heat_loss.fix(
382 0 # Default fixed value
383 )
384 self.pressure_loss = Var(
385 self.flowsheet().time,
386 domain=NonNegativeReals,
387 doc="Pressure loss. Default: 0 Pa.",
388 units=units_meta.get_derived_units("pressure")
389 )
390 self.pressure_loss.fix(
391 0 # Default fixed value
392 )
393 # User defined: Boiler feed water temperature entering the desuperheater
394 self.bfw_temperature = Var(
395 self.flowsheet().time,
396 domain=NonNegativeReals,
397 doc="The target amount of subcooling of the condensate after process heating. Default: 0 K.",
398 units=units_meta.get_derived_units("temperature"),
399 )
400 self.bfw_temperature.fix(
401 (110 + 273.15) * UNIT.K # Default fixed value
402 )
403 # User defined: Target degree of superheat at the outlet of the desuperheater
404 self.deltaT_superheat = Var(
405 self.flowsheet().time,
406 domain=NonNegativeReals,
407 doc="The target amount of superheat present in the steam after desuperheating before use. Default: 0 K.",
408 units=units_meta.get_derived_units("temperature"),
409 )
410 self.deltaT_superheat.fix(
411 0 # Default fixed value
412 )
414 def _create_expressions(self) -> None:
415 """Create helper Expressions.
417 Creates:
418 - ``flow_ratio``
419 """
420 # Calculated, always show
421 self.flow_ratio = Expression(
422 self.flowsheet().time,
423 rule=lambda b, t: (
424 b.inlet_water_state[t].flow_mol
425 /
426 (b.inlet_steam_state[t].flow_mol + 1e-9)
427 ),
428 doc="Ratio of water to steam flows.",
429 )
431 # ------------------------------------------------------------------
432 # Balances
433 # ------------------------------------------------------------------
434 def _add_material_balances(self) -> None:
435 """Material balance equations summary.
437 Balances / Constraints:
438 - ``overall_material_balance``
439 """
440 @self.Constraint(
441 self.flowsheet().time,
442 doc="Overall material balance",
443 )
444 def overall_material_balance(b, t):
445 return (
446 sum(
447 o[t].flow_mol
448 for o in b.outlet_blocks
449 )
450 ==
451 sum(
452 i[t].flow_mol
453 for i in b.inlet_blocks
454 )
455 )
457 def _add_energy_balances(self) -> None:
458 """Energy balance equations summary.
460 Balances / Constraints:
461 - ``overall_energy_balance``
462 - ``saturated_vap_enthalpy_eq``
463 """
464 @self.Constraint(
465 self.flowsheet().time,
466 doc="Overall energy balance",
467 )
468 def overall_energy_balance(b, t):
469 return (
470 sum(
471 i[t].flow_mol * i[t].enth_mol
472 for i in b.inlet_blocks
473 )
474 ==
475 sum(
476 i[t].flow_mol * i[t].enth_mol
477 for i in b.outlet_blocks
478 )
479 +
480 b.heat_loss[t]
481 )
482 @self.Constraint(
483 self.flowsheet().time,
484 doc="Saturated vapour enthalpy",
485 )
486 def saturated_vap_enthalpy_eq(b, t):
487 return (
488 b.outlet_state[t].enth_mol_sat_phase["Vap"]
489 ==
490 b._int_sat_vap_state[t].enth_mol
491 )
493 def _add_momentum_balances(self) -> None:
494 """Momentum balance equations summary.
496 Balances / Constraints:
497 - ``overall_momentum_balance``
498 - ``intlet_water_momentum_balance``
499 - ``saturated_vap_pressure_eq``
500 """
501 @self.Constraint(
502 self.flowsheet().time,
503 doc="Overall momentum balance",
504 )
505 def overall_momentum_balance(b, t):
506 return (
507 b.inlet_steam_state[t].pressure
508 ==
509 b.outlet_state[t].pressure
510 +
511 b.pressure_loss[t]
512 )
513 @self.Constraint(
514 self.flowsheet().time,
515 doc="Inlet water momentum balance",
516 )
517 def intlet_water_momentum_balance(b, t):
518 return (
519 b.inlet_water_state[t].pressure
520 ==
521 b.inlet_steam_state[t].pressure
522 )
523 @self.Constraint(
524 self.flowsheet().time,
525 doc="Saturated vapour pressure",
526 )
527 def saturated_vap_pressure_eq(b, t):
528 return (
529 b._int_sat_vap_state[t].pressure
530 ==
531 b.outlet_state[t].pressure
532 )
534 def _add_additional_constraints(self) -> None:
535 """Add auxiliary constraints and bounds.
537 Constraints:
538 - ``inlet_water_temperature_eq``
539 - ``desuperheating_temperature_eq``
540 """
541 @self.Constraint(
542 self.flowsheet().time,
543 doc="Inlet water temperature",
544 )
545 def inlet_water_temperature_eq(b, t):
546 return (
547 b.inlet_water_state[t].temperature
548 ==
549 b.bfw_temperature[t]
550 )
551 @self.Constraint(
552 self.flowsheet().time,
553 doc="Temperature after desuperheating",
554 )
555 def desuperheating_temperature_eq(b, t):
556 return (
557 b.outlet_state[t].temperature
558 ==
559 b._int_sat_vap_state[t].temperature + b.deltaT_superheat[t]
560 )
562 def calculate_scaling_factors(self):
563 """Assign scaling factors to improve numerical conditioning.
565 Sets scaling factors for performance and auxiliary variables.
566 """
567 super().calculate_scaling_factors()
568 scaling.set_scaling_factor(self.heat_loss, 1e-3) # kW scale
569 scaling.set_scaling_factor(self.pressure_loss, 1e-3) # kPa scale
571 def _get_stream_table_contents(self, time_point=0):
572 """Create a stream table for all inlets and outlets.
574 Args:
575 time_point (int | float): Time index at which to extract stream data.
577 Returns:
578 pandas.DataFrame: A tabular view suitable for reporting via
579 ``create_stream_table_dataframe``.
580 """
581 io_dict = {}
583 for inlet_name in self.inlet_list:
584 io_dict[inlet_name] = getattr(self, inlet_name)
586 for outlet_name in self.outlet_list:
587 io_dict[outlet_name] = getattr(self, outlet_name)
589 return create_stream_table_dataframe(io_dict, time_point=time_point)
591 def _get_performance_contents(self, time_point=0):
592 """Collect performance variables for reporting.
594 Args:
595 time_point (int | float): Time index at which to report values.
597 Returns:
598 dict: Mapping used by IDAES reporters, containing human-friendly labels
599 to Vars/References (e.g., heat/pressure loss, mixed-state properties).
600 """
601 return {
602 "vars": {
603 "BFW temperature [K]": self.bfw_temperature[time_point],
604 "Degree of superheat target [K]": self.deltaT_superheat[time_point],
605 "Heat loss [W]": self.heat_loss[time_point],
606 "Pressure loss [Pa]": self.pressure_loss[time_point],
607 },
608 "exprs": {
609 "Water-to-steam flow ratio": self.flow_ratio[time_point],
610 },
611 }
613 def initialize(self, *args, **kwargs):
614 """Initialize the Desuperheater unit using :class:`DesuperheaterInitializer`.
616 Args:
617 *args: Forwarded to ``DesuperheaterInitializer.initialize``.
618 **kwargs: Forwarded to ``DesuperheaterInitializer.initialize`` (e.g., solver, options).
620 Returns:
621 pyomo.opt.results.results_.SolverResults: Results from the initializer's solve.
622 """
623 init = DesuperheaterInitializer()
624 return init.initialize(self, *args, **kwargs)