Domain Model

Once a problem is prepared, OpenPinch operates on a small set of in-memory domain objects. Understanding these classes is the key to using the package as more than a black box.

Core Objects

Zone: Hierarchical analysis boundary containing streams, utilities, targets, and graphs.
Stream: Process or utility stream with supply/target states, shifted temperatures, and active/base dt_cont behavior.
StreamCollection: Ordered container with hot/cold filtering and utility inversion helpers.
ProblemTable: Numerical temperature-interval table behind composite curves, pinch temperatures, utility cascades, and several advanced targeting routines.

These are the objects you inspect when you need to understand how a case was prepared or why a target changed after mutating the in-memory model.

Key Classes

class OpenPinch.classes.zone.Zone(name='Zone', type='Process Zone', zone_config=None, parent_zone=None)[source]

Bases: object

Hierarchical analysis boundary containing streams, utilities, and targets.

Zones form the backbone of the in-memory OpenPinch model. Each zone can own process streams, utility streams, solved targets, generated graphs, and nested child zones. Direct and indirect integration routines progressively populate this structure as the analysis moves from local process scopes up to site-style aggregation.

Initialise an empty zone with stream, target, and graph containers.

Parameters:

name (str)
type (str)
zone_config (Configuration | None)
parent_zone (Zone)

property name: Display name used when addressing the zone in the hierarchy.

property type: Zone type type from ZoneType.

property config: Configuration object controlling analysis behaviour for this zone.

property parent_zone: Direct parent zone in the site hierarchy, if any.

property active: bool: Whether the zone participates in the current analysis.

property state_ids: dict[str, int] | None: Canonical state_id -> idx lookup for this zone.

property weights: Canonical state weights for this zone.

property num_states: Number of distinct states for this zone.

property address: str: Slash-delimited path from the root zone to this zone.

property dt_cont_multiplier: float: Effective multiplier applied to stream and utility dt_cont values.

property hot_streams: Process streams that release heat within this zone.

property cold_streams: Process streams that require heat within this zone.

property net_hot_streams: Net hot streams derived from zonal aggregation.

property net_cold_streams: Net cold streams derived from zonal aggregation.

property hot_utilities: Hot utility streams assigned to the zone.

property cold_utilities: Cold utility streams assigned to the zone.

property graphs: Graphs generated for this zone.

property subzones: Immediate child zones keyed by name.

property targets: Energy targets keyed by target name.

property process_streams: Combined hot and cold process streams for the zone.

property net_process_streams: Combined net hot and net cold process streams for the zone.

property utility_streams: Combined hot and cold utility streams for the zone.

property all_streams: All process and utility streams defined on the zone.

set_state_context(state_ids, weights, num_states)[source]

Set the canonical state lookup owned by this zone and propagate refs.

Parameters:

state_ids (dict[str, int] | list[str] | tuple[str, ...] | None)
num_states (int | None)

Return type:

None

add_graph(name, result)[source]

Store a graph result under name for later export or display.

Parameters:: name (str)

add_zone(zone_to_add, sub=True)[source]

Add a single zone object keyed by its name.

If the zone name already exists: - If the zone is identical (e.g. same stream and utility objects), skip. - If it’s different, add it with a suffix like ‘_1’, ‘_2’, etc.

Parameters:: sub (bool)

add_target(target_to_add)[source]

Add one target to a specific zone.

Parameters:: target_to_add (BaseTargetModel)

add_targets(targets=[])[source]

Add multiple targets to a specific zone.

Parameters:: targets (list)

get_subzone(loc=None)[source]

Resolve a slash-delimited zone path relative to this zone.

Parameters:: loc (str)
Return type:: Zone

calc_utility_cost()[source]: Calculate and cache the annual utility cost across assigned utilities.

import_hot_and_cold_streams_from_sub_zones(get_net_streams=False, is_n_zone_depth=True, is_new_stream_collection=True)[source]

Get referenced hot and cold streams across multiple subzones.

Parameters:

get_net_streams (bool)
is_n_zone_depth (bool)
is_new_stream_collection (bool)

get_target_zone(zone_name)[source]

Resolve zone_name to the concrete zone that should receive a target.

Parameters:: zone_name (str | list | None)
Return type:: Zone

lock_dt_cont_multiplier()[source]: Lock the dt_cont_multiplier to prevent further changes.

class OpenPinch.classes.stream.Stream(name='Stream', t_supply=None, t_target=None, p_supply=None, p_target=None, h_supply=None, h_target=None, dt_cont=0.0, dt_cont_multiplier=1.0, heat_flow=0.0, htc=1.0, price=0.0, is_process_stream=True)[source]

