"""Direct heat integration entry point for process and unit-level targeting."""
from __future__ import annotations
from typing import List, Tuple
import numpy as np
from ...domain.configuration import tol
from ...domain.enums import GraphType, ProblemTableLabel, StreamType, TargetType
from ...domain.problem_table import ProblemTable
from ...domain.stream import Stream
from ...domain.stream_collection import StreamCollection
from ...domain.targets import DirectIntegrationTarget
from ...domain.zone import Zone
from ..numerics import delta_vals, get_period_index
from .area_cost import (
get_area_targets,
get_balanced_CC,
get_capital_cost_targets,
get_min_number_hx,
)
from .cascade import (
get_heat_recovery_target_from_pt,
get_process_heat_cascade,
set_zonal_targets,
)
from .grand_composite import get_additional_GCCs
from .utilities import get_utility_targets
__all__ = ["compute_direct_integration_targets"]
################################################################################
# Public API
################################################################################
[docs]
def compute_direct_integration_targets(
zone: Zone,
args: dict | None = None,
) -> DirectIntegrationTarget:
"""Populate a ``Zone`` with detailed direct heat integration pinch targets.
The function aggregates Problem Table calculations, multi-utility targeting,
pinch temperature detection, and graph preparation. Results are cached on
the provided ``zone`` and used later by site and regional aggregation routines.
"""
idx, sid = get_period_index(period_ids=zone.period_ids, args=args)
all_streams = zone.all_streams
pt = get_process_heat_cascade(
hot_streams=zone.hot_streams,
cold_streams=zone.cold_streams,
all_streams=all_streams,
is_shifted=True,
period_idx=idx,
)
pt_real = get_process_heat_cascade(
hot_streams=zone.hot_streams,
cold_streams=zone.cold_streams,
all_streams=all_streams,
is_shifted=False,
known_heat_recovery=get_heat_recovery_target_from_pt(pt),
period_idx=idx,
)
hot_pinch, cold_pinch = pt.pinch_temperatures()
direct = zone.config.direct
calculate_area_cost = bool((args or {}).get("_calculate_area_cost", False))
pt = get_additional_GCCs(
pt,
do_vert_cc_calc=direct.vertical_gcc_enabled,
do_assisted_ht_calc=direct.assisted_ht_enabled,
assisted_ht_dt_cut=direct.assisted_ht_dt,
)
get_utility_targets(
pt=pt,
pt_real=pt_real,
hot_utilities=zone.hot_utilities,
cold_utilities=zone.cold_utilities,
is_direct_integration=True,
idx=idx,
)
zone.net_hot_streams, zone.net_cold_streams = (
_create_net_hot_and_cold_stream_collections_for_site_analysis(
T_vals=pt[ProblemTableLabel.T],
H_vals=pt[ProblemTableLabel.H_NET_A],
hot_utilities=zone.hot_utilities,
cold_utilities=zone.cold_utilities,
idx=idx,
)
)
if should_update_balanced_composite_curves(direct, calculate_area_cost):
pt.update(
**get_balanced_CC(
T_col=pt[ProblemTableLabel.T],
H_hot=pt[ProblemTableLabel.H_HOT],
H_cold=pt[ProblemTableLabel.H_COLD],
H_hot_ut=pt[ProblemTableLabel.H_HOT_UT],
H_cold_ut=pt[ProblemTableLabel.H_COLD_UT],
)
)
pt_real.update(
**get_balanced_CC(
T_col=pt_real[ProblemTableLabel.T],
H_hot=pt_real[ProblemTableLabel.H_HOT],
H_cold=pt_real[ProblemTableLabel.H_COLD],
H_hot_ut=pt_real[ProblemTableLabel.H_HOT_UT],
H_cold_ut=pt_real[ProblemTableLabel.H_COLD_UT],
dT_vals=pt_real[ProblemTableLabel.DELTA_T],
RCP_hot=pt_real[ProblemTableLabel.RCP_HOT],
RCP_cold=pt_real[ProblemTableLabel.RCP_COLD],
RCP_hot_ut=pt_real[ProblemTableLabel.RCP_HOT_UT],
RCP_cold_ut=pt_real[ProblemTableLabel.RCP_COLD_UT],
)
)
area_data = build_area_cost_target_data(
pt,
pt_real,
zone,
idx,
enabled=calculate_area_cost,
)
else:
area_data = {}
target_data = (
set_zonal_targets(
pt=pt,
pt_real=pt_real,
)
| {
"zone_name": zone.name,
"type": TargetType.DI.value,
"parent_zone": zone.parent_zone,
"config": zone.config,
"pt": pt,
"pt_real": pt_real,
"graphs": _save_graph_data(pt, pt_real),
"hot_utilities": zone.hot_utilities,
"cold_utilities": zone.cold_utilities,
"hot_pinch": hot_pinch,
"cold_pinch": cold_pinch,
"period_id": sid,
"period_idx": idx,
}
| area_data
)
return DirectIntegrationTarget.model_validate(target_data)
################################################################################
# Helper functions
################################################################################
def should_update_balanced_composite_curves(
direct_config,
calculate_area_cost: bool = False,
) -> bool:
"""Return True when balanced composite curves are needed downstream."""
return bool(direct_config.balanced_cc_enabled or calculate_area_cost)
def build_area_cost_target_data(
pt: ProblemTable,
pt_real: ProblemTable,
zone: Zone,
idx: int | None,
*,
enabled: bool,
) -> dict:
"""Calculate direct-integration area and capital-cost target fields."""
if not enabled:
return {}
num_units = get_min_number_hx(
T_vals=pt[ProblemTableLabel.T],
H_hot_bal=pt[ProblemTableLabel.H_HOT_BAL],
H_cold_bal=pt[ProblemTableLabel.H_COLD_BAL],
hot_streams=zone.hot_streams,
cold_streams=zone.cold_streams,
hot_utilities=zone.hot_utilities,
cold_utilities=zone.cold_utilities,
idx=idx,
)
area = get_area_targets(
T_vals=pt_real[ProblemTableLabel.T],
H_hot_bal=pt_real[ProblemTableLabel.H_HOT_BAL],
H_cold_bal=pt_real[ProblemTableLabel.H_COLD_BAL],
R_hot_bal=pt_real[ProblemTableLabel.R_HOT_BAL],
R_cold_bal=pt_real[ProblemTableLabel.R_COLD_BAL],
)
capital_cost, annual_capital_cost = get_capital_cost_targets(
area=area,
num_units=num_units,
config=zone.config,
)
return {
"area": area,
"num_units": num_units,
"capital_cost": capital_cost,
"total_cost": annual_capital_cost,
}
def _create_net_hot_and_cold_stream_collections_for_site_analysis(
T_vals: np.ndarray,
H_vals: np.ndarray,
hot_utilities: StreamCollection,
cold_utilities: StreamCollection,
idx: int | None = None,
) -> Tuple[StreamCollection, StreamCollection]:
"""Construct net stream segments requiring utility input by interval."""
net_hot_streams = StreamCollection()
net_cold_streams = StreamCollection()
hot_utilities_seq = list(hot_utilities)
cold_utilities_seq = list(cold_utilities)
hot_remaining = [
StreamCollection._value_at_idx(utility._heat_flow, idx)
for utility in hot_utilities_seq
]
cold_remaining = [
StreamCollection._value_at_idx(utility._heat_flow, idx)
for utility in cold_utilities_seq
]
if np.nansum(hot_remaining) + np.nansum(cold_remaining) < tol:
# If no utility is needed, there is no net streams for indirect integration.
return net_hot_streams, net_cold_streams
if delta_vals(T_vals).min() < tol:
raise ValueError("Infeasible temperature interval detected in _store_TSP_data")
T_vals = T_vals
dh_vals = delta_vals(H_vals)
hu_idx = _initialise_utility_index(hot_utilities_seq, hot_remaining)
cu_idx = _initialise_utility_index(cold_utilities_seq, cold_remaining)
k = 1
for i, dh in enumerate(dh_vals):
if dh > tol and hu_idx >= 0:
hu_idx, k = _add_net_segment_period(
T_ub=T_vals[i],
T_lb=T_vals[i + 1],
curr_idx=hu_idx,
dh_req=dh,
utilities=hot_utilities_seq,
remaining=hot_remaining,
net_streams=net_cold_streams,
k=k,
idx=idx,
)
elif -dh > tol and cu_idx >= 0:
cu_idx, k = _add_net_segment_period(
T_ub=T_vals[i],
T_lb=T_vals[i + 1],
curr_idx=cu_idx,
dh_req=abs(dh),
utilities=cold_utilities_seq,
remaining=cold_remaining,
net_streams=net_hot_streams,
k=k,
idx=idx,
)
return net_hot_streams, net_cold_streams
def _add_net_segment_period(
T_ub: float,
T_lb: float,
curr_idx: int,
dh_req: float,
utilities: StreamCollection,
remaining: List[float],
net_streams: StreamCollection,
k: int,
j: int = 0,
idx: int | None = None,
) -> Tuple[int, int]:
"""Add net utility segments, splitting iteratively across utilities."""
if curr_idx < 0 or not utilities or dh_req <= tol:
return curr_idx, k
segment_upper = float(T_ub)
segment_lower = float(T_lb)
remaining_dh = float(dh_req)
split_idx = int(j)
while remaining_dh > tol and curr_idx >= 0:
available = float(remaining[curr_idx]) if curr_idx < len(remaining) else 0.0
if available <= tol:
next_idx = _find_next_available_utility(curr_idx + 1, utilities, remaining)
if next_idx == curr_idx:
break
curr_idx = next_idx
continue
curr_u = utilities[curr_idx]
dh_curr = float(min(remaining_dh, available))
remaining[curr_idx] = max(available - dh_curr, 0.0)
span = segment_upper - segment_lower
split_temp = (
segment_upper - (dh_curr / remaining_dh) * span
if span and remaining_dh > tol
else segment_lower
)
net_streams.add(
Stream(
name=f"Segment {k}" if split_idx == 0 else f"Segment {k}-{split_idx}",
supply_temperature=(
split_temp
if curr_u.stream_type == StreamType.Hot.value
else segment_upper
),
target_temperature=(
segment_upper
if curr_u.stream_type == StreamType.Hot.value
else split_temp
),
heat_flow=dh_curr,
delta_t_contribution=StreamCollection._value_at_idx(
curr_u._dt_cont, idx
),
delta_t_contribution_multiplier=curr_u.delta_t_contribution_multiplier,
heat_transfer_coefficient=1.0,
is_process_stream=True,
)
)
remaining_dh -= dh_curr
if remaining_dh <= tol:
break
segment_upper = split_temp
curr_idx = _find_next_available_utility(curr_idx + 1, utilities, remaining)
split_idx += 1
next_idx = (
curr_idx
if 0 <= curr_idx < len(remaining) and remaining[curr_idx] > tol
else _find_next_available_utility(curr_idx + 1, utilities, remaining)
)
return next_idx, k + 1
def _initialise_utility_index(
utilities: StreamCollection, remaining: List[float]
) -> int:
"""Returns the index of the first available utility with remaining capacity."""
for idx, residual in enumerate(remaining):
if residual > tol:
return idx
return len(utilities) - 1 if utilities else -1
def _find_next_available_utility(
start: int, utilities: StreamCollection, remaining: List[float]
) -> int:
"""Return the index of the next utility that still has remaining duty."""
if not utilities:
return -1
for idx in range(max(start, 0), len(utilities)):
if remaining[idx] > tol:
return idx
return len(utilities) - 1
def _save_graph_data(pt: ProblemTable, pt_real: ProblemTable) -> dict:
"""Assemble the Problem Table slices required for composite/comparison plots."""
pt.round(decimals=4)
pt_real.round(decimals=4)
return {
GraphType.CC.value: pt_real.slice(
[ProblemTableLabel.T, ProblemTableLabel.H_HOT, ProblemTableLabel.H_COLD]
),
GraphType.SCC.value: pt.slice(
[ProblemTableLabel.T, ProblemTableLabel.H_HOT, ProblemTableLabel.H_COLD]
),
GraphType.BCC.value: pt_real.slice(
[
ProblemTableLabel.T,
ProblemTableLabel.H_HOT_BAL,
ProblemTableLabel.H_COLD_BAL,
]
),
GraphType.GCC.value: pt.slice(
[
ProblemTableLabel.T,
ProblemTableLabel.H_NET,
ProblemTableLabel.H_NET_NP,
ProblemTableLabel.H_NET_V,
ProblemTableLabel.H_NET_A,
ProblemTableLabel.H_NET_UT,
]
),
GraphType.GCC_R.value: pt_real.slice(
[ProblemTableLabel.T, ProblemTableLabel.H_NET, ProblemTableLabel.H_NET_UT]
),
GraphType.NLP.value: pt.slice(
[
ProblemTableLabel.T,
ProblemTableLabel.H_NET_HOT,
ProblemTableLabel.H_NET_COLD,
ProblemTableLabel.H_HOT_UT,
ProblemTableLabel.H_COLD_UT,
ProblemTableLabel.H_HOT_HP,
ProblemTableLabel.H_COLD_HP,
]
),
GraphType.GCC_HP.value: pt.slice(
[
ProblemTableLabel.T,
ProblemTableLabel.H_NET_W_AIR,
ProblemTableLabel.H_NET_HP,
]
),
}