Coverage for backend/idaes_service/solver/custom/simple_separator.py: 48%
235 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-11-06 23:27 +0000
1#################################################################################
2# The Institute for the Design of Advanced Energy Systems Integrated Platform
3# Framework (IDAES IP) was produced under the DOE Institute for the
4# Design of Advanced Energy Systems (IDAES).
5#
6# Copyright (c) 2018-2024 by the software owners: The Regents of the
7# University of California, through Lawrence Berkeley National Laboratory,
8# National Technology & Engineering Solutions of Sandia, LLC, Carnegie Mellon
9# University, West Virginia University Research Corporation, et al.
10# All rights reserved. Please see the files COPYRIGHT.md and LICENSE.md
11# for full copyright and license information.
12#################################################################################
13"""
14General purpose separator block for IDAES models
15"""
17from enum import Enum
18from pandas import DataFrame
20from pyomo.environ import (Block, Set)
21from pyomo.network import Port
22from pyomo.common.config import ConfigBlock, ConfigValue, In, ListOf, Bool
24from idaes.core import (
25 declare_process_block_class,
26 UnitModelBlockData,
27 useDefault,
28 MaterialBalanceType,
29 MomentumBalanceType,
30)
31from idaes.core.util.config import (
32 is_physical_parameter_block,
33 is_state_block,
34)
35from idaes.core.util.exceptions import (
36 BurntToast,
37 ConfigurationError,
38)
39from idaes.core.solvers import get_solver
40from idaes.core.util.tables import create_stream_table_dataframe
41from idaes.core.util.model_statistics import degrees_of_freedom
42import idaes.logger as idaeslog
43import idaes.core.util.scaling as iscale
44from idaes.core.util.units_of_measurement import report_quantity
45from idaes.core.initialization import ModularInitializerBase
47__author__ = "Team Ahuora"
50# Set up logger
51_log = idaeslog.getLogger(__name__)
54# Enumerate options for balances
55class SplittingType(Enum):
56 """
57 Enum of supported material split types.
58 """
60 totalFlow = 1
61 phaseFlow = 2
62 componentFlow = 3
63 phaseComponentFlow = 4
66class EnergySplittingType(Enum):
67 """
68 Enum of support energy split types.
69 """
71 none = 0
72 equal_molar_enthalpy = 2
75class SimpleSeparatorInitializer(ModularInitializerBase):
76 """
77 Initializer for Separator blocks.
79 """
81 def initialization_routine(
82 self,
83 model: Block,
84 ):
85 """
86 Initialization routine for Separator Blocks.
88 This routine starts by initializing the feed and outlet streams using simple rules.
90 Args:
91 model: model to be initialized
93 Returns:
94 None
96 """
97 init_log = idaeslog.getInitLogger(
98 model.name, self.get_output_level(), tag="unit"
99 )
100 solve_log = idaeslog.getSolveLogger(
101 model.name, self.get_output_level(), tag="unit"
102 )
104 # Create solver
105 solver = self._get_solver()
106 # Initialize mixed state block
108 mblock = model.mixed_state
109 self.get_submodel_initializer(mblock).initialize(mblock)
111 res = None
112 if degrees_of_freedom(model) != 0:
113 with idaeslog.solver_log(solve_log, idaeslog.DEBUG) as slc:
114 res = solver.solve(model, tee=slc.tee)
115 init_log.info(
116 "Initialization Step 1 Complete: {}".format(idaeslog.condition(res))
117 )
119 for c, s in component_status.items():
120 if s:
121 c.activate()
124 # Initialize outlet StateBlocks
125 outlet_list = model._create_outlet_list()
127 # Initializing outlet states
128 for o in outlet_list:
129 # Get corresponding outlet StateBlock
130 o_block = getattr(model, o + "_state")
132 # Create dict to store fixed status of state variables
133 for t in model.flowsheet().time:
134 # Calculate values for state variables
135 s_vars = o_block[t].define_state_vars()
136 for var_name_port, var_obj in s_vars.items():
137 for k in var_obj:
138 # If fixed, use current value
139 # otherwise calculate guess from mixed state and fix
140 if not var_obj[k].fixed:
141 m_var = getattr(mblock[t], var_obj.local_name)
142 if "flow" in var_name_port:
143 # Leave initial value
144 pass
145 else:
146 # Otherwise intensive, equate to mixed stream
147 var_obj[k].set_value(m_var[k].value)
149 # Call initialization routine for outlet StateBlock
150 self.get_submodel_initializer(o_block).initialize(o_block)
152 init_log.info("Initialization Complete.")
154 return res
157@declare_process_block_class("SimpleSeparator")
158class SimpleSeparatorData(UnitModelBlockData):
159 """
160 This is a simple Splitter block with the IDAES modeling framework.
161 Unlike the generic Separator, this block avoids use of split fractions.
163 This model creates a number of StateBlocks to represent the outgoing
164 streams, then writes a set of phase-component material balances, an
165 overall enthalpy balance (2 options), and a momentum balance (2 options)
166 linked to a mixed-state StateBlock. The mixed-state StateBlock can either
167 be specified by the user (allowing use as a sub-model), or created by the
168 Separator.
169 """
171 default_initializer = SimpleSeparatorInitializer
173 CONFIG = UnitModelBlockData.CONFIG()
174 CONFIG.declare(
175 "property_package",
176 ConfigValue(
177 default=useDefault,
178 domain=is_physical_parameter_block,
179 description="Property package to use for mixer",
180 doc="""Property parameter object used to define property
181calculations,
182**default** - useDefault.
183**Valid values:** {
184**useDefault** - use default package from parent model or flowsheet,
185**PropertyParameterObject** - a PropertyParameterBlock object.}""",
186 ),
187 )
188 CONFIG.declare(
189 "property_package_args",
190 ConfigBlock(
191 implicit=True,
192 description="Arguments to use for constructing property packages",
193 doc="""A ConfigBlock with arguments to be passed to a property
194block(s) and used when constructing these,
195**default** - None.
196**Valid values:** {
197see property package for documentation.}""",
198 ),
199 )
200 CONFIG.declare(
201 "outlet_list",
202 ConfigValue(
203 domain=ListOf(str),
204 description="List of outlet names",
205 doc="""A list containing names of outlets,
206**default** - None.
207**Valid values:** {
208**None** - use num_outlets argument,
209**list** - a list of names to use for outlets.}""",
210 ),
211 )
212 CONFIG.declare(
213 "num_outlets",
214 ConfigValue(
215 domain=int,
216 description="Number of outlets to unit",
217 doc="""Argument indicating number (int) of outlets to construct,
218not used if outlet_list arg is provided,
219**default** - None.
220**Valid values:** {
221**None** - use outlet_list arg instead, or default to 2 if neither argument
222provided,
223**int** - number of outlets to create (will be named with sequential integers
224from 1 to num_outlets).}""",
225 ),
226 )
227 CONFIG.declare(
228 "material_balance_type",
229 ConfigValue(
230 default=MaterialBalanceType.useDefault,
231 domain=In(MaterialBalanceType),
232 description="Material balance construction flag",
233 doc="""Indicates what type of mass balance should be constructed,
234**default** - MaterialBalanceType.useDefault.
235**Valid values:** {
236**MaterialBalanceType.useDefault - refer to property package for default
237balance type
238**MaterialBalanceType.none** - exclude material balances,
239**MaterialBalanceType.componentPhase** - use phase component balances,
240**MaterialBalanceType.componentTotal** - use total component balances,
241**MaterialBalanceType.elementTotal** - use total element balances,
242**MaterialBalanceType.total** - use total material balance.}""",
243 ),
244 )
245 CONFIG.declare(
246 "momentum_balance_type",
247 ConfigValue(
248 default=MomentumBalanceType.pressureTotal,
249 domain=In(MomentumBalanceType),
250 description="Momentum balance construction flag",
251 doc="""Indicates what type of momentum balance should be constructed,
252 **default** - MomentumBalanceType.pressureTotal.
253 **Valid values:** {
254 **MomentumBalanceType.none** - exclude momentum balances,
255 **MomentumBalanceType.pressureTotal** - pressure in all outlets is equal,
256 **MomentumBalanceType.pressurePhase** - not yet supported,
257 **MomentumBalanceType.momentumTotal** - not yet supported,
258 **MomentumBalanceType.momentumPhase** - not yet supported.}""",
259 ),
260 )
261 CONFIG.declare(
262 "has_phase_equilibrium",
263 ConfigValue(
264 default=False,
265 domain=Bool,
266 description="Calculate phase equilibrium in mixed stream",
267 doc="""Argument indicating whether phase equilibrium should be
268calculated for the resulting mixed stream,
269**default** - False.
270**Valid values:** {
271**True** - calculate phase equilibrium in mixed stream,
272**False** - do not calculate equilibrium in mixed stream.}""",
273 ),
274 )
276 def build(self):
277 """
278 General build method for SeparatorData. This method calls a number
279 of sub-methods which automate the construction of expected attributes
280 of unit models.
282 Inheriting models should call `super().build`.
284 Args:
285 None
287 Returns:
288 None
289 """
290 # Call super.build()
291 super(SimpleSeparatorData, self).build()
293 # Call setup methods from ControlVolumeBlockData
294 self._get_property_package()
295 self._get_indexing_sets()
297 # Create list of inlet names
298 outlet_list = self._create_outlet_list()
300 mixed_block = self._add_mixed_state_block()
302 # Add inlet port
303 self._add_inlet_port_objects(mixed_block)
305 # Build StateBlocks for outlet
306 outlet_blocks = self._add_outlet_state_blocks(outlet_list)
307 self.outlet_idx = Set(initialize=outlet_list)
309 # Construct splitting equations
310 self._add_material_balance(mixed_block, outlet_blocks)
311 self._add_energy_balance(mixed_block, outlet_blocks)
312 self._add_momentum_balance(mixed_block, outlet_blocks)
314 # Construct outlet port objects
315 self._add_outlet_port_objects(outlet_list)
317 def _create_outlet_list(self):
318 """
319 Create list of outlet stream names based on config arguments.
321 Returns:
322 list of strings
323 """
324 if self.config.outlet_list is not None and self.config.num_outlets is not None: 324 ↛ 326line 324 didn't jump to line 326 because the condition on line 324 was never true
325 # If both arguments provided and not consistent, raise Exception
326 if len(self.config.outlet_list) != self.config.num_outlets:
327 raise ConfigurationError(
328 "{} Separator provided with both outlet_list and "
329 "num_outlets arguments, which were not consistent ("
330 "length of outlet_list was not equal to num_outlets). "
331 "Please check your arguments for consistency, and "
332 "note that it is only necessry to provide one of "
333 "these arguments.".format(self.name)
334 )
335 elif self.config.outlet_list is None and self.config.num_outlets is None: 335 ↛ 337line 335 didn't jump to line 337 because the condition on line 335 was never true
336 # If no arguments provided for outlets, default to num_outlets = 2
337 self.config.num_outlets = 2
339 # Create a list of names for outlet StateBlocks
340 if self.config.outlet_list is not None:
341 outlet_list = self.config.outlet_list
342 else:
343 outlet_list = [
344 "outlet_" + str(n) for n in range(1, self.config.num_outlets + 1)
345 ]
347 return outlet_list
349 def _add_outlet_state_blocks(self, outlet_list):
350 """
351 Construct StateBlocks for all outlet streams.
353 Args:
354 list of strings to use as StateBlock names
356 Returns:
357 list of StateBlocks
358 """
359 # Setup StateBlock argument dict
360 tmp_dict = dict(**self.config.property_package_args)
361 tmp_dict["has_phase_equilibrium"] = False
362 tmp_dict["defined_state"] = False
364 # Create empty list to hold StateBlocks for return
365 outlet_blocks = []
367 # Create an instance of StateBlock for all outlets
368 for o in outlet_list:
369 o_obj = self.config.property_package.build_state_block(
370 self.flowsheet().time, doc="Material properties at outlet", **tmp_dict
371 )
373 setattr(self, o + "_state", o_obj)
375 outlet_blocks.append(getattr(self, o + "_state"))
377 return outlet_blocks
379 def _add_mixed_state_block(self):
380 """
381 Constructs StateBlock to represent mixed stream.
383 Returns:
384 New StateBlock object
385 """
386 # Setup StateBlock argument dict
387 tmp_dict = dict(**self.config.property_package_args)
388 tmp_dict["has_phase_equilibrium"] = False
389 tmp_dict["defined_state"] = True
391 self.mixed_state = self.config.property_package.build_state_block(
392 self.flowsheet().time, doc="Material properties of mixed stream", **tmp_dict
393 )
395 return self.mixed_state
397 def _add_inlet_port_objects(self, mixed_block):
398 """ Adds inlet Port object."""
399 self.add_port(name="inlet", block=mixed_block, doc="Inlet Port")
401 def _add_outlet_port_objects(self, outlet_list):
402 """Adds outlet Port objects."""
403 for p in outlet_list:
404 o_state = getattr(self, p + "_state")
405 self.add_port(name=p, block=o_state, doc="Outlet Port")
407 def _add_material_balance(self, mixed_block, outlet_blocks):
408 """Add overall material balance equation."""
409 # Get phase component list(s)
410 pc_set = mixed_block.phase_component_set
412 # Write phase-component balances
413 @self.Constraint(self.flowsheet().time, doc="Material balance equation")
414 def material_balance_equation(b, t):
415 return 0 == sum(
416 sum(
417 mixed_block[t].get_material_flow_terms(p, j)
418 -
419 sum(
420 o[t].get_material_flow_terms(p, j)
421 for o in outlet_blocks
422 )
423 for j in mixed_block.component_list
424 if (p, j) in pc_set
425 )
426 for p in mixed_block.phase_list
427 )
429 def _add_energy_balance(self, mixed_block, outlet_blocks):
430 """
431 Creates constraints for splitting the energy flows.
432 """
433 # split basis is equal_molar_enthalpy
434 @self.Constraint(
435 self.flowsheet().time,
436 self.outlet_idx,
437 doc="Molar enthalpy equality constraint",
438 )
439 def molar_enthalpy_equality_eqn(b, t, o):
440 o_block = getattr(self, o + "_state")
441 return mixed_block[t].enth_mol == o_block[t].enth_mol
443 def _add_momentum_balance(self, mixed_block, outlet_blocks):
444 """
445 Creates constraints for splitting the momentum flows - done by equating
446 pressures in outlets.
447 """
448 if self.config.momentum_balance_type is MomentumBalanceType.pressureTotal: 448 ↛ exitline 448 didn't return from function '_add_momentum_balance' because the condition on line 448 was always true
449 @self.Constraint(
450 self.flowsheet().time,
451 self.outlet_idx,
452 doc="Pressure equality constraint",
453 )
454 def pressure_equality_eqn(b, t, o):
455 o_block = getattr(self, o + "_state")
456 return mixed_block[t].pressure == o_block[t].pressure
458 def model_check(blk):
459 """
460 This method executes the model_check methods on the associated state
461 blocks (if they exist). This method is generally called by a unit model
462 as part of the unit's model_check method.
464 Args:
465 None
467 Returns:
468 None
469 """
470 # Try property block model check
471 for t in blk.flowsheet().time:
472 try:
473 blk.mixed_state[t].model_check()
474 except AttributeError:
475 _log.warning(
476 "{} Separator inlet state block has no "
477 "model check. To correct this, add a "
478 "model_check method to the associated "
479 "StateBlock class.".format(blk.name)
480 )
482 try:
483 outlet_list = blk._create_outlet_list()
484 for o in outlet_list:
485 o_block = getattr(blk, o + "_state")
486 o_block[t].model_check()
487 except AttributeError:
488 _log.warning(
489 "{} Separator outlet state block has no "
490 "model checks. To correct this, add a model_check"
491 " method to the associated StateBlock class.".format(blk.name)
492 )
494 def initialize_build(
495 blk, outlvl=idaeslog.NOTSET, optarg=None, solver=None, hold_state=False
496 ):
497 """
498 Initialization routine for separator
500 Keyword Arguments:
501 outlvl : sets output level of initialization routine
502 optarg : solver options dictionary object (default=None, use
503 default solver options)
504 solver : str indicating which solver to use during
505 initialization (default = None, use default solver)
506 hold_state : flag indicating whether the initialization routine
507 should unfix any state variables fixed during
508 initialization, **default** - False. **Valid values:**
509 **True** - states variables are not unfixed, and a dict of
510 returned containing flags for which states were fixed
511 during initialization, **False** - state variables are
512 unfixed after initialization by calling the release_state
513 method.
515 Returns:
516 If hold_states is True, returns a dict containing flags for which
517 states were fixed during initialization.
518 """
519 init_log = idaeslog.getInitLogger(blk.name, outlvl, tag="unit")
520 solve_log = idaeslog.getSolveLogger(blk.name, outlvl, tag="unit")
522 # Create solver
523 opt = get_solver(solver, optarg)
525 mblock = blk.mixed_state
526 flags = mblock.initialize(
527 outlvl=outlvl,
528 optarg=optarg,
529 solver=solver,
530 hold_state=True,
531 )
533 if degrees_of_freedom(blk) != 0: 533 ↛ 534line 533 didn't jump to line 534 because the condition on line 533 was never true
534 with idaeslog.solver_log(solve_log, idaeslog.DEBUG) as slc:
535 res = opt.solve(blk, tee=slc.tee)
536 init_log.info(
537 "Initialization Step 1 Complete: {}".format(idaeslog.condition(res))
538 )
540 # Initialize outlet StateBlocks
541 outlet_list = blk._create_outlet_list()
543 # Premises for initializing outlet states:
544 for o in outlet_list:
545 # Get corresponding outlet StateBlock
546 o_block = getattr(blk, o + "_state")
548 # Create dict to store fixed status of state variables
549 o_flags = {}
550 for t in blk.flowsheet().time:
552 # Calculate values for state variables
553 s_vars = o_block[t].define_state_vars()
554 for v in s_vars:
555 for k in s_vars[v]:
556 # Record whether variable was fixed or not
557 o_flags[t, v, k] = s_vars[v][k].fixed
559 # If fixed, use current value
560 # otherwise calculate guess from mixed state and fix
561 if not s_vars[v][k].fixed:
562 m_var = getattr(mblock[t], s_vars[v].local_name)
563 if "flow" in v:
564 # Leave initial value, but avoid negative flows
565 if s_vars[v][k].value < 1e-4:
566 s_vars[v][k].set_value(1e-2)
567 else:
568 # Otherwise intensive, equate to mixed stream
569 s_vars[v][k].set_value(m_var[k].value)
571 # Call initialization routine for outlet StateBlock
572 o_block.initialize(
573 outlvl=outlvl,
574 optarg=optarg,
575 solver=solver,
576 hold_state=False,
577 )
579 # Revert fixed status of variables to what they were before
580 for t in blk.flowsheet().time:
581 s_vars = o_block[t].define_state_vars()
582 for v in s_vars:
583 for k in s_vars[v]:
584 s_vars[v][k].fixed = o_flags[t, v, k]
586 init_log.info("Initialization Complete.")
587 return flags
589 def release_state(blk, flags, outlvl=idaeslog.NOTSET):
590 """
591 Method to release state variables fixed during initialization.
593 Keyword Arguments:
594 flags : dict containing information of which state variables
595 were fixed during initialization, and should now be
596 unfixed. This dict is returned by initialize if
597 hold_state = True.
598 outlvl : sets output level of logging
600 Returns:
601 None
602 """
603 mblock = blk.mixed_state
604 mblock.release_state(flags, outlvl=outlvl)
606 def calculate_scaling_factors(self):
607 mb_type = self.config.material_balance_type
608 mixed_state = self.mixed_state
609 if mb_type == MaterialBalanceType.useDefault:
610 t_ref = self.flowsheet().time.first()
611 mb_type = mixed_state[t_ref].default_material_balance_type()
612 super().calculate_scaling_factors()
614 if hasattr(self, "temperature_equality_eqn"):
615 for (t, i), c in self.temperature_equality_eqn.items():
616 s = iscale.get_scaling_factor(
617 mixed_state[t].temperature, default=1, warning=True
618 )
619 iscale.constraint_scaling_transform(c, s)
621 if hasattr(self, "pressure_equality_eqn"):
622 for (t, i), c in self.pressure_equality_eqn.items():
623 s = iscale.get_scaling_factor(
624 mixed_state[t].pressure, default=1, warning=True
625 )
626 iscale.constraint_scaling_transform(c, s)
628 if hasattr(self, "material_splitting_eqn"):
629 if mb_type == MaterialBalanceType.componentPhase:
630 for (t, _, p, j), c in self.material_splitting_eqn.items():
631 flow_term = mixed_state[t].get_material_flow_terms(p, j)
632 s = iscale.get_scaling_factor(flow_term, default=1)
633 iscale.constraint_scaling_transform(c, s)
634 elif mb_type == MaterialBalanceType.componentTotal:
635 for (t, _, j), c in self.material_splitting_eqn.items():
636 s = None
637 for p in mixed_state.phase_list:
638 try:
639 ft = mixed_state[t].get_material_flow_terms(p, j)
640 except KeyError:
641 # This component does not exist in this phase
642 continue
643 if s is None:
644 s = iscale.get_scaling_factor(ft, default=1)
645 else:
646 _s = iscale.get_scaling_factor(ft, default=1)
647 s = _s if _s < s else s
648 iscale.constraint_scaling_transform(c, s)
649 elif mb_type == MaterialBalanceType.total:
650 pc_set = mixed_state.phase_component_set
651 for (t, _), c in self.material_splitting_eqn.items():
652 for i, (p, j) in enumerate(pc_set):
653 ft = mixed_state[t].get_material_flow_terms(p, j)
654 if i == 0:
655 s = iscale.get_scaling_factor(ft, default=1)
656 else:
657 _s = iscale.get_scaling_factor(ft, default=1)
658 s = _s if _s < s else s
659 iscale.constraint_scaling_transform(c, s)
661 def _get_performance_contents(self, time_point=0):
662 if hasattr(self, "split_fraction"):
663 var_dict = {}
664 for k, v in self.split_fraction.items():
665 if k[0] == time_point:
666 var_dict[f"Split Fraction [{str(k[1:])}]"] = v
667 return {"vars": var_dict}
668 else:
669 return None
671 def _get_stream_table_contents(self, time_point=0):
672 outlet_list = self._create_outlet_list()
674 io_dict = {}
675 io_dict["Inlet"] = self.mixed_state
677 for o in outlet_list:
678 io_dict[o] = getattr(self, o + "_state")
680 return create_stream_table_dataframe(io_dict, time_point=time_point)