Skip to content

Workflow

Analysis tasks often need more than a single compute step. Radial distribution functions require neighbour lists; cluster analysis feeds on per-frame descriptors; mean-squared displacements chain into autocorrelation functions. Doing this by hand — calling each node, threading outputs into inputs, checking that upstream ran before downstream — is tedious and easy to get wrong.

molpy.compute.Workflow makes this mechanical. You register named compute nodes, declare which parameters come from which upstream node (or from an external input), and the workflow handles topological order and result caching. It adds zero dependencies beyond Python's standard library.

Mental model

A workflow is a directed acyclic graph (DAG) of Compute callables:

  • Each node is a named Compute instance.
  • Each node has an inputs map: parameter name → source name.
  • A source is either another node’s name (upstream result) or an external key you pass to run(**externals).
  • Execution order is the topological sort; each node runs once and its result is cached for all dependents.

This is the same idea as a build system or a dataflow pipeline — without parallelism, conditionals, or streaming (see When not to use Workflow).

A linear chain

The simplest workflow is a straight pipeline. Node a runs first; node b takes a's output as its x parameter.

from molpy.compute import (
    Workflow,
    WorkflowCycleError,
    WorkflowDuplicateNodeError,
    WorkflowMissingInputError,
)
from molpy.compute.base import Compute

class Square(Compute):
    """Return the square of a number."""

    def __init__(self):
        super().__init__()

    def __call__(self, x):
        return x * x

class AddOne(Compute):
    """Add one."""

    def __init__(self):
        super().__init__()

    def __call__(self, x):
        return x + 1

wf = Workflow()
wf.add("square", Square(), inputs={"x": "x"})
wf.add("add_one", AddOne(), inputs={"x": "square"})

results = wf.run(x=3)
print(results)  # {'square': 9, 'add_one': 10}

wf.add() returns the workflow, so calls chain:

wf = Workflow()
(
    wf.add("square", Square(), inputs={"x": "x"}).add(
        "add_one", AddOne(), inputs={"x": "square"}
    )
)

External inputs

Every parameter a node needs must appear in its inputs — including the ones that come from outside the workflow. When the source name does not match any registered node, the workflow treats it as an external input, and you supply it to run().

wf = Workflow()
wf.add("square", Square(), inputs={"x": "x"})

# "x" maps to no registered node, so it is an external input
results = wf.run(x=5)
print(results)  # {'square': 25}

If you forget an external input, run() raises WorkflowMissingInputError before executing any node — no partial execution.

try:
    wf.run()
except WorkflowMissingInputError as exc:
    print(exc.missing)  # {'x'}

Diamond reuse

When two downstream nodes share an upstream, the upstream runs exactly once. This is not a special code path — it falls out of the result cache naturally.

class Count(Compute):
    """Counts how many times it was called."""

    def __init__(self):
        super().__init__()
        self.call_count = 0

    def __call__(self, x):
        self.call_count += 1
        return x

wf = Workflow()
upstream = Count()
wf.add("upstream", upstream, inputs={"x": "x"})
wf.add("branch_a", AddOne(), inputs={"x": "upstream"})
wf.add("branch_b", AddOne(), inputs={"x": "upstream"})

results = wf.run(x=1)
assert upstream.call_count == 1  # not 2

Real example: radial distribution function

A radial distribution function \(g(r)\) needs both a frame (for the box) and a neighbour list (for the pair distances). We express this as a two-node workflow.

import numpy as np
from molpy.compute import Workflow, NeighborList, RDF
import molpy as mp

# Build a simple test frame — 10 atoms in a 10 Å cube
rng = np.random.default_rng(42)
xyz = rng.uniform(0.0, 10.0, size=(10, 3))
frame = mp.Frame()
frame["atoms"] = {"x": xyz[:, 0], "y": xyz[:, 1], "z": xyz[:, 2]}
frame.box = mp.Box.cubic(10.0)

wf = Workflow()
wf.add("nlist", NeighborList(cutoff=5.0), inputs={"frame": "frame"})
wf.add(
    "rdf",
    RDF(n_bins=100, r_max=5.0),
    inputs={"frames": "frame", "neighbors": "nlist"},
)

results = wf.run(frame=frame)
rdf_array = np.asarray(results["rdf"].rdf)
print(f"g(r) has {len(rdf_array)} bins, max value {rdf_array.max():.3f}")

NeighborList needs only the frame, so frame is its one external input. RDF needs both the original frame (for box dimensions) and the neighbour list — so its inputs map references both "frame" (external) and "nlist" (upstream node). Parameter names in inputs must match the __call__ signature (frame vs frames / neighbors).

Introspection

You can inspect the workflow before running it.

wf.nodes                 # ['nlist', 'rdf'] — insertion order
wf.external_inputs       # {'frame'} — all unregistered source names
wf.topological_order()   # ['nlist', 'rdf'] — execution order
wf.predecessors("rdf")   # {'nlist'} — node predecessors only (no externals)

predecessors() deliberately excludes external inputs — it describes the internal DAG topology, not every dependency.

Cycle detection

If adding a node would create a cycle, add() raises WorkflowCycleError and rolls back — the workflow state is unchanged.

wf = Workflow()
wf.add("a", Square(), inputs={"x": "x"})

# b depends on a → OK
wf.add("b", Square(), inputs={"x": "a"})

# Linear extension remains acyclic
try:
    wf.add("c", Square(), inputs={"x": "b"})  # OK, linear
except WorkflowCycleError:
    pass  # not reached — this is valid

Cycle detection happens at registration time, not at execution time. You get immediate feedback.

Immutability contract

The workflow never mutates registered compute nodes. Calling run() twice with the same external inputs produces identical results, and node.dump() (which serialises the node's configuration) returns the same dictionary before and after.

results1 = wf.run(x=5)
results2 = wf.run(x=5)
assert results1 == results2  # always true

When not to use Workflow

Workflow is serial and synchronous. It does not (yet) support parallel execution, conditional nodes, or streaming. For those patterns, use TopologicalSorter.get_ready() / .done() directly, or wait for a future version.

See also