bpod_core.com¶

Module providing extended serial communication functionality.

Classes¶

class bpod_core.com.ChunkedSerialReader ¶

Bases: Protocol

A protocol for reading chunked data from a serial port.

This class provides methods to buffer incoming data and retrieve it in chunks.

Parameters:

chunk_size (int) – The fixed size of chunks to emit to the callback function when enough data has accumulated in the buffer.
callback (Callable[[bytearray], None]) – A function to call with each chunk of data.
buffer (bytearray | None, default: None) – Pre-allocated buffer to use for accumulation. If None, a new bytearray is created.

connection_lost(exc) ¶

Called when the serial port is closed or the reader loop terminated otherwise.

Parameters:: exc (BaseException | None) – The exception that caused the connection to be closed, if any.

connection_made(transport) ¶

Called when a connection is made.

Parameters:: transport (ReaderThread[Self]) – The reader thread that created this protocol instance.

data_received(data) ¶

Called with snippets received from the serial port.

Parameters:: data (bytes) – The binary data received from the serial port.

class bpod_core.com.ExtendedSerial ¶

Bases: Serial

Enhances pySerial’s Serial with additional functionality.

query(query, size=1) ¶

Query data from the serial port.

This method is a combination of write() and read().

Parameters:

query (Buffer) – Query to be sent to the serial port.
size (int, default: 1) – The number of bytes to receive from the serial port.

Returns:

Data returned by the serial device in response to the query.

Return type:

bytes

Examples

Send a command and read back multiple bytes:

response = serial_port.query(b'\x4A', size=4)

query_struct(query, fmt) ¶

Query structured data from the serial port.

This method queries a specified number of bytes from the serial port and unpacks it into a tuple according to the provided format.

Parameters:

query (Buffer) – Query to be sent to the serial port.
fmt (str | Struct) – A pre-compiled struct or a format string compatible with the struct module’s format specifications.

Returns:

A tuple containing the unpacked data read from the serial port. The structure of the tuple corresponds to the format specified in fmt.

Return type:

tuple[Any, ...]

Examples

Send a command and unpack the response as two unsigned 8-bit integers:

major, minor = serial_port.query_struct(b'\x4a', 'BB')

Pass a pre-compiled struct.Struct to avoid re-parsing the format string on every call:

fmt = struct.Struct('<2H')
while acquiring:
    value, flag = serial_port.query_struct(b'\x4a', fmt)

read_bool() ¶

Read a boolean value from the serial port.

Returns:: True if the byte read is non-zero, False otherwise.
Return type:: bool

read_int16() ¶

Read a 16-bit signed integer from the serial port (little-endian).

Returns:: The 16-bit signed integer read from the port.
Return type:: int

read_int32() ¶

Read a 32-bit signed integer from the serial port (little-endian).

Returns:: The 32-bit signed integer read from the port.
Return type:: int

read_int64() ¶

Read a 64-bit signed integer from the serial port (little-endian).

Returns:: The 64-bit signed integer read from the port.
Return type:: int

read_int8() ¶

Read an 8-bit signed integer from the serial port.

Returns:: The 8-bit signed integer read from the port.
Return type:: int

read_struct(fmt) ¶

Read structured data from the serial port.

This method reads a specified number of bytes from the serial port and unpacks it into a tuple according to the provided format.

Parameters:: fmt (str | Struct) – A pre-compiled struct or a format string compatible with the struct module’s format specifications.
Returns:: A tuple containing the unpacked data read from the serial port. The structure of the tuple corresponds to the format specified in fmt.
Return type:: tuple[Any, ...]
Raises:: struct.error – If fmt is invalid or the data cannot be unpacked.

Examples

Read one unsigned 16-bit integer followed by two unsigned 8-bit integers:

major, minor, patch = serial_port.read_struct('<HBB')

Pass a pre-compiled struct.Struct to avoid re-parsing the format string on every call:

fmt = struct.Struct('<HBB')
while acquiring:
    major, minor, patch = serial_port.read_struct(fmt)

See also

Format specifications used by the struct module.

read_struct_iter(fmt, n=1, *, flatten=False) ¶

Read structured data from the serial port as an iterator.

Parameters:

fmt (str | Struct) – A pre-compiled struct or a format string compatible with the struct module’s format specifications.
n (int, default: 1) – Number of records to read.
flatten (bool, default: False) – If True, yield individual values instead of tuples.

Yields:

tuple or Any – Each unpacked record as a tuple, or individual values if flatten=True.

Return type:

Iterator[tuple[Any, ...]] | Iterator[Any]

Notes

All bytes are read in a single call before any records are yielded. Use stream_struct() instead if records should be yielded as they arrive.