Bases: object

Generic thermal stream used for both process and utility duties.

A Stream stores supply/target states together with derived values such as hot/cold classification, shifted temperature bounds, heat-capacity flow rate, and simple economic attributes. The same class is reused for process streams, utilities, and derived net streams created during site- level aggregation.

Initialise a stream and infer hot/cold classification.

Parameters:

name (str)
t_supply (Optional[MaybeVU])
t_target (Optional[MaybeVU])
p_supply (Optional[MaybeVU])
p_target (Optional[MaybeVU])
h_supply (Optional[MaybeVU])
h_target (Optional[MaybeVU])
dt_cont (MaybeVU)
dt_cont_multiplier (float)
heat_flow (MaybeVU)
htc (MaybeVU)
price (MaybeVU)
is_process_stream (bool)

property name: str: Stream name.

property is_process_stream: bool: Process or utility stream.

property type: str | None: Stream type (Hot, Cold, Both).

property num_states: int | None: Number of states.

property t_supply: Value | None: Supply temperature (e.g., degC).

property t_target: Value | None: Target temperature (e.g., degC).

property p_supply: Value | None: Supply pressure (e.g., kPa).

property p_target: Value | None: Target pressure (e.g., kPa).

property h_supply: Value | None: Supply enthalpy (e.g., kJ/kg).

property h_target: Value | None: Target enthalpy (e.g., kJ/kg).

property dt_cont: Value: Preserved base delta-T contribution before any zone multiplier.

property dt_cont_act: Value: Effective delta-T contribution used in shifted-temperature calculations.

property dt_cont_multiplier: float: Effective delta-T contribution used in shifted-temperature calculations.

property dt_cont_multiplier_locked: bool: Whether the delta-T contribution multiplier is locked against changes.

property heat_flow: Value: Stream heat flow view over a scalar or stateful duty value.

property htc: float: Heat transfer coefficient (e.g., kW/m^2/K).

property htr: Value | None: Heat transfer resistance (e.g., m^2.K/kW).

property price: Value: Unit energy price (e.g., $/MWh).

property ut_cost: Value | None: Utility cost (e.g., $/y).

property CP: Value | None: Heat capacity flowrate (e.g., kW/K).

property rCP: Value | None: Resistance-capacity product (1/heat transfer rate).

property active: bool: Whether the stream is active in analysis.

property t_min: Value | None: Minimum temperature (supply or target depending on hot/cold).

property t_max: Value | None: Maximum temperature (supply or target depending on hot/cold).

property t_min_star: Value | None: Shifted minimum temperature.

property t_max_star: Value | None: Shifted maximum temperature.

property stream_type: str | None: Alias for the stream thermal type.

property is_active: bool: Alias for whether the stream participates in analysis.

property supply_temperature: Value | None: Alias for the supply temperature.

property target_temperature: Value | None: Alias for the target temperature.

property minimum_temperature: Value | None: Alias for the minimum stream temperature.

property maximum_temperature: Value | None: Alias for the maximum stream temperature.

property shifted_minimum_temperature: Value | None: Alias for the shifted minimum stream temperature.

property shifted_maximum_temperature: Value | None: Alias for the shifted maximum stream temperature.

property supply_pressure: Value | None: Alias for the supply pressure.

property target_pressure: Value | None: Alias for the target pressure.

property supply_enthalpy: Value | None: Alias for the supply enthalpy.

property target_enthalpy: Value | None: Alias for the target enthalpy.

property delta_t_contribution: Value: Alias for the base shifted-temperature contribution.

property delta_t_contribution_multiplier: float: Alias for the shifted-temperature contribution multiplier.

property effective_delta_t_contribution: Value: Alias for the effective shifted-temperature contribution.

property heat_transfer_coefficient: Value: Alias for the heat transfer coefficient.

property heat_transfer_resistance: Value | None: Alias for the heat-transfer resistance.

property utility_cost: Value | None: Alias for the derived utility cost.

property heat_capacity_flow_rate: Value | None: Alias for the derived stream heat-capacity flow rate.

property resistance_capacity_product: Value | None: Alias for the stream resistance-capacity product.

invert()[source]

Flip a utility stream into its generating process-stream analogue.

Return type:: None

class OpenPinch.classes.stream_collection.StreamCollection(streams=None)[source]

Bases: object

A dynamic, ordered collection of streams.

Key features include:

Add and remove streams by name.
Prevent overwriting existing streams by auto-renaming.
Configure sort keys as attributes or callables.
Iterate efficiently with lazy sorting.
Support ascending or descending ordering.

Initialise an empty collection sorted by descending supply temperature.

