anthriq-bxi-interface

Python SDK for BXI devices. Drives device registers, motors, and live EEG/impedance streams by talking to the C++ BXI interface bridge over ZeroMQ.

This is a port of the Node @anthriq_dev/bxi-interface package and speaks the identical wire protocol, so both SDKs can drive the same bridge daemon.

For pipelines, operator/pipeline registries, and recording, see the companion package anthriq-services.

Install

pip install anthriq-bxi-interface

pyzmq is the only runtime dependency. The binary stream protobuf is decoded by a small built-in reader, so there is no protobuf runtime to keep in step and no codegen step.

Requires Python 3.9 or newer, and a BXI interface bridge on the same host.

Quick start

import asyncio
from anthriq_bxi_interface import BxiClient

async def main():
    client = BxiClient(device_type="anthriq-instinct")
    await client.initialize("localhost")
    await client.connect()

    state = await client.execute_operation("system", "get_state", {})
    print(state.data)

    await client.shutdown()

asyncio.run(main())

initialize() spawns or attaches to the bridge and queries capabilities. connect() opens the device link. Both are required, in that order.

Read and write registers

response = await client.execute_operation(
    "registers", "read", {"type": "synap", "synap_id": 0}
)
fields = response.data["fields"]

from anthriq_bxi_interface import GainStage1, LpfSetting

await client.execute_operation(
    "registers",
    "write",
    {
        "type": "synap",
        "synap_id": 0,
        "fields": {
            "enabled": 1,
            "gain_stage_1": GainStage1.GAIN_20G,
            "lpf_setting": LpfSetting.HZ_300,
        },
    },
)

Writes are partial: send only the fields you intend to change. A write returns once the device accepts it, not once the value has propagated, so allow roughly 200 ms before reading back to verify.

Stream EEG

The stream lifecycle is six steps, and skipping the last three leaves the device transmitting after your process exits.

import asyncio
from anthriq_bxi_interface import BxiClient

async def record(duration_s: float):
    client = BxiClient(device_type="anthriq-instinct")
    await client.initialize("localhost")
    await client.connect()

    frames = 0

    def on_frame(update):
        nonlocal frames
        if update.success:
            frames += 1

    stream_id = 20
    subscription_id = None
    try:
        await client.execute_operation("eeg", "add_stream", {
            "stream": {
                "stream_id": stream_id,
                "protocol": "websocket",
                "host": "127.0.0.1",
                "port": 9020,
                "elements_before_flush": 30,
            }
        })
        await asyncio.sleep(1)

        subscription = await client.subscribe_to_operation(
            "eeg", "stream", on_frame, {"stream_id": stream_id}
        )
        subscription_id = subscription.subscription_id
        await asyncio.sleep(2)

        await client.execute_operation("eeg", "start_stream", {"stream_id": stream_id})
        await asyncio.sleep(duration_s)
        await client.execute_operation("eeg", "stop_stream", {"stream_id": stream_id})
        await asyncio.sleep(1.5)  # let buffered frames drain

        print(f"received {frames} frames")
    finally:
        if subscription_id:
            await client.unsubscribe_from_operation(
                "eeg", "stream", {"subscriptionId": subscription_id}
            )
        await client.execute_operation("eeg", "remove_stream", {"stream_id": stream_id})
        await client.shutdown()

asyncio.run(record(10))

Unsubscribe with the same feature and operation used to subscribe — "eeg" and "stream", not "eeg" and "subscribe".

stream_type (0x2100 for EEG, 0x2200 for impedance) is injected by the plugin. Do not set it yourself.

Control motors

response = await client.execute_operation("motors", "move", {
    "motor_id": 0,
    "displacement": 5,
    "operation": "forward",
})

from anthriq_bxi_interface import MotorStatus, motor_status_to_string

status = response.data["status"]
if status & MotorStatus.STALL:
    await client.execute_operation("motors", "stop", {"motor_id": 0})
    print(motor_status_to_string(status))

displacement is a 7-bit field, so 0–127 is accepted. Values above 20 emit a UserWarning because they likely exceed actuator travel; values outside 0–127 raise BxiValidationError before anything is sent.

Motor calibration can exceed the default 30 s timeout. Override it per call:

await client.invoke({
    "feature": "motors",
    "operation": "calibrate",
    "payload": {"motor_id": 0},
    "timeoutMs": 120_000,
})

Without an event loop

SyncBxiClient runs a loop on a background thread for scripts and notebooks:

import queue
from anthriq_bxi_interface import SyncBxiClient

frames = queue.Queue()

with SyncBxiClient(device_type="anthriq-instinct") as client:
    client.initialize("localhost")
    client.connect()
    client.subscribe_to_operation("eeg", "stream", frames.put, {"stream_id": 20})
    client.execute_operation("eeg", "start_stream", {"stream_id": 20})

Subscription callbacks run on that background thread, so keep them short and thread-safe. A queue.Queue handoff, as above, is the reliable pattern.

Responses and errors

Every operation returns an SdkResponse. A device-reported failure comes back as success=False; transport faults raise.

response = await client.execute_operation("motors", "read", {"motor_ids": [0]})

if response.success:
    print(response.data)
else:
    print(response.error.code, response.error.message)

unwrap() raises BxiOperationError instead, when a failure should abort the caller:

data = (await client.execute_operation("motors", "read", {"motor_ids": [0]})).unwrap()

Error codes come from three namespaces: hex codes from the bridge (0x01 is its generic failure), hex codes from the device stub, and uppercase SDK identifiers such as SUBSCRIBE_FAILED. The code may be an empty string, so branch on success and log message.

Several devices at once

Without instance_id, clients of the same device type share one bridge daemon. Set it to key the daemon per device, and pass device-selecting environment variables through bridge_env rather than mutating os.environ, which is shared and races on concurrent connects:

daq1 = BxiClient(device_type="ni-usb-daq", instance_id="dev1",
                 bridge_env={"NI_DEVICE_NAME": "Dev1"})
daq2 = BxiClient(device_type="ni-usb-daq", instance_id="dev2",
                 bridge_env={"NI_DEVICE_NAME": "Dev2"})

Configuration

Field	Default	Purpose
zmq_address	ipc://<tmp>/bxi-interface.sock	Bridge request socket
zmq_log_address	ipc://<tmp>/bxi-interface-logs.sock	Log socket
bridge_path	Auto-detected	Bridge executable
timeout	30000	Request timeout, milliseconds
debug	False	Raise SDK logging to DEBUG
device_type	None	Device identifier
instance_id	None	Per-device daemon key
bridge_env	{}	Environment for the spawned bridge

On Windows the socket paths resolve under %TEMP%.

The bridge executable is searched in order: $BXI_BRIDGE_PATH, the local core/cpp/build/bin/{Release,Debug} build, ~/.bxi-interface/bin, then PATH.

Development

uv venv --python 3.11
uv pip install -e ".[dev]"

pytest          # tests
mypy            # type check
ruff check src  # lint
python -m build # wheel + sdist

License

MIT