quicwire 0.1.0

Python SDK for Quicwire — stream arbitrary robot data over MoQ/QUIC with recovery, integrity, and observability.

Quicwire Python SDK

Stream arbitrary robot data over MoQ/QUIC with per-track delivery guarantees, loss recovery, byte-level integrity, and observability — a WebRTC alternative built for robot teleop and data collection.

You declare N tracks, each with its own delivery mode and priority, and publish opaque bytes (JSON, CBOR, protobuf, raw sensor data — anything). The SDK is pure stdlib; the transport ships as the publisher / ingest-bridge binaries.

Install

pip install quicwire
# the SDK needs the two Rust binaries on hand; point at them if not auto-found:
export QW_PUBLISHER_BIN=/path/to/publisher
export QW_BRIDGE_BIN=/path/to/ingest-bridge

Publish

import quicwire as qw

pub = qw.connect(
    relay_url="https://relay.example.com:443",     # one value — host + port
    tracks=[
        qw.Track("lidar",     mode="frames",   priority=3, rate_hz=10),
        qw.Track("telemetry", mode="datagram", priority=1, rate_hz=50),
    ],
)
pub.publish("telemetry", b"...your opaque bytes...")
pub.publish("lidar", jpeg_or_pointcloud_bytes)
pub.close()                       # or: with qw.connect(...) as pub: ...

Subscribe

import quicwire as qw

sub = qw.subscribe(
    relay_url="https://relay.example.com:443",
    tracks=["lidar", "telemetry"],
    forward_body=True,            # also deliver the opaque payload bytes
)

# iterator style
for obj in sub:
    print(obj.track, obj.object_id, obj.crc_ok, obj.latency_ms, obj.body)

# or callback style
@sub.on("telemetry")
def handle(obj):
    process(obj.body)

@sub.on("*")                      # all tracks
def audit(obj): ...

ReceivedObject fields: track, object_id, group_id, mode, tai_us, latency_ms, crc_ok, body, size, raw.

Multi-robot / namespaces (fleet)

Each robot publishes under its own namespace (a path — its address on a shared relay). One consumer subscribes across many namespaces over a single connection — that's fleet fan-in without an SFU.

# each robot
qw.connect(relay_url="https://relay:443", namespace="acme/warehouse-3/robot-007",
           tracks=[qw.Track("joints", mode="datagram")])

# one fleet consumer ← N robots, one session
sub = qw.subscribe(relay_url="https://relay:443", tracks=["joints"],
                   namespaces=["acme/warehouse-3/robot-007",
                               "acme/warehouse-3/robot-008", ...])
for obj in sub:
    print(obj.namespace, obj.track, obj.object_id)   # obj.namespace = which robot

Namespace defaults to "robo" (single-robot, back-compat). The hierarchical path (org/site/robot) is the unit for multi-tenancy + access control.

Video (codecs + hardware encode + adaptive bitrate)

Like WebRTC, Quicwire can encode video for you — declare a codec on the track and work in frames instead of bytes. The transport stays codec-agnostic; encoding lives in the SDK (pip install quicwire[media], which pulls FFmpeg via PyAV).

import quicwire as qw
from quicwire.abr import AdaptiveBitrate

cam = qw.Track("camera", mode="frames", priority=3, codec="vp9", rate_hz=30)
pub = qw.connect(relay_url="https://relay:443", tracks=[cam])

abr = AdaptiveBitrate(start_kbps=2000)             # WebRTC-style congestion control
while True:
    pub.publish_frame("camera", rgb_ndarray)        # encodes (VP9) and publishes
    pub.set_video_bitrate("camera", abr.update(loss=measured_loss, rtt_ms=measured_rtt))

# consumer — decoded frames, not bytes
sub = qw.subscribe(relay_url="https://relay:443", tracks=[qw.Track("camera", codec="vp9")])
@sub.on_frame("camera")
def show(track, frame):       # frame = decoded RGB ndarray
    cv2.imshow(track, frame[:, :, ::-1])

• Codecs: vp8, vp9, av1 (royalty-free), h264, hevc. Pass a hardware encoder name directly for offload — e.g. codec="h264_videotoolbox" (macOS) or "h264_nvenc" (NVIDIA/Jetson).
• Adaptive bitrate: AdaptiveBitrate adjusts target bitrate to loss + delay (Google-Congestion-Control-style); feed it observations and push the result into set_video_bitrate.

Delivery modes (per track)

mode	transport	use for	loss
datagram	unreliable QUIC datagram	high-rate telemetry, setpoints	dropped, latest-wins, recoverable from cache
reliable	ordered stream (rolling groups)	commands, events	retransmitted in order
frames	reliable, one group per object	camera frames, large messages	relay may drop a whole unit under congestion

datagram bodies must fit one packet (~1200 B). Larger → reliable / frames.

Configuration — nothing hardcoded

What	How	Default
relay endpoint	relay_url= or QW_RELAY_URL	https://localhost:4443 (dev)
recovery address	recovery_addr= or QW_RECOVERY_ADDR	derived: <relay-host>:5601
TLS verification	tls_verify=True	off (dev)
binaries	publisher_bin= / bridge_bin= or QW_PUBLISHER_BIN / QW_BRIDGE_BIN	auto-discovered

You typically set one value — relay_url — and everything else derives from it.

Ports & firewalls

• The robot/consumer only ever dials out to the relay — no inbound ports to open, no NAT traversal, no TURN (unlike WebRTC).
• For a hosted/production relay, run it on UDP 443 (QUIC/HTTP-3): traffic looks like HTTPS and passes most corporate firewalls. Pass any port via relay_url for self-hosting.

Wire format

Each object is [8-byte LE TAI µs][4-byte LE CRC32(body)][body]. The SDK frames and verifies for you; quicwire.wire.frame() / parse() are exposed for integrators working in raw bytes (or implementing it in another language). See docs/TRACKS_API.md.