diff --git a/.env.example b/.env.example index d3afeb9..87fa44e 100644 --- a/.env.example +++ b/.env.example @@ -5,25 +5,9 @@ CONTINUUM_GATEWAY_TOKEN=replace-with-provisioning-token # Cloud uplink CONTINUUM_UPLINK_URL=wss://api.continuum.local/ws/edge/v1 -CONTINUUM_API_BASE=https://api.continuum.local - -# Local state -CONTINUUM_SQLITE_PATH=./data/edge-cache.db # go2rtc process management CONTINUUM_GO2RTC_BIN=/usr/local/bin/go2rtc CONTINUUM_GO2RTC_CONFIG=./go2rtc.yaml -CONTINUUM_GO2RTC_API_PORT=1984 - -# Bambu Lab printers (LAN mode) — comma-separated "serial:ip:access_code" triples, -# normally discovered via SSDP but overridable for static configs. -CONTINUUM_BAMBU_STATIC_PRINTERS= - -# PrusaLink / Moonraker discovery -CONTINUUM_DISCOVERY_INTERVAL_SECS=30 - -# plate changer hardware -CONTINUUM_PLATE_CHANGER_SERIAL_PORT=/dev/ttyUSB0 -CONTINUUM_PLATE_CHANGER_BAUD=115200 RUST_LOG=info,continuum_proxy=debug diff --git a/Cargo.toml b/Cargo.toml index bb845ae..11ef271 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,30 +20,15 @@ path = "examples/printer_polymorphism.rs" [dependencies] tokio = { version = "1.40", features = ["full"] } tokio-tungstenite = { version = "0.24", features = ["native-tls"] } -tokio-util = { version = "0.7", features = ["codec", "compat"] } futures-util = "0.3" -rumqttc = "0.24" -reqwest = { version = "0.12", features = ["stream", "json"] } -suppaftp = { version = "6", features = ["async-native-tls"] } serde = { version = "1", features = ["derive"] } serde_json = "1" -rusqlite = { version = "0.32", features = ["bundled"] } -tokio-serial = "5.4" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } anyhow = "1" thiserror = "1" -uuid = { version = "1", features = ["v4", "serde"] } -rand = "0.8" dotenvy = "0.15" -# continuum-common's rust-core crate (error types, tracing init, G-code helpers) -# lives in the sibling `continuum-common` repo. Once that crate is published to -# a registry or exposed over git, add it here, e.g.: -# continuum-common = { git = "https://git.octoturge.com/Continuum/continuum-common.git", package = "continuum-core" } - -[profile.release] -opt-level = "z" -lto = true -codegen-units = 1 -panic = "abort" +# Real printer protocol clients (MQTT for Bambu, HTTP for PrusaLink/ +# Moonraker, FTPS for gcode transfer) go here once you're ready to build +# past the stubs in src/printer/ — see that module's doc comment. diff --git a/README.md b/README.md index eea74ae..9ce5a14 100644 --- a/README.md +++ b/README.md @@ -1,51 +1,52 @@ # continuum-proxy -Edge gateway daemon for the Continuum print farm platform. Runs on a Linux SBC -on-site, discovers and talks to printers over the LAN, and maintains a -resilient uplink to the cloud control plane (`continuum-backend`). +Edge gateway daemon for the Continuum print farm platform. Runs on a Linux +SBC on-site, talks to printers over the LAN, and keeps a connection to the +cloud control plane (`continuum-backend`). -## Responsibilities +## This is a learning-stage boilerplate -- **Uplink** (`src/uplink/`): persistent WebSocket connection to - `wss://api.domain.com/ws/edge/v1` with exponential-backoff reconnects, - heartbeat ping/pong, and inbound task dispatch. -- **Printer adapters** (`src/adapters/`): protocol clients per printer vendor — - Bambu Lab (MQTT 8883 + FTPS 990), PrusaLink (REST), Klipper (Moonraker WS). -- **Plate changer** (`src/plate_changer/`): mechanical cycle execution and - sensor validation over GPIO/serial for automated plate-swap hardware. -- **Local cache** (`rusqlite`): buffers telemetry and job state through - connectivity gaps so nothing is lost if the uplink drops. -- **go2rtc watchdog**: supervises the bundled `go2rtc` process for camera - restreaming (RTSP/USB → WebRTC). +This repo is deliberately minimal right now — real printer protocol clients +(Bambu MQTT+FTPS, PrusaLink REST, Klipper/Moonraker WebSocket) are **not** +implemented yet. Each vendor is a stub that just prints what it would do +(`src/printer/bambu.rs`, `prusa.rs`, `klipper.rs`). The idea is to learn +Rust's polymorphism pattern (trait + enum, since Rust has no class +inheritance) on something simple before adding real networking on top. ## Getting started ```bash cp .env.example .env -cargo run +cargo run # the daemon: cloud uplink + go2rtc watchdog +cargo run --example printer_polymorphism # standalone demo, no network/env needed ``` ## Structure ``` src/ - main.rs Multi-threaded async engine: uplink, discovery, go2rtc watchdog - uplink/ Cloud WebSocket client (reconnect, heartbeat, dispatch) - adapters/ bambu.rs, prusalink.rs, moonraker.rs — wire-protocol clients - printer/ Domain model: Printer trait + PrinterBase + PrinterHandle enum - plate_changer/ Mechanical cycle control + sensor validation - discovery/ LAN printer discovery (SSDP / mDNS / static config) + main.rs Runs the uplink and the go2rtc watchdog side by side + config.rs Loads settings from environment variables + uplink/ WebSocket client to continuum-backend: connect, heartbeat, reconnect on drop + printer/ Printer trait + PrinterBase + PrinterHandle enum + one stub per vendor + go2rtc.rs Restarts the go2rtc camera-restreaming process if it dies +examples/ + printer_polymorphism.rs Runs all three printer stubs through one `connect()` call site ``` -`src/adapters/` talks the raw wire protocols (MQTT, FTPS, HTTP, WS). -`src/printer/` is the vendor-agnostic domain model layered on top: a -`Printer` trait plus one struct per vendor (`BambuPrinter`, `PrusaPrinter`, -`KlipperPrinter`) wrapped in a closed `PrinterHandle` enum. Rust has no class -inheritance, so this trait+enum combo is what stands in for a -`GenericPrinter -> BambuPrinter` hierarchy — composition for shared fields, -a trait for shared/overridable behavior, an enum for the closed set of -vendors. See `examples/printer_polymorphism.rs`: +`src/printer/` is where the "inheritance" question lives — see that +module's doc comment for the trait+enum pattern this project uses instead +of class inheritance, and run the example above to see it work. -```bash -cargo run --example printer_polymorphism -``` +## What's not here yet (on purpose) + +- Real MQTT/FTPS/HTTP/WebSocket printer clients — `src/printer/*.rs` has a + `println!` where each of these will go. +- Local SQLite buffering for telemetry across connectivity gaps. +- LAN printer discovery (SSDP/mDNS). +- The mechanical plate-changer interface (serial/GPIO). + +Add these back in one at a time as you get comfortable with the Rust +underneath them — each is its own small lesson (async I/O, a new crate's +API, error handling for a real protocol) rather than something to absorb +all at once. diff --git a/src/adapters/bambu.rs b/src/adapters/bambu.rs deleted file mode 100644 index 3c03d67..0000000 --- a/src/adapters/bambu.rs +++ /dev/null @@ -1,144 +0,0 @@ -use std::path::Path; -use std::time::Duration; - -use rumqttc::{AsyncClient, Event, MqttOptions, Packet, QoS, TlsConfiguration, Transport}; -use serde::{Deserialize, Serialize}; -use suppaftp::async_native_tls::TlsConnector; -use suppaftp::{AsyncNativeTlsConnector, AsyncNativeTlsFtpStream}; -use tokio::sync::mpsc; -use tokio_util::compat::TokioAsyncReadCompatExt; -use tracing::{debug, info, warn}; - -const MQTT_PORT: u16 = 8883; -const FTPS_PORT: u16 = 990; - -#[derive(Debug, Clone)] -pub struct BambuPrinter { - pub serial: String, - pub host: String, - pub access_code: String, -} - -/// Telemetry pushed from the printer's `report` MQTT topic, decoded into the -/// subset of fields the control plane cares about. -#[derive(Debug, Clone, Deserialize)] -pub struct BambuReport { - #[serde(default)] - pub nozzle_temper: Option, - #[serde(default)] - pub bed_temper: Option, - #[serde(default)] - pub mc_percent: Option, - #[serde(default)] - pub gcode_state: Option, -} - -#[derive(Debug, Serialize)] -struct BambuCommandEnvelope<'a> { - print: BambuCommand<'a>, -} - -#[derive(Debug, Serialize)] -struct BambuCommand<'a> { - sequence_id: &'a str, - command: &'a str, -} - -/// Connects over local-LAN MQTTS to a single Bambu printer and streams -/// decoded telemetry reports out on `tx`. Bambu's LAN-mode broker uses a -/// self-signed cert, so a permissive TLS config is required. -pub async fn run_telemetry(printer: BambuPrinter, tx: mpsc::Sender<(String, BambuReport)>) { - loop { - if let Err(err) = telemetry_session(&printer, &tx).await { - warn!(serial = %printer.serial, ?err, "bambu MQTT session ended, reconnecting"); - } - tokio::time::sleep(Duration::from_secs(5)).await; - } -} - -async fn telemetry_session( - printer: &BambuPrinter, - tx: &mpsc::Sender<(String, BambuReport)>, -) -> anyhow::Result<()> { - let mut opts = MqttOptions::new( - format!("continuum-proxy-{}", printer.serial), - printer.host.clone(), - MQTT_PORT, - ); - opts.set_credentials("bblp", printer.access_code.clone()); - opts.set_keep_alive(Duration::from_secs(20)); - // Bambu LAN-mode uses a self-signed certificate; the client trusts it - // explicitly because the connection never leaves the local network. - opts.set_transport(Transport::Tls(TlsConfiguration::default())); - - let (client, mut event_loop) = AsyncClient::new(opts, 16); - let report_topic = format!("device/{}/report", printer.serial); - client.subscribe(&report_topic, QoS::AtMostOnce).await?; - - loop { - match event_loop.poll().await? { - Event::Incoming(Packet::Publish(publish)) if publish.topic == report_topic => { - match serde_json::from_slice::(&publish.payload) { - Ok(value) => { - if let Some(print) = value.get("print") { - if let Ok(report) = serde_json::from_value::(print.clone()) { - debug!(serial = %printer.serial, ?report, "bambu telemetry"); - if tx.send((printer.serial.clone(), report)).await.is_err() { - return Ok(()); - } - } - } - } - Err(err) => warn!(?err, "failed to decode bambu report payload"), - } - } - Event::Incoming(Packet::Disconnect) => { - return Err(anyhow::anyhow!("printer closed MQTT connection")); - } - _ => {} - } - } -} - -/// Sends a G-code print-control command (pause/resume/stop/etc.) to the -/// printer over its LAN MQTT channel. -pub async fn send_command(printer: &BambuPrinter, sequence_id: &str, command: &str) -> anyhow::Result<()> { - let mut opts = MqttOptions::new(format!("continuum-proxy-cmd-{}", printer.serial), printer.host.clone(), MQTT_PORT); - opts.set_credentials("bblp", printer.access_code.clone()); - opts.set_transport(Transport::Tls(TlsConfiguration::default())); - - let (client, mut event_loop) = AsyncClient::new(opts, 4); - let request_topic = format!("device/{}/request", printer.serial); - - let envelope = BambuCommandEnvelope { - print: BambuCommand { sequence_id, command }, - }; - client - .publish(&request_topic, QoS::AtLeastOnce, false, serde_json::to_vec(&envelope)?) - .await?; - - // Pump the event loop once so the publish actually flushes before we drop the client. - let _ = tokio::time::timeout(Duration::from_secs(3), event_loop.poll()).await; - Ok(()) -} - -/// Uploads a sliced `.gcode.3mf` project file to the printer's local storage -/// over FTPS (port 990, implicit TLS) ahead of a print job. -pub async fn dispatch_file(printer: &BambuPrinter, local_path: &Path, remote_name: &str) -> anyhow::Result<()> { - info!(serial = %printer.serial, remote_name, "dispatching file via FTPS"); - - let ftp = AsyncNativeTlsFtpStream::connect(format!("{}:{FTPS_PORT}", printer.host)).await?; - let mut ftp = ftp - .into_secure(AsyncNativeTlsConnector::from(TlsConnector::new()), &printer.host) - .await?; - ftp.login("bblp", &printer.access_code).await?; - - // suppaftp wants a `futures::io::AsyncRead`; tokio::fs::File only - // implements tokio's own AsyncRead, so bridge it with tokio-util's - // `.compat()` adapter rather than pulling in a second async-fs stack. - let mut file = tokio::fs::File::open(local_path).await?.compat(); - ftp.put_file(remote_name, &mut file).await?; - ftp.quit().await?; - - Ok(()) -} diff --git a/src/adapters/mod.rs b/src/adapters/mod.rs deleted file mode 100644 index 22e782d..0000000 --- a/src/adapters/mod.rs +++ /dev/null @@ -1,3 +0,0 @@ -pub mod bambu; -pub mod moonraker; -pub mod prusalink; diff --git a/src/adapters/moonraker.rs b/src/adapters/moonraker.rs deleted file mode 100644 index 8653101..0000000 --- a/src/adapters/moonraker.rs +++ /dev/null @@ -1,74 +0,0 @@ -use futures_util::{SinkExt, StreamExt}; -use serde::{Deserialize, Serialize}; -use tokio_tungstenite::tungstenite::Message; -use tracing::warn; - -#[derive(Debug, Clone)] -pub struct MoonrakerPrinter { - pub host: String, -} - -#[derive(Debug, Serialize)] -struct JsonRpcRequest<'a> { - jsonrpc: &'static str, - method: &'a str, - params: serde_json::Value, - id: u64, -} - -#[derive(Debug, Clone, Deserialize)] -pub struct MoonrakerNotification { - pub method: String, - #[serde(default)] - pub params: Vec, -} - -/// Subscribes to a Klipper/Moonraker printer's `printer.objects.subscribe` -/// WebSocket feed and forwards decoded status notifications upstream. -pub async fn run(printer: MoonrakerPrinter, tx: tokio::sync::mpsc::Sender) { - loop { - if let Err(err) = session(&printer, &tx).await { - warn!(host = %printer.host, ?err, "moonraker session ended, reconnecting"); - } - tokio::time::sleep(std::time::Duration::from_secs(5)).await; - } -} - -async fn session( - printer: &MoonrakerPrinter, - tx: &tokio::sync::mpsc::Sender, -) -> anyhow::Result<()> { - let url = format!("ws://{}/websocket", printer.host); - let (ws_stream, _) = tokio_tungstenite::connect_async(&url).await?; - let (mut write, mut read) = ws_stream.split(); - - let subscribe = JsonRpcRequest { - jsonrpc: "2.0", - method: "printer.objects.subscribe", - params: serde_json::json!({ - "objects": { - "extruder": ["temperature", "target"], - "heater_bed": ["temperature", "target"], - "print_stats": ["state", "progress"], - } - }), - id: 1, - }; - write.send(Message::Text(serde_json::to_string(&subscribe)?)).await?; - - while let Some(frame) = read.next().await { - match frame? { - Message::Text(text) => { - if let Ok(notification) = serde_json::from_str::(&text) { - if tx.send(notification).await.is_err() { - return Ok(()); - } - } - } - Message::Close(_) => return Ok(()), - _ => {} - } - } - - Ok(()) -} diff --git a/src/adapters/prusalink.rs b/src/adapters/prusalink.rs deleted file mode 100644 index e0abaa9..0000000 --- a/src/adapters/prusalink.rs +++ /dev/null @@ -1,56 +0,0 @@ -use serde::Deserialize; -use tracing::warn; - -#[derive(Debug, Clone)] -pub struct PrusaLinkPrinter { - pub host: String, - pub api_key: String, -} - -#[derive(Debug, Clone, Deserialize)] -pub struct PrusaLinkStatus { - pub printer: PrusaLinkPrinterState, -} - -#[derive(Debug, Clone, Deserialize)] -pub struct PrusaLinkPrinterState { - pub state: String, - pub temp_nozzle: f32, - pub temp_bed: f32, -} - -/// Polls PrusaLink's REST API for the current printer state. -/// PrusaLink has no push/streaming transport, so the proxy polls it on a -/// short interval and forwards deltas upstream as synthetic telemetry. -pub async fn fetch_status(client: &reqwest::Client, printer: &PrusaLinkPrinter) -> anyhow::Result { - let url = format!("http://{}/api/v1/status", printer.host); - let response = client - .get(url) - .header("X-Api-Key", &printer.api_key) - .send() - .await? - .error_for_status()?; - - let status = response.json::().await?; - Ok(status) -} - -pub async fn poll_loop( - client: reqwest::Client, - printer: PrusaLinkPrinter, - interval: std::time::Duration, - tx: tokio::sync::mpsc::Sender<(String, PrusaLinkStatus)>, -) { - let mut ticker = tokio::time::interval(interval); - loop { - ticker.tick().await; - match fetch_status(&client, &printer).await { - Ok(status) => { - if tx.send((printer.host.clone(), status)).await.is_err() { - return; - } - } - Err(err) => warn!(host = %printer.host, ?err, "prusalink poll failed"), - } - } -} diff --git a/src/cache.rs b/src/cache.rs deleted file mode 100644 index 216ba79..0000000 --- a/src/cache.rs +++ /dev/null @@ -1,59 +0,0 @@ -use rusqlite::Connection; -use tracing::info; - -/// Local SQLite cache that buffers telemetry and job state across uplink -/// outages so nothing is lost between the printer and the cloud. -pub struct EdgeCache { - conn: Connection, -} - -impl EdgeCache { - pub fn open(path: &str) -> anyhow::Result { - if let Some(parent) = std::path::Path::new(path).parent() { - std::fs::create_dir_all(parent)?; - } - - let conn = Connection::open(path)?; - conn.execute_batch( - "CREATE TABLE IF NOT EXISTS pending_telemetry ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - printer_id TEXT NOT NULL, - payload TEXT NOT NULL, - created_at TEXT NOT NULL DEFAULT (datetime('now')) - ); - CREATE TABLE IF NOT EXISTS job_state ( - job_id TEXT PRIMARY KEY, - status TEXT NOT NULL, - updated_at TEXT NOT NULL DEFAULT (datetime('now')) - );", - )?; - - info!(path, "opened edge cache"); - Ok(Self { conn }) - } - - pub fn enqueue_telemetry(&self, printer_id: &str, payload: &str) -> anyhow::Result<()> { - self.conn.execute( - "INSERT INTO pending_telemetry (printer_id, payload) VALUES (?1, ?2)", - (printer_id, payload), - )?; - Ok(()) - } - - pub fn drain_telemetry(&self, limit: u32) -> anyhow::Result> { - let mut stmt = self - .conn - .prepare("SELECT id, printer_id, payload FROM pending_telemetry ORDER BY id ASC LIMIT ?1")?; - let rows = stmt - .query_map([limit], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))? - .collect::, _>>()?; - Ok(rows) - } - - pub fn ack_telemetry(&self, ids: &[i64]) -> anyhow::Result<()> { - for id in ids { - self.conn.execute("DELETE FROM pending_telemetry WHERE id = ?1", [id])?; - } - Ok(()) - } -} diff --git a/src/config.rs b/src/config.rs index d2fd5a8..b1c471d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -6,12 +6,8 @@ pub struct Config { pub farm_id: String, pub gateway_token: String, pub uplink_url: String, - pub api_base: String, - pub sqlite_path: String, pub go2rtc_bin: String, pub go2rtc_config: String, - pub go2rtc_api_port: u16, - pub discovery_interval_secs: u64, } impl Config { @@ -21,12 +17,8 @@ impl Config { farm_id: require("CONTINUUM_FARM_ID")?, gateway_token: require("CONTINUUM_GATEWAY_TOKEN")?, uplink_url: env_or("CONTINUUM_UPLINK_URL", "wss://api.continuum.local/ws/edge/v1"), - api_base: env_or("CONTINUUM_API_BASE", "https://api.continuum.local"), - sqlite_path: env_or("CONTINUUM_SQLITE_PATH", "./data/edge-cache.db"), go2rtc_bin: env_or("CONTINUUM_GO2RTC_BIN", "go2rtc"), go2rtc_config: env_or("CONTINUUM_GO2RTC_CONFIG", "./go2rtc.yaml"), - go2rtc_api_port: env_or("CONTINUUM_GO2RTC_API_PORT", "1984").parse()?, - discovery_interval_secs: env_or("CONTINUUM_DISCOVERY_INTERVAL_SECS", "30").parse()?, }) } } diff --git a/src/discovery/mod.rs b/src/discovery/mod.rs deleted file mode 100644 index 717e0b9..0000000 --- a/src/discovery/mod.rs +++ /dev/null @@ -1,44 +0,0 @@ -use std::time::Duration; - -use serde::{Deserialize, Serialize}; -use tracing::{debug, info}; - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub enum PrinterVendor { - Bambu, - Prusa, - Klipper, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct DiscoveredPrinter { - pub vendor: PrinterVendor, - pub host: String, - pub serial: Option, -} - -/// Periodically sweeps the LAN for printers (SSDP for Bambu, mDNS for -/// PrusaLink/Moonraker) and reports newly-seen devices upstream. -/// -/// This is intentionally a stub: production discovery would bind a UDP -/// multicast socket per protocol. It's structured so each protocol's probe -/// can be dropped in independently without touching the polling loop. -pub async fn run(interval: Duration, tx: tokio::sync::mpsc::Sender) { - let mut ticker = tokio::time::interval(interval); - loop { - ticker.tick().await; - debug!("running LAN discovery sweep"); - - for printer in sweep().await { - info!(host = %printer.host, ?printer.vendor, "discovered printer"); - if tx.send(printer).await.is_err() { - return; - } - } - } -} - -async fn sweep() -> Vec { - // TODO: SSDP probe (Bambu), mDNS `_prusalink._tcp` / `_moonraker._tcp` probes. - Vec::new() -} diff --git a/src/main.rs b/src/main.rs index ebc9ff2..8f3ea00 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,118 +1,33 @@ -mod adapters; -mod cache; mod config; -mod discovery; mod go2rtc; -mod plate_changer; +mod printer; mod uplink; -use std::time::Duration; - -use tokio::sync::mpsc; -use tracing::{info, warn}; use tracing_subscriber::EnvFilter; -use crate::cache::EdgeCache; use crate::config::Config; -use crate::uplink::ClientMessage; #[tokio::main] async fn main() -> anyhow::Result<()> { tracing_subscriber::fmt() .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"))) - .json() .init(); - dotenvy_load(); - + dotenvy::dotenv().ok(); let config = Config::from_env()?; - info!(gateway_id = %config.gateway_id, farm_id = %config.farm_id, "starting continuum-proxy"); - let cache = EdgeCache::open(&config.sqlite_path)?; - - // Channels wiring the printer adapters -> uplink, and uplink -> task - // handlers for cloud-issued commands. - let (telemetry_tx, telemetry_rx) = mpsc::channel::(256); - let (task_tx, mut task_rx) = mpsc::channel::(64); - let (discovered_tx, mut discovered_rx) = mpsc::channel::(32); - - let uplink_handle = tokio::spawn(uplink::run(config.clone(), telemetry_rx, task_tx)); - - let go2rtc_handle = tokio::spawn(go2rtc::watchdog(config.go2rtc_bin.clone(), config.go2rtc_config.clone())); - - let discovery_handle = tokio::spawn(discovery::run( - Duration::from_secs(config.discovery_interval_secs), - discovered_tx, - )); - - // Cloud-issued task dispatcher: pause/resume/plate-change/etc. commands - // arriving over the uplink get routed to the right adapter here. - let task_dispatcher = tokio::spawn(async move { - while let Some(msg) = task_rx.recv().await { - match msg { - uplink::ServerMessage::Task { task_id, kind, payload } => { - info!(task_id, kind, ?payload, "received task from cloud"); - // TODO: route to adapters::bambu / prusalink / moonraker or plate_changer - // based on `kind`, then send a ClientMessage::TaskResult back upstream. - } - uplink::ServerMessage::HelloAck { session_id } => { - info!(session_id, "uplink session established"); - } - uplink::ServerMessage::HeartbeatAck { seq } => { - tracing::debug!(seq, "heartbeat acked"); - } - } - } - }); - - // Newly-discovered printers get their telemetry adapters spawned on the fly. - let telemetry_tx_for_discovery = telemetry_tx.clone(); - let discovery_dispatcher = tokio::spawn(async move { - while let Some(printer) = discovered_rx.recv().await { - info!(host = %printer.host, "spawning adapter for discovered printer"); - let _ = &telemetry_tx_for_discovery; - // TODO: match printer.vendor and spawn adapters::bambu::run_telemetry / moonraker::run / prusalink::poll_loop - } - }); - - // Periodically flush anything the SQLite cache buffered while the uplink was down. - let cache_flusher = tokio::spawn(async move { - let mut ticker = tokio::time::interval(Duration::from_secs(10)); - loop { - ticker.tick().await; - match cache.drain_telemetry(100) { - Ok(rows) if !rows.is_empty() => { - let ids: Vec = rows.iter().map(|(id, _, _)| *id).collect(); - for (_, printer_id, payload) in &rows { - let payload: serde_json::Value = serde_json::from_str(payload).unwrap_or_default(); - let _ = telemetry_tx - .send(ClientMessage::Telemetry { printer_id: printer_id.clone(), payload }) - .await; - } - if let Err(err) = cache.ack_telemetry(&ids) { - warn!(?err, "failed to ack drained telemetry rows"); - } - } - Ok(_) => {} - Err(err) => warn!(?err, "failed to drain edge cache"), - } - } - }); + tracing::info!(gateway_id = %config.gateway_id, farm_id = %config.farm_id, "starting continuum-proxy"); + // Two things run at once for now: the cloud uplink, and the go2rtc + // camera-restreaming process. More will join them later — printer + // discovery, the plate changer — as you build past src/printer/'s + // stubs. `tokio::select!` runs both and stops as soon as either one + // returns; both loop forever below, so in practice this runs until the + // process is killed. tokio::select! { - res = uplink_handle => warn!(?res, "uplink task exited"), - res = go2rtc_handle => warn!(?res, "go2rtc watchdog exited"), - res = discovery_handle => warn!(?res, "discovery task exited"), - res = task_dispatcher => warn!(?res, "task dispatcher exited"), - res = discovery_dispatcher => warn!(?res, "discovery dispatcher exited"), - res = cache_flusher => warn!(?res, "cache flusher exited"), + _ = uplink::run(config.clone()) => {} + _ = go2rtc::watchdog(config.go2rtc_bin.clone(), config.go2rtc_config.clone()) => {} } Ok(()) } - -/// Loads `.env` if present; a missing file is fine (env vars may already be -/// set by the process supervisor), so the error is deliberately ignored. -fn dotenvy_load() { - let _ = dotenvy::dotenv(); -} diff --git a/src/plate_changer/gpio.rs b/src/plate_changer/gpio.rs deleted file mode 100644 index 6f6dffe..0000000 --- a/src/plate_changer/gpio.rs +++ /dev/null @@ -1,20 +0,0 @@ -/// Snapshot of the plate changer's discrete sensor inputs, read after a -/// cycle completes to confirm the mechanism actually did what it reported. -#[derive(Debug, Clone, Copy, Default)] -pub struct SensorState { - pub plate_present: bool, - pub bed_clear: bool, -} - -/// Reads the plate-present and bed-clear sensors. -/// -/// On the target hardware (Raspberry Pi / SBC GPIO header) this would read -/// two debounced digital inputs via `rppal` or sysfs GPIO. Stubbed here so -/// the crate builds without hardware access; swap in a real backend behind -/// this same function signature. -pub async fn read_sensor_state() -> SensorState { - SensorState { - plate_present: true, - bed_clear: true, - } -} diff --git a/src/plate_changer/mod.rs b/src/plate_changer/mod.rs deleted file mode 100644 index c82e316..0000000 --- a/src/plate_changer/mod.rs +++ /dev/null @@ -1,68 +0,0 @@ -mod gpio; -mod serial; - -use thiserror::Error; -use tracing::{info, warn}; - -#[derive(Debug, Error)] -pub enum PlateChangerError { - #[error("plate changer hardware not responding")] - NotResponding, - #[error("sensor validation failed: {0}")] - SensorMismatch(String), - #[error("serial transport error: {0}")] - Serial(#[from] serial::SerialError), -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum CycleOutcome { - Success, - Retried, -} - -/// Drives a mechanical plate-swap cycle: signal the changer over serial, -/// wait for it to report completion, then cross-check the physical sensors -/// (plate-present, bed-clear) before releasing the print queue to continue. -pub struct PlateChanger { - port: serial::SerialPort, -} - -impl PlateChanger { - pub fn open(path: &str, baud: u32) -> Result { - Ok(Self { - port: serial::SerialPort::open(path, baud)?, - }) - } - - pub async fn run_cycle(&mut self, slot: u8) -> Result { - info!(slot, "starting plate change cycle"); - - self.port.send_command(&serial::Command::Eject).await?; - self.port.await_ack(std::time::Duration::from_secs(30)).await?; - - self.port.send_command(&serial::Command::LoadSlot(slot)).await?; - self.port.await_ack(std::time::Duration::from_secs(30)).await?; - - match self.validate_sensors().await { - Ok(()) => Ok(CycleOutcome::Success), - Err(err) => { - warn!(?err, slot, "sensor validation failed, retrying cycle once"); - self.port.send_command(&serial::Command::LoadSlot(slot)).await?; - self.port.await_ack(std::time::Duration::from_secs(30)).await?; - self.validate_sensors().await?; - Ok(CycleOutcome::Retried) - } - } - } - - async fn validate_sensors(&mut self) -> Result<(), PlateChangerError> { - let state = gpio::read_sensor_state().await; - if !state.plate_present { - return Err(PlateChangerError::SensorMismatch("plate not detected on bed".into())); - } - if !state.bed_clear { - return Err(PlateChangerError::SensorMismatch("bed obstruction detected".into())); - } - Ok(()) - } -} diff --git a/src/plate_changer/serial.rs b/src/plate_changer/serial.rs deleted file mode 100644 index 2c392f8..0000000 --- a/src/plate_changer/serial.rs +++ /dev/null @@ -1,64 +0,0 @@ -use std::time::Duration; - -use thiserror::Error; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio_serial::SerialPortBuilderExt; - -#[derive(Debug, Error)] -pub enum SerialError { - #[error("failed to open serial port: {0}")] - Open(#[from] tokio_serial::Error), - #[error("i/o error: {0}")] - Io(#[from] std::io::Error), - #[error("timed out waiting for hardware acknowledgement")] - Timeout, - #[error("hardware reported a fault: {0}")] - Fault(String), -} - -pub enum Command { - Eject, - LoadSlot(u8), -} - -impl Command { - fn encode(&self) -> String { - match self { - Command::Eject => "EJECT\n".to_string(), - Command::LoadSlot(slot) => format!("LOAD {slot}\n"), - } - } -} - -/// Thin line-protocol wrapper around the plate changer's serial control -/// board (an ASCII command set over a USB-serial link, e.g. an Arduino/RP2040 -/// running the changer firmware). -pub struct SerialPort { - inner: tokio_serial::SerialStream, -} - -impl SerialPort { - pub fn open(path: &str, baud: u32) -> Result { - let inner = tokio_serial::new(path, baud).timeout(Duration::from_millis(500)).open_native_async()?; - Ok(Self { inner }) - } - - pub async fn send_command(&mut self, command: &Command) -> Result<(), SerialError> { - self.inner.write_all(command.encode().as_bytes()).await?; - Ok(()) - } - - pub async fn await_ack(&mut self, timeout: Duration) -> Result<(), SerialError> { - let mut buf = [0u8; 64]; - let read = tokio::time::timeout(timeout, self.inner.read(&mut buf)) - .await - .map_err(|_| SerialError::Timeout)??; - - let response = String::from_utf8_lossy(&buf[..read]); - if response.trim() == "OK" { - Ok(()) - } else { - Err(SerialError::Fault(response.trim().to_string())) - } - } -} diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index c23d341..03f3421 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -1,3 +1,12 @@ +//! The cloud connection: a WebSocket to `continuum-backend`'s `/ws/edge/v1`. +//! +//! This is intentionally the simplest version of "a client that reconnects +//! when it drops": no backoff curve, no channels routing messages in from +//! other parts of the program yet — just connect, say hello, send a +//! heartbeat on a timer, and print whatever comes back. Once you're +//! comfortable with this, growing it (real backoff, forwarding printer +//! telemetry, dispatching cloud-issued commands) is additive, not a rewrite. + mod protocol; pub use protocol::{ClientMessage, ServerMessage}; @@ -5,109 +14,61 @@ pub use protocol::{ClientMessage, ServerMessage}; use std::time::Duration; use futures_util::{SinkExt, StreamExt}; -use tokio::sync::mpsc; use tokio_tungstenite::tungstenite::Message; use tracing::{info, warn}; use crate::config::Config; -const MIN_BACKOFF: Duration = Duration::from_secs(1); -const MAX_BACKOFF: Duration = Duration::from_secs(60); +const RETRY_DELAY: Duration = Duration::from_secs(5); const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15); -/// Runs the resilient uplink loop forever: connect, authenticate, exchange -/// heartbeats and telemetry/task messages, and reconnect with exponential -/// backoff (plus jitter) whenever the connection drops. -pub async fn run( - config: Config, - mut telemetry_rx: mpsc::Receiver, - task_tx: mpsc::Sender, -) { - let mut backoff = MIN_BACKOFF; - +/// Runs forever: connect, serve the connection until it drops, wait, retry. +pub async fn run(config: Config) { loop { info!(url = %config.uplink_url, "connecting to cloud uplink"); - match connect_and_serve(&config, &mut telemetry_rx, &task_tx).await { - Ok(()) => { - info!("uplink connection closed cleanly"); - backoff = MIN_BACKOFF; - } - Err(err) => { - warn!(?err, backoff_secs = backoff.as_secs(), "uplink connection failed, retrying"); - } + if let Err(err) = connect_and_serve(&config).await { + warn!(?err, "uplink connection failed, retrying in {RETRY_DELAY:?}"); } - let jitter = Duration::from_millis(rand::random::() % 500); - tokio::time::sleep(backoff + jitter).await; - backoff = (backoff * 2).min(MAX_BACKOFF); + tokio::time::sleep(RETRY_DELAY).await; } } -async fn connect_and_serve( - config: &Config, - telemetry_rx: &mut mpsc::Receiver, - task_tx: &mpsc::Sender, -) -> anyhow::Result<()> { - let (ws_stream, _resp) = tokio_tungstenite::connect_async(&config.uplink_url).await?; +async fn connect_and_serve(config: &Config) -> anyhow::Result<()> { + let (ws_stream, _response) = tokio_tungstenite::connect_async(&config.uplink_url).await?; let (mut write, mut read) = ws_stream.split(); - send(&mut write, &ClientMessage::Hello { + let hello = ClientMessage::Hello { gateway_id: config.gateway_id.clone(), farm_id: config.farm_id.clone(), token: config.gateway_token.clone(), - version: env!("CARGO_PKG_VERSION"), - }) - .await?; + }; + write.send(Message::Text(serde_json::to_string(&hello)?)).await?; let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL); let mut seq: u64 = 0; - heartbeat.tick().await; // consume the immediate first tick loop { tokio::select! { _ = heartbeat.tick() => { seq += 1; - send(&mut write, &ClientMessage::Heartbeat { seq }).await?; - } - - Some(msg) = telemetry_rx.recv() => { - send(&mut write, &msg).await?; + let msg = ClientMessage::Heartbeat { seq }; + write.send(Message::Text(serde_json::to_string(&msg)?)).await?; } frame = read.next() => { match frame { Some(Ok(Message::Text(text))) => { - match serde_json::from_str::(&text) { - Ok(server_msg) => { - if task_tx.send(server_msg).await.is_err() { - return Ok(()); - } - } - Err(err) => warn!(?err, "failed to decode server message"), + if let Ok(server_msg) = serde_json::from_str::(&text) { + info!(?server_msg, "received from cloud"); } } - Some(Ok(Message::Ping(payload))) => { - write.send(Message::Pong(payload)).await?; - } - Some(Ok(Message::Close(frame))) => { - info!(?frame, "server closed uplink"); - return Ok(()); - } - Some(Ok(_)) => {} + Some(Ok(Message::Close(_))) | None => return Ok(()), Some(Err(err)) => return Err(err.into()), - None => return Ok(()), + _ => {} } } } } } - -async fn send( - write: &mut (impl SinkExt + Unpin), - msg: &ClientMessage, -) -> anyhow::Result<()> { - let payload = serde_json::to_string(msg)?; - write.send(Message::Text(payload)).await?; - Ok(()) -} diff --git a/src/uplink/protocol.rs b/src/uplink/protocol.rs index 54b8b8a..2fe1315 100644 --- a/src/uplink/protocol.rs +++ b/src/uplink/protocol.rs @@ -1,42 +1,17 @@ use serde::{Deserialize, Serialize}; -/// Messages sent from this gateway up to `continuum-backend`'s `/ws/edge/v1` route. +/// What this gateway sends up to `continuum-backend`'s `/ws/edge/v1` route. #[derive(Debug, Clone, Serialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum ClientMessage { - Hello { - gateway_id: String, - farm_id: String, - token: String, - version: &'static str, - }, - Heartbeat { - seq: u64, - }, - Telemetry { - printer_id: String, - payload: serde_json::Value, - }, - TaskResult { - task_id: String, - ok: bool, - error: Option, - }, + Hello { gateway_id: String, farm_id: String, token: String }, + Heartbeat { seq: u64 }, } -/// Messages received from `continuum-backend` over the same connection. +/// What `continuum-backend` sends back over the same connection. #[derive(Debug, Clone, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum ServerMessage { - HelloAck { - session_id: String, - }, - HeartbeatAck { - seq: u64, - }, - Task { - task_id: String, - kind: String, - payload: serde_json::Value, - }, + HelloAck { session_id: String }, + HeartbeatAck { seq: u64 }, }