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.
This commit is contained in:
+324
@@ -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}");
|
||||
}
|
||||
}
|
||||
@@ -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"),
|
||||
})
|
||||
}
|
||||
|
||||
+37
-7
@@ -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(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?");
|
||||
}
|
||||
|
||||
+5
-2
@@ -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;
|
||||
|
||||
+28
-10
@@ -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()) => {}
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
|
||||
+1
-1
@@ -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;
|
||||
|
||||
@@ -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),
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
+26
-1
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user