Coverage for backend/ahuora-builder/src/ahuora_builder/custom/thermal_utility_systems/temp.py: 14%
145 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1# TODO:
2# backend/idaes_service/solver/custom/steam_header.py [DONE]
3# backend/idaes_service/solver/custom/tests/test_steam_header.py [DONE]
4# backend/idaes_service/solver/methods/adapter_library.py
5# backend/idaes_factory/adapters/unit_models/header_adapter.py
6# backend/idaes_factory/adapters/adapter_library.py
7# backend/flowsheetInternals/unitops/config/config_base.py
8# backend/flowsheetInternals/unitops/config/objects/header_config.py
9# backend/django/flowsheetInternals/unitops/config/objects/header_config.json
11from pyomo.environ import Suffix, Var
12from pyomo.common.config import ConfigBlock, ConfigValue
13from pyomo.network import Arc
14from pyomo.core.base.reference import Reference
15import pyomo.environ as pyo
17# Import IDAES cores
18from idaes.core import (
19 declare_process_block_class,
20 UnitModelBlockData,
21 useDefault,
22)
23from idaes.core.util.config import is_physical_parameter_block
24import idaes.logger as idaeslog
25from idaes.core.util.tables import create_stream_table_dataframe
26from idaes.models.unit_models.mixer import Mixer
27from idaes.models.unit_models.heater import Heater
29# Set up logger
30_log = idaeslog.getLogger(__name__)
33@declare_process_block_class("Desuperheater")
34class DesuperheaterData(UnitModelBlockData):
35 """
36 Desuperheater unit operation:
37 Superheated steam + water -> Mixer -> Losses -> Saturated Vapour
38 Desuperheater aims to remove steam superhear by direct contact with boiler feed water.
39 Uses Sequential Decomposition for optimal initialization order.
40 """
42 CONFIG = UnitModelBlockData.CONFIG()
43 CONFIG.declare(
44 "property_package",
45 ConfigValue(
46 default=useDefault,
47 domain=is_physical_parameter_block,
48 description="Property package to use for control volume",
49 ),
50 )
51 CONFIG.declare(
52 "property_package_args",
53 ConfigBlock(
54 implicit=True,
55 description="Arguments to use for constructing property packages",
56 ),
57 )
60 def build(self):
61 super().build()
62 self.scaling_factor = Suffix(direction=Suffix.EXPORT)
64 # Create internal units
65 self.mixer = Mixer(
66 property_package=self.config.property_package,
67 property_package_args=self.config.property_package_args,
68 num_inlets=2
69 )
70 self.losses = Heater(
71 property_package=self.config.property_package,
72 property_package_args=self.config.property_package_args,
73 has_pressure_change=True,
74 )
76 # Build saturated vapour state block
77 tmp_dict = dict(**self.config.property_package_args)
78 tmp_dict["has_phase_equilibrium"] = True
79 tmp_dict["defined_state"] = False
81 self._properties_sat = self.config.property_package.build_state_block(
82 self.flowsheet().time,
83 doc="saturated vapour properties at outlet",
84 **tmp_dict
85 )
87 self.unit_ops = [self.mixer, self.losses, self._properties_sat]
89 # Create internal arcs
90 self.condenser_to_subcooler = Arc(
91 source=self.mixer.outlet,
92 destination=self.losses.inlet,
93 )
95 # Expand arcs
96 pyo.TransformationFactory("network.expand_arcs").apply_to(self)
98 # Identify the exposed ports
99 inlet_exposed_ls = [
100 (self.mixer, "steam_inlet", "inlet_1"),
101 (self.mixer, "water_inlet", "inlet_2"),
102 ]
103 outlet_exposed_ls = [
104 (self.losses, "steam_outlet", "outlet"),
105 ]
107 self.inlet_list, self.inlet_blocks = self._construct_exposed_ports(inlet_exposed_ls)
108 self.outlet_list, self.outlet_blocks = self._construct_exposed_ports(outlet_exposed_ls)
110 # Declare additional variables and provide alias of existing ones to expose to the user
111 self.deltaT_supheat = Var(
112 self.flowsheet().time,
113 domain=pyo.NonNegativeReals,
114 # initialize=0.0,
115 units=pyo.units.K,
116 doc="The degree of steam superheat remaining."
117 )
118 self.water_temperature = Reference(
119 self.water_inlet_state[:].temperature
120 )
121 self.water_flow_mol = Reference(
122 self.water_inlet_state[:].flow_mol
123 )
124 self.water_flow_mass = Reference(
125 self.water_inlet_state[:].flow_mass
126 )
127 self.heat_duty = Reference(
128 self.losses.heat_duty[:]
129 )
130 self.deltaP = Reference(
131 self.losses.deltaP[:]
132 )
134 # Declare internal variables and references for internal use
135 self._T_sat = Reference(
136 self._properties_sat[:].temperature
137 )
139 # Additional bounds and constraints
140 self._additional_constraints()
142 def _additional_constraints(self):
143 """
144 Additional constraints.
145 """
147 """
148 1. Pressure equality for saturation state block
149 """
150 @self.Constraint(
151 self.flowsheet().time, doc="Pressure equality for saturation state block"
152 )
153 def outlet_pressure_eql(b, t):
154 return (
155 b._properties_sat[t].pressure == b.losses.control_volume.properties_out[t].pressure
156 )
158 """
159 2. Vapour quality for saturation state block
160 """
161 @self.Constraint(
162 self.flowsheet().time, doc="Pressure equality for saturation state block"
163 )
164 def vapour_quality_eq(b, t):
165 return (
166 b._properties_sat[t].vapor_frac == 0.5
167 )
169 """
170 3. Mixer pressure inlet equality for steam and water
171 """
172 @self.Constraint(
173 self.flowsheet().time, doc="Mixer pressure inlet equality for steam and water"
174 )
175 def mixer_inlet_pressure_eql(b, t):
176 return (
177 b.steam_inlet_state[t].pressure == b.water_inlet_state[t].pressure
178 )
180 """
181 4. Superheat temperature balance
182 """
183 @self.Constraint(self.flowsheet().time, doc="Temperature superheat balance")
184 def superheat_temperature_eqn(b, t):
185 return (
186 b.deltaT_supheat[t] + 0.001 == b.steam_outlet_state[t].temperature - b._T_sat[t]
187 )
189 def _construct_exposed_ports(self, exposed_ls):
190 """
191 exposed_ls: list of [(unit, public_name, internal_basename)]
192 """
193 names, states = [], []
195 for unit, public_name, internal_basename in exposed_ls:
196 # 1) Locate the state block on the child unit
197 # Try <basename>_state (Mixer & many IDAES units)
198 state = getattr(unit, f"{internal_basename}_state", None)
200 # Fall back to control_volume properties for CV-based units (e.g., Heater)
201 if state is None and hasattr(unit, "control_volume"):
202 cv = unit.control_volume
203 if internal_basename.startswith("inlet") and hasattr(cv, "properties_in"):
204 state = cv.properties_in
205 elif internal_basename.startswith("outlet") and hasattr(cv, "properties_out"):
206 state = cv.properties_out
208 if state is None:
209 raise AttributeError(
210 f"{unit.name} has no state for '{internal_basename}'. "
211 f"Tried '{internal_basename}_state' and control_volume properties."
212 )
214 # 2) Expose the state on this wrapper
215 # (a) keep a convenient Reference to the time-indexed state block
216 setattr(self, f"{public_name}_state", Reference(state[:]))
218 # (b) create a public Port mapped to that state
219 # NOTE: UnitModelBlock.add_port builds a Port with the vars
220 # defined by the property package’s port members.
221 self.add_port(name=public_name, block=state)
223 names.append(public_name)
224 states.append(state)
226 return names, states
228 def calculate_scaling_factors(self):
229 super().calculate_scaling_factors()
230 for o in self.unit_ops:
231 if hasattr(o, "calculate_scaling_factors"):
232 o.calculate_scaling_factors()
234 def _get_stream_table_contents(self, time_point=0):
235 """
236 Create stream table showing all inlets and outlets
237 """
238 io_dict = {}
240 for inlet_name in self.inlet_list:
241 io_dict[inlet_name] = getattr(self, inlet_name)
243 for outlet_name in self.outlet_list:
244 io_dict[outlet_name] = getattr(self, outlet_name)
246 return create_stream_table_dataframe(io_dict, time_point=time_point)
248 def _copy_port_state(self, src_port, snk_port, t):
249 for name in ("flow_mol", "pressure", "enth_mol", "temperature", "enth_mass", "vapor_frac"):
250 if hasattr(src_port, name) and hasattr(snk_port, name):
251 src = getattr(src_port, name)
252 snk = getattr(snk_port, name)
253 if (t in src) and (t in snk) and pyo.is_variable_type(snk[t]):
254 try:
255 snk[t].set_value(pyo.value(src[t]))
256 except Exception:
257 pass
259 def _hold_state_vars(self, sb):
260 held = {}
261 if hasattr(sb, "define_state_vars"):
262 for k, v in sb.define_state_vars().items():
263 if pyo.is_variable_type(v) and not v.fixed:
264 v.fix()
265 held[k] = False
266 else:
267 held[k] = True
268 return held
270 def _release_state_vars(self, sb, held):
271 if hasattr(sb, "define_state_vars"):
272 for k, was_fixed in held.items():
273 v = sb.define_state_vars()[k]
274 if was_fixed is False and pyo.is_variable_type(v):
275 v.unfix()
277 def initialize_build(self, outlvl=idaeslog.NOTSET, **kwargs):
278 """
279 Initialize the Desuperheater without using state_args for sub-units.
281 Steps:
282 1) Initialize Mixer (no state_args; it reads from connected inlet ports).
283 2) Initialize Heater with temporary ΔP=0 and Q=0, then restore.
284 3) Seed and (if supported) initialize saturated-vapor state block.
285 4) Seed deltaT_supheat.
286 """
287 init_log = idaeslog.getInitLogger(self.name, outlvl, tag="unit")
288 solve_log = idaeslog.getSolveLogger(self.name, outlvl, tag="unit")
290 t0 = list(self.flowsheet().time)[0]
292 # 1) Mixer
293 init_log.info("Initializing Mixer (no state_args)...")
294 try:
295 # Inlet states should already be provided via self.steam_inlet / self.water_inlet
296 # and are wired to mixer.inlet_1 / mixer.inlet_2 via the exposed ports.
297 self.water_flow_mol[t0].fix(0.0)
298 self.water_flow_mol[t0].unfix()
299 self.mixer.inlet_2.pressure[t0].fix(
300 pyo.value(
301 self.mixer.inlet_1.pressure[t0]
302 )
303 )
304 self.mixer.inlet_2.pressure[t0].unfix()
305 self.mixer.initialize(outlvl=outlvl)
306 except Exception as err:
307 init_log.error(f"Mixer initialization failed: {err}")
308 raise
310 # 2) Heater (losses)
311 init_log.info("Initializing Heater (losses) with temporary ΔP=0 and Q=0...")
312 self._copy_port_state(
313 src_port=self.mixer.outlet,
314 snk_port=self.losses.inlet,
315 t=t0
316 )
317 held = self._hold_state_vars(self.losses.control_volume.properties_in[t0])
318 try:
319 self.losses.initialize(outlvl=outlvl)
320 finally:
321 self._release_state_vars(self.losses.control_volume.properties_in[t0], held)
323 # 3) Saturated-vapor state block seeding
324 init_log.info("Seeding saturated-vapor state block...")
325 try:
326 # Set seed values directly
327 self._properties_sat[t0].pressure.set_value(
328 pyo.value(self.losses.outlet.pressure[t0])
329 )
330 self._properties_sat[t0].enth_mol.set_value(
331 pyo.value(self.losses.outlet.enth_mol[t0])
332 )
334 # If the block exposes its own initialize(outlvl=...), call it
335 if hasattr(self._properties_sat[t0], "initialize"):
336 # Many property packages implement initialize on the *indexed* block,
337 # i.e. call on the parent indexed component:
338 try:
339 self._properties_sat.initialize(outlvl=outlvl)
340 except TypeError:
341 # Some implementations only support per-timepoint initialize
342 self._properties_sat[t0].initialize(outlvl=outlvl)
344 except Exception as err:
345 init_log.error(f"Saturation state initialization failed: {err}")
346 raise
348 init_log.info("Desuperheater initialization complete.")