Parameters:: streams (List[Stream] | None)

property state_ids: dict[str, int] | None: Return the canonical state identifiers for this collection.

property weights: ndarray | None: Return the canonical state weights for this collection.

property num_states: int | None: Return the number of states for this collection.

add(stream, key=None, prevent_overwrite=True)[source]

Insert a stream, optionally renaming the key to avoid collisions.

Parameters:

stream (Stream)
key (str)
prevent_overwrite (bool)

Return type:

str

add_many(streams, keys=None, prevent_overwrite=True)[source]

Insert several streams, optionally using explicit keys for each stream.

Parameters:

streams (List[Stream])
prevent_overwrite (bool)

remove(stream_name)[source]

Remove a stream by name.

Parameters:: stream_name (str)

sum_stream_attribute(attr_name, idx=None)[source]

Return the total of a specified attribute for streams in the collection.

Parameters:

attr_name (str)
idx (int | None)

Return type:

float

set_common_stream_attribute(attr_name, value, *, idx=None)[source]

Set a common attribute across all streams in the collection.

Parameters:

attr_name (str)
value (Any)
idx (int | None)

set_sort_key(key, reverse=False)[source]

Set the sorting key. Supports attribute names or custom lambdas.

Parameters:

key (str | List[str] | Callable)
reverse (bool)

copy(*, deep=False)[source]

Return a copy of the collection, optionally deep-copying streams.

Parameters:: deep (bool)
Return type:: StreamCollection

set_state_context(state_ids, weights, num_states=None)[source]

Persist the canonical shared state model for this collection.

Parameters:

state_ids (dict[str, int] | list[str] | tuple[str, ...] | None)
weights (ndarray | list[float] | tuple[float, ...] | None)
num_states (int | None)

Return type:

None

get_index(stream)[source]

Return the position (index) of a stream object in the sorted stream list.

Return type:: int

items()[source]: Return the underlying keyed stream items in insertion order.

export_to_csv(filename='heat pump streams')[source]

Export stream data to results/<filename> and return the path written.

Parameters:: filename (str)
Return type:: Path

get_hot_streams(include_process_streams=True, include_utility_streams=True, invert_utility=False, sort_attr=None)[source]

Return a new collection containing only hot streams.

Parameters:

include_process_streams (bool)
include_utility_streams (bool)
invert_utility (bool)
sort_attr (str | None)

get_cold_streams(include_process_streams=True, include_utility_streams=True, invert_utility=False, sort_attr=None)[source]

Return a new collection containing only cold streams.

Parameters:

include_process_streams (bool)
include_utility_streams (bool)
invert_utility (bool)
sort_attr (str | None)

get_process_streams(sort_attr=None)[source]

Return a new collection containing only process streams.

Parameters:: sort_attr (str | None)

get_hot_process_streams(sort_attr=None)[source]

Return a new collection containing only hot process streams.

Parameters:: sort_attr (str | None)

get_cold_process_streams(sort_attr=None)[source]

Return a new collection containing only cold process streams.

Parameters:: sort_attr (str | None)

get_utility_streams(sort_attr=None)[source]

Return a new collection containing only utility streams.

Parameters:: sort_attr (str | None)

get_hot_utility_streams(sort_attr=None)[source]

Return a new collection containing only hot utility streams.

Parameters:: sort_attr (str | None)

get_cold_utility_streams(sort_attr=None)[source]

Return a new collection containing only cold utility streams.

Parameters:: sort_attr (str | None)

get_inverted_hot_utility_streams(sort_attr=None)[source]

Return a new collection containing only inverted hot utility streams.

Parameters:: sort_attr (str | None)

get_inverted_cold_utility_streams(sort_attr=None)[source]

Return a new collection containing only inverted cold utility streams.

Parameters:: sort_attr (str | None)

class OpenPinch.classes.problem_table.ProblemTable(data_input=None, add_default_labels=True)[source]

Bases: object

NumPy-backed pinch problem table with enum-friendly accessors.

Initialise the table from a dictionary or list-of-columns structure.

Parameters:

data_input (dict[str | ProblemTableLabel, object] | list | None)
add_default_labels (bool)

class ColumnViewByIndex(parent)[source]

Bases: object

Expose read/write access to columns addressed by integer index.

Parameters:: parent (ProblemTable)

property icol: Return a view for column access by integer position.

class ColumnViewByName(parent)[source]

Bases: object

Expose read/write access to columns addressed by label or enum.

Parameters:: parent (ProblemTable)

property col: Return a view for column access by string label or ProblemTableLabel.

class ColumnsViewByName(parent)[source]

