Service Layer

The service layer is the boundary between validated input data and the prepared/solved in-memory model. It is the right integration surface when you want more control than PinchProblem provides but do not want to invoke individual low-level algorithms directly.

Layering

The service stack is designed in three steps:

  1. validate or receive typed request data

  2. prepare the inputs into a Zone hierarchy

  3. dispatch direct, indirect, HPR, exergy, cogeneration, or area/cost targeting

Use Cases

Use the service layer when you need to:

  • embed OpenPinch in another application with a stable request/response boundary

  • prepare a zone hierarchy once and run multiple advanced studies against it

  • inspect the prepared model before solving

  • mutate a live prepared model with process components before rerunning targets

  • bypass file handling entirely and work with typed inputs

  • apply exergy or cogeneration as post-processing on already solved targets

Main Service Surface

Public service-layer entry points and reusable targeting helpers.

OpenPinch.services.data_preprocessing_service(input_data, project_name='Site')[source]

Validate raw input data and construct the in-memory zone tree.

Parameters:

  • input_data (Any)

  • project_name (str)

Return type:

Zone

OpenPinch.services.direct_heat_integration_service(zone, args=None)[source]

Run direct heat integration targeting for a prepared zone.

Parameters:
Return type:

Zone

OpenPinch.services.exergy_targeting_service(zone, args=None)[source]

Run exergy enrichment on one compatible target family.

Parameters:
Return type:

Zone

OpenPinch.services.indirect_heat_integration_service(zone, args=None)[source]

Run indirect heat integration targeting for an aggregated zone.

Parameters:
Return type:

Zone

OpenPinch.services.direct_heat_pump_service(zone, args=None)[source]

Run direct heat pump targeting for a prepared zone.

Parameters:
Return type:

Zone

OpenPinch.services.indirect_heat_pump_service(zone, args=None)[source]

Run indirect heat pump targeting for an aggregated zone.

Parameters:
Return type:

Zone

OpenPinch.services.direct_refrigeration_service(zone, args=None)[source]

Run direct refrigeration targeting for a prepared zone.

Parameters:
Return type:

Zone

OpenPinch.services.indirect_refrigeration_service(zone, args=None)[source]

Run indirect refrigeration targeting for an aggregated zone.

Parameters:
Return type:

Zone

OpenPinch.services.power_cogeneration_service(zone, args=None)[source]

Run turbine cogeneration targeting for a prepared zone.

Parameters:
Return type:

Zone

OpenPinch.services.area_cost_targeting_service(zone, args=None)[source]

Recompute direct integration targets with area/cost targeting enabled.

Parameters:
Return type:

Zone

OpenPinch.services.energy_transfer_analysis_service(zone, args=None)[source]

Create energy-transfer diagram and surplus/deficit table outputs.

Parameters:
Return type:

Zone

OpenPinch.services.heat_exchanger_network_controllability_service(network, **kwargs)[source]

Quantify steady-state controllability for a heat exchanger network.

Parameters:
Return type:

Any

OpenPinch.services.get_area_targets(T_vals, H_hot_bal, H_cold_bal, R_hot_bal, R_cold_bal)[source]

Estimate heat-transfer area targets with vectorised counter-current logic.

Parameters:
Return type:

dict

OpenPinch.services.get_capital_cost_targets(area, num_units, config)[source]

Estimate equipment and annualized capital costs from area/unit targets.

Parameters:

  • area (float) – Total heat-transfer area target from balanced composite curves.

  • num_units (int) – Minimum exchanger count estimate for the same targeting scenario.

  • config (Configuration) – Active configuration containing fixed/variable cost coefficients, capital exponent, discount rate, and service life assumptions.

Returns:

(capital_cost, annual_capital_cost).

Return type:

tuple[float, float]

OpenPinch.services.get_output_graph_data(zone, graph_sets=None)[source]

Returns Json data points for each process.

Parameters:
Return type:

dict

OpenPinch.services.get_utility_targets(pt, pt_real=None, hot_utilities=None, cold_utilities=None, is_direct_integration=True, idx=None)[source]

Target utility usage and compute GCC variants for a zone.

