What-if studies and typed refusals¶

This notebook changes one specification of the Rankine cycle at a time, through commit_change, and solves each revision. It then asks for two things the software does not support, and shows how each is refused: a state outside the property model's domain, and a unit on a property basis it cannot bind to.

The sweep here is a loop of ordinary solves. It is not the parametric-sensitivity machinery of the studies layer, which the user guide describes.

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 cycle, as a function of the boiler temperature¶

In [2]:

WATER = "benchmarks/s05/components.yaml"


def rankine(boiler_t):
    return 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", boiler_t, "temperature"),
            fix("turbine-P", "instance", "T", "outlet.P", 1.0e4, "pressure"),
        ],
    )


app = LocalApplication.in_memory()
commit(app, rankine(673.15), "rankine-base")

WATER = "benchmarks/s05/components.yaml" def rankine(boiler_t): return 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", boiler_t, "temperature"), fix("turbine-P", "instance", "T", "outlet.P", 1.0e4, "pressure"), ], ) app = LocalApplication.in_memory() commit(app, rankine(673.15), "rankine-base")

Out[2]:

'rev-000001'

A sweep by edits¶

Each step is a Change holding one edit — a JSON path into the revision and its new value — committed against the current head. Every revision stays in the project's history.

In [3]:

document = rankine(673.15)
boiler = next(i for i, s in enumerate(document["specifications"]) if s["id"] == "boiler-T")
efficiencies = {}
for t in (573.15, 623.15, 673.15, 723.15, 773.15):
    head = app.get_project().head
    change = Change(edits=(Edit("set", ("specifications", boiler, "value"), t),))
    result = app.commit_change(change, head, f"boiler-{t}")
    run = app.solve(result.revision_id, "default")
    state, _ = solved_state(app, run)
    efficiencies[t] = -(state["T.W"] + state["P.W"]) / state["B.Q"]
    print(
        f"{t:7.2f} K  {run.verification_status:9s} efficiency {efficiencies[t]:.4f}"
        f"  turbine exit vapour fraction {state['S4.sat.V']:.4f}"
    )

document = rankine(673.15) boiler = next(i for i, s in enumerate(document["specifications"]) if s["id"] == "boiler-T") efficiencies = {} for t in (573.15, 623.15, 673.15, 723.15, 773.15): head = app.get_project().head change = Change(edits=(Edit("set", ("specifications", boiler, "value"), t),)) result = app.commit_change(change, head, f"boiler-{t}") run = app.solve(result.revision_id, "default") state, _ = solved_state(app, run) efficiencies[t] = -(state["T.W"] + state["P.W"]) / state["B.Q"] print( f"{t:7.2f} K {run.verification_status:9s} efficiency {efficiencies[t]:.4f}" f" turbine exit vapour fraction {state['S4.sat.V']:.4f}" )

 573.15 K  VERIFIED  efficiency 0.2786  turbine exit vapour fraction 0.8435

 623.15 K  VERIFIED  efficiency 0.2835  turbine exit vapour fraction 0.8743

 673.15 K  VERIFIED  efficiency 0.2892  turbine exit vapour fraction 0.9017

 723.15 K  VERIFIED  efficiency 0.2954  turbine exit vapour fraction 0.9272

 773.15 K  VERIFIED  efficiency 0.3020  turbine exit vapour fraction 0.9513

Outside the domain¶

IAPWS-95 in this software covers 273.16–1273.15 K. Asking the boiler for 1400 K does not extrapolate: the run ends with a typed outcome, and no certificate is issued.

In [4]:

too_hot = app.commit_change(
    Change(edits=(Edit("set", ("specifications", boiler, "value"), 1400.0),)),
    app.get_project().head,
    "boiler-1400",
)
run = app.solve(too_hot.revision_id, "default")
print(run.job_status, run.outcome, run.verification_status)
print([output.kind for output in run.outputs])

too_hot = app.commit_change( Change(edits=(Edit("set", ("specifications", boiler, "value"), 1400.0),)), app.get_project().head, "boiler-1400", ) run = app.solve(too_hot.revision_id, "default") print(run.job_status, run.outcome, run.verification_status) print([output.kind for output in run.outputs])

completed INITIALIZATION_FAILED None
['failure_bundle', 'run_manifest', 'replay_bundle']

A unit on a basis it does not support¶

A pump needs a liquid volume, which the Wilson ideal-gas basis does not provide. The revision validates as DRAFT, the structural checks say why, and solve refuses it.

In [5]:

on_wilson = rankine(673.15)
on_wilson["component_set"] = {
    "record_source": "benchmarks/s10/records.yaml",
    "components": ["CH3OH", "C2H5OH", "H2O"],
}
rid = commit(app, on_wilson, "rankine-on-wilson")
report = app.validate(rid, "simulation")
print(report.status)
check = next(c for c in report.checks if c.result != "PASS")
print(check.id, check.result, check.message)
try:
    app.solve(rid, "default")
except Exception as error:
    print(type(error).__name__, error)
app.close()

on_wilson = rankine(673.15) on_wilson["component_set"] = { "record_source": "benchmarks/s10/records.yaml", "components": ["CH3OH", "C2H5OH", "H2O"], } rid = commit(app, on_wilson, "rankine-on-wilson") report = app.validate(rid, "simulation") print(report.status) check = next(c for c in report.checks if c.result != "PASS") print(check.id, check.result, check.message) try: app.solve(rid, "default") except Exception as error: print(type(error).__name__, error) app.close()

DRAFT
STR-01 NOT_RUN not run: this revision cannot be analysed by T01's binding: model_unsupported(thermal.pump). thermal.pump binds on the coolprop-heos-v1, iapws95-v1 bases only, not on this revision's wilson-ig-v2; wilson-ig-v2 has no liquid volume, which a pump needs (L-GP-1)
ApplicationError No model with the id "thermal.pump" is available in this version.
  Why:     Each unit names the model that computes it. This version computes only the models it lists, and a unit's model id must match one of them exactly, prefix included (for example thermal.heater).
  Do:      Compare the unit's model id with the available models (the list_models operation, or openflowsheet api list_models): correct its spelling, or choose a listed model that does what the unit needs.
  Kind:    Not supported in this version: a limitation of OpenFlowsheet, not an error in your flowsheet.
  Code:    model_unsupported, in revision_not_ready (openflowsheet explain model_unsupported)
  Docs:    https://computational-chemical-engineering.github.io/openflowsheet/messages/refusals/#model_unsupported
  Details: revision_not_ready: revision 'rev-000008' validates DRAFT, not READY_FOR_SIMULATION

What this example shows¶

Across 573.15–773.15 K the efficiency rises from 0.2786 to 0.3020, every point VERIFIED. At 1400 K the run ends INITIALIZATION_FAILED, and the pump on the Wilson basis is refused with model_unsupported(thermal.pump). The test suite checks these numbers every time this page is built. This is a demonstration of the software, not validation evidence.