Metadata-Version: 2.4
Name: anthriq-services
Version: 0.1.0
Summary: Python SDK for the BXI backend services — pipeline execution, operator and pipeline registries, recording and playback
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,bxi,executor,operator-registry,pipeline-registry,recording,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-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`](../anthriq-bxi-interface).

## Install

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

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

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

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

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

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

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

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

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

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

```python
from anthriq_services import get_endpoint_info
print(get_endpoint_info())
```

## Spawning servers

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

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

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

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

`unwrap()` raises `ServiceOperationError` instead:

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

## Development

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

pytest
mypy
ruff check src
python -m build
```

## License

MIT
