fix(inventory): fix usage tracking for remapped AMS slots and slicer prints

Three fixes for inventory spool usage not updating after prints:

1. Store ams_mapping from print command (reprint, library print, queue)
   so the usage tracker maps 3MF slots to the actual physical trays
   the user selected, not the default slicer mapping.

2. Track last_loaded_tray on printer state — the last valid tray_now
   (0-253) seen during printing. On H2D printers, tray_now is always
   255 in AMS data; the real tray resolves via snow field ~44s after
   print start but reverts to "unloaded" at completion. The fallback
   chain is: tray_now_at_start > current tray_now > last_loaded_tray.

3. Use color similarity (Euclidean RGB distance, threshold 50) instead
   of exact hex match for auto-unlink fingerprint checks. RFID sensors
   report slightly different shades across reads (e.g. distance ~43.6
   for the same spool), causing false assignment unlinks after prints.

CHANGELOG.md

@@ -46,6 +46,9 @@ All notable changes to Bambuddy will be documented in this file.
 - **Schedule Print Allows No Plate Selected for Multi-Plate Files** ([#394](https://github.com/maziggy/bambuddy/issues/394)) — When scheduling a multi-plate file from the file manager, the modal showed a "Selection required" warning but still allowed submission without selecting a plate. The job defaulted to plate 1, but the queue item didn't indicate which plate, and editing showed no plate selected. Now auto-selects the first plate by default when plates load, and the submit button validation applies to both archive and library files.
 - **3MF Usage Tracking Broken for Queue Prints from File Manager** ([#364](https://github.com/maziggy/bambuddy/issues/364)) — When a print was queued from the file manager (library file), the scheduler did not create an archive or register the expected print. The `on_print_start` callback had to re-download the 3MF from the printer via FTP, and if that failed, a fallback archive was created without the 3MF file — making 3MF-based filament usage tracking impossible. The queue item's `archive_id` also remained NULL, so the usage tracker could not find the queue's AMS slot mapping for correct spool resolution. The scheduler now creates an archive from the library file before uploading, links it to the queue item, and registers it as an expected print — matching the behavior of the direct library print route.
 - **Printer Queue Widget Shows "Archive #null" for File Manager Prints** ([#364](https://github.com/maziggy/bambuddy/issues/364)) — The "Next in queue" widget on the printer card only checked `archive_name` and `archive_id` when displaying the queued item name. Queue items from the file manager have `library_file_name` and `library_file_id` instead, so the widget displayed "Archive #null". Now falls back to `library_file_name` and `library_file_id`, matching the Queue page display logic.
+- **Inventory Usage Not Tracked for Remapped AMS Slots** ([#364](https://github.com/maziggy/bambuddy/issues/364)) — When reprinting an archive with a different AMS slot mapping (e.g. changing from slot A1 to C4 in the mapping modal), the usage tracker used the default 3MF slot-to-tray mapping instead of the actual mapping from the print command. The `ams_mapping` from reprint, library print, and queue print commands is now stored and used as the highest-priority mapping source for usage tracking.
+- **Inventory Usage Not Tracked for Slicer-Initiated Prints on H2D** ([#364](https://github.com/maziggy/bambuddy/issues/364)) — On H2D printers, the AMS `tray_now` field is always 255 in MQTT data. The actual tray is resolved via the snow field ~44 seconds after print start, but reverts to "unloaded" when the AMS retracts filament at completion. The usage tracker now tracks `last_loaded_tray` — the last valid tray seen during printing — as a fallback when both `tray_now` at start and at completion are invalid. Also captures `tray_now` at print start for printers that report a valid value before the RUNNING state.
+- **Spool Assignments Falsely Unlinked After Print Due to Color Variation** — The auto-unlink logic compared AMS tray colors against saved fingerprints using exact hex match. RFID sensors report slightly different color values across reads (e.g. `7CC4D5FF` vs `56B7E6FF` for the same spool, Euclidean distance ~43.6). Now uses a color similarity function with a tolerance threshold of 50, preventing false unlinks from minor RFID/firmware color variations while still detecting genuinely different spools.
 
 ### Improved
 - **Skip Objects: Click-to-Enlarge Lightbox** ([#396](https://github.com/maziggy/bambuddy/issues/396)) — The skip objects modal's small 208px image panel made it difficult to distinguish object markers when parts are small or close together. Clicking the image now opens a fullscreen lightbox overlay with the same image and markers at a much larger size (up to 600px). The 24px marker circles are proportionally smaller relative to the enlarged image, solving the overlap problem. Close via X button, Escape key, or clicking the backdrop. Escape cascades correctly — closes lightbox first, then the modal.

backend/app/api/routes/archives.py

@@ -2819,7 +2819,7 @@ async def reprint_archive(
         )
 
     # Register this as an expected print so we don't create a duplicate archive
-    register_expected_print(printer_id, remote_filename, archive_id)
+    register_expected_print(printer_id, remote_filename, archive_id, ams_mapping=body.ams_mapping)
 
     # Use plate_id from request if provided, otherwise auto-detect from 3MF file
     if body.plate_id is not None:

backend/app/api/routes/library.py

@@ -1868,7 +1868,7 @@ async def print_library_file(
         )
 
     # Register this as an expected print so we don't create a duplicate archive
-    register_expected_print(printer_id, remote_filename, archive.id)
+    register_expected_print(printer_id, remote_filename, archive.id, ams_mapping=body.ams_mapping)
 
     # Determine plate ID
     if body.plate_id is not None:

backend/app/main.py

@@ -249,6 +249,10 @@ _print_energy_start: dict[int, float] = {}
 # Track reprints to add costs on completion: {archive_id}
 _reprint_archives: set[int] = set()
 
+# Track AMS mapping for prints: {archive_id: [global_tray_id_per_slot]}
+# Used by usage tracker to map 3MF slots to physical AMS trays
+_print_ams_mappings: dict[int, list[int]] = {}
+
 # Track progress milestones for notifications: {printer_id: last_milestone_notified}
 # Milestones are 25, 50, 75. Value of 0 means no milestone notified yet for current print.
 _last_progress_milestone: dict[int, int] = {}
@@ -292,7 +296,7 @@ async def _get_plug_energy(plug, db) -> dict | None:
         return await tasmota_service.get_energy(plug)
 
 
-def register_expected_print(printer_id: int, filename: str, archive_id: int):
+def register_expected_print(printer_id: int, filename: str, archive_id: int, ams_mapping: list[int] | None = None):
     """Register an expected print from reprint/scheduled so we don't create duplicate archives."""
     # Store with multiple filename variations to catch different naming patterns
     _expected_prints[(printer_id, filename)] = archive_id
@@ -301,8 +305,11 @@ def register_expected_print(printer_id: int, filename: str, archive_id: int):
         base = filename[:-4]
         _expected_prints[(printer_id, base)] = archive_id
         _expected_prints[(printer_id, f"{base}.gcode")] = archive_id
+    # Store AMS mapping for usage tracking at print completion
+    if ams_mapping is not None:
+        _print_ams_mappings[archive_id] = ams_mapping
     logging.getLogger(__name__).info(
-        f"Registered expected print: printer={printer_id}, file={filename}, archive={archive_id}"
+        f"Registered expected print: printer={printer_id}, file={filename}, archive={archive_id}, ams_mapping={ams_mapping}"
     )
 
 
@@ -537,6 +544,8 @@ async def on_ams_change(printer_id: int, ams_data: list):
     except Exception as e:
         logger.warning("Failed to broadcast AMS change for printer %s: %s", printer_id, e)
 
+    from backend.app.utils.color_utils import colors_similar as _colors_similar
+
     # Auto-unlink spool assignments with stale fingerprints
     try:
         async with async_session() as db:
@@ -604,14 +613,14 @@ async def on_ams_change(printer_id: int, ams_data: list):
                     cur_type = current_tray.get("tray_type", "")
                     fp_color = assignment.fingerprint_color or ""
                     fp_type = assignment.fingerprint_type or ""
-                    if cur_color.upper() != fp_color.upper() or cur_type.upper() != fp_type.upper():
+                    if not _colors_similar(cur_color, fp_color) or cur_type.upper() != fp_type.upper():
                         # Fingerprint mismatch — but check if tray now matches the
                         # assigned spool (e.g. auto-configure changed the tray).
                         spool = assignment.spool
                         if spool:
                             spool_color = (spool.rgba or "FFFFFFFF").upper()
                             spool_type = (spool.material or "").upper()
-                            if cur_color.upper() == spool_color and cur_type.upper() == spool_type:
+                            if _colors_similar(cur_color, spool_color) and cur_type.upper() == spool_type:
                                 # Tray was reconfigured to match the spool — update fingerprint
                                 logger.info(
                                     "Auto-unlink: spool %d AMS%d-T%d — fingerprint mismatch but tray matches spool, updating fp",
@@ -2194,6 +2203,7 @@ async def on_print_complete(printer_id: int, data: dict):
 
     # Track filament consumption from AMS remain% deltas (skip if Spoolman handles usage)
     usage_results: list[dict] = []
+    stored_ams_mapping = _print_ams_mappings.pop(archive_id, None) if archive_id else None
     try:
         async with async_session() as db:
             from backend.app.api.routes.settings import get_setting
@@ -2204,7 +2214,12 @@ async def on_print_complete(printer_id: int, data: dict):
 
             async with async_session() as db:
                 usage_results = await usage_on_print_complete(
-                    printer_id, data, printer_manager, db, archive_id=archive_id
+                    printer_id,
+                    data,
+                    printer_manager,
+                    db,
+                    archive_id=archive_id,
+                    ams_mapping=stored_ams_mapping,
                 )
                 if usage_results:
                     await ws_manager.broadcast(

backend/app/services/bambu_mqtt.py

@@ -130,6 +130,8 @@ class PrinterState:
     active_extruder: int = 0
     # Currently loaded tray (global ID): 254/255 = external spools, 255 = no filament on legacy printers
     tray_now: int = 255
+    # Last valid tray_now (0-253) — survives unload (255) for usage tracking after print completes
+    last_loaded_tray: int = -1
     # Pending load target - used to track what tray we're loading for H2D disambiguation
     pending_tray_target: int | None = None
     # AMS status for filament change tracking (from print.ams.ams_status field)
@@ -927,6 +929,10 @@ class BambuMQTTClient:
                     # Trust the printer's reported value.
                     self.state.tray_now = parsed_tray_now
 
+                # Track last valid tray for usage tracking (survives retract → 255 at print end)
+                if 0 <= self.state.tray_now <= 253:
+                    self.state.last_loaded_tray = self.state.tray_now
+
                 logger.debug("[%s] tray_now updated: %s", self.serial_number, self.state.tray_now)
 
             # NOTE: ams_status is parsed BEFORE tray_now (see above) to ensure correct
@@ -1832,10 +1838,12 @@ class BambuMQTTClient:
                         if "diameter" in nozzle:
                             self.state.nozzles[idx].nozzle_diameter = str(nozzle["diameter"])
 
-        # Preserve AMS, vt_tray, and ams_extruder_map data when updating raw_data
+        # Preserve AMS, vt_tray, ams_extruder_map, and mapping data when updating raw_data
+        # (these fields aren't sent in every MQTT push, only when changed)
         ams_data = self.state.raw_data.get("ams")
         vt_tray_data = self.state.raw_data.get("vt_tray")
         ams_extruder_map_data = self.state.raw_data.get("ams_extruder_map")
+        mapping_data = self.state.raw_data.get("mapping")
         self.state.raw_data = data
         if ams_data is not None:
             self.state.raw_data["ams"] = ams_data
@@ -1843,6 +1851,12 @@ class BambuMQTTClient:
             self.state.raw_data["vt_tray"] = vt_tray_data
         if ams_extruder_map_data is not None:
             self.state.raw_data["ams_extruder_map"] = ams_extruder_map_data
+        if mapping_data is not None and "mapping" not in data:
+            self.state.raw_data["mapping"] = mapping_data
+
+        # Log mapping data when received (for usage tracking debugging)
+        if "mapping" in data:
+            logger.debug("[%s] MQTT mapping field: %s", self.serial_number, data["mapping"])
 
         # Log state transitions for debugging
         if "gcode_state" in data:

backend/app/services/print_scheduler.py

@@ -1035,13 +1035,6 @@ class PrintScheduler:
             await self._power_off_if_needed(db, item)
             return
 
-        # Register as expected print so we don't create a duplicate archive
-        # Only applicable for archive-based prints
-        if archive:
-            from backend.app.main import register_expected_print
-
-            register_expected_print(item.printer_id, remote_filename, archive.id)
-
         # Parse AMS mapping if stored
         ams_mapping = None
         if item.ams_mapping:
@@ -1050,6 +1043,13 @@ class PrintScheduler:
             except json.JSONDecodeError:
                 logger.warning("Queue item %s: Invalid AMS mapping JSON, ignoring", item.id)
 
+        # Register as expected print so we don't create a duplicate archive
+        # Only applicable for archive-based prints
+        if archive:
+            from backend.app.main import register_expected_print
+
+            register_expected_print(item.printer_id, remote_filename, archive.id, ams_mapping=ams_mapping)
+
         # IMPORTANT: Set status to "printing" BEFORE sending the print command.
         # This prevents phantom reprints if the backend crashes/restarts after the
         # print command is sent but before the status update is committed.

backend/app/services/usage_tracker.py

@@ -28,6 +28,8 @@ class PrintSession:
     print_name: str
     started_at: datetime
     tray_remain_start: dict[tuple[int, int], int] = field(default_factory=dict)
+    # tray_now at print start (correct value, unlike at completion where it's 255)
+    tray_now_at_start: int = -1
 
 
 # Module-level storage, keyed by printer_id
@@ -58,6 +60,9 @@ async def on_print_start(printer_id: int, data: dict, printer_manager) -> None:
 
     print_name = data.get("subtask_name", "") or data.get("filename", "unknown")
 
+    # Capture tray_now at print start (reliable, unlike at completion where it's 255)
+    tray_now_at_start = state.tray_now if state else -1
+
     # Always create session (even without valid remain data) so print_name
     # is available at completion for 3MF-based tracking
     session = PrintSession(
@@ -65,8 +70,10 @@ async def on_print_start(printer_id: int, data: dict, printer_manager) -> None:
         print_name=print_name,
         started_at=datetime.now(timezone.utc),
         tray_remain_start=tray_remain_start,
+        tray_now_at_start=tray_now_at_start,
     )
     _active_sessions[printer_id] = session
+    logger.info("[UsageTracker] Captured tray_now=%d at print start for printer %d", tray_now_at_start, printer_id)
 
     if tray_remain_start:
         logger.info(
@@ -85,6 +92,7 @@ async def on_print_complete(
     printer_manager,
     db: AsyncSession,
     archive_id: int | None = None,
+    ams_mapping: list[int] | None = None,
 ) -> list[dict]:
     """Compute consumption deltas and update spool weight_used/last_used.
 
@@ -99,13 +107,29 @@ async def on_print_complete(
     results = []
     handled_trays: set[tuple[int, int]] = set()
 
+    logger.info(
+        "[UsageTracker] on_print_complete: printer=%d, archive=%s, session=%s, ams_mapping=%s",
+        printer_id,
+        archive_id,
+        "yes" if session else "no",
+        ams_mapping,
+    )
+
     # --- Path 1 (PRIMARY): 3MF per-filament estimates ---
     if archive_id:
         print_name = (
             (session.print_name if session else None) or data.get("subtask_name", "") or data.get("filename", "unknown")
         )
         threemf_results = await _track_from_3mf(
-            printer_id, archive_id, status, print_name, handled_trays, printer_manager, db
+            printer_id,
+            archive_id,
+            status,
+            print_name,
+            handled_trays,
+            printer_manager,
+            db,
+            ams_mapping=ams_mapping,
+            tray_now_at_start=session.tray_now_at_start if session else -1,
         )
         results.extend(threemf_results)
 
@@ -213,6 +237,8 @@ async def _track_from_3mf(
     handled_trays: set[tuple[int, int]],
     printer_manager,
     db: AsyncSession,
+    ams_mapping: list[int] | None = None,
+    tray_now_at_start: int = -1,
 ) -> list[dict]:
     """Track usage from 3MF per-filament slicer data (primary path).
 
@@ -221,9 +247,10 @@ async def _track_from_3mf(
     then falls back to linear scaling by progress.
 
     Slot-to-tray mapping priority:
-    1. Queue item ams_mapping (for queue-initiated prints)
-    2. tray_now from printer state (for single-filament non-queue prints)
-    3. Default mapping: slot_id - 1 = global_tray_id (last resort)
+    1. Stored ams_mapping from print command (reprints/direct prints)
+    2. Queue item ams_mapping (for queue-initiated prints)
+    3. tray_now from printer state (for single-filament non-queue prints)
+    4. Default mapping: slot_id - 1 = global_tray_id (last resort)
     """
     from backend.app.core.config import settings as app_settings
     from backend.app.models.archive import PrintArchive
@@ -233,43 +260,76 @@ async def _track_from_3mf(
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
     archive = result.scalar_one_or_none()
     if not archive or not archive.file_path:
+        logger.info("[UsageTracker] 3MF: archive %s has no file_path, skipping", archive_id)
         return []
 
     file_path = app_settings.base_dir / archive.file_path
     if not file_path.exists():
+        logger.info("[UsageTracker] 3MF: file not found: %s", file_path)
         return []
 
     filament_usage = extract_filament_usage_from_3mf(file_path)
     if not filament_usage:
+        logger.info("[UsageTracker] 3MF: no filament usage data in %s", file_path)
         return []
 
+    logger.info("[UsageTracker] 3MF: archive %s, filament_usage=%s", archive_id, filament_usage)
+
     # --- Resolve slot-to-tray mapping ---
-    # 1. Try queue item ams_mapping (queue-initiated prints store the exact mapping)
-    slot_to_tray = None
-    queue_result = await db.execute(
-        select(PrintQueueItem)
-        .where(PrintQueueItem.archive_id == archive_id)
-        .where(PrintQueueItem.status.in_(["printing", "completed", "failed"]))
+    # 1. Use stored ams_mapping from the print command (reprints/direct prints)
+    slot_to_tray = ams_mapping
+
+    # 2. Try queue item ams_mapping (queue-initiated prints store the exact mapping)
+    if not slot_to_tray:
+        queue_result = await db.execute(
+            select(PrintQueueItem)
+            .where(PrintQueueItem.archive_id == archive_id)
+            .where(PrintQueueItem.status.in_(["printing", "completed", "failed"]))
+        )
+        queue_item = queue_result.scalar_one_or_none()
+        if queue_item and queue_item.ams_mapping:
+            try:
+                slot_to_tray = json.loads(queue_item.ams_mapping)
+            except (json.JSONDecodeError, TypeError):
+                pass
+
+    logger.info(
+        "[UsageTracker] 3MF: slot_to_tray=%s (source: %s)",
+        slot_to_tray,
+        "print_cmd" if ams_mapping else ("queue" if slot_to_tray else "none"),
     )
-    queue_item = queue_result.scalar_one_or_none()
-    if queue_item and queue_item.ams_mapping:
-        try:
-            slot_to_tray = json.loads(queue_item.ams_mapping)
-        except (json.JSONDecodeError, TypeError):
-            pass
-
-    # 2. For single-filament non-queue prints, use tray_now from printer state
+
+    # 3. For single-filament non-queue prints, use tray_now from printer state
+    #    Priority: tray_now_at_start > current tray_now > last_loaded_tray > vt_tray check
     nonzero_slots = [u for u in filament_usage if u.get("used_g", 0) > 0]
     tray_now_override: int | None = None
     if not slot_to_tray and len(nonzero_slots) == 1:
         state = printer_manager.get_status(printer_id)
-        if state and 0 <= state.tray_now <= 254:
+        # Try tray_now_at_start first (captured at print start)
+        if 0 <= tray_now_at_start <= 254:
+            tray_now_override = tray_now_at_start
+            logger.info("[UsageTracker] 3MF: using tray_now_at_start=%d (single-filament fallback)", tray_now_at_start)
+        elif state and 0 <= state.tray_now <= 254:
+            # Current state is valid (printer didn't retract yet)
             tray_now_override = state.tray_now
+            logger.info("[UsageTracker] 3MF: using current tray_now=%d", state.tray_now)
+        elif state and 0 <= state.last_loaded_tray <= 253:
+            # Last valid tray before retract (H2D retracts before completion callback)
+            tray_now_override = state.last_loaded_tray
+            logger.info("[UsageTracker] 3MF: using last_loaded_tray=%d (post-retract fallback)", state.last_loaded_tray)
         elif state and state.tray_now == 255:
             # 255 = "no filament" on legacy printers, but valid 2nd external spool on H2-series
             vt_tray = state.raw_data.get("vt_tray") or []
             if any(int(vt.get("id", 0)) == 255 for vt in vt_tray if isinstance(vt, dict)):
                 tray_now_override = state.tray_now
+                logger.info("[UsageTracker] 3MF: using tray_now=255 (H2-series external spool)")
+        if tray_now_override is None:
+            logger.info(
+                "[UsageTracker] 3MF: no valid tray_now (at_start=%d, current=%s, last_loaded=%s)",
+                tray_now_at_start,
+                state.tray_now if state else "N/A",
+                state.last_loaded_tray if state else "N/A",
+            )
 
     # Scale factor for partial prints (failed/aborted)
     if status == "completed":
@@ -338,6 +398,16 @@ async def _track_from_3mf(
             ams_id = global_tray_id // 4
             tray_id = global_tray_id % 4
 
+        logger.info(
+            "[UsageTracker] 3MF: slot_id=%d -> global_tray=%d -> AMS%d-T%d (used_g=%.1f, tray_now_override=%s)",
+            slot_id,
+            global_tray_id,
+            ams_id,
+            tray_id,
+            used_g,
+            tray_now_override,
+        )
+
         key = (ams_id, tray_id)
         if key in handled_trays:
             continue
@@ -352,6 +422,7 @@ async def _track_from_3mf(
         )
         assignment = assign_result.scalar_one_or_none()
         if not assignment:
+            logger.info("[UsageTracker] 3MF: no spool assignment at printer %d AMS%d-T%d", printer_id, ams_id, tray_id)
             continue
 
         # Load spool
@@ -401,6 +472,8 @@ async def _track_from_3mf(
         # Determine mapping source for debug logging
         if tray_now_override is not None:
             map_src = ", tray_now"
+        elif slot_to_tray and ams_mapping:
+            map_src = ", print_cmd_map"
         elif slot_to_tray:
             map_src = ", queue_map"
         else:

backend/app/utils/color_utils.py

@@ -0,0 +1,24 @@
+"""Color comparison utilities for RFID/firmware color matching."""
+
+
+def colors_similar(hex_a: str, hex_b: str, threshold: int = 50) -> bool:
+    """Compare two RRGGBB(AA) hex colors with tolerance for RFID/firmware variations.
+
+    Uses Euclidean RGB distance. Alpha channel (bytes 7-8) is ignored.
+    Default threshold of 50 accommodates typical RFID read variations
+    (e.g. 7CC4D5 vs 56B7E6 = distance ~43.6) while rejecting clearly
+    different colors (e.g. red vs blue = distance ~360).
+    """
+    a = hex_a.strip().upper()
+    b = hex_b.strip().upper()
+    if a == b:
+        return True
+    if len(a) < 6 or len(b) < 6:
+        return False
+    try:
+        ra, ga, ba = int(a[0:2], 16), int(a[2:4], 16), int(a[4:6], 16)
+        rb, gb, bb = int(b[0:2], 16), int(b[2:4], 16), int(b[4:6], 16)
+    except ValueError:
+        return False
+    dist = ((ra - rb) ** 2 + (ga - gb) ** 2 + (ba - bb) ** 2) ** 0.5
+    return dist <= threshold

backend/tests/unit/test_color_utils.py

@@ -0,0 +1,54 @@
+"""Unit tests for color_utils — hex color similarity comparison."""
+
+from backend.app.utils.color_utils import colors_similar
+
+
+class TestColorsSimilar:
+    """Tests for colors_similar()."""
+
+    def test_exact_match(self):
+        assert colors_similar("FF0000FF", "FF0000FF") is True
+
+    def test_exact_match_case_insensitive(self):
+        assert colors_similar("ff0000ff", "FF0000FF") is True
+
+    def test_similar_colors_within_threshold(self):
+        # Real-world case: RFID read variation (distance ~43.6)
+        assert colors_similar("7CC4D5FF", "56B7E6FF") is True
+
+    def test_different_colors_beyond_threshold(self):
+        # Red vs blue (distance ~360)
+        assert colors_similar("FF0000FF", "0000FFFF") is False
+
+    def test_ignores_alpha_channel(self):
+        # Same RGB, different alpha — should match
+        assert colors_similar("FF000000", "FF0000FF") is True
+
+    def test_six_digit_hex(self):
+        assert colors_similar("FF0000", "FF0000") is True
+
+    def test_short_string_returns_false(self):
+        assert colors_similar("FFF", "FF0000") is False
+        assert colors_similar("", "FF0000") is False
+
+    def test_empty_strings_match(self):
+        """Two empty strings are exact match (both missing data)."""
+        assert colors_similar("", "") is True
+
+    def test_invalid_hex_returns_false(self):
+        assert colors_similar("ZZZZZZ", "FF0000") is False
+
+    def test_whitespace_stripped(self):
+        assert colors_similar(" FF0000 ", "FF0000") is True
+
+    def test_custom_threshold(self):
+        # Distance ~43.6 — within 50 but outside 30
+        assert colors_similar("7CC4D5FF", "56B7E6FF", threshold=30) is False
+        assert colors_similar("7CC4D5FF", "56B7E6FF", threshold=50) is True
+
+    def test_black_and_near_black(self):
+        # (10, 10, 10) distance from (0, 0, 0) = ~17.3
+        assert colors_similar("000000", "0A0A0A") is True
+
+    def test_white_and_off_white(self):
+        assert colors_similar("FFFFFF", "F0F0F0") is True

backend/tests/unit/test_usage_tracker.py

@@ -104,7 +104,8 @@ class TestOnPrintStart:
         """Captures AMS remain% at print start."""
         printer_manager = MagicMock()
         printer_manager.get_status.return_value = SimpleNamespace(
-            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}, {"id": 1, "remain": 50}]}]}
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}, {"id": 1, "remain": 50}]}]},
+            tray_now=5,
         )
 
         await on_print_start(1, {"subtask_name": "Benchy"}, printer_manager)
@@ -114,12 +115,39 @@ class TestOnPrintStart:
         assert session.print_name == "Benchy"
         assert session.tray_remain_start == {(0, 0): 80, (0, 1): 50}
 
+    @pytest.mark.asyncio
+    async def test_captures_tray_now_at_start(self):
+        """Captures tray_now at print start for later use in usage tracking."""
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]},
+            tray_now=9,
+        )
+
+        await on_print_start(1, {"subtask_name": "Test"}, printer_manager)
+
+        assert _active_sessions[1].tray_now_at_start == 9
+
+    @pytest.mark.asyncio
+    async def test_tray_now_at_start_255_when_unloaded(self):
+        """Captures tray_now=255 when printer has no filament loaded at start."""
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]},
+            tray_now=255,
+        )
+
+        await on_print_start(1, {"subtask_name": "Test"}, printer_manager)
+
+        assert _active_sessions[1].tray_now_at_start == 255
+
     @pytest.mark.asyncio
     async def test_creates_session_without_remain(self):
         """Creates session even without valid remain data (for 3MF tracking)."""
         printer_manager = MagicMock()
         printer_manager.get_status.return_value = SimpleNamespace(
-            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": -1}]}]}
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": -1}]}]},
+            tray_now=255,
         )
 
         await on_print_start(1, {"subtask_name": "Test"}, printer_manager)
@@ -657,6 +685,157 @@ class TestTrackFrom3mf:
         assert results[0]["ams_id"] == 0
         assert results[0]["tray_id"] == 0
 
+    @pytest.mark.asyncio
+    async def test_stored_ams_mapping_overrides_all(self):
+        """Stored ams_mapping from print command takes priority over queue and tray_now."""
+        # Spool at AMS2-T1 (global_tray_id=9)
+        spool = _make_spool(spool_id=10, label_weight=1000)
+        assignment = _make_assignment(spool_id=10, ams_id=2, tray_id=1)
+        archive = _make_archive(archive_id=50)
+
+        # db: archive, assignment, spool (no queue lookup when ams_mapping provided)
+        db = _mock_db_sequential([archive, assignment, spool])
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            progress=100,
+            layer_num=50,
+            tray_now=0,  # Different from mapped tray — should be ignored
+            last_loaded_tray=0,
+        )
+
+        filament_usage = [{"slot_id": 2, "used_g": 1.57, "type": "PLA", "color": "#FFFFFF"}]
+        handled_trays: set[tuple[int, int]] = set()
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=filament_usage,
+            ),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            # ams_mapping: slot 2 (index 1) -> tray 9 (AMS2-T1)
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=50,
+                status="completed",
+                print_name="Test",
+                handled_trays=handled_trays,
+                printer_manager=printer_manager,
+                db=db,
+                ams_mapping=[-1, 9],
+            )
+
+        assert len(results) == 1
+        assert results[0]["spool_id"] == 10
+        assert results[0]["ams_id"] == 2
+        assert results[0]["tray_id"] == 1
+        assert results[0]["weight_used"] == 1.6  # rounded
+
+    @pytest.mark.asyncio
+    async def test_last_loaded_tray_fallback(self):
+        """Falls back to last_loaded_tray when tray_now_at_start and current tray_now are both 255."""
+        # Spool at AMS2-T1 (global_tray_id=9)
+        spool = _make_spool(spool_id=11, label_weight=1000)
+        assignment = _make_assignment(spool_id=11, ams_id=2, tray_id=1)
+        archive = _make_archive(archive_id=60)
+
+        # db: archive, queue_item(None), assignment, spool
+        db = _mock_db_sequential([archive, None, assignment, spool])
+
+        # H2D scenario: tray_now=255 at completion, but last_loaded_tray=9
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            progress=100,
+            layer_num=50,
+            tray_now=255,
+            last_loaded_tray=9,
+        )
+
+        filament_usage = [{"slot_id": 6, "used_g": 1.52, "type": "PLA", "color": "#7CC4D5"}]
+        handled_trays: set[tuple[int, int]] = set()
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=filament_usage,
+            ),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=60,
+                status="completed",
+                print_name="Cube",
+                handled_trays=handled_trays,
+                printer_manager=printer_manager,
+                db=db,
+                tray_now_at_start=255,  # H2D: 255 at start too
+            )
+
+        assert len(results) == 1
+        assert results[0]["spool_id"] == 11
+        assert results[0]["ams_id"] == 2
+        assert results[0]["tray_id"] == 1
+
+    @pytest.mark.asyncio
+    async def test_tray_now_at_start_preferred_over_last_loaded(self):
+        """tray_now_at_start is used before last_loaded_tray fallback."""
+        spool = _make_spool(spool_id=3, label_weight=1000)
+        assignment = _make_assignment(spool_id=3, ams_id=1, tray_id=1)
+        archive = _make_archive(archive_id=70)
+
+        db = _mock_db_sequential([archive, None, assignment, spool])
+
+        # tray_now_at_start=5 (valid), last_loaded_tray=9 (different) — should use 5
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            progress=100,
+            layer_num=50,
+            tray_now=255,
+            last_loaded_tray=9,
+        )
+
+        filament_usage = [{"slot_id": 1, "used_g": 5.0, "type": "PLA", "color": ""}]
+        handled_trays: set[tuple[int, int]] = set()
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=filament_usage,
+            ),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=70,
+                status="completed",
+                print_name="Test",
+                handled_trays=handled_trays,
+                printer_manager=printer_manager,
+                db=db,
+                tray_now_at_start=5,  # AMS1-T1
+            )
+
+        assert len(results) == 1
+        assert results[0]["ams_id"] == 1
+        assert results[0]["tray_id"] == 1
+
 
 class TestNotificationVariables:
     """Tests for filament_details formatting in notifications."""

static/index.html

@@ -23,7 +23,7 @@
 
     <!-- Splash screens for iOS -->
     <link rel="apple-touch-startup-image" href="/img/android-chrome-512x512.png" />
-    <script type="module" crossorigin src="/assets/index-B2wPwovd.js"></script>
+    <script type="module" crossorigin src="/assets/index-yL-bH2h4.js"></script>
     <link rel="stylesheet" crossorigin href="/assets/index-DF7TfzH1.css">
   </head>
   <body>