A Rankine cycle on IAPWS-95 water¶

One mole per second of water runs once through a simple Rankine cycle: a pump (3 MPa, isentropic efficiency 0.75), a boiler that heats it to 673.15 K, a turbine back to 10 kPa (efficiency 0.85) and a condenser that returns saturated liquid. The water properties are the IAPWS-95 formulation (iapws95-v1), selected by the revision's record source.

The notebook uses only the public application contract: LocalApplication, commit_change, validate, solve and the artifacts a run registers.

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 [1]:

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¶

thermal.* units evaluate their property calls on the pure-fluid basis. Each unit carries the semantic_role of its kind (pump, heater, turbine); the role is descriptive and the binder does not read it.

In [2]:

WATER = "benchmarks/s05/components.yaml"  # selects the IAPWS-95 basis, `iapws95-v1`

rankine = revision(
    "rankine",
    WATER,
    ["H2O"],
    [
        unit("F", "syn001.feed_source", "feed"),
        unit(
            "P",
            "thermal.pump",
            "pump",
            efficiency=parameter(0.75, "dimensionless", "isentropic efficiency"),
        ),
        unit("B", "thermal.heater", "heater"),
        unit(
            "T",
            "thermal.turbine",
            "turbine",
            efficiency=parameter(0.85, "dimensionless", "isentropic efficiency"),
        ),
        unit(
            "C",
            "thermal.heater",
            "heater",
            outlet_vapor_fraction=parameter(0.0, "dimensionless", "saturated liquid out"),
        ),
        unit("K", "syn001.product_sink", "product_sink"),
    ],
    [
        stream("S1", ("F", "outlet"), ("P", "inlet"), "liquid"),
        stream("S2", ("P", "outlet"), ("B", "inlet"), "liquid"),
        stream("S3", ("B", "outlet"), ("T", "inlet"), "vapor"),
        stream("S4", ("T", "outlet"), ("C", "inlet"), "vapor_liquid"),
        stream("S5", ("C", "outlet"), ("K", "inlet"), "vapor_liquid"),
    ],
    [
        fix("feed-n", "connection", "S1", "state.n", 1.0, "molar_flow", component="H2O"),
        fix("feed-T", "connection", "S1", "state.T", 318.15, "temperature"),
        fix("feed-P", "connection", "S1", "state.P", 1.0e4, "pressure"),
        fix("pump-P", "instance", "P", "outlet.P", 3.0e6, "pressure"),
        fix("boiler-T", "instance", "B", "outlet.T", 673.15, "temperature"),
        fix("turbine-P", "instance", "T", "outlet.P", 1.0e4, "pressure"),
    ],
)

WATER = "benchmarks/s05/components.yaml" # selects the IAPWS-95 basis, `iapws95-v1` rankine = revision( "rankine", WATER, ["H2O"], [ unit("F", "syn001.feed_source", "feed"), unit( "P", "thermal.pump", "pump", efficiency=parameter(0.75, "dimensionless", "isentropic efficiency"), ), unit("B", "thermal.heater", "heater"), unit( "T", "thermal.turbine", "turbine", efficiency=parameter(0.85, "dimensionless", "isentropic efficiency"), ), unit( "C", "thermal.heater", "heater", outlet_vapor_fraction=parameter(0.0, "dimensionless", "saturated liquid out"), ), unit("K", "syn001.product_sink", "product_sink"), ], [ stream("S1", ("F", "outlet"), ("P", "inlet"), "liquid"), stream("S2", ("P", "outlet"), ("B", "inlet"), "liquid"), stream("S3", ("B", "outlet"), ("T", "inlet"), "vapor"), stream("S4", ("T", "outlet"), ("C", "inlet"), "vapor_liquid"), stream("S5", ("C", "outlet"), ("K", "inlet"), "vapor_liquid"), ], [ fix("feed-n", "connection", "S1", "state.n", 1.0, "molar_flow", component="H2O"), fix("feed-T", "connection", "S1", "state.T", 318.15, "temperature"), fix("feed-P", "connection", "S1", "state.P", 1.0e4, "pressure"), fix("pump-P", "instance", "P", "outlet.P", 3.0e6, "pressure"), fix("boiler-T", "instance", "B", "outlet.T", 673.15, "temperature"), fix("turbine-P", "instance", "T", "outlet.P", 1.0e4, "pressure"), ], )