Parameters:

  • pt (ProblemTable) – Shifted and real problem tables used for constructing composite curves.

  • pt_real (ProblemTable) – Shifted and real problem tables used for constructing composite curves.

  • hot_utilities (StreamCollection) – Candidate utility collections that will be targeted across temperature intervals.

  • cold_utilities (StreamCollection) – Candidate utility collections that will be targeted across temperature intervals.

  • is_direct_integration (bool) – When True (default) the function assumes the zone represents a process area and applies additional targeting logic appropriate for that context.

  • idx (int | None)

Returns:

Updated (pt, pt_real, hot_utilities, cold_utilities) collections with derived profiles embedded.

Return type:

tuple

Preparation Entry Point

The preparation stage is the key boundary between external inputs and the internal model. It validates configuration choices, builds the zone tree, applies dt_cont multipliers, instantiates process and utility streams, and produces the Zone object consumed by the solver stack.

Period-valued inputs remain period-aware after preparation, but period selection does not happen inside prepare_problem(...). Instead, the selected period is applied later through the targeting-service args dictionaries or the higher level problem.target.*(..., period_id=...) wrappers.

OpenPinch.services.input_data_processing.data_preparation.prepare_problem(streams=None, utilities=None, options=None, project_name='Site', zone_tree=None)[source]

Build the top-level zone hierarchy for analysis.

Parameters:
Return type:

Zone

Heat Exchanger Network Synthesis Entry

Heat exchanger network synthesis is problem-rooted. User code should normally enter through PinchProblem.design:

from OpenPinch.lib import HENDesignMethod

problem.design.enhanced_synthesis_method(quality_tier=2)
problem.design.open_hens_method()
problem.design.heat_exchanger_network_synthesis()
problem.design.heat_exchanger_network_synthesis(
    method=HENDesignMethod.NetworkEvolution,
    initial_networks=(existing_network,),
)

The internal service entry point owns method dispatch and final result caching. It dispatches to the same direct services exposed by the design accessor. Use enhanced_synthesis_method(quality_tier=...) as the public quality-tier selector, open_hens_method() for original tier 1 OpenHENS, and heat_exchanger_network_synthesis() for the generic fast tier 0 default or explicit enum dispatch.

Service entry and dispatch for heat exchanger network synthesis.

OpenPinch.services.heat_exchanger_network_synthesis.heat_exchanger_network_synthesis_entry.heat_exchanger_network_evolution_method_service(problem, *, initial_networks=None, options=None, workspace_variant=None, executor=None)[source]

Run only seeded network evolution and update the problem cache.

Parameters:
Return type:

HeatExchangerNetworkSynthesisResult

OpenPinch.services.heat_exchanger_network_synthesis.heat_exchanger_network_synthesis_entry.heat_exchanger_network_open_hens_method_service(problem, *, options=None, workspace_variant=None, executor=None)[source]

Run the original tier-1 OpenHENS PDM -> TDM -> EVM sequence.

Parameters:
Return type:

HeatExchangerNetworkSynthesisResult

OpenPinch.services.heat_exchanger_network_synthesis.heat_exchanger_network_synthesis_entry.heat_exchanger_network_pinch_design_method_service(problem, *, options=None, workspace_variant=None, executor=None)[source]

Run only the pinch design method and update the problem cache.

Parameters:
Return type:

HeatExchangerNetworkSynthesisResult

OpenPinch.services.heat_exchanger_network_synthesis.heat_exchanger_network_synthesis_entry.heat_exchanger_network_synthesis_service(problem, *, method=None, initial_networks=None, options=None, workspace_variant=None, executor=None)[source]

Dispatch one HEN design method and update the problem cache.

Parameters:
Return type:

HeatExchangerNetworkSynthesisResult

OpenPinch.services.heat_exchanger_network_synthesis.heat_exchanger_network_synthesis_entry.heat_exchanger_network_thermal_derivative_method_service(problem, *, initial_networks=None, options=None, workspace_variant=None, executor=None)[source]

Run only seeded TDM and update the problem cache.

Parameters:
Return type:

HeatExchangerNetworkSynthesisResult

The HEN synthesis package is intentionally method-oriented:

  • targeting_services contains method-specific orchestration.

  • common.execution contains settings, task builders, executor contracts, and fallback policy.

  • common.results contains result assembly and seed lookup.

  • common.reporting contains ranking, verification, and export helpers.

  • common.solver contains optional-dependency checks, array adapters, backend calls, and network extraction.

  • unit_models contains the equation/unit model layer for pinch design and stagewise models.

