A vapour-compression refrigeration cycle on CoolProp (R134a)¶

R134a vapour at 0.2 MPa and 270 K is compressed to 1 MPa (isentropic efficiency 0.75), condensed to saturated liquid, throttled back to 0.2 MPa and evaporated to saturated vapour. The properties are CoolProp's Helmholtz equations of state (coolprop-heos-v1), one fluid per revision, which need the optional coolprop extra.

In [1]:

import importlib.util

HAVE_COOLPROP = importlib.util.find_spec("CoolProp") is not None
if not HAVE_COOLPROP:
    print(
        "SKIPPED: this example needs the optional `coolprop` extra "
        "(pip install 'openflowsheet[coolprop]')."
    )

import importlib.util HAVE_COOLPROP = importlib.util.find_spec("CoolProp") is not None if not HAVE_COOLPROP: print( "SKIPPED: this example needs the optional `coolprop` extra " "(pip install 'openflowsheet[coolprop]')." )

Writing the revision document¶

A flowsheet is a ProcessRevision document: a component set, unit instances, material connections and the specifications that fix its degrees of freedom. The helpers below write its members as plain dictionaries — nothing here bypasses validation; the application checks the whole document when it is committed and again before it solves. Every value is in SI.

In [2]:

import json

from openflowsheet.application.local import LocalApplication
from openflowsheet.application.types import Change, Edit

#: (SI unit, dimension exponents in the schema's base order) by kind, for instance parameters.
KINDS = {
    "dimensionless": ("1", [0, 0, 0, 0, 0, 0, 0]),
    "pressure": ("Pa", [-1, 1, -2, 0, 0, 0, 0]),
    "molar_flow": ("mol/s", [0, 0, -1, 0, 1, 0, 0]),
}
#: The tolerance a fixed specification carries, by kind (ADR 0001's registered values).
TOLERANCE = {
    "molar_flow": ({"absolute": 1e-9, "relative": 1e-8, "scale": 3.0}, "mol/s"),
    "temperature": ({"absolute": 1e-6}, "K"),
    "pressure": ({"absolute": 0.01}, "Pa"),
    "heat_rate": ({"absolute": 1e-5, "relative": 1e-8, "scale": 1e5}, "W"),
}


def parameter(value, kind, meaning):
    unit, dimension = KINDS[kind]
    return {
        "value": value,
        "unit": unit,
        "dimension": dimension,
        "kind": kind,
        "meaning": meaning,
        "role": "fixed",
    }


def unit(unit_id, model, role, **parameters):
    return {
        "id": unit_id,
        "model": {"id": model, "version": "0.0.0-declared", "artifact_ref": None},
        "semantic_role": role,
        "parameters": parameters,
        "policy": {"fidelity": "example", "validity": "example"},
    }


def stream(stream_id, source, target, phase):
    return {
        "id": stream_id,
        "kind": "material",
        "from": {"instance": source[0], "port": source[1]},
        "to": {"instance": target[0], "port": target[1]},
        "state_definition": "nTP-v1",
        "component_mapping": "revision_component_set",
        "phase_capability": phase,
        "notes": stream_id,
    }


def fix(spec_id, object_type, object_id, path, value, kind, component=None):
    tolerance, si_unit = TOLERANCE[kind]
    target = {"object_type": object_type, "object_id": object_id, "path": path}
    if component is not None:
        target["component"] = component
    return {
        "id": spec_id,
        "target": target,
        "kind": kind,
        "unit": si_unit,
        "value": value,
        "tolerance": tolerance,
        "role": "fixed",
        "provenance": "example",
    }


def revision(title, record_source, components, instances, connections, specifications):
    return {
        "schema_version": 1,
        "revision_id": title,
        "parent_revision": None,
        "title": title,
        "description": title,
        "component_set": {"record_source": record_source, "components": list(components)},
        "instances": instances,
        "connections": connections,
        "specifications": specifications,
        "provenance": {
            "actor": "example",
            "operation": "create",
            "timestamp": "2026-10-29T00:00:00Z",
            "source_references": [],
            "notes": "built in an OpenFlowsheet example",
        },
    }


def commit(app, document, key):
    """Commit `document` as the project's next revision; its revision id."""
    edits = tuple(Edit("set", (member,), value) for member, value in document.items())
    head = app.get_project().head
    result = app.commit_change(Change(edits=edits), head, key)
    assert result.status == "committed", result
    return result.revision_id


