lotse-client

Async Python client for the control API of lotse, the media daemon for RTSP cameras and WebRTC viewers. It speaks the API over the daemon's Unix socket with aiohttp; the payload types are frozen dataclasses decoded with mashumaro.

from lotse_client import LotseClient
from lotse_client.models import SourceSpec

async with LotseClient("/run/lotse/lotse.sock") as client:
    await client.stream.put(stream_id="front", sources=[SourceSpec(url="rtsp://cam/h264")])
    async with await client.stream.subscribe(stream_id="front") as events:
        async for event in events:
            print(event)

Every command is a keyword-only method named after it: stream/put is client.stream.put(...), session/close is client.session.close(...), and a command without a slash (ping, info, schema, unsubscribe) is a method of the client. A command that subscribes (stream.subscribe, webrtc.offer, session.adopt) returns a Subscription: an async iterator of its events, which ends after a session's closed event or on unsubscribe. A failed command raises CommandError with the stable error code.

Generated code

src/lotse_client/models.py is generated by datamodel-code-generator from the daemon's JSON Schema bundle (../schema/api.json, copied from lotse's crates/lotse-api-types/schema/api.json, which is itself generated from the Rust types); its options are in pyproject.toml. src/lotse_client/_commands.py is then generated by scripts/generate_commands.py from the bundle and those models: each command's fields and their types come from its generated command class, its result type and whether it subscribes from the bundle's results and events. Do not edit either: change the Rust types in lotse, then copy the bundle and regenerate with

mise run schema    # from the repository root; or, here,
uv run datamodel-codegen && uv run python scripts/generate_commands.py

The python-client hook fails when either is stale, and tests/test_commands.py checks the generator's output and what it refuses.

Development

uv sync
uv run pytest       # contract tests too when LOTSE_BIN names a lotse binary built with source-fake (`mise run contract`)
uv run ruff check && uv run ruff format --check && uv run mypy