A methanol–water distillation column on the Wilson γ–φ basis¶

A 20-stage column at 101.325 kPa separates 1 mol/s of a saturated 40/60 methanol–water liquid feed on stage 10. The column has a total condenser with reflux ratio 1.5 and a reboiler drawing 0.6 mol/s of bottoms. The liquid is described by the Wilson activity model with an ideal-gas vapour (wilson-ig-v2); its three components are methanol, ethanol and water, and ethanol's feed flow is exactly zero here.

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 column¶

syn001.column takes its stage count, the feed stage of each feed connection (feed_stage.<connection id>), the pressure of every stage, and one specification each for the condenser and the reboiler.

In [2]:

WILSON = "benchmarks/s10/records.yaml"  # selects the Wilson γ–φ basis, `wilson-ig-v2`
COMPONENTS = ["CH3OH", "C2H5OH", "H2O"]
STAGES = 20

column_parameters = {
    "stage_count": parameter(float(STAGES), "dimensionless", "equilibrium stages"),
    "feed_stage.S-FEED": parameter(10.0, "dimensionless", "feed stage, counted from the top"),
    "condenser_reflux_ratio": parameter(1.5, "dimensionless", "reflux ratio"),
    "reboiler_bottoms_rate": parameter(0.6, "molar_flow", "bottoms molar flow"),
    **{
        f"stage_pressure.{k}": parameter(101325.0, "pressure", "stage pressure")
        for k in range(1, STAGES + 1)
    },
}
feed = {"CH3OH": 0.4, "C2H5OH": 0.0, "H2O": 0.6}

column = revision(
    "methanol-water-column",
    WILSON,
    COMPONENTS,
    [
        unit("FEED", "syn001.feed_source", "feed"),
        unit("COL", "syn001.column", "separator", **column_parameters),
        unit("DIST", "syn001.product_sink", "product_sink"),
        unit("BOT", "syn001.product_sink", "product_sink"),
    ],
    [
        stream("S-FEED", ("FEED", "outlet"), ("COL", "feed"), "liquid"),
        stream("S-DIST", ("COL", "distillate"), ("DIST", "inlet"), "liquid"),
        stream("S-BOT", ("COL", "bottoms"), ("BOT", "inlet"), "liquid"),
    ],
    [
        *(
            fix(f"feed-n-{c}", "connection", "S-FEED", "state.n", n, "molar_flow", component=c)
            for c, n in feed.items()
        ),
        fix("feed-T", "connection", "S-FEED", "state.T", 348.6155803604359, "temperature"),
        fix("feed-P", "connection", "S-FEED", "state.P", 101325.0, "pressure"),
    ],
)

WILSON = "benchmarks/s10/records.yaml" # selects the Wilson γ–φ basis, `wilson-ig-v2` COMPONENTS = ["CH3OH", "C2H5OH", "H2O"] STAGES = 20 column_parameters = { "stage_count": parameter(float(STAGES), "dimensionless", "equilibrium stages"), "feed_stage.S-FEED": parameter(10.0, "dimensionless", "feed stage, counted from the top"), "condenser_reflux_ratio": parameter(1.5, "dimensionless", "reflux ratio"), "reboiler_bottoms_rate": parameter(0.6, "molar_flow", "bottoms molar flow"), **{ f"stage_pressure.{k}": parameter(101325.0, "pressure", "stage pressure") for k in range(1, STAGES + 1) }, } feed = {"CH3OH": 0.4, "C2H5OH": 0.0, "H2O": 0.6} column = revision( "methanol-water-column", WILSON, COMPONENTS, [ unit("FEED", "syn001.feed_source", "feed"), unit("COL", "syn001.column", "separator", **column_parameters), unit("DIST", "syn001.product_sink", "product_sink"), unit("BOT", "syn001.product_sink", "product_sink"), ], [ stream("S-FEED", ("FEED", "outlet"), ("COL", "feed"), "liquid"), stream("S-DIST", ("COL", "distillate"), ("DIST", "inlet"), "liquid"), stream("S-BOT", ("COL", "bottoms"), ("BOT", "inlet"), "liquid"), ], [ *( fix(f"feed-n-{c}", "connection", "S-FEED", "state.n", n, "molar_flow", component=c) for c, n in feed.items() ), fix("feed-T", "connection", "S-FEED", "state.T", 348.6155803604359, "temperature"), fix("feed-P", "connection", "S-FEED", "state.P", 101325.0, "pressure"), ], )

Solve¶

In [3]:

app = LocalApplication.in_memory()
run = app.solve(commit(app, column, "column-1"), "default")
print(run.outcome, run.verification_status)
state, certificate = solved_state(app, run)

app = LocalApplication.in_memory() run = app.solve(commit(app, column, "column-1"), "default") print(run.outcome, run.verification_status) state, certificate = solved_state(app, run)

CONVERGED VERIFIED

Products¶

In [4]:

def composition(stream_id):
    flows = {c: state[f"{stream_id}.n.{c}"] for c in COMPONENTS}
    total = sum(flows.values())
    return total, {c: n / total for c, n in flows.items()}


d_total, x_d = composition("S-DIST")
b_total, x_b = composition("S-BOT")
print(f"distillate {d_total:.4f} mol/s, x(CH3OH) = {x_d['CH3OH']:.4f}")
print(f"bottoms    {b_total:.4f} mol/s, x(CH3OH) = {x_b['CH3OH']:.6f}")
print(f"methanol recovered to the distillate: {state['S-DIST.n.CH3OH'] / feed['CH3OH']:.4f}")
app.close()

def composition(stream_id): flows = {c: state[f"{stream_id}.n.{c}"] for c in COMPONENTS} total = sum(flows.values()) return total, {c: n / total for c, n in flows.items()} d_total, x_d = composition("S-DIST") b_total, x_b = composition("S-BOT") print(f"distillate {d_total:.4f} mol/s, x(CH3OH) = {x_d['CH3OH']:.4f}") print(f"bottoms {b_total:.4f} mol/s, x(CH3OH) = {x_b['CH3OH']:.6f}") print(f"methanol recovered to the distillate: {state['S-DIST.n.CH3OH'] / feed['CH3OH']:.4f}") app.close()

distillate 0.4000 mol/s, x(CH3OH) = 0.9941
bottoms    0.6000 mol/s, x(CH3OH) = 0.003916
methanol recovered to the distillate: 0.9941

What this example shows¶

The distillate is 0.4000 mol/s at a methanol mole fraction of 0.9941, and the bottoms carry a methanol mole fraction of 0.003916. The test suite checks these numbers every time this page is built. The Wilson model's comparison with measured vapour–liquid equilibrium is described in the user guide; this example is a demonstration, not validation evidence.