From 6a0d80a3814696f0ddcd7ce77bdda8731258fd62 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Thu, 15 Oct 2026 23:03:02 +0000 Subject: [PATCH] Generate go2rtc's config from what the printers report go2rtc has been supervised since before this commit, with nothing to serve: `CONTINUUM_GO2RTC_CONFIG` pointed at `./go2rtc.yaml`, and that file did not exist in the repository or anywhere else. A watchdog was restarting a process that had no cameras configured. `camera.rs` writes it, from the `rtsp_url` a printer publishes in its own `print.push_status` `ipcam` block rather than from `printers.toml`. The URL is already something the machine tells us, so asking an operator to keep a second copy of it - and to update it by hand when a printer's address changes - would be asking for drift. Two tasks, not one: the builder decides what should be served, the watchdog keeps the thing serving it alive. They meet at one `Notify`, because go2rtc reads its configuration only at startup, so a changed config needs a restart. The watchdog now selects the child's exit against that notification and skips its backoff on a deliberate change. What this does *not* cover is most printers, and that is the honest limit. Only the X1/H2 class publishes an RTSP URL - confirmed against two real machines: the H2C reports `rtsps://192.168.0.21:322/streaming/live/1`, the P1P reports an `ipcam` block with `ipcam_dev: "1"` and no URL at all. "A camera exists" and "a camera you can connect to" are different facts. A P1/A1 needs Bambu's proprietary port-6000 JPEG stream, which this crate does not speak, so those printers report no `camera_stream` and a UI should offer a player only when one is present. Details worth keeping: - The config is rendered sorted so an unchanged fleet is byte-identical, and on the first pass it is compared against the file on disk - without that, every daemon restart rewrites an identical file and drops every live viewer. - Written through a `.tmp` and renamed. The watchdog can restart go2rtc at any moment, and a half-written config fails to parse, which takes every working stream down rather than one. - Never logged, and gitignored: the credentials go in the URL because that is the only place go2rtc takes them for an RTSP source, so the file holds printer access codes. - A reported URL with no scheme is refused rather than repaired. One bad entry makes go2rtc fail to load the whole file. - No `rtsp:` or `webrtc:` block. The API listener is what serves WebRTC to a browser, which is the only consumer here; opening go2rtc's own RTSP server would publish every printer's camera on the LAN unauthenticated. - `stream_base_url` is reported in `hello`, not derived from the listen address: `:1984` is a bind address and not something a browser can open. Unset means "advertise nothing", which is the right default, because a wrong URL makes the UI offer a player that cannot connect. It is a LAN address either way - there is no relay, so this works for an operator in the workshop and not for anyone outside it. 63 tests, clippy unchanged at the two deliberate variant-size warnings, `cargo deny check licenses` ok. go2rtc itself stays operator-installed and is not distributed. --- .env.example | 13 +- .gitignore | 5 + README.md | 16 +- src/camera.rs | 324 ++++++++++++++++++++++++++++ src/config.rs | 18 ++ src/go2rtc.rs | 44 +++- src/lib.rs | 7 +- src/main.rs | 38 +++- src/printer/bambu_commands/mod.rs | 2 +- src/printer/bambu_commands/print.rs | 28 +++ src/printer/mod.rs | 2 +- src/uplink/mod.rs | 11 + src/uplink/protocol.rs | 27 ++- tests/uplink_protocol.rs | 34 +++ 14 files changed, 542 insertions(+), 27 deletions(-) create mode 100644 src/camera.rs diff --git a/.env.example b/.env.example index d87f092..6ab6529 100644 --- a/.env.example +++ b/.env.example @@ -6,9 +6,20 @@ CONTINUUM_GATEWAY_TOKEN=replace-with-provisioning-token # Cloud uplink CONTINUUM_UPLINK_URL=wss://api.continuum.local/ws/edge/v1 -# go2rtc process management +# Camera restreaming. The config file is GENERATED by this daemon from what +# the printers themselves report - don't write it by hand, and keep it out of +# version control: it holds printer access codes. CONTINUUM_GO2RTC_BIN=/usr/local/bin/go2rtc CONTINUUM_GO2RTC_CONFIG=./go2rtc.yaml +CONTINUUM_GO2RTC_API_LISTEN=:1984 + +# What a browser can reach the streams at, reported to the cloud. A bind +# address is not one, which is why this is separate from the listen address +# above; only you know this gateway's reachable name. Note this is a LAN +# address: a browser on this network can play the stream, one outside cannot. +# Leave it unset to advertise no streams at all, which is the safe default - +# a wrong URL here makes the UI offer a player that cannot connect. +# CONTINUUM_STREAM_BASE_URL=http://gateway.local:1984 # Fleet config - see printers.example.toml. Bambu MQTTS certificate trust # (per-printer ca_cert_path/require_valid_cert, and automatic diff --git a/.gitignore b/.gitignore index 08bd6ac..428f1b3 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,8 @@ Cargo.lock *.log printers.toml /certs/pinned/ + +# Generated by the daemon from what the printers report, and it holds their +# access codes. +go2rtc.yaml +go2rtc.yaml.tmp diff --git a/README.md b/README.md index 46e83c1..a574d00 100644 --- a/README.md +++ b/README.md @@ -27,8 +27,15 @@ cloud control plane (`continuum-backend`). - **Cloud uplink** - a WebSocket client to `continuum-backend` that connects, sends a heartbeat on a timer, and reconnects on drop (`src/uplink/`). -- **go2rtc watchdog** - restarts the camera-restreaming subprocess if it - dies (`src/go2rtc.rs`). +- **Camera restreaming** - generates go2rtc's configuration from the camera + URLs the printers report, and restarts go2rtc when it changes + (`src/camera.rs` + `src/go2rtc.rs`). Only the X1/H2 class publishes an + RTSP URL: confirmed against two real machines, an H2C reports + `rtsps://:322/streaming/live/1` and a P1P reports a camera with no URL + for it. A P1/A1 camera needs Bambu's proprietary port-6000 JPEG stream, + which this crate does not speak, so those printers advertise no stream + rather than one that will not play. Note the stream is reachable on the + gateway's LAN only - there is no relay. - **Bambu file upload** - implicit FTPS (port 990) into the printer's own `cache/` directory (`src/printer/bambu_ftps.rs`, via `BambuPrinter::upload_file`). **Confirmed live against a P1P on @@ -97,11 +104,12 @@ cargo run --example bambu_commands_demo # round-trips hand-written structs ag ``` src/ - main.rs Loads+connects the printer fleet, then runs the uplink and go2rtc watchdog side by side + main.rs Loads+connects the printer fleet, then runs the uplink, camera config, watchdog and completion watcher side by side config.rs Loads settings from environment variables fleet.rs Loads printers.toml into ready-to-use PrinterHandles uplink/ WebSocket client to continuum-backend: connect, heartbeat, reconnect on drop - go2rtc.rs Restarts the go2rtc camera-restreaming process if it dies + camera.rs Writes go2rtc's config from the camera URLs the printers report + go2rtc.rs Restarts the go2rtc camera-restreaming process if it dies, or when the config changes printer/ mod.rs GenericPrinter trait + PrinterBase + PrinterHandle enum - the trait+enum "inheritance" pattern this project uses instead of class inheritance diff --git a/src/camera.rs b/src/camera.rs new file mode 100644 index 0000000..95252c6 --- /dev/null +++ b/src/camera.rs @@ -0,0 +1,324 @@ +//! Turns the printers' own camera URLs into a go2rtc configuration. +//! +//! `go2rtc.rs` has supervised a go2rtc process since before this module +//! existed, and go2rtc has never had anything to serve: nothing wrote its +//! config file. `CONTINUUM_GO2RTC_CONFIG` pointed at `./go2rtc.yaml`, which +//! did not exist in the repository or anywhere else. This writes it. +//! +//! ## Where the URL comes from +//! +//! Not from `printers.toml`. A printer reports its own camera URL in +//! `print.push_status`'s `ipcam` block, so asking an operator to copy it into +//! a config file would be asking them to keep a second copy of something the +//! machine already tells us - and to update it by hand when the printer's +//! address changes. +//! +//! ## What does not work, which is most printers +//! +//! Only the X1/H2 class publishes an `rtsp_url`. Confirmed against two real +//! machines: an H2C reports `rtsps://:322/streaming/live/1`, and a P1P +//! reports an `ipcam` block with no URL in it at all. A P1/A1 camera is +//! reachable only over Bambu's proprietary port-6000 JPEG stream, which this +//! crate does not speak, so those printers get no stream here and say so +//! rather than advertising one that will not play. +//! +//! That is also why this module reads the *reported* URL and never builds one +//! from the printer's IP: the fact that a printer has a camera and the fact +//! that the camera has an RTSP endpoint are different facts, and only the +//! second one is useful. + +use std::collections::BTreeMap; +use std::time::Duration; + +use tokio::sync::Notify; +use tracing::{info, warn}; + +use crate::printer::{GenericPrinter, PrinterHandle}; +use crate::uplink::SharedFleet; + +/// How often to rebuild the configuration from what the printers report. +/// +/// A camera URL changes when a printer's address changes, which is a DHCP +/// lease event, not something that needs watching closely. The first useful +/// rebuild is the one a few seconds after startup, once the first reports +/// have arrived - before that the fleet has nothing to say. +const REBUILD_INTERVAL: Duration = Duration::from_secs(30); + +/// The username every Bambu uses for its own services - MQTT, FTPS, and the +/// camera. Not a guess: it is what `bambu_mqtt` and `bambu_ftps` already +/// authenticate with. +const CAMERA_USER: &str = "bblp"; + +/// One printer's playable stream. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CameraSource { + /// Our own printer id, which doubles as the go2rtc stream name. Using it + /// for both means the cloud can name a stream without the gateway having + /// to report a mapping: `camera_stream` in telemetry is this id. + pub stream_name: String, + /// The full source URL, credentials included. + pub url: String, +} + +/// Builds the source URL for one printer, or `None` if it has no stream. +/// +/// The credentials go in the URL because that is the only place go2rtc takes +/// them for an RTSP source. The consequence is that the printer's access code +/// ends up in a file on disk - which is not a new exposure (`printers.toml` +/// already holds it, and that file is gitignored for this reason) but it does +/// mean the generated config deserves the same treatment, and must never be +/// logged. +pub fn source_for(printer: &PrinterHandle, report_url: Option<&str>) -> Option { + let PrinterHandle::Bambu(bambu) = printer else { + // Nothing else has a camera path here yet. A Prusa's is a separate + // product, and a Klipper camera is whatever the user configured. + return None; + }; + + let url = report_url?; + if url.is_empty() { + return None; + } + + // `rtsps://1.2.3.4:322/streaming/live/1` -> `rtsps://bblp:code@1.2.3.4:322/...` + let with_credentials = match url.split_once("://") { + Some((scheme, rest)) => { + // A URL that already carries credentials is left alone rather + // than having a second set spliced in. + if rest.contains('@') { + url.to_string() + } else { + format!("{scheme}://{CAMERA_USER}:{}@{rest}", bambu.access_code) + } + } + // Not a URL shape this understands. Refused rather than guessed at: + // a malformed source makes go2rtc fail to load the whole file, which + // would take every *working* stream down with it. + None => { + warn!( + printer = %bambu.display_name(), + "the printer reported a camera URL with no scheme; ignoring it" + ); + return None; + } + }; + + Some(CameraSource { + stream_name: printer.base().id.clone(), + url: with_credentials, + }) +} + +/// Renders a go2rtc configuration. +/// +/// Hand-written YAML rather than a serializer, for the same reason the +/// `.3mf` reader is hand-rolled: it is a fixed, tiny shape, and a dependency +/// for ten lines of output is a dependency to license and audit. The values +/// are quoted and the keys are our own printer ids, which `fleet.rs` +/// constrains. +/// +/// `listen` is explicit rather than left to go2rtc's defaults, because the +/// cloud is told this address and a default that changes between go2rtc +/// versions would change where the stream lives without anything saying so. +pub fn render_config(api_listen: &str, sources: &[CameraSource]) -> String { + let mut out = String::from( + "# Generated by continuum-proxy. Do not edit: it is rewritten from\n\ + # what the printers themselves report, so an edit here is lost on\n\ + # the next rebuild. Contains printer access codes - treat it as a\n\ + # secret and keep it out of version control.\n", + ); + + // Sorted, so an unchanged fleet renders byte-identical output and the + // change detection below does not restart go2rtc on every rebuild. + let ordered: BTreeMap<&str, &str> = sources + .iter() + .map(|s| (s.stream_name.as_str(), s.url.as_str())) + .collect(); + + out.push_str("streams:\n"); + if ordered.is_empty() { + // An empty mapping, written explicitly. `streams:` with nothing + // under it is a null in YAML, and go2rtc is entitled to treat that + // differently from an empty map. + out.push_str(" {}\n"); + } + for (name, url) in &ordered { + out.push_str(&format!(" {name}: \"{url}\"\n")); + } + + out.push_str(&format!("api:\n listen: \"{api_listen}\"\n")); + // No `rtsp:` or `webrtc:` block: the API listener is what serves WebRTC + // to a browser, which is the only consumer here. Opening go2rtc's own + // RTSP server would publish every printer's camera on the LAN + // unauthenticated, which is not something to do by default. + out +} + +/// Collects every printer that currently has a playable stream. +pub async fn sources(fleet: &SharedFleet) -> Vec { + let fleet = fleet.lock().await; + let mut found = Vec::new(); + + for printer in fleet.iter() { + let report_url = printer + .print_status() + .await + .and_then(|report| report.ipcam) + .and_then(|ipcam| ipcam.rtsp_url); + + if let Some(source) = source_for(printer, report_url.as_deref()) { + found.push(source); + } + } + + found +} + +/// Rewrites the config whenever what the printers report changes, and asks +/// the watchdog to restart go2rtc when it does. +/// +/// Restart rather than reload: go2rtc can reload some of its configuration +/// over its own API, but a restart is the behaviour we can be sure of, and +/// the watchdog was already built to restart it. A stream interruption at the +/// moment a printer's camera URL changes is acceptable; a stream that is +/// silently serving the old address is not. +pub async fn run( + fleet: SharedFleet, + config_path: String, + api_listen: String, + restart: std::sync::Arc, +) { + let mut ticker = tokio::time::interval(REBUILD_INTERVAL); + let mut last_written: Option = None; + + loop { + ticker.tick().await; + + let desired = render_config(&api_listen, &sources(&fleet).await); + if last_written.as_deref() == Some(desired.as_str()) { + continue; + } + + // Compare against the file too, not just our own memory of it: on + // startup `last_written` is `None`, and rewriting an identical file + // would restart go2rtc - and therefore drop every live viewer - for + // no reason, every time the daemon restarts. + if last_written.is_none() + && let Ok(existing) = tokio::fs::read_to_string(&config_path).await + && existing == desired + { + last_written = Some(desired); + continue; + } + + match write_atomically(&config_path, &desired).await { + Ok(()) => { + let count = desired.matches("\n ").count().saturating_sub(1); + // The content is never logged: it holds access codes. + info!(path = %config_path, streams = count, "wrote go2rtc config, restarting it"); + last_written = Some(desired); + restart.notify_one(); + } + Err(err) => warn!(path = %config_path, ?err, "could not write the go2rtc config"), + } + } +} + +/// Writes through a temporary file and renames. +/// +/// go2rtc may be reading this file at any moment - the watchdog restarts it +/// on a five-second loop - and a partially written config is a config that +/// fails to parse, which takes every working stream down. A rename is atomic +/// on the same filesystem, so a reader sees either the old file or the new +/// one. +async fn write_atomically(path: &str, contents: &str) -> std::io::Result<()> { + let temp = format!("{path}.tmp"); + tokio::fs::write(&temp, contents).await?; + tokio::fs::rename(&temp, path).await +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::printer::{BambuPrinter, BambuProtocolPreference, BambuTls}; + + fn printer(id: &str) -> PrinterHandle { + PrinterHandle::Bambu(BambuPrinter::new( + id, + "Test Printer", + "192.168.0.21", + "12345678", + "SN123", + BambuTls::Insecure, + BambuProtocolPreference::Auto, + )) + } + + /// The URL shape confirmed from a real H2C's `ipcam` block. + #[test] + fn a_reported_rtsps_url_gets_the_printer_credentials_spliced_in() { + let source = source_for(&printer("h2c"), Some("rtsps://192.168.0.21:322/streaming/live/1")) + .expect("an rtsps URL is playable"); + assert_eq!(source.stream_name, "h2c"); + assert_eq!(source.url, "rtsps://bblp:12345678@192.168.0.21:322/streaming/live/1"); + } + + /// What a P1P reports: an `ipcam` block that says a camera exists without + /// giving anything to connect to. Advertising a stream here would make a + /// UI offer a player that can never load. + #[test] + fn a_printer_that_reports_no_url_has_no_stream() { + assert_eq!(source_for(&printer("p1p"), None), None); + assert_eq!(source_for(&printer("p1p"), Some("")), None); + } + + /// Splicing a second set of credentials into a URL that already has some + /// would produce `bblp:code@user:pass@host`, which is not a host. + #[test] + fn a_url_that_already_carries_credentials_is_left_alone() { + let source = source_for(&printer("h2c"), Some("rtsps://someone:else@10.0.0.5:322/live")) + .expect("still playable"); + assert_eq!(source.url, "rtsps://someone:else@10.0.0.5:322/live"); + } + + /// Refused rather than guessed at, because one bad entry makes go2rtc + /// fail to load the whole file and takes every working stream with it. + #[test] + fn a_url_with_no_scheme_is_refused_rather_than_repaired() { + assert_eq!(source_for(&printer("h2c"), Some("192.168.0.21:322/streaming/live/1")), None); + } + + /// `streams:` with nothing under it is YAML for null, not for an empty + /// map, and a gateway with no streamable camera is the normal state of a + /// P1-only farm - so this is the output that must be valid, not a corner. + #[test] + fn an_empty_fleet_renders_an_explicit_empty_map() { + let config = render_config(":1984", &[]); + assert!(config.contains("streams:\n {}\n"), "got:\n{config}"); + assert!(config.contains("listen: \":1984\""), "got:\n{config}"); + } + + /// The change detection in `run` compares rendered output, so an + /// unchanged fleet in a different order must render identically or + /// go2rtc gets restarted - dropping every live viewer - every 30s. + #[test] + fn the_same_fleet_in_a_different_order_renders_identically() { + let a = CameraSource { stream_name: "aaa".into(), url: "rtsps://one".into() }; + let b = CameraSource { stream_name: "zzz".into(), url: "rtsps://two".into() }; + assert_eq!( + render_config(":1984", &[a.clone(), b.clone()]), + render_config(":1984", &[b, a]), + ); + } + + /// Opening go2rtc's own RTSP or WebRTC listener would publish every + /// printer's camera on the LAN unauthenticated. The absence is the + /// decision, so it is worth a test that fails if someone adds one. + #[test] + fn the_generated_config_opens_no_server_but_the_api() { + let source = CameraSource { stream_name: "h2c".into(), url: "rtsps://x".into() }; + let config = render_config(":1984", &[source]); + assert!(!config.contains("rtsp:"), "got:\n{config}"); + assert!(!config.contains("webrtc:"), "got:\n{config}"); + } +} diff --git a/src/config.rs b/src/config.rs index 352a42f..4c00cae 100644 --- a/src/config.rs +++ b/src/config.rs @@ -8,6 +8,22 @@ pub struct Config { pub uplink_url: String, pub go2rtc_bin: String, pub go2rtc_config: String, + /// What go2rtc's API listens on, which is also what serves WebRTC to a + /// browser. Written into the generated config rather than left to + /// go2rtc's default, because the cloud is told this address and a + /// default that moves between go2rtc versions would move the stream + /// without anything saying so. + pub go2rtc_api_listen: String, + /// The base URL a browser can reach this gateway's streams at, reported + /// to the cloud in `hello`. + /// + /// Not derived from `go2rtc_api_listen`: that is a bind address, often + /// `:1984` or `0.0.0.0:1984`, and neither is something a browser can + /// open. Only the operator knows the gateway's reachable name. Unset + /// means "do not advertise a stream", which is the right default - a + /// wrong URL here is worse than none, because a UI would offer a player + /// that cannot connect. + pub stream_base_url: Option, /// Path to the fleet config file - see `fleet.rs` and /// `printers.example.toml` for the shape. Bambu certificate trust /// (per-printer `ca_cert_path`/`require_valid_cert`, and automatic @@ -26,6 +42,8 @@ impl Config { uplink_url: env_or("CONTINUUM_UPLINK_URL", "wss://api.continuum.local/ws/edge/v1"), go2rtc_bin: env_or("CONTINUUM_GO2RTC_BIN", "go2rtc"), go2rtc_config: env_or("CONTINUUM_GO2RTC_CONFIG", "./go2rtc.yaml"), + go2rtc_api_listen: env_or("CONTINUUM_GO2RTC_API_LISTEN", ":1984"), + stream_base_url: env::var("CONTINUUM_STREAM_BASE_URL").ok().filter(|v| !v.is_empty()), printers_path: env_or("CONTINUUM_PRINTERS_PATH", "printers.toml"), }) } diff --git a/src/go2rtc.rs b/src/go2rtc.rs index d2fd31b..8a6925d 100644 --- a/src/go2rtc.rs +++ b/src/go2rtc.rs @@ -1,12 +1,21 @@ use std::process::Stdio; +use std::sync::Arc; use std::time::Duration; use tokio::process::Command; +use tokio::sync::Notify; use tracing::{error, info, warn}; -/// Supervises the go2rtc subprocess used for camera restreaming (RTSP/USB -> WebRTC). -/// Restarts it with a fixed backoff whenever it exits, for as long as the daemon runs. -pub async fn watchdog(bin: String, config_path: String) { +/// Supervises the go2rtc subprocess used for camera restreaming +/// (RTSP -> WebRTC). Restarts it with a fixed backoff whenever it exits, and +/// immediately when `restart` fires. +/// +/// `restart` is what `camera::run` rings after rewriting the config: go2rtc +/// reads its configuration once at startup, so a config that changed while it +/// was running would otherwise be ignored until the process happened to die. +/// Before that existed this watchdog supervised a process whose config file +/// nothing wrote - it restarted go2rtc forever over a file that did not exist. +pub async fn watchdog(bin: String, config_path: String, restart: Arc) { loop { info!(%bin, %config_path, "starting go2rtc"); @@ -19,10 +28,31 @@ pub async fn watchdog(bin: String, config_path: String) { .spawn(); match spawned { - Ok(mut child) => match child.wait().await { - Ok(status) => warn!(%status, "go2rtc exited, restarting after backoff"), - Err(err) => error!(?err, "failed to wait on go2rtc process"), - }, + Ok(mut child) => { + tokio::select! { + // Whichever happens first. `kill_on_drop` means the + // child dies when `child` goes out of scope at the end + // of this arm, so the restart path needs no explicit + // kill - and cannot leave an orphan serving the old + // config. + waited = child.wait() => match waited { + Ok(status) => warn!(%status, "go2rtc exited, restarting after backoff"), + Err(err) => error!(?err, "failed to wait on go2rtc process"), + }, + _ = restart.notified() => { + info!("go2rtc config changed, restarting it"); + // Explicitly, rather than relying on the drop: this + // awaits the kill so the next iteration cannot race + // the old process for the API port. + if let Err(err) = child.kill().await { + warn!(?err, "could not stop go2rtc for a config change"); + } + // No backoff on this path - a config change is a + // deliberate act, not a crash loop. + continue; + } + } + } Err(err) => { error!(?err, "failed to spawn go2rtc, is it installed and on PATH?"); } diff --git a/src/lib.rs b/src/lib.rs index 7694c2b..3b33232 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,9 +3,12 @@ //! integration tests in `tests/` use it too. //! //! Everything the daemon needs lives here rather than in the binary, so it is -//! compiled once. `go2rtc` is the one exception: the camera-restreaming -//! watchdog is the daemon's own concern and nothing else calls it. +//! compiled once. `go2rtc` is the one exception: supervising that subprocess +//! is the daemon's own concern and nothing else calls it. Deciding *what* it +//! should serve is not - that is `camera`, which reads the fleet and is worth +//! testing, so it lives here. +pub mod camera; pub mod completion; pub mod config; pub mod fleet; diff --git a/src/main.rs b/src/main.rs index 7053fc8..ccf7703 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,7 +10,7 @@ use tracing_subscriber::EnvFilter; use continuum_proxy::config::Config; use continuum_proxy::printer::{GenericPrinter, PrinterHandle}; -use continuum_proxy::{completion, fleet, uplink}; +use continuum_proxy::{camera, completion, fleet, uplink}; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -36,24 +36,42 @@ async fn main() -> anyhow::Result<()> { // see `uplink::SharedFleet` for why that type and what it costs. let fleet: uplink::SharedFleet = std::sync::Arc::new(tokio::sync::Mutex::new(fleet)); - // Three long-running tasks: the cloud uplink, the go2rtc - // camera-restreaming watchdog, and the print-completion watcher that - // blinks a printer's lights when it finishes. Every printer's MQTT - // session is a fourth thing running concurrently, but implicitly - each - // one has its own background poll task spawned by `connect()`, kept - // alive by the fleet above. More will join these later (printer - // discovery, the plate changer). + // Four long-running tasks: the cloud uplink, the go2rtc subprocess + // watchdog, the camera-config builder that tells it what to serve, and + // the print-completion watcher that blinks a printer's lights when it + // finishes. Every printer's MQTT session is a fifth thing running + // concurrently, but implicitly - each one has its own background poll + // task spawned by `connect()`, kept alive by the fleet above. More will + // join these later (printer discovery, the plate changer). // // The completion watcher is deliberately its own task rather than part // of the uplink: it must keep working with the cloud connection down, // because the person it is for is standing next to the printer. See // `completion::run`. // - // `tokio::select!` stops as soon as any returns; all three loop forever, + // The camera builder and the watchdog are two tasks and not one because + // they answer different questions - what should be served, and keeping + // the thing that serves it alive. They meet at one `Notify`: the builder + // rings it after writing a new config, because go2rtc only reads its + // configuration at startup. + // + // `tokio::select!` stops as soon as any returns; all four loop forever, // so in practice this runs until the process is killed. + let restart_go2rtc = std::sync::Arc::new(tokio::sync::Notify::new()); + tokio::select! { _ = uplink::run(config.clone(), fleet.clone()) => {} - _ = go2rtc::watchdog(config.go2rtc_bin.clone(), config.go2rtc_config.clone()) => {} + _ = go2rtc::watchdog( + config.go2rtc_bin.clone(), + config.go2rtc_config.clone(), + restart_go2rtc.clone(), + ) => {} + _ = camera::run( + fleet.clone(), + config.go2rtc_config.clone(), + config.go2rtc_api_listen.clone(), + restart_go2rtc.clone(), + ) => {} _ = completion::run(fleet.clone()) => {} } diff --git a/src/printer/bambu_commands/mod.rs b/src/printer/bambu_commands/mod.rs index c832b99..ce6a7a1 100644 --- a/src/printer/bambu_commands/mod.rs +++ b/src/printer/bambu_commands/mod.rs @@ -27,7 +27,7 @@ mod pushall; pub use get_version::{GetVersionRequest, GetVersionResponse, ModuleInfo}; pub use print::{ - AmsState, AmsTray, AmsUnit, EmptyTray, LightReport, LoadedTray, PrintReport, VtTray, + AmsState, AmsTray, AmsUnit, EmptyTray, Ipcam, LightReport, LoadedTray, PrintReport, VtTray, XcamSettings, }; pub use pushall::PushallRequest; diff --git a/src/printer/bambu_commands/print.rs b/src/printer/bambu_commands/print.rs index 11be487..732990f 100644 --- a/src/printer/bambu_commands/print.rs +++ b/src/printer/bambu_commands/print.rs @@ -123,6 +123,9 @@ pub struct PrintReport { /// Observed on v1 despite only the v2 schema declaring it; the v1 /// schema was simply missing it. pub lights_report: Option>, + /// The printer's camera settings - and, on the models that have one, the + /// URL of its stream. See `Ipcam`. + pub ipcam: Option, // There are still ~60 more top-level fields in a real v2 payload (see // `tests/fixtures/push_status_v2_h2c_printing.json` for the full set): @@ -193,9 +196,34 @@ impl PrintReport { self.stg = newer.stg.or(self.stg.take()); self.vt_tray = newer.vt_tray.or(self.vt_tray.take()); self.lights_report = newer.lights_report.or(self.lights_report.take()); + self.ipcam = newer.ipcam.or(self.ipcam.take()); } } +/// The camera block, which is where a printer says whether it has a stream +/// anything else can connect to. +/// +/// Only `rtsp_url` is modelled of the dozen-odd fields in here, because it is +/// the only one that is actionable: the rest are recording and timelapse +/// preferences the printer manages itself. +/// +/// **It is absent on most printers, and that is the headline.** Confirmed +/// from live captures of two machines: an H2C reports +/// `rtsps://:322/streaming/live/1`, while a P1P's block carries +/// `ipcam_dev: "1"`, `ipcam_record: "enable"` and *no URL at all*. So the +/// P1/A1 class has a camera that simply cannot be reached this way - it +/// needs Bambu's proprietary port-6000 JPEG stream, which this crate does +/// not implement. `ipcam_dev: "1"` means "a camera exists", not "a camera +/// you can connect to", and reading it as the latter would have the gateway +/// advertise streams that never play. +#[derive(serde::Deserialize, Debug, Clone)] +pub struct Ipcam { + /// `rtsps://:322/streaming/live/1` on the models that have it. Note + /// it carries no credentials: a client has to add `bblp:` + /// itself. + pub rtsp_url: Option, +} + /// One light's state, as the printer reports it rather than as we asked for /// it. `mode` is the same vocabulary `ledctrl` accepts (`"on"`, `"off"`, /// `"flashing"`), and `node` the same light name - so a round-trip is diff --git a/src/printer/mod.rs b/src/printer/mod.rs index 15d4408..8efd543 100644 --- a/src/printer/mod.rs +++ b/src/printer/mod.rs @@ -49,7 +49,7 @@ pub use bambu::{ }; pub use bambu_commands::{ AmsState, AmsTray, AmsUnit, BambuReport, BambuRequest, CommandEnvelope, EmptyTray, GetVersionRequest, - GetVersionResponse, LightReport, LoadedTray, ModuleInfo, PrintReport, PushallRequest, VtTray, + GetVersionResponse, Ipcam, LightReport, LoadedTray, ModuleInfo, PrintReport, PushallRequest, VtTray, XcamSettings, }; pub use klipper::KlipperPrinter; diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index d1d6e39..f2ae581 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -79,6 +79,7 @@ async fn connect_and_serve(config: &Config, fleet: &SharedFleet) -> anyhow::Resu gateway_id: config.gateway_id.clone(), farm_id: config.farm_id.clone(), token: config.gateway_token.clone(), + stream_base_url: config.stream_base_url.clone(), }; write.send(Message::Text(serde_json::to_string(&hello)?)).await?; @@ -209,6 +210,16 @@ async fn collect_telemetry(fleet: &SharedFleet) -> Vec { bed_target: report.bed_target_temper, print_error: report.print_error, active_alerts: report.hms.as_ref().map(|h| h.len()), + // Derived from the same reported URL `camera::sources` builds + // the go2rtc config from, so a printer is advertised as + // streamable here only if it is also actually being served + // there. Asking `camera` rather than keeping a second copy of + // the rule means the two cannot drift apart. + camera_stream: crate::camera::source_for( + printer, + report.ipcam.as_ref().and_then(|c| c.rtsp_url.as_deref()), + ) + .map(|source| source.stream_name), }); } diff --git a/src/uplink/protocol.rs b/src/uplink/protocol.rs index 72f79f0..e779f1f 100644 --- a/src/uplink/protocol.rs +++ b/src/uplink/protocol.rs @@ -6,7 +6,21 @@ use crate::printer::PrintControl; #[derive(Debug, Clone, Serialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum ClientMessage { - Hello { gateway_id: String, farm_id: String, token: String }, + Hello { + gateway_id: String, + farm_id: String, + token: String, + /// Where a browser on this LAN can reach the gateway's camera + /// streams, if the operator has configured one. See + /// `Config::stream_base_url`. + /// + /// Reported in the handshake rather than per-printer in telemetry + /// because it is a property of the gateway, and repeating it in + /// every snapshot of every printer would be the same string sent + /// hundreds of times an hour. + #[serde(skip_serializing_if = "Option::is_none")] + stream_base_url: Option, + }, Heartbeat { seq: u64 }, /// One snapshot per printer, on a timer. Sent as a batch rather than a /// message per printer: a farm has one uplink and many printers, and the @@ -95,6 +109,17 @@ pub struct PrinterTelemetry { /// into messages needs Bambu's HMS tables, which we don't have yet. #[serde(skip_serializing_if = "Option::is_none")] pub active_alerts: Option, + /// The go2rtc stream name for this printer's camera, if it has one that + /// can actually be played. Combined with the farm's `stream_base_url` + /// this is everything a browser needs. + /// + /// Absent is the common case, not the exception: only the X1/H2 class + /// publishes an RTSP URL. A P1 or A1 reports that it *has* a camera + /// without giving anything to connect to, and this stays `None` for + /// those - a UI should offer a player only when this is present. See + /// `crate::camera`. + #[serde(skip_serializing_if = "Option::is_none")] + pub camera_stream: Option, } /// What `continuum-backend` sends back over the same connection. diff --git a/tests/uplink_protocol.rs b/tests/uplink_protocol.rs index 60fb2f4..7c22f87 100644 --- a/tests/uplink_protocol.rs +++ b/tests/uplink_protocol.rs @@ -185,6 +185,9 @@ fn empty_telemetry_fields_are_omitted_not_null() { bed_target: Some(0.0), print_error: Some(0), active_alerts: Some(0), + // The common case, and the third field that must be omitted: a + // P1P reports that it has a camera without giving a URL for it. + camera_stream: None, }], }; @@ -222,6 +225,7 @@ fn a_zero_stage_is_sent_rather_than_treated_as_absent() { bed_target: Some(80.0), print_error: Some(0), active_alerts: Some(0), + camera_stream: Some("p1p".into()), }], }; @@ -251,3 +255,33 @@ fn a_failed_command_result_carries_the_reason() { assert!(json.contains(r#""ok":false"#)); assert!(json.contains("no printer with id 'nope'")); } + +/// A gateway with no configured stream base URL must omit the field, for the +/// same reason the telemetry fields are omitted: the backend reads a missing +/// key and an explicit null differently, and an unconfigured gateway is the +/// default case, not an edge case. +#[test] +fn a_hello_without_a_stream_base_url_omits_it() { + let json = serde_json::to_string(&ClientMessage::Hello { + gateway_id: "gw-1".into(), + farm_id: "farm-1".into(), + token: "t".into(), + stream_base_url: None, + }) + .expect("should serialize"); + assert!(!json.contains("stream_base_url"), "got:\n {json}"); + // Without this the assertion above would also pass on an empty object. + assert!(json.contains(r#""gateway_id":"gw-1""#)); +} + +#[test] +fn a_hello_carries_the_stream_base_url_when_one_is_configured() { + let json = serde_json::to_string(&ClientMessage::Hello { + gateway_id: "gw-1".into(), + farm_id: "farm-1".into(), + token: "t".into(), + stream_base_url: Some("http://gateway.local:1984".into()), + }) + .expect("should serialize"); + assert!(json.contains(r#""stream_base_url":"http://gateway.local:1984""#), "got:\n {json}"); +}