respyra.core.breath_belt

Threaded Vernier Go Direct Respiration Belt reader.

Wraps the gdx convenience module with a background thread and queue so that PsychoPy’s frame loop (typically 60 Hz) is never blocked by the sensor’s blocking read() call.

Typical usage

with BreathBelt(connection=’ble’, period_ms=100) as belt:

# inside PsychoPy frame loop sample = belt.get_latest() if sample is not None:

timestamp, force = sample

Or without a context manager:

belt = BreathBelt() belt.start() try:

…

finally:

belt.stop()

Notes

• gdx.read() blocks for the full sampling period (e.g. 100 ms at 10 Hz). The background thread absorbs that wait, pushing samples into a thread-safe queue that the main thread drains without blocking.

• Always call stop() (or use the context manager) to ensure the device is cleanly disconnected. Failure to do so leaves the belt streaming, requiring a physical power-cycle.

• The gdx wrapper uses class-level state, so only one BreathBelt instance should exist at a time.

• Windows BLE + PsychoPy: On Windows, importing PsychoPy (pyglet/wxPython) puts the main thread into COM STA mode, which breaks Bleak’s BLE scanner. Bleak also requires the main thread for Windows Runtime callbacks. Therefore, start() must be called on the main thread before PsychoPy is imported.

exception respyra.core.breath_belt.BreathBeltError

Bases: Exception

Raised when the belt encounters a fatal error.

class respyra.core.breath_belt.BreathBelt(connection='ble', device_to_open='proximity_pairing', period_ms=100, sensors=None)

Bases: object

Non-blocking interface to the Vernier Go Direct Respiration Belt.

Parameters:

• connection (str) – 'ble' or 'usb'.

• device_to_open (str or None) – Device identifier passed to gdx.open(). 'proximity_pairing' connects to the nearest BLE device. A specific name like 'GDX-RB 081000A1' targets a known belt. None auto-connects for USB (single device) or prompts for BLE (avoid in automated scripts).

• period_ms (int) – Sampling interval in milliseconds. Minimum is 10 (100 Hz). Default 100 (10 Hz) gives good resolution for respiration waveforms.

• sensors (list[int]) – Channel numbers to enable. Default [1] enables the raw Force channel, which is the primary respiration signal.

start()

Open the device, configure sensors, and launch the reader thread.

On Windows, Bleak’s BLE scanner requires the main thread and a COM MTA apartment. Importing PsychoPy sets COM to STA, so start() must be called before import psychopy.

Raises:

BreathBeltError – If the belt is already started or the device fails to open.

Return type:

None

get_latest()

Return the most recent sample, discarding older ones.

Returns:

(timestamp, force_value) where timestamp is time.time() at the moment gdx.read() returned, and force_value is the reading (in Newtons) from the first enabled channel. Returns None if no samples are available.

Return type:

tuple[float, float] or None

Raises:

BreathBeltError – If the reader thread has recorded an error.

get_all()

Drain and return all queued samples since the last call.

Returns:

A list of (timestamp, force_value) tuples in chronological order. May be empty if no new samples have arrived.

Return type:

list[tuple[float, float]]

Raises:

BreathBeltError – If the reader thread has recorded an error.

stop()

Signal the reader thread to stop, join it, and close the device.

Safe to call multiple times – subsequent calls are no-ops.

Return type:

None

property is_running: bool

True if the reader thread is alive and no error has occurred.

property has_error: bool

True if the reader thread has recorded an error.

property error: BaseException | None

The exception recorded by the reader thread, or None.