Metadata-Version: 2.4
Name: synapse-runtime-sdk
Version: 0.3.4
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"

node = Synapse("agent", "sim-1")                              # type agent|observer|web|hmi; the name is minted when absent; idle
await node.connect("https://localhost:4470")                  # bounded first connect; the newest connect() wins
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)
sub = node.subscribe(imu, on_imu)                             # the callback form: the SDK runs that loop, on_imu(batch) (def or async def)
view = node.view(imu, stale_time_ms=200); await view.ready(); print(view.get().data)   # the value now; changed() / watch() / on_change(cb)
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", step=4); board["phase"]   # a value kept current, read like a dict
@node.every(hz=50)                                            # a fixed grid while the node lives (polling bridges)
def poll(): pub.publish(read_imu())
await node.close()
```

- **Identity**: `Synapse(type="agent", name=None, *, user_id=None, credential=None, ...)`: who the node is, never where it
  connects. `type` bounds the role the runtime grants on the ladder `observer < agent < operator < hmi < admin` (`agent` →
  `agent`; `hmi` → `hmi` from admission, never promoted or demoted; `web` → `observer` until promoted to `operator` on a
  project); a `web`/`hmi` node without `user_id` mints one. The node is usable at once and `idle` until `connect()`.
  `node.info()` has `name`, `prefix`, `role`, `robot`, `runtime_id`, `broker_version`.
- **Connecting**: `await node.connect(url)`, the one connect verb: the first connect, and how an HMI moves from one
  robot's runtime to another (the current session is torn down first; every handle re-binds to the new one). `url` is the
  runtime, `https://host:4470`. Through a relay, give the relay with the runtime's path,
  `https://relay:4443/runtimes/<runtime_id>` (`runtime_id` from `node.info()` of a node connected directly, or the runtime's
  banner), and `token=` the relay token (a `jwt=` in the URL is taken as the token; neither is ever printed). `type=`,
  `name=`, `user_id=`, `credential=` given to `connect()` change the identity there; left out, the node's are kept.
  Operator intent is never lost under quick clicking: the newest call wins (a superseded `connect()` returns quietly);
  the target already declared is a no-op. With no url, `connect()` settles with the target already declared, else
  connects to `$SYNAPSE_ENDPOINT` (set by the runtime for every task it starts), else raises `invalid_input`.
  `await node.disconnect()` leaves the target: the node rests `idle` with every handle kept, until the next `connect()`.
  A name a live node already holds is refused `admission_refused` (retryable); no answer within `connect_timeout_ms` is
  `not_connected` (through a relay: check the token); a relay address without the runtime's path is `invalid_input`. The
  first connect tries a retryable failure again for up to 30 s (a hard bound), then raises it; the node rests `failed`,
  still usable. For cleanup on every exit path, `async with await Synapse(...).connect(url) as node:` (or `async with
  Synapse(...) as node:` in a task, which connects to `$SYNAPSE_ENDPOINT`).
