Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 13 additions & 8 deletions biosteam/facilities/hxn/_heat_exchanger_network.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,14 +149,18 @@ def _cost(self):
flowsheet = bst.Flowsheet(sys.ID + '_HXN')
use_cached_network = False
if self.cache_network and hasattr(self, 'original_heat_utils'):
hxs = [hu.unit for hu in hx_utils]
# Units are a stable key to compare whether system has changed configuration.
hu_by_unit = {hu.unit: hu for hu in hx_utils}
use_cached_network = (
sorted(hxs, key=lambda x: x.ID)
== sorted(self.original_heat_exchangers, key=lambda x: x.ID)
hu_by_unit.keys() == set(self.original_heat_exchangers)
)
with flowsheet.temporary(), bst.IgnoreDockingWarnings():
if use_cached_network:
hx_heat_utils_rearranged = [i.heat_utilities[0] for i in hxs]
if use_cached_network:
# Keep the same configuration, but update stream life cycles and heat exchanger specifications.
# Note that stream_life_cycles are aligned with original_heat_exchangers (from synthesize_network).
# Heat utils are rearranged so that they align with stream_life_cycles too.
hxs = self.original_heat_exchangers
hx_heat_utils_rearranged = [hu_by_unit[hx] for hx in hxs]
stream_life_cycles = self.stream_life_cycles
new_HXs = self.new_HXs
new_HX_utils = self.new_HX_utils
Expand Down Expand Up @@ -248,8 +252,9 @@ def _cost(self):
np.testing.assert_allclose(s_util.imol[IDs], s_lc.imol[IDs])
np.testing.assert_allclose(P, s_lc.P, rtol=1e-3, atol=0.1)
np.testing.assert_allclose(s_util.H, s_lc.H, rtol=1e-3, atol=1.)
except:
msg = ("heat exchanger network cache algorithm failed, cached network ignored")
except AssertionError as e:
msg = ("heat exchanger network cache algorithm failed, "
f"cached network ignored: {e}")
warn(msg, RuntimeWarning, stacklevel=2)
del self.original_heat_utils
self._cost()
Expand Down Expand Up @@ -337,7 +342,7 @@ def _energy_balance_error_contributions(self):
if ignored and callable(ignored): ignored = ignored()
energy_balance_errors = {}
for hu in self._get_original_heat_utilties():
self.ignored = ignored + [hu.unit]
self.ignored = list(ignored or ()) + [hu.unit]
if hasattr(hu.unit, 'owner'):
ID = hu.unit.owner.ID, hu.unit.ID
else:
Expand Down
96 changes: 96 additions & 0 deletions tests/test_hxn.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,96 @@
# -*- coding: utf-8 -*-
# BioSTEAM: The Biorefinery Simulation and Techno-Economic Analysis Modules
# Copyright (C) 2026-, Sarang Bhagwat <sarangbhagwat.developer@gmail.com>
#
# This module is under the UIUC open-source license. See
# github.com/BioSTEAMDevelopmentGroup/biosteam/blob/master/LICENSE.txt
# for license details.
"""
Tests for the heat exchanger network facility.
"""
import warnings
import biosteam as bst
import numpy as np
from numpy.testing import assert_allclose

def build_system(N_columns=1):
"""Doctest system of HeatExchangerNetwork; `N_columns > 1` adds more
ShortcutColumns so auxiliary heat exchangers have duplicate IDs."""
bst.settings.set_thermo(['Water', 'Methanol', 'Glycerol'], cache=True)
bst.main_flowsheet.set_flowsheet('test_hxn')
units = []
feeds = []
for i in range(N_columns):
feed = bst.Stream(f'feed{i}', flow=(8000, 100 * (i + 1), 25))
D = bst.ShortcutColumn(f'D{i}', ins=feed, LHK=('Methanol', 'Water'),
y_top=0.99, x_bot=0.01, k=2, is_divided=True)
H1 = bst.HXutility(f'D{i}_H1', ins=D.outs[1], T=300)
H2 = bst.HXutility(f'D{i}_H2', ins=D.outs[0], T=300)
units.extend([D, H1, H2])
feeds.append(feed)
feed2 = bst.Stream('feed_flash', flow=(10000, 1000, 10))
F1 = bst.Flash('F1', ins=feed2, V=0.9, P=101325)
HXN = bst.HeatExchangerNetwork('HXN', T_min_app=5.)
sys = bst.System.from_units('sys', units=[*units, F1, HXN])
return sys, HXN, feeds[0]

def network_results(HXN):
return dict(
heat=HXN.actual_heat_util_load,
cool=HXN.actual_cool_util_load,
Q=np.array([hx.Q for hx in HXN.new_HXs]),
installed=HXN.installed_costs['Heat exchangers'],
)

def assert_same_results(a, b, rtol=2e-3):
for key in a:
assert_allclose(a[key], b[key], rtol=rtol, err_msg=key)

def simulate_cached(sys, HXN):
HXN.cache_network = True
HXN_sys = HXN.HXN_sys
with warnings.catch_warnings():
warnings.simplefilter('error', RuntimeWarning)
sys.simulate()
assert HXN.HXN_sys is HXN_sys, 'cached network was not used'

def test_cache_network_matches_fresh_synthesis():
sys, HXN, feed = build_system()
sys.simulate()
fresh = network_results(HXN)
assert HXN.actual_heat_util_load < 0.9 * HXN.original_heat_util_load
simulate_cached(sys, HXN)
assert_same_results(network_results(HXN), fresh)

def test_cache_network_perturbed_feed():
sys, HXN, feed = build_system()
sys.simulate()
feed.F_mass *= 1.000001
simulate_cached(sys, HXN)
cached = network_results(HXN)
HXN.cache_network = False
sys.simulate()
assert_same_results(cached, network_results(HXN), rtol=1e-3)

def test_cache_network_duplicate_IDs():
sys, HXN, feed = build_system(N_columns=2)
sys.simulate()
IDs = [hx.ID for hx in HXN.original_heat_exchangers]
assert len(IDs) != len(set(IDs)), 'test needs duplicate auxiliary IDs'
fresh = network_results(HXN)
simulate_cached(sys, HXN)
assert_same_results(network_results(HXN), fresh)

def test_energy_balance_error_contributions_ignored_none():
sys, HXN, feed = build_system()
sys.simulate()
N = len(HXN.original_heat_utils)
errors = HXN._energy_balance_error_contributions()
assert len(errors) == N
assert HXN.ignored is None

if __name__ == '__main__':
test_cache_network_matches_fresh_synthesis()
test_cache_network_perturbed_feed()
test_cache_network_duplicate_IDs()
test_energy_balance_error_contributions_ignored_none()
Loading