anthriq-services

Python SDK for the BXI backend services. Runs processing pipelines, manages the operator and pipeline registries, and handles recording and playback by talking to the four C++ service servers over ZeroMQ.

This is a port of the Node @anthriq_dev/services package and speaks the identical wire protocol, so both SDKs can drive the same backend.

For device control, registers, motors, and live streaming, see the companion package anthriq-bxi-interface.

Install

pip install anthriq-services

pyzmq is the only runtime dependency. Requires Python 3.9 or newer, and the BXI backend servers on the same host.

Quick start

import asyncio
from anthriq_services import BxiServicesClient

async def main():
    async with BxiServicesClient() as services:
        result = await services.executor.create_pipeline(
            pipeline_id="eeg-pipeline",
            config={
                "nodes": [
                    {"id": "eeg", "type": "eegstream", "config": {"channels": 8}},
                    {"id": "fft", "type": "fft", "config": {"windowSize": 256}},
                ],
                "pipes": [{"source": "eeg", "destination": "fft"}],
            },
        )
        print(result.data["nodeCount"])

        await services.executor.start_pipeline("eeg-pipeline")
        await asyncio.sleep(30)
        await services.executor.stop_pipeline("eeg-pipeline")
        await services.executor.destroy_pipeline("eeg-pipeline")

asyncio.run(main())

The context manager connects on entry and shuts down on exit. Use await services.initialize() instead when you want a health check first, and auto_spawn=True to start servers that are not running.

The four clients hang off the unified client:

Member	Class
services.executor	ExecutorClient
services.operator_registry	OperatorRegistryClient
services.pipeline_registry	PipelineRegistryClient
services.recording	RecordingClient

Pipelines

validation = await services.executor.validate_pipeline(config)
if not validation.data["valid"]:
    raise RuntimeError(validation.data["message"])

await services.executor.create_pipeline("run-1", config)
await services.executor.start_pipeline("run-1")

await services.executor.signal_pipeline(
    "run-1", signal="set_gain", node_ids=["eeg"], args={"gain": 24}
)

info = await services.executor.get_pipeline_info("run-1")
for node in info.data.get("nodes", []):
    print(node["id"], node["state"])

Destroy every pipeline you create, including on the error path. A pipeline that is not destroyed keeps its nodes, sockets, and device claims allocated in the server, which outlives your process.

Recording and playback

started = await services.recording.start_recording(
    recording_id="session-001",
    device_id="instinct-a1",
    sample_rate=1000,
    num_channels=8,
    channels=[{"id": 0, "label": "Fp1", "unit": "uV"}],
    kind="experiment",
)
batch_size = started.data["batchSize"]

await services.recording.write_batch(
    recording_id="session-001",
    values=[[1.2, 1.3], [0.8, 0.9]],   # channels-first
    timestamps=[0, 1000],
)

await services.recording.stop_recording("session-001")

values is channels-first, so values[channel][sample]. Match the batch length to batchSize from the start response to avoid partial row groups.

A recording that is never stopped leaves its final segment unclosed: the samples are on disk, but the duration and segment index are incomplete and playback of that segment fails.

Playback is pull-based:

playback = await services.recording.start_playback("session-001", channels=[0, 1])
playback_id = playback.data["playbackId"]

while True:
    batch = await services.recording.get_next_batch(playback_id)
    if batch.data.get("state") == "completed":
        break
    process(batch.data.get("values"), batch.data.get("timestamps"))

await services.recording.stop_playback(playback_id)

seek and marker times are microseconds from the start of the recording — the same base — and deliberately not wall-clock, so nothing needs re-anchoring on replay.

Markers

Markers use a dictionary/occurrence split: a MarkerDef describes a kind of marker once, and each occurrence references it by defId. A 1500-epoch run therefore carries a handful of definitions rather than 1500.

await services.recording.add_markers(
    recording_id="session-001",
    defs=[
        {"id": 1, "name": "Stimulus", "color": "#FF5C00", "kind": "event", "source": "hardware"},
        {"id": 2, "name": "Trial", "color": "#634391", "kind": "epoch", "source": "experiment"},
    ],
    markers=[
        {"defId": 1, "tUs": 1_000_000},
        {"defId": 2, "tUs": 1_000_000, "durUs": 2_000_000, "data": {"rep": 7}},
    ],
)

