ReferencesBioModule API

BioModule API

BioModule is the single runnable unit in the communication-step kernel. Temporal modules advance state across windows; finite modules implement input-to-output computation behind the same contract.

Invocation policies require biosimulant>=0.0.26. Model and Lab manifests stay on the existing schema and do not declare the policy.

Contract

class BioModule:
    execution_policy = ExecutionPolicy.EACH_WINDOW
 
    def setup(self, config: dict[str, Any] | None = None) -> None: ...
    def reset(self) -> None: ...
    def set_inputs(self, signals: dict[str, BioSignal]) -> None: ...
    def execute(
        self,
        inputs: Mapping[str, BioSignal],
        *,
        context: ExecutionContext,
    ) -> Mapping[str, Any | BioSignal]: ...
    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: ...

New modules override the canonical execute() hook and explicitly declare a policy. Existing temporal modules may continue overriding advance_window() and normally get_outputs(). A module must not override both computation hooks.

Required semantics

  • execute(inputs, *, context) is the canonical hook for new finite and temporal modules.
  • advance_window(start, end) is the supported temporal compatibility hook and is not deprecated.
  • ExecutionPolicy.EACH_WINDOW is the compatibility default.
  • Use ONCE_BEFORE_RUN for preprocessing, finite inference, and pure one-shot pipelines; use ONCE_AFTER_RUN for final analysis and export.
  • Use canonical EACH_WINDOW when AI computation consumes evolving state or a new settle-independent temporal module advances through positive windows.
  • A canonical module that implicitly inherits EACH_WINDOW is accepted with a registration warning; first-party and generated modules declare it explicitly.
  • Canonical get_outputs() returns only the latest result successfully normalized and committed by BioWorld. Direct execute() calls do not update this cache.
  • Existing temporal get_outputs() must return declared typed signals.
  • 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.

Canonical temporal example

import biosimulant as biosim
 
 
class Gain(biosim.BioModule):
    execution_policy = biosim.ExecutionPolicy.EACH_WINDOW
 
    def __init__(self, gain: float = 1.0) -> None:
        self.gain = float(gain)
 
    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 execute(self, inputs, *, context):
        latest = inputs.get("x")
        if latest is None:
            return {}
        assert context.window_start is not None
        assert context.window_end is not None
        return {"y": float(latest.value) * self.gain}
 
    def snapshot(self):
        return {"gain": self.gain}
 
    def restore(self, snapshot):
        self.gain = float(snapshot["gain"])

BioWorld stamps the result at context.window_end. Canonical modules never run during zero-time settle() in 0.0.26; retain the temporal compatibility hook when settling is scientifically required.

Finite-computation example

class Predictor(biosim.BioModule):
    execution_policy = biosim.ExecutionPolicy.ONCE_BEFORE_RUN
 
    def inputs(self):
        return {"features": biosim.SignalSpec.array(dtype="float32", shape=(4,))}
 
    def outputs(self):
        return {"score": biosim.SignalSpec.scalar(dtype="float64")}
 
    def execute(self, inputs, *, context):
        return {"score": float(predict(inputs["features"].value))}

BioWorld wraps raw results using the output specs and stamps them at the current BioWorld boundary. Connected non-optional inputs must be available before invocation. Completed once-policy modules are not polled or retimestamped.

Valid phase edges move from before-run to each-window to after-run. Cycles remain valid only inside the each-window phase.

ExecutionContext

ExecutionContext is immutable. run_start and run_end describe the positive public BioWorld.run() call. window_start and window_end are present only for EACH_WINDOW; simulated_time resolves to the start, window end, or run end for before, window, or after invocation respectively. Wall-clock IDs, retries, and platform scheduling metadata are deliberately excluded.

BioWorld.run(duration <= 0) remains setup-only, so once policies require a positive-duration run. ONCE_AFTER_RUN runs only after all positive windows commit successfully.

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.

See Also

How Biosimulant WorksBioWorld API