bpod_core.bpod¶

Module for interfacing with the Bpod Finite State Machine.

Submodules¶

`abc`	Abstract base classes used by the bpod module.
`constants`	Constants used by the bpod module.
`structs`	Data structures used by the bpod module.

Exceptions¶

exception bpod_core.bpod.BpodError ¶

Bases: Exception

Raised for errors specific to Bpod device operations.

exception bpod_core.bpod.BpodKeyError ¶

Bases: BpodError, KeyError

Exception class for Bpod-related key errors.

Classes¶

class bpod_core.bpod.Bpod ¶

Bases: SerialDevice, AbstractBpod

Interface to a Bpod Finite State Machine.

Connects to the Bpod hardware over USB. If neither port nor serial_number is given, the first idle Bpod found on any USB port is used.

Parameters:

port (str | None, default: None) – USB serial port of the device (e.g., ‘/dev/ttyACM0’ or ‘COM3’). Mutually exclusive with serial_number.
serial_number (str | None, default: None) – Serial number of the device to connect to. Mutually exclusive with port.
remote (bool, default: False) – Advertise the ZeroMQ service via Zeroconf so that other processes can connect to this Bpod instance remotely.

Raises:

BpodError – If no idle Bpod is found, the indicated port does not exist, or the device is not a supported Bpod model.

Examples

Connect to a Bpod on COM3:

with Bpod(port='COM3') as bpod:
    # do things

close() ¶

Close the connection to the Bpod.

Waits for any running trial to finish before closing the serial port.

Note

Prefer using Bpod as a context manager, which opens and closes the connection automatically.

Raises:: SerialException – If the port could not be closed.

get_data(*, concat=True, rechunk=False, lazy=False) ¶

Return trial data from the data queue.

Parameters:

concat (bool, default: True) – If True, pop and concatenate all DataFrames currently in the queue into a single DataFrame, blocking until at least one is available. If False, pop and return one DataFrame, blocking until one is available.
rechunk (bool, default: False) – If True, make sure that the result data is in contiguous memory. Only applies when concat=True.
lazy (bool, default: False) – If True, return a polars.LazyFrame. If False, return a polars.DataFrame.

Returns:

One trial’s data, or all available trials concatenated when concat=True.

Columns:

time (Datetime) – absolute Bpod timestamp.
trial (UInt16) – zero-based trial index.
state machine (Categorical) – state machine hash, see hash().
state (Categorical) – state name.
type (Enum) – event type.
event (Categorical) – input event name.
channel (Categorical) – channel name.
value (UInt8) – channel value.

Return type:

polars.DataFrame or polars.LazyFrame

Raises:

BpodError – If the queue is empty and no state machine is currently running.

open() ¶

Open the connection to the Bpod and perform a handshake.

Note

Prefer using Bpod as a context manager, which opens and closes the connection automatically.

Raises:

SerialException – If the port could not be opened.
BpodError – If the handshake fails.

peek_data(trigger_states=None, *, lazy=False) ¶

Return a snapshot of the current trial’s data.

Parameters:

trigger_states (Collection[str] | None, default: None) – Block until at least one of the given states has been entered, then return the snapshot. If None (default), returns immediately.
lazy (bool, default: False) – If True, return a polars.LazyFrame. If False (default), return a polars.DataFrame.

Returns:

Events recorded so far in the current trial. Returns an empty DataFrame if no trial is running.

Return type:

polars.DataFrame or polars.LazyFrame

Raises:

ValueError – If one or several of the trigger states are not part of the state machine.

reset_session_clock() ¶

Reset the Bpod session clock.

Returns:: True if the Bpod acknowledged the command.
Return type:: bool
Raises:: BpodError – When the method is called while a state machine is running.

run(state_machine=None, *, trial_number=None, validate=True) ¶

Run a state machine on the Bpod.

