Core API

The core API is the supported front door to OpenPinch. It is intentionally small: one high-level service function, two convenience wrapper classes, and a small number of lower-level orchestration helpers for callers that need to operate on prepared zone trees directly.

Service-Layer Example

from OpenPinch import pinch_analysis_service
from OpenPinch.lib.schemas.io import StreamSchema, TargetInput

input_data = TargetInput(
    streams=[
        StreamSchema(
            zone="Process",
            name="Hot Feed",
            t_supply=180.0,
            t_target=80.0,
            heat_flow=2500.0,
            dt_cont=10.0,
        )
    ],
    utilities=[],
)

result = pinch_analysis_service(input_data, project_name="Example")

Top-Level Package Re-Exports

The OpenPinch package re-exports the small subset of classes and helpers that are intended to be imported most often. This makes it practical to work from from OpenPinch import ... in notebooks and scripts without traversing the full package layout.

In particular, the package root exposes:

Package Entrypoints

OpenPinch public API.

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

class OpenPinch.classes.pinch_problem.PinchProblem(source=None, *, project_name='Site')[source]

Bases: object

Typed orchestrator for loading input data and running targeting.

Parameters:

  • source (TargetInput | JsonDict | PathLike | tuple[PathLike, PathLike] | None)

  • project_name (Optional[str])

load(source=None)[source]

Load problem inputs from JSON, Excel, CSV, or an in-memory object.

Parameters:

source (TargetInput | Dict[str, Any] | str | Path | tuple[str | Path, str | Path] | None)

Return type:

Zone | None

property period_ids: dict[str, int]

Return the canonical period_id -> idx lookup for the loaded problem.

target_all_periods(*, parallel=False, max_workers=None, preserve_cached_results=True)[source]

Run default targeting once per canonical period id.

Parameters:

  • parallel (bool | str) – False runs serially. True and "process" use a process pool, while "thread" uses a thread pool which is suitable for no-GIL Python builds.

  • max_workers (int | None) – Optional executor worker limit for parallel runs.

  • preserve_cached_results (bool) – Restore the original results cache after the batch run when True.

Return type:

dict[str, TargetOutput]

validate()[source]

Validate the currently loaded problem data without running targeting.

Return type:

TargetInput

validation_report()[source]

Return structured validation results without raising for bad inputs.

Return type:

ValidationReport

summary_frame(*, detailed=False, format=None, periods='selected')[source]

Return the solved target summary as a pandas DataFrame.

Parameters:

  • detailed (bool)

  • format (str | None)

  • periods (str)

Return type:

DataFrame

metrics(*, solve=True, periods='selected')[source]

Return typed summary metrics for the current solved result.

Parameters:
Return type:

list[ReportMetric]

report(*, solve=True, periods='selected')[source]

Return a typed report without writing any files.

Parameters:
Return type:

ProblemReport

export_excel(results_dir=None, *, periods='selected')[source]

Export the solved target summary and problem tables to an Excel file.

Parameters:
Return type:

Path

compare_to(other_problem, *, target_name=None, base_label='Base case', other_label='Scenario')[source]

Compare numeric summary metrics of two solved problems.

Parameters:
Return type:

DataFrame

property problem_filepath: Path | None

Return the filepath of the problem that was loaded or supplied.

property problem_data: Dict[str, Any] | TargetInput | None

Return the raw problem definition that was loaded or supplied.

property results: TargetOutput | None

Return the cached targeting results, if targeting has been executed.

property master_zone: Zone | None

Return the prepared root zone after a successful load() pass.

property process_components: dict[str, Any]

Memory-only process components applied to the prepared model.

property hot_streams: StreamCollection

Hot process streams on the root analysis zone.

property cold_streams: StreamCollection

Cold process streams on the root analysis zone.

property hot_utilities: StreamCollection

Hot utility streams on the root analysis zone.

property cold_utilities: StreamCollection

Cold utility streams on the root analysis zone.

property project_name: str

Return the project label used for the root zone and exports.

classmethod from_json(data)[source]

Build from an in-memory mapping and apply the normal input cleaners.

Parameters:

data (Dict[str, Any])

Return type:

PinchProblem

to_problem_json(*, canonical=False)[source]

Return the currently loaded problem inputs.

Parameters:

canonical (bool)

Return type:

Dict[str, Any]

canonical_problem_json()[source]

Return canonical mutable problem inputs with an explicit zone tree.

Return type:

Dict[str, Any]

set_dt_cont_multiplier(value, *, zone_name=None)[source]

Update one zone-tree multiplier and rebuild the prepared analysis state.

Parameters:
Return type:

Zone

update_options(options, *, replace=False)[source]

Update the problem options in-place and rebuild the analysis state.

Parameters:
Return type:

Zone

show_dashboard(*, zone=None, graph_data=None, page_title='OpenPinch Dashboard', value_rounding=2)[source]

Launch the Streamlit dashboard for the analysed problem.

Parameters:

  • zone (Zone | None)

  • graph_data (Dict[str, Any] | None)

  • page_title (str | None)

  • value_rounding (int)

Return type:

None

class OpenPinch.classes.pinch_workspace.PinchWorkspace(source=None, *, project_name='Site', baseline_name='baseline')[source]

Bases: object

Manage multiple named PinchProblem cases with a script-native API.

Parameters:
classmethod load_bundle(path)[source]

Load a previously persisted workspace bundle.

Parameters:

path (str | Path)

Return type:

PinchWorkspace

load(source, *, case_name=None, activate=True, project_name=None)[source]

Load or replace a named case and return a live validated case.

Parameters:
Return type:

PinchProblem | None

list_variants()[source]

Return the case names in stable insertion order.

Return type:

list[str]

get_variant_input(name)[source]

Return a defensive copy of one stored case input.

Parameters:

name (str)

Return type:

Dict[str, Any]

input_view(name)[source]

Return a frontend-friendly editable case input view.

Parameters:

name (str)

Return type:

VariantInputView

validate_variant(name)[source]

Return a structured validation report for one case input.

Parameters:

name (str)

validation_report(case_name=None)[source]

Return a structured validation report for one case input.

Parameters:

case_name (str | None)

set_variant_input(name, case_input, *, base=None)[source]

Create or replace one stored case input.

Parameters:
Return type:

Dict[str, Any]

solve_variant(name, *, workflow='target', workflow_options=None)[source]

Solve one case and return a serializable frontend-facing view.

Parameters:
Return type:

ScenarioVariantView

compare_variants(variant_names=None, *, base=None)[source]

Return a deterministic comparison view across solved variants.

Parameters:
list_cases()[source]

Return the loaded case names in stable insertion order.

Return type:

list[str]

case(name=None)[source]

Return the live PinchProblem for one named case.

Parameters:

name (str | None)

Return type:

PinchProblem

use_case(name)[source]

Activate one named case and return it.

Parameters:

name (str)

Return type:

PinchProblem

copy_case(*, source_name='baseline', new_name='new', activate=False)[source]

Clone one existing case into a new named case.

Parameters:

  • source_name (str)

  • new_name (str)

  • activate (bool)

Return type:

PinchProblem

scenario(name, *, base=None, options=None, replace_options=False, dt_cont_multiplier=None, activate=False, solve=False, workflow='target', workflow_options=None)[source]

Create a named scenario from a base case and optional edits.

Parameters:
Return type:

PinchProblem

get_case_input(name=None, *, canonical=True)[source]

Return one case input, optionally normalised to canonical form.

Parameters:

  • name (str | None)

  • canonical (bool)

Return type:

Dict[str, Any]

to_problem_json(*, case_name=None, canonical=True)[source]

Return the case input for one case using PinchProblem naming.

Parameters:

  • case_name (str | None)

  • canonical (bool)

Return type:

Dict[str, Any]

property active_case_name: str | None

Return the currently active case name.

property target

Delegate the target accessor to the active case.

property plot

Delegate the plot accessor to the active case.

property problem_data

Return the active case input.

property problem_filepath

Return the active case filepath when available.

property results

Return the active case results when available.

property master_zone

Return the active case master zone when available.

validate(case_name=None)[source]

Validate one case input.

Parameters:

case_name (str | None)

summary_frame(*, case_name=None, detailed=False, format=None, periods='selected')[source]

Return the solved summary for one case.

Parameters:

  • case_name (str | None)

  • detailed (bool)

  • format (str | None)

  • periods (str)

Return type:

DataFrame

metrics(*, case_name=None, solve=True, periods='selected')[source]

Return typed metrics for one case.

Parameters:

  • case_name (str | None)

  • solve (bool)

  • periods (str)

report(*, case_name=None, solve=True, periods='selected')[source]

Return a typed report for one case.

