BioModule API

BioModule is the runnable unit in the communication-step kernel. Modules declare typed ports, receive committed inputs at communication boundaries, advance across windows, and emit typed signals for atomic commit.

Contract

class BioModule:
    def setup(self, config: dict[str, Any] | None = None) -> None: ...
    def reset(self) -> None: ...
    def set_inputs(self, signals: dict[str, BioSignal]) -> None: ...
    def advance_window(self, start: float, end: float) -> None: ...
    def get_outputs(self) -> dict[str, BioSignal]: ...
    def inputs(self) -> Mapping[str, SignalSpec]: ...
    def outputs(self) -> Mapping[str, SignalSpec]: ...
    def snapshot(self) -> dict[str, Any]: ...
    def restore(self, snapshot: Mapping[str, Any]) -> None: ...
    def visualize(self) -> VisualSpec | list[VisualSpec] | None: ...

Only advance_window() and get_outputs() are abstract. The rest have defaults, but production modules should usually implement typed port declarations plus snapshot/restore.

Required semantics

advance_window(start, end) is the world-facing simulation hook.
get_outputs() must return a mapping of declared output port name to typed signal.
inputs() and outputs() must return port -> SignalSpec mappings when the module participates in validated wiring.
Signals should be emitted as ScalarSignal, ArraySignal, RecordSignal, or EventSignal.
snapshot() / restore() should round-trip the module state needed for deterministic continuation from a communication boundary.

Minimal example

import biosimulant as biosim
 
 
class Gain(biosim.BioModule):
    def __init__(self, gain: float = 1.0) -> None:
        self.gain = float(gain)
        self._latest = None
        self._window_end = 0.0
 
    def inputs(self):
        return {
            "x": biosim.SignalSpec.scalar(
                dtype="float64",
                max_age=0.2,
                stale_policy="warn",
            )
        }
 
    def outputs(self):
        return {"y": biosim.SignalSpec.scalar(dtype="float64")}
 
    def set_inputs(self, signals):
        self._latest = signals.get("x")
 
    def advance_window(self, start: float, end: float) -> None:
        self._window_end = end
 
    def get_outputs(self):
        if self._latest is None:
            return {}
        return {
            "y": biosim.ScalarSignal(
                source="gain",
                name="y",
                value=float(self._latest.value) * self.gain,
                emitted_at=self._window_end,
                spec=self.outputs()["y"],
            )
        }
 
    def snapshot(self):
        return {"gain": self.gain}
 
    def restore(self, snapshot):
        self.gain = float(snapshot["gain"])

When a module wants its output timestamp to reflect the boundary rather than the upstream source, cache the boundary in advance_window() and emit that value from get_outputs().

Convenience subclasses

BioModule remains the escape hatch for full control. Two optional subclasses remove common adapter boilerplate without changing the base contract.

`SignalEmitterBioModule`

Use SignalEmitterBioModule when your module owns its lifecycle but wants shared output storage and typed signal wrapping.

class Reporter(biosim.SignalEmitterBioModule):
    def outputs(self):
        return {"summary": biosim.SignalSpec.record(schema={"count": "int"})}
 
    def advance_window(self, start, end):
        self.publish_outputs(end, {"summary": {"count": 3}})

It provides source_name(), emit_signal(), publish_outputs(), get_outputs(), and clear_outputs().

`StatefulBioModule`

Use StatefulBioModule for fixed-step stateful models. Subclasses implement biology hooks and let the base handle window stepping, input overrides, and bounded history.

class Counter(biosim.StatefulBioModule):
    def __init__(self):
        super().__init__(integration_step=0.1, record_initial_state=True)
        self.count = 0
 
    def outputs(self):
        return {"count": biosim.SignalSpec.scalar(dtype="int64")}
 
    def step(self, h):
        self.count += 1
 
    def record_state(self, t):
        self._history.append({"t": t, "count": self.count})
 
    def output_payload(self, t):
        return {"count": self.count}

Port declarations

def inputs(self):
    return {
        "state": biosim.SignalSpec.record(
            schema={"v": "float64", "u": "float64"},
            max_age=0.001,
            stale_policy="error",
        )
    }
 
def outputs(self):
    return {
        "spikes": biosim.SignalSpec.event(
            schema={"ids": "list[int]"},
            description="Discrete spike deliveries",
        )
    }

Outputs declare the emitted profile. Inputs declare the accepted profile and freshness policy.

Durability

reset() is an optional convenience hook for UI-driven workflows you own. The kernel durability contract is snapshot() / restore(), which is what BioWorld.branch() and snapshot-based reruns rely on.

Notes

The world binds emitted outputs to the registered module name and the declared output port.
Empty output mappings do not clear prior committed state outputs from that module.
visualize() is optional and should return JSON-serializable visual specs.