Validates, compiles, sends, and queues a state machine for immediate execution. If the Bpod is currently running a state machine, the new one is queued to run as soon as the current one finishes, with no inter-trial gap.

If called without an argument, the previously sent state machine is re-sent from the compilation cache and queued for immediate back-to-back execution.

Parameters:

state_machine (StateMachine | None, default: None) – The state machine to run. If not provided, the previously sent state machine is repeated.
trial_number (int | None, default: None) – The trial number to assign to the state machine. If not provided, the trial number is automatically incremented with each run.
validate (bool, default: True) – If False, the state machine will not be validated prior to compilation. This will speed up the process, but may result in errors or unexpected behavior if the state machine is invalid. Use with caution.

Raises:

RuntimeError – If called without an argument and no state machine has been run yet.
ValueError – If the state machine is invalid or exceeds hardware limitations.
ValidationError – If function arguments don’t match type hints.

Notes

This method returns once the state machine has been queued on the Bpod. The Bpod will then begin executing the state machine as soon as possible — immediately if the device is idle or right after the current state machine finishes. Subsequent calls of this method will result in continuous acquisition and zero inter-trial downtime (as long as the Bpod’s run queue stays filled).

send_softcode(softcode) ¶

Send a softcode to the state machine.

Can be used to trigger transitions.

Parameters:: softcode (int) – The softcode value to send.
Raises:: ValueError – If softcode is out of range.

set_softcode_handler(softcode_handler=None) ¶

Set the handler function for softcodes sent from the Bpod.

Parameters:: softcode_handler (Callable[[int], None] | None, default: None) – The function to call when a softcode is received.

set_status_led(enable) ¶

Enable or disable the Bpod status LED.

Parameters:: enable (bool) – True to turn the LED on, False to turn it off.
Returns:: True if the Bpod acknowledged the command.
Return type:: bool

stop_state_machine() ¶

Stop the currently running state machine.

update_modules() ¶

Update the list of connected modules and their configurations.

validate_state_machine(state_machine) ¶

Validate the provided state machine for compatibility with the hardware.

Parameters:: state_machine (StateMachine) – The state machine to validate.
Raises:: ValueError – If the state machine is invalid or not compatible with the hardware.

wait() ¶

Wait for the currently running state machine to finish.

Blocks until the state machine thread completes. If no state machine is currently running, this method returns immediately.

property address: str ¶: The ZeroMQ address of the Bpod.

property input_event_names: list[str] ¶: Names of all hardware input events.

inputs: dict[str, Input]¶: Dictionary of available input channels, keyed by name.

property is_queued: bool ¶: Check if a state machine is queued to run after the current one.

property is_ready: bool ¶: Check if a compiled state machine is loaded and ready to run.

property is_running: bool ¶: Check if the Bpod is currently running a state machine.

property location: str | None ¶: The Bpod’s user-defined location, or None if not set.

modules: dict[str, Module]¶: Dictionary of available modules, keyed by name.

property name: str | None ¶: The Bpod’s user-defined name, or None if not set.

outputs: dict[str, Output]¶: Dictionary of available output channels, keyed by name.

property port: str ¶

The name of the serial port.

Returns:: The device path of the serial port (e.g., ‘/dev/ttyACM0’).
Return type:: str

property serial0: ExtendedSerial ¶: Primary serial device for communication with the Bpod.

serial1: ExtendedSerial | None = None¶: Secondary serial device for communication with the Bpod.

serial2: ExtendedSerial | None = None¶: Tertiary serial device for communication with the Bpod - used by Bpod 2+ only.

property serial_number: str ¶: The Bpod’s unique serial number.

property version: VersionInfo ¶: Version information of the Bpod’s firmware and hardware.

class bpod_core.bpod.Channel ¶

Bases: object

Base class representing a channel on the Bpod device.

Parameters:

bpod (Bpod) – The Bpod instance associated with the channel.
name (str) – The name of the channel.
io_key (bytes) – The I/O type of the channel (e.g., b’B’, b’V’, b’P’).
index (int) – The index of the channel.