Parameters:

  • case_name (str | None)

  • solve (bool)

  • periods (str)

export_excel(results_dir=None, *, case_name=None, periods='selected')[source]

Export one case to an Excel workbook.

Parameters:

  • results_dir (str | Path | None)

  • case_name (str | None)

  • periods (str)

Return type:

Path

set_dt_cont_multiplier(value, *, zone_name=None, case_name=None)[source]

Update one case multiplier and keep the stored case input in sync.

Parameters:

  • value (float)

  • zone_name (str | None)

  • case_name (str | None)

update_options(options, *, case_name=None, replace=False)[source]

Update one case’s options and keep the stored case input in sync.

Parameters:
Return type:

PinchProblem

show_dashboard(*, case_name=None, zone=None, graph_data=None, page_title='OpenPinch Dashboard', value_rounding=2)[source]

Launch the dashboard for one case.

Parameters:

  • case_name (str | None)

  • graph_data (dict[str, Any] | None)

  • page_title (str | None)

  • value_rounding (int)

Return type:

None

compare_to(other_problem, *, case_name=None, other_case_name=None, target_name=None, base_label='Base case', other_label='Scenario')[source]

Compare one workspace case to another problem or workspace case.

Parameters:

  • other_problem (PinchProblem | 'PinchWorkspace')

  • case_name (Optional[str])

  • other_case_name (Optional[str])

  • target_name (Optional[str])

  • base_label (str)

  • other_label (str)

Return type:

pd.DataFrame

compare_cases(base_case, other_case, *, target_name=None, base_label=None, other_label=None)[source]

Compare two cases in the same workspace.

Parameters:

  • base_case (str)

  • other_case (str)

  • target_name (str | None)

  • base_label (str | None)

  • other_label (str | None)

Return type:

DataFrame

save_bundle(path)[source]

Persist the current workspace, syncing any live case edits first.

Parameters:

path (str | Path)

Return type:

Path

classmethod configuration_field_metadata()[source]

Return declarative metadata for editable configuration fields.

Return type:

list[ConfigurationFieldMetadata]

Core Service Functions

OpenPinch.main is the thin orchestration layer above the analysis modules.

  • pinch_analysis_service() validates the incoming input data, prepares the zone hierarchy, runs the appropriate direct and indirect targeting steps, and returns a structured response.

Private dispatch and result-extraction helpers support the public problem and service entry points but are not documented as stable user interfaces.

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.

PinchProblem Convenience Wrapper

PinchProblem adds file loading, cached execution state, tabular summaries, graph generation, Excel export, and Streamlit dashboard integration on top of the core service layer. It also owns the add_component accessor used for memory-only process mutations such as direct gas/vapour MVR before targets are rerun.

Those rendered graph, Excel, and dashboard surfaces require the openpinch[notebook] or openpinch[dashboard] extra.

Use it when you want:

  • a single object that owns the problem definition and solved result

  • support for JSON, workbook, and CSV-bundle inputs

  • simple summary, graph, export, and dashboard hooks without manually wiring the lower-level functions

The main user-facing methods are:

The wrapper is intentionally light. Once targeting has run, the same solved Zone hierarchy and TargetOutput objects remain available for direct inspection.

The problem.target.* accessor is the explicit advanced post-processing entrypoint family. Each named workflow returns the affected BaseTargetModel and refreshes cached TargetOutput results on the same PinchProblem instance. Heat pump and refrigeration targets also surface HPR summary fields such as hpr_cycle, hpr_utility_total, hpr_work, hpr_external_utility, and StreamCollection objects on hpr_hot_streams and hpr_cold_streams. Simulated-cycle targets also expose annualized cost fields: hpr_operating_cost, hpr_capital_cost, hpr_annualized_capital_cost, hpr_total_annualized_cost, hpr_compressor_capital_cost, and hpr_heat_exchanger_capital_cost. For multiperiod input data, the same named entry points also accept period_id=... and the refreshed summaries, exports, and graph metadata carry that selected period context forward. Weighted HPR summaries average operating fields, retain maximum capital fields, and recompute total annualized cost from weighted operating cost plus maximum annualized capital. Per-period summary replay uses fresh copies of one baseline zone and restores the original problem state on both success and failure.

The problem.add_component.* accessor is different: it mutates the prepared stream model before targeting. The direct process MVR component deactivates selected source streams, activates generated compressed-vapour replacement streams, stores per-period stage results, and contributes process-component work to later target summaries.