Flowsheet & recycle¶

The recycle/tear solver and the Flowsheet container. Recycle loops are converged to a fixed point and differentiated by the implicit function theorem, so gradients flow through the converged flowsheet rather than the iteration.

flowsheet ¶

Sequential-modular flowsheet solving with a differentiable recycle/tear solver.

A flowsheet with a recycle is an implicit problem: the value of a torn recycle stream must equal what the flowsheet computes for it once that guess is fed back in. Writing the single forward pass as g(tear, theta) -> tear (mix the feed with the recycle guess, run the units, return the recomputed recycle), the converged flowsheet is the fixed point tear* = g(tear*, theta).

tear_solve finds that fixed point with a Wegstein-accelerated iteration (the workhorse of sequential-modular simulators, far more robust than plain direct substitution on tight recycles) and differentiates the converged solution by the implicit function theorem (a hand-written custom_vjp). The forward iteration count never appears in the backward pass, so a gradient of any product spec with respect to an operating variable costs one adjoint solve, no matter how many recycle iterations were needed. That is what makes whole-process, recycle-closed gradient optimisation tractable.

The tear state can be any JAX pytree (a Stream, a list of them, a dict, ...); it is flattened internally. Convergence is judged on a relative norm, so mixed-scale states (flows, temperature, pressure) all converge to the same relative tolerance without manual scaling.

Flowsheet is a thin declarative wrapper: register feeds and unit functions, mark a tear, and call Flowsheet.solve.

Classes:

Name	Description
`Flowsheet`	A small declarative flowsheet: named streams produced by connected units.

Functions:

Name	Description
`tear_solve`	Converge a recycle by solving the tear fixed point `tear = g(tear, theta)`.

Flowsheet `dataclass` ¶

Flowsheet(
    feeds: dict[str, Stream] = dict(),
    units: list[_Unit] = list(),
    tears: dict[str, Stream] = dict(),
)

A small declarative flowsheet: named streams produced by connected units.

Build a flowsheet by registering feeds and units, then designate a recycle tear and call solve. Each unit is a plain function of its input streams (and the shared theta) returning one or more output streams; the flowsheet evaluates the units in registration order, which the caller arranges to be a valid sequential-modular order with the recycle torn.

Example::

fs = Flowsheet()
fs.feed("fresh", fresh_stream)
fs.unit("mixer", lambda fresh, rec, th: mix([fresh, rec]),
        inputs=("fresh", "recycle"), outputs=("mixed",))
fs.unit("drum", lambda mixed, th: flash_drum(mixed, th["T"], th["P"]),
        inputs=("mixed",), outputs=("vapor", "liquid"))
fs.unit("split", lambda liq, th: splitter(liq, [th["r"], 1 - th["r"]]),
        inputs=("liquid",), outputs=("recycle", "purge"))
fs.tear("recycle", recycle_guess)
streams = fs.solve({"T": 320.0, "P": 2e6, "r": 0.6})
product = streams["vapor"]

Methods:

Name	Description
`feed`	Register a fresh feed stream by name. Returns `self` for chaining.
`unit`	Register a unit `fn(*input_streams, theta) -> output stream(s)`.
`tear`	Designate stream `name` as a recycle tear with an initial `guess`.
`solve`	Solve the flowsheet (closing any recycle) and return all named streams.

feed ¶

feed(name: str, stream: Stream) -> Flowsheet

Register a fresh feed stream by name. Returns self for chaining.

unit ¶

unit(
    name: str,
    fn: UnitFn,
    *,
    inputs: Sequence[str],
    outputs: Sequence[str],
) -> Flowsheet

Register a unit fn(*input_streams, theta) -> output stream(s).

fn receives the named input streams positionally followed by the shared theta pytree, and returns either a single Stream (for one output name) or a tuple/list of streams aligned with outputs.

tear ¶

tear(name: str, guess: Stream) -> Flowsheet

Designate stream name as a recycle tear with an initial guess.

solve ¶

solve(
    theta: Any = None, **tear_solve_kwargs: Any
) -> dict[str, Stream]

Solve the flowsheet (closing any recycle) and return all named streams.

theta is the differentiable parameter pytree passed to every unit; any output stream is differentiable with respect to it. Extra keyword arguments are forwarded to tear_solve.

tear_solve ¶

tear_solve(
    g: Callable[[Any, Any], Any],
    tear0: Any,
    theta: Any = None,
    *,
    q_min: float = -5.0,
    q_max: float = 0.0,
    tol: float = 1e-10,
    atol: float = 1e-12,
    max_iter: int = 200,
) -> Any

Converge a recycle by solving the tear fixed point tear = g(tear, theta).

Parameters:

Name	Type	Description	Default
`g`	`Callable[[Any, Any], Any]`	One sequential-modular pass of the flowsheet. Given a tear-stream guess (any pytree) and the parameter pytree `theta`, it runs the units and returns the recomputed tear stream(s) in the same pytree structure.	required
`tear0`	`Any`	Initial guess for the torn stream(s).	required
`theta`	`Any`	Differentiable parameter pytree (operating conditions, specs, feed). Pass the quantities you want to differentiate through here; gradients flow to `theta` by implicit differentiation. Closed-over constants are fine but are treated as non-differentiable.	`None`
`q_min`	`float`	Lower bound on the Wegstein acceleration factor.	`-5.0`
`q_max`	`float`	Upper bound on the Wegstein acceleration factor. The default `[-5, 0]` accelerates without over-damping; widen `q_max` toward `1` to damp oscillatory recycles.	`0.0`
`tol`	`float`	Relative tolerance for the convergence norm.	`1e-10`
`atol`	`float`	Absolute floor for the convergence norm.	`1e-12`
`max_iter`	`int`	Iteration cap for both the forward and adjoint solves.	`200`

Returns:

Type	Description
`Any`	The converged tear stream(s), in the structure of `tear0`. Differentiable
`Any`	with respect to `theta`.