From 9359add6aaaa133d7e57ae6a66105f39a878f657 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Fri, 30 Oct 2026 07:49:31 +0000 Subject: [PATCH] Drive Klipper, OctoPrint and PrusaLink for real The three were println stubs. Each is now an HTTP client (Moonraker, OctoPrint's REST API, PrusaLink's v1 API) behind a small shared helper: status polled in the background, pause/resume/cancel, G-code lines, fan speed, and upload-and-start of a sliced file. A slicer project (.3mf) is refused for these, and so is an upload while a print is running. Telemetry and the completion watcher now read a vendor-neutral PrinterStatus instead of Bambu's PrintReport, so these printers report progress, temperatures and job state to the cloud. Written from the public API docs and tested against mock servers only; no real Klipper, OctoPrint or PrusaLink machine has been on the bench. --- src/completion.rs | 2 +- src/printer/http.rs | 202 +++++++++++++++++++ src/printer/klipper.rs | 297 ++++++++++++++++++++++++---- src/printer/mock_http.rs | 118 +++++++++++ src/printer/mod.rs | 72 ++++++- src/printer/octoprint.rs | 417 +++++++++++++++++++++++++++++++++++++++ src/printer/polled.rs | 47 +++++ src/printer/prusa.rs | 106 ---------- src/provision.rs | 7 +- src/uplink/mod.rs | 39 ++-- 10 files changed, 1141 insertions(+), 166 deletions(-) create mode 100644 src/printer/http.rs create mode 100644 src/printer/mock_http.rs create mode 100644 src/printer/octoprint.rs create mode 100644 src/printer/polled.rs diff --git a/src/completion.rs b/src/completion.rs index 6a7df76..0e3d4b4 100644 --- a/src/completion.rs +++ b/src/completion.rs @@ -105,7 +105,7 @@ pub async fn run(fleet: SharedFleet) { if let Err(err) = printer.refresh_status().await { warn!(printer = %printer.base().id, %err, "could not request a status refresh"); } - if let Some(state) = printer.print_status().await.and_then(|r| r.gcode_state) { + if let Some(state) = printer.status().await.and_then(|s| s.state) { states.push((printer.base().id.clone(), state)); } } diff --git a/src/printer/http.rs b/src/printer/http.rs new file mode 100644 index 0000000..3255e42 --- /dev/null +++ b/src/printer/http.rs @@ -0,0 +1,202 @@ +//! The HTTP plumbing the REST-speaking vendors (Klipper's Moonraker, +//! OctoPrint, PrusaLink) share, so each driver is only its own endpoints. +//! +//! reqwest is built without its `json` and `multipart` features (see +//! Cargo.toml), so bodies are made by hand here: JSON through `serde_json` +//! and the one multipart upload as a few lines of bytes. +use std::time::Duration; + +use reqwest::Method; +use serde_json::Value; + +use super::PrinterError; + +const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +/// A project file over a Pi's Wi-Fi can be tens of megabytes. +const UPLOAD_TIMEOUT: Duration = Duration::from_secs(600); + +#[derive(Clone)] +pub struct Http { + client: reqwest::Client, + base: String, + api_key: Option, +} + +/// `host` as the operator wrote it: a bare `192.168.0.9`, `host:port`, or a +/// full `http(s)://` URL. A bare host gets `default_port`. +fn base_url(host: &str, default_port: u16) -> String { + let host = host.trim().trim_end_matches('/'); + if host.starts_with("http://") || host.starts_with("https://") { + host.to_string() + } else if host.contains(':') { + format!("http://{host}") + } else { + format!("http://{host}:{default_port}") + } +} + +impl Http { + pub fn new(host: &str, default_port: u16, api_key: Option<&str>) -> Self { + Self { + client: reqwest::Client::new(), + base: base_url(host, default_port), + api_key: api_key.filter(|k| !k.is_empty()).map(str::to_string), + } + } + + async fn send( + &self, + method: Method, + path: &str, + content_type: Option<&str>, + extra_headers: &[(&str, &str)], + body: Option>, + timeout: Duration, + ) -> Result { + let url = format!("{}{path}", self.base); + let mut request = self.client.request(method.clone(), &url).timeout(timeout); + if let Some(key) = &self.api_key { + request = request.header("X-Api-Key", key); + } + if let Some(content_type) = content_type { + request = request.header("Content-Type", content_type); + } + for (name, value) in extra_headers { + request = request.header(*name, *value); + } + if let Some(body) = body { + request = request.body(body); + } + + // The URL is left out of errors on purpose: it is the printer's + // address, which the caller already names. + let response = request.send().await.map_err(|err| PrinterError(format!("{method} {path}: {err}")))?; + let status = response.status(); + let bytes = response.bytes().await.map_err(|err| PrinterError(format!("{method} {path}: {err}")))?; + if !status.is_success() { + let text = String::from_utf8_lossy(&bytes); + let text: String = text.trim().chars().take(200).collect(); + return Err(PrinterError(format!("{method} {path}: HTTP {} {text}", status.as_u16()))); + } + // An empty body (204, or a 201 with nothing) is a success, not a + // parse error; so is a non-JSON one, which only the callers that + // read the answer care about. + Ok(serde_json::from_slice(&bytes).unwrap_or(Value::Null)) + } + + pub async fn get_json(&self, path: &str) -> Result { + self.send(Method::GET, path, None, &[], None, REQUEST_TIMEOUT).await + } + + pub async fn post_json(&self, path: &str, body: &Value) -> Result { + let body = body.to_string().into_bytes(); + self.send(Method::POST, path, Some("application/json"), &[], Some(body), REQUEST_TIMEOUT).await + } + + pub async fn post_empty(&self, path: &str) -> Result { + self.send(Method::POST, path, None, &[], None, REQUEST_TIMEOUT).await + } + + pub async fn put_empty(&self, path: &str) -> Result { + self.send(Method::PUT, path, None, &[], None, REQUEST_TIMEOUT).await + } + + pub async fn delete(&self, path: &str) -> Result { + self.send(Method::DELETE, path, None, &[], None, REQUEST_TIMEOUT).await + } + + pub async fn put_bytes(&self, path: &str, headers: &[(&str, &str)], data: Vec) -> Result { + self.send(Method::PUT, path, Some("application/octet-stream"), headers, Some(data), UPLOAD_TIMEOUT).await + } + + /// A `multipart/form-data` POST: the text `fields`, then `data` as the + /// part named `file_field`. + pub async fn post_file( + &self, + path: &str, + fields: &[(&str, &str)], + file_field: &str, + file_name: &str, + data: &[u8], + ) -> Result { + let boundary = format!( + "continuum-{:x}{:x}", + std::process::id(), + std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos()).unwrap_or(0) + ); + let body = multipart_body(&boundary, fields, file_field, file_name, data); + let content_type = format!("multipart/form-data; boundary={boundary}"); + self.send(Method::POST, path, Some(&content_type), &[], Some(body), UPLOAD_TIMEOUT).await + } +} + +fn multipart_body(boundary: &str, fields: &[(&str, &str)], file_field: &str, file_name: &str, data: &[u8]) -> Vec { + let mut body = Vec::with_capacity(data.len() + 512); + for (name, value) in fields { + body.extend_from_slice( + format!("--{boundary}\r\nContent-Disposition: form-data; name=\"{name}\"\r\n\r\n{value}\r\n").as_bytes(), + ); + } + body.extend_from_slice( + format!( + "--{boundary}\r\nContent-Disposition: form-data; name=\"{file_field}\"; filename=\"{file_name}\"\r\nContent-Type: application/octet-stream\r\n\r\n" + ) + .as_bytes(), + ); + body.extend_from_slice(data); + body.extend_from_slice(format!("\r\n--{boundary}--\r\n").as_bytes()); + body +} + +/// The file name to give the printer: the last path component, and nothing +/// that could break out of a multipart header or a URL path. +pub fn safe_file_name(file_name: &str) -> Result { + let name = file_name.rsplit(['/', '\\']).next().unwrap_or("").trim(); + if name.is_empty() || name.starts_with('.') || name.chars().any(|c| c.is_control() || c == '"') { + return Err(PrinterError(format!("'{file_name}' is not a usable file name"))); + } + Ok(name.to_string()) +} + +/// Percent-encodes `name` for use as one URL path segment. +pub fn encode_segment(name: &str) -> String { + let mut out = String::new(); + for byte in name.bytes() { + if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') { + out.push(byte as char); + } else { + out.push_str(&format!("%{byte:02X}")); + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_bare_host_gets_the_default_port() { + assert_eq!(base_url("192.168.0.9", 7125), "http://192.168.0.9:7125"); + assert_eq!(base_url("printer.lan:80", 7125), "http://printer.lan:80"); + assert_eq!(base_url("https://p.example/", 7125), "https://p.example"); + } + + #[test] + fn file_names_are_reduced_to_a_safe_last_component() { + assert_eq!(safe_file_name("jobs/../a b.gcode").unwrap(), "a b.gcode"); + assert!(safe_file_name("..").is_err()); + assert!(safe_file_name("x\".gcode").is_err()); + assert!(safe_file_name("").is_err()); + assert_eq!(encode_segment("a b+c.gcode"), "a%20b%2Bc.gcode"); + } + + #[test] + fn the_multipart_body_has_fields_then_the_file() { + let body = multipart_body("B", &[("print", "true")], "file", "x.gcode", b"G1"); + let text = String::from_utf8(body).unwrap(); + assert!(text.starts_with("--B\r\nContent-Disposition: form-data; name=\"print\"\r\n\r\ntrue\r\n")); + assert!(text.contains("name=\"file\"; filename=\"x.gcode\"")); + assert!(text.ends_with("G1\r\n--B--\r\n")); + } +} diff --git a/src/printer/klipper.rs b/src/printer/klipper.rs index b9c28fc..c98d2d2 100644 --- a/src/printer/klipper.rs +++ b/src/printer/klipper.rs @@ -1,75 +1,304 @@ -use super::{GenericPrinter, PrinterBase, PrinterError}; +//! Klipper, through Moonraker's HTTP API (default port 7125). +//! +//! Written from Moonraker's published API and exercised against a mock +//! server in the tests below - no real Klipper machine has been on the +//! bench, so treat the field names as unconfirmed until one has. +use std::time::Duration; + +use serde_json::{json, Value}; + +use super::http::{safe_file_name, Http}; +use super::polled::StatusFeed; +use super::{GenericPrinter, PrintControl, PrinterBase, PrinterError, PrinterStatus}; + +const MOONRAKER_PORT: u16 = 7125; +const POLL_EVERY: Duration = Duration::from_secs(2); + +/// What one `objects/query` asks for - everything `parse_status` reads. +const STATUS_QUERY: &str = + "/printer/objects/query?print_stats&display_status&extruder&heater_bed&virtual_sdcard"; pub struct KlipperPrinter { base: PrinterBase, + http: Http, + feed: Option, } impl KlipperPrinter { pub fn new(id: &str, name: &str, host: &str) -> Self { Self { base: PrinterBase { id: id.into(), name: name.into(), host: host.into() }, + http: Http::new(host, MOONRAKER_PORT, None), + feed: None, + } + } + + pub fn status(&self) -> Option { + self.feed.as_ref().and_then(StatusFeed::latest) + } +} + +/// Moonraker's `print_stats.state` in the vocabulary the cloud already +/// understands (the strings Bambu reports; see `mapStatus` in the backend). +fn map_state(state: &str) -> Option<&'static str> { + match state { + "standby" => Some("IDLE"), + "printing" => Some("RUNNING"), + "paused" => Some("PAUSE"), + "complete" => Some("FINISH"), + // A cancelled print leaves nothing running; the machine is free. + "cancelled" => Some("IDLE"), + "error" => Some("FAILED"), + _ => None, + } +} + +/// The `result.status` object of an `objects/query` answer. +fn parse_status(answer: &Value) -> Option { + let status = answer.get("result")?.get("status")?; + let stats = status.get("print_stats")?; + let state = stats.get("state")?.as_str()?; + let number = |v: Option<&Value>| v.and_then(Value::as_f64).map(|n| n as f32); + + let progress = number(status.get("virtual_sdcard").and_then(|s| s.get("progress"))).unwrap_or(0.0); + let percent = match state { + "complete" => 100, + "printing" | "paused" => (progress * 100.0).round() as i32, + _ => 0, + }; + let duration = number(stats.get("print_duration")).unwrap_or(0.0); + // Klipper has no estimate of its own; this is elapsed time scaled by + // progress, which is wrong early in a print and settles as it goes. + let remaining_minutes = (matches!(state, "printing" | "paused") && progress > 0.01) + .then(|| ((duration / progress - duration) / 60.0).round() as i32); + let layer_info = stats.get("info"); + let int = |v: Option<&Value>| v.and_then(Value::as_i64).map(|n| n as i32); + + Some(PrinterStatus { + state: map_state(state).map(str::to_string), + percent: Some(percent), + layer: int(layer_info.and_then(|i| i.get("current_layer"))), + total_layers: int(layer_info.and_then(|i| i.get("total_layer"))), + remaining_minutes, + job_name: stats + .get("filename") + .and_then(Value::as_str) + .filter(|f| !f.is_empty()) + .map(str::to_string), + nozzle_temp: number(status.get("extruder").and_then(|e| e.get("temperature"))), + nozzle_target: number(status.get("extruder").and_then(|e| e.get("target"))), + bed_temp: number(status.get("heater_bed").and_then(|e| e.get("temperature"))), + bed_target: number(status.get("heater_bed").and_then(|e| e.get("target"))), + print_error: (state == "error").then_some(1), + ..PrinterStatus::default() + }) +} + +async fn poll(http: &Http) -> Option { + match http.get_json(STATUS_QUERY).await { + Ok(answer) => parse_status(&answer), + // Moonraker answers 503 to object queries while Klipper itself is + // down, but still answers `/printer/info` - which is the + // difference between "off" and "faulted". + Err(_) => { + let info = http.get_json("/printer/info").await.ok()?; + let state = info.get("result")?.get("state")?.as_str()?; + matches!(state, "error" | "shutdown").then(|| PrinterStatus { + state: Some("FAILED".into()), + print_error: Some(1), + ..PrinterStatus::default() + }) } } } impl GenericPrinter for KlipperPrinter { - /// OVERRIDE: real version is `POST /printer/print/pause`, `/resume` or - /// `/cancel` on Moonraker - note Klipper calls it "cancel" where Bambu - /// calls it "stop". - async fn control_print(&mut self, action: super::PrintControl) -> Result<(), PrinterError> { + /// Note Klipper calls it "cancel" where Bambu calls it "stop". + async fn control_print(&mut self, action: PrintControl) -> Result<(), PrinterError> { let endpoint = match action { - super::PrintControl::Pause => "pause", - super::PrintControl::Resume => "resume", - super::PrintControl::Stop => "cancel", + PrintControl::Pause => "pause", + PrintControl::Resume => "resume", + PrintControl::Stop => "cancel", }; - println!("[{}] POST /printer/print/{endpoint}", self.base.name); - Ok(()) + self.http.post_empty(&format!("/printer/print/{endpoint}")).await.map(|_| ()) } - fn base(&self) -> &PrinterBase { &self.base } - /// OVERRIDE: a third, again-different body - Moonraker's real feed is a - /// WebSocket (see `adapters::moonraker`). This also shows the `Err` - /// path: an empty `host` fails immediately, same as any other vendor - /// would if it couldn't reach its printer. + /// Checks Moonraker answers, then starts the status poller. Klipper + /// itself being down is not a connect failure: Moonraker is up and + /// will report it. async fn connect(&mut self) -> Result<(), PrinterError> { - if self.base.host.is_empty() { + if self.base.host.trim().is_empty() { return Err(PrinterError(format!("{} has no host configured", self.base.name))); } - println!("[{}] connecting over WebSocket (Moonraker)", self.base.name); + self.http + .get_json("/printer/info") + .await + .map_err(|err| PrinterError(format!("{} did not answer as Moonraker: {err}", self.base.name)))?; + let http = self.http.clone(); + self.feed = Some(StatusFeed::start(POLL_EVERY, move || { + let http = http.clone(); + async move { poll(&http).await } + })); Ok(()) } + /// Nothing to publish to: Moonraker is request/response. async fn publish(&mut self, _payload: String) -> Result<(), PrinterError> { - // TODO: add publishing via HTTP - Ok(()) + Err(PrinterError("Klipper has no publish channel".into())) } - /// OVERRIDE: Moonraker's own **hypothetical** native API, not gcode injection - no - /// percent-to-PWM conversion needed since it already takes a - /// percentage directly. Real version: - /// `POST /printer/gcode/script?script=SET_FAN_SPEED SPEED=0.NN`. + /// `M106` works on Klipper for the part-cooling fan, same as elsewhere. async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> { - println!("[{}] Moonraker: SET_FAN_SPEED SPEED={:.2}", self.base.name, percent as f32 / 100.0); - Ok(()) + self.send_gcode_line(format!("M106 S{}", super::percent_to_pwm(percent))).await } - /// OVERRIDE: real version is a multipart `POST /server/files/upload` - /// with `data` as the file part - Moonraker stores it on the printer's - /// virtual SD card, same upload-then-start model as Bambu/PrusaLink. + /// Uploads to the printer's `gcodes` root and starts it (`print=true`), + /// like Bambu's upload-then-`project_file`. Klipper runs plain G-code, + /// so a slicer project (`.3mf`) is refused here rather than uploaded + /// and left to fail on the machine. async fn upload_project_file(&mut self, file_name: String, data: Vec) -> Result<(), PrinterError> { - println!("[{}] POST /server/files/upload {file_name} ({} bytes)", self.base.name, data.len()); + let name = safe_file_name(&file_name)?; + let lower = name.to_lowercase(); + if !(lower.ends_with(".gcode") || lower.ends_with(".gco") || lower.ends_with(".g")) { + return Err(PrinterError(format!( + "{name} is not G-code - Klipper needs a sliced .gcode file, not a project" + ))); + } + if let Some(status) = self.status() + && matches!(status.state.as_deref(), Some("RUNNING" | "PAUSE")) + { + return Err(PrinterError(format!("{} is already printing", self.base.name))); + } + let answer = self + .http + .post_file("/server/files/upload", &[("root", "gcodes"), ("print", "true")], "file", &name, &data) + .await?; + if answer.get("print_started").and_then(Value::as_bool) == Some(false) { + return Err(PrinterError(format!("{name} was uploaded but Klipper did not start it"))); + } Ok(()) } - /// OVERRIDE: Moonraker's gcode-script endpoint, one line per call - same - /// idea as `set_fan_speed`'s `SET_FAN_SPEED` call above, just with an - /// arbitrary line instead of a hardcoded one. async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> { - println!("[{}] POST /printer/gcode/script?script={gcode_line}", self.base.name); - Ok(()) + self.http.post_json("/printer/gcode/script", &json!({ "script": gcode_line })).await.map(|_| ()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::printer::mock_http::MockServer; + + const PRINTING: &str = r#"{"result":{"status":{ + "print_stats":{"state":"printing","filename":"cube.gcode","print_duration":600.0,"info":{"current_layer":12,"total_layer":60}}, + "virtual_sdcard":{"progress":0.25}, + "extruder":{"temperature":214.8,"target":215.0}, + "heater_bed":{"temperature":59.9,"target":60.0}}}}"#; + + #[test] + fn a_printing_machine_maps_to_the_shared_vocabulary() { + let status = parse_status(&serde_json::from_str(PRINTING).unwrap()).unwrap(); + assert_eq!(status.state.as_deref(), Some("RUNNING")); + assert_eq!(status.percent, Some(25)); + assert_eq!((status.layer, status.total_layers), (Some(12), Some(60))); + // 600 s for a quarter: 1800 s more. + assert_eq!(status.remaining_minutes, Some(30)); + assert_eq!(status.job_name.as_deref(), Some("cube.gcode")); + assert_eq!(status.nozzle_temp, Some(214.8)); + assert_eq!(status.bed_target, Some(60.0)); + assert_eq!(status.print_error, None); + } + + #[test] + fn complete_is_100_percent_and_error_carries_a_code() { + let done = r#"{"result":{"status":{"print_stats":{"state":"complete","filename":"a.gcode"},"virtual_sdcard":{"progress":1.0}}}}"#; + let status = parse_status(&serde_json::from_str(done).unwrap()).unwrap(); + assert_eq!((status.state.as_deref(), status.percent), (Some("FINISH"), Some(100))); + + let failed = r#"{"result":{"status":{"print_stats":{"state":"error"}}}}"#; + let status = parse_status(&serde_json::from_str(failed).unwrap()).unwrap(); + assert_eq!((status.state.as_deref(), status.print_error), (Some("FAILED"), Some(1))); + } + + #[tokio::test] + async fn control_gcode_and_fan_go_to_the_right_endpoints() { + let mock = MockServer::start().await; + mock.answer("GET /printer/info", 200, r#"{"result":{"state":"ready"}}"#); + mock.answer("POST /printer/print/cancel", 200, r#"{"result":"ok"}"#); + mock.answer("POST /printer/gcode/script", 200, r#"{"result":"ok"}"#); + let mut printer = KlipperPrinter::new("k1", "Voron", &mock.host()); + printer.connect().await.unwrap(); + + printer.control_print(PrintControl::Stop).await.unwrap(); + assert_eq!(mock.sent("POST", "/printer/print/cancel").len(), 1); + + printer.set_fan_speed(50).await.unwrap(); + let sent = mock.sent("POST", "/printer/gcode/script"); + assert_eq!(sent[0].text(), r#"{"script":"M106 S127"}"#); + } + + #[tokio::test] + async fn connect_fails_when_nothing_answers_as_moonraker() { + let mock = MockServer::start().await; // every route is a 404 + let mut printer = KlipperPrinter::new("k1", "Voron", &mock.host()); + assert!(printer.connect().await.is_err()); + } + + #[tokio::test] + async fn the_poller_feeds_status_and_clears_it_when_the_printer_goes() { + let mock = MockServer::start().await; + mock.answer("GET /printer/info", 200, r#"{"result":{"state":"ready"}}"#); + mock.answer("GET /printer/objects/query", 200, PRINTING); + let mut printer = KlipperPrinter::new("k1", "Voron", &mock.host()); + printer.connect().await.unwrap(); + + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!(printer.status().unwrap().percent, Some(25)); + + // Klipper shuts down: queries 503, /printer/info says why. + mock.answer("GET /printer/objects/query", 503, "{}"); + mock.answer("GET /printer/info", 200, r#"{"result":{"state":"shutdown"}}"#); + tokio::time::sleep(Duration::from_millis(2500)).await; + assert_eq!(printer.status().unwrap().state.as_deref(), Some("FAILED")); + } + + #[tokio::test] + async fn upload_sends_the_file_with_print_true_and_refuses_a_project() { + let mock = MockServer::start().await; + mock.answer("GET /printer/info", 200, r#"{"result":{"state":"ready"}}"#); + mock.answer("POST /server/files/upload", 201, r#"{"print_started":true}"#); + let mut printer = KlipperPrinter::new("k1", "Voron", &mock.host()); + printer.connect().await.unwrap(); + + printer.upload_project_file("jobs/cube.gcode".into(), b"G28\nG1 X1\n".to_vec()).await.unwrap(); + let sent = mock.sent("POST", "/server/files/upload"); + assert_eq!(sent.len(), 1); + let body = sent[0].text(); + assert!(body.contains("name=\"print\"\r\n\r\ntrue")); + assert!(body.contains("filename=\"cube.gcode\"")); + assert!(body.contains("G28\nG1 X1\n")); + assert!(sent[0].header("content-type").unwrap().starts_with("multipart/form-data; boundary=")); + + let err = printer.upload_project_file("x.3mf".into(), vec![1]).await.unwrap_err(); + assert!(err.0.contains("not G-code"), "{err}"); + assert_eq!(mock.sent("POST", "/server/files/upload").len(), 1); + } + + #[tokio::test] + async fn an_upload_is_refused_while_a_print_runs() { + let mock = MockServer::start().await; + mock.answer("GET /printer/info", 200, r#"{"result":{"state":"ready"}}"#); + mock.answer("GET /printer/objects/query", 200, PRINTING); + let mut printer = KlipperPrinter::new("k1", "Voron", &mock.host()); + printer.connect().await.unwrap(); + tokio::time::sleep(Duration::from_millis(300)).await; + + let err = printer.upload_project_file("b.gcode".into(), vec![1]).await.unwrap_err(); + assert!(err.0.contains("already printing"), "{err}"); + assert!(mock.sent("POST", "/server/files/upload").is_empty()); } } diff --git a/src/printer/mock_http.rs b/src/printer/mock_http.rs new file mode 100644 index 0000000..395753f --- /dev/null +++ b/src/printer/mock_http.rs @@ -0,0 +1,118 @@ +//! A tiny HTTP server for the REST drivers' tests: canned answers per +//! `METHOD /path` (query string ignored), and a record of what was sent. +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; + +#[derive(Debug, Clone)] +pub struct Recorded { + pub method: String, + pub path: String, + pub headers: Vec<(String, String)>, + pub body: Vec, +} + +impl Recorded { + pub fn header(&self, name: &str) -> Option<&str> { + self.headers.iter().find(|(k, _)| k.eq_ignore_ascii_case(name)).map(|(_, v)| v.as_str()) + } + pub fn text(&self) -> String { + String::from_utf8_lossy(&self.body).into_owned() + } +} + +#[derive(Clone)] +pub struct MockServer { + pub addr: SocketAddr, + routes: Arc>>, + requests: Arc>>, +} + +impl MockServer { + pub async fn start() -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let server = Self { + addr: listener.local_addr().unwrap(), + routes: Arc::default(), + requests: Arc::default(), + }; + let shared = server.clone(); + tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { return }; + let shared = shared.clone(); + tokio::spawn(async move { shared.handle(stream).await }); + } + }); + server + } + + pub fn host(&self) -> String { + self.addr.to_string() + } + + /// `route` is e.g. `"GET /printer/info"`. + pub fn answer(&self, route: &str, status: u16, body: &str) { + self.routes.lock().unwrap().insert(route.to_string(), (status, body.to_string())); + } + + pub fn requests(&self) -> Vec { + self.requests.lock().unwrap().clone() + } + + pub fn sent(&self, method: &str, path: &str) -> Vec { + self.requests().into_iter().filter(|r| r.method == method && r.path == path).collect() + } + + async fn handle(&self, mut stream: tokio::net::TcpStream) { + let mut buf = Vec::new(); + let mut chunk = [0u8; 8192]; + let head_end = loop { + if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") { + break i + 4; + } + match stream.read(&mut chunk).await { + Ok(0) | Err(_) => return, + Ok(n) => buf.extend_from_slice(&chunk[..n]), + } + }; + let head = String::from_utf8_lossy(&buf[..head_end]).into_owned(); + let mut lines = head.split("\r\n"); + let mut first = lines.next().unwrap_or("").split(' '); + let method = first.next().unwrap_or("").to_string(); + let target = first.next().unwrap_or("").to_string(); + let headers: Vec<(String, String)> = lines + .filter_map(|l| l.split_once(':').map(|(k, v)| (k.trim().to_string(), v.trim().to_string()))) + .collect(); + let length: usize = headers + .iter() + .find(|(k, _)| k.eq_ignore_ascii_case("content-length")) + .and_then(|(_, v)| v.parse().ok()) + .unwrap_or(0); + while buf.len() < head_end + length { + match stream.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(n) => buf.extend_from_slice(&chunk[..n]), + } + } + let path = target.split('?').next().unwrap_or("").to_string(); + let body = buf[head_end..].to_vec(); + let (status, answer) = self + .routes + .lock() + .unwrap() + .get(&format!("{method} {path}")) + .cloned() + .unwrap_or((404, "{}".to_string())); + self.requests.lock().unwrap().push(Recorded { method, path, headers, body }); + let response = format!( + "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{answer}", + answer.len() + ); + let _ = stream.write_all(response.as_bytes()).await; + let _ = stream.shutdown().await; + } +} diff --git a/src/printer/mod.rs b/src/printer/mod.rs index 6fcba5f..8a9eafa 100644 --- a/src/printer/mod.rs +++ b/src/printer/mod.rs @@ -24,16 +24,21 @@ //! Real MQTT is built now - `bambu.rs`/`bambu_mqtt.rs` for Bambu's //! protocol, `bambu_ftps.rs` for getting files onto a Bambu printer, //! `marlin_serial.rs` for a real Marlin serial connection - see each -//! module's own doc comment. PrusaLink/Klipper's HTTP clients are still -//! stubs; that's the next piece to build past this module. +//! module's own doc comment. Klipper (`klipper.rs`) and OctoPrint/PrusaLink +//! (`octoprint.rs`) are real HTTP clients, tested against mock servers only. mod bambu; mod bambu_camera; mod bambu_commands; mod bambu_ftps; mod bambu_mqtt; +mod http; mod klipper; mod marlin_serial; +#[cfg(test)] +pub(crate) mod mock_http; +mod octoprint; +mod polled; mod prusa; // Reading a `.3mf` well enough to know which filament a plate wants. Kept @@ -54,7 +59,54 @@ pub use bambu_commands::{ XcamSettings, }; pub use klipper::KlipperPrinter; -pub use prusa::{OctoPrintFlavour, OctoPrintPrinter, PrusaSerialPrinter}; +pub use octoprint::{OctoPrintFlavour, OctoPrintPrinter}; +pub use polled::StatusFeed; +pub use prusa::PrusaSerialPrinter; + +/// What the cloud is told about a printer's job, whatever the vendor. +/// +/// `state` uses the strings Bambu reports (`IDLE`, `RUNNING`, `PAUSE`, +/// `FINISH`, `FAILED`) because the cloud's `mapStatus` and the completion +/// watcher already speak them; the other vendors translate into them. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct PrinterStatus { + pub state: Option, + pub percent: Option, + pub stage: Option, + pub layer: Option, + pub total_layers: Option, + pub remaining_minutes: Option, + pub job_name: Option, + pub nozzle_temp: Option, + pub nozzle_target: Option, + pub bed_temp: Option, + pub bed_target: Option, + pub print_error: Option, + pub active_alerts: Option, +} + +impl From<&PrintReport> for PrinterStatus { + fn from(report: &PrintReport) -> Self { + Self { + state: report.gcode_state.clone(), + percent: report.mc_percent, + stage: report.stg_cur, + layer: report.layer_num, + total_layers: report.total_layer_num, + remaining_minutes: report.mc_remaining_time, + // An idle printer reports `""` rather than omitting the field; + // empty-string-as-absent is worth normalising here so the cloud + // doesn't have to know that. + job_name: report.subtask_name.clone().filter(|s| !s.is_empty()), + nozzle_temp: report.nozzle_temper, + nozzle_target: report.nozzle_target_temper, + bed_temp: report.bed_temper, + bed_target: report.bed_target_temper, + print_error: report.print_error, + active_alerts: report.hms.as_ref().map(|h| h.len()), + } + } +} /// A tiny error type: just a message. `String` would work too - this exists /// mainly so `?` has something concrete to convert into everywhere. @@ -204,6 +256,20 @@ impl PrinterHandle { } } + /// The latest vendor-neutral status, if the printer has one. Bambu's + /// comes from its MQTT cache, the HTTP vendors' from their pollers; the + /// serial printer has none. + pub async fn status(&self) -> Option { + match self { + PrinterHandle::Bambu(p) => p.print_status().await.as_ref().map(PrinterStatus::from), + PrinterHandle::OctoPrint(p) => p.status(), + PrinterHandle::Klipper(p) => p.status(), + PrinterHandle::PrusaSerial(_) => None, + } + } + + /// Bambu's full report, for what is Bambu-specific (camera, AMS). + /// /// The latest status this process has for the printer, if it has one. /// /// An inherent method rather than a `GenericPrinter` trait method, diff --git a/src/printer/octoprint.rs b/src/printer/octoprint.rs new file mode 100644 index 0000000..703a503 --- /dev/null +++ b/src/printer/octoprint.rs @@ -0,0 +1,417 @@ +//! OctoPrint and PrusaLink over HTTP (`X-Api-Key`). +//! +//! Written from the two projects' published REST APIs and exercised against +//! a mock server in the tests below - no real OctoPrint or PrusaLink machine +//! has been on the bench, so treat the field names as unconfirmed until one +//! has. +use std::time::Duration; + +use serde_json::{json, Value}; + +use super::http::{encode_segment, safe_file_name, Http}; +use super::polled::StatusFeed; +use super::{GenericPrinter, PrintControl, PrinterBase, PrinterError, PrinterStatus}; + +const POLL_EVERY: Duration = Duration::from_secs(3); + +/// Which implementation of the OctoPrint REST API is on the other end. +/// +/// PrusaLink deliberately implements OctoPrint's API, so one client serves +/// both - but not identically. PrusaLink adds its own `/api/v1/*` (`info`, +/// `status`, `job`) which is where the useful detail lives, while OctoPrint +/// proper has the older `/api/printer` and `/api/job` and a plugin surface +/// PrusaLink does not have. +/// +/// This is recorded from config now, while the client is still a stub, +/// because **no probe can settle it**: both answer `/api/version`, and every +/// PrusaLink identity endpoint needs a credential, so from outside the two +/// are indistinguishable (see `discovery::net::classify_octoprint_api`). It +/// is the operator's answer to a question only they can answer, so it needs +/// somewhere to live the moment they are asked - otherwise the real client +/// would have to ask them a second time. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OctoPrintFlavour { + /// Prusa's own firmware-hosted implementation, on an MK4/XL/MINI. + PrusaLink, + /// OctoPrint itself, usually on a Raspberry Pi in front of any printer. + Generic, +} + +impl OctoPrintFlavour { + pub fn as_str(self) -> &'static str { + match self { + OctoPrintFlavour::PrusaLink => "PrusaLink", + OctoPrintFlavour::Generic => "OctoPrint", + } + } +} + + +pub struct OctoPrintPrinter { + base: PrinterBase, + pub flavour: OctoPrintFlavour, + http: Http, + feed: Option, +} + +impl OctoPrintPrinter { + pub fn new(id: &str, name: &str, host: &str, api_key: &str, flavour: OctoPrintFlavour) -> Self { + Self { + base: PrinterBase { id: id.into(), name: name.into(), host: host.into() }, + flavour, + // Port 80: OctoPrint's usual front (haproxy) and PrusaLink both + // answer there. An unusual port goes in `host`. + http: Http::new(host, 80, Some(api_key)), + feed: None, + } + } + + pub fn status(&self) -> Option { + self.feed.as_ref().and_then(StatusFeed::latest) + } +} + +fn number(v: Option<&Value>) -> Option { + v.and_then(Value::as_f64).map(|n| n as f32) +} + +/// OctoPrint's `GET /api/job`, plus `GET /api/printer` for temperatures +/// (which is a 409 whenever the printer is not operational, hence `Option`). +fn octoprint_status(job: &Value, printer: Option<&Value>) -> Option { + let state = job.get("state")?.as_str()?; + let completion = number(job.get("progress").and_then(|p| p.get("completion"))); + let left = number(job.get("progress").and_then(|p| p.get("printTimeLeft"))); + + let (mapped, percent) = if state.starts_with("Printing") || state == "Cancelling" { + ("RUNNING", completion.unwrap_or(0.0)) + } else if state == "Pausing" || state == "Paused" { + ("PAUSE", completion.unwrap_or(0.0)) + } else if state == "Operational" { + // OctoPrint has no "finished" state: the job simply stays at 100%. + match completion { + Some(c) if c >= 100.0 => ("FINISH", 100.0), + _ => ("IDLE", 0.0), + } + } else if state.contains("rror") && !state.starts_with("Offline") { + ("FAILED", 0.0) + } else { + // Offline, still opening the serial port, ...: not a state worth + // reporting as the printer's. + return None; + }; + + let tool = printer.and_then(|p| p.get("temperature")).and_then(|t| t.get("tool0")); + let bed = printer.and_then(|p| p.get("temperature")).and_then(|t| t.get("bed")); + Some(PrinterStatus { + state: Some(mapped.into()), + percent: Some(percent.round() as i32), + remaining_minutes: (mapped == "RUNNING" || mapped == "PAUSE").then_some(left).flatten().map(|s| (s / 60.0).round() as i32), + job_name: job + .get("job") + .and_then(|j| j.get("file")) + .and_then(|f| f.get("display").or_else(|| f.get("name"))) + .and_then(Value::as_str) + .map(str::to_string), + nozzle_temp: number(tool.and_then(|t| t.get("actual"))), + nozzle_target: number(tool.and_then(|t| t.get("target"))), + bed_temp: number(bed.and_then(|t| t.get("actual"))), + bed_target: number(bed.and_then(|t| t.get("target"))), + print_error: (mapped == "FAILED").then_some(1), + ..PrinterStatus::default() + }) +} + +/// PrusaLink's `GET /api/v1/status`, with `GET /api/v1/job` for the file name +/// when a job exists. +fn prusalink_status(status: &Value, job: Option<&Value>) -> Option { + let printer = status.get("printer")?; + let state = printer.get("state")?.as_str()?; + let mapped = match state { + "IDLE" | "STOPPED" => "IDLE", + "BUSY" => "PREPARE", + "PRINTING" => "RUNNING", + "PAUSED" | "ATTENTION" => "PAUSE", + "FINISHED" => "FINISH", + "ERROR" => "FAILED", + _ => return None, + }; + let running = matches!(mapped, "RUNNING" | "PAUSE"); + let progress = number(status.get("job").and_then(|j| j.get("progress"))); + let left = number(status.get("job").and_then(|j| j.get("time_remaining"))); + Some(PrinterStatus { + state: Some(mapped.into()), + percent: Some(match mapped { + "FINISH" => 100, + _ if running => progress.unwrap_or(0.0).round() as i32, + _ => 0, + }), + remaining_minutes: running.then_some(left).flatten().map(|s| (s / 60.0).round() as i32), + job_name: job + .and_then(|j| j.get("file")) + .and_then(|f| f.get("display_name").or_else(|| f.get("name"))) + .and_then(Value::as_str) + .map(str::to_string), + nozzle_temp: number(printer.get("temp_nozzle")), + nozzle_target: number(printer.get("target_nozzle")), + bed_temp: number(printer.get("temp_bed")), + bed_target: number(printer.get("target_bed")), + print_error: (mapped == "FAILED").then_some(1), + ..PrinterStatus::default() + }) +} + +async fn poll(http: &Http, flavour: OctoPrintFlavour) -> Option { + match flavour { + OctoPrintFlavour::Generic => { + let job = http.get_json("/api/job").await.ok()?; + let printer = http.get_json("/api/printer").await.ok(); + octoprint_status(&job, printer.as_ref()) + } + OctoPrintFlavour::PrusaLink => { + let status = http.get_json("/api/v1/status").await.ok()?; + let has_job = status.get("job").is_some_and(|j| !j.is_null()); + let job = if has_job { http.get_json("/api/v1/job").await.ok() } else { None }; + prusalink_status(&status, job.as_ref()) + } + } +} + +impl GenericPrinter for OctoPrintPrinter { + /// OctoPrint: `POST /api/job` with a command. PrusaLink addresses the + /// running job by id, so the current job is fetched first. + async fn control_print(&mut self, action: PrintControl) -> Result<(), PrinterError> { + match self.flavour { + OctoPrintFlavour::Generic => { + let body = match action { + PrintControl::Pause => json!({ "command": "pause", "action": "pause" }), + PrintControl::Resume => json!({ "command": "pause", "action": "resume" }), + PrintControl::Stop => json!({ "command": "cancel" }), + }; + self.http.post_json("/api/job", &body).await.map(|_| ()) + } + OctoPrintFlavour::PrusaLink => { + let status = self.http.get_json("/api/v1/status").await?; + let id = status + .get("job") + .and_then(|j| j.get("id")) + .and_then(Value::as_i64) + .ok_or_else(|| PrinterError(format!("{} has no job to {}", self.base.name, action.as_str())))?; + match action { + PrintControl::Pause => self.http.put_empty(&format!("/api/v1/job/{id}/pause")).await, + PrintControl::Resume => self.http.put_empty(&format!("/api/v1/job/{id}/resume")).await, + PrintControl::Stop => self.http.delete(&format!("/api/v1/job/{id}")).await, + } + .map(|_| ()) + } + } + } + + fn base(&self) -> &PrinterBase { + &self.base + } + + /// Checks the printer answers with this key, then starts the poller. + async fn connect(&mut self) -> Result<(), PrinterError> { + if self.base.host.trim().is_empty() { + return Err(PrinterError(format!("{} has no host configured", self.base.name))); + } + // `/api/version` needs no permissions on either, so it would pass + // with a wrong key; these two are the first that check it. + let probe = match self.flavour { + OctoPrintFlavour::Generic => "/api/job", + OctoPrintFlavour::PrusaLink => "/api/v1/status", + }; + self.http + .get_json(probe) + .await + .map_err(|err| PrinterError(format!("{} did not answer as {}: {err}", self.base.name, self.flavour.as_str())))?; + let (http, flavour) = (self.http.clone(), self.flavour); + self.feed = Some(StatusFeed::start(POLL_EVERY, move || { + let http = http.clone(); + async move { poll(&http, flavour).await } + })); + Ok(()) + } + + async fn publish(&mut self, _payload: String) -> Result<(), PrinterError> { + Err(PrinterError(format!("{} has no publish channel", self.flavour.as_str()))) + } + + async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> { + self.send_gcode_line(format!("M106 S{}", super::percent_to_pwm(percent))).await + } + + /// Uploads and starts the print. OctoPrint takes a multipart POST to + /// `/api/files/local`; PrusaLink a raw `PUT /api/v1/files/local/{name}` + /// with `Print-After-Upload`. Both run sliced G-code, not a project. + async fn upload_project_file(&mut self, file_name: String, data: Vec) -> Result<(), PrinterError> { + let name = safe_file_name(&file_name)?; + let lower = name.to_lowercase(); + let printable = [".gcode", ".gco", ".g", ".bgcode"].iter().any(|ext| lower.ends_with(ext)); + if !printable { + return Err(PrinterError(format!( + "{name} is not G-code - {} needs a sliced file, not a project", + self.flavour.as_str() + ))); + } + if let Some(status) = self.status() + && matches!(status.state.as_deref(), Some("RUNNING" | "PAUSE" | "PREPARE")) + { + return Err(PrinterError(format!("{} is already printing", self.base.name))); + } + match self.flavour { + OctoPrintFlavour::Generic => self + .http + .post_file("/api/files/local", &[("select", "true"), ("print", "true")], "file", &name, &data) + .await + .map(|_| ()), + OctoPrintFlavour::PrusaLink => self + .http + .put_bytes( + &format!("/api/v1/files/local/{}", encode_segment(&name)), + &[("Print-After-Upload", "?1"), ("Overwrite", "?1")], + data, + ) + .await + .map(|_| ()), + } + } + + async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> { + match self.flavour { + OctoPrintFlavour::Generic => self + .http + .post_json("/api/printer/command", &json!({ "commands": [gcode_line] })) + .await + .map(|_| ()), + // PrusaLink's API has no G-code endpoint. + OctoPrintFlavour::PrusaLink => { + Err(PrinterError("PrusaLink cannot be sent G-code lines".into())) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::printer::mock_http::MockServer; + + const JOB: &str = r#"{"job":{"file":{"name":"cube.gcode","display":"cube.gcode"}}, + "progress":{"completion":40.0,"printTime":600,"printTimeLeft":900},"state":"Printing"}"#; + const TEMPS: &str = + r#"{"temperature":{"tool0":{"actual":209.5,"target":210.0},"bed":{"actual":60.0,"target":60.0}}}"#; + + #[test] + fn octoprint_states_map_to_the_shared_vocabulary() { + let status = octoprint_status(&serde_json::from_str(JOB).unwrap(), serde_json::from_str(TEMPS).ok().as_ref()).unwrap(); + assert_eq!(status.state.as_deref(), Some("RUNNING")); + assert_eq!(status.percent, Some(40)); + assert_eq!(status.remaining_minutes, Some(15)); + assert_eq!(status.job_name.as_deref(), Some("cube.gcode")); + assert_eq!(status.nozzle_target, Some(210.0)); + + // Finished: still "Operational", job stays at 100%. No temperatures + // (the printer endpoint was a 409) is fine. + let done = r#"{"state":"Operational","progress":{"completion":100.0}}"#; + let status = octoprint_status(&serde_json::from_str(done).unwrap(), None).unwrap(); + assert_eq!((status.state.as_deref(), status.percent, status.nozzle_temp), (Some("FINISH"), Some(100), None)); + + let idle = r#"{"state":"Operational","progress":{"completion":null}}"#; + assert_eq!(octoprint_status(&serde_json::from_str(idle).unwrap(), None).unwrap().state.as_deref(), Some("IDLE")); + + let offline = r#"{"state":"Offline","progress":{}}"#; + assert!(octoprint_status(&serde_json::from_str(offline).unwrap(), None).is_none()); + let broken = r#"{"state":"Error","progress":{}}"#; + assert_eq!(octoprint_status(&serde_json::from_str(broken).unwrap(), None).unwrap().print_error, Some(1)); + } + + #[test] + fn prusalink_states_map_to_the_shared_vocabulary() { + let status = r#"{"printer":{"state":"PRINTING","temp_nozzle":215.1,"target_nozzle":215.0,"temp_bed":60.2,"target_bed":60.0}, + "job":{"id":7,"progress":33.0,"time_remaining":1200,"time_printing":300}}"#; + let job = r#"{"id":7,"file":{"name":"CUBE~1.GCO","display_name":"cube.gcode"}}"#; + let parsed = prusalink_status(&serde_json::from_str(status).unwrap(), serde_json::from_str(job).ok().as_ref()).unwrap(); + assert_eq!(parsed.state.as_deref(), Some("RUNNING")); + assert_eq!((parsed.percent, parsed.remaining_minutes), (Some(33), Some(20))); + assert_eq!(parsed.job_name.as_deref(), Some("cube.gcode")); + assert_eq!(parsed.bed_temp, Some(60.2)); + + let idle = r#"{"printer":{"state":"IDLE","temp_nozzle":25.0,"temp_bed":24.0}}"#; + let parsed = prusalink_status(&serde_json::from_str(idle).unwrap(), None).unwrap(); + assert_eq!((parsed.state.as_deref(), parsed.percent), (Some("IDLE"), Some(0))); + } + + #[tokio::test] + async fn octoprint_controls_upload_and_send_the_key() { + let mock = MockServer::start().await; + mock.answer("GET /api/job", 200, r#"{"state":"Operational","progress":{}}"#); + mock.answer("POST /api/job", 204, ""); + mock.answer("POST /api/printer/command", 204, ""); + mock.answer("POST /api/files/local", 201, r#"{"done":true}"#); + let mut printer = OctoPrintPrinter::new("o1", "Ender", &mock.host(), "KEY123", OctoPrintFlavour::Generic); + printer.connect().await.unwrap(); + + printer.control_print(PrintControl::Resume).await.unwrap(); + let sent = mock.sent("POST", "/api/job"); + assert_eq!(sent[0].text(), r#"{"action":"resume","command":"pause"}"#); + assert_eq!(sent[0].header("x-api-key"), Some("KEY123")); + printer.control_print(PrintControl::Stop).await.unwrap(); + assert_eq!(mock.sent("POST", "/api/job")[1].text(), r#"{"command":"cancel"}"#); + + printer.set_fan_speed(100).await.unwrap(); + assert_eq!(mock.sent("POST", "/api/printer/command")[0].text(), r#"{"commands":["M106 S255"]}"#); + + printer.upload_project_file("a.gcode".into(), b"G28".to_vec()).await.unwrap(); + let body = mock.sent("POST", "/api/files/local")[0].text(); + assert!(body.contains("name=\"print\"\r\n\r\ntrue") && body.contains("filename=\"a.gcode\"")); + assert!(printer.upload_project_file("a.3mf".into(), vec![1]).await.is_err()); + } + + #[tokio::test] + async fn a_wrong_key_fails_connect() { + let mock = MockServer::start().await; + mock.answer("GET /api/job", 403, r#"{"error":"forbidden"}"#); + let mut printer = OctoPrintPrinter::new("o1", "Ender", &mock.host(), "bad", OctoPrintFlavour::Generic); + assert!(printer.connect().await.unwrap_err().0.contains("403")); + } + + #[tokio::test] + async fn prusalink_addresses_the_running_job_by_id() { + let mock = MockServer::start().await; + mock.answer("GET /api/v1/status", 200, r#"{"printer":{"state":"PRINTING"},"job":{"id":7,"progress":10.0}}"#); + mock.answer("GET /api/v1/job", 200, r#"{"id":7,"file":{"display_name":"c.gcode"}}"#); + mock.answer("PUT /api/v1/job/7/pause", 204, ""); + mock.answer("DELETE /api/v1/job/7", 204, ""); + let mut printer = OctoPrintPrinter::new("p1", "MK4", &mock.host(), "K", OctoPrintFlavour::PrusaLink); + printer.connect().await.unwrap(); + + printer.control_print(PrintControl::Pause).await.unwrap(); + printer.control_print(PrintControl::Stop).await.unwrap(); + assert_eq!(mock.sent("PUT", "/api/v1/job/7/pause").len(), 1); + assert_eq!(mock.sent("DELETE", "/api/v1/job/7").len(), 1); + assert!(printer.send_gcode_line("M106".into()).await.is_err()); + + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!(printer.status().unwrap().job_name.as_deref(), Some("c.gcode")); + // A job is running, so an upload is refused before anything is sent. + let err = printer.upload_project_file("n.bgcode".into(), vec![1]).await.unwrap_err(); + assert!(err.0.contains("already printing"), "{err}"); + assert!(mock.sent("PUT", "/api/v1/files/local/n.bgcode").is_empty()); + } + + #[tokio::test] + async fn prusalink_upload_is_a_put_with_print_after_upload() { + let mock = MockServer::start().await; + mock.answer("GET /api/v1/status", 200, r#"{"printer":{"state":"IDLE"}}"#); + mock.answer("PUT /api/v1/files/local/my%20part.gcode", 201, ""); + let mut printer = OctoPrintPrinter::new("p1", "MK4", &mock.host(), "K", OctoPrintFlavour::PrusaLink); + printer.connect().await.unwrap(); + tokio::time::sleep(Duration::from_millis(300)).await; + + printer.upload_project_file("my part.gcode".into(), b"G28".to_vec()).await.unwrap(); + let sent = mock.sent("PUT", "/api/v1/files/local/my%20part.gcode"); + assert_eq!(sent[0].header("print-after-upload"), Some("?1")); + assert_eq!(sent[0].body, b"G28"); + } +} diff --git a/src/printer/polled.rs b/src/printer/polled.rs new file mode 100644 index 0000000..2dbb799 --- /dev/null +++ b/src/printer/polled.rs @@ -0,0 +1,47 @@ +//! Status for vendors that have no push feed: a background task asks the +//! printer on an interval and keeps the last answer for `status()` to read. +//! (Bambu pushes over MQTT and keeps its own cache.) +use std::future::Future; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use tokio::task::JoinHandle; + +use super::PrinterStatus; + +pub struct StatusFeed { + latest: Arc>>, + task: JoinHandle<()>, +} + +impl StatusFeed { + /// `fetch` returning `None` means "no usable answer" (printer + /// unreachable, or not ready), and clears the cache: a stale status + /// shown as current is worse than none. + pub fn start(interval: Duration, fetch: F) -> Self + where + F: Fn() -> Fut + Send + 'static, + Fut: Future> + Send, + { + let latest = Arc::new(Mutex::new(None)); + let shared = latest.clone(); + let task = tokio::spawn(async move { + loop { + let status = fetch().await; + *shared.lock().unwrap() = status; + tokio::time::sleep(interval).await; + } + }); + Self { latest, task } + } + + pub fn latest(&self) -> Option { + self.latest.lock().unwrap().clone() + } +} + +impl Drop for StatusFeed { + fn drop(&mut self) { + self.task.abort(); + } +} diff --git a/src/printer/prusa.rs b/src/printer/prusa.rs index 7c73090..800aa33 100644 --- a/src/printer/prusa.rs +++ b/src/printer/prusa.rs @@ -1,110 +1,4 @@ use super::{GenericPrinter, PrinterBase, PrinterError}; - -/****************** OctoPrint API family ******************/ - -/// Which implementation of the OctoPrint REST API is on the other end. -/// -/// PrusaLink deliberately implements OctoPrint's API, so one client serves -/// both - but not identically. PrusaLink adds its own `/api/v1/*` (`info`, -/// `status`, `job`) which is where the useful detail lives, while OctoPrint -/// proper has the older `/api/printer` and `/api/job` and a plugin surface -/// PrusaLink does not have. -/// -/// This is recorded from config now, while the client is still a stub, -/// because **no probe can settle it**: both answer `/api/version`, and every -/// PrusaLink identity endpoint needs a credential, so from outside the two -/// are indistinguishable (see `discovery::net::classify_octoprint_api`). It -/// is the operator's answer to a question only they can answer, so it needs -/// somewhere to live the moment they are asked - otherwise the real client -/// would have to ask them a second time. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum OctoPrintFlavour { - /// Prusa's own firmware-hosted implementation, on an MK4/XL/MINI. - PrusaLink, - /// OctoPrint itself, usually on a Raspberry Pi in front of any printer. - Generic, -} - -impl OctoPrintFlavour { - pub fn as_str(self) -> &'static str { - match self { - OctoPrintFlavour::PrusaLink => "PrusaLink", - OctoPrintFlavour::Generic => "OctoPrint", - } - } -} - -pub struct OctoPrintPrinter { - base: PrinterBase, - pub api_key: String, - pub flavour: OctoPrintFlavour, -} - -impl OctoPrintPrinter { - pub fn new(id: &str, name: &str, host: &str, api_key: &str, flavour: OctoPrintFlavour) -> Self { - Self { - base: PrinterBase { id: id.into(), name: name.into(), host: host.into() }, - api_key: api_key.into(), - flavour, - } - } -} - -impl GenericPrinter for OctoPrintPrinter { - /// OVERRIDE: real version is `PUT /api/v1/job/{id}/pause`, `/resume`, or - /// `DELETE /api/v1/job/{id}` to cancel - PrusaLink addresses the running - /// job by id rather than the printer, so a real implementation has to - /// fetch the current job first. - async fn control_print(&mut self, action: super::PrintControl) -> Result<(), PrinterError> { - println!("[{}] {} job control: {}", self.base.name, self.flavour.as_str(), action.as_str()); - Ok(()) - } - - - fn base(&self) -> &PrinterBase { - &self.base - } - - /// OVERRIDE: a completely different body from Bambu's - PrusaLink is a - /// plain REST API (real version: `GET /api/v1/status` with the API - /// key, see `adapters::prusalink`), no persistent session to hold. - async fn connect(&mut self) -> Result<(), PrinterError> { - println!("[{}] connecting over HTTP ({})", self.base.name, self.flavour.as_str()); - Ok(()) - } - - async fn publish(&mut self, _payload: String) -> Result<(), PrinterError> { - // TODO: add publishing via HTTP - Ok(()) - } - - /// OVERRIDE: gcode via PrusaLink's command-injection endpoint - same - /// underlying M106 as Bambu, reusing the shared percent->PWM - /// conversion, but sent over HTTP instead of MQTT. - async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> { - let pwm = super::percent_to_pwm(percent); - println!("[{}] POST /api/printer/command {{\"commands\":[\"M106 S{pwm}\"]}}", self.base.name); - Ok(()) - } - - /// OVERRIDE: real version is `PUT /api/v1/files/local/{file_name}` with - /// `data` as the request body - PrusaLink stores it on the printer's own - /// SD card, same "upload to storage the firmware owns" model as Bambu. - async fn upload_project_file(&mut self, file_name: String, data: Vec) -> Result<(), PrinterError> { - println!("[{}] PUT /api/v1/files/local/{file_name} ({} bytes)", self.base.name, data.len()); - Ok(()) - } - - /// OVERRIDE: PrusaLink's command-injection endpoint, one line per call - - /// same endpoint `set_fan_speed` above already uses for M106, just with - /// an arbitrary line instead of a hardcoded one. - async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> { - println!("[{}] POST /api/printer/command {{\"commands\":[\"{gcode_line}\"]}}", self.base.name); - Ok(()) - } -} - - /****************** PrusaSerial ******************/ pub struct PrusaSerialPrinter { diff --git a/src/provision.rs b/src/provision.rs index f494652..a6befa0 100644 --- a/src/provision.rs +++ b/src/provision.rs @@ -201,8 +201,8 @@ mod behaviour { let _ = std::fs::remove_file(path); } - /// The success path, using the OctoPrint client because its `connect` is - /// the one that succeeds without a real machine (the driver is a stub). + /// The success path, using the OctoPrint client against a mock server, + /// because no real machine is needed to connect to one. /// What this proves is the plumbing, not a vendor: a connected printer is /// saved, joins the running fleet, and is still there after a "restart" /// that reopens the store and rebuilds the fleet from it. @@ -210,7 +210,10 @@ mod behaviour { async fn an_added_printer_joins_the_fleet_and_survives_a_restart() { let (store, path) = store(); let fleet = empty_fleet(); + let mock = crate::printer::mock_http::MockServer::start().await; + mock.answer("GET /api/job", 200, r#"{"state":"Operational","progress":{}}"#); let mut req = request("octo1"); + req.host = Some(mock.host()); req.vendor = "octoprint".into(); req.sn = None; req.access_code = None; diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index 4fb2b08..78fe94c 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -374,29 +374,28 @@ async fn collect_telemetry(fleet: &SharedFleet) -> Vec { for printer in fleet.iter() { let base = printer.base(); - let Some(report) = printer.print_status().await else { + let Some(status) = printer.status().await else { continue; }; + // Only Bambu has a camera report to derive these from. + let report = printer.print_status().await; out.push(PrinterTelemetry { printer_id: base.id.clone(), name: base.name.clone(), - state: report.gcode_state.clone(), - percent: report.mc_percent, - stage: report.stg_cur, - layer: report.layer_num, - total_layers: report.total_layer_num, - remaining_minutes: report.mc_remaining_time, - // An idle printer reports `""` rather than omitting the field; - // empty-string-as-absent is worth normalising here so the cloud - // doesn't have to know that. - job_name: report.subtask_name.clone().filter(|s| !s.is_empty()), - nozzle_temp: report.nozzle_temper, - nozzle_target: report.nozzle_target_temper, - bed_temp: report.bed_temper, - bed_target: report.bed_target_temper, - print_error: report.print_error, - active_alerts: report.hms.as_ref().map(|h| h.len()), + state: status.state, + percent: status.percent, + stage: status.stage, + layer: status.layer, + total_layers: status.total_layers, + remaining_minutes: status.remaining_minutes, + job_name: status.job_name, + nozzle_temp: status.nozzle_temp, + nozzle_target: status.nozzle_target, + bed_temp: status.bed_temp, + bed_target: status.bed_target, + print_error: status.print_error, + active_alerts: status.active_alerts, // 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 @@ -404,13 +403,13 @@ async fn collect_telemetry(fleet: &SharedFleet) -> Vec { // 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()), + report.as_ref().and_then(|r| r.ipcam.as_ref()).and_then(|c| c.rtsp_url.as_deref()), ) .map(|source| source.stream_name), camera_snapshot: crate::camera::wants_snapshot( printer, - report.ipcam.is_some(), - report.ipcam.as_ref().and_then(|c| c.rtsp_url.as_deref()), + report.as_ref().is_some_and(|r| r.ipcam.is_some()), + report.as_ref().and_then(|r| r.ipcam.as_ref()).and_then(|c| c.rtsp_url.as_deref()), ) .then_some(true), });