Commit, validate, solve¶

A project holds revisions and runs. in_memory() opens a throw-away one; create(path) and open(path) keep it on disk. solve with the policy default uses the route's registered policy.

In [3]:

app = LocalApplication.in_memory()
revision_id = commit(app, rankine, "rankine-1")
report = app.validate(revision_id, "simulation")
print(report.status)

app = LocalApplication.in_memory() revision_id = commit(app, rankine, "rankine-1") report = app.validate(revision_id, "simulation") print(report.status)

READY_FOR_SIMULATION

In [4]:

run = app.solve(revision_id, "default")
print(run.policy_id, run.outcome, run.verification_status)

run = app.solve(revision_id, "default") print(run.policy_id, run.outcome, run.verification_status)

T06-revision-v2 CONVERGED VERIFIED

The result¶

The run's replay bundle holds the solved state (solution-state.json) and the certificate the independent verifier issued (solution-certificate.json). Works and duties are in W, positive into the fluid.

In [5]:

state, certificate = solved_state(app, run)
w_turbine, w_pump, q_boiler, q_condenser = state["T.W"], state["P.W"], state["B.Q"], state["C.Q"]
efficiency = -(w_turbine + w_pump) / q_boiler
print(f"turbine work   {w_turbine:12.1f} W")
print(f"pump work      {w_pump:12.1f} W")
print(f"boiler duty    {q_boiler:12.1f} W")
print(f"condenser duty {q_condenser:12.1f} W")
print(f"turbine exit vapour fraction {state['S4.sat.V']:.4f}")
print(f"thermal efficiency {efficiency:.4f}")

state, certificate = solved_state(app, run) w_turbine, w_pump, q_boiler, q_condenser = state["T.W"], state["P.W"], state["B.Q"], state["C.Q"] efficiency = -(w_turbine + w_pump) / q_boiler print(f"turbine work {w_turbine:12.1f} W") print(f"pump work {w_pump:12.1f} W") print(f"boiler duty {q_boiler:12.1f} W") print(f"condenser duty {q_condenser:12.1f} W") print(f"turbine exit vapour fraction {state['S4.sat.V']:.4f}") print(f"thermal efficiency {efficiency:.4f}")

turbine work       -15905.1 W
pump work              72.5 W
boiler duty         54752.5 W
condenser duty     -38859.2 W
turbine exit vapour fraction 0.9017
thermal efficiency 0.2892

The certificate¶

VERIFIED means every residual, balance and specification check passed at its registered tolerance. Two checks are not_applicable: the independent phase-split check is degenerate at a saturated pure-fluid state. The certificate states what it does not claim:

In [6]:

failed = [c["id"] for c in certificate["checks"] if c["result"] != "pass"]
print(
    certificate["verification_status"],
    len(certificate["checks"]),
    "checks,",
    len(failed),
    "not passed",
)
for statement in certificate["statements"]:
    print("-", statement)
app.close()

failed = [c["id"] for c in certificate["checks"] if c["result"] != "pass"] print( certificate["verification_status"], len(certificate["checks"]), "checks,", len(failed), "not passed", ) for statement in certificate["statements"]: print("-", statement) app.close()

VERIFIED 75 checks, 2 not passed
- VERIFIED certifies the residuals at the registered tolerances; at first order the scaled solution is determined to within the recorded solution-error bound b; no solution accuracy beyond b is claimed.
- This is a target-model verification of the stated equations at the stated state. It is not an experimental validation of any thermodynamic model or correlation, and it is not an optimality claim.
- No finite test suite constitutes a proof that the verifier has no defects.
- shares the property provider with the solver: provider_id=iapws95-v1, implementation_sha256=21fe37db62d751829316c244343736fa1067910754c96d7c915b1e26d2dc306b, data_sha256=b9e93877e37874f1b3a483e298a45692e264d70686d4c158372084b96abbe6ed, reference_convention=IAPWS95-ref-v1; this is a consistency check of process bookkeeping, not a validation of the enthalpy correlations

What this example shows¶

With the stated inputs the cycle's thermal efficiency is 0.2892, the turbine delivers 15905.1 W and the boiler takes 54752.5 W; the turbine exhaust has a vapour fraction of 0.9017. These numbers are checked by the test suite every time this page is built. This is a demonstration of the software, not validation evidence.