enhancements to PO-API

benchmark/acetic_acid_system.py

@@ -60,7 +60,7 @@ def fresh_solvent_flow_rate():
  0, broth * solvent_feed_ratio - EtAc_recycle
  )
 
- @solvent.equation('material')
+ @solvent.material_balance
  def fresh_solvent_flow_rate():
  s = np.ones(chemicals.size)
  r = np.zeros(chemicals.size)
@@ -104,7 +104,7 @@ def create_acetic_acid_complex_system(alg):
  distillate = bst.Stream('distillate')
  distillate_2 = bst.Stream('distillate_2')
  with bst.System(algorithm=alg) as sys:
- # @ethyl_acetate.equation('material')
+ # @ethyl_acetate.material_balance
  # def fresh_solvent_flow_rate():
  # f = np.ones(chemicals.size)
  # r = np.zeros(chemicals.size)
@@ -119,7 +119,7 @@ def create_acetic_acid_complex_system(alg):
  # distillate_2: r},
  # v
  # )
- @ethyl_acetate.equation('material')
+ @ethyl_acetate.material_balance
  def fresh_solvent_flow_rate():
  f = np.ones(chemicals.size)
  r = np.zeros(chemicals.size)

biosteam/_system.py

@@ -73,7 +73,7 @@ def __init__(self, f, args):
  def __call__(self): self.f(*self.args)
 
 
-# %% Reconfiguration for phenomena oriented simulation
+# %% Reconfiguration for phenomena oriented simulation and convergence
 
 class Configuration:
  __slots__ = ('stages', 'nodes', 'streams', 'stream_ref', 'connections')
@@ -85,42 +85,13 @@ def __init__(self, stages, nodes, streams, stream_ref, connections):
  self.stream_ref = stream_ref
  self.connections = connections
 
- def solve_variable(self, variable):
- return solve_linear_equations(variable, self.nodes, self.stream_ref)
-
- def __enter__(self):
- units = self.stages
- streams = self.streams
- sinks = {}
- sources = {}
- for u in units:
- for i in u.ins: 
- i._sink = u
- sinks[i.imol] = u
- for i in u.outs: 
- i._source = u
- sources[i.imol] = u
- units_set = set(units)
- for i in streams:
- if i.source and i.source not in units_set: i._source = sources[i.imol]
- if i.sink and i.sink not in units_set: i._sink = sinks[i.imol]
- return self
-
- def __exit__(self, type, exception, traceback):
- for s, source, sink in self.connections:
- s._source = source
- s._sink = sink
- if exception: raise exception
-
-
-# %% Phenomenological convergence tools
-
-def solve_linear_equations(variable, nodes, streams):
- A = []
- b = []
- if variable == 'material':
+ def solve_material_flows(self):
+ nodes = self.nodes
+ streams = self.stream_ref
+ A = []
+ b = []
  for node in nodes:
- for coefficients, value in node._create_linear_equations(variable):
+ for coefficients, value in node._create_material_balance_equations():
  coefficients = {
  streams[i.imol]: j for i, j in coefficients.items()
  }
@@ -140,7 +111,7 @@ def solve_linear_equations(variable, nodes, streams):
  b_ready = np.array([i[ready_index] for i in b], float)
  values = solve(A_ready, b_ready.T).T
  for obj, value in zip(objs, values): 
- obj._update_decoupled_variable(variable, value, ready_index)
+ obj._update_material_flows(value, ready_index)
  A_delayed = [{i: j[delayed_index] for i, j in dct.items()} for dct in A]
  A_delayed, objs = dictionaries2array(A_delayed)
  b_delayed = np.array([
@@ -149,29 +120,57 @@ def solve_linear_equations(variable, nodes, streams):
  ], float)
  values = solve(A_delayed, b_delayed.T).T
  for obj, value in zip(objs, values): 
- obj._update_decoupled_variable(variable, value, delayed_index)
+ obj._update_material_flows(value, delayed_index)
  else:
  A, objs = dictionaries2array(A)
  values = solve(A, np.array(b).T).T
  for obj, value in zip(objs, values): 
- obj._update_decoupled_variable(variable, value)
+ obj._update_material_flows(value)
  for i in nodes: 
  if hasattr(i, '_update_auxiliaries'):
  i._update_auxiliaries()
  elif hasattr(i, 'stages'):
  for i in i.stages:
  if hasattr(i, '_update_auxiliaries'):
  i._update_auxiliaries()
- else:
+
+ def solve_energy_departures(self):
+ nodes = self.nodes
+ A = []
+ b = []
  for node in nodes:
- for coefficients, value in node._create_linear_equations(variable):
+ for coefficients, value in node._create_energy_departure_equations():
  A.append(coefficients)
  b.append(value)
  A, objs = dictionaries2array(A)
  values = solve(A, np.array(b).T).T
  for obj, value in zip(objs, values): 
- obj._update_decoupled_variable(variable, value)
- return objs, values
+ obj._update_energy_variable(value)
+
+ def __enter__(self):
+ units = self.stages
+ streams = self.streams
+ sinks = {}
+ sources = {}
+ for u in units:
+ for i in u.ins: 
+ i._sink = u
+ sinks[i.imol] = u
+ for i in u.outs: 
+ i._source = u
+ sources[i.imol] = u
+ units_set = set(units)
+ for i in streams:
+ if i.source and i.source not in units_set: i._source = sources[i.imol]
+ if i.sink and i.sink not in units_set: i._sink = sinks[i.imol]
+ return self
+
+ def __exit__(self, type, exception, traceback):
+ for s, source, sink in self.connections:
+ s._source = source
+ s._sink = sink
+ if exception: raise exception
+
 
 # %% Sequential modular convergence tools
 
@@ -715,9 +714,6 @@ class System:
  #: Method definitions for convergence
  available_methods: Methods[str, tuple[Callable, bool, dict]] = Methods()
 
- #: Variable solution priority for phenomena oriented simulation.
- variable_priority: list[str] = ['equilibrium', 'material', 'energy', 'material']
-
  @classmethod
  def register_method(cls, name, solver, conditional=False, **kwargs):
  """
@@ -2264,7 +2260,7 @@ def stage_configuration(self, aggregated=False):
  else:
  return self._stage_configuration
  except:
- feeds = [i for i in self.feeds if i.equations]
+ feeds = [i for i in self.feeds if i.material_equations]
  stages = self.stages
  streams = list(set([i for s in stages for i in s.ins + s.outs]))
  connections = [(i, i.source, i.sink) for i in streams]
@@ -2287,22 +2283,30 @@ def stage_configuration(self, aggregated=False):
  def run_phenomena(self):
  """Decouple and linearize material, equilibrium, summation, enthalpy,
  and reaction phenomena and iteratively solve them."""
- path = self.unit_path
+ *path, last = self.unit_path
  try:
  for n, i in enumerate(path):
  i.run()
  with self.stage_configuration(aggregated=False) as conf:
- for variable in ('material', 'energy'): conf.solve_variable(variable)
+ conf.solve_material_flows()
+ conf.solve_energy_departures()
  except:
  for i in path[n+1:]: i.run()
+ last.run()
  for i in self.stages: 
- if getattr(i, 'phases', None) == ('g', 'l'): i._create_linear_equations('equilibrium')
+ if (hasattr(i, '_update_equilibrium_variables') 
+ and getattr(i, 'phases', None) == ('g', 'l')):
+ i._update_equilibrium_variables()
+ if hasattr(i, '_update_reaction_conversion'): 
+ i._update_reaction_conversion()
  try:
  with self.stage_configuration(aggregated=False) as conf:
- for variable in ('material', 'energy'): conf.solve_variable(variable)
+ conf.solve_energy_departures()
+ conf.solve_material_flows()
  except: 
  with self.stage_configuration(aggregated=True) as conf:
- for variable in ('material', 'energy'): conf.solve_variable(variable)
+ conf.solve_energy_departures()
+ conf.solve_material_flows()
 
  def _solve(self):
  """Solve the system recycle iteratively."""

biosteam/_unit.py

@@ -39,12 +39,6 @@
 # stream = Union[Annotated[Union[Stream, str, None], 1], Union[Stream, str, None]]
 # stream_sequence = Collection[Union[Stream, str, None]]
 
-# %% Defaults
-
-def _get_energy_departure_coefficient(self, stream): return
-
-def _update_decoupled_variable(self, variable, value): return
-
 
 # %% Unit Operation
 
@@ -121,32 +115,10 @@ class Unit(AbstractUnit):
  def __init_subclass__(cls,
  isabstract=False,
  new_graphics=True,
- does_nothing=None,
- phenomena_oriented=None):
+ does_nothing=None):
  super().__init_subclass__()
  if does_nothing: return 
  dct = cls.__dict__
- if phenomena_oriented:
- signatures = (
- ('_create_linear_equations', ('self', 'variable'), None),
- ('_update_decoupled_variable', ('self', 'variable', 'value'), _update_decoupled_variable),
- ('_get_energy_departure_coefficient', ('self', 'stream'), _get_energy_departure_coefficient),
- )
- for name, params, default in signatures:
- if hasattr(cls, name):
- f = signature(getattr(cls, name))
- if tuple(f.parameters) != params:
- raise NotImplementedError(
- 'invalid signtaure {name}.{str(f)}; correct signature is '
- '{name}({', '.join(params)})'
- )
- elif default:
- setattr(cls, name, default)
- else:
- raise NotImplementedError(
- f'unit must implement a `{name}` '
- 'method for phenomena-oriented simulation'
- )
  if '_N_heat_utilities' in dct:
  warn("'_N_heat_utilities' class attribute is scheduled for deprecation; "
  "use the `add_heat_utility` method instead",
@@ -258,14 +230,26 @@ def __init_subclass__(cls,
  #: Add itemized purchase costs to the :attr:`~Unit.baseline_purchase_costs` dictionary.
  _cost = AbstractMethod
 
- #: For embodied emissions (e.g., unit construction) in LCA
+ #: Add embodied emissions (e.g., unit construction) in LCA
  _lca = AbstractMethod
 
- #: dict[Unit, float] Return unit coefficients for the given variable.
- _create_linear_equations = AbstractMethod
+ #: Create material balance equations for phenomena-oriented simulation.
+ _create_material_balance_equations = AbstractMethod
+
+ #: Create energy departure equations for phenomena-oriented simulation.
+ _create_energy_departure_equations = AbstractMethod
+
+ #: Return energy departure coefficient of a stream for phenomena-oriented simulation.
+ _get_energy_departure_coefficient = AbstractMethod
+
+ #: Update energy variable being solved in energy departure equations for phenomena-oriented simulation.
+ _update_energy_variable = AbstractMethod
+
+ #: Update equilibrium variables for phenomena-oriented simulation.
+ _update_equilibrium_variables = AbstractMethod
 
- #: Update internal variables related to mass and energy balances.
- _update_variables = AbstractMethod
+ #: Update reaction conversion for phenomena-oriented simulation.
+ _update_reaction_conversion = AbstractMethod
 
  Inlets = piping.Inlets
  Outlets = piping.Outlets

biosteam/units/compressor.py

@@ -370,7 +370,7 @@ def _design(self):
  self._set_power(ideal_power / self.eta)
 
 
-class IsentropicCompressor(Compressor, new_graphics=False, phenomena_oriented=True):
+class IsentropicCompressor(Compressor, new_graphics=False):
  """
  Create an isentropic compressor.
 