def solved_state(app, run):
    """The solution state and the certificate of a converged run, read from its replay bundle."""
    bundle = next(o.artifact_id for o in run.outputs if o.kind == "replay_bundle")
    state = json.loads(app.artifact_bytes(f"{bundle}/solution-state.json"))["variables"]
    certificate = json.loads(app.artifact_bytes(f"{bundle}/solution-certificate.json"))
    return state, certificate

import json from openflowsheet.application.local import LocalApplication from openflowsheet.application.types import Change, Edit #: (SI unit, dimension exponents in the schema's base order) by kind, for instance parameters. KINDS = { "dimensionless": ("1", [0, 0, 0, 0, 0, 0, 0]), "pressure": ("Pa", [-1, 1, -2, 0, 0, 0, 0]), "molar_flow": ("mol/s", [0, 0, -1, 0, 1, 0, 0]), } #: The tolerance a fixed specification carries, by kind (ADR 0001's registered values). TOLERANCE = { "molar_flow": ({"absolute": 1e-9, "relative": 1e-8, "scale": 3.0}, "mol/s"), "temperature": ({"absolute": 1e-6}, "K"), "pressure": ({"absolute": 0.01}, "Pa"), "heat_rate": ({"absolute": 1e-5, "relative": 1e-8, "scale": 1e5}, "W"), } def parameter(value, kind, meaning): unit, dimension = KINDS[kind] return { "value": value, "unit": unit, "dimension": dimension, "kind": kind, "meaning": meaning, "role": "fixed", } def unit(unit_id, model, role, **parameters): return { "id": unit_id, "model": {"id": model, "version": "0.0.0-declared", "artifact_ref": None}, "semantic_role": role, "parameters": parameters, "policy": {"fidelity": "example", "validity": "example"}, } def stream(stream_id, source, target, phase): return { "id": stream_id, "kind": "material", "from": {"instance": source[0], "port": source[1]}, "to": {"instance": target[0], "port": target[1]}, "state_definition": "nTP-v1", "component_mapping": "revision_component_set", "phase_capability": phase, "notes": stream_id, } def fix(spec_id, object_type, object_id, path, value, kind, component=None): tolerance, si_unit = TOLERANCE[kind] target = {"object_type": object_type, "object_id": object_id, "path": path} if component is not None: target["component"] = component return { "id": spec_id, "target": target, "kind": kind, "unit": si_unit, "value": value, "tolerance": tolerance, "role": "fixed", "provenance": "example", } def revision(title, record_source, components, instances, connections, specifications): return { "schema_version": 1, "revision_id": title, "parent_revision": None, "title": title, "description": title, "component_set": {"record_source": record_source, "components": list(components)}, "instances": instances, "connections": connections, "specifications": specifications, "provenance": { "actor": "example", "operation": "create", "timestamp": "2026-10-29T00:00:00Z", "source_references": [], "notes": "built in an OpenFlowsheet example", }, } def commit(app, document, key): """Commit `document` as the project's next revision; its revision id.""" edits = tuple(Edit("set", (member,), value) for member, value in document.items()) head = app.get_project().head result = app.commit_change(Change(edits=edits), head, key) assert result.status == "committed", result return result.revision_id def solved_state(app, run): """The solution state and the certificate of a converged run, read from its replay bundle.""" bundle = next(o.artifact_id for o in run.outputs if o.kind == "replay_bundle") state = json.loads(app.artifact_bytes(f"{bundle}/solution-state.json"))["variables"] certificate = json.loads(app.artifact_bytes(f"{bundle}/solution-certificate.json")) return state, certificate

The flowsheet¶

In [3]:

REFRIGERANTS = "benchmarks/s07/components.yaml"  # selects CoolProp's HEOS, `coolprop-heos-v1`

cycle = revision(
    "r134a-cycle",
    REFRIGERANTS,
    ["R134a"],
    [
        unit("F", "syn001.feed_source", "feed"),
        unit(
            "G",
            "thermal.compressor",
            "compressor",
            efficiency=parameter(0.75, "dimensionless", "isentropic efficiency"),
        ),
        unit(
            "C",
            "thermal.heater",
            "heater",
            outlet_vapor_fraction=parameter(0.0, "dimensionless", "saturated liquid out"),
        ),
        unit("V", "thermal.valve", "valve"),
        unit(
            "E",
            "thermal.heater",
            "heater",
            outlet_vapor_fraction=parameter(1.0, "dimensionless", "saturated vapour out"),
        ),
        unit("K", "syn001.product_sink", "product_sink"),
    ],
    [
        stream("S1", ("F", "outlet"), ("G", "inlet"), "vapor"),
        stream("S2", ("G", "outlet"), ("C", "inlet"), "vapor"),
        stream("S3", ("C", "outlet"), ("V", "inlet"), "vapor_liquid"),
        stream("S4", ("V", "outlet"), ("E", "inlet"), "vapor_liquid"),
        stream("S5", ("E", "outlet"), ("K", "inlet"), "vapor_liquid"),
    ],
    [
        fix("feed-n", "connection", "S1", "state.n", 1.0, "molar_flow", component="R134a"),
        fix("feed-T", "connection", "S1", "state.T", 270.0, "temperature"),
        fix("feed-P", "connection", "S1", "state.P", 2.0e5, "pressure"),
        fix("compressor-P", "instance", "G", "outlet.P", 1.0e6, "pressure"),
        fix("valve-P", "instance", "V", "outlet.P", 2.0e5, "pressure"),
    ],
)