Always batch. Re-sending defs on every batch is safe — they upsert, which keeps each batch self-contained. Supply seq explicitly to replace an occurrence, which makes a retry idempotent.

delete_markers with no seqs deletes every marker for the recording, definitions included.

Registries

await services.operator_registry.install(name="fft", version="1.0.0")

listing = await services.operator_registry.list()
for op in listing.data["operators"]:
    print(op["name"], op["version"], op["installPath"])

meta = await services.operator_registry.info(name="fft", version="1.0.0")
for signal in meta.data.get("signals", []):
    print(signal["name"])   # cross-check before signal_pipeline

Operators are keyed by name, pipelines by id — the one asymmetry between the two registries.

The executor reports a missing operator as a pipeline creation failure, not a registry error, so check operator status first when create_pipeline fails on a definition that used to work.

Log streaming

Service-level logs (lifecycle, pipeline create and destroy) come from one callback:

services = BxiServicesClient(
    on_log=lambda entry: print(f"[{entry['service']}] {entry['level']}: {entry['message']}")
)

Per-pipeline logs (node output and pipeline lifecycle) arrive on each pipeline's own socket. Pass on_pipeline_log and the client subscribes on create_pipeline and unsubscribes on destroy_pipeline:

services = BxiServicesClient(
    on_pipeline_log=lambda pid, entry: print(f"[{pid}] {entry['level']}: {entry['message']}")
)

To attach to a pipeline this client did not create, pass the logEndpoint from create or info — a socket address, not a port:

await services.executor.subscribe_to_pipeline_logs("existing", "ipc:///tmp/bxi-pipe-existing.sock")

Endpoints

Endpoints are IPC sockets, derived from the resolved backend installation:

Service	Request socket	Log socket
Executor	ipc:///tmp/bxi-executor.sock	ipc:///tmp/bxi-executor-logs.sock
Operator Registry	ipc:///tmp/bxi-operator-registry.sock	…-logs.sock
Pipeline Registry	ipc:///tmp/bxi-pipeline-registry.sock	…-logs.sock
Recording	ipc:///tmp/bxi-recording.sock	…-logs.sock

On Windows the base is %TEMP% with forward slashes, which ZeroMQ IPC requires.

A CLI-installed backend gets a version suffix so several versions can run side by side, for example ipc:///tmp/bxi-executor-0.1.0.sock. Bundled and development backends use plain names.

When every request times out, the SDK and backend usually disagree on an endpoint. ZeroMQ queues messages to an absent peer instead of refusing, so a wrong endpoint surfaces as a timeout rather than a connection error:

from anthriq_services import get_endpoint_info
print(get_endpoint_info())

Spawning servers

from anthriq_services import ServiceProcessManager, ServiceProcessConfig

manager = ServiceProcessManager(
    services=[
        ServiceProcessConfig(
            name="executor",
            binary_path="/opt/bxi/bin/executor_server",
            endpoint="ipc:///tmp/bxi-executor.sock",
            log_endpoint="ipc:///tmp/bxi-executor-logs.sock",
        ),
    ],
    auto_spawn=True,
)

statuses = await manager.ensure_running()
await manager.stop_all()   # stops only what this manager spawned

The 20-second readiness budget accommodates cold start on Windows, where on-access virus scanning routinely adds 3–8 seconds before the server binds. A process that exits early is reported with its stderr rather than waiting out the budget.

Backend versions

bxi-backend list
bxi-backend info 0.1.0
bxi-backend install backend-0.2.0.tar.gz --suffix dev --set-default
bxi-backend default 0.2.0_dev
bxi-backend uninstall 0.1.0
bxi-backend endpoints          # print resolved endpoints

Environment variable	Effect
BXI_BACKEND_VERSION	Select a version, overriding the default
BXI_BACKEND_PATH	Override path resolution entirely

bxi-backend clean clears the registry but leaves installed files on disk.

Responses and errors

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

result = await services.executor.start_pipeline("run-1")

if result.success:
    print(result.data["state"])
else:
    print(result.error.message)

unwrap() raises ServiceOperationError instead:

state = (await services.executor.start_pipeline("run-1")).unwrap()["state"]

Development

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

pytest
mypy
ruff check src
python -m build

License

MIT