Tutorial 0 — A Process in 12 Lines

The fastest possible tour of process-bigraph: define a process as a typed function, wire it, and run it — all on one screen.

The @process decorator infers a process's config_schema from the decorated function's keyword-only parameters, so the config values are the function's arguments. No config_schema dict, no initialize recast. The class-based Process API stays available as the escape hatch for stateful processes.

from process_bigraph import allocate_core, Composite, process

1. Define a process

A process is a typed function of (state, interval) plus keyword-only config. Here, exponential decay of a species S at some rate. The rate: float = 0.1 keyword-only parameter becomes the process's config — its type and default are read straight off the signature.

@process(inputs={"S": "float"}, outputs={"S": "float"})
def decay(state, interval, *, rate: float = 0.1):
    return {"S": -rate * state["S"] * interval}   # additive delta

The config_schema was inferred for you — rate is a float defaulting to 0.1:

decay.config_schema

{'rate': {'_type': 'float', '_default': 0.1}}

2. Wire and run

Register the process on a core, drop it into a Composite document wired to a shared store S, and run. The document is the same {_type, address, config, inputs, outputs} dict the engine consumes — the decorator changed only how the process is authored, not the wire format. Identity wires ("S": ["S"]) connect the S port to the top-level S store.

core = allocate_core()
core.register_link("decay", decay)

sim = Composite({"state": {
    "S": 10.0,
    "decay": {
        "_type": "process",
        "address": "local:decay",
        "config": {"rate": 0.2},          # override the inferred default
        "inputs":  {"S": ["S"]},
        "outputs": {"S": ["S"]},
    },
}}, core=core)

sim.run(10.0)                              # advance 10 time units
sim.state["S"]

module `pbg_emitters.parquet_emitter` not found during dynamic import

1.0737418240000003

That is the whole loop: define → wire → run.

3. Ports carry units

A port can declare _units, and the engine auto-converts across a wire whenever two compatible-but-different units meet. Here a process emits mass in femtograms (fg) into a store declared in picograms (pg) — the engine scales 1000 fg → 1 pg at the wire, with no code on your part.

@process(inputs={}, outputs={"mass": {"_type": "float", "_units": "fg"}})
def emit_fg(state, interval, *, amount: float = 1000.0):
    return {"mass": amount}               # 1000 femtograms

core = allocate_core()
core.register_link("emit_fg", emit_fg)

sim = Composite({"state": {
    "mass": {"_type": "float", "_units": "pg"},   # store reads picograms
    "emitter": {
        "_type": "process",
        "address": "local:emit_fg",
        "config": {"amount": 1000.0},
        "inputs":  {},
        "outputs": {"mass": ["mass"]},
    },
}}, core=core)

sim.run(1.0)
sim.state["mass"]                          # 1000 fg auto-converted to 1.0 pg

1.0

Escape hatch: the class-based API

@process targets the ~70% of processes that are a pure typed function of (state, interval) + config. Stateful or initialize-heavy processes still use the class-based Process, which stays fully supported — the two forms interoperate freely in the same Composite:

from process_bigraph import Process

class Decay(Process):
    config_schema = {"rate": {"_type": "float", "_default": 0.1}}

    def initialize(self, config):
        self.rate = float(config["rate"])

    def inputs(self):  return {"S": "float"}
    def outputs(self): return {"S": "float"}

    def update(self, state, interval):
        return {"S": -self.rate * state["S"] * interval}

Next steps

• Tutorial 1 — Process-Bigraph Basics: Steps, workflows, emitters.
• Tutorial 2 — Wrapping an ODE Solver: expose an existing scientific API.
• Tutorial 4 — Composing a Biological Model: the central dogma from four small processes over shared molecular state.