REFRIGERANTS = "benchmarks/s07/components.yaml" # selects CoolProp's HEOS, `coolprop-heos-v1` cycle = revision( "r134a-cycle", REFRIGERANTS, ["R134a"], [ unit("F", "syn001.feed_source", "feed"), unit( "G", "thermal.compressor", "compressor", efficiency=parameter(0.75, "dimensionless", "isentropic efficiency"), ), unit( "C", "thermal.heater", "heater", outlet_vapor_fraction=parameter(0.0, "dimensionless", "saturated liquid out"), ), unit("V", "thermal.valve", "valve"), unit( "E", "thermal.heater", "heater", outlet_vapor_fraction=parameter(1.0, "dimensionless", "saturated vapour out"), ), unit("K", "syn001.product_sink", "product_sink"), ], [ stream("S1", ("F", "outlet"), ("G", "inlet"), "vapor"), stream("S2", ("G", "outlet"), ("C", "inlet"), "vapor"), stream("S3", ("C", "outlet"), ("V", "inlet"), "vapor_liquid"), stream("S4", ("V", "outlet"), ("E", "inlet"), "vapor_liquid"), stream("S5", ("E", "outlet"), ("K", "inlet"), "vapor_liquid"), ], [ fix("feed-n", "connection", "S1", "state.n", 1.0, "molar_flow", component="R134a"), fix("feed-T", "connection", "S1", "state.T", 270.0, "temperature"), fix("feed-P", "connection", "S1", "state.P", 2.0e5, "pressure"), fix("compressor-P", "instance", "G", "outlet.P", 1.0e6, "pressure"), fix("valve-P", "instance", "V", "outlet.P", 2.0e5, "pressure"), ], )

Solve and read the cycle¶

In [4]:

if HAVE_COOLPROP:
    app = LocalApplication.in_memory()
    run = app.solve(commit(app, cycle, "cycle-1"), "default")
    print(run.outcome, run.verification_status)
    state, certificate = solved_state(app, run)
    w_compressor, q_evaporator, q_condenser = state["G.W"], state["E.Q"], state["C.Q"]
    cop = q_evaporator / w_compressor
    print(f"compressor work   {w_compressor:10.1f} W")
    print(f"evaporator duty   {q_evaporator:10.1f} W")
    print(f"condenser duty    {q_condenser:10.1f} W")
    print(f"valve outlet vapour fraction {state['S4.sat.V']:.4f}")
    print(f"coefficient of performance {cop:.3f}")
    app.close()

if HAVE_COOLPROP: app = LocalApplication.in_memory() run = app.solve(commit(app, cycle, "cycle-1"), "default") print(run.outcome, run.verification_status) state, certificate = solved_state(app, run) w_compressor, q_evaporator, q_condenser = state["G.W"], state["E.Q"], state["C.Q"] cop = q_evaporator / w_compressor print(f"compressor work {w_compressor:10.1f} W") print(f"evaporator duty {q_evaporator:10.1f} W") print(f"condenser duty {q_condenser:10.1f} W") print(f"valve outlet vapour fraction {state['S4.sat.V']:.4f}") print(f"coefficient of performance {cop:.3f}") app.close()

CONVERGED VERIFIED
compressor work       4728.6 W
evaporator duty      13990.9 W
condenser duty      -19321.4 W
valve outlet vapour fraction 0.3344
coefficient of performance 2.959

What this example shows¶

With CoolProp installed, the cycle's coefficient of performance is 2.959, with 4728.6 W of compressor work and 13990.9 W taken up in the evaporator; the throttled stream leaves the valve with a vapour fraction of 0.3344. The test suite checks these numbers when CoolProp is installed, and reports the example as skipped when it is not. This is a demonstration of the software, not validation evidence.