Metadata-Version: 2.4
Name: anthriq-bxi-interface
Version: 0.1.0
Summary: Python SDK for BXI devices — device control, registers, motors, and live EEG/impedance streaming over the BXI interface bridge
Project-URL: Homepage, https://anthriq.com
Project-URL: Documentation, https://docs.anthriq.com/bxi-studio/node-sdk/overview
Project-URL: Source, https://github.com/Anthriq/sw-bxi-studio
Author: Anthriq
License: MIT
Keywords: anthriq,biosignal,bxi,eeg,neural-interface,sdk,zmq
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: MacOS
Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: pyzmq>=25.0
Requires-Dist: typing-extensions>=4.5; python_version < '3.12'
Provides-Extra: dev
Requires-Dist: mypy>=1.8; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=7.4; extra == 'dev'
Requires-Dist: ruff>=0.3; extra == 'dev'
Description-Content-Type: text/markdown

# 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`](../anthriq-services).

## Install

```bash
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

```python
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

```python
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.

```python
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

```python
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:

```python
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:

```python
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.

```python
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:

```python
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:

```python
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

```bash
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