class bpod_core.bpod.Input ¶

Bases: Channel

Input channel class representing a digital input channel.

enable(enable) ¶

Enable or disable the input channel.

Parameters:: enable (bool) – True to enable the input channel, False to disable.
Returns:: True if the operation was successful, False otherwise.
Return type:: bool

override(state) ¶

Override the state of the input channel.

Parameters:: state (bool) – The state to set for the input channel.

read() ¶

Read the state of the input channel.

Returns:: True if the input channel is active, False otherwise.
Return type:: bool

property enabled: bool ¶

Check if the input channel is enabled.

Returns:: True if the input channel is enabled, False otherwise.
Return type:: bool

class bpod_core.bpod.Module ¶

Bases: object

Represents a Bpod module with its configuration and event names.

load_serial_message(message_id, message_bytes) ¶

Load a serial message targeting the module.

Serial messages are byte sequences targeting a specific module that can be triggered as output actions during a state machine run. Each message is identified by a message_id.

Parameters:

message_id (int) – Identifier for the message, in the range [0, 254].
message_bytes (bytes) – The message payload (1 to 3 bytes).

Returns:

True if the Bpod acknowledged the message, False otherwise.

Return type:

bool

Raises:

ValidationError – If the provided parameters cannot be validated or coerced to the expected type.
ValueError – If message_id, or message_bytes length is out of range.

set_relay(enabled) ¶

Enable or disable the serial relay for the module.

Parameters:: enabled (bool) – True to enable the relay, False to disable it.

property event_names: list[str] ¶: A list of event names associated with the module.

firmware_version: int | None = None¶: The firmware version of the module.

index: int¶: The index of the module.

is_connected: bool = False¶: Whether the module is connected.

n_events: int = 15¶: The number of events assigned to the module.

name: str¶: The name of the module.

property relay: bool ¶

Whether the serial relay for this module is enabled.

When True, the Bpod forwards bytes from the module to the host via the Bpod’s USB serial port.

class bpod_core.bpod.Output ¶

Bases: Channel

Output channel class representing a digital output channel.

override(state) ¶

Override the state of the output channel.

Parameters:: state (bool | int) – The state to set for the output channel. For binary I/O types, provide a bool. For pulse width modulation (PWM) I/O types, provide an int (0-255).

class bpod_core.bpod.RemoteBpod ¶

Bases: AbstractBpod

Proxy for a Bpod instance running in another process.

Use this when the Bpod hardware is managed by a separate process that was started with remote=True. RemoteBpod discovers that process via Zeroconf and forwards all method calls over ZeroMQ.

Note

This class is not yet fully functional. Some methods may be missing or incomplete.

Parameters:

address (str | None, default: None) – ZeroMQ address of the remote Bpod service. Discovered automatically if not given.
name (str | None, default: None) – Zeroconf service name to filter by during discovery.
serial_number (str | None, default: None) – Serial number of the target Bpod to filter by during discovery.
location (str | None, default: None) – Zeroconf location string to filter by during discovery.
timeout (float, default: 10.0) – Discovery timeout in seconds.

Raises:

TimeoutError – If no matching remote Bpod is found within timeout seconds.

close() ¶

Close the connection to the remote Bpod.

get_data(*, concat=True, rechunk=False, lazy=False) ¶

Return trial data from the data queue.

Parameters:

concat (bool, default: True) – If True, pop and concatenate all DataFrames currently in the queue into a single DataFrame, blocking until at least one is available. If False, pop and return one DataFrame, blocking until one is available.
rechunk (bool, default: False) – If True, make sure that the result data is in contiguous memory. Only applies when concat=True.
lazy (bool, default: False) – If True, return a polars.LazyFrame. If False, return a polars.DataFrame.

Returns:

One trial’s data, or all available trials concatenated when concat=True.

Columns:

time (Datetime) – absolute Bpod timestamp.
trial (UInt16) – zero-based trial index.
state machine (Categorical) – state machine hash, see hash().
state (Categorical) – state name.
type (Enum) – event type.
event (Categorical) – input event name.
channel (Categorical) – channel name.
value (UInt8) – channel value.

Return type:

polars.DataFrame or polars.LazyFrame

Raises:

BpodError – If the queue is empty and no state machine is currently running.

reset_session_clock() ¶

Reset the Bpod session clock.

Returns:: True if the Bpod acknowledged the command.
Return type:: bool
Raises:: BpodError – When the method is called while a state machine is running.

run(state_machine=None, *, trial_number=None, validate=True) ¶

Run a state machine on the Bpod.

If called without an argument, the previously sent state machine is re-sent from the compilation cache and queued for immediate back-to-back execution.

Parameters:

state_machine (StateMachine | None, default: None) – The state machine to run. If not provided, the previously sent state machine is repeated.
trial_number (int | None, default: None) – The trial number to assign to the state machine. If not provided, the trial number is automatically incremented with each run.
validate (bool, default: True) – If False, the state machine will not be validated prior to compilation. This will speed up the process, but may result in errors or unexpected behavior if the state machine is invalid. Use with caution.

Raises:

RuntimeError – If called without an argument and no state machine has been run yet.
ValueError – If the state machine is invalid or exceeds hardware limitations.
ValidationError – If function arguments don’t match type hints.

Notes

set_status_led(enable) ¶

Enable or disable the Bpod’s status LED.

Parameters:: enable (bool) – True to enable the status LED, False to disable.
Returns:: True if the operation was successful, False otherwise.
Return type:: bool

stop_state_machine() ¶

Stop the currently running state machine.

update_modules() ¶

Update the list of connected modules and their configurations.

property location: str | None ¶: The Bpod’s user-defined location, or None if not set.

property name: str | None ¶: The Bpod’s user-defined name, or None if not set.

property serial_number: str ¶: The Bpod’s unique serial number.

property version: VersionInfo ¶: Version information of the Bpod’s firmware and hardware.

Functions¶

bpod_core.bpod.discover_bpod(port=None, serial_number=None) ¶

Identify available Bpod devices connected via USB.

Scans for USB serial ports matching Bpod vendor/product IDs and verifies each device responds to a discovery message. Yields information about identified devices.

Parameters:

port (str | None, default: None) – Filter by specific device path (e.g., ‘/dev/ttyACM0’ or ‘COM3’).
serial_number (str | None, default: None) – Filter by USB serial number.

Yields:

BpodInfo – Information structure describing a Bpod device.

Return type:

Iterator[BpodInfo]

Examples

Iterate over available Bpods:

for device in discover_bpod():
    print(f"Found Bpod at {device}")

Get as a list:

devices = list(discover_bpod())

bpod_core.bpod.discover_remote_bpod(name=None, serial_number=None, location=None, timeout=10.0, poll_interval=1.0, *, local=True, remote=True) ¶

Identify available Bpod devices connected via ZeroMQ.

Parameters:

name (str | None, default: None) – Name of the Bpod device.
serial_number (str | None, default: None) – Serial number of the Bpod device.
location (str | None, default: None) – Location of the Bpod device.
timeout (float | None, default: 10.0) – How many seconds to monitor. Pass None to monitor indefinitely until the iterator is closed.
poll_interval (float, default: 1.0) – How often to poll for local service changes, in seconds. Default is 1.
local (bool, default: True) – Whether to search for services on the local machine.
remote (bool, default: True) – Whether to also search for services on the network.

Yields:

ServiceEvent – A named tuple with the following fields:

kind: str, either ‘added’ or ‘removed’
address: str, the service address, e.g., ‘tcp://192.168.1.10:1234’
properties: dict, the service properties, e.g., {‘name’: ‘MyDevice’}

Return type:

Iterator[ServiceEvent]