Network Grid Diagrams

The selected heat exchanger network can construct its own grid diagram through OpenPinch.classes.heat_exchanger_network.HeatExchangerNetwork.build_grid_diagram():

design = problem.results.design
period_id = design.network.period_ids[0]
diagram = design.network.build_grid_diagram(period_id=period_id)

The standalone service remains available for batch rendering one or more HeatExchangerNetwork objects, for example when displaying several ranked candidates.

Heat exchanger network grid diagram service.

class OpenPinch.services.network_grid_diagram.GridDiagramMatch(exchanger, state, source_stream, sink_stream, stage, duty)[source]

Bases: object

One active exchanger match placed in the process-stream grid.

Parameters:
class OpenPinch.services.network_grid_diagram.HeatExchangerNetworkGridDiagram(fig, ax, network, grid_model)[source]

Bases: object

Rendered heat exchanger network grid diagram.

Parameters:
show()[source]

Display the Plotly figure.

Return type:

None

save(path='grid_diagram.png')[source]

Save the Plotly figure to path.

Parameters:

path (str | Path)

Return type:

None

class OpenPinch.services.network_grid_diagram.HeatExchangerNetworkGridModel(network, period_id, hot_streams, cold_streams, stages, recovery_matches, hot_utility_matches, cold_utility_matches, branch_counts)[source]

Bases: object

Normalized topology for a heat exchanger network grid diagram.

Parameters:
OpenPinch.services.network_grid_diagram.build_grid_diagram(networks, *, index=None, period_id=None, stream_line_width=5.0, temperature_scaled=False)[source]

Return OpenHENS-style grid diagrams for one or more networks.

Parameters:
Return type:

HeatExchangerNetworkGridDiagram | tuple[HeatExchangerNetworkGridDiagram, …]

OpenPinch.services.network_grid_diagram.build_grid_model(network, *, period_id=None)[source]

Normalize an OpenPinch network into the OpenHENS grid topology.

Parameters:
Return type:

HeatExchangerNetworkGridModel

Network Controllability

Solved heat exchanger networks can also be screened for steady-state controllability. The service treats process-stream outlet temperatures as controlled outputs and practical bypass or utility-flow adjustments as manipulated variables, then scores the resulting duty-normalised interaction matrix.

For multiperiod networks, pass period_id to both diagram and controllability services. OpenPinch does not silently select period zero.

design = problem.results.design
period_id = design.network.period_ids[0]
assessment = design.network.quantify_controllability(period_id=period_id)
assessment.score
assessment.components.rank

The score is a screening metric rather than a dynamic closed-loop simulation. It is intended for comparing candidate HEN topologies and identifying networks with weak actuator coverage, poor pairing, low thermal margin, or insufficient control redundancy.

Heat exchanger network controllability service.

class OpenPinch.services.heat_exchanger_network_controllability.HeatExchangerNetworkControllabilityActuator(*, actuator_id, exchanger_id=None, kind, source_stream, sink_stream, stage=None, manipulated_variable, duty)[source]

Bases: BaseModel

One manipulated variable available to control the HEN.

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:

  • actuator_id (str)

  • exchanger_id (str | None)

  • kind (HeatExchangerKind)

  • source_stream (str)

  • sink_stream (str)

  • stage (int | None)

  • manipulated_variable (Literal['recovery_bypass_fraction', 'hot_utility_flow', 'cold_utility_flow'])

  • duty (float)

