synapse-runtime-sdk

Python SDK for Synapse: an asyncio API over the Rust client (synapse-client) through PyO3, plus the H.264 video path (synapse_sdk.media: encoder/decoder = the Rust synapse-media codec over our FFmpeg bundle, zero-copy pictures via the buffer protocol / numpy). Explanations for D: docs/explained_for_d/python_sdk.md.

The surface in one screen

from pydantic import BaseModel
from synapse_sdk import Synapse, Topic, Service, Api, ServiceError, SynapseError

class Imu(BaseModel):
    accel: tuple[float, float, float]

imu = Topic("imu", schema=Imu)                                   # datagram (default): one datagram per message, newest wins
log = Topic("log", delivery="stream")                             # ordered, nothing dropped
api = Api(arm=dict(get_pose=Service(input=dict, output=dict)))    # api.arm.get_pose.name == "arm/get_pose"

async with Synapse("https://localhost:4470/synapse", "agent", "sim-1") as node:   # type agent|observer|web|hmi; the name is minted when absent
    pub = await node.publisher(imu)                               # resolves once the runtime admitted it on the current session
    pub.publish(Imu(accel=(0, 0, 9.8)))                           # synchronous; validated by the schema; dropped+counted while unbound
    async for batch in node.subscribe(imu):                       # one read = the newest message of each publisher (queue_depth=0: every message)
        for msg in batch:
            print(msg.publisher, msg.data.accel, msg.t_us)
    view = node.view(imu, stale_time_ms=200); await view.ready(); print(view.get().data)
    reg = await node.serve(api.arm.get_pose, lambda inp, ctx: {"frame": inp["frame"]})   # (input, ctx) -> result; sync runs in a thread, async on the loop
    pose = await node.call(api.arm.get_pose, {"frame": "tool0"}, deadline_ms=1000)       # await the Run for the result
    run = node.call("arm/move_to", {"target": 1}); async for u in run.updates(): ...; await run.cancel()
    board = await node.store("mission/board", access="public"); await board.set("phase", "loading"); board.value["phase"]   # a value kept current
    @node.every(hz=50)                                            # a fixed grid while the node lives (polling bridges)
    def poll(): pub.publish(read_imu())

• Identity (WP §3.1): Synapse(url, type, name=None, user_id=None, credential=None). type bounds the role the runtime grants (agent → agent; web/hmi → observer until promoted on a project); a web/hmi node without user_id mints one. node.info() has name, prefix (client/<name>/<sid>), role, dom, broker_dom, broker_version.
• Sessions come and go, the node stays: a lost session is replaced (reconnect=True, backoff reconnect_initial_ms → reconnect_max_ms), every publisher, registration, subscription and namespace replays its intent on the new session; in-flight calls fail connection_lost; nothing is resumed. node.status / async for s in node.watch_status() / node.on_status(cb). reconnect=False ends the node closed on a drop.
• The runtime decides: publisher(), serve(), store() resolve on admission; a refusal (already_registered, unauthorized, owner_taken, ...) is asked again, paced, with a warning each time (warn= callback, default logging).
• Time: t_us is the publisher's clock, microseconds. node.clock(), node.time_offsets(), node.to_local_us(t_us, dom), node.age_us(msg); deadline_ms= on a subscription drops a message older than that in our own time domain.
• Errors: one SynapseError(code, message, data, retryable); ServiceError(code, message, data) raised by a handler crosses verbatim (data.app for an application code); the table is error_codes.json, shared with Rust and JS.
• Media: Topic(name, content_type="video/h264", fec=True, layout={...}) (raw bytes in FEC symbols); VideoPublisher.open(node, topic, encoder), VideoSubscriber(node, topic, decoder, on_image=); see below.
• The certificate pin (broker_cert_sha256=) is discovered when absent (SYNAPSE_BROKER_CERT_SHA256, then http://host:port/certificate.sha256). SIGINT/SIGTERM close every open node, then re-raise (signals=False opts out).
• Timeouts and periods are _ms integers everywhere (one vocabulary with Rust, JS and the wire).

Build and test (all Python through the uv project)

export CARGO_HOME=/media/data1/tmp/cargo-home
export VIRTUAL_ENV=/media/data1/phoenix_ws_2/python_work/.venv
cd /media/data1/phoenix_ws_2/synapse/py
uvx maturin develop --release --features media,test-hooks            # crates/synapse-py (feature `media` on) → synapse_sdk/_core.abi3.so; test-hooks for the reconnect tests
uv run --project /media/data1/phoenix_ws_2/python_work python -m pytest -q   # starts conformance/broker.sh if needed

The module links libavcodec from ../dependencies/ffmpeg-n8.1.1-11 through the rpath in .cargo/config.toml (check: ldd synapse_sdk/_core.abi3.so | grep libav). Workspace cargo test never links FFmpeg: the media feature is only enabled by pyproject.toml.

Wheel with the FFmpeg closure inside (synapse_sdk.libs/ + FFMPEG-LICENSE.txt, verified by ldd and an install test): py/build_wheel.sh [/media/data1/tmp/synapse_wheels].

Examples (conformance/broker.sh start first)

Getting started, one verb per file (≤ 15 lines each, defaults only, certificate discovered): examples/getting-started/python/01_connect.py … 11_store.py. The richer demos:

uv run --project /media/data1/phoenix_ws_2/python_work python examples/quickstart.py
uv run --project /media/data1/phoenix_ws_2/python_work python examples/fec_stream.py 5
uv run --project /media/data1/phoenix_ws_2/python_work python examples/media_relay.py demo/cam demo/cam-relay 10
uv run --project /media/data1/phoenix_ws_2/python_work python examples/mujoco_video.py video/mujoco0 10 --size 640x360 --backend auto [--threaded]
uv run --project /media/data1/phoenix_ws_2/python_work python examples/video_view.py video/mujoco0 10       # OpenCV window if cv2 is installed

Video in five lines

from synapse_sdk import Synapse, now_us
from synapse_sdk.media import H264Encoder, H264Decoder, VideoPublisher, VideoSubscriber

enc = H264Encoder(640, 360, 30, bitrate_kbps=2000, backend="auto")   # nvenc when a GPU is there, else x264
vp = await VideoPublisher.open(node, "cam/h264", enc)                # video/h264 datagram topic with fec + layout; serves `cam/h264/idr`
vp.push(i420_uint8_array, now_us())                                   # (h*3/2, w) numpy, or (y, u, v) planes; no copy
vs = VideoSubscriber(node, "cam/h264", H264Decoder(), on_image=lambda img: np.asarray(img.y))  # zero-copy planes