diff --git a/examples/bambu_mapping_check.rs b/examples/bambu_mapping_check.rs index 4b0d159..a542870 100644 --- a/examples/bambu_mapping_check.rs +++ b/examples/bambu_mapping_check.rs @@ -47,6 +47,10 @@ async fn main() -> anyhow::Result<()> { for f in continuum_proxy::printer::plate_filaments(&bytes, 1)? { println!("project filament {}: {} {} #{}", f.index, f.filament_type, f.tray_info_idx, f.color); } + if let Some(slot) = std::env::var("AMS_SLOT").ok().and_then(|v| v.parse::().ok()) { + println!("(with AMS_SLOT={slot} chosen explicitly)"); + bambu.set_ams_slot_override(Some(slot)); + } match bambu.filament_source_for(&bytes, 1).await { Ok(source) => println!("mapping the production path would use: {source:?}"), Err(err) => println!("the production path would REFUSE: {err}"), diff --git a/src/printer/bambu.rs b/src/printer/bambu.rs index c27ccea..39372e0 100644 --- a/src/printer/bambu.rs +++ b/src/printer/bambu.rs @@ -246,6 +246,8 @@ pub struct BambuPrinter { /// alongside the protocol version it settled on, since the two are /// decided together and always used together - see `bambu_mqtt.rs`. connection: Option<(super::bambu_mqtt::BambuMqttSession, BambuProtocolVersion)>, + /// See `set_ams_slot_override`. + ams_slot_override: Option, } /// Which plate of a multi-plate project gets printed. @@ -491,9 +493,17 @@ impl BambuPrinter { tls, preference, connection: None, + ams_slot_override: None, } } + /// The operator's chosen AMS slot for the next print (0-based across all + /// units), or `None` to map by the project's own filament. Set by the + /// uplink around one `start_print` and cleared after it. + pub fn set_ams_slot_override(&mut self, slot: Option) { + self.ams_slot_override = slot; + } + pub fn base(&self) -> &PrinterBase { &self.base } @@ -758,7 +768,8 @@ impl BambuPrinter { .map_err(|err| PrinterError(format!("failed to build pushall request: {err}")))?; body["push_target"] = serde_json::Value::from(1); - session.publish("pushing", &body).await + // Unacked: nothing waits on a reply to this, and see `publish_unacked`. + session.publish_unacked("pushing", &body).await } /// Just the FTPS transfer: puts `data` in the printer's `cache/` @@ -874,7 +885,18 @@ impl BambuPrinter { // naming a filament that also happens to be on the spool should // still use the AMS, because that is what it was sliced for. if !trays.is_empty() { - match super::map_to_trays(&filaments, &trays) { + let mapped = match self.ams_slot_override { + Some(slot) => super::map_to_slot(&filaments, &trays, i32::from(slot)), + None => super::map_to_trays(&filaments, &trays), + }; + // An explicit slot that cannot be honoured is an error, never a + // fallback to some other spool. + if self.ams_slot_override.is_some() { + if let Err(err) = &mapped { + return Err(PrinterError(format!("{}: {err}", self.base.name))); + } + } + match mapped { Ok(mapping) => { tracing::info!( printer = %self.base.name, diff --git a/src/printer/bambu_mqtt.rs b/src/printer/bambu_mqtt.rs index 146cf37..de77903 100644 --- a/src/printer/bambu_mqtt.rs +++ b/src/printer/bambu_mqtt.rs @@ -76,6 +76,9 @@ use super::PrinterError; type PendingTable = Arc>>>; const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); + +/// How long a publish may wait for room in the MQTT request queue. +const PUBLISH_TIMEOUT: Duration = Duration::from_secs(10); /// Both rumqttc clients default to a 10KB max packet size in *both* /// directions - fine for a request, nowhere near enough for a reply: /// confirmed live against a real H2C, whose `get_version` reply alone @@ -359,18 +362,38 @@ impl BambuMqttSession { /// `publish` above and `GenericPrinter::publish` (a raw pass-through, /// no envelope opinion at all) are built on. pub async fn publish_raw(&self, payload: String) -> Result<(), PrinterError> { + self.publish_with_qos(payload, true).await + } + + /// Like `publish`, but QoS 0: for requests that need no acknowledgement + /// (a status refresh). QoS 1 publishes the printer never acknowledges + /// fill rumqttc's in-flight window, after which every publish blocks + /// forever - found on the first real dispatch, where that froze the whole + /// proxy about 4.5 minutes in, because callers hold the fleet lock. + pub async fn publish_unacked(&self, category: &str, body: &Value) -> Result<(), PrinterError> { + self.publish_with_qos(serde_json::json!({ category: body }).to_string(), false).await + } + + async fn publish_with_qos(&self, payload: String, acked: bool) -> Result<(), PrinterError> { let topic = format!("device/{}/request", self.sn); - match &self.client { - BambuMqttClient::V1(client) => client - .publish(topic, QoS::AtLeastOnce, false, payload) - .await - .map_err(|err| PrinterError(format!("publish failed: {err}"))), - BambuMqttClient::V2(client) => client - .publish(topic, QoSV5::AtLeastOnce, false, payload) - .await - .map_err(|err| PrinterError(format!("publish failed: {err}"))), - } + // Bounded: a publish waits for room in rumqttc's request queue, and an + // unbounded wait here wedges every caller that holds the fleet lock. + let publishing = async { + match &self.client { + BambuMqttClient::V1(client) => client + .publish(topic, if acked { QoS::AtLeastOnce } else { QoS::AtMostOnce }, false, payload) + .await + .map_err(|err| PrinterError(format!("publish failed: {err}"))), + BambuMqttClient::V2(client) => client + .publish(topic, if acked { QoSV5::AtLeastOnce } else { QoSV5::AtMostOnce }, false, payload) + .await + .map_err(|err| PrinterError(format!("publish failed: {err}"))), + } + }; + tokio::time::timeout(PUBLISH_TIMEOUT, publishing) + .await + .map_err(|_| PrinterError(format!("{} publish timed out after {PUBLISH_TIMEOUT:?} - the MQTT queue is not draining", self.sn)))? } /// The current in-memory `print.push_status` object, if at least one diff --git a/src/printer/mod.rs b/src/printer/mod.rs index d0d03b8..6fcba5f 100644 --- a/src/printer/mod.rs +++ b/src/printer/mod.rs @@ -40,7 +40,7 @@ mod prusa; // free of printer types so it is testable without a printer. mod project_file; pub use project_file::{ - map_to_trays, normalise_color, plate_filaments, AvailableTray, ProjectFilament, + map_to_slot, map_to_trays, normalise_color, plate_filaments, AvailableTray, ProjectFilament, ProjectFileError, }; diff --git a/src/printer/project_file.rs b/src/printer/project_file.rs index ca07ef5..4bb3374 100644 --- a/src/printer/project_file.rs +++ b/src/printer/project_file.rs @@ -71,6 +71,8 @@ pub enum ProjectFileError { /// material or none. #[error("no loaded tray matches {wanted} (filament index {index}); loaded: {loaded}")] NoMatchingTray { index: usize, wanted: String, loaded: String }, + #[error("cannot use AMS slot {0}: {1}")] + BadSlot(i32, String), } /// Normalises a colour to six uppercase hex digits. @@ -271,6 +273,36 @@ pub fn map_to_trays( } /// Match quality, lower is better; `None` means no match. +/// An explicit choice of slot instead of the colour-based pick: the operator +/// said which spool to use. Deliberately narrow - a single-filament project +/// only (which filament goes where is ambiguous otherwise), onto a slot that +/// is loaded and holds the same *material type*, because the type decides +/// the temperatures. A different colour is allowed: that is the point. +pub fn map_to_slot( + filaments: &[ProjectFilament], + trays: &[AvailableTray], + slot: i32, +) -> Result, ProjectFileError> { + let [filament] = filaments else { + return Err(ProjectFileError::BadSlot( + slot, + format!("the project uses {} filaments; a chosen slot is only for single-filament projects", filaments.len()), + )); + }; + let tray = trays.iter().find(|t| t.slot == slot).ok_or_else(|| { + ProjectFileError::BadSlot(slot, "no filament is loaded in that slot".into()) + })?; + if !tray.filament_type.eq_ignore_ascii_case(&filament.filament_type) { + return Err(ProjectFileError::BadSlot( + slot, + format!("it holds {} but the project was sliced for {}", tray.filament_type, filament.filament_type), + )); + } + let mut mapping = vec![-1; filament.index + 1]; + mapping[filament.index] = slot; + Ok(mapping) +} + fn tier(filament: &ProjectFilament, tray: &AvailableTray) -> Option { let same_product = filament.tray_info_idx.eq_ignore_ascii_case(&tray.tray_info_idx); let same_color = filament.color == tray.color; @@ -288,6 +320,18 @@ mod tests { use super::*; use std::io::Write; + #[test] + fn a_chosen_slot_overrides_the_colour_pick_but_not_the_material() { + let f = [filament(0, "GFG99", "PETG", "161616")]; + let t = [tray(2, "GFG99", "PETG", "161616"), tray(3, "GFG99", "PETG", "2850E0"), tray(1, "GFL99", "PLA", "D3C5A3")]; + assert_eq!(map_to_trays(&f, &t).unwrap(), vec![2]); + assert_eq!(map_to_slot(&f, &t, 3).unwrap(), vec![3]); + assert!(map_to_slot(&f, &t, 1).is_err(), "PLA slot for a PETG project"); + assert!(map_to_slot(&f, &t, 4).is_err(), "empty slot"); + let two = [filament(0, "GFG99", "PETG", "161616"), filament(1, "GFG99", "PETG", "2850E0")]; + assert!(map_to_slot(&two, &t, 3).is_err(), "multi-filament project"); + } + const REAL: &str = r##"
diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index 29372e1..4fb2b08 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -652,7 +652,7 @@ async fn dispatch(fleet: &SharedFleet, printer_id: &str, command: PrinterCommand PrinterCommand::ControlPrint { control } => { printer.control_print(control).await.map_err(|err| err.to_string()) } - PrinterCommand::StartPrint { job_id, file_name, .. } => { + PrinterCommand::StartPrint { job_id, file_name, ams_slot, .. } => { let bytes = downloaded.ok_or_else(|| "internal error: start_print reached dispatch with no downloaded file".to_string())?; let byte_count = bytes.len(); info!(%printer_id, %job_id, %file_name, byte_count, "job file downloaded, sending to printer"); @@ -661,10 +661,19 @@ async fn dispatch(fleet: &SharedFleet, printer_id: &str, command: PrinterCommand // all as one error - so the stage here is "printer", not finer. // The printer's own message says which part (mapping, upload of // the file, or start) it was. - printer + if let PrinterHandle::Bambu(bambu) = &mut *printer { + bambu.set_ams_slot_override(ams_slot); + } else if ams_slot.is_some() { + return Err("this printer cannot be told which AMS slot to use".into()); + } + let result = printer .upload_project_file(file_name, bytes) .await - .map_err(|err| format!("printer upload/start failed (download of {byte_count} bytes succeeded): {err}")) + .map_err(|err| format!("printer upload/start failed (download of {byte_count} bytes succeeded): {err}")); + if let PrinterHandle::Bambu(bambu) = &mut *printer { + bambu.set_ams_slot_override(None); + } + result } } } diff --git a/src/uplink/protocol.rs b/src/uplink/protocol.rs index 90bfba7..950b35c 100644 --- a/src/uplink/protocol.rs +++ b/src/uplink/protocol.rs @@ -329,7 +329,14 @@ pub enum PrinterCommand { /// hides it. `file_name` is what the file is called on the printer; /// `.3mf` is treated as a sliced project, anything else as bare gcode. /// `job_id` is only for lining gateway logs up against a cloud job. - StartPrint { job_id: String, file_name: String, url: String }, + StartPrint { + job_id: String, + file_name: String, + url: String, + /// Use this AMS slot (0-based, all units) instead of the project's own pick. + #[serde(default)] + ams_slot: Option, + }, } impl PrinterCommand { diff --git a/tests/uplink_protocol.rs b/tests/uplink_protocol.rs index 823cbcc..16cdc8c 100644 --- a/tests/uplink_protocol.rs +++ b/tests/uplink_protocol.rs @@ -106,7 +106,7 @@ fn a_vendor_specific_control_verb_is_rejected() { fn a_start_print_command_arrives_with_its_fields_intact() { let raw = r#"{"type":"command","command_id":"c","printer_id":"p1p","command":{"action":"start_print","job_id":"5b0e7c1a-0000-4000-8000-000000000000","file_name":"plate_1.3mf","url":"https://objects.example.com/jobs/abc.3mf?X-Amz-Signature=deadbeef&X-Amz-Expires=300"}}"#; match serde_json::from_str::(raw).expect("should parse") { - ServerMessage::Command { command: PrinterCommand::StartPrint { job_id, file_name, url }, .. } => { + ServerMessage::Command { command: PrinterCommand::StartPrint { job_id, file_name, url, .. }, .. } => { assert_eq!(job_id, "5b0e7c1a-0000-4000-8000-000000000000"); assert_eq!(file_name, "plate_1.3mf"); assert_eq!(