@@ -567,11 +567,8 @@ def _run(self):
  def _design(self):
  super()._design()
  self._set_power(self.design_results['Ideal power'] / self.eta)
-
- def _create_energy_departure_equations(self):
- return [self._create_energy_departure_equation()]
 
- def _create_energy_departure_equation(self):
+ def _create_energy_departure_equations(self):
  # Ll: C1dT1 - Ce2*dT2 - Cr0*dT0 - hv2*L2*dB2 = Q1 - H_out + H_in
  # gl: hV1*L1*dB1 - hv2*L2*dB2 - Ce2*dT2 - Cr0*dT0 = Q1 + H_in - H_out
  feed = self.ins[0]
@@ -582,11 +579,9 @@ def _create_energy_departure_equation(self):
  J = R / (2. * product.Cn * self.eta) * log(feed.P / self.P)
  coeff = {self: (1 - J) / (1 + J) * product.Cn}
  for i in self.ins: i._update_energy_departure_coefficient(coeff)
- return (coeff, 0)
+ return [(coeff, 0)]
 
  def _create_material_balance_equations(self):
- top_split = self.top_split
- bottom_split = self.bottom_split
  inlets = self.ins
  fresh_inlets = [i for i in inlets if i.isfeed() and not i.equations]
  process_inlets = [i for i in inlets if not i.isfeed() or i.equations]
@@ -605,18 +600,8 @@ def _create_material_balance_equations(self):
 
  return equations
 
- def _create_linear_equations(self, variable):
- # list[dict[Unit|Stream, float]]
- if variable == 'material':
- eqs = self._create_material_balance_equations()
- elif variable == 'energy':
- eqs = self._create_energy_departure_equations()
- else:
- eqs = []
- return eqs
-
- def _update_decoupled_variable(self, variable, value):
- if variable == 'energy': self.outs[0].T += value
+ def _update_energy_variable(self, variable, value):
+ self.outs[0].T += value
 
 
 class PolytropicCompressor(Compressor, new_graphics=False):