BioWorld & Modules

from bsim import BioWorld, FixedStepSolver
 
world = BioWorld(solver=FixedStepSolver())

from bsim import BioWorldEvent
 
# Event sequence during simulation:
# LOADED → BEFORE_SIMULATION → STEP (repeated) → AFTER_SIMULATION

# Basic run
result = world.simulate(steps=1000, dt=0.01)
 
# Listen to events
def listener(event, payload):
    print(event.name, payload)
 
world.on(listener)
world.simulate(steps=100, dt=0.1)
world.off(listener)

# Pause at step boundaries
world.request_pause()
 
# Resume from pause
world.request_resume()
 
# Stop gracefully
world.request_stop()

from bsim import BioModule, BioWorldEvent
 
class MyPopulation(BioModule):
    def __init__(self, initial_population: int):
        super().__init__()
        self.population = initial_population
        self.history = []
 
    def inputs(self) -> set[str]:
        """Declare input ports."""
        return {"births", "deaths"}
 
    def outputs(self) -> set[str]:
        """Declare output ports."""
        return {"population_state"}
 
    def subscriptions(self) -> set[BioWorldEvent]:
        """Events to listen to (returning a set filters events)."""
        return {BioWorldEvent.STEP}
 
    def on_event(self, event: BioWorldEvent, payload: dict, world):
        """Handle lifecycle events."""
        if event != BioWorldEvent.STEP:
            return
        t = float(payload.get("t", 0.0))
        self.history.append([t, self.population])
        world.publish_biosignal(
            self,
            topic="population_state",
            payload={"count": self.population, "t": t},
        )
 
    def on_signal(self, topic: str, payload: dict, source, world):
        """Handle incoming signals from other modules."""
        if topic == "births":
            self.population += int(payload.get("count", 0))
        elif topic == "deaths":
            self.population -= int(payload.get("count", 0))
 
    def visualize(self) -> dict:
        """Return visualization data."""
        return {
            "render": "timeseries",
            "data": {
                "title": "Population Over Time",
                "series": [{"name": "Population", "points": self.history}],
            },
        }
 
    def reset(self):
        """Reset state for fresh run."""
        self.history = []

class Neuron(BioModule):
    def inputs(self):
        return {"current", "synaptic_input"}
 
    def outputs(self):
        return {"spikes", "membrane_potential"}

def on_event(self, event, payload, world):
    if event == BioWorldEvent.STEP:
        # Emit to all connected modules
        world.publish_biosignal(self, topic="spikes", payload={"t": payload["t"], "ids": spike_times})
        world.publish_biosignal(self, topic="membrane_potential", payload={"t": payload["t"], "v": self.v})

def on_signal(self, topic: str, payload: dict, source, world):
    if topic == "current":
        self.I_ext = payload.get("I", 0.0)
    elif topic == "synaptic_input":
        self.I_syn += payload.get("I", 0.0)

from bsim import WiringBuilder
from bsim.packs.neuro import StepCurrent, IzhikevichPopulation, SpikeMonitor
 
wb = WiringBuilder(world)
 
# Add modules
wb.add("input", StepCurrent(I=10.0))
wb.add("neurons", IzhikevichPopulation(n=100))
wb.add("monitor", SpikeMonitor())
 
# Connect ports
wb.connect("input.out.current", ["neurons.in.current"])
wb.connect("neurons.out.spikes", ["monitor.in.spikes"])
 
# Apply to world
wb.apply()

# Single target
wb.connect("source.out.port", ["target.in.port"])
 
# Multiple targets (fan-out)
wb.connect("source.out.port", [
    "target1.in.port",
    "target2.in.port"
])
 
# Port naming convention: "{module_name}.{direction}.{port_name}"
# direction: "in" or "out"

# This will raise an error if ports don't exist
wb.connect("source.out.nonexistent", ["target.in.port"])
# Error: Module 'source' has no output port 'nonexistent'

from bsim import FixedStepSolver
 
solver = FixedStepSolver()
world = BioWorld(solver=solver)
 
# Each step advances by dt
result = world.simulate(steps=1000, dt=0.01)  # 10 time units total

from bsim import DefaultBioSolver, TemperatureParams, ScalarRateParams
 
solver = DefaultBioSolver(
    temperature=TemperatureParams(
        initial=300.0,         # Kelvin
        delta_per_step=0.1,    # Add per step
        rate_per_time=0.5,     # Add per second
        bounds=(273.15, 315.15)
    ),
    water=ScalarRateParams(name="water", initial=1.0, rate_per_time=-0.1, bounds=(0.0, 1.0)),
    oxygen=ScalarRateParams(name="oxygen", initial=0.3, rate_per_time=-0.05, bounds=(0.0, 1.0)),
)
 
world = BioWorld(solver=solver)

from bsim import Solver, BioWorldEvent
 
class AdaptiveStepSolver(Solver):
    def simulate(self, *, steps: int, dt: float, emit) -> dict:
        t = 0.0
        for i in range(steps):
            # Adaptive logic
            current_dt = self.compute_dt(t)
 
            # Emit step event
            emit(BioWorldEvent.STEP, {"i": i, "t": t, "dt": current_dt})
 
            t += current_dt
 
        return {"time": t, "steps": i + 1}

def visualize(self):
    return {
        "render": "timeseries",
        "data": {
            "title": "Membrane Potential",
            "xlabel": "Time (ms)",
            "ylabel": "Voltage (mV)",
            "series": [
                {"name": "Neuron 1", "points": self.v1_history},
                {"name": "Neuron 2", "points": self.v2_history}
            ]
        }
    }

def visualize(self):
    return {
        "render": "bar",
        "data": {
            "title": "Spike Counts",
            "items": [
                {"label": "Excitatory", "value": self.exc_count},
                {"label": "Inhibitory", "value": self.inh_count}
            ]
        }
    }

def visualize(self):
    return {
        "render": "table",
        "data": {
            "columns": ["Metric", "Value"],
            "rows": [
                ["Mean Rate", f"{self.mean_rate:.2f} Hz"],
                ["Total Spikes", str(self.total_spikes)]
            ]
        }
    }

def visualize(self):
    return {
        "render": "image",
        "data": {
            "src": self.raster_image_base64,
            "alt": "Spike Raster",
            "width": 800,
            "height": 400
        }
    }