From 661ae4258feb8ff4c6c3cdfd2c33521d18114c97 Mon Sep 17 00:00:00 2001 From: nityam Date: Thu, 10 Sep 2026 19:32:47 +0530 Subject: [PATCH 1/3] mediad: a websocket for programs, next to the datachannel for people --- docs/design/architecture.md | 5 + docs/design/remote-webrtc.md | 10 +- mediad/Cargo.toml | 2 +- mediad/src/agent.rs | 229 +++++++++++++++++++++++++++++++++++ mediad/src/lib.rs | 3 + mediad/src/main.rs | 14 ++- mediad/src/route.rs | 6 +- mediad/src/session.rs | 5 +- mediad/src/web.rs | 28 +++-- 9 files changed, 284 insertions(+), 18 deletions(-) create mode 100644 mediad/src/agent.rs diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 1b2c1e1b..9266072d 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -422,6 +422,11 @@ Same API behind all of them. "Run an LLM on a server that controls the robot" becomes: open a WebSocket, poll a frame, send intents — a few dozen lines, no media stack. That is what makes it genuinely easy. +`mediad::agent` is that socket, at `/agent` on the port the console is already +served from. Same API is not a promise here, it is the implementation: a line +arriving there runs through `session::run` against `route`, which is the same +dispatcher and the same table the datachannel uses. + Note also that LLM latency (hundreds of ms to seconds) means the agent is a **high-level** controller: "go to the kitchen", "look at the person". Reactive control stays local in `robotd`. This is the correct split regardless of diff --git a/docs/design/remote-webrtc.md b/docs/design/remote-webrtc.md index d1b1b1f9..c8f4a9bd 100644 --- a/docs/design/remote-webrtc.md +++ b/docs/design/remote-webrtc.md @@ -571,10 +571,12 @@ along with a transport. `remote-access-design.md` §9 carries it as open. ## 12. Deferred, with reasons -- **A WebSocket surface for server-side programs** (`architecture.md` §5.3). Same JSON-RPC, no - media stack, `get_frame` returning a JPEG. It is a few dozen lines once §5's routing exists, and - it is what makes "an LLM drives the robot" easy — but it is a second transport and the first one - should work. +- ~~**A WebSocket surface for server-side programs**~~ — built. `mediad::agent` serves it at + `/agent` on the console's own listener, and it was the few dozen lines this predicted: §5's + routing already existed, and `session::run` never knew what carried its lines, so the whole of + the transport is pumping text frames into one channel and out of the other. What an agent may + call is what a console peer may call, decided in `route` for both — a method is routed once, for + every transport, or refused everywhere. `refusal` stopped naming WebRTC for the same reason. - **The `teleop` datachannel.** Not the near-term priority; §6 covers what deferring it removes, what it costs in the meantime, and the sequence numbers it will need. - **Multi-peer video.** One media session at a time, plus control-only clients. Simulcast and diff --git a/mediad/Cargo.toml b/mediad/Cargo.toml index 0581fdc3..5712853d 100644 --- a/mediad/Cargo.toml +++ b/mediad/Cargo.toml @@ -39,7 +39,7 @@ image = { version = "0.25", default-features = false, features = ["jpeg"] } # # Default features off: this serves one file over GET, and JSON extractors, multipart, forms and # tracing middleware are all weight for routes that do not exist. -axum = { version = "0.8.9", default-features = false, features = ["http1", "tokio"] } +axum = { version = "0.8.9", default-features = false, features = ["http1", "tokio", "ws"] } clap = { workspace = true, features = ["derive"] } # The relay's half of the bridge: HTTPS to the rendezvous service — SSE inbound, `POST /send` # outbound. Same line `updater` uses, minus `form` (nothing here posts a form), so the TLS stack diff --git a/mediad/src/agent.rs b/mediad/src/agent.rs new file mode 100644 index 00000000..f4153cce --- /dev/null +++ b/mediad/src/agent.rs @@ -0,0 +1,229 @@ +//! A WebSocket for programs, next to the WebRTC console for people. +//! +//! `architecture.md` §5.3 argues that an LLM-driven controller should not be pushed through +//! WebRTC: an agent does not want a 30 fps H.264 track to decode, it wants a frame every second or +//! two and a state blob, and making it do ICE and DTLS and a decode pipeline first is a poor +//! trade. What it wants is "open a socket, poll a frame, send intents". That is this. +//! +//! **It is the same API, not a second one.** A line arriving here goes through +//! [`crate::session::run`] — the same dispatcher the datachannel uses, against the same +//! [`crate::route`] table — so what an agent may call is what a console peer may call, decided in +//! one place. Adding a method here is not a thing anybody can do; a method is routed once, for +//! every transport, or it is refused everywhere. +//! +//! That is also why this is small. `session::run` takes a pair of `String` channels and knows +//! nothing about what carries them, so the whole of the transport is pumping text frames into one +//! and out of the other. + +use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; +use axum::response::Response; +use futures_util::{SinkExt, StreamExt}; +use tokio::sync::{mpsc, watch}; + +use crate::session::{self, Media}; +use crate::upstream::{Pool, Sockets}; + +/// Lines buffered in each direction before a slow peer starts costing something. +/// +/// An agent's own calls are request-response and never queue, so this is really the size of a +/// subscription's backlog — `robot.state` at the control rate is the fast one. Deep enough to ride +/// out a garbage collection at the other end, shallow enough that a peer which has stopped reading +/// is noticed rather than buffered for ever. +const QUEUE: usize = 256; + +/// What an agent connection needs to serve: where the daemons are, and the media that may not +/// exist yet. +/// +/// The media arrives late by construction — the console is served before the pipeline is built, +/// because a page that cannot bind must not cost the video — so it comes as a watch rather than a +/// value. A connection reads it once, at accept, which is also when a `None` is honest: no +/// pipeline, no frames to hand out. +#[derive(Clone)] +pub struct State { + pub sockets: Sockets, + pub video: watch::Receiver>, +} + +/// The upgrade handler, for the router to hang on a path. +pub async fn upgrade(ws: WebSocketUpgrade, state: State) -> Response { + ws.on_upgrade(move |socket| serve(socket, state)) +} + +/// One agent, until it goes away. +async fn serve(socket: WebSocket, state: State) { + let (mut sink, mut stream) = socket.split(); + let (inbound, inbound_rx) = mpsc::channel::(QUEUE); + let (outbound, mut outbound_rx) = mpsc::channel::(QUEUE); + + // Its own connections to the daemons, per peer rather than shared, for the reason the + // datachannel's are: one peer's minutes-long update must not silence another's telemetry. + let pool = Pool::new(state.sockets.clone(), outbound.clone()); + let media = state.video.borrow().clone(); + if media.is_none() { + // Worth saying once. Every control call still works; `media.*` is what will refuse, and + // "the camera answers nothing" is otherwise a silent property of having connected early. + tracing::debug!("an agent connected before the pipeline; frames are not available yet"); + } + + let writer = tokio::spawn(async move { + while let Some(line) = outbound_rx.recv().await { + if sink.send(Message::Text(line.into())).await.is_err() { + break; + } + } + }); + + let session = tokio::spawn(session::run(inbound_rx, outbound, pool, media)); + + while let Some(message) = stream.next().await { + match message { + // One JSON-RPC object per message, which is what a WebSocket already frames for us — + // so unlike every other transport here there is no newline to reassemble. + Ok(Message::Text(text)) => { + if inbound.send(text.to_string()).await.is_err() { + break; + } + } + Ok(Message::Close(_)) => break, + // Ping and pong are the runtime's; binary is not something this speaks. + Ok(_) => {} + Err(e) => { + tracing::debug!(error = %e, "an agent's socket ended"); + break; + } + } + } + + // Dropping `inbound` ends `session::run`, which drops the last `outbound` and ends the writer. + drop(inbound); + let _ = session.await; + writer.abort(); + tracing::debug!("agent disconnected"); +} + +#[cfg(test)] +mod tests { + use super::*; + use duck_ipc_proto as proto; + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + + /// A daemon that reads one line and answers with a canned one, so a call that is *routed* can + /// be told from one that is merely accepted. + fn fake_daemon(path: std::path::PathBuf, reply: String) { + let listener = std::os::unix::net::UnixListener::bind(&path).unwrap(); + listener.set_nonblocking(true).unwrap(); + let listener = tokio::net::UnixListener::from_std(listener).unwrap(); + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let reply = reply.clone(); + tokio::spawn(async move { + let (read, mut write) = stream.into_split(); + let mut lines = BufReader::new(read).lines(); + while let Ok(Some(_)) = lines.next_line().await { + let _ = write.write_all(format!("{reply}\n").as_bytes()).await; + let _ = write.flush().await; + } + }); + } + }); + } + + /// Serve the real router on an ephemeral port and open a real WebSocket to it. + async fn connect( + dir: &std::path::Path, + ) -> tokio_tungstenite::WebSocketStream> + { + let sockets = Sockets { + robot: dir.join("robotd.sock"), + updater: dir.join("updaterd.sock"), + config: dir.join("configd.sock"), + ..Sockets::default() + }; + + let (_video_tx, video) = watch::channel(None); + let state = State { sockets, video }; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let router = axum::Router::new().route( + "/agent", + axum::routing::get(move |ws| upgrade(ws, state.clone())), + ); + tokio::spawn(async move { + let _ = axum::serve(listener, router).await; + }); + + let (socket, _) = tokio_tungstenite::connect_async(format!("ws://127.0.0.1:{port}/agent")) + .await + .expect("the agent socket accepts a websocket"); + socket + } + + async fn ask( + socket: &mut tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, + >, + call: &proto::Call, + ) -> serde_json::Value { + let request = proto::Request::call(proto::Id::Number(1), call); + socket + .send(tokio_tungstenite::tungstenite::Message::Text( + serde_json::to_string(&request).unwrap().into(), + )) + .await + .unwrap(); + let answer = socket.next().await.unwrap().unwrap(); + serde_json::from_str(answer.to_text().unwrap()).unwrap() + } + + /// **The point of the whole module: a program gets the surface a person gets.** + /// + /// `hello` is answered, a call `route` permits reaches the daemon that owns it, and one it + /// refuses is refused here too — all through `session::run`, which is what makes "same API + /// behind all of them" true rather than a thing to keep in step by hand. + #[tokio::test] + async fn an_agent_gets_the_same_surface_a_console_peer_gets() { + let dir = tempfile::tempdir().unwrap(); + fake_daemon( + dir.path().join("robotd.sock"), + r#"{"jsonrpc":"2.0","id":1,"result":{"healthy":true}}"#.to_owned(), + ); + // `hello` belongs to updaterd, like it does on every other transport. + fake_daemon( + dir.path().join("updaterd.sock"), + format!( + r#"{{"jsonrpc":"2.0","id":1,"result":{{"api_version":{}}}}}"#, + proto::API_VERSION + ), + ); + let mut socket = connect(dir.path()).await; + + let hello = ask( + &mut socket, + &proto::Call::Hello(proto::HelloParams { + api_version: proto::API_VERSION, + }), + ) + .await; + assert_eq!( + hello["result"]["api_version"], + proto::API_VERSION, + "a version handshake must work before anything else: {hello}" + ); + + let health = ask(&mut socket, &proto::Call::RobotHealth).await; + assert_eq!( + health["result"]["healthy"], true, + "a permitted call must reach robotd: {health}" + ); + + // Refused over WebRTC, so refused here: the PIN authorises a phone, and a transport that + // hands it out is a transport that hands out the robot. + let pin = ask(&mut socket, &proto::Call::SystemPairingPin).await; + assert_eq!( + pin["error"]["code"], + proto::code::METHOD_NOT_FOUND, + "the agent socket must not widen what a transport may call: {pin}" + ); + } +} diff --git a/mediad/src/lib.rs b/mediad/src/lib.rs index 4c0bea8f..0f47148e 100644 --- a/mediad/src/lib.rs +++ b/mediad/src/lib.rs @@ -17,6 +17,9 @@ //! signalling server in this process, `mpph264enc` in front of it, and a `control` datachannel per //! peer wired to [`session::run`]. +/// The WebSocket a program drives the robot over, as opposed to the datachannel a person does. +/// `architecture.md` §5.3 has the argument; the surface is the same one either way. +pub mod agent; /// What the camera's geometry is — the intrinsics a consumer needs to turn pixels into /// directions, and which sensor mode they belong to. pub mod camera; diff --git a/mediad/src/main.rs b/mediad/src/main.rs index 448c9e42..c5dd7829 100644 --- a/mediad/src/main.rs +++ b/mediad/src/main.rs @@ -276,10 +276,19 @@ fn main() -> ExitCode { // a port already in use, which `Restart=always` cannot fix by trying again; a robot that // streams and answers control calls with no console is much better than one that does // neither. So this is logged at error and the daemon carries on. + // Made here rather than beside the relay, because the console's listener now carries the + // agent socket too and both want the same late-arriving media. + let (video_tx, video_rx) = + tokio::sync::watch::channel::>(None); + let page = mediad::web::page(args.port); let (web_host, web_port) = (args.host.clone(), args.web_port); + let agent = mediad::agent::State { + sockets: args.sockets(), + video: video_rx.clone(), + }; tokio::spawn(async move { - if let Err(e) = mediad::web::serve(&web_host, web_port, page).await { + if let Err(e) = mediad::web::serve(&web_host, web_port, page, agent).await { tracing::error!( error = %format!("{e:#}"), "the console is not being served; video and control are unaffected" @@ -317,9 +326,6 @@ fn main() -> ExitCode { // once something has tried to set it, and there are no frames to encode before then — and // the relay below is spawned before that on purpose, so the answer has to be able to // arrive late rather than be a value passed in now. - let (video_tx, video_rx) = - tokio::sync::watch::channel::>(None); - // The outward half of remote access, and it is deliberately *after* the producer is // learned: the name a client sees in the service's listing comes from the same place the // local `meta` gets it, and a relay that registered first would publish an unnamed robot diff --git a/mediad/src/route.rs b/mediad/src/route.rs index 30cef1b5..a641d64f 100644 --- a/mediad/src/route.rs +++ b/mediad/src/route.rs @@ -305,10 +305,14 @@ pub fn route_for(call: &proto::Call) -> Route { } /// The refusal a peer gets, naming the method so a client can report which call was declined. +/// +/// "this transport" rather than "WebRTC": the table is shared with the agent socket +/// (`crate::agent`), and a websocket client told its call was refused over WebRTC would go looking +/// for a second table that does not exist. pub fn refusal(call: &proto::Call) -> proto::Error { proto::Error::new( proto::code::METHOD_NOT_FOUND, - format!("{} is not available over WebRTC", call.method()), + format!("{} is not available over this transport", call.method()), ) } diff --git a/mediad/src/session.rs b/mediad/src/session.rs index 6c65132e..974d3205 100644 --- a/mediad/src/session.rs +++ b/mediad/src/session.rs @@ -466,7 +466,10 @@ mod tests { .unwrap(); let reply = h.to_peer.recv().await.unwrap(); - assert!(reply.contains("not available over WebRTC"), "{reply}"); + assert!( + reply.contains("not available over this transport"), + "{reply}" + ); assert!(reply.contains(r#""id":7"#), "{reply}"); assert!( config_seen.try_recv().is_err(), diff --git a/mediad/src/web.rs b/mediad/src/web.rs index 75922f03..6da123b9 100644 --- a/mediad/src/web.rs +++ b/mediad/src/web.rs @@ -72,7 +72,7 @@ pub fn page(signalling_port: u32) -> String { /// Returns only on failure — a bind that was refused, or a listener that died. The caller decides /// what that costs; in `mediad` it costs the page and not the video, because a robot that streams /// and answers control calls with no console is a great deal better than one that does neither. -pub async fn serve(host: &str, port: u16, page: String) -> Result<()> { +pub async fn serve(host: &str, port: u16, page: String, agent: crate::agent::State) -> Result<()> { let address: SocketAddr = format!("{host}:{port}") .parse() .with_context(|| format!("{host}:{port} is not an address to listen on"))?; @@ -80,15 +80,24 @@ pub async fn serve(host: &str, port: u16, page: String) -> Result<()> { .await .with_context(|| format!("could not listen on {address}"))?; - tracing::info!(%address, "serving the console"); - axum::serve(listener, router(page)) + tracing::info!(%address, "serving the console and the agent socket"); + axum::serve(listener, router(page, agent)) .await .context("the console's listener stopped") } -/// One route, returning `page`. -fn router(page: String) -> Router { - Router::new().route("/", get(move || std::future::ready(Html(page)))) +/// The page, and the WebSocket a program drives the robot over. +/// +/// One listener for both because they are one interface seen from two sides, and because a second +/// port is a second thing to configure, open and explain. `/` is for a person, `/agent` is for a +/// program, and `crate::agent` says why the second one exists. +fn router(page: String, agent: crate::agent::State) -> Router { + Router::new() + .route("/", get(move || std::future::ready(Html(page)))) + .route( + "/agent", + get(move |ws| crate::agent::upgrade(ws, agent.clone())), + ) } #[cfg(test)] @@ -161,7 +170,12 @@ mod tests { .expect("a loopback port"); let address = listener.local_addr().expect("the port it took"); tokio::spawn(async move { - let _ = axum::serve(listener, router(page(8443))).await; + let (_video_tx, video) = tokio::sync::watch::channel(None); + let agent = crate::agent::State { + sockets: crate::upstream::Sockets::default(), + video, + }; + let _ = axum::serve(listener, router(page(8443), agent)).await; }); let mut stream = tokio::net::TcpStream::connect(address) From 2db2be34b74d3745246da55f906c022a3e9e25dc Mon Sep 17 00:00:00 2001 From: nityam Date: Sat, 12 Sep 2026 10:02:29 +0530 Subject: [PATCH 2/3] sdk: a python client, so a script can fetch a frame and send an intent --- docs/project/roadmap.md | 24 ++- sdk/README.md | 46 +++++ sdk/duck/__init__.py | 24 +++ sdk/duck/client.py | 175 ++++++++++++++++++ sdk/duck/rpc.py | 168 +++++++++++++++++ .../fetch_a_frame_and_send_an_intent.py | 57 ++++++ sdk/microduck.egg-info/PKG-INFO | 55 ++++++ sdk/microduck.egg-info/SOURCES.txt | 10 + sdk/microduck.egg-info/dependency_links.txt | 1 + sdk/microduck.egg-info/requires.txt | 1 + sdk/microduck.egg-info/top_level.txt | 1 + sdk/pyproject.toml | 15 ++ sdk/tests/test_client.py | 87 +++++++++ 13 files changed, 654 insertions(+), 10 deletions(-) create mode 100644 sdk/README.md create mode 100644 sdk/duck/__init__.py create mode 100644 sdk/duck/client.py create mode 100644 sdk/duck/rpc.py create mode 100644 sdk/examples/fetch_a_frame_and_send_an_intent.py create mode 100644 sdk/microduck.egg-info/PKG-INFO create mode 100644 sdk/microduck.egg-info/SOURCES.txt create mode 100644 sdk/microduck.egg-info/dependency_links.txt create mode 100644 sdk/microduck.egg-info/requires.txt create mode 100644 sdk/microduck.egg-info/top_level.txt create mode 100644 sdk/pyproject.toml create mode 100644 sdk/tests/test_client.py diff --git a/docs/project/roadmap.md b/docs/project/roadmap.md index 4a559df5..0d958845 100644 --- a/docs/project/roadmap.md +++ b/docs/project/roadmap.md @@ -153,15 +153,18 @@ because the first-party device-code client takes no `scope` parameter. Narrowing `openid profile read-repos` is a public OAuth app in the org and one constant. **The SDK, and a small Python client.** §5.3 designs it as WebSocket plus snapshot: the same -JSON-RPC, no media stack, `get_frame` returning a JPEG, a few dozen lines — and `mediad`'s -session layer was built so that surface reuses it unchanged (`mediad/src/session.rs`). A Python -client over **WebRTC** instead gets live video and the `control` datachannel from one -connection, at the cost of `aiortc`, an ICE negotiation and a signalling round trip for a caller -who only wants to send an intent and read a frame. **The investigation is whether one client -covers both** — WebSocket for control and snapshots, WebRTC only when the caller asks for a -stream — or whether the WebSocket surface alone is what a script wants and live video stays in -the console. Answer that before writing either, because it decides whether the SDK is fifty -lines or a project. +JSON-RPC, no media stack, a few dozen lines — and `mediad`'s session layer was built so that +surface reuses it unchanged (`mediad/src/session.rs`). Both halves are built: `mediad::agent` +serves the socket at `/agent`, and `sdk/` is the client. + +**The investigation it was waiting on is answered, and the answer was fifty lines rather than a +project.** One client does cover both, because the transport was never the hard part: +`spaces/shared/control.py` had already carried the same JSON-RPC over three of them — a +datachannel through the rendezvous, a datachannel on the LAN, and HTTP with no WebRTC at all — so +the SDK binds a WebSocket to that same object rather than being a fourth implementation of the +wire. Live video stays in the console: a viewer wants WebRTC, and a script that wants frames gets +them the way `stream.rs` already sends them, outbound to a socket it opens, which needs no relay +candidate and no ICE. **Privacy, and it is now two items rather than one.** *Consent* — explicit per-session approval before a stream starts — is a `mediad` session-layer change and is not blocked on anything. The @@ -171,7 +174,8 @@ be asked of the hardware rather than parked on a software milestone. `architectu right that both are cheap now and expensive later, so consent should not wait for the LED. **Done when:** telepresence works from outside the LAN, and a server-side script can fetch a -frame and send an intent in a few dozen lines. +frame and send an intent in a few dozen lines. The second half is done — +`sdk/examples/fetch_a_frame_and_send_an_intent.py` is that sentence as a program. ### M6 — Ship readiness diff --git a/sdk/README.md b/sdk/README.md new file mode 100644 index 00000000..f2b1b5a8 --- /dev/null +++ b/sdk/README.md @@ -0,0 +1,46 @@ +# Driving a duck from a script + +```python +from duck import Duck + +with Duck("robot.local") as duck: + print(duck.health()["healthy"]) + duck.move(vx=0.1) +``` + +`pip install -e .`, and the robot needs `mediad` running — the socket is `/agent` on the port the +console is served from, 8080 by default. + +## What it is + +One WebSocket, the same JSON-RPC every other transport speaks, and no media stack. +`architecture.md` §5.3 is the argument for why a program should not have to negotiate ICE and +decode H.264 to send an intent. + +It is deliberately small. Every method is one call, and the ones that exist are the ones +`mediad::route` permits — so what a script may do is what a browser on the LAN may do, decided in +one place on the robot rather than twice. + +## Frames + +The robot does not serve frames, it **sends** them to a socket you open: + +```python +from duck import Duck, receive +import threading + +threading.Thread(target=receive, args=(8099, print_frame), daemon=True).start() +with Duck("robot.local") as duck: + duck.frames(url="ws://192.168.1.20:8099", fps=1) +``` + +That direction is the point. A robot behind a home router and a script anywhere else cannot pair +without a relay candidate, and the robot dialling out means NAT is not a participant. +`mediad/src/stream.rs` has the whole argument. + +`examples/fetch_a_frame_and_send_an_intent.py` is both halves in about thirty lines. + +## What it does not do + +Live video. A viewer wants WebRTC and the console already is one. This is for when the consumer +is a program. diff --git a/sdk/duck/__init__.py b/sdk/duck/__init__.py new file mode 100644 index 00000000..532d7b2d --- /dev/null +++ b/sdk/duck/__init__.py @@ -0,0 +1,24 @@ +"""Drive a duck from a script. + +`architecture.md` §5.3 argues a server-side program should not be pushed through WebRTC: an agent +does not want a 30 fps H.264 track to decode, it wants a frame every second or two and a state +blob, and ICE and DTLS in front of that is a poor trade. `mediad`'s `/agent` WebSocket is that +surface. This is the client for it, and it is the "few dozen lines" M5 asks for: + + from duck import Duck + + with Duck("robot.local") as duck: + print(duck.health()["healthy"]) + duck.move(vx=0.1) + +Every method here is one JSON-RPC call against the same `route` table the console page uses, so +what a script may do is what a browser on the LAN may do, decided in one place on the robot. There +is deliberately no method for anything the robot refuses that transport. +""" + +from __future__ import annotations + +from .client import Duck, receive +from .rpc import RpcError + +__all__ = ["Duck", "RpcError", "receive"] diff --git a/sdk/duck/client.py b/sdk/duck/client.py new file mode 100644 index 00000000..753d1d10 --- /dev/null +++ b/sdk/duck/client.py @@ -0,0 +1,175 @@ +"""One duck, over the agent WebSocket. + +The transport is the thin part: a background thread owns the socket, `Rpc` owns the id space and +the pending table, and everything below is one call each. What is *not* thin is which calls exist +— those are the ones `mediad::route` permits, and a method missing here is a method the robot +would refuse anyway. +""" + +from __future__ import annotations + +import json +import threading +from typing import Any, Callable + +from websockets.sync.client import connect + +from .rpc import Rpc, RpcError + +__all__ = ["Duck", "RpcError"] + +#: Where `mediad` serves the console, and the agent socket beside it. +DEFAULT_PORT = 8080 + + +class Duck: + """A connected duck. + + Blocking, on purpose: a script that fetches a frame and sends an intent has nothing else to do + while it waits, and `asyncio` in front of that is a tax on the simplest possible caller. The + socket is read on its own thread so notifications — `robot.state`, `media.detections` — arrive + while a call is in flight rather than behind it. + """ + + def __init__(self, host: str, port: int = DEFAULT_PORT, timeout: float = 30.0): + self.url = f"ws://{host}:{port}/agent" + self._rpc = Rpc(timeout=timeout) + self._socket = connect(self.url) + self._closing = threading.Event() + self._rpc.bound_to(self._send) + self._reader = threading.Thread(target=self._read, name="duck-agent", daemon=True) + self._reader.start() + + # ── the transport ──────────────────────────────────────────────────────── + + def _send(self, message: dict[str, Any]) -> bool: + try: + self._socket.send(json.dumps(message)) + return True + except Exception: + return False + + def _read(self) -> None: + try: + for message in self._socket: + self._rpc.on_message(message) + except Exception as e: + # A closed socket during `close()` is the ordinary way this ends, and failing every + # call in flight over it would be a lie. + if not self._closing.is_set(): + self._rpc.abandon(str(e)) + return + if not self._closing.is_set(): + self._rpc.abandon("the robot closed the connection") + + def close(self) -> None: + self._closing.set() + self._socket.close() + + def __enter__(self) -> "Duck": + return self + + def __exit__(self, *_: object) -> None: + self.close() + + # ── asking ─────────────────────────────────────────────────────────────── + + def call(self, method: str, **params: Any) -> Any: + """Any method the robot routes to this transport, for the ones without a wrapper below.""" + return self._rpc.call(method, params or None) + + def health(self) -> dict[str, Any]: + """Is the robot alright — the loop's rate, the bus, the battery, the motors.""" + return self._rpc.call("robot.health") + + def state(self) -> dict[str, Any] | None: + """The last `robot.state` that arrived, or `None` before the first. + + A read of what has already been pushed rather than a call: `subscribe` starts the stream + and this is where it lands. + """ + return self._rpc.notifications.get("robot.state") + + def subscribe(self, hz: int | None = None) -> Any: + """Start the `robot.state` stream, so `state()` has something to answer with.""" + return self._rpc.call("robot.subscribe", {"hz": hz} if hz else {}) + + def video(self) -> dict[str, Any]: + """What the camera is sending: the frame size, and how far it is mounted from upright.""" + return self._rpc.call("media.video") + + # ── telling ────────────────────────────────────────────────────────────── + + def move(self, vx: float = 0.0, vy: float = 0.0, vyaw: float = 0.0) -> Any: + """Walk. Metres per second forward and left, radians per second about up. + + One call is one intent, and `robotd`'s deadman zeroes the twist when they stop arriving — + so a script that wants the robot to keep walking has to keep saying so, which is the + property that makes a crashed script a robot that stops. + """ + return self._rpc.call("robot.move", {"vx": vx, "vy": vy, "vyaw": vyaw}) + + def stop(self) -> Any: + """Zero the twist now rather than waiting for the deadman.""" + return self._rpc.call("robot.stop") + + def look(self, x: float, y: float, z: float) -> Any: + """Point the camera at a trunk-frame point: x forward, y left, z up, metres.""" + return self._rpc.call("robot.look", {"x": x, "y": y, "z": z}) + + def head(self, yaw: float = 0.0, pitch: float = 0.0, roll: float = 0.0) -> Any: + """Pose the head directly, radians.""" + return self._rpc.call("robot.head", {"yaw": yaw, "pitch": pitch, "roll": roll}) + + def enable(self, on: bool = True) -> Any: + """Hand the robot to its policy, or take it back.""" + return self._rpc.call("robot.enable", {"on": on}) + + def do(self, skill: str) -> Any: + """Run a one-shot skill by name — whatever `robot.health` says this robot has.""" + return self._rpc.call("robot.do", {"skill": skill}) + + # ── frames ─────────────────────────────────────────────────────────────── + + def frames(self, url: str, fps: float | None = None, longest: int | None = None) -> Any: + """Tell the robot to send JPEG frames to a WebSocket it dials. + + **Outbound, and that is the point.** The robot is behind somebody's router and the script + may be anywhere; a robot that dials out needs no relay candidate and no NAT traversal. + `stream.rs` has the argument. `receive` below is the other end of it. + """ + params: dict[str, Any] = {"url": url} + if fps is not None: + params["fps"] = fps + if longest is not None: + params["longest"] = longest + return self._rpc.call("media.stream", params) + + def frames_stop(self) -> Any: + """Stop streaming.""" + return self._rpc.call("media.stream", {"url": None}) + + def frames_status(self) -> Any: + """What is streaming, if anything.""" + return self._rpc.call("media.stream") + + +def receive(port: int, on_frame: Callable[[bytes], None], host: str = "0.0.0.0") -> None: + """Listen for the frames a duck was told to send, and hand each JPEG to `on_frame`. + + The half a script would otherwise have to write itself, and the reason `frames()` is usable + without a Space: `mediad` sends one text message describing what is coming and then one binary + message per frame, so everything here is "ignore the first kind, pass on the second". + + Blocks. Runs until the socket closes or the callback raises. + """ + from websockets.sync.server import serve + + def handler(connection: Any) -> None: + for message in connection: + # The opening text frame says the size and the rotation; the rest are JPEGs. + if isinstance(message, bytes): + on_frame(message) + + with serve(handler, host, port) as server: + server.serve_forever() diff --git a/sdk/duck/rpc.py b/sdk/duck/rpc.py new file mode 100644 index 00000000..9a787e3a --- /dev/null +++ b/sdk/duck/rpc.py @@ -0,0 +1,168 @@ +"""JSON-RPC 2.0 over whatever carries it: ids out, answers and notifications back. + +Taken from `spaces/shared/control.py` unchanged, because a fourth implementation of the same wire +is a fourth thing to keep in step. It is transport-agnostic already — three of them have been +carried on it — and `Duck` binds a WebSocket to it the way a Space binds a datachannel. + +`duck-ipc-proto`'s own wire, one object per line — the same lines `robotctl` sends over a unix +socket, the console page sends over a datachannel, and `mediad`'s control lane relays inside a +`peer` envelope. Ids are handed out here and answers matched to them, which is what lets a Gradio +callback block on one. + +**Transport-agnostic, and that is now load-bearing rather than tidy.** Three of them have been +tried: a datachannel over the rendezvous, a datachannel on the LAN, and JSON-RPC relayed as HTTP +with no WebRTC at all. Every one of them hands lines to this object and takes lines back, so none +of the page above it changed when the first was replaced by the third. + +It once also held a shim over `ReachyCentralConsumer`, whose `pc.on("datachannel")` handler drops +any label but `"data"` while `mediad` opens `"control"` — `remote-access-design.md` §5.1 records +it, and it is still true of their client. It is gone because the transport that needed it is gone: +a rendezvous session that has to negotiate ICE cannot connect from a data centre while §6 stands, +so `wire.py` replaced it, and with it the only dependency this file had on another package's +private method. +""" + +from __future__ import annotations + +import asyncio +import itertools +import json +import logging +import threading +from concurrent.futures import Future +from concurrent.futures import TimeoutError as FutureTimeout +from typing import Any, Callable + +logger = logging.getLogger(__name__) + +# What `mediad` calls the channel it opens on every consumer's peer connection. +CONTROL_LABEL = "control" + + +class RpcError(Exception): + """A refusal from the robot, carrying the code it refused with. + + Distinct from a transport failure on purpose: `policy.fetch` answering "obs_len 51, this + robot is 61" is the robot working correctly and is the most useful thing this Space can + show, while "no answer in 30s" means something else entirely. + """ + + def __init__(self, method: str, error: dict[str, Any]): + self.method = method + self.code = error.get("code") + self.message = error.get("message") or "refused, with nothing said about why" + super().__init__(f"{method}: {self.message}") + + +class Rpc: + """Requests out, answers and notifications in, over one data channel. + + Not a session and not a connection — those are the consumer's. This owns the id space and + the pending table, and it survives a channel closing: the futures are failed rather than + left for a caller to time out on one at a time. + """ + + def __init__(self, timeout: float = 30.0): + self.timeout = timeout + self._ids = itertools.count(1) + # The method rides along with the future so a refusal can name what was refused: a + # reply carries an id and nothing else, and "call: no such method" is a worse sentence + # than "robot.setSkill: no such method" for the sake of one tuple. + self._pending: dict[int, tuple[str, Future]] = {} + self._lock = threading.Lock() + self._send: Callable[[dict[str, Any]], bool] | None = None + # The last one of each notification. A duck streams `robot.state` at whatever rate it was + # asked for, so keeping them all is a leak and keeping the last is what a panel shows. + self.notifications: dict[str, Any] = {} + + def bound_to(self, send: Callable[[dict[str, Any]], bool]) -> None: + self._send = send + + def is_open(self) -> bool: + return self._send is not None + + def call( + self, method: str, params: dict[str, Any] | None = None, timeout: float | None = None + ) -> Any: + """Send one request and wait for its answer. + + Blocking, because every caller here is a Gradio callback with nothing else to do. The + timeout is not a budget on the robot's behalf: a promise nobody settles is a leak, and a + method that never answers is worth seeing rather than a panel that quietly stopped. + `policy.fetch` is the one call that wants its own — it is a download over the robot's + wifi, not a question about state. + """ + budget = self.timeout if timeout is None else timeout + send = self._send + if send is None: + raise RpcError(method, {"message": "no control channel — connect first"}) + + call_id = next(self._ids) + future: Future = Future() + with self._lock: + self._pending[call_id] = (method, future) + logger.info("→ %s %s", method, json.dumps(params or {})) + + if not send({"jsonrpc": "2.0", "id": call_id, "method": method, "params": params or {}}): + with self._lock: + self._pending.pop(call_id, None) + raise RpcError(method, {"message": "the control channel would not take the request"}) + + try: + return future.result(timeout=budget) + except FutureTimeout: + with self._lock: + self._pending.pop(call_id, None) + raise RpcError( + method, {"message": f"no answer in {budget:.0f}s"} + ) from None + + def on_message(self, raw: Any) -> None: + """One line off the channel. Runs on the consumer's event loop, so it does not block.""" + if isinstance(raw, bytes): + raw = raw.decode("utf-8", "replace") + try: + message = json.loads(raw) + except (TypeError, ValueError): + logger.warning("← unparseable: %s", str(raw)[:160]) + return + + call_id = message.get("id") + if call_id is None: + # A notification. `robot.state` streams, `media.video` arrives once when the channel + # opens, and `media.detections` a couple of times a second — none of them answer + # anything anybody asked for. + method = message.get("method") + if method: + self.notifications[method] = message.get("params") + # DEBUG, not INFO: `media.detections` arrives a couple of times a second and + # `robot.state` at whatever rate it was asked for, and a log that scrolls is a + # log nobody reads the top of. + logger.debug("← %s %s", method, json.dumps(message.get("params"))[:200]) + return + + with self._lock: + waiting = self._pending.pop(call_id, None) + if waiting is None: + # A duck does not correlate replies (`remote-webrtc.md` §5), so this is ordinary: + # a fire-and-forget intent's answer, or one that arrived after its timeout. + return + method, future = waiting + if "error" in message: + error = message["error"] or {} + logger.warning("← %s refused: %s", method, error.get("message")) + future.set_exception(RpcError(method, error)) + else: + result = message.get("result") + logger.info("← %s %s", method, json.dumps(result)[:240]) + future.set_result(result) + + def abandon(self, why: str) -> None: + """Fail everything in flight. A closed channel answers nothing, ever.""" + logger.info("control channel gone: %s", why) + with self._lock: + pending, self._pending = self._pending, {} + self._send = None + for method, future in pending.values(): + if not future.done(): + future.set_exception(RpcError(method, {"message": why})) diff --git a/sdk/examples/fetch_a_frame_and_send_an_intent.py b/sdk/examples/fetch_a_frame_and_send_an_intent.py new file mode 100644 index 00000000..d09d025b --- /dev/null +++ b/sdk/examples/fetch_a_frame_and_send_an_intent.py @@ -0,0 +1,57 @@ +"""M5's sentence, as a program: a server-side script fetches a frame and sends an intent. + + python examples/fetch_a_frame_and_send_an_intent.py robot.local + +Frames come *outbound* from the robot to a socket this script opens, so the robot dials us: no +relay candidate, no NAT traversal, and it works from anywhere the robot can reach. +""" + +import sys +import threading + +from duck import Duck, receive + +FRAME_PORT = 8099 + + +def main(host: str) -> None: + seen = threading.Event() + + def on_frame(jpeg: bytes) -> None: + if not seen.is_set(): + with open("frame.jpg", "wb") as f: + f.write(jpeg) + print(f"got a frame, {len(jpeg)} bytes, written to frame.jpg") + seen.set() + + threading.Thread(target=receive, args=(FRAME_PORT, on_frame), daemon=True).start() + + with Duck(host) as duck: + print("healthy:", duck.health()["healthy"]) + + # Where to send them. The robot dials this, so it has to be our address as the robot + # sees it rather than a loopback one. + duck.frames(url=f"ws://{local_address(host)}:{FRAME_PORT}", fps=1) + seen.wait(timeout=10) + duck.frames_stop() + + # An intent. One call is one intent, and the deadman stops the robot when they stop + # arriving — so this walks for about a second and then the robot stops on its own. + duck.move(vx=0.1) + print("told it to walk") + + +def local_address(host: str) -> str: + """Our address on the route to the robot, which is what the robot should dial back.""" + import socket + + probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + probe.connect((host, 80)) + return probe.getsockname()[0] + finally: + probe.close() + + +if __name__ == "__main__": + main(sys.argv[1] if len(sys.argv) > 1 else "robot.local") diff --git a/sdk/microduck.egg-info/PKG-INFO b/sdk/microduck.egg-info/PKG-INFO new file mode 100644 index 00000000..3bce69f1 --- /dev/null +++ b/sdk/microduck.egg-info/PKG-INFO @@ -0,0 +1,55 @@ +Metadata-Version: 2.4 +Name: microduck +Version: 0.1.0 +Summary: Drive a Microduck from a script, over the agent WebSocket +License-Expression: Apache-2.0 +Requires-Python: >=3.10 +Description-Content-Type: text/markdown +Requires-Dist: websockets>=13 + +# Driving a duck from a script + +```python +from duck import Duck + +with Duck("robot.local") as duck: + print(duck.health()["healthy"]) + duck.move(vx=0.1) +``` + +`pip install -e .`, and the robot needs `mediad` running — the socket is `/agent` on the port the +console is served from, 8080 by default. + +## What it is + +One WebSocket, the same JSON-RPC every other transport speaks, and no media stack. +`architecture.md` §5.3 is the argument for why a program should not have to negotiate ICE and +decode H.264 to send an intent. + +It is deliberately small. Every method is one call, and the ones that exist are the ones +`mediad::route` permits — so what a script may do is what a browser on the LAN may do, decided in +one place on the robot rather than twice. + +## Frames + +The robot does not serve frames, it **sends** them to a socket you open: + +```python +from duck import Duck, receive +import threading + +threading.Thread(target=receive, args=(8099, print_frame), daemon=True).start() +with Duck("robot.local") as duck: + duck.frames(url="ws://192.168.1.20:8099", fps=1) +``` + +That direction is the point. A robot behind a home router and a script anywhere else cannot pair +without a relay candidate, and the robot dialling out means NAT is not a participant. +`mediad/src/stream.rs` has the whole argument. + +`examples/fetch_a_frame_and_send_an_intent.py` is both halves in about thirty lines. + +## What it does not do + +Live video. A viewer wants WebRTC and the console already is one. This is for when the consumer +is a program. diff --git a/sdk/microduck.egg-info/SOURCES.txt b/sdk/microduck.egg-info/SOURCES.txt new file mode 100644 index 00000000..9e66e5ba --- /dev/null +++ b/sdk/microduck.egg-info/SOURCES.txt @@ -0,0 +1,10 @@ +README.md +pyproject.toml +duck/__init__.py +duck/client.py +duck/rpc.py +microduck.egg-info/PKG-INFO +microduck.egg-info/SOURCES.txt +microduck.egg-info/dependency_links.txt +microduck.egg-info/requires.txt +microduck.egg-info/top_level.txt \ No newline at end of file diff --git a/sdk/microduck.egg-info/dependency_links.txt b/sdk/microduck.egg-info/dependency_links.txt new file mode 100644 index 00000000..8b137891 --- /dev/null +++ b/sdk/microduck.egg-info/dependency_links.txt @@ -0,0 +1 @@ + diff --git a/sdk/microduck.egg-info/requires.txt b/sdk/microduck.egg-info/requires.txt new file mode 100644 index 00000000..9473ab6d --- /dev/null +++ b/sdk/microduck.egg-info/requires.txt @@ -0,0 +1 @@ +websockets>=13 diff --git a/sdk/microduck.egg-info/top_level.txt b/sdk/microduck.egg-info/top_level.txt new file mode 100644 index 00000000..ed891705 --- /dev/null +++ b/sdk/microduck.egg-info/top_level.txt @@ -0,0 +1 @@ +duck diff --git a/sdk/pyproject.toml b/sdk/pyproject.toml new file mode 100644 index 00000000..63e7ca99 --- /dev/null +++ b/sdk/pyproject.toml @@ -0,0 +1,15 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "microduck" +version = "0.1.0" +description = "Drive a Microduck from a script, over the agent WebSocket" +readme = "README.md" +license = "Apache-2.0" +requires-python = ">=3.10" +dependencies = ["websockets>=13"] + +[tool.setuptools.packages.find] +include = ["duck*"] diff --git a/sdk/tests/test_client.py b/sdk/tests/test_client.py new file mode 100644 index 00000000..ed06b361 --- /dev/null +++ b/sdk/tests/test_client.py @@ -0,0 +1,87 @@ +"""What a caller can rely on, without a robot. + +The wire is checked against a fake socket rather than a duck: these are about the shape of what +goes out and what comes back, and the half that needs a real `mediad` is `README.md`'s example. +""" + +import json +import threading + +import pytest + +from duck.rpc import Rpc, RpcError + + +def bound() -> tuple[Rpc, list]: + """An `Rpc` writing into a list instead of a socket.""" + rpc = Rpc(timeout=2.0) + sent: list = [] + rpc.bound_to(lambda message: (sent.append(message), True)[1]) + return rpc, sent + + +def test_a_call_goes_out_as_jsonrpc_and_its_answer_comes_back(): + rpc, sent = bound() + + def answer(): + while not sent: + pass + rpc.on_message(json.dumps({"jsonrpc": "2.0", "id": sent[0]["id"], "result": {"ok": True}})) + + threading.Thread(target=answer, daemon=True).start() + assert rpc.call("robot.health") == {"ok": True} + assert sent[0]["method"] == "robot.health" + assert sent[0]["jsonrpc"] == "2.0" + + +def test_a_refusal_names_the_method_it_refused(): + """A reply carries an id and nothing else, so the method has to be remembered here — and + "robot.setMode: not available over this transport" is a better sentence than "call:".""" + rpc, sent = bound() + + def refuse(): + while not sent: + pass + rpc.on_message( + json.dumps( + { + "jsonrpc": "2.0", + "id": sent[0]["id"], + "error": {"code": -32601, "message": "not available over this transport"}, + } + ) + ) + + threading.Thread(target=refuse, daemon=True).start() + with pytest.raises(RpcError) as refused: + rpc.call("robot.setMode", {"mode": "roller"}) + assert "robot.setMode" in str(refused.value) + + +def test_notifications_are_kept_by_method_and_never_answer_a_call(): + """`robot.state` streams. A caller reads the last one; nothing here waits for it.""" + rpc, _ = bound() + rpc.on_message(json.dumps({"jsonrpc": "2.0", "method": "robot.state", "params": {"t": 1.0}})) + rpc.on_message(json.dumps({"jsonrpc": "2.0", "method": "robot.state", "params": {"t": 2.0}})) + assert rpc.notifications["robot.state"] == {"t": 2.0} + + +def test_a_closed_socket_fails_everything_in_flight(): + """A promise nobody settles is a leak, so a dropped connection fails the callers rather than + leaving each to time out on its own.""" + rpc, sent = bound() + failed: list = [] + + def call(): + try: + rpc.call("robot.health") + except RpcError as e: + failed.append(e) + + caller = threading.Thread(target=call, daemon=True) + caller.start() + while not sent: + pass + rpc.abandon("the robot closed the connection") + caller.join(timeout=3) + assert failed and "closed" in str(failed[0]) From 548fc5fc3f965d855dd5f7c50b0cf43ec72d1d4e Mon Sep 17 00:00:00 2001 From: nityam Date: Sat, 12 Sep 2026 10:34:33 +0530 Subject: [PATCH 3/3] sdk: watch(), so a behaviour is a loop instead of three subscriptions --- .gitignore | 3 + sdk/README.md | 27 ++++ sdk/duck/__init__.py | 13 +- sdk/duck/client.py | 147 +++++++++++++++++- sdk/duck/view.py | 114 ++++++++++++++ .../wander_without_bumping_into_things.py | 52 +++++++ sdk/microduck.egg-info/PKG-INFO | 55 ------- sdk/microduck.egg-info/SOURCES.txt | 10 -- sdk/microduck.egg-info/dependency_links.txt | 1 - sdk/microduck.egg-info/requires.txt | 1 - sdk/microduck.egg-info/top_level.txt | 1 - sdk/tests/test_view.py | 69 ++++++++ 12 files changed, 422 insertions(+), 71 deletions(-) create mode 100644 sdk/duck/view.py create mode 100644 sdk/examples/wander_without_bumping_into_things.py delete mode 100644 sdk/microduck.egg-info/PKG-INFO delete mode 100644 sdk/microduck.egg-info/SOURCES.txt delete mode 100644 sdk/microduck.egg-info/dependency_links.txt delete mode 100644 sdk/microduck.egg-info/requires.txt delete mode 100644 sdk/microduck.egg-info/top_level.txt create mode 100644 sdk/tests/test_view.py diff --git a/.gitignore b/.gitignore index ad6ab663..17fa59f2 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,6 @@ __pycache__/ # holding `*` so it self-ignores; this is for the person who reaches for `python -m venv`, which # does not, and it is the same accident `__pycache__` above was added for. .venv/ + +# a local editable install leaves this behind +sdk/*.egg-info/ diff --git a/sdk/README.md b/sdk/README.md index f2b1b5a8..61da53c6 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -21,6 +21,33 @@ It is deliberately small. Every method is one call, and the ones that exist are `mediad::route` permits — so what a script may do is what a browser on the LAN may do, decided in one place on the robot rather than twice. +## A behaviour is a loop + +What the robot can see arrives on three streams at three different rates — state at the control +rate, depth at 15 Hz, frames at whatever you asked for. `watch` merges them and hands you the +latest of each, once per tick: + +```python +with Duck("robot.local") as duck: + for view in duck.watch(fps=4, camera=False): + if view.nearest and view.nearest < 0.4: + duck.move(vyaw=0.8) + else: + duck.move(vx=0.12) +``` + +`view.nearest` is the closest thing the depth sensor can see, in metres, or `None` when it sees +nothing usable. The interpretation is `tof::Frame::zone`'s, including the part that bites: the +sensor returns negative distances on a failed convergence and still flags them valid, so taken at +face value they are the nearest thing in the room. + +`examples/wander_without_bumping_into_things.py` is a duck that walks around a room, in about +twenty lines of behaviour. + +Nothing here looks at a picture. `view.frame` is the JPEG as it arrived and what is in it is your +model to run — an SDK that shipped one would be a much bigger dependency and a worse guess than +whatever you already have. + ## Frames The robot does not serve frames, it **sends** them to a socket you open: diff --git a/sdk/duck/__init__.py b/sdk/duck/__init__.py index 532d7b2d..fc837ec3 100644 --- a/sdk/duck/__init__.py +++ b/sdk/duck/__init__.py @@ -11,6 +11,16 @@ print(duck.health()["healthy"]) duck.move(vx=0.1) +And a behaviour is a loop over what the robot can see, which `watch` merges from the three streams +it arrives on: + + with Duck("robot.local") as duck: + for view in duck.watch(fps=2): + if view.nearest and view.nearest < 0.3: + duck.stop() + else: + duck.move(vx=0.15) + Every method here is one JSON-RPC call against the same `route` table the console page uses, so what a script may do is what a browser on the LAN may do, decided in one place on the robot. There is deliberately no method for anything the robot refuses that transport. @@ -19,6 +29,7 @@ from __future__ import annotations from .client import Duck, receive +from .view import View from .rpc import RpcError -__all__ = ["Duck", "RpcError", "receive"] +__all__ = ["Duck", "RpcError", "View", "receive"] diff --git a/sdk/duck/client.py b/sdk/duck/client.py index 753d1d10..31c2e788 100644 --- a/sdk/duck/client.py +++ b/sdk/duck/client.py @@ -10,13 +10,15 @@ import json import threading -from typing import Any, Callable +import time +from typing import Any, Callable, Iterator from websockets.sync.client import connect from .rpc import Rpc, RpcError +from .view import View -__all__ = ["Duck", "RpcError"] +__all__ = ["Duck", "RpcError", "View"] #: Where `mediad` serves the console, and the agent socket beside it. DEFAULT_PORT = 8080 @@ -153,6 +155,37 @@ def frames_status(self) -> Any: """What is streaming, if anything.""" return self._rpc.call("media.stream") + # ── the loop ───────────────────────────────────────────────────────────── + + def watch( + self, + fps: float = 2.0, + camera: bool = True, + depth: bool = True, + frame_port: int = 8099, + ) -> "Iterator[View]": + """Yield one [`View`] per tick, with whatever each stream sent most recently. + + This is the loop a robot program is: + + for view in duck.watch(): + if view.nearest and view.nearest < 0.3: + duck.stop() + + It subscribes to `robot.state`, starts the depth stream, and tells the robot to send + frames to a socket it opens here — then merges the three and hands over the latest of + each. Everything is turned off again when the loop ends, including when it ends because + the caller raised. + + `camera=False` skips the frames, which is what a behaviour that only needs state and + depth wants: the robot stops encoding JPEGs for nobody, and no inbound port is opened. + + **The rate is the behaviour's, not the sensors'.** `robot.state` arrives at the control + rate and depth at 15 Hz; ticking at those would make the loop the fastest thing rather + than the one deciding. `fps` is how often the caller wants to think. + """ + return _watch(self, fps=fps, camera=camera, depth=depth, frame_port=frame_port) + def receive(port: int, on_frame: Callable[[bytes], None], host: str = "0.0.0.0") -> None: """Listen for the frames a duck was told to send, and hand each JPEG to `on_frame`. @@ -173,3 +206,113 @@ def handler(connection: Any) -> None: with serve(handler, host, port) as server: server.serve_forever() + + +def _watch( + duck: Duck, fps: float, camera: bool, depth: bool, frame_port: int +) -> Iterator[View]: + """[`Duck.watch`]'s body, as a generator so its `finally` runs when the caller stops.""" + latest: dict[str, Any] = {"frame": None} + started = time.monotonic() + + # A notification lands on the reader thread. Nothing here locks: each of these is one + # assignment of one reference, and a tick reading a field mid-swap gets the old value or the + # new one, never half of either. + def on_frame(jpeg: bytes) -> None: + latest["frame"] = jpeg + + server = None + if camera: + server = _FrameServer(frame_port, on_frame) + server.start() + duck.frames(url=f"ws://{_address_the_robot_can_reach(duck.url)}:{frame_port}", fps=fps) + + duck.subscribe() + if depth: + # `tofd` streams to whoever asked; the notifications land in `Rpc.notifications` beside + # `robot.state`, so there is nothing further to wire up. + try: + duck.call("tof.stream") + except RpcError: + # A duck with no ToF fitted refuses this, and that is not a reason to stop: a + # behaviour that wanted depth gets `None` and can say so itself. + pass + + tick = 0 + period = 1.0 / fps if fps > 0 else 0.0 + try: + while True: + tick += 1 + yield View( + frame=latest["frame"], + state=duck._rpc.notifications.get("robot.state"), + depth=duck._rpc.notifications.get("tof.frame"), + elapsed=time.monotonic() - started, + tick=tick, + ) + if period: + time.sleep(period) + finally: + # Whatever ended the loop — a `break`, an exception, the caller simply stopping — the + # robot should not be left encoding frames for a socket that has gone. + if camera: + try: + duck.frames_stop() + except RpcError: + pass + if server is not None: + server.stop() + + +class _FrameServer: + """The socket the robot dials, run on a thread so the loop above stays in charge.""" + + def __init__(self, port: int, on_frame: Callable[[bytes], None]): + self._port = port + self._on_frame = on_frame + self._server: Any = None + self._thread: threading.Thread | None = None + + def start(self) -> None: + from websockets.sync.server import serve + + ready = threading.Event() + + def run() -> None: + def handler(connection: Any) -> None: + for message in connection: + if isinstance(message, bytes): + self._on_frame(message) + + with serve(handler, "0.0.0.0", self._port) as server: + self._server = server + ready.set() + server.serve_forever() + + self._thread = threading.Thread(target=run, name="duck-frames", daemon=True) + self._thread.start() + # Bound before telling the robot where to dial, or the first connection is refused and + # `stream.rs` backs off before anybody is listening. + ready.wait(timeout=5) + + def stop(self) -> None: + if self._server is not None: + self._server.shutdown() + + +def _address_the_robot_can_reach(url: str) -> str: + """Our address on the route to the robot. + + Not `localhost`: the robot dials this, so it has to be the address *it* would use, which on + any machine with more than one interface is not something a caller should have to work out. + """ + import socket + from urllib.parse import urlparse + + host = urlparse(url).hostname or "127.0.0.1" + probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + probe.connect((host, 80)) + return str(probe.getsockname()[0]) + finally: + probe.close() diff --git a/sdk/duck/view.py b/sdk/duck/view.py new file mode 100644 index 00000000..652467f0 --- /dev/null +++ b/sdk/duck/view.py @@ -0,0 +1,114 @@ +"""What the robot can see right now, as one object. + +A robot program is a loop — look, decide, act — and the three things it looks at arrive on three +different schedules: `robot.state` at whatever rate it was asked for, depth at 15 Hz, JPEG frames +at whatever `media.stream` was told. Merging those is the part every caller would otherwise write +for itself, and it is the reason the surface below `Duck.watch` exists at all. + +**The merge is last-one-wins, not a queue.** A behaviour wants the freshest reading, not every +reading: a frame from two ticks ago is a frame from somewhere the robot no longer is, and a +program that falls behind should skip rather than accumulate a backlog it will act on late. + +**Nothing here interprets a picture.** `frame` is the JPEG bytes as they arrived, and what is in +it is the caller's model to run — an SDK that shipped one would be a much larger dependency, and a +worse guess than whatever the caller already has. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +#: Status codes ST documents as a usable range: valid, and valid with a large pulse. The same two +#: `tof::STATUS_VALID` carries — the protocol says consumers should use `Frame::zone`'s rules +#: rather than re-deriving the thresholds, so these are those rules and not new ones. +STATUS_VALID = (5, 9) + +#: "Measured, and nothing is there." Distinct from a failed measurement, which is why the wire +#: carries the status byte rather than a magic distance. +STATUS_NO_TARGET = 255 + + +@dataclass +class View: + """One tick's worth of everything, whatever arrived most recently. + + Every field can be `None` or empty: a robot with no camera has no frame, a duck whose ToF is + not fitted has no depth, and the first tick of a loop may have neither yet. A behaviour that + checks is a behaviour that survives a sensor going away, which on a real robot happens. + """ + + #: The last JPEG, exactly as the robot sent it. `None` until one arrives. + frame: bytes | None = None + + #: The last `robot.state`: joints, odometry, the loop's rate, whether it has fallen. + state: dict[str, Any] | None = None + + #: The last depth frame, as the wire carries it — millimetres and ST's status byte. + depth: dict[str, Any] | None = None + + #: Seconds since the loop started, so a behaviour can time itself without a clock of its own. + elapsed: float = 0.0 + + #: Ticks so far, starting at one. + tick: int = 0 + + #: Where a zone is out of range or failed, `ranges` holds `None` in its place. + _ranges: list[float | None] = field(default_factory=list, repr=False) + + # ── depth, interpreted ─────────────────────────────────────────────────── + + @property + def ranges(self) -> list[float | None]: + """Every depth zone in metres, row-major, `None` where there is no usable range. + + The interpretation is `tof::Frame::zone`'s, including the part that is not obvious: a + negative distance comes back from the sensor on a failed convergence and is not a range + whatever the status byte says. + """ + if self._ranges: + return self._ranges + if not self.depth: + return [] + distances = self.depth.get("distance_mm") or [] + statuses = self.depth.get("status") or [] + out: list[float | None] = [] + for i, distance in enumerate(distances): + status = statuses[i] if i < len(statuses) else STATUS_NO_TARGET + out.append(distance / 1000.0 if status in STATUS_VALID and distance > 0 else None) + self._ranges = out + return out + + @property + def nearest(self) -> float | None: + """The closest thing the depth sensor can see, in metres, or `None` if it sees nothing. + + The one number an avoidance behaviour wants. `None` means no zone had a usable range — + an empty room and an unfitted sensor look the same from here, which is why `depth` is + there for a caller that needs to tell them apart. + """ + seen = [r for r in self.ranges if r is not None] + return min(seen) if seen else None + + # ── state, unwrapped ───────────────────────────────────────────────────── + + @property + def fallen(self) -> bool: + """Whether the robot is down. `False` when nothing has said yet — a behaviour should not + act on a fall it has no evidence for.""" + return bool((self.state or {}).get("safety", {}).get("fallen", False)) + + @property + def position(self) -> tuple[float, float, float] | None: + """Where contact odometry believes the robot is, metres, in the frame it booted in.""" + odom = (self.state or {}).get("odom") + if not odom or "position" not in odom: + return None + x, y, z = odom["position"] + return (x, y, z) + + @property + def yaw(self) -> float | None: + """Which way the robot is facing, radians, relative to where it booted.""" + odom = (self.state or {}).get("odom") + return odom.get("yaw") if odom else None diff --git a/sdk/examples/wander_without_bumping_into_things.py b/sdk/examples/wander_without_bumping_into_things.py new file mode 100644 index 00000000..28784423 --- /dev/null +++ b/sdk/examples/wander_without_bumping_into_things.py @@ -0,0 +1,52 @@ +"""A duck that walks around a room and does not walk into it. + + python examples/wander_without_bumping_into_things.py robot.local + +The whole behaviour is the loop below. Everything it reads — depth, state, whether the robot has +fallen — arrives on a different stream at a different rate, and `watch` is what makes that one +object per tick instead of three subscriptions to merge by hand. +""" + +import sys + +from duck import Duck + +#: How close something has to be before turning away from it, metres. The ToF sees about 2 m, so +#: this is "in the way" rather than "visible". +TOO_CLOSE = 0.4 + +#: Metres per second, and radians per second. A duck's top speed is not the interesting part of a +#: wander; being able to stop is. +WALK = 0.12 +TURN = 0.8 + + +def main(host: str) -> None: + with Duck(host) as duck: + duck.enable(True) + + for view in duck.watch(fps=4, camera=False): + if view.fallen: + print("down — stopping and letting somebody pick it up") + duck.stop() + break + + near = view.nearest + if near is None: + # No usable range: either nothing in front or no sensor. Walking on a reading + # that does not exist is how a duck finds a wall with its face, so it turns. + duck.move(vyaw=TURN) + elif near < TOO_CLOSE: + print(f"{near:.2f} m ahead, turning") + duck.move(vyaw=TURN) + else: + duck.move(vx=WALK) + + if view.elapsed > 60: + print("that is enough for now") + duck.stop() + break + + +if __name__ == "__main__": + main(sys.argv[1] if len(sys.argv) > 1 else "robot.local") diff --git a/sdk/microduck.egg-info/PKG-INFO b/sdk/microduck.egg-info/PKG-INFO deleted file mode 100644 index 3bce69f1..00000000 --- a/sdk/microduck.egg-info/PKG-INFO +++ /dev/null @@ -1,55 +0,0 @@ -Metadata-Version: 2.4 -Name: microduck -Version: 0.1.0 -Summary: Drive a Microduck from a script, over the agent WebSocket -License-Expression: Apache-2.0 -Requires-Python: >=3.10 -Description-Content-Type: text/markdown -Requires-Dist: websockets>=13 - -# Driving a duck from a script - -```python -from duck import Duck - -with Duck("robot.local") as duck: - print(duck.health()["healthy"]) - duck.move(vx=0.1) -``` - -`pip install -e .`, and the robot needs `mediad` running — the socket is `/agent` on the port the -console is served from, 8080 by default. - -## What it is - -One WebSocket, the same JSON-RPC every other transport speaks, and no media stack. -`architecture.md` §5.3 is the argument for why a program should not have to negotiate ICE and -decode H.264 to send an intent. - -It is deliberately small. Every method is one call, and the ones that exist are the ones -`mediad::route` permits — so what a script may do is what a browser on the LAN may do, decided in -one place on the robot rather than twice. - -## Frames - -The robot does not serve frames, it **sends** them to a socket you open: - -```python -from duck import Duck, receive -import threading - -threading.Thread(target=receive, args=(8099, print_frame), daemon=True).start() -with Duck("robot.local") as duck: - duck.frames(url="ws://192.168.1.20:8099", fps=1) -``` - -That direction is the point. A robot behind a home router and a script anywhere else cannot pair -without a relay candidate, and the robot dialling out means NAT is not a participant. -`mediad/src/stream.rs` has the whole argument. - -`examples/fetch_a_frame_and_send_an_intent.py` is both halves in about thirty lines. - -## What it does not do - -Live video. A viewer wants WebRTC and the console already is one. This is for when the consumer -is a program. diff --git a/sdk/microduck.egg-info/SOURCES.txt b/sdk/microduck.egg-info/SOURCES.txt deleted file mode 100644 index 9e66e5ba..00000000 --- a/sdk/microduck.egg-info/SOURCES.txt +++ /dev/null @@ -1,10 +0,0 @@ -README.md -pyproject.toml -duck/__init__.py -duck/client.py -duck/rpc.py -microduck.egg-info/PKG-INFO -microduck.egg-info/SOURCES.txt -microduck.egg-info/dependency_links.txt -microduck.egg-info/requires.txt -microduck.egg-info/top_level.txt \ No newline at end of file diff --git a/sdk/microduck.egg-info/dependency_links.txt b/sdk/microduck.egg-info/dependency_links.txt deleted file mode 100644 index 8b137891..00000000 --- a/sdk/microduck.egg-info/dependency_links.txt +++ /dev/null @@ -1 +0,0 @@ - diff --git a/sdk/microduck.egg-info/requires.txt b/sdk/microduck.egg-info/requires.txt deleted file mode 100644 index 9473ab6d..00000000 --- a/sdk/microduck.egg-info/requires.txt +++ /dev/null @@ -1 +0,0 @@ -websockets>=13 diff --git a/sdk/microduck.egg-info/top_level.txt b/sdk/microduck.egg-info/top_level.txt deleted file mode 100644 index ed891705..00000000 --- a/sdk/microduck.egg-info/top_level.txt +++ /dev/null @@ -1 +0,0 @@ -duck diff --git a/sdk/tests/test_view.py b/sdk/tests/test_view.py new file mode 100644 index 00000000..821a99f1 --- /dev/null +++ b/sdk/tests/test_view.py @@ -0,0 +1,69 @@ +"""What a behaviour reads off a tick. + +The depth rules are `tof::Frame::zone`'s, and these hold the Python to them: the protocol says a +consumer should use that interpretation rather than re-deriving the thresholds, which only means +anything if somebody checks the copy. +""" + +from duck.view import STATUS_NO_TARGET, View + + +def depth(distances, statuses): + return {"rows": 1, "cols": len(distances), "distance_mm": distances, "status": statuses} + + +def test_a_tick_with_nothing_in_it_is_readable(): + """The first tick of any loop, and every tick on a duck with no sensors. A behaviour that + has to guard every field would be a behaviour nobody writes correctly.""" + view = View() + assert view.frame is None + assert view.ranges == [] + assert view.nearest is None + assert view.position is None + assert view.fallen is False + + +def test_only_the_two_valid_statuses_are_a_range(): + """5 and 9 are what ST documents as usable — valid, and valid with a large pulse. Everything + else is the sensor saying it did not measure, and reporting it as a distance would put a + number a behaviour acts on where there is no measurement at all.""" + view = View(depth=depth([400, 800, 1200, 300], [5, 9, 255, 4])) + assert view.ranges == [0.4, 0.8, None, None] + assert view.nearest == 0.4 + + +def test_a_negative_distance_is_not_a_range_whatever_the_status_says(): + """The non-obvious half of `Frame::zone`: the sensor returns negative distances on a failed + convergence and still flags them valid. Taken at face value they are the nearest thing in + the room, so an avoidance behaviour would stop for something that is not there.""" + view = View(depth=depth([-120, 600], [5, 5])) + assert view.ranges == [None, 0.6] + assert view.nearest == 0.6 + + +def test_an_empty_room_and_an_absent_sensor_both_read_as_nothing_seen(): + """Both are `nearest is None`, deliberately — a behaviour does the same thing either way. + `depth` is still there for a caller that needs to tell them apart.""" + empty = View(depth=depth([0, 0], [STATUS_NO_TARGET, STATUS_NO_TARGET])) + assert empty.nearest is None and empty.depth is not None + absent = View() + assert absent.nearest is None and absent.depth is None + + +def test_state_is_unwrapped_where_a_behaviour_would_reach_for_it(): + view = View( + state={ + "safety": {"fallen": True}, + "odom": {"position": [1.0, 2.0, 0.1], "yaw": 0.5}, + } + ) + assert view.fallen is True + assert view.position == (1.0, 2.0, 0.1) + assert view.yaw == 0.5 + + +def test_a_robot_that_has_not_said_it_fell_has_not_fallen(): + """`False` rather than `None`, because a behaviour should not act on a fall it has no + evidence for — and `if view.fallen` is what everybody will write.""" + assert View(state={}).fallen is False + assert View().fallen is False