From 2807b2b0673c8de98cd900dde034c2fae79928c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Fri, 28 Aug 2026 16:25:26 +0000 Subject: [PATCH] Initial boilerplate scaffold for continuum-proxy --- .env.example | 29 ++++++++ .gitignore | 8 +++ Cargo.toml | 41 +++++++++++ README.md | 37 ++++++++++ src/adapters/bambu.rs | 139 ++++++++++++++++++++++++++++++++++++ src/adapters/mod.rs | 3 + src/adapters/moonraker.rs | 74 +++++++++++++++++++ src/adapters/prusalink.rs | 56 +++++++++++++++ src/cache.rs | 59 +++++++++++++++ src/config.rs | 40 +++++++++++ src/discovery/mod.rs | 44 ++++++++++++ src/go2rtc.rs | 33 +++++++++ src/main.rs | 130 +++++++++++++++++++++++++++++++++ src/plate_changer/gpio.rs | 20 ++++++ src/plate_changer/mod.rs | 70 ++++++++++++++++++ src/plate_changer/serial.rs | 64 +++++++++++++++++ src/uplink/mod.rs | 113 +++++++++++++++++++++++++++++ src/uplink/protocol.rs | 42 +++++++++++ 18 files changed, 1002 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 Cargo.toml create mode 100644 README.md create mode 100644 src/adapters/bambu.rs create mode 100644 src/adapters/mod.rs create mode 100644 src/adapters/moonraker.rs create mode 100644 src/adapters/prusalink.rs create mode 100644 src/cache.rs create mode 100644 src/config.rs create mode 100644 src/discovery/mod.rs create mode 100644 src/go2rtc.rs create mode 100644 src/main.rs create mode 100644 src/plate_changer/gpio.rs create mode 100644 src/plate_changer/mod.rs create mode 100644 src/plate_changer/serial.rs create mode 100644 src/uplink/mod.rs create mode 100644 src/uplink/protocol.rs diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..d3afeb9 --- /dev/null +++ b/.env.example @@ -0,0 +1,29 @@ +# Identity +CONTINUUM_GATEWAY_ID=gw-farm01-rpi01 +CONTINUUM_FARM_ID=replace-with-farm-uuid +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/.gitignore b/.gitignore new file mode 100644 index 0000000..55e567b --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +/target +Cargo.lock +*.db +*.db-journal +*.sqlite +*.sqlite3 +.env +*.log diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..5f20b2c --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,41 @@ +[package] +name = "continuum-proxy" +version = "0.1.0" +edition = "2024" +description = "Continuum edge gateway daemon — runs on-site, bridges printers to the cloud control plane" +license = "UNLICENSED" + +[[bin]] +name = "continuum-proxy" +path = "src/main.rs" + +[dependencies] +tokio = { version = "1.40", features = ["full"] } +tokio-tungstenite = { version = "0.24", features = ["native-tls"] } +tokio-util = { version = "0.7", features = ["codec"] } +futures-util = "0.3" +rumqttc = "0.24" +reqwest = { version = "0.12", features = ["stream", "json"] } +suppaftp = { version = "6", features = ["async-native-tls"] } +native-tls = "0.2" +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" + +# 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" diff --git a/README.md b/README.md new file mode 100644 index 0000000..d55b107 --- /dev/null +++ b/README.md @@ -0,0 +1,37 @@ +# 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`). + +## Responsibilities + +- **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). + +## Getting started + +```bash +cp .env.example .env +cargo run +``` + +## 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 printer clients + plate_changer/ Mechanical cycle control + sensor validation + discovery/ LAN printer discovery (SSDP / mDNS / static config) +``` diff --git a/src/adapters/bambu.rs b/src/adapters/bambu.rs new file mode 100644 index 0000000..808be97 --- /dev/null +++ b/src/adapters/bambu.rs @@ -0,0 +1,139 @@ +use std::path::Path; +use std::time::Duration; + +use rumqttc::{AsyncClient, Event, MqttOptions, Packet, QoS, TlsConfiguration, Transport}; +use serde::{Deserialize, Serialize}; +use suppaftp::{AsyncNativeTlsConnector, AsyncNativeTlsFtpStream}; +use tokio::sync::mpsc; +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(native_tls::TlsConnector::new()?), &printer.host) + .await?; + ftp.login("bblp", &printer.access_code).await?; + + let mut file = tokio::fs::File::open(local_path).await?; + ftp.put_file(remote_name, &mut file).await?; + ftp.quit().await?; + + Ok(()) +} diff --git a/src/adapters/mod.rs b/src/adapters/mod.rs new file mode 100644 index 0000000..22e782d --- /dev/null +++ b/src/adapters/mod.rs @@ -0,0 +1,3 @@ +pub mod bambu; +pub mod moonraker; +pub mod prusalink; diff --git a/src/adapters/moonraker.rs b/src/adapters/moonraker.rs new file mode 100644 index 0000000..8653101 --- /dev/null +++ b/src/adapters/moonraker.rs @@ -0,0 +1,74 @@ +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 new file mode 100644 index 0000000..e0abaa9 --- /dev/null +++ b/src/adapters/prusalink.rs @@ -0,0 +1,56 @@ +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 new file mode 100644 index 0000000..216ba79 --- /dev/null +++ b/src/cache.rs @@ -0,0 +1,59 @@ +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 new file mode 100644 index 0000000..d2fd5a8 --- /dev/null +++ b/src/config.rs @@ -0,0 +1,40 @@ +use std::env; + +#[derive(Debug, Clone)] +pub struct Config { + pub gateway_id: String, + 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 { + pub fn from_env() -> anyhow::Result { + Ok(Self { + gateway_id: require("CONTINUUM_GATEWAY_ID")?, + 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()?, + }) + } +} + +fn require(key: &str) -> anyhow::Result { + env::var(key).map_err(|_| anyhow::anyhow!("missing required env var {key}")) +} + +fn env_or(key: &str, default: &str) -> String { + env::var(key).unwrap_or_else(|_| default.to_string()) +} diff --git a/src/discovery/mod.rs b/src/discovery/mod.rs new file mode 100644 index 0000000..717e0b9 --- /dev/null +++ b/src/discovery/mod.rs @@ -0,0 +1,44 @@ +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/go2rtc.rs b/src/go2rtc.rs new file mode 100644 index 0000000..d2fd31b --- /dev/null +++ b/src/go2rtc.rs @@ -0,0 +1,33 @@ +use std::process::Stdio; +use std::time::Duration; + +use tokio::process::Command; +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) { + loop { + info!(%bin, %config_path, "starting go2rtc"); + + let spawned = Command::new(&bin) + .arg("-config") + .arg(&config_path) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .kill_on_drop(true) + .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"), + }, + Err(err) => { + error!(?err, "failed to spawn go2rtc, is it installed and on PATH?"); + } + } + + tokio::time::sleep(Duration::from_secs(5)).await; + } +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..86d171b --- /dev/null +++ b/src/main.rs @@ -0,0 +1,130 @@ +mod adapters; +mod cache; +mod config; +mod discovery; +mod go2rtc; +mod plate_changer; +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(); + + 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"), + } + } + }); + + 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"), + } + + Ok(()) +} + +/// Loads a `.env` file if present, without pulling in a heavyweight config +/// crate. No-op (and safe to ignore errors) when the file doesn't exist. +fn dotenvy_load() { + if let Ok(contents) = std::fs::read_to_string(".env") { + for line in contents.lines() { + let line = line.trim(); + if line.is_empty() || line.starts_with('#') { + continue; + } + if let Some((key, value)) = line.split_once('=') { + if std::env::var(key).is_err() { + std::env::set_var(key, value); + } + } + } + } +} diff --git a/src/plate_changer/gpio.rs b/src/plate_changer/gpio.rs new file mode 100644 index 0000000..6f6dffe --- /dev/null +++ b/src/plate_changer/gpio.rs @@ -0,0 +1,20 @@ +/// 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 new file mode 100644 index 0000000..d611e66 --- /dev/null +++ b/src/plate_changer/mod.rs @@ -0,0 +1,70 @@ +mod gpio; +mod serial; + +pub use gpio::SensorState; + +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 new file mode 100644 index 0000000..2c392f8 --- /dev/null +++ b/src/plate_changer/serial.rs @@ -0,0 +1,64 @@ +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 new file mode 100644 index 0000000..c23d341 --- /dev/null +++ b/src/uplink/mod.rs @@ -0,0 +1,113 @@ +mod protocol; + +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 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; + + 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"); + } + } + + let jitter = Duration::from_millis(rand::random::() % 500); + tokio::time::sleep(backoff + jitter).await; + backoff = (backoff * 2).min(MAX_BACKOFF); + } +} + +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?; + let (mut write, mut read) = ws_stream.split(); + + send(&mut write, &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?; + + 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?; + } + + 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"), + } + } + 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(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 new file mode 100644 index 0000000..54b8b8a --- /dev/null +++ b/src/uplink/protocol.rs @@ -0,0 +1,42 @@ +use serde::{Deserialize, Serialize}; + +/// Messages sent from this gateway 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, + }, +} + +/// Messages received from `continuum-backend` 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, + }, +}