- **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` is a state value, read like every handle:
  `node.status.get()`, `async for s in node.status.watch()`, `await node.status.changed()`, `node.status.on_change(cb)` (so are `node.catalog` and
  `node.project`). `reconnect=False` rests the node `failed` on a drop (not ended: `connect()` dials again).
- **The runtime decides**: `publisher()`, `serve()`, `store()`, `scene()` resolve on admission (made before the first
  session, or between two, they resolve on the next one); 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.now_us()`, `node.age_us(msg)` (how old a message
  is in our clock, with its error bound, or `None` when the clocks are not comparable); `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 (`connect(url, 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.

## Store

A namespace is a value the runtime keeps current: `{key: value}`, shared by every node of the active project. The handle
reads like a dict that changes under you, as `os.environ` does: each read is the value now. Writes go to the runtime and
are awaited, so they are verbs, never assignments (`board["phase"] = "x"` raises `TypeError` naming the verb).

```python
board = await node.store("mission/board", access="public")   # resolves once the value is current
board["phase"]; "phase" in board; len(board); board.items()  # the value now; board["nope"] raises KeyError
board.get("phase", "idle")                     # that key, or the default
snapshot = board.get()                         # the snapshot: the whole value in one read, a plain dict that stays as it was
await board.set(phase="loading", step=4)      # one atomic write; also set({"with space": 1}) and set("phase", "x")
await board.delete("step"); await board.clear()   # each resolves once a read shows it
async for value in board.watch(): ...          # the whole value after each change; watch("phase") for one key
value = await board.changed()                  # the whole value after its next change
off = board.on_change(lambda value: ...)       # the callback form: cb(value) after each change; off() removes it
board.synced                                   # current or not
board.status.get()                             # why: the handle's own state, a state value (connected, pending, mode, owner, ended, ...);
async for st in board.status.watch(): ...      # ... after each change; await board.status.changed() for the next
```

The runtime decides every write: `not_owner`, `unauthorized`, `store_gone` raise from the awaited verb. `dict(board)` also
copies, but key by key: while someone writes, it can mix two versions; `board.get()` keeps keys written together together.

## Scene

The active project's world: one set of entities (frames, robots, objects, cameras …) the runtime keeps for the project.
Writes go through two namespaces. `scene.authored` writes what exists — the hierarchy and the components — and every write
is one atomic batch the runtime checks. `scene.live` writes what is happening now — poses, joint positions and state of
entities, and live entities of this node's own — 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. Names are global:
one name, one entity, whichever namespace wrote it.

```python
from synapse_sdk import Kind                   # Kind.FRAME / ROBOT / MODEL / SCAN; every kind parameter also takes any string

scene = await node.scene()                     # the active project's scene; detached (empty) while no project is active
await asyncio.wait_for(scene.ready(), 10)     # until a project's scene is in
world = scene.get()                            # {id: entity}, read once to build a simulation from it
scene.tree(); scene.query(lambda e: e["kind"] == "robot")   # the value now; tree() = [{id, children}]

# a builder's one verb, keyed by name: spawned when no entity has it, else patched only where it differs (nothing sent
# when equal; live overrides never count); parent is for the spawn only; the name under another kind: the spawn is sent
# and the runtime answers (name_taken while the name is held)
tool = await scene.authored.upsert(Kind.FRAME, name="tool", parent=None, xform={"t": [0, 0, 1], "q": [0, 0, 0, 1]})
tool.id                                        # the entity's view comes back (by id); every entity argument takes an id or a view

# one entity as a view: what upsert returns, or by id or by name (a view by name follows the name to whichever entity holds it)
tool = scene.entity(name="tool")               # or scene.entity(id): merged, authored with its live overrides
tool["xform"]; "xform" in tool; len(tool)      # read like a dict over its components, now
tool.get("xform", None); tool.get()            # a component or the default; get(): the entity in one read (a copy), None when absent
tool.authored.get(); tool.live.get()           # one plane: the entity as declared; its live row ({entity, pose, joints, state, stamp})
await tool.authored.set(props={"speed": 2})    # one patch of the authored entity (also set({...}), set("props", {...}))
tool.live.set(xform={"t": [0, 0, 1.1], "q": [0, 0, 0, 1]}, joints={"j1": 0.3}, state={"badge": "on"}, lifetime_ms=500)   # its live lanes, synchronous
await tool.live.remove()                       # the release call: its live row gone, back to authored
async for xform in tool.watch("xform"): ...    # that component each time it changes (latest wins); watch() for the whole entity
landed = tool.changed()                        # the entity after its next change, armed at the call (arm it before the write)
scene.status.get()                             # a state value: state (attached | detached | closed), synced, pending, version, hash,
                                               # detached_drops, unclaimed_drops; status.watch() / status.changed(); scene.synced

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

# live: decisions are calls, data is streams
cup = await scene.live.upsert("detection", name="cup", parent=tool, lifetime_ms=500, props={"score": 0.93})   # the claim call: the view of the runtime's id
await scene.live.remove(cup)                   # the release call: an override back to authored; a live entity gone
scene.live.batch(poses=[{"id": arm, "t": [0, 0, 1], "q": [0, 0, 0, 1]}], joints=[{"id": arm, "j": {"shoulder": 0.3}}], lifetime_ms=500)   # a sim's tick: lanes only, synchronous
scene.live.batch(state=[{"id": arm, "s": {"badge": "serving"}}])

async for world in scene.watch(): ...          # the entities after each change (await scene.changed() for the next; scene.on_change(cb))
                                               # engine feeds: watch_live_removals(), watch_live_denied()
```

- An entity view reads like a dict that changes under you (as the store handle does); an entity that is not there (no
  entity holds the name, the id is gone, the scene is detached) reads as an empty mapping, never an error. Writes are
  verbs: `tool["xform"] = x` raises `TypeError` naming `await tool.authored.set(...)`. The view's two planes,
  `tool.authored` (`get`, `set`) and `tool.live` (`get`, `set`, `remove`), are one object each for the view's life and have
  no `watch`/`on_change`/`changed` of their own: the view's cover the entity, whichever plane changed.
- `view.live.set(...)` writes three lanes: `xform` (`{"t", "q"}`, both, nothing else), `joints`, `state`; anything else is
  `invalid_input`. The first lane write for an id this node has not claimed asks the claim (`upsert {id}`, a call) and the
  latest row per lane is held until the runtime answers, then sent: a single write lands. A refusal drops the held rows
  (counted in `status.get().unclaimed_drops`), is a node notice (`node.notices()`) and is asked again, paced, by a later
  write. `await scene.live.upsert(...)` is the claim call (`name` optional): it returns the view of the id the runtime
  answered (a name this node already claimed keeps its id; a taken name raises `name_taken {name, id, holder}`, another
  session's id `not_owner`, nothing kept). A granted claim is claimed again, same id, on every reconnect and project switch
  (a refusal asked again, paced, with a warning) until `await scene.live.remove(x)` or `await view.live.remove()`; closing
  the node's last scene handle releases them all. The id is never a component: `id=` on either `upsert` is `invalid_input`.
- Keyword names: `name` and `parent` (and `lifetime_ms` on `scene.live.upsert`) are parameters of `upsert`, so a component
  of that name cannot be given there by keyword; `await view.authored.set({...})` and `patch(e, {...})` take any component
  name.
- Two kinds of users. A **node author** (a driver, a sim, a task) needs `node.store()` with `get`/`set`/`watch`,
  `scene.authored.upsert()` keyed by name (its view back), `scene.entity(id)` / `scene.entity(name=...)` with
  `get`/`watch`/`authored.set`/`live.set`, `scene.live.upsert()` for a live entity of its own and `scene.live.batch()` for
  a sim's one write per tick. An **engine or editor builder** adds `scene.authored.spawn`/`patch`/`rename`/`reparent`/
  `despawn`/`set_locked`, `txn`/`apply`, one plane alone (`view.authored.get()`, `view.live.get()`) and the engine feeds
  (`watch_live_removals`, `watch_live_denied`). Every handle (store, scene) has one shape: `get()`, `ready()`, `status` (a
  state value: `get()`, `async for … in watch()`, `changed()`, `on_change(cb)`), `synced`, `watch()`, `changed()`, `on_change(cb)`,
  `close()`.
- 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 session that claimed an id holds it; another session's claim
  of it is refused `not_owner` (a node notice, and its ids on `watch_live_denied()`). A name another entity holds is refused
  `name_taken`, its `data["holder"]` the holder's id.
- While detached, authored verbs raise `project_inactive` and lane writes are dropped (counted in `status.get().detached_drops`).
  A closed handle reads nothing (`get()` is `{}`, views are empty, `tree()` is `[]`) and its verbs raise `not_connected`.
- 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.authored.txn() as t:          # when a scene is loaded: no poses, so nothing is spawned or moved
    report = materialize_frames(t, robot, scene.get())
async with scene.authored.txn() as t:          # after an edit of the robot: the rest poses are your FK at its joints
    report = materialize_frames(t, robot, scene.get(), {"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.authored.txn() as t:          # hardpoint frames take their pose from the hardpoint itself
    materialize_hardpoints(t, entity, scene.get())
```

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.

## Tasks

A task is a program the runtime starts and keeps running for you (a driver, a simulator, a planner). You declare
which tasks to run; an operator runs and stops them. There are two kinds, and stopping one kind never touches the other:

- **Robot tasks** belong to the robot and keep running when the active project changes (an arm driver). The runtime
  keeps their declarations across restarts.
- **Project tasks** belong to the active project: the `tasks` component of its scene's entities (one component; it
  replaces the prototype's `behaviors`). Switching projects stops them.

A declaration is `{ref, enabled?, config?, schema?}` under an instance name: `ref` is the task's address (as the task
tool lists it), `enabled: False` leaves it out of the next run, `config` is passed to the task, and `schema` pins the
task's schema hash (a different one fails the run). Whether a task has a process to run is the task's own spec: one
with nothing to run is *passive*, counted and never started (something else uses its config, a simulator say).

Everything is under `node.tasks`, the same object for the node's life:

```python
t = node.tasks

# robot tasks (operator role)
cam = await t.declarations.set({"ref": "python:camera:driver", "config": {"fps": 30}})   # the SDK mints the instance; returned
await t.declarations.set("arm", {"ref": "python:ur-driver:driver", "config": {"host": "10.0.0.5"}})   # or a stable key of yours
await t.declarations.set("arm", {**t.declarations["arm"], "enabled": False})         # off for the next run

# project tasks: the `tasks` component of a scene entity
scene = await node.scene()
await scene.authored.patch(sim_id, tasks={"sim": {"ref": "python:mujoco-sim:sim"}})

r = await t.run_project_tasks()        # {"launched": 2, "passive": 3}
await t.stop_project_tasks()           # None, also when nothing ran
await t.run_robot_tasks()
await t.stop_robot_tasks()

await t.status.ready()                                          # the value is in
t.status["project"]["state"]                                    # "starting" | "running" | "stopping" | "stopped" | None (not said yet)
t.status["project"]["tasks"].get(entity_id, {}).get("sim")      # {"scope", "addr", "instance", "entity", "state": "up" | "down", "since", "restarts", "exit"?, "log"}
t.status.get()                                                  # the whole status: {"robot": ..., "project": ...}
t.catalog["python:ur-driver:driver"]                            # {"addr", "workspace", "pkg"?, "name"?, "inputs"?, "schemaHash"?, ...}

async for s in t.status.watch():                                # the whole status after each change
    arm = s["robot"]["tasks"].get("arm")
    if arm and arm["state"] == "down":
        print(f"arm exited {arm.get('exit')}, {arm['restarts']} restarts; log {arm['log']}")
```

- Each of `declarations`, `status` and `catalog` reads like a dict, as a store handle does (`t.status["robot"]`,
  `"arm" in t.declarations`), beside `get()` (the whole value in one read). `declarations` writes with awaited verbs (`set`,
  `delete`, `clear`; also `synced`, `status`, `watch`, `on_change`, `ready`); a write waits for the value to be current, then resolves
  once a read shows it; `set(declaration)` mints the instance (a ulid) and returns it, `set(instance, declaration)` writes
  the one named, `set({instance: declaration, ...})` several at once (one mapping is a declaration when its `ref` is a
  string; the same program may be declared many times). `status` and `catalog` are read-only: the runtime writes them
  (`synced`, `ready()`, `watch()`, `on_change(cb)`; item assignment raises `TypeError`). Each opens the first time you touch it; the verbs open nothing.
- `status.get()` is `{"robot": {"state", "since"?, "reason"?, "tasks": {instance: task}}, "project": {..., "tasks":
  {entity_id: {instance: task}}}}`. `reason` says why a run stopped or failed (`project_switch`, or an error code).
  `exit` is `{"code", "at"}`, `{"signal", "at"}` or `{"at"}` (Unix ms). A part that did not change keeps its object
  (`is` tells what changed). `watch()` is state, not history: a slow reader sees the latest status, and `restarts` is
  the count to trust.
- A run checks every enabled declaration first and starts nothing unless all pass. A failed check raises
  `preflight_failed`, a task that could not be started `launch_failed` (what had started is stopped); both carry
  `e.data["report"]`, one entry per declaration at fault: `{scope, entity?, entity_name?, instance, addr, reason, detail?}`
  with `reason` one of `missing` (unknown address), `schema_mismatch`, `spec_error`, `invalid_input` (the config);
  `is_tasks_refusal(e)` tells them apart. A second run while one is running raises `already_running`; a project run
  with no active project `project_inactive`. A run's reply is the runtime's answer; what runs afterwards is `status`.
- A task that exits is started again after a delay that grows (by default from 1 s up to 10 s) while it keeps exiting.
  A crash loop never ends the run; `restarts` climbs.
- Running and stopping need the operator role or above: an `hmi` node, or a `web` node promoted on the active project.
  Robot tasks need only that role; project tasks also need an active project (`project_inactive` otherwise).

## Writing a task

**A task owes the runtime nothing.** A task is any program the task tool can describe. The runtime asks the tool for
the task's spec at every run and starts the spec's command directly; your declaration's `config` reaches it as the
spec's flags, positionals or environment variables. The program never has to connect, report or handshake. What the
runtime does to it:

- **Environment**: cleared, then exactly the spec's `env`, then the stamps below. There is no `PATH` or `HOME` unless
  the spec sets them. Working directory: the spec's `cwd`. No stdin.
- **Output**: stdout and stderr are appended to a log file, with one header line per start; its path is `log` in the
  status. Python buffers stdout when it is a file: set `PYTHONUNBUFFERED=1` in the spec's env, or `print(..., flush=True)`.
- **Stop**: SIGTERM to the program's process group, then SIGKILL 5 s later. Handle SIGTERM and leave within 5 s.
- **Exit**: any exit is followed by a new start after a delay (1 s doubling to 10 s, back to 1 s once a run lasted
  longer than 10 s); `restarts` and `exit` in the status show it.
- **Runtime gone**: the program gets SIGTERM.

The stamps (optional to read):

| variable | value |
|---|---|
| `SYNAPSE_ENDPOINT` | the runtime's URL; `await node.connect()` with no url connects to it |
| `SYNAPSE_ROBOT` | the robot the runtime serves |
| `SYNAPSE_INSTANCE` | the declaration's instance name |
| `SYNAPSE_ADDR` | the task's address (`ref`) |
| `SYNAPSE_CONFIG` | the declaration's `config` as JSON (`{}` when none) |
| `SYNAPSE_ENTITY`, `SYNAPSE_ENTITY_NAME` | project tasks only: the entity's id and name |

A plain task (no SDK at all):

```python
# serve.py; its spec: argv ["python3", "serve.py"], mapping port → {as: "flag", flag: "--port"}
import argparse, http.server, signal, sys
p = argparse.ArgumentParser(); p.add_argument("--port", type=int, default=8080); a = p.parse_args()
srv = http.server.ThreadingHTTPServer(("0.0.0.0", a.port), http.server.SimpleHTTPRequestHandler)
signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))          # the runtime stops us with SIGTERM
srv.serve_forever()
```

A task that talks to the runtime (a simulator's shape):

```python
import asyncio, json, os, signal
from synapse_sdk import Synapse, Topic
from synapse_sim.util import JointStates                         # the simulator's own message model

config = json.loads(os.environ.get("SYNAPSE_CONFIG", "{}"))     # or flags from the spec's mapping

async def main():
    stop = asyncio.Event()
    for s in (signal.SIGINT, signal.SIGTERM):                    # our own graceful path; the SDK leaves an installed handler alone
        asyncio.get_running_loop().add_signal_handler(s, stop.set)
    node = Synapse(type="agent", name="sim-server-1")
    await node.connect()                                                     # no url: $SYNAPSE_ENDPOINT
    state = await node.store("mujoco/state", access="protected")             # ours alone; a second sim waits
    scene = await node.scene(); await asyncio.wait_for(scene.ready(), 10)
    joints = await node.publisher(Topic("ur5e/joint_states", schema=JointStates))   # once; publish() is sync
    async def control(req, ctx):                                             # (input, ctx); ctx.update() is sync
        ...; return {"result": "stopped"}
    await node.serve("mujoco/control", control)
    await state.set({"simulation_status": "stopped", "sim_time": 0.0})       # one atomic write, resolves when visible
    # loop: joints.publish(...); scene.live.batch(joints=[...], poses=[...], lifetime_ms=2000)
    await stop.wait()
    await node.close()

asyncio.run(main())
```

A name is not taken from `SYNAPSE_INSTANCE` for you: two entities with an instance `sim` would collide. Give each
task's node its own name (from its config, say).

## Examples

Getting started, one verb per file: `examples/getting-started/python/01_connect.py` … `14_live.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.