Examples

Read three records as tuples:

for value, flag in serial_port.read_struct_iter('<HB', 3):
    print(value, flag)

Read two records as individual integers:

v1, f1, v2, f2 = serial_port.read_struct_iter('<HB', 2, flatten=True)

See also

Format specifications used by the struct module.

read_uint16() ¶

Read a 16-bit unsigned integer from the serial port (little-endian).

Returns:: The 16-bit unsigned integer read from the port.
Return type:: int

read_uint32() ¶

Read a 32-bit unsigned integer from the serial port (little-endian).

Returns:: The 32-bit unsigned integer read from the port.
Return type:: int

read_uint64() ¶

Read a 64-bit unsigned integer from the serial port (little-endian).

Returns:: The 64-bit unsigned integer read from the port.
Return type:: int

read_uint8() ¶

Read an 8-bit unsigned integer from the serial port.

Returns:: The 8-bit unsigned integer read from the port.
Return type:: int

stream_struct(fmt, n, *, flatten=True) ¶

Stream structured data from the serial port, yielding one record at a time.

Parameters:

fmt (str | Struct) – A pre-compiled struct or a format string compatible with the struct module’s format specifications.
n (int) – Number of records to read.
flatten (bool, default: True) – If True, yield individual values instead of tuples.

Yields:

Any or tuple – Individual values if flatten=True, otherwise one tuple per record.

Return type:

Iterator[Any] | Iterator[tuple[Any, ...]]

Notes

Each record is read and yielded as soon as its bytes arrive, using one readinto() call per record. Use read_struct_iter() instead if all data is available upfront and a single read call is preferred.

Examples

Stream 100 uint32 samples, yielding each as it arrives:

for sample in serial_port.stream_struct('<I', 100):
    process(sample)

Stream 10 records of 3 floats as tuples:

for x, y, z in serial_port.stream_struct('<3f', 10, flatten=False):
    print(x, y, z)

See also

Format specifications used by the struct module.

verify(query=b'', expected_response=b'\\x01') ¶

Verify the response of the serial port.

This method sends a query to the serial port and checks if the response matches the expected response.

Parameters:

query (Buffer, default: b'') – The query to be sent to the serial port.
expected_response (bytes, default: b'\\x01') – The expected response from the serial port.

Returns:

True if the response matches the expected response, False otherwise.

Return type:

bool

Examples

Send a handshake byte and check for acknowledgement:

if not serial_port.verify(b'\x48'):
    raise RuntimeError('Device did not acknowledge handshake')

write_bool(value) ¶

Write a boolean value to the serial port.

Parameters:: value (bool) – The boolean value to write (True → 0x01, False → 0x00).
Returns:: Number of bytes written, or None if the write fails.
Return type:: int | None

write_int16(value) ¶

Write a 16-bit signed integer to the serial port (little-endian).

Parameters:: value (int) – An integer in the range [-32768, 32767].
Returns:: Number of bytes written, or None if the write fails.
Return type:: int | None
Raises:: struct.error – If value is out of range.

write_int32(value) ¶

Write a 32-bit signed integer to the serial port (little-endian).

Parameters:: value (int) – An integer in the range [-2147483648, 2147483647].
Returns:: Number of bytes written, or None if the write fails.
Return type:: int | None
Raises:: struct.error – If value is out of range.

write_int64(value) ¶

Write a 64-bit signed integer to the serial port (little-endian).

Parameters:: value (int) – An integer in the range [-9223372036854775808, 9223372036854775807].
Returns:: Number of bytes written, or None if the write fails.
Return type:: int | None
Raises:: struct.error – If value is out of range.

write_int8(value) ¶

Write an 8-bit signed integer to the serial port.

Parameters:: value (int) – An integer in the range [-128, 127].
Returns:: Number of bytes written, or None if the write fails.
Return type:: int | None
Raises:: struct.error – If value is out of range.

write_struct(fmt, *data) ¶

Write structured data to the serial port.

This method packs the provided data into a binary format according to the specified format string and writes it to the serial port.

Parameters:

fmt (str | Struct) – A pre-compiled struct or a format string compatible with the struct module’s format specifications.
*data (Any) – Variable-length arguments representing the data to be packed and written, corresponding to the format specifiers in format_string.

Returns:

The number of bytes written to the serial port, or None if the write operation fails.

Return type:

int | None

Raises:

struct.error – If data cannot be packed with the given format_string.
SerialTimeoutException – In case a write timeout is configured for the port and the time is exceeded.

Examples

Write a command byte followed by a 16-bit unsigned integer:

serial_port.write_struct('<BH', 0x4A, 1000)

See also

Format specifications used by the struct module.

write_uint16(value) ¶

