Metadata-Version: 2.4
Name: synapse-runtime-sdk
Version: 0.3.2
Classifier: Programming Language :: Rust
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Framework :: AsyncIO
Requires-Dist: msgpack>=1.0
Requires-Dist: pydantic>=2.5
Requires-Dist: pytest>=8 ; extra == 'test'
Requires-Dist: pytest-asyncio>=0.24 ; extra == 'test'
Requires-Dist: pytest-timeout>=2.3 ; extra == 'test'
Requires-Dist: numpy>=1.26 ; extra == 'test'
Requires-Dist: synapse-runtime-sdk-video==0.3.0 ; extra == 'video'
Requires-Dist: numpy>=1.26 ; extra == 'video'
Provides-Extra: test
Provides-Extra: video
Summary: Python SDK for the Synapse runtime: topics, services, store and H.264 video from asyncio
License: UNLICENSED
Requires-Python: >=3.11
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM

# synapse-runtime-sdk

The Python SDK for the Synapse runtime: publish and subscribe to topics, serve and call services, keep shared
values current with the store, and move H.264 video — all from asyncio.

```sh
pip install synapse-runtime-sdk              # the SDK
pip install "synapse-runtime-sdk[video]"     # + the H.264 encoder/decoder (CPU or NVIDIA)
```

Linux x86_64, Python ≥ 3.11. A Synapse runtime to connect to: `curl -fsSL <release host>/install.sh | bash`.

## The surface in one screen

```python
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**: `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`, `role`, `robot`, `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 same error codes in every language.
- **Video**: `cam = await node.video_publisher("gs/video")` then `cam.publish(image)`; `async for image in
  node.video_subscriber("gs/video")`; see below. A video topic is `Topic(name, content_type="video/h264")`: datagrams with
  loss repair unless `fec=` says otherwise; `pub.publish(access_unit)` on it sends one H.264 access unit, and a subscriber
  that names the topic gets bytes with `msg.key` / `msg.config` / `msg.recovery`.
- **Bodies follow the topic**: a subscription decodes what the runtime says the topic carries — msgpack topics give values
  (validated by your schema when the `Topic` has one), any other content type gives bytes.
- 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, in every language.

## Scene

The active project's world: one set of entities (frames, robots, objects, cameras …) the runtime keeps for the project,
in two planes. The **authored** plane is what exists — the hierarchy and the components — and every write to it is one
atomic batch the runtime checks. The **live** plane is what is happening now — poses, joint positions and state of
entities, written many times a second, never saved. What you read is both merged: a live pose replaces `xform.t`/`xform.q`
of the entity it names; removing it brings the authored value back.

```python
scene = await node.scene()                     # the active project's scene; detached (empty) while no project is active
await scene.wait_attached(timeout=10)          # until a project's scene is in
world = scene.entities                         # {id: entity}, read once to build a simulation from it
scene.entity(id); scene.tree(); scene.query(lambda e: e["kind"] == "robot")   # the value now; tree() = [{id, children}]
scene.status                                   # state (attached | detached | closed), synced, pending, version, hash, detached_drops

# author: each resolves once scene.entities shows it; a refusal (locked, name_taken, invalid_tree, ...) raises
id = await scene.spawn("box", parent=None, name="crate", xform={"t": [1, 0, 0]}, visual=[{"kind": "box", "size": [1, 1, 1]}])
await scene.patch(id, props={"mass": 2}); await scene.rename(id, "crate-1"); await scene.reparent(id, other, before=sibling)
await scene.set_locked(id, True); await scene.despawn(id, cascade=True)      # cascade: with its subtree
async with scene.txn() as t:                   # one atomic batch when the block ends
    a = t.spawn("frame", name="a"); t.patch(a, props={"k": 1})

# live: synchronous, like publish — one call per tick
scene.live(poses=[{"id": arm, "t": [0, 0, 1], "q": [0, 0, 0, 1]}], joints=[{"id": arm, "j": {"shoulder": 0.3}}], lifetime_ms=500)
scene.live(state=[{"id": arm, "s": {"badge": "serving"}}]); scene.live(remove=[arm])   # remove: back to authored

async for ops in scene.watch_changes(): ...    # authored changes; also watch_live_removals(), watch_live_denied(), watch_state()
```

