"""Spoolman per-filament usage tracking for active prints. Captures AMS tray state and G-code data at print start, then reports per-filament usage to the correct Spoolman spools at print completion. Supports accurate partial usage reporting for failed/cancelled prints. """ import json import logging from sqlalchemy import delete, select from backend.app.core.config import settings as app_settings from backend.app.core.database import async_session from backend.app.services.spoolman import ( SpoolmanClientError, SpoolmanNotFoundError, SpoolmanUnavailableError, get_spoolman_client, init_spoolman_client, ) logger = logging.getLogger(__name__) # Zero UUID used by Bambu printers for empty/unset tray_uuid _ZERO_UUID = "00000000000000000000000000000000" _ZERO_TAG_UID = "0000000000000000" # Highest global tray id that names a real slot. 255 does not: it is # ``PrinterState.tray_now``'s initial value, what an unparseable reading falls # back to, and what the field reads while nothing is loaded. The external spool # reports 254 when it is actually in use, and ``bambu_mqtt`` applies the same # cut-off when it seeds the tray-change log. Treating 255 as a slot would put # ``(255, 1)`` into the "slots this print used" evidence and exclude every real # one -- silently disabling the very fallback this guard protects (#1820). # # Applied to ``tray_now`` only. A 255 in the print's mapping or its tray-change # log was written there by a print and is evidence, however odd; a 255 in # ``tray_now`` is the field at rest, which is the absence of evidence. _MAX_REAL_TRAY_ID = 254 def _is_non_zero_identifier(value: str) -> bool: """Return True when identifier is non-empty and not all zeros.""" if not value: return False return set(value) != {"0"} def _to_fixed_hex(value: int, width: int) -> str: """Mirror frontend toFixedHex(): uppercase, zero-padded, fixed width.""" safe = max(0, int(value)) return format(safe, "X").zfill(width)[-width:] def _hash_serial_to_hex32(serial: str) -> str: """Mirror frontend hashSerialToHex32() exactly (32-bit FNV-1a).""" input_str = (serial or "").strip().upper() hash_value = 0x811C9DC5 for char in input_str: hash_value ^= ord(char) hash_value = (hash_value * 0x01000193) & 0xFFFFFFFF return format(hash_value, "X").zfill(8) def _global_tray_id_to_ams_slot(global_tray_id: int) -> tuple[int, int]: """Convert global tray id to (ams_id, tray_id) tuple for fallback tag generation.""" # External spool slots use IDs 254/255 and map to ams_id=255 tray_id=0/1. if global_tray_id >= 254: return 255, max(0, global_tray_id - 254) # AMS-HT units are addressed by ams_id directly and have a single tray. if global_tray_id >= 128: return global_tray_id, 0 # Standard AMS units: four trays each. return global_tray_id // 4, global_tray_id % 4 def _get_fallback_spool_tag(printer_serial: str, global_tray_id: int) -> str: """Mirror frontend getFallbackSpoolTag(serial, amsId, trayId) exactly.""" if not printer_serial: return "" ams_id, tray_id = _global_tray_id_to_ams_slot(global_tray_id) return get_fallback_spool_tag_for_slot(printer_serial, ams_id, tray_id) def get_fallback_spool_tag_for_slot(printer_serial: str, ams_id: int, tray_id: int) -> str: """Public helper matching frontend getFallbackSpoolTag(serial, amsId, trayId). Used by stale-tag cleanup (#1457) to detect Spoolman spools still holding this slot's deterministic fallback tag in extra.tag. """ if not printer_serial: return "" return f"{_hash_serial_to_hex32(printer_serial)}{_to_fixed_hex(ams_id, 4)}{_to_fixed_hex(tray_id, 4)}" def _resolve_spool_tag(tray_info: dict, printer_serial: str = "", global_tray_id: int | None = None) -> str: """Get the best spool identifier from tray info (prefer tray_uuid over tag_uid). Returns empty string if no usable identifier is found. """ tray_uuid = str(tray_info.get("tray_uuid", "") or "") tag_uid = str(tray_info.get("tag_uid", "") or "") if tray_uuid and tray_uuid != _ZERO_UUID and _is_non_zero_identifier(tray_uuid): return tray_uuid if tag_uid and tag_uid != _ZERO_TAG_UID and _is_non_zero_identifier(tag_uid): return tag_uid if global_tray_id is not None: return _get_fallback_spool_tag(printer_serial, global_tray_id) return "" async def _get_printer_serial(printer_id: int) -> str: """Get printer serial for deterministic fallback tag generation.""" from backend.app.models.printer import Printer from backend.app.services.printer_manager import printer_manager printer_info = printer_manager.get_printer(printer_id) if printer_info and printer_info.serial_number: return printer_info.serial_number async with async_session() as db: result = await db.execute(select(Printer.serial_number).where(Printer.id == printer_id)) serial_number = result.scalar_one_or_none() return serial_number or "" def _resolve_global_tray_id(slot_id: int, slot_to_tray: list | None, ams_trays: dict | None = None) -> int: """Map a 1-based slot_id to a global_tray_id using optional custom mapping. Custom mapping: slot_to_tray[slot_id - 1] is used when >= 0. A value of -1 in the custom mapping means the slicer routed this slot to the external spool. BambuStudio converts virtual tray IDs (254/255) to -1 in the flat ams_mapping array before sending to the printer — see start_print() in bambu_mqtt.py which documents this convention. We mirror it here: when -1 is seen, look up the external spool's actual global_tray_id (254/255) in ams_trays rather than falling through to the position-based default (which would map slot_id=1 to the first AMS tray and credit an unrelated spool — see #1276, regression of #853). Position-based default: uses sorted ams_trays keys so external spools (ID 254/255) naturally follow standard AMS trays, matching the slicer's slot numbering. Final fallback: slot_id - 1 (legacy, works for pure AMS without external spools). """ if slot_to_tray and slot_id <= len(slot_to_tray): mapped_tray = slot_to_tray[slot_id - 1] if mapped_tray >= 0: return mapped_tray if mapped_tray == -1 and ams_trays: # -1 means external spool. 254 = VIRTUAL_TRAY_DEPUTY_ID (main on # single-nozzle, left/deputy on H2D dual-nozzle); 255 = # VIRTUAL_TRAY_MAIN_ID. Prefer 254 when both exist since that's # what single-nozzle printers report via tray_now. for ext_id in (254, 255): if ext_id in ams_trays: return ext_id # Position-based default: sort available tray IDs so external spools (254/255) # come after standard AMS trays, matching the slicer's slot assignment order. if ams_trays: sorted_tray_ids = sorted(ams_trays.keys()) if slot_id <= len(sorted_tray_ids): return sorted_tray_ids[slot_id - 1] return slot_id - 1 def _resolve_slot_to_tray_fallback(printer_id: int, filament_usage: list[dict]) -> tuple[list[int] | None, str]: """Recover a slot-to-tray mapping at completion when print start captured none. ``store_print_data`` can only learn the mapping from two sources: the ``ams_mapping`` Bambuddy intercepts on the printer's local request topic, and a queue item's stored mapping. Neither exists for a print dispatched from Bambu Studio while the printer is cloud-bound — the command travels through Bambu's broker and never appears on the local topic we subscribe to. With ``slot_to_tray`` left NULL, ``_resolve_global_tray_id`` guesses by position: slicer slot 1 to the first loaded tray, slot 2 to the second, and so on. An AMS that isn't loaded in slicer order then charges every slot to the wrong spool, and the archive's filament is rewritten to match, so the print silently changes colour when it finishes (#2768). The printer knows the real answer. Its ``mapping`` field carries the actual slot-to-tray assignment for the running job, and for the models that never publish it (A1, P1S, P2S) the 3MF's per-slot colours can be matched against the loaded trays instead. The built-in inventory writer has consulted both for as long as it has resolved mappings at completion; this gives the Spoolman writer the same two fallbacks at the same moment. Deliberately at completion rather than inside ``store_print_data``: the printer keeps publishing ``mapping`` long after a job ends — it is still in the status payload while the printer sits idle — so reading it at print start risks stamping the *previous* job's mapping onto this one before the printer has pushed the update. At completion the field unambiguously describes the job that just ran. Args: printer_id: Printer whose live state is consulted. filament_usage: The 3MF's per-slot estimates, needed by the colour match. Only the ``slot_id``/``color`` keys are read. Returns: ``(mapping, source)``, or ``(None, "none")`` when neither fallback produced anything and the positional default stands. """ from backend.app.services.printer_manager import printer_manager from backend.app.services.usage_tracker import _decode_mqtt_mapping, _match_slots_by_color state = printer_manager.get_status(printer_id) raw_data = getattr(state, "raw_data", None) if state else None if not raw_data: return None, "none" decoded = _decode_mqtt_mapping(raw_data.get("mapping")) if decoded: return decoded, "mqtt" matched = _match_slots_by_color(filament_usage, raw_data.get("ams")) if matched: return matched, "color_match" return None, "none" def build_ams_tray_lookup(raw_data: dict) -> dict[int, dict]: """Build lookup of global_tray_id -> tray info from printer state. Returns: {0: {"tray_uuid": "...", "tag_uid": "...", "tray_type": "..."}, ...} """ lookup = {} ams_data = raw_data.get("ams", []) for ams_unit in ams_data: ams_id = int(ams_unit.get("id", 0)) for tray in ams_unit.get("tray", []): tray_id = int(tray.get("id", 0)) # AMS-HT units have IDs starting at 128 with a single tray global_tray_id = ams_id if ams_id >= 128 else ams_id * 4 + tray_id lookup[global_tray_id] = { "tray_uuid": tray.get("tray_uuid", ""), "tag_uid": tray.get("tag_uid", ""), "tray_type": tray.get("tray_type", ""), } # External spool(s) (vt_tray is a list, global_tray_id from each entry's "id") for vt in raw_data.get("vt_tray") or []: if vt.get("tray_type"): tray_id = int(vt.get("id", 254)) lookup[tray_id] = { "tray_uuid": vt.get("tray_uuid", ""), "tag_uid": vt.get("tag_uid", ""), "tray_type": vt.get("tray_type", ""), } return lookup def _snapshot_tray_remain(raw_data: dict, skipped_out: list[str] | None = None) -> dict[str, dict]: """Capture per-slot ``remain%`` + ``tray_uuid`` at print start so the completion path can compute a remain-delta when 3MF data doesn't cover the slot (or there's no 3MF at all — #1820). Returns ``{"-": {"remain": int, "tray_uuid": str}}``. Only slots whose ``remain`` is a valid 0..100 int are included; invalid values mean the AMS hasn't read the spool yet and a delta would be meaningless. Mirrors the gate in ``usage_tracker.on_print_start:309``. A rejected slot is appended to *skipped_out* when one is supplied, so the caller can say which slots this print will not be able to charge. That is not hypothetical: an AMS reports a negative ``remain`` on a nearly empty spool, so the gate can drop the one slot that is about to do the printing (#1820). """ snapshot: dict[str, dict] = {} ams_raw = raw_data.get("ams", []) ams_data = ams_raw.get("ams", []) if isinstance(ams_raw, dict) else ams_raw if isinstance(ams_raw, list) else [] for ams_unit in ams_data: if not isinstance(ams_unit, dict): continue ams_id = int(ams_unit.get("id", 0)) for tray in ams_unit.get("tray", []): if not isinstance(tray, dict): continue tray_id = int(tray.get("id", 0)) remain = tray.get("remain", -1) if isinstance(remain, int) and 0 <= remain <= 100: snapshot[f"{ams_id}-{tray_id}"] = { "remain": remain, "tray_uuid": tray.get("tray_uuid", "") or "", } elif skipped_out is not None: skipped_out.append(f"AMS{ams_id}-T{tray_id}(remain={remain})") vt_tray_raw = raw_data.get("vt_tray") or [] if isinstance(vt_tray_raw, dict): vt_tray_raw = [vt_tray_raw] for vt in vt_tray_raw: if not isinstance(vt, dict): continue vt_id = int(vt.get("id", 254)) # 254 → (255, 0), 255 → (255, 1) — matches usage_tracker's encoding. vt_tray_id = vt_id - 254 remain = vt.get("remain", -1) if isinstance(remain, int) and 0 <= remain <= 100: snapshot[f"255-{vt_tray_id}"] = { "remain": remain, "tray_uuid": vt.get("tray_uuid", "") or "", } elif skipped_out is not None: skipped_out.append(f"VT{vt_id}(remain={remain})") return snapshot async def store_print_data( printer_id: int, archive_id: int, file_path: str, db, printer_manager, ams_mapping: list[int] | None = None, plate_id: int | None = None, ): """Store Spoolman tracking data at print start (persisted to database). Per-print tracking is the primary weight-update path for Spoolman, mirroring how the internal Filament Inventory works. The legacy AMS-remain%-based sync is no longer used as a weight writer (#1119), so this runs whenever Spoolman is enabled regardless of the deprecated `spoolman_disable_weight_sync` flag. ``plate_id``, when set, scopes the 3MF filament extract to a single plate so queue / direct-Print dispatch of plate N of a multi-plate file doesn't attribute every plate's filament to the printed spool (#1697). When unset, the queue item's plate_id (if any) is used; otherwise the whole-file sum is extracted, which is correct for direct prints that target the first/only plate of a single-plate file. """ from backend.app.api.routes.settings import get_setting from backend.app.models.active_print_spoolman import ActivePrintSpoolman from backend.app.models.print_queue import PrintQueueItem from backend.app.utils.threemf_tools import ( extract_filament_properties_from_3mf, extract_filament_usage_from_3mf, extract_layer_filament_usage_from_3mf, ) # Check if Spoolman is enabled spoolman_enabled = await get_setting(db, "spoolman_enabled") if not spoolman_enabled or spoolman_enabled.lower() != "true": return # Get current AMS tray state up front — needed both for the 3MF path's # ams_trays field and for the remain%-delta snapshot (#1820 fallback for # no-3MF "Untitled" prints, mirroring usage_tracker.on_print_start). state = printer_manager.get_status(printer_id) ams_trays: dict[int, dict] = {} tray_remain_start: dict[str, dict] = {} if state and state.raw_data: ams_trays = build_ams_tray_lookup(state.raw_data) skipped_slots: list[str] = [] tray_remain_start = _snapshot_tray_remain(state.raw_data, skipped_slots) if skipped_slots: # Matches what usage_tracker.on_print_start reports for the # internal inventory, so both backends name the slots that this # print will not be able to charge at AMS granularity. logger.info( "[SPOOLMAN] Printer %s: slots with no usable remain%% at print start: %s", printer_id, ", ".join(skipped_slots), ) # Try to read per-slot filament estimates from the 3MF. Two paths can # leave ``filament_usage`` empty: (1) fallback archive (no .gcode.3mf # was downloadable from the printer — "Untitled" prints, see #1820), # (2) 3MF present but slice_info missing per-filament estimates. # Both fall through to the remain%-delta path at completion. filament_usage: list | None = None layer_usage_json: dict | None = None filament_properties: dict | None = None full_path = ( app_settings.base_dir / file_path ) # SEC-PATH-OK: file_path is archive.file_path / library_file.file_path — DB-stored, internally generated threemf_available = bool(file_path) and full_path.exists() queue_item = None if threemf_available: # Resolve the queue item once — used both for the plate-scoped 3MF parsing # fallback (#1697: multi-plate file dispatched for one plate must only count # that plate's filament) and for the ams_mapping fallback below. queue_result = await db.execute( select(PrintQueueItem) .where(PrintQueueItem.archive_id == archive_id) .where(PrintQueueItem.status == "printing") ) queue_item = queue_result.scalar_one_or_none() # Caller-supplied plate_id wins (direct-Print path); fall back to the queue # item's plate_id (queue dispatch path). effective_plate_id = ( plate_id if plate_id is not None else (queue_item.plate_id if queue_item is not None else None) ) filament_usage = extract_filament_usage_from_3mf(full_path, effective_plate_id) or None layer_usage = extract_layer_filament_usage_from_3mf(full_path, effective_plate_id) if layer_usage: # Convert int keys to string for JSON serialization layer_usage_json = {str(k): v for k, v in layer_usage.items()} logger.debug("[SPOOLMAN] Parsed %s layers from G-code", len(layer_usage)) filament_properties = extract_filament_properties_from_3mf(full_path) else: # No 3MF on disk — common for "Untitled" prints whose .gcode.3mf # was never on the printer's FTP. Logged at debug since the # fallback path below picks up the slack when remain% is available. logger.debug("[SPOOLMAN] 3MF file not available: %s", full_path) # If neither path has anything useful, there's nothing to track. if not filament_usage and not tray_remain_start: if threemf_available: logger.debug("[SPOOLMAN] No filament usage data in 3MF for archive %s", archive_id) return # Prefer the explicit mapping captured from the print command, then fall back # to any queue mapping stored for scheduled/reprint jobs. slot_to_tray = ams_mapping if ams_mapping is not None else None mapping_source = "print_cmd" if slot_to_tray else None if not slot_to_tray and queue_item and queue_item.ams_mapping: try: slot_to_tray = json.loads(queue_item.ams_mapping) mapping_source = "queue" except json.JSONDecodeError: pass # Ignore malformed AMS mapping; fall back to default slot assignment # Delete any existing row for this printer/archive (shouldn't exist, but just in case) await db.execute( delete(ActivePrintSpoolman) .where(ActivePrintSpoolman.printer_id == printer_id) .where(ActivePrintSpoolman.archive_id == archive_id) ) # Insert new tracking data. ``filament_usage`` may be None for the # no-3MF case; report_usage falls back to ``tray_remain_start``. tracking = ActivePrintSpoolman( printer_id=printer_id, archive_id=archive_id, filament_usage=filament_usage, ams_trays=ams_trays, slot_to_tray=slot_to_tray, layer_usage=layer_usage_json, filament_properties=filament_properties, tray_remain_start=tray_remain_start or None, # Which slot the printer was drawing from when this print began. For a # print with no ams_mapping -- one started from the printer's own # screen, which is the case this whole fallback exists for -- it is the # only evidence of which slot the print used (#1820). tray_now_at_start=getattr(state, "tray_now", None) if state else None, ) db.add(tracking) await db.commit() logger.info( "[SPOOLMAN] Stored tracking data for print: printer=%s, archive=%s (3mf=%s, remain_snapshot=%d slot(s))", printer_id, archive_id, "yes" if filament_usage else "no", len(tray_remain_start), ) logger.debug("[SPOOLMAN] Filament usage: %s", filament_usage) logger.debug("[SPOOLMAN] AMS trays: %s", list(ams_trays.keys())) # Logged at info even when there is no mapping: "source: none" here is the # signal that completion will have to fall back, which is the single most # useful line in the log when a print is charged to the wrong spool (#2768). logger.info( "[SPOOLMAN] Print start: archive %s slot_to_tray=%s (source: %s)", archive_id, slot_to_tray, mapping_source or "none", ) if layer_usage_json: logger.debug("[SPOOLMAN] Layer usage data available for partial tracking") async def cleanup_tracking( printer_id: int, archive_id: int, db, last_layer_num: int | None = None, last_progress: int | None = None, ): """Report partial usage and clean up Spoolman tracking data for failed/aborted prints.""" from backend.app.models.active_print_spoolman import ActivePrintSpoolman # Get tracking data first (needed for partial usage reporting) result = await db.execute( select(ActivePrintSpoolman) .where(ActivePrintSpoolman.printer_id == printer_id) .where(ActivePrintSpoolman.archive_id == archive_id) ) tracking = result.scalar_one_or_none() if not tracking: logger.debug("[SPOOLMAN] No tracking data to clean up for printer=%s, archive=%s", printer_id, archive_id) return # Try to report partial usage before cleanup try: await _report_partial_usage( printer_id, tracking, last_layer_num=last_layer_num, last_progress=last_progress, ) except Exception as e: logger.warning("[SPOOLMAN] Partial usage report failed: %s", e) # Delete tracking data await db.execute( delete(ActivePrintSpoolman) .where(ActivePrintSpoolman.printer_id == printer_id) .where(ActivePrintSpoolman.archive_id == archive_id) ) await db.commit() logger.debug("[SPOOLMAN] Cleaned up tracking data for printer=%s, archive=%s", printer_id, archive_id) async def _get_spoolman_client_with_fallback(): """Get Spoolman client, initializing from settings if needed. Returns (client, is_healthy) tuple. Client may be None. """ client = await get_spoolman_client() if not client: async with async_session() as db: from backend.app.api.routes.settings import get_setting spoolman_url = await get_setting(db, "spoolman_url") if spoolman_url: try: client = await init_spoolman_client(spoolman_url) except ValueError as exc: logger.warning("Spoolman URL %r rejected by SSRF guard: %s", spoolman_url, exc) return None if not client: return None if not await client.health_check(): logger.warning("Spoolman health check failed; skipping usage reporting") return None return client async def _resolve_spool_id_via_slot_assignment(printer_id: int, ams_id: int, tray_id: int) -> int | None: """Look up the Spoolman spool ID locally bound to (printer, ams, tray). Fallback path for #1459: when a tag-less spool was assigned via the Bambuddy UI, the user's deterministic fallback tag is intentionally NOT written to Spoolman's extra.tag (kept clean per #1457), so find_spool_by_tag misses. The local spoolman_slot_assignments table is the authoritative binding for those spools. """ from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment async with async_session() as db: result = await db.execute( select(SpoolmanSlotAssignment.spoolman_spool_id).where( SpoolmanSlotAssignment.printer_id == printer_id, SpoolmanSlotAssignment.ams_id == ams_id, SpoolmanSlotAssignment.tray_id == tray_id, ) ) return result.scalar_one_or_none() async def _report_spool_usage_for_slots( client, filament_usage_items: list[tuple[int, float]], ams_trays: dict[int, dict], slot_to_tray: list | None, method_label: str, printer_serial: str = "", printer_id: int | None = None, slot_colors_out: dict[int, str] | None = None, slot_materials_out: dict[int, str] | None = None, ) -> int: """Report usage to Spoolman for a list of (slot_id, grams) pairs. Resolution order per slot: (1) Spoolman extra.tag match against the tray's RFID or deterministic fallback tag, (2) #1459 fallback — local spoolman_slot_assignments table keyed by (printer_id, ams_id, tray_id). Without (2), tag-less spools assigned via the Bambuddy UI never get their weight decremented because their extra.tag is empty on the Spoolman side. When ``slot_colors_out`` is provided it is populated with ``{slot_id: color_hex}`` for every resolved spool — used by :func:`report_usage` to stamp the archive's filament colour from the Spoolman spool rather than the slicer's 3MF value (#1494). Returns number of spools successfully updated. """ spools_updated = 0 for slot_id, grams_used in filament_usage_items: if grams_used <= 0: continue global_tray_id = _resolve_global_tray_id(slot_id, slot_to_tray, ams_trays) tray_info = ams_trays.get(global_tray_id) if not tray_info: logger.debug("[SPOOLMAN] Slot %s: no tray at global_tray_id %s", slot_id, global_tray_id) continue is_external = global_tray_id >= 254 tray_type = tray_info.get("tray_type", "") logger.debug( "[SPOOLMAN] Slot %s resolved to global_tray_id %s (tray_type=%s, external=%s)", slot_id, global_tray_id, tray_type or "unknown", is_external, ) spool_id_to_use: int | None = None resolution_path = "" # color_hex + material of the resolved spool's filament, for the #1494 # archive colour rewrite and the #2563 type rewrite. The tag path # already has the full spool object; the slot-assignment path only # yields an id and is fetched below. spool_color_hex: str | None = None spool_material: str | None = None spool_tag = _resolve_spool_tag(tray_info, printer_serial, global_tray_id) if spool_tag: spool = await client.find_spool_by_tag(spool_tag) if spool: spool_id_to_use = spool["id"] resolution_path = "tag" spool_color_hex = (spool.get("filament") or {}).get("color_hex") spool_material = (spool.get("filament") or {}).get("material") if spool_id_to_use is None and printer_id is not None: ams_id, tray_id = _global_tray_id_to_ams_slot(global_tray_id) spool_id_to_use = await _resolve_spool_id_via_slot_assignment(printer_id, ams_id, tray_id) if spool_id_to_use is not None: resolution_path = "slot-assignment" if spool_id_to_use is None: logger.debug( "[SPOOLMAN] Slot %s: no spool resolved (tag=%s, no slot-assignment)", slot_id, spool_tag[:16] if spool_tag else "none", ) continue # Record the spool's filament colour + material for the archive # rewrites (#1494, #2563). The slot-assignment path resolved only an # id, so fetch the spool once for whichever value is still missing. # Strictly best-effort: a fetch failure must never abort the weight # reporting for the remaining slots, so the catch is broad. if slot_colors_out is not None or slot_materials_out is not None: need_color = slot_colors_out is not None and spool_color_hex is None need_material = slot_materials_out is not None and spool_material is None if need_color or need_material: try: _fil = (await client.get_spool(spool_id_to_use)).get("filament") or {} if need_color: spool_color_hex = _fil.get("color_hex") if need_material: spool_material = _fil.get("material") except Exception as exc: # noqa: BLE001 — colour/material are non-critical logger.debug("[SPOOLMAN] Slot %s: could not fetch spool filament: %s", slot_id, exc) if slot_colors_out is not None and spool_color_hex: slot_colors_out[slot_id] = spool_color_hex if slot_materials_out is not None and spool_material: slot_materials_out[slot_id] = spool_material try: await client.use_spool(spool_id_to_use, grams_used) logger.info( "[SPOOLMAN] %s: slot %s: %sg -> spool %s (via %s)", method_label, slot_id, grams_used, spool_id_to_use, resolution_path, ) spools_updated += 1 except (SpoolmanNotFoundError, SpoolmanClientError, SpoolmanUnavailableError) as exc: logger.warning("[SPOOLMAN] Failed to record usage for spool %s: %s", spool_id_to_use, exc) return spools_updated async def _report_spool_usage_split_by_tray_changes( client, filament_usage: list[dict], tray_changes: list[tuple[int, int]], ams_trays: dict[int, dict], layer_usage: dict[int, dict[int, float]] | None, filament_properties: dict | None, total_layers: int, last_layer_num: int, method_label: str, printer_serial: str, printer_id: int, slot_colors_out: dict[int, str] | None = None, slot_materials_out: dict[int, str] | None = None, ) -> tuple[int, set[int]]: """Split each slot's grams across ``tray_changes`` and charge per-segment. Mirrors ``usage_tracker`` Path 1's tray-switch branch so Spoolman and the internal Spool inventory attribute mid-print AMS-backup switches identically (#1793 — reporter's origin spool was over-charged the whole print because this path didn't exist). ``compute_tray_split_grams`` holds the shared segment-math; this function wraps the per-segment spool resolution + ``use_spool`` sink for the Spoolman side. Returns ``(spools_updated, handled_global_tray_ids)`` — the caller passes ``handled_global_tray_ids`` into the remain-delta fallback so a tray attributed here is not double-charged there. """ from backend.app.utils.tray_split import compute_tray_split_grams spools_updated = 0 handled_global_tray_ids: set[int] = set() for usage in filament_usage: slot_id = usage.get("slot_id", 0) total_weight = usage.get("used_g", 0) if total_weight <= 0 or slot_id <= 0: continue props = (filament_properties or {}).get(str(slot_id)) or (filament_properties or {}).get(slot_id) or {} segments = compute_tray_split_grams( tray_changes=tray_changes, total_weight=float(total_weight), slot_id=slot_id, layer_usage=layer_usage, density=float(props.get("density", 1.24)), diameter=float(props.get("diameter", 1.75)), total_layers=total_layers, last_layer_num=last_layer_num, ) for seg_idx, tray_global, segment_grams in segments: if segment_grams <= 0: continue # Mark this tray as handled BEFORE the resolution attempt so # remain-delta doesn't double-charge it, even if we fail to # find a spool below. Matches usage_tracker behaviour: the # tray was physically fed from during this print, whether or # not Spoolman happens to have a matching row. handled_global_tray_ids.add(tray_global) tray_info = ams_trays.get(tray_global) or {} spool_id_to_use: int | None = None resolution_path = "" spool_color_hex: str | None = None spool_material: str | None = None spool_tag = _resolve_spool_tag(tray_info, printer_serial, tray_global) if tray_info else "" if spool_tag: spool = await client.find_spool_by_tag(spool_tag) if spool: spool_id_to_use = spool["id"] resolution_path = "tag" spool_color_hex = (spool.get("filament") or {}).get("color_hex") spool_material = (spool.get("filament") or {}).get("material") if spool_id_to_use is None: seg_ams_id, seg_tray_id = _global_tray_id_to_ams_slot(tray_global) spool_id_to_use = await _resolve_spool_id_via_slot_assignment(printer_id, seg_ams_id, seg_tray_id) if spool_id_to_use is not None: resolution_path = "slot-assignment" if spool_id_to_use is None: logger.info( "[SPOOLMAN] Split slot %s seg %s tray=%d: no spool resolved — %.2fg lost from split accounting", slot_id, seg_idx, tray_global, segment_grams, ) continue # Colour (#1494) + material (#2563) rewrite — first segment for a # slot wins. The UI displays a single colour/type per slot, so # later segments on the same slot don't overwrite (a backup swap # can differ but the archive card stays consistent with the origin). need_color = slot_colors_out is not None and slot_id not in slot_colors_out and spool_color_hex is None need_material = ( slot_materials_out is not None and slot_id not in slot_materials_out and spool_material is None ) if need_color or need_material: try: _fil = (await client.get_spool(spool_id_to_use)).get("filament") or {} if need_color: spool_color_hex = _fil.get("color_hex") if need_material: spool_material = _fil.get("material") except Exception as exc: # noqa: BLE001 — colour/material are non-critical logger.debug("[SPOOLMAN] Split slot %s: could not fetch spool filament: %s", slot_id, exc) if slot_colors_out is not None and slot_id not in slot_colors_out and spool_color_hex: slot_colors_out[slot_id] = spool_color_hex if slot_materials_out is not None and slot_id not in slot_materials_out and spool_material: slot_materials_out[slot_id] = spool_material try: await client.use_spool(spool_id_to_use, round(segment_grams, 2)) logger.info( "[SPOOLMAN] %s: slot %s seg %s tray=%d: %.2fg -> spool %s (via %s)", method_label, slot_id, seg_idx, tray_global, segment_grams, spool_id_to_use, resolution_path, ) spools_updated += 1 except (SpoolmanNotFoundError, SpoolmanClientError, SpoolmanUnavailableError) as exc: logger.warning( "[SPOOLMAN] Split slot %s seg %s: failed to record usage for spool %s: %s", slot_id, seg_idx, spool_id_to_use, exc, ) return spools_updated, handled_global_tray_ids async def _report_partial_usage( printer_id: int, tracking, last_layer_num: int | None = None, last_progress: int | None = None, ): """Report partial filament usage based on actual G-code layer data. Uses per-layer cumulative extrusion from G-code parsing for accurate multi-material tracking. Falls back to linear interpolation if G-code data is unavailable. """ from backend.app.services.printer_manager import printer_manager from backend.app.utils.threemf_tools import get_cumulative_usage_at_layer, mm_to_grams async with async_session() as db: from backend.app.api.routes.settings import get_setting # Check if partial usage reporting is enabled (default: true) report_partial = await get_setting(db, "spoolman_report_partial_usage") if report_partial and report_partial.lower() == "false": logger.debug("[SPOOLMAN] Partial usage reporting disabled by setting") return # Check if Spoolman is enabled spoolman_enabled = await get_setting(db, "spoolman_enabled") if not spoolman_enabled or spoolman_enabled.lower() != "true": return # Get current printer state for layer progress. # On failed/aborted prints the firmware may already reset to IDLE with layer=0, # so we fall back to completion-time hints captured from MQTT. state = printer_manager.get_status(printer_id) current_layer = state.layer_num if state else None total_layers = state.total_layers if state else None if (not current_layer or current_layer <= 0) and last_layer_num and last_layer_num > 0: current_layer = last_layer_num logger.debug("[SPOOLMAN] Using captured last_layer_num=%s for partial usage", current_layer) progress_ratio_from_event = None if last_progress is not None: try: progress_ratio_from_event = min(max(float(last_progress), 0.0), 100.0) / 100.0 except (TypeError, ValueError): progress_ratio_from_event = None if (not current_layer or current_layer <= 0) and progress_ratio_from_event and total_layers and total_layers > 0: current_layer = max(1, int(round(total_layers * progress_ratio_from_event))) logger.debug( "[SPOOLMAN] Estimated layer from last_progress=%s%% and total_layers=%s -> %s", last_progress, total_layers, current_layer, ) if not current_layer or current_layer <= 0: logger.debug( "[SPOOLMAN] No progress to report (layer 0/unknown, last_layer_num=%s, last_progress=%s)", last_layer_num, last_progress, ) return logger.info("[SPOOLMAN] Reporting partial usage at layer %s/%s", current_layer, total_layers or "?") # Get tracking data layer_usage = tracking.layer_usage filament_properties = tracking.filament_properties or {} filament_usage = tracking.filament_usage or [] ams_trays = {int(k): v for k, v in (tracking.ams_trays or {}).items()} slot_to_tray = tracking.slot_to_tray tray_remain_start = tracking.tray_remain_start or {} printer_serial = await _get_printer_serial(printer_id) client = await _get_spoolman_client_with_fallback() if not client: logger.warning("[SPOOLMAN] Not reachable for partial usage reporting") return # No-3MF aborted print (#1820 mirror of the completion path): nothing in # filament_usage or layer_usage to base partial estimates on, but the # remain%-delta snapshot we captured at start still describes consumption # up to the abort moment. Write it the same way report_usage's fallback # does, then return — there's no 3MF-derived partial to layer on top. # ``state`` was already fetched at the top of the function for current_layer. if not filament_usage and not layer_usage and tray_remain_start: current_lookup = _snapshot_tray_remain(state.raw_data) if state and state.raw_data else {} await _report_remain_delta_for_slots( client, printer_id=printer_id, tray_remain_start=tray_remain_start, current_lookup=current_lookup, handled_global_tray_ids=set(), archive_id=getattr(tracking, "archive_id", -1), print_used_keys=_print_used_tray_keys(slot_to_tray, getattr(tracking, "tray_now_at_start", None), state), ) return # Same recovery the completion path does, for the same reason: a print # dispatched from Studio over the cloud left print start with no mapping to # store, and both paths below feed ``slot_to_tray`` to # ``_resolve_global_tray_id`` (#2768). An aborted print charges the wrong # spool just as readily as a finished one. if not slot_to_tray: slot_to_tray, _partial_mapping_source = _resolve_slot_to_tray_fallback(printer_id, filament_usage) logger.info( "[SPOOLMAN] Partial usage: slot_to_tray=%s (source: %s)", slot_to_tray, _partial_mapping_source, ) # Try to use accurate G-code parsed data if layer_usage: layer_usage_int = { int(layer): {int(fid): mm for fid, mm in filaments.items()} for layer, filaments in layer_usage.items() } usage_mm = get_cumulative_usage_at_layer(layer_usage_int, current_layer) if usage_mm: logger.info("[SPOOLMAN] Using G-code parsed data for layer %s", current_layer) # Build (slot_id, grams) list using Spoolman densities with 3MF fallback usage_items = [] for filament_id, mm_used in usage_mm.items(): slot_id = filament_id + 1 # filament_id is 0-based, slot_id is 1-based # Get density from Spoolman (most accurate), fall back to 3MF, then PLA default global_tray_id = _resolve_global_tray_id(slot_id, slot_to_tray, ams_trays) tray_info = ams_trays.get(global_tray_id) density = None diameter = 1.75 if tray_info: spool_tag = _resolve_spool_tag(tray_info, printer_serial, global_tray_id) if spool_tag: spool = await client.find_spool_by_tag(spool_tag) if spool: filament_data = spool.get("filament", {}) density = filament_data.get("density") diameter = filament_data.get("diameter", 1.75) if not density: props = filament_properties.get(str(slot_id), filament_properties.get(slot_id, {})) density = props.get("density", 1.24) logger.debug("[SPOOLMAN] Using fallback density %s for slot %s", density, slot_id) grams_used = round(mm_to_grams(mm_used, diameter, density), 2) usage_items.append((slot_id, grams_used)) spools_updated = await _report_spool_usage_for_slots( client, usage_items, ams_trays, slot_to_tray, "Partial (G-code)", printer_serial, printer_id=printer_id, ) if spools_updated > 0: logger.info("[SPOOLMAN] Reported partial usage to %s spool(s) using G-code data", spools_updated) return # Fallback: linear interpolation (if no G-code data available) progress_ratio = None if total_layers and total_layers > 0: progress_ratio = min(current_layer / total_layers, 1.0) elif progress_ratio_from_event is not None: progress_ratio = progress_ratio_from_event if progress_ratio is None: logger.debug( "[SPOOLMAN] Cannot use linear fallback: total_layers=%s, last_progress=%s", total_layers, last_progress, ) return logger.info("[SPOOLMAN] Falling back to linear interpolation (%s)", progress_ratio) usage_items = [] for usage in filament_usage: slot_id = usage.get("slot_id", 0) total_used_g = usage.get("used_g", 0) if total_used_g > 0: partial_used_g = round(total_used_g * progress_ratio, 2) usage_items.append((slot_id, partial_used_g)) spools_updated = await _report_spool_usage_for_slots( client, usage_items, ams_trays, slot_to_tray, "Partial (linear)", printer_serial, printer_id=printer_id, ) if spools_updated > 0: logger.info("[SPOOLMAN] Reported partial usage to %s spool(s) using linear interpolation", spools_updated) async def report_usage(printer_id: int, archive_id: int): """Report filament usage to Spoolman after print completion. Two writers, mirroring the internal-inventory split in usage_tracker: 1. **3MF path (primary)** — per-filament slice estimates captured at print start drive a precise per-slot ``use_spool`` call. 2. **AMS remain%-delta (fallback)** — for slots the 3MF path didn't handle (including the no-3MF "Untitled" case from #1820): compute ``start_remain - current_remain``, multiply by the resolved Spoolman filament's reference weight, and write the delta. Mirrors ``usage_tracker.on_print_complete`` Path 2 (line 517). """ async with async_session() as db: from backend.app.api.routes.settings import get_setting from backend.app.models.active_print_spoolman import ActivePrintSpoolman # Get tracking data stored at print start result = await db.execute( select(ActivePrintSpoolman) .where(ActivePrintSpoolman.printer_id == printer_id) .where(ActivePrintSpoolman.archive_id == archive_id) ) tracking = result.scalar_one_or_none() if not tracking: logger.info("[SPOOLMAN] No tracking data for print (printer=%s, archive=%s)", printer_id, archive_id) return filament_usage = tracking.filament_usage or [] ams_trays = {int(k): v for k, v in (tracking.ams_trays or {}).items()} slot_to_tray = tracking.slot_to_tray tray_remain_start = tracking.tray_remain_start or {} # ``layer_usage`` and ``filament_properties`` were added later than # the base tracking fields; use ``getattr`` so tests that stub # ``tracking`` as a lightweight SimpleNamespace stay valid, and # historic ORM rows loaded without these columns can't AttributeError # on read. layer_usage_raw = getattr(tracking, "layer_usage", None) or {} filament_properties = getattr(tracking, "filament_properties", None) or {} tray_now_at_start = getattr(tracking, "tray_now_at_start", None) printer_serial = await _get_printer_serial(printer_id) # Delete tracking row (we're done with it) await db.delete(tracking) await db.commit() if not filament_usage and not tray_remain_start: logger.debug("[SPOOLMAN] No usage data or remain-snapshot for archive %s", archive_id) return # Check if Spoolman is enabled spoolman_enabled = await get_setting(db, "spoolman_enabled") if not spoolman_enabled or spoolman_enabled.lower() != "true": return client = await _get_spoolman_client_with_fallback() if not client: logger.warning("[SPOOLMAN] Not reachable for usage reporting") return # Consult the live printer state for the tray-change log written by # ``bambu_mqtt.py`` on every mid-print ``tray_now`` change (#957). # When there's more than one entry, the print traversed >1 AMS tray # and the split path attributes each segment to the tray that was # loaded at the time — matches the internal Spool inventory writer # in ``usage_tracker.py``. Without this, an AMS-backup runout switch # charges the whole slot to the origin spool and pushes it past # ``initial_weight`` (#1793). # # Split only for SINGLE-slot prints — same gate as # ``usage_tracker.py:1002``. Multi-slot (multi-colour) prints # naturally cycle trays for every colour change, so splitting each # slot's grams across ALL tray_change_log entries would attribute # slot 1's grams to the segments where slot 2's tray was loaded and # vice versa. Multi-slot prints fall through to the existing # single-tray path (which uses the stable ``slot_to_tray`` mapping). nonzero_slots = [u for u in filament_usage if u.get("used_g", 0) > 0] tray_changes: list[tuple[int, int]] = [] _state = None if len(nonzero_slots) == 1: from backend.app.services.printer_manager import printer_manager as _pm _state = _pm.get_status(printer_id) if _state is not None: tray_changes = list(getattr(_state, "tray_change_log", []) or []) _total_layers = int(getattr(_state, "total_layers", 0) or 0) if _state else 0 _current_layer = int(getattr(_state, "layer_num", 0) or 0) if _state else 0 # For the linear-fallback denominator when total_layers is 0 (P1S # firmware resets it at print end). At completion the current layer # is the print's last valid layer. _layer_denom_hint = _total_layers or _current_layer # Recover the mapping when print start had nothing to store — the # cloud-dispatched Studio print of #2768. Only the 3MF path consumes # ``slot_to_tray``; the remain-delta path below resolves spools from the # AMS slot directly, so there is nothing to recover for it. mapping_source = "stored" if slot_to_tray else "none" if filament_usage and not slot_to_tray: slot_to_tray, mapping_source = _resolve_slot_to_tray_fallback(printer_id, filament_usage) logger.info( "[SPOOLMAN] Archive %s: slot_to_tray=%s (source: %s)", archive_id, slot_to_tray, mapping_source, ) slot_colors: dict[int, str] = {} slot_materials: dict[int, str] = {} handled_global_tray_ids: set[int] = set() spools_updated = 0 # --- Path 1: 3MF per-slot estimates ----------------------------- if filament_usage: if len(tray_changes) > 1: # Tray-split path — attribute per-segment to the tray that # was loaded at that time. logger.info( "[SPOOLMAN] Reporting per-filament usage for archive %s with tray-split " "(tray_change_log=%s, denom_layers=%d)", archive_id, tray_changes, _layer_denom_hint, ) # ``tracking.layer_usage`` was serialized to JSON so int keys # come back as strings. Restore them for the split math. layer_usage = None if layer_usage_raw: try: layer_usage = { int(layer): {int(fid): mm for fid, mm in filaments.items()} for layer, filaments in layer_usage_raw.items() } except (TypeError, ValueError, AttributeError): # AttributeError catches ``inner.items()`` when the # inner value isn't dict-shaped (corrupt JSON row). # Missing gcode falls through to the linear-ratio # branch inside ``compute_tray_split_grams`` — still # gives a correct split, just less precise. layer_usage = None split_updated, split_handled = await _report_spool_usage_split_by_tray_changes( client, filament_usage, tray_changes, ams_trays, layer_usage, filament_properties, _total_layers, _layer_denom_hint, f"Archive {archive_id}", printer_serial, printer_id=printer_id, slot_colors_out=slot_colors, slot_materials_out=slot_materials, ) spools_updated += split_updated handled_global_tray_ids |= split_handled else: logger.info("[SPOOLMAN] Reporting per-filament usage for archive %s", archive_id) usage_items = [(u.get("slot_id", 0), u.get("used_g", 0)) for u in filament_usage] spools_updated = await _report_spool_usage_for_slots( client, usage_items, ams_trays, slot_to_tray, f"Archive {archive_id}", printer_serial, printer_id=printer_id, slot_colors_out=slot_colors, slot_materials_out=slot_materials, ) # Track which physical slots the 3MF path already covered so # Path 2 doesn't double-charge them. for u in filament_usage: slot_id = u.get("slot_id", 0) handled_global_tray_ids.add(_resolve_global_tray_id(slot_id, slot_to_tray, ams_trays)) # --- Path 2: AMS remain%-delta for slots 3MF didn't cover ------- # Triggered for no-3MF "Untitled" prints (#1820) AND for partial # 3MF coverage (slots whose filament_id wasn't in slice_info). if tray_remain_start: from backend.app.services.printer_manager import printer_manager current = printer_manager.get_status(printer_id) current_lookup = _snapshot_tray_remain(current.raw_data) if current and current.raw_data else {} fallback_updates = await _report_remain_delta_for_slots( client, printer_id=printer_id, tray_remain_start=tray_remain_start, current_lookup=current_lookup, handled_global_tray_ids=handled_global_tray_ids, archive_id=archive_id, print_used_keys=_print_used_tray_keys(slot_to_tray, tray_now_at_start, current), slot_colors_out=slot_colors, slot_materials_out=slot_materials, ) spools_updated += fallback_updates if spools_updated == 0: logger.info("[SPOOLMAN] Archive %s: no spools updated", archive_id) else: logger.info("[SPOOLMAN] Archive %s: updated %s spool(s)", archive_id, spools_updated) # Stamp the archive's filament colour from the matched Spoolman spools # so it reflects the curated inventory colour, not the slicer's 3MF # value (#1494) — mirrors the built-in inventory path in usage_tracker. await _apply_spool_colors_to_archive(db, archive_id, filament_usage, slot_colors) # Same for the material: a slot mapped to a differently-typed spool than # it was sliced for otherwise records the sliced type (#2563). await _apply_spool_types_to_archive(db, archive_id, filament_usage, slot_materials) def _print_used_tray_keys( slot_to_tray: list | None, tray_now_at_start: int | None, state, ) -> set[tuple[int, int]]: """Which AMS slots this print actually drew from, as far as we can tell. Mirrors the guard the internal tracker has carried since #1269. Without it, swapping a spool in a slot the print never touched drops that slot's ``remain%``, and the remain-delta path reads the drop as consumption and charges it to whoever the slot is assigned to. That is a phantom write to an uninvolved spool, and it is likeliest on exactly the prints this fallback serves -- ones with no 3MF, where nothing else limits which slots are considered. Three sources, matching the internal tracker's: - the print's ``ams_mapping``, stored here as ``slot_to_tray``; - every tray the printer switched to mid-print; - the tray it was drawing from at the start. An empty result means no evidence, not "no slots" -- callers must then consider every slot, as before, or a printer that reports none of the three would silently stop being tracked at all. Takes the two stored values rather than the tracking row: the caller deletes that row before it gets this far, and everything read off it is read into locals beforehand. """ keys: set[tuple[int, int]] = set() for global_tray_id in list(slot_to_tray or []): if isinstance(global_tray_id, int) and global_tray_id >= 0: keys.add(_global_tray_id_to_ams_slot(global_tray_id)) for change in getattr(state, "tray_change_log", None) or []: if isinstance(change, (tuple, list)) and change: global_tray_id = change[0] if isinstance(global_tray_id, int) and global_tray_id >= 0: keys.add(_global_tray_id_to_ams_slot(global_tray_id)) if isinstance(tray_now_at_start, int) and 0 <= tray_now_at_start <= _MAX_REAL_TRAY_ID: keys.add(_global_tray_id_to_ams_slot(tray_now_at_start)) return keys async def _report_remain_delta_for_slots( client, *, printer_id: int, tray_remain_start: dict[str, dict], current_lookup: dict[str, dict], handled_global_tray_ids: set[int], archive_id: int, print_used_keys: set[tuple[int, int]] | None = None, slot_colors_out: dict[int, str] | None = None, slot_materials_out: dict[int, str] | None = None, ) -> int: """AMS remain%-delta path: write ``(start - current) * filament.weight`` grams to Spoolman for slots the 3MF path didn't cover. Mirrors ``usage_tracker.on_print_complete`` Path 2: per-slot, gated on a valid current ``remain%``, skipped on spool swap (``tray_uuid`` changed), using the resolved spool's filament reference weight rather than MQTT's unreliable ``tray_weight`` (which is the failure mode #1119 documented). """ spools_updated = 0 not_in_print: list[str] = [] for slot_key, start in tray_remain_start.items(): try: ams_id_str, tray_id_str = slot_key.split("-", 1) ams_id, tray_id = int(ams_id_str), int(tray_id_str) except (ValueError, AttributeError): continue # Skip slots already handled by the 3MF path. Encoding mirrors # build_ams_tray_lookup: VT trays land at 254/255, AMS-HT keeps # its native id (>=128), regular AMS slots are ams_id*4+tray_id. if ams_id == 255: global_tray_id = 254 + tray_id elif ams_id >= 128: global_tray_id = ams_id else: global_tray_id = ams_id * 4 + tray_id if global_tray_id in handled_global_tray_ids: continue # Slots the print never touched (#1269's guard, see _print_used_tray_keys). # Only enforced when there is evidence of which slots it did use. # Collected rather than logged per slot: on a four-AMS farm a # single-colour print leaves fifteen of these, and they are the # expected case, unlike the "consumed but charged nothing" lines below. if print_used_keys and (ams_id, tray_id) not in print_used_keys: not_in_print.append(f"AMS{ams_id}-T{tray_id}") continue current = current_lookup.get(slot_key) if not current: # Reported at info, like the internal tracker's equivalent: on a # near-empty spool the AMS reports a negative remain%, which the # snapshot gate rejects, and the slot that was actually printing # disappears from this path entirely (#1820). logger.info( "[SPOOLMAN] AMS%d-T%d: no valid remain%% at completion, nothing charged for this slot", ams_id, tray_id ) continue # Spool swap mid-print — tray_uuid changed. We don't know how much # of the print went to which spool; skip rather than mis-attribute. start_uuid = (start.get("tray_uuid") or "").lower() cur_uuid = (current.get("tray_uuid") or "").lower() if start_uuid and cur_uuid and start_uuid != cur_uuid: logger.info( "[SPOOLMAN] AMS%d-T%d: spool swapped mid-print (uuid changed), skipping remain-delta", ams_id, tray_id ) continue delta_pct = start["remain"] - current["remain"] if delta_pct <= 0: # A fresh spool reads 100% for the first tens of grams and the AMS # estimate drifts upward on its own, so this covers a real print # that simply left no trace at AMS granularity -- not only a refill. # Said out loud so it can be told apart from having nothing to # charge, which is what "no spools updated" alone looked like. logger.info( "[SPOOLMAN] AMS%d-T%d: remain%% did not fall over the print (%d%% -> %d%%), nothing charged", ams_id, tray_id, start["remain"], current["remain"], ) continue # No consumption captured at AMS granularity, or refilled spool_id = await _resolve_spool_id_via_slot_assignment(printer_id, ams_id, tray_id) if spool_id is None: logger.info( "[SPOOLMAN] AMS%d-T%d: consumed %d%% but has no Spoolman slot assignment, nothing charged", ams_id, tray_id, delta_pct, ) continue # Look up the spool's filament reference weight. Use a fresh GET so # we don't depend on a stale cached_spools list. Failure here is # silent-skip rather than fatal — other slots can still be written. try: spool = await client.get_spool(spool_id) except Exception as exc: # noqa: BLE001 logger.debug("[SPOOLMAN] AMS%d-T%d: get_spool(%s) failed: %s", ams_id, tray_id, spool_id, exc) continue filament = spool.get("filament") or {} ref_weight = filament.get("weight") if not ref_weight or ref_weight <= 0: logger.debug( "[SPOOLMAN] AMS%d-T%d: spool %s has no filament.weight, skipping remain-delta", ams_id, tray_id, spool_id, ) continue grams_used = round((delta_pct / 100.0) * ref_weight, 2) if grams_used <= 0: continue try: await client.use_spool(spool_id, grams_used) except Exception as exc: # noqa: BLE001 logger.warning( "[SPOOLMAN] AMS%d-T%d: use_spool(%s, %.2fg) failed: %s", ams_id, tray_id, spool_id, grams_used, exc ) continue spools_updated += 1 # No 3MF slot_id for this path — use the AMS slot key so the maps can # still be inspected by callers if needed. The archive rewrites # (#1494 colour, #2563 type) key on 3MF slot_ids, so remain-delta-only # prints intentionally don't participate (matches usage_tracker's # slot_id=None). if slot_colors_out is not None: color = filament.get("color_hex") if color: slot_colors_out[-(global_tray_id + 1)] = color if slot_materials_out is not None: material = filament.get("material") if material: slot_materials_out[-(global_tray_id + 1)] = material logger.info( "[SPOOLMAN] Archive %s AMS%d-T%d: %.2fg via remain-delta (%d%% of %.0fg) -> spool %s", archive_id, ams_id, tray_id, grams_used, delta_pct, ref_weight, spool_id, ) if not_in_print: logger.info( "[SPOOLMAN] Archive %s: slots not part of this print, left alone: %s", archive_id, ", ".join(not_in_print), ) return spools_updated async def _apply_spool_colors_to_archive( db, archive_id: int, filament_usage: list[dict], slot_colors: dict[int, str], ) -> None: """Overwrite an archive's ``filament_color`` with the colours of the Spoolman spools that fed the print (#1494). All-or-nothing, exactly like the built-in inventory path: the colour is only rewritten when every used slot resolved to a spool that carries a colour, so a partial match never drops slots from the archive. """ if not slot_colors: return from backend.app.models.archive import PrintArchive from backend.app.services.usage_tracker import ( _archive_colors_from_spools, _spool_color_to_hex, ) results = [{"slot_id": sid, "color": _spool_color_to_hex(hex_)} for sid, hex_ in slot_colors.items()] colors = _archive_colors_from_spools(filament_usage, results) if not colors: return archive = (await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))).scalar_one_or_none() if archive is None: return joined = ",".join(colors) if joined != archive.filament_color: logger.info( "[SPOOLMAN] Archive %s filament_color %r -> %r (from Spoolman spools)", archive_id, archive.filament_color, joined, ) archive.filament_color = joined await db.commit() async def _apply_spool_types_to_archive( db, archive_id: int, filament_usage: list[dict], slot_materials: dict[int, str], ) -> None: """Overwrite an archive's ``filament_type`` with the materials of the Spoolman spools that fed the print (#2563). All-or-nothing, exactly like the colour path and the built-in inventory path: the type is only rewritten when every used slot resolved to a spool that carries a material, so a partial match never drops slots from the archive or the material statistics. """ if not slot_materials: return from backend.app.models.archive import PrintArchive from backend.app.services.usage_tracker import _archive_types_from_spools results = [{"slot_id": sid, "material": material} for sid, material in slot_materials.items()] types = _archive_types_from_spools(filament_usage, results) if not types: return archive = (await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))).scalar_one_or_none() if archive is None: return joined = ",".join(types) if joined != archive.filament_type: logger.info( "[SPOOLMAN] Archive %s filament_type %r -> %r (from Spoolman spools)", archive_id, archive.filament_type, joined, ) archive.filament_type = joined await db.commit()