Adapters
Integrate external simulators and ML models into B-Simulant compositions.
Overview
Adapters wrap external simulation tools and run under the TimeBroker runtime. They can
be used standalone, or synchronized together in multi-adapter simulations.
| Adapter | Purpose | Dependency |
|---|---|---|
TelluriumAdapter | SBML biochemical models | tellurium |
MLAdapter | ONNX neural networks | onnxruntime |
NeuroMLAdapter | NeuroML neural models | pyneuroml |
SimulatorAdapter Protocol
All adapters implement this interface:
class SimulatorAdapter(Protocol):
def setup(self, config: dict) -> None:
"""Initialize with configuration."""
...
def advance_to(self, t: float) -> None:
"""Advance simulation to time t."""
...
def set_inputs(self, signals: dict[str, BioSignal]) -> None:
"""Receive inputs from other modules."""
...
def get_outputs(self) -> dict[str, BioSignal]:
"""Emit outputs to other modules."""
...
def get_state(self) -> dict:
"""Checkpoint current state."""
...
def reset(self) -> None:
"""Reset to initial state."""
...TelluriumAdapter
Run SBML biochemical models via tellurium/libroadrunner.
Installation
pip install bsim[tellurium]
# or
pip install telluriumBasic Usage
from bsim.adapters import TelluriumAdapter
# Load from file
adapter = TelluriumAdapter(model_path="glycolysis.xml", expose=["S1", "S2"])
adapter.setup({})
# Load from string
adapter = TelluriumAdapter(sbml_string=sbml_content, expose=["S1", "S2"])
adapter.setup({})Standalone Simulation
# Advance simulation
adapter.advance_to(10.0)
# Access outputs
outputs = adapter.get_outputs()
print(outputs["S1"].value)In Composition
from bsim.adapters import TelluriumAdapter, TimeBroker
sbml = TelluriumAdapter(
model_path="pathway.xml",
expose=["ATP", "glucose", "pyruvate"],
)
broker = TimeBroker()
broker.register("metabolism", sbml, time_scale="seconds")
broker.setup()
for _t in broker.run(duration=10.0, dt=1.0):
passParameter Injection
# Override model parameters
adapter = TelluriumAdapter(model_path="model.xml", parameters={"k1": 0.5, "Km": 10.0})
adapter.setup({})
# Or via set_inputs() (map signal names to model inputs via the adapter config)Exposed Outputs
By default, all floating species are exposed. Customize with expose:
adapter = TelluriumAdapter(
model_path="model.xml",
expose=["ATP", "ADP", "glucose"] # Only these species
)
# Outputs available:
# - metabolism.out.ATP
# - metabolism.out.ADP
# - metabolism.out.glucose
# (assuming the adapter module is named "metabolism")MLAdapter
Run ONNX-format machine learning models in simulation loops.
Installation
pip install bsim[ml]
# or
pip install onnxruntimeBasic Usage
from bsim.adapters import MLAdapter
# Load ONNX model
adapter = MLAdapter(model_path="model.onnx")
adapter.setup({})
# Get model info
print(adapter.input_names()) # ['x1', 'x2']
print(adapter.output_names()) # ['y']Inference
from bsim.adapters import BioSignal
# Run inference
adapter.set_inputs({
"x1": BioSignal(source="client", name="x1", value=[1.0, 2.0, 3.0], time=0.0),
"x2": BioSignal(source="client", name="x2", value=[4.0, 5.0, 6.0], time=0.0),
})
adapter.advance_to(0.0)
outputs = adapter.get_outputs()
print(outputs)In Composition
from bsim.adapters import TelluriumAdapter, MLAdapter, TimeBroker
sbml = TelluriumAdapter(
model_path="pathway.xml",
expose=["ATP", "glucose"],
)
ml = MLAdapter(
model_path="drug_response.onnx",
inputs={"ATP": "x1", "glucose": "x2"},
outputs={"y": "efficacy"},
)
broker = TimeBroker()
broker.register("pathway", sbml, time_scale="seconds")
broker.register("predictor", ml, time_scale="seconds")
broker.connect("pathway.ATP", "predictor.ATP")
broker.connect("pathway.glucose", "predictor.glucose")
broker.setup()
for _t in broker.run(duration=10.0, dt=1.0):
passNeuroMLAdapter
Run NeuroML neural models via pyNeuroML.
Installation
pip install pyneuromlBasic Usage
from bsim.adapters import NeuroMLAdapter
adapter = NeuroMLAdapter(model_path="network.nml", expose=["v"])
adapter.setup({})
# Advance simulation
adapter.advance_to(50.0) # ms
outputs = adapter.get_outputs()In Composition
import bsim
from bsim.adapters import NeuroMLAdapter, TimeBroker
network = NeuroMLAdapter(model_path="circuit.nml", expose=["population_exc", "population_inh"])
broker = TimeBroker()
broker.register("network", network, time_scale="ms")
broker.setup()
for _t in broker.run(duration=50.0, dt=1.0):
passBioSignal
The message format for inter-adapter communication:
from bsim.adapters import BioSignal, SignalMetadata
signal = BioSignal(
source="metabolism", # Source module name
name="ATP", # Signal name
value=2.5, # Payload (any type)
time=100.0, # Simulation time
metadata=SignalMetadata(
units="mM",
min_value=0.0,
max_value=10.0
),
)TimeBroker
Synchronize adapters with different time scales:
from bsim.adapters import TimeBroker, AdaptiveTimeBroker
broker = TimeBroker()
broker.register("sbml_model", sbml_adapter, time_scale="seconds")
broker.register("neuro_model", neuro_adapter, time_scale="ms")
broker.connect("sbml_model.glucose", "neuro_model.energy")
broker.setup()
for t in broker.run(duration=1.0, dt=0.001):
signals = broker.get_all_signals()
# Adaptive (adjusts dt based on signal change)
adaptive = AdaptiveTimeBroker(min_dt=1e-4, max_dt=0.1, error_tolerance=0.01)Creating Custom Adapters
Wrap your own simulator:
from bsim.adapters import SimulatorAdapter, BioSignal
class MySimulatorAdapter(SimulatorAdapter):
def __init__(self, model_path: str, **config):
self.model = load_my_simulator(model_path)
self.config = config
self.state = {}
self.inputs = {}
self.current_time = 0.0
def setup(self, config: dict):
self.model.configure(**config)
def advance_to(self, t: float):
# Run simulation to time t
self.model.integrate_to(t)
self.current_time = t
# Update internal state
self.state = {
"output1": self.model.get_output1(),
"output2": self.model.get_output2()
}
def set_inputs(self, signals: dict[str, BioSignal]):
for name, signal in signals.items():
self.model.set_input(name, signal.value)
def get_outputs(self) -> dict[str, BioSignal]:
return {
"output1": BioSignal(
source="my_adapter",
name="output1",
value=self.state["output1"],
time=self.current_time,
)
}
def get_state(self) -> dict:
return self.model.checkpoint()
def reset(self):
self.model.reset()
self.state = {}Best Practices
Performance
- Batch Where Possible - Adapters should process arrays, not scalars
- Minimize I/O - Only expose necessary variables
- Cache Compiled Models - Avoid recompilation on reset
Composition
- Match Time Scales - Use TimeBroker for heterogeneous simulations
- Validate Connections - Check signal types match expected inputs
- Handle Missing Inputs - Provide sensible defaults
Testing
def test_adapter_roundtrip():
adapter = TelluriumAdapter(model_path="model.xml")
# Run simulation
adapter.advance_to(10.0)
result1 = adapter.get_outputs()
# Reset and run again
adapter.reset()
adapter.advance_to(10.0)
result2 = adapter.get_outputs()
# Should be identical
assert result1['ATP'].value == result2['ATP'].value