Bases: object

Vectorised view over multiple labelled columns or enums.

Parameters:: parent (ProblemTable)

property cols: Return a vectorised view over multiple labelled columns or enums.

class LocationByRowByColName(parent)[source]

Bases: object

Row/column accessor mirroring DataFrame.loc semantics.

Parameters:: parent (ProblemTable)

property loc: Expose row/column access using label semantics (loc).

class LocationByRowByCol(parent)[source]

Bases: object

Row/column accessor mirroring DataFrame.iloc semantics.

Parameters:: parent (ProblemTable)

property iloc: Expose row/column access using positional semantics (iloc).

slice(keys)[source]

Return a new ProblemTable containing only the requested columns.

Parameters:: keys (str | ProblemTableLabel | Sequence[str | ProblemTableLabel])
Return type:: ProblemTable

property shape: Tuple describing (rows, columns) for the buffer.

property to_dataframe: DataFrame: Convert the buffer into a pandas DataFrame.

property copy: Return a deep copy of the table.

to_list(col=None)[source]

Return table data as Python lists; optionally restrict to a single column.

Parameters:: col (str | ProblemTableLabel | None)

round(decimals)[source]: Round the underlying NumPy buffer in-place.

pinch_idx(col=ProblemTableLabel.H_NET)[source]

Return the row indices of the hot and cold pinch temperatures.

Parameters:: col (int | str | ProblemTableLabel)
Return type:: Tuple[int, int, bool]

pinch_temperatures(col_T=ProblemTableLabel.T, col_H=ProblemTableLabel.H_NET)[source]

Determine the hottest hot and coldest cold pinch temperatures.

Parameters:

col_T (str | ProblemTableLabel)
col_H (int | str | ProblemTableLabel)

Return type:

Tuple[float | None, float | None]

shift_heat_cascade(dh, col)[source]

Shift a heat-cascade column by dh and return a table copy.

Parameters:

dh (float)
col (int | str | ProblemTableLabel)

Return type:

ProblemTable

share_temperature_intervals(other)[source]

Mutate both tables so they use the union of their temperature intervals.

Returns a tuple containing (rows_inserted_into_self, rows_inserted_into_other).

Parameters:: other (ProblemTable)
Return type:: Tuple[int, int]

insert_temperature_interval(T_ls)[source]

Insert any missing temperature intervals and return count inserted.

Parameters:: T_ls (List[float] | float)
Return type:: int

insert(row_dict, index)[source]

Insert a single row (dict of column: value) at the specified index.

Parameters:

row_dict (dict)
index (int)

update_row(index, row_dict)[source]

Update selected columns for one row using values from row_dict.

Parameters:

index (int)
row_dict (dict)

update(updates=None, T_col=None)[source]

Assign aligned column values in-place using an explicit source T column.

Parameters:

updates (dict[str | ProblemTableLabel, ndarray] | None)
T_col (ndarray | None)

Return type:

ProblemTable

delete_row(index)[source]

Remove a row at index from the buffer.

Parameters:: index (int)

sort_by_column(column, ascending=True)[source]

Sort rows in-place by the given column.

Parameters:

column (str | ProblemTableLabel)
ascending (bool)

export(filename='problem_table', sheet_name='ProblemTable', include_index=False)[source]

Export the table to results/<filename>.xlsx and return the path.

Parameters:

filename (str)
sheet_name (str)
include_index (bool)

Return type:

Path

Solved Target Records

Targets are stored on zones and normalized through schema models before export. The base target schema is a useful reference when you are programmatically comparing cases or consuming target results in another tool.

class OpenPinch.lib.schemas.targets.BaseTargetModel(*, zone_name=None, state_id=None, state_idx=None, name, type, parent_zone=None, config=<factory>, active=True)[source]

Bases: BaseModel

Shared metadata for all solved target objects.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

zone_name (str | None)
state_id (str | None)
state_idx (int | None)
name (str)
type (str)
parent_zone (Any)
config (Configuration)
active (bool)

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

to_target_results(isTotal=False)[source]

Convert the runtime target into the exported reporting schema.

Parameters:: isTotal (bool)
Return type:: TargetResults

serialize_json(isTotal=False)[source]

Serialise the reporting-schema view of this target to plain Python.

Parameters:: isTotal (bool)
Return type:: dict[str, Any]

How These Objects Relate

The usual flow is:

input schemas describe the external inputs
preparation turns those inputs into Zone and Stream objects
targeting populates ProblemTable objects, zone targets, and graph data
result schemas serialize the solved state back out

That layering is what lets the package support both high-level scripted use and deeper programmatic inspection.