Write a 16-bit unsigned integer to the serial port (little-endian).

Parameters:: value (int) – An integer in the range [0, 65535].
Returns:: Number of bytes written, or None if the write fails.
Return type:: int | None
Raises:: struct.error – If value is out of range.

write_uint32(value) ¶

Write a 32-bit unsigned integer to the serial port (little-endian).

Parameters:: value (int) – An integer in the range [0, 4294967295].
Returns:: Number of bytes written, or None if the write fails.
Return type:: int | None
Raises:: struct.error – If value is out of range.

write_uint64(value) ¶

Write a 64-bit unsigned integer to the serial port (little-endian).

Parameters:: value (int) – An integer in the range [0, 18446744073709551615].
Returns:: Number of bytes written, or None if the write fails.
Return type:: int | None
Raises:: struct.error – If value is out of range.

write_uint8(value) ¶

Write an 8-bit unsigned integer to the serial port.

Parameters:: value (int) – An integer in the range [0, 255].
Returns:: Number of bytes written, or None if the write fails.
Return type:: int | None
Raises:: struct.error – If value is out of range.

class bpod_core.com.SerialDevice ¶

Bases: AbstractContextManager

Base class for implementing drivers for USB serial devices.

Handles connection lifecycle — opening, closing, and cleanup on garbage collection — and provides subclasses with an ExtendedSerial connection and port metadata. Derive from this class instead of using serial.Serial directly to get automatic resource management and consistent error handling.

Parameters:

port (str) – The serial port device path (e.g., ‘/dev/ttyUSB0’ or ‘COM3’).
open_connection (bool, default: True) – Whether to open the connection immediately.

Raises:

serial.SerialException – If the specified port does not exist.

Examples

Subclass SerialDevice and use _serial to communicate:

class MyDevice(SerialDevice):
    _serial_device_name = 'My Device'

    def ping(self) -> bool:
        return self._serial.verify(b'\x48')

with MyDevice('/dev/ttyACM0') as dev:
    assert dev.ping()

_rename_serial_device(new_name) ¶

Rename the serial device.

Use this method to change the name of the serial device after instantiation. It will ensure that the finalizer is updated to reflect the new name.

Parameters:: new_name (str) – The new name of the serial device.

close() ¶

Close the serial connection.

If the connection is already closed, this method does nothing.

Raises:: serial.SerialException – If the connection cannot be closed.

open() ¶

Open the serial connection.

If the connection is already open, this method does nothing.

Raises:: serial.SerialException – If the connection cannot be opened.

_port_info: serial.tools.list_ports_common.ListPortInfo¶: Information about the serial port associated with the device.

_serial: ExtendedSerial¶: The serial connection to the device.

_serial_device_name: str = 'serial device'¶: Name of the serial device.

property port: str ¶

The name of the serial port.

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

Functions¶

bpod_core.com.find_ports(**filters) ¶

Find serial ports matching specified criteria.

Multiple filters use AND logic. Iterables within a single filter use OR logic.

Parameters:

Port attributes to filter by. Values can be:

Scalar: exact match
Sequence: match any item (OR logic)
re.Pattern: regex match (use re.compile())

Returns:

Ports matching all criteria.

Return type:

list[ListPortInfo]

Examples

Find by vendor ID:

find_ports(vid=0x16C0)

Find using regex pattern:

find_ports(device=re.compile(r'/dev/ttyACM\d+'))

Find multiple values:

find_ports(pid=[0x0483, 0x048B])

Combine filters:

find_ports(vid=0x16C0, device=re.compile(r'/dev/ttyACM\d+'))

Notes

Strings use exact matching. Use re.compile() for regex patterns.

bpod_core.com.verify_serial_discovery(port, expected_message, timeout=1, trigger=None) ¶

Check if a device sends an expected discovery message on a serial port.

Opens the specified serial port and waits to receive bytes matching the expected discovery message. Optionally executes a function first, which can be used to trigger the device’s discovery routine, e.g., by sending a command.

Parameters:

port (str) – The serial port to read from (e.g., ‘/dev/ttyUSB0’ or ‘COM3’).
expected_message (bytes) – The exact byte sequence expected from the device.
timeout (float, default: 1) – Maximum time (in seconds) to wait for the discovery message.
trigger (Callable[[], Any] | None, default: None) – A function to call before reading. Use this to trigger the device’s discovery routine.

Returns:

True if the device sent the expected message within the timeout period, False otherwise (including if the port cannot be opened).

Return type:

bool

Attributes¶

bpod_core.com.FilterValue = str | int | re.Pattern[str] | None | collections.abc.Sequence['FilterValue']¶: Type for filter values used in find_ports().