- A live row with `lifetime_ms` goes away that long after its last write (`timeout` on `watch_live_removals()`), and when its
  writer leaves; one without it stays (a parked robot). The first writer of an id owns its live rows; another writer's rows
  for it are dropped and reported on its `watch_live_denied()`.
- While detached, author verbs raise `project_inactive` and `live()` drops its batch (counted in `status.detached_drops`).
- Helpers: `ulid()` (a fresh entity id), `order_between/order_before/order_after` (sibling order values),
  `derived_frame_name(parent, frame)`.

### Derived frames

A robot's `robot.frames` (names of links) and an entity's `hardpoints` (named mounting points with a pose) each stand for
`frame` children of that entity, marked `flags.derived`. `materialize_frames` and `materialize_hardpoints` keep those
children in step with the declaration: they spawn what is missing, rename the frames after the entity is renamed, set a
changed pose, and remove a frame that is no longer declared (refused, as a veto, when something is parented under it).
They only add ops to a txn and return a report of what they did; nothing is sent until the block ends.

```python
from synapse_sdk import materialize_frames, materialize_hardpoints

async with scene.txn() as t:                   # when a scene is loaded: no poses, so nothing is spawned or moved
    report = materialize_frames(t, robot, scene.entities)
async with scene.txn() as t:                   # after an edit of the robot: the rest poses are your FK at its joints
    report = materialize_frames(t, robot, scene.entities, {"tool0": {"t": [0, 0, 1.2], "q": [0, 0, 0, 1]}})
    if not report.clean:                       # raising here sends nothing
        raise RuntimeError("; ".join(report.errors + report.vetoes))
async with scene.txn() as t:                   # hardpoint frames take their pose from the hardpoint itself
    materialize_hardpoints(t, entity, scene.entities)
```

The report: `frames` (each declared name → its frame's id), `spawned`, `updated`, `removed`, `vetoes`, `errors`, and
`clean` (no vetoes, no errors). A declared frame that does not exist yet and has no pose given is an error, never a spawn.

## Examples

Getting started, one verb per file: `examples/getting-started/python/01_connect.py` … `12_scene.py` in the repository.

## Video

Needs the extra: `pip install "synapse-runtime-sdk[video]"` (without it the two calls below raise `MediaError` saying so).

```python
cam = await node.video_publisher("gs/video")       # resolves once the topic is admitted
cam.publish(image)                                  # numpy uint8 (h, w, 3) RGB: convert + encode + publish in one call
cam.publish(image, t_us=capture_us)                 # the capture time, if you have one (default: now)
cam.close()

async for image in node.video_subscriber("gs/video"):   # decoded pictures
    print(image.width, image.height, image.t_us)
    frame = image_to_numpy(image)                       # (h*3/2, w) I420; np.asarray(image.y) is the luma plane, no copy
node.video_subscriber("gs/video", on_image=show)       # or a callback per picture (def or async def)
```

- `publish` is synchronous and queues nothing: the caller waits for one conversion + encode (about a millisecond at
  640x360). The picture's format comes from its shape — `(h, w, 3)` RGB, `(h, w, 4)` RGBA, `(h*3/2, w)` I420, a tuple of
  planes `(y, u, v)` (I420) or `(y, uv)` (NV12); `pixel_format="bgr24"` for OpenCV pictures.
- Options of `video_publisher`, all optional: `pixel_format`, `fps` (nominal, for rate control; 30), `bitrate_kbps`
  (2000), `backend` (`"auto"`: NVIDIA when present, else CPU).
- One publisher, one size: the encoder is made from the first picture; a picture of another size raises
  `SynapseError("invalid_input")`.
- A subscriber that joins mid-stream starts at the next picture it can decode, and asks the publisher for one when it
  waits too long; `close()` stops it.
- The encoder and decoder are also there on their own (`H264Encoder`, `H264Decoder`, `TestPattern` in
  `synapse_sdk.media`) for a pipeline of your own.