model_config = {'extra': 'forbid', 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class OpenPinch.services.heat_exchanger_network_controllability.HeatExchangerNetworkControllabilityComponents(*, rank, pairing, authority, conditioning, redundancy, thermal_margin=None)[source]

Bases: BaseModel

Component scores contributing to the composite controllability score.

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:
model_config = {'extra': 'forbid', 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class OpenPinch.services.heat_exchanger_network_controllability.HeatExchangerNetworkControllabilityEndpoint(*, output_id, stream_id, side, exchanger_count, total_duty)[source]

Bases: BaseModel

One process-stream outlet temperature treated as a controlled output.

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:

  • output_id (str)

  • stream_id (str)

  • side (Literal['source', 'sink'])

  • exchanger_count (int)

  • total_duty (float)

model_config = {'extra': 'forbid', 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class OpenPinch.services.heat_exchanger_network_controllability.HeatExchangerNetworkControllabilityPairing(*, output_id, actuator_id, interaction)[source]

Bases: BaseModel

Best steady-state output/actuator pairing entry.

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:

  • output_id (str)

  • actuator_id (str)

  • interaction (float)

model_config = {'extra': 'forbid', 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class OpenPinch.services.heat_exchanger_network_controllability.HeatExchangerNetworkControllabilityResult(*, score, rating, components, outputs=<factory>, actuators=<factory>, interaction_matrix=<factory>, pairings=<factory>, matrix_rank=0, condition_number=None, singular_values=<factory>, minimum_approach_temperature=None, diagnostics=<factory>)[source]

Bases: BaseModel

Quantified controllability assessment for a heat exchanger network.

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:

  • score (float)

  • rating (Literal['strong', 'moderate', 'weak', 'poor'])

  • components (HeatExchangerNetworkControllabilityComponents)

  • outputs (tuple[HeatExchangerNetworkControllabilityEndpoint, ...])

  • actuators (tuple[HeatExchangerNetworkControllabilityActuator, ...])

  • interaction_matrix (tuple[tuple[float, ...], ...])

  • pairings (tuple[HeatExchangerNetworkControllabilityPairing, ...])

  • matrix_rank (int)

  • condition_number (float | None)

  • singular_values (tuple[float, ...])

  • minimum_approach_temperature (float | None)

  • diagnostics (tuple[str, ...])

model_config = {'extra': 'forbid', 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

OpenPinch.services.heat_exchanger_network_controllability.quantify_heat_exchanger_network_controllability(network, *, period_id=None, active_only=True, include_utility_actuators=True, minimum_interaction=1e-09, minimum_approach_temperature=5.0, desired_redundancy=2, rank_tolerance=None, condition_warning_threshold=25.0)[source]

Return a 0-1 controllability assessment for a solved HEN.

The service builds a steady-state interaction matrix from available network data. Rows are process-stream outlet temperatures, columns are practical manipulated variables: recovery bypass fractions and utility flow rates. Entries are duty-normalised thermal authority values, which provide a deterministic controllability proxy when no dynamic HEN model is available.

Parameters:
Return type:

HeatExchangerNetworkControllabilityResult

Typical Preparation and Solve Pattern

from OpenPinch.lib.schemas.io import TargetInput
from OpenPinch.services import (
    data_preprocessing_service,
    direct_heat_integration_service,
    indirect_heat_integration_service,
)

source_data = {"streams": [...], "utilities": [...]}
input_data = TargetInput.model_validate(source_data)
zone = data_preprocessing_service(input_data, project_name="Example")

direct_heat_integration_service(zone, {"period_id": "peak"})
indirect_heat_integration_service(zone, {"period_id": "peak"})

Each targeting service mutates the prepared zone in place, records the requested period metadata on the zone, and adds or refreshes the corresponding target model.

The exergy service follows a slightly different contract from the base thermal targeting services: it enriches an already existing compatible target for the requested period instead of re-solving direct or indirect targeting internally.

Direct High-Level Orchestration

For callers that want one function rather than an object wrapper, the root orchestration helper remains available:

High-level orchestration for running an OpenPinch analysis.

The functions in this module wire together data validation, pinch targeting, and output formatting. They act as the main entry points used by the PinchProblem helper class as well as external callers embedding OpenPinch in larger workflows.

OpenPinch.main.pinch_analysis_service(data, project_name='Project')[source]

Validate input data, run targeting, and return TargetOutput.

Parameters:

  • data (Any) – Raw request data matching OpenPinch.lib.schemas.io.TargetInput. Dictionaries, Pydantic models, and dataclass-like objects are accepted.

  • project_name (str) – Optional label used in generated graphs and result files.

Returns:

Validated response data containing solved targets and graph data.

Return type:

TargetOutput

Choosing Between Interfaces

  • Use OpenPinch.main.pinch_analysis_service() when you want a typed request/response contract.

  • Use problem.add_component.process_mvr(...) when the study needs direct gas/vapour MVR stream replacement before ordinary target reruns.

  • Use OpenPinch.services when you want to prepare a zone once and run several advanced analyses against the same prepared state.

  • Use PinchProblem when you want a notebook- and file-oriented convenience wrapper with summaries and exports.

The service layer is also the best place to look when you are trying to understand how the narrative workflow maps onto the actual analysis pipeline.