Continuum/continuum-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Generate go2rtc's config from what the printers report

Browse Source 
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.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
octoturge
2026-10-15 23:03:02 +00:00
co-authored by Claude Opus 5
parent 48672b1f45
commit bda5e95b66
14 changed files with 542 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+12 -1
View File
@@ -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
.gitignore
+5
View File
@@ -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
README.md
+12 -4
View File
@@ -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://<ip>: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
src/camera.rs
+324
View File
@@ -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://<ip>: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<CameraSource> {
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<CameraSource> {
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<Notify>,
) {
let mut ticker = tokio::time::interval(REBUILD_INTERVAL);
let mut last_written: Option<String> = 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}");
}
}
src/config.rs
+18
View File
@@ -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<String>,
/// 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"),
})
}
src/go2rtc.rs
+34 -4
View File
@@ -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<Notify>) {
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(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?");
}
src/lib.rs
+5 -2
View File
@@ -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;
src/main.rs
+28 -10
View File
@@ -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()) => {}
}
src/printer/bambu_commands/mod.rs
+1 -1
View File
@@ -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;
src/printer/bambu_commands/print.rs
+28
View File
@@ -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<Vec<LightReport>>,
/// The printer's camera settings - and, on the models that have one, the
/// URL of its stream. See `Ipcam`.
pub ipcam: Option<Ipcam>,
// 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://<ip>: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://<ip>:322/streaming/live/1` on the models that have it. Note
/// it carries no credentials: a client has to add `bblp:<access_code>`
/// itself.
pub rtsp_url: Option<String>,
}
/// 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
src/printer/mod.rs
+1 -1
View File
@@ -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;
src/uplink/mod.rs
+11
View File
@@ -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<PrinterTelemetry> {
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),
});
}
src/uplink/protocol.rs
+26 -1
View File
@@ -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<String>,
},
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<usize>,
/// 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<String>,
}
/// What `continuum-backend` sends back over the same connection.
tests/uplink_protocol.rs
+34
View File
@@ -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}");
}