Refactor Spoolman per-filament tracking (PR #277 follow-up)

Extract Spoolman tracking from main.py into dedicated service module,
DRY up repeated helpers, add i18n for new settings UI, and add tests.

- Move ~460 lines from main.py to services/spoolman_tracking.py
- Extract _resolve_spool_tag, _resolve_global_tray_id, build_ams_tray_lookup helpers
- Replace fragile tuple return in get_spoolman_settings() with dict
- Wire 4 new UI strings through i18n (en/de/ja)
- Add 42 backend unit tests (spoolman tracking helpers + 3MF parsing)
- Add 6 frontend tests for weight sync and partial usage toggles
- Update CHANGELOG, wiki, and website docs

Closes PR #277

CHANGELOG.md

@@ -23,6 +23,13 @@ All notable changes to Bambuddy will be documented in this file.
   - Security scan results visible in GitHub Security tab
 
 ### Enhanced
+- **Per-Filament Spoolman Usage Tracking** (PR #277):
+  - Reports exact filament consumption per spool to Spoolman after each print
+  - Parses G-code from 3MF files for layer-by-layer extrusion data (multi-material support)
+  - New setting: "Disable AMS Estimated Weight Sync" to prefer Spoolman usage tracking over AMS weight estimates
+  - New setting: "Report Partial Usage for Failed Prints" estimates filament used up to the failure point based on layer progress
+  - Persists tracking data in SQLite for reliability across restarts
+  - Extracted Spoolman tracking into dedicated service module with DRY helpers
 - **3D Model Viewer Improvements** (PR #262):
   - Added plate selector for multi-plate 3MF files with thumbnail previews
   - Object count display shows number of objects per plate and total

backend/app/api/routes/spoolman.py

@@ -52,29 +52,31 @@ class SyncResult(BaseModel):
     errors: list[str]
 
 
-async def get_spoolman_settings(db: AsyncSession) -> tuple[bool, str, str, bool]:
+async def get_spoolman_settings(db: AsyncSession) -> dict:
     """Get Spoolman settings from database.
 
     Returns:
-        Tuple of (enabled, url, sync_mode, disable_weight_sync)
+        Dict with keys: enabled, url, sync_mode, disable_weight_sync
     """
-    enabled = False
-    url = ""
-    sync_mode = "auto"
-    disable_weight_sync = False
+    settings = {
+        "enabled": False,
+        "url": "",
+        "sync_mode": "auto",
+        "disable_weight_sync": False,
+    }
 
     result = await db.execute(select(Settings))
     for setting in result.scalars().all():
         if setting.key == "spoolman_enabled":
-            enabled = setting.value.lower() == "true"
+            settings["enabled"] = setting.value.lower() == "true"
         elif setting.key == "spoolman_url":
-            url = setting.value
+            settings["url"] = setting.value
         elif setting.key == "spoolman_sync_mode":
-            sync_mode = setting.value
+            settings["sync_mode"] = setting.value
         elif setting.key == "spoolman_disable_weight_sync":
-            disable_weight_sync = setting.value.lower() == "true"
+            settings["disable_weight_sync"] = setting.value.lower() == "true"
 
-    return enabled, url, sync_mode, disable_weight_sync
+    return settings
 
 
 @router.get("/status", response_model=SpoolmanStatus)
@@ -83,7 +85,8 @@ async def get_spoolman_status(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """Get Spoolman integration status."""
-    enabled, url, _, _ = await get_spoolman_settings(db)
+    sm = await get_spoolman_settings(db)
+    enabled, url = sm["enabled"], sm["url"]
 
     client = await get_spoolman_client()
     connected = False
@@ -103,7 +106,8 @@ async def connect_spoolman(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
 ):
     """Connect to Spoolman server using configured URL."""
-    enabled, url, _, _ = await get_spoolman_settings(db)
+    sm = await get_spoolman_settings(db)
+    enabled, url = sm["enabled"], sm["url"]
 
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
@@ -147,7 +151,8 @@ async def sync_printer_ams(
 ):
     """Sync AMS data from a specific printer to Spoolman."""
     # Check if Spoolman is enabled and connected
-    enabled, url, _, disable_weight_sync = await get_spoolman_settings(db)
+    sm = await get_spoolman_settings(db)
+    enabled, url, disable_weight_sync = sm["enabled"], sm["url"], sm["disable_weight_sync"]
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -288,7 +293,8 @@ async def sync_all_printers(
 ):
     """Sync AMS data from all connected printers to Spoolman."""
     # Check if Spoolman is enabled
-    enabled, url, _, disable_weight_sync = await get_spoolman_settings(db)
+    sm = await get_spoolman_settings(db)
+    enabled, url, disable_weight_sync = sm["enabled"], sm["url"], sm["disable_weight_sync"]
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -417,7 +423,8 @@ async def get_spools(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """Get all spools from Spoolman."""
-    enabled, url, _, _ = await get_spoolman_settings(db)
+    sm = await get_spoolman_settings(db)
+    enabled, url = sm["enabled"], sm["url"]
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -441,7 +448,8 @@ async def get_filaments(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """Get all filaments from Spoolman."""
-    enabled, url, _, _ = await get_spoolman_settings(db)
+    sm = await get_spoolman_settings(db)
+    enabled, url = sm["enabled"], sm["url"]
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -476,7 +484,8 @@ async def get_unlinked_spools(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """Get all Spoolman spools that don't have a tag (not linked to AMS)."""
-    enabled, url, _, _ = await get_spoolman_settings(db)
+    sm = await get_spoolman_settings(db)
+    enabled, url = sm["enabled"], sm["url"]
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -521,7 +530,8 @@ async def get_linked_spools(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """Get a map of tag -> spool_id for all Spoolman spools that have a tag assigned."""
-    enabled, url, _, _ = await get_spoolman_settings(db)
+    sm = await get_spoolman_settings(db)
+    enabled, url = sm["enabled"], sm["url"]
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -565,7 +575,8 @@ async def link_spool(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_UPDATE),
 ):
     """Link a Spoolman spool to an AMS tray by setting the tag to tray_uuid."""
-    enabled, url, _, _ = await get_spoolman_settings(db)
+    sm = await get_spoolman_settings(db)
+    enabled, url = sm["enabled"], sm["url"]
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 

backend/app/main.py

@@ -1,5 +1,4 @@
 import asyncio
-import json
 import logging
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta, timezone
@@ -225,6 +224,11 @@ from backend.app.services.printer_manager import (
 )
 from backend.app.services.smart_plug_manager import smart_plug_manager
 from backend.app.services.spoolman import close_spoolman_client, get_spoolman_client, init_spoolman_client
+from backend.app.services.spoolman_tracking import (
+    cleanup_tracking as _cleanup_spoolman_tracking,
+    report_usage as _report_spoolman_usage,
+    store_print_data as _store_spoolman_print_data,
+)
 from backend.app.services.tasmota import tasmota_service
 
 # Track active prints: {(printer_id, filename): archive_id}
@@ -295,466 +299,6 @@ _last_status_broadcast: dict[int, str] = {}
 _nozzle_count_updated: set[int] = set()  # Track printers where we've updated nozzle_count
 
 
-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": "..."}, 1: {...}, ...}
-    """
-    lookup = {}
-    ams_data = raw_data.get("ams", [])
-    for ams_unit in ams_data:
-        ams_id = ams_unit.get("id", 0)
-        for tray in ams_unit.get("tray", []):
-            tray_id = tray.get("id", 0)
-            global_tray_id = 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 (global_tray_id = 254)
-    vt_tray = raw_data.get("vt_tray")
-    if vt_tray and vt_tray.get("tray_type"):
-        lookup[254] = {
-            "tray_uuid": vt_tray.get("tray_uuid", ""),
-            "tag_uid": vt_tray.get("tag_uid", ""),
-            "tray_type": vt_tray.get("tray_type", ""),
-        }
-
-    return lookup
-
-
-async def _store_spoolman_print_data(printer_id: int, archive_id: int, file_path: str, db, logger):
-    """Store Spoolman tracking data at print start (persisted to database)."""
-    from backend.app.api.routes.settings import get_setting
-    from backend.app.models.active_print_spoolman import ActivePrintSpoolman
-    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
-
-    # Only store tracking data if "Disable AMS Weight Sync" is enabled
-    # (otherwise we're using AMS percentage-based estimates, not per-usage tracking)
-    disable_weight_sync_str = await get_setting(db, "spoolman_disable_weight_sync")
-    disable_weight_sync = disable_weight_sync_str and disable_weight_sync_str.lower() == "true"
-    if not disable_weight_sync:
-        logger.debug("[SPOOLMAN] Weight sync enabled, skipping per-usage tracking data storage")
-        return
-
-    # Get 3MF file path
-    full_path = app_settings.base_dir / file_path
-    if not full_path.exists():
-        logger.debug(f"[SPOOLMAN] 3MF file not found: {full_path}")
-        return
-
-    # Extract per-filament usage from 3MF (total usage per slot)
-    filament_usage = extract_filament_usage_from_3mf(full_path)
-    if not filament_usage:
-        logger.debug(f"[SPOOLMAN] No filament usage data in 3MF for archive {archive_id}")
-        return
-
-    # Get current AMS tray state
-    state = printer_manager.get_status(printer_id)
-    ams_trays = {}
-    if state and state.raw_data:
-        ams_trays = _build_ams_tray_lookup(state.raw_data)
-
-    # Get custom slot-to-tray mapping from queue item (if this is a queued print)
-    slot_to_tray = None
-    from backend.app.models.print_queue import PrintQueueItem
-
-    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()
-    if queue_item and queue_item.ams_mapping:
-        try:
-            slot_to_tray = json.loads(queue_item.ams_mapping)
-        except json.JSONDecodeError:
-            pass
-
-    # Parse G-code for per-layer filament usage (for accurate partial usage tracking)
-    layer_usage = extract_layer_filament_usage_from_3mf(full_path)
-    layer_usage_json = None
-    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(f"[SPOOLMAN] Parsed {len(layer_usage)} layers from G-code")
-
-    # Extract filament properties (density, diameter) for mm -> grams conversion
-    filament_properties = extract_filament_properties_from_3mf(full_path)
-
-    # Delete any existing row for this printer/archive (shouldn't exist, but just in case)
-    from sqlalchemy import delete
-
-    await db.execute(
-        delete(ActivePrintSpoolman)
-        .where(ActivePrintSpoolman.printer_id == printer_id)
-        .where(ActivePrintSpoolman.archive_id == archive_id)
-    )
-
-    # Insert new tracking data
-    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,
-    )
-    db.add(tracking)
-    await db.commit()
-
-    logger.info(f"[SPOOLMAN] Stored tracking data for print: printer={printer_id}, archive={archive_id}")
-    logger.debug(f"[SPOOLMAN] Filament usage: {filament_usage}")
-    logger.debug(f"[SPOOLMAN] AMS trays: {list(ams_trays.keys())}")
-    if slot_to_tray:
-        logger.debug(f"[SPOOLMAN] Custom slot mapping: {slot_to_tray}")
-    if layer_usage_json:
-        logger.debug("[SPOOLMAN] Layer usage data available for partial tracking")
-
-
-async def _cleanup_spoolman_tracking(printer_id: int, archive_id: int, db, logger):
-    """Report partial usage and clean up Spoolman tracking data for failed/aborted prints."""
-    from sqlalchemy import delete, select
-
-    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(f"[SPOOLMAN] No tracking data to clean up for printer={printer_id}, archive={archive_id}")
-        return
-
-    # Try to report partial usage before cleanup
-    try:
-        await _report_partial_spoolman_usage(printer_id, tracking, logger)
-    except Exception as e:
-        logger.warning(f"[SPOOLMAN] Partial usage report failed: {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(f"[SPOOLMAN] Cleaned up tracking data for printer={printer_id}, archive={archive_id}")
-
-
-async def _report_partial_spoolman_usage(printer_id: int, tracking, logger):
-    """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.
-
-    This is called when a print fails or is cancelled.
-    """
-    from backend.app.api.routes.settings import get_setting
-    from backend.app.utils.threemf_tools import get_cumulative_usage_at_layer, mm_to_grams
-
-    async with async_session() as db:
-        # 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
-    state = printer_manager.get_status(printer_id)
-    if not state:
-        logger.debug("[SPOOLMAN] No printer state available for partial usage")
-        return
-
-    current_layer = state.layer_num
-    total_layers = state.total_layers
-
-    # Need current layer to calculate usage
-    if not current_layer or current_layer <= 0:
-        logger.debug("[SPOOLMAN] No progress to report (layer 0 or unknown)")
-        return
-
-    logger.info(f"[SPOOLMAN] Reporting partial usage at layer {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 = tracking.ams_trays or {}
-    slot_to_tray = tracking.slot_to_tray
-
-    # Convert ams_trays keys from string to int (JSON serialization)
-    ams_trays = {int(k): v for k, v in ams_trays.items()}
-
-    async with async_session() as db:
-        from backend.app.api.routes.settings import get_setting
-
-        # Get Spoolman client
-        client = await get_spoolman_client()
-        if not client:
-            spoolman_url = await get_setting(db, "spoolman_url")
-            if spoolman_url:
-                client = await init_spoolman_client(spoolman_url)
-
-        if not client or not await client.health_check():
-            logger.warning("[SPOOLMAN] Not reachable for partial usage reporting")
-            return
-
-        spools_updated = 0
-
-        # Try to use accurate G-code parsed data
-        if layer_usage:
-            # Convert string keys back to int (JSON serialization issue)
-            # Both outer (layer) and inner (filament_id) keys need conversion
-            layer_usage_int = {
-                int(layer): {int(fid): mm for fid, mm in filaments.items()} for layer, filaments in layer_usage.items()
-            }
-
-            # Get cumulative usage at current layer
-            usage_mm = get_cumulative_usage_at_layer(layer_usage_int, current_layer)
-
-            if usage_mm:
-                logger.info(f"[SPOOLMAN] Using G-code parsed data for layer {current_layer}")
-
-                # Process each filament's usage
-                for filament_id, mm_used in usage_mm.items():
-                    # filament_id is 0-based from G-code, slot_id is 1-based
-                    slot_id = filament_id + 1
-
-                    # Determine which tray was used for this slot
-                    global_tray_id = slot_id - 1  # Default: slot 1 -> tray 0
-                    if slot_to_tray and slot_id <= len(slot_to_tray):
-                        mapped_tray = slot_to_tray[slot_id - 1]
-                        if mapped_tray >= 0:
-                            global_tray_id = mapped_tray
-
-                    tray_info = ams_trays.get(global_tray_id)
-                    if not tray_info:
-                        logger.debug(f"[SPOOLMAN] Slot {slot_id}: no tray at global_tray_id {global_tray_id}")
-                        continue
-
-                    # Get spool identifier (prefer tray_uuid over tag_uid)
-                    tray_uuid = tray_info.get("tray_uuid", "")
-                    tag_uid = tray_info.get("tag_uid", "")
-                    spool_tag = tray_uuid if tray_uuid and tray_uuid != "00000000000000000000000000000000" else tag_uid
-
-                    if not spool_tag:
-                        logger.debug(f"[SPOOLMAN] Slot {slot_id}: no identifier for tray {global_tray_id}")
-                        continue
-
-                    # Find the spool in Spoolman
-                    spool = await client.find_spool_by_tag(spool_tag)
-                    if not spool:
-                        logger.debug(f"[SPOOLMAN] Slot {slot_id}: no spool for tag {spool_tag[:16]}...")
-                        continue
-
-                    # Get density from Spoolman's filament data (most accurate)
-                    # Falls back to 3MF properties, then to default PLA density
-                    filament_data = spool.get("filament", {})
-                    density = filament_data.get("density")
-                    diameter = filament_data.get("diameter", 1.75)
-
-                    if not density:
-                        # Fallback to 3MF properties
-                        props = filament_properties.get(str(slot_id), filament_properties.get(slot_id, {}))
-                        density = props.get("density", 1.24)
-                        logger.debug(f"[SPOOLMAN] Using fallback density {density} for slot {slot_id}")
-
-                    # Convert mm to grams using Spoolman's filament density
-                    grams_used = round(mm_to_grams(mm_used, diameter, density), 2)
-
-                    if grams_used <= 0:
-                        continue
-
-                    # Report usage to Spoolman
-                    result = await client.use_spool(spool["id"], grams_used)
-                    if result:
-                        logger.info(
-                            f"[SPOOLMAN] Partial (G-code): slot {slot_id}: {grams_used}g ({mm_used:.1f}mm, d={density}) -> spool {spool['id']}"
-                        )
-                        spools_updated += 1
-
-                if spools_updated > 0:
-                    logger.info(f"[SPOOLMAN] Reported partial usage to {spools_updated} spool(s) using G-code data")
-                return
-
-        # Fallback: linear interpolation (if no G-code data available)
-        if not total_layers or total_layers <= 0:
-            logger.debug(f"[SPOOLMAN] Cannot use linear fallback: total_layers={total_layers}")
-            return
-
-        progress_ratio = min(current_layer / total_layers, 1.0)
-        logger.info(f"[SPOOLMAN] Falling back to linear interpolation ({progress_ratio:.1%})")
-
-        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:
-                continue
-
-            # Calculate partial usage using linear interpolation
-            partial_used_g = round(total_used_g * progress_ratio, 2)
-
-            if partial_used_g <= 0:
-                continue
-
-            # Determine which tray was used for this slot
-            global_tray_id = slot_id - 1
-            if slot_to_tray and slot_id <= len(slot_to_tray):
-                mapped_tray = slot_to_tray[slot_id - 1]
-                if mapped_tray >= 0:
-                    global_tray_id = mapped_tray
-
-            tray_info = ams_trays.get(global_tray_id)
-            if not tray_info:
-                continue
-
-            # Get spool identifier
-            tray_uuid = tray_info.get("tray_uuid", "")
-            tag_uid = tray_info.get("tag_uid", "")
-            spool_tag = tray_uuid if tray_uuid and tray_uuid != "00000000000000000000000000000000" else tag_uid
-
-            if not spool_tag:
-                continue
-
-            # Find and update spool
-            spool = await client.find_spool_by_tag(spool_tag)
-            if spool:
-                result = await client.use_spool(spool["id"], partial_used_g)
-                if result:
-                    logger.info(
-                        f"[SPOOLMAN] Partial (linear): slot {slot_id}: {partial_used_g}g/{total_used_g}g -> spool {spool['id']}"
-                    )
-                    spools_updated += 1
-
-        if spools_updated > 0:
-            logger.info(f"[SPOOLMAN] Reported partial usage to {spools_updated} spool(s) using linear interpolation")
-
-
-async def _report_spoolman_usage(printer_id: int, archive_id: int, logger):
-    """Report filament usage to Spoolman after print completion.
-
-    Uses per-filament usage data captured at print start to report
-    usage to the correct spools.
-    """
-    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(f"[SPOOLMAN] No tracking data for print (printer={printer_id}, archive={archive_id})")
-            return
-
-        filament_usage = tracking.filament_usage or []
-        ams_trays = tracking.ams_trays or {}
-        slot_to_tray = tracking.slot_to_tray
-
-        # Delete tracking row (we're done with it)
-        await db.delete(tracking)
-        await db.commit()
-
-        if not filament_usage:
-            logger.debug(f"[SPOOLMAN] No filament usage data for archive {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
-
-        # Get Spoolman client
-        client = await get_spoolman_client()
-        if not client:
-            spoolman_url = await get_setting(db, "spoolman_url")
-            if spoolman_url:
-                client = await init_spoolman_client(spoolman_url)
-
-        if not client or not await client.health_check():
-            logger.warning("[SPOOLMAN] Not reachable for usage reporting")
-            return
-
-        logger.info(f"[SPOOLMAN] Reporting per-filament usage for archive {archive_id}")
-
-        # Convert ams_trays keys from string to int (JSON serialization converts int keys to strings)
-        ams_trays = {int(k): v for k, v in ams_trays.items()}
-
-        spools_updated = 0
-        for usage in filament_usage:
-            slot_id = usage.get("slot_id", 0)
-            used_g = usage.get("used_g", 0)
-
-            if used_g <= 0:
-                continue
-
-            # Determine which tray was used for this slot
-            # Default: slot_id 1 -> global_tray_id 0, etc.
-            global_tray_id = slot_id - 1
-
-            # Apply custom mapping if provided
-            if slot_to_tray and slot_id <= len(slot_to_tray):
-                mapped_tray = slot_to_tray[slot_id - 1]
-                if mapped_tray >= 0:  # -1 = unmapped
-                    global_tray_id = mapped_tray
-
-            tray_info = ams_trays.get(global_tray_id)
-            if not tray_info:
-                logger.debug(f"[SPOOLMAN] Slot {slot_id}: no tray at global_tray_id {global_tray_id}")
-                continue
-
-            # Get spool identifier (prefer tray_uuid over tag_uid)
-            tray_uuid = tray_info.get("tray_uuid", "")
-            tag_uid = tray_info.get("tag_uid", "")
-            spool_tag = tray_uuid if tray_uuid and tray_uuid != "00000000000000000000000000000000" else tag_uid
-
-            if not spool_tag:
-                logger.debug(f"[SPOOLMAN] Slot {slot_id}: no identifier for tray {global_tray_id}")
-                continue
-
-            # Find and update spool
-            spool = await client.find_spool_by_tag(spool_tag)
-            if spool:
-                spool_result = await client.use_spool(spool["id"], used_g)
-                if spool_result:
-                    logger.info(f"[SPOOLMAN] Slot {slot_id}: {used_g}g -> spool {spool['id']} (tray {global_tray_id})")
-                    spools_updated += 1
-            else:
-                logger.debug(f"[SPOOLMAN] Slot {slot_id}: no spool for tag {spool_tag[:16]}...")
-
-        if spools_updated == 0:
-            logger.info(f"[SPOOLMAN] Archive {archive_id}: no spools updated")
-        else:
-            logger.info(f"[SPOOLMAN] Archive {archive_id}: updated {spools_updated} spool(s)")
-
-
 async def on_printer_status_change(printer_id: int, state: PrinterState):
     """Handle printer status changes - broadcast via WebSocket."""
     # Only broadcast if something meaningful changed (reduce WebSocket spam)
@@ -1394,7 +938,7 @@ async def on_print_start(printer_id: int, data: dict):
 
                 # Store Spoolman tracking data for per-filament usage reporting
                 try:
-                    await _store_spoolman_print_data(printer_id, archive.id, archive.file_path, db, logger)
+                    await _store_spoolman_print_data(printer_id, archive.id, archive.file_path, db, printer_manager)
                 except Exception as e:
                     logger.warning(f"[SPOOLMAN] Failed to store tracking data: {e}")
 
@@ -1695,7 +1239,7 @@ async def on_print_start(printer_id: int, data: dict):
                 # Store Spoolman tracking data (may not work for fallback since no 3MF)
                 try:
                     await _store_spoolman_print_data(
-                        printer_id, fallback_archive.id, fallback_archive.file_path, db, logger
+                        printer_id, fallback_archive.id, fallback_archive.file_path, db, printer_manager
                     )
                 except Exception as e:
                     logger.debug(f"[SPOOLMAN] Could not store tracking for fallback archive: {e}")
@@ -1814,7 +1358,7 @@ async def on_print_start(printer_id: int, data: dict):
 
                 # Store Spoolman tracking data for per-filament usage reporting
                 try:
-                    await _store_spoolman_print_data(printer_id, archive.id, archive.file_path, db, logger)
+                    await _store_spoolman_print_data(printer_id, archive.id, archive.file_path, db, printer_manager)
                 except Exception as e:
                     logger.warning(f"[SPOOLMAN] Failed to store tracking data: {e}")
         finally:
@@ -2164,7 +1708,7 @@ async def on_print_complete(printer_id: int, data: dict):
     # Report filament usage to Spoolman if print completed successfully
     if data.get("status") == "completed":
         try:
-            await _report_spoolman_usage(printer_id, archive_id, logger)
+            await _report_spoolman_usage(printer_id, archive_id)
             log_timing("Spoolman usage report")
         except Exception as e:
             logger.warning(f"Spoolman usage reporting failed: {e}")
@@ -2172,7 +1716,7 @@ async def on_print_complete(printer_id: int, data: dict):
         # Report partial usage if tracking data exists (only stored when weight sync is disabled)
         try:
             async with async_session() as db:
-                await _cleanup_spoolman_tracking(printer_id, archive_id, db, logger)
+                await _cleanup_spoolman_tracking(printer_id, archive_id, db)
         except Exception as e:
             logger.debug(f"[SPOOLMAN] Cleanup failed: {e}")
 

backend/app/services/spoolman_tracking.py

@@ -0,0 +1,442 @@
+"""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 get_spoolman_client, init_spoolman_client
+
+logger = logging.getLogger(__name__)
+
+# Zero UUID used by Bambu printers for empty/unset tray_uuid
+_ZERO_UUID = "00000000000000000000000000000000"
+
+
+def _resolve_spool_tag(tray_info: dict) -> 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 = tray_info.get("tray_uuid", "")
+    tag_uid = tray_info.get("tag_uid", "")
+    if tray_uuid and tray_uuid != _ZERO_UUID:
+        return tray_uuid
+    return tag_uid
+
+
+def _resolve_global_tray_id(slot_id: int, slot_to_tray: list | None) -> int:
+    """Map a 1-based slot_id to a global_tray_id using optional custom mapping.
+
+    Default mapping: slot 1 -> tray 0, slot 2 -> tray 1, etc.
+    Custom mapping (from print queue): slot_to_tray[slot_id - 1] overrides default.
+    A value of -1 in custom mapping means unmapped (uses default).
+    """
+    global_tray_id = slot_id - 1
+    if slot_to_tray and slot_id <= len(slot_to_tray):
+        mapped_tray = slot_to_tray[slot_id - 1]
+        if mapped_tray >= 0:
+            global_tray_id = mapped_tray
+    return global_tray_id
+
+
+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 = ams_unit.get("id", 0)
+        for tray in ams_unit.get("tray", []):
+            tray_id = tray.get("id", 0)
+            global_tray_id = 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 (global_tray_id = 254)
+    vt_tray = raw_data.get("vt_tray")
+    if vt_tray and vt_tray.get("tray_type"):
+        lookup[254] = {
+            "tray_uuid": vt_tray.get("tray_uuid", ""),
+            "tag_uid": vt_tray.get("tag_uid", ""),
+            "tray_type": vt_tray.get("tray_type", ""),
+        }
+
+    return lookup
+
+
+async def store_print_data(printer_id: int, archive_id: int, file_path: str, db, printer_manager):
+    """Store Spoolman tracking data at print start (persisted to database).
+
+    Only stores data when Spoolman is enabled and AMS weight sync is disabled
+    (i.e., we're using per-usage tracking instead of AMS percentage estimates).
+    """
+    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
+
+    # Only store tracking data if "Disable AMS Weight Sync" is enabled
+    disable_weight_sync_str = await get_setting(db, "spoolman_disable_weight_sync")
+    disable_weight_sync = disable_weight_sync_str and disable_weight_sync_str.lower() == "true"
+    if not disable_weight_sync:
+        logger.debug("[SPOOLMAN] Weight sync enabled, skipping per-usage tracking data storage")
+        return
+
+    # Get 3MF file path
+    full_path = app_settings.base_dir / file_path
+    if not full_path.exists():
+        logger.debug(f"[SPOOLMAN] 3MF file not found: {full_path}")
+        return
+
+    # Extract per-filament usage from 3MF (total usage per slot)
+    filament_usage = extract_filament_usage_from_3mf(full_path)
+    if not filament_usage:
+        logger.debug(f"[SPOOLMAN] No filament usage data in 3MF for archive {archive_id}")
+        return
+
+    # Get current AMS tray state
+    state = printer_manager.get_status(printer_id)
+    ams_trays = {}
+    if state and state.raw_data:
+        ams_trays = build_ams_tray_lookup(state.raw_data)
+
+    # Get custom slot-to-tray mapping from queue item (if this is a queued print)
+    slot_to_tray = None
+    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()
+    if queue_item and queue_item.ams_mapping:
+        try:
+            slot_to_tray = json.loads(queue_item.ams_mapping)
+        except json.JSONDecodeError:
+            pass
+
+    # Parse G-code for per-layer filament usage (for accurate partial usage tracking)
+    layer_usage = extract_layer_filament_usage_from_3mf(full_path)
+    layer_usage_json = None
+    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(f"[SPOOLMAN] Parsed {len(layer_usage)} layers from G-code")
+
+    # Extract filament properties (density, diameter) for mm -> grams conversion
+    filament_properties = extract_filament_properties_from_3mf(full_path)
+
+    # 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
+    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,
+    )
+    db.add(tracking)
+    await db.commit()
+
+    logger.info(f"[SPOOLMAN] Stored tracking data for print: printer={printer_id}, archive={archive_id}")
+    logger.debug(f"[SPOOLMAN] Filament usage: {filament_usage}")
+    logger.debug(f"[SPOOLMAN] AMS trays: {list(ams_trays.keys())}")
+    if slot_to_tray:
+        logger.debug(f"[SPOOLMAN] Custom slot mapping: {slot_to_tray}")
+    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):
+    """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(f"[SPOOLMAN] No tracking data to clean up for printer={printer_id}, archive={archive_id}")
+        return
+
+    # Try to report partial usage before cleanup
+    try:
+        await _report_partial_usage(printer_id, tracking)
+    except Exception as e:
+        logger.warning(f"[SPOOLMAN] Partial usage report failed: {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(f"[SPOOLMAN] Cleaned up tracking data for printer={printer_id}, archive={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:
+                client = await init_spoolman_client(spoolman_url)
+
+    if not client or not await client.health_check():
+        return None
+
+    return client
+
+
+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,
+) -> int:
+    """Report usage to Spoolman for a list of (slot_id, grams) pairs.
+
+    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)
+        tray_info = ams_trays.get(global_tray_id)
+        if not tray_info:
+            logger.debug(f"[SPOOLMAN] Slot {slot_id}: no tray at global_tray_id {global_tray_id}")
+            continue
+
+        spool_tag = _resolve_spool_tag(tray_info)
+        if not spool_tag:
+            logger.debug(f"[SPOOLMAN] Slot {slot_id}: no identifier for tray {global_tray_id}")
+            continue
+
+        spool = await client.find_spool_by_tag(spool_tag)
+        if not spool:
+            logger.debug(f"[SPOOLMAN] Slot {slot_id}: no spool for tag {spool_tag[:16]}...")
+            continue
+
+        result = await client.use_spool(spool["id"], grams_used)
+        if result:
+            logger.info(f"[SPOOLMAN] {method_label}: slot {slot_id}: {grams_used}g -> spool {spool['id']}")
+            spools_updated += 1
+
+    return spools_updated
+
+
+async def _report_partial_usage(printer_id: int, tracking):
+    """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
+    state = printer_manager.get_status(printer_id)
+    if not state:
+        logger.debug("[SPOOLMAN] No printer state available for partial usage")
+        return
+
+    current_layer = state.layer_num
+    total_layers = state.total_layers
+
+    if not current_layer or current_layer <= 0:
+        logger.debug("[SPOOLMAN] No progress to report (layer 0 or unknown)")
+        return
+
+    logger.info(f"[SPOOLMAN] Reporting partial usage at layer {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
+
+    client = await _get_spoolman_client_with_fallback()
+    if not client:
+        logger.warning("[SPOOLMAN] Not reachable for partial usage reporting")
+        return
+
+    # 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(f"[SPOOLMAN] Using G-code parsed data for layer {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)
+                tray_info = ams_trays.get(global_tray_id)
+                density = None
+                diameter = 1.75
+
+                if tray_info:
+                    spool_tag = _resolve_spool_tag(tray_info)
+                    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(f"[SPOOLMAN] Using fallback density {density} for slot {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)"
+            )
+            if spools_updated > 0:
+                logger.info(f"[SPOOLMAN] Reported partial usage to {spools_updated} spool(s) using G-code data")
+            return
+
+    # Fallback: linear interpolation (if no G-code data available)
+    if not total_layers or total_layers <= 0:
+        logger.debug(f"[SPOOLMAN] Cannot use linear fallback: total_layers={total_layers}")
+        return
+
+    progress_ratio = min(current_layer / total_layers, 1.0)
+    logger.info(f"[SPOOLMAN] Falling back to linear interpolation ({progress_ratio:.1%})")
+
+    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)"
+    )
+    if spools_updated > 0:
+        logger.info(f"[SPOOLMAN] Reported partial usage to {spools_updated} spool(s) using linear interpolation")
+
+
+async def report_usage(printer_id: int, archive_id: int):
+    """Report filament usage to Spoolman after print completion.
+
+    Uses per-filament usage data captured at print start to report
+    usage to the correct spools.
+    """
+    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(f"[SPOOLMAN] No tracking data for print (printer={printer_id}, archive={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
+
+        # Delete tracking row (we're done with it)
+        await db.delete(tracking)
+        await db.commit()
+
+        if not filament_usage:
+            logger.debug(f"[SPOOLMAN] No filament usage data for archive {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
+
+        logger.info(f"[SPOOLMAN] Reporting per-filament usage for archive {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}"
+        )
+
+        if spools_updated == 0:
+            logger.info(f"[SPOOLMAN] Archive {archive_id}: no spools updated")
+        else:
+            logger.info(f"[SPOOLMAN] Archive {archive_id}: updated {spools_updated} spool(s)")

backend/app/utils/threemf_tools.py

@@ -10,6 +10,7 @@ import math
 import re
 import zipfile
 from pathlib import Path
+
 import defusedxml.ElementTree as ET
 
 # Default filament properties

backend/tests/unit/services/test_spoolman_tracking.py

@@ -0,0 +1,122 @@
+"""Unit tests for Spoolman tracking service helpers."""
+
+import pytest
+
+from backend.app.services.spoolman_tracking import (
+    _resolve_global_tray_id,
+    _resolve_spool_tag,
+    build_ams_tray_lookup,
+)
+
+
+class TestResolveSpoolTag:
+    """Tests for _resolve_spool_tag()."""
+
+    def test_prefers_tray_uuid(self):
+        tray = {"tray_uuid": "A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4", "tag_uid": "DEADBEEF"}
+        assert _resolve_spool_tag(tray) == "A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4"
+
+    def test_falls_back_to_tag_uid(self):
+        tray = {"tray_uuid": "", "tag_uid": "DEADBEEF"}
+        assert _resolve_spool_tag(tray) == "DEADBEEF"
+
+    def test_skips_zero_uuid(self):
+        tray = {"tray_uuid": "00000000000000000000000000000000", "tag_uid": "DEADBEEF"}
+        assert _resolve_spool_tag(tray) == "DEADBEEF"
+
+    def test_empty_both(self):
+        tray = {"tray_uuid": "", "tag_uid": ""}
+        assert _resolve_spool_tag(tray) == ""
+
+    def test_missing_keys(self):
+        assert _resolve_spool_tag({}) == ""
+
+    def test_zero_uuid_no_tag(self):
+        tray = {"tray_uuid": "00000000000000000000000000000000", "tag_uid": ""}
+        assert _resolve_spool_tag(tray) == ""
+
+
+class TestResolveGlobalTrayId:
+    """Tests for _resolve_global_tray_id()."""
+
+    def test_default_mapping(self):
+        """slot 1 -> tray 0, slot 2 -> tray 1, etc."""
+        assert _resolve_global_tray_id(1, None) == 0
+        assert _resolve_global_tray_id(2, None) == 1
+        assert _resolve_global_tray_id(4, None) == 3
+
+    def test_custom_mapping(self):
+        """Custom slot_to_tray overrides default."""
+        mapping = [5, 2, -1, 0]
+        assert _resolve_global_tray_id(1, mapping) == 5
+        assert _resolve_global_tray_id(2, mapping) == 2
+        assert _resolve_global_tray_id(4, mapping) == 0
+
+    def test_unmapped_slot(self):
+        """Slot with -1 in mapping uses default."""
+        mapping = [5, -1, 2, 0]
+        assert _resolve_global_tray_id(2, mapping) == 1  # default: slot 2 -> tray 1
+
+    def test_slot_beyond_mapping(self):
+        """Slot beyond mapping length uses default."""
+        mapping = [5, 2]
+        assert _resolve_global_tray_id(3, mapping) == 2  # default: slot 3 -> tray 2
+
+    def test_empty_mapping(self):
+        mapping = []
+        assert _resolve_global_tray_id(1, mapping) == 0
+
+
+class TestBuildAmsTrayLookup:
+    """Tests for build_ams_tray_lookup()."""
+
+    def test_single_ams_unit(self):
+        raw = {
+            "ams": [
+                {
+                    "id": 0,
+                    "tray": [
+                        {"id": 0, "tray_uuid": "AAA", "tag_uid": "111", "tray_type": "PLA"},
+                        {"id": 1, "tray_uuid": "BBB", "tag_uid": "222", "tray_type": "ABS"},
+                    ],
+                }
+            ]
+        }
+        lookup = build_ams_tray_lookup(raw)
+        assert lookup[0] == {"tray_uuid": "AAA", "tag_uid": "111", "tray_type": "PLA"}
+        assert lookup[1] == {"tray_uuid": "BBB", "tag_uid": "222", "tray_type": "ABS"}
+
+    def test_multiple_ams_units(self):
+        raw = {
+            "ams": [
+                {"id": 0, "tray": [{"id": 0, "tray_uuid": "A", "tag_uid": "", "tray_type": "PLA"}]},
+                {"id": 1, "tray": [{"id": 0, "tray_uuid": "B", "tag_uid": "", "tray_type": "PETG"}]},
+            ]
+        }
+        lookup = build_ams_tray_lookup(raw)
+        assert 0 in lookup  # AMS 0, tray 0
+        assert 4 in lookup  # AMS 1, tray 0 (1*4+0)
+        assert lookup[4]["tray_uuid"] == "B"
+
+    def test_external_spool(self):
+        raw = {
+            "ams": [],
+            "vt_tray": {"tray_uuid": "EXT", "tag_uid": "X", "tray_type": "TPU"},
+        }
+        lookup = build_ams_tray_lookup(raw)
+        assert 254 in lookup
+        assert lookup[254]["tray_type"] == "TPU"
+
+    def test_empty_external_spool_skipped(self):
+        raw = {"ams": [], "vt_tray": {"tray_type": ""}}
+        lookup = build_ams_tray_lookup(raw)
+        assert 254 not in lookup
+
+    def test_no_ams_data(self):
+        assert build_ams_tray_lookup({}) == {}
+        assert build_ams_tray_lookup({"ams": []}) == {}
+
+    def test_missing_fields_default(self):
+        raw = {"ams": [{"id": 0, "tray": [{"id": 0}]}]}
+        lookup = build_ams_tray_lookup(raw)
+        assert lookup[0] == {"tray_uuid": "", "tag_uid": "", "tray_type": ""}

backend/tests/unit/test_threemf_tools.py

@@ -0,0 +1,251 @@
+"""Unit tests for 3MF parsing utilities (threemf_tools.py).
+
+Tests G-code parsing, filament length-to-weight conversion,
+and cumulative layer usage lookup.
+"""
+
+import math
+
+import pytest
+
+from backend.app.utils.threemf_tools import (
+    get_cumulative_usage_at_layer,
+    mm_to_grams,
+    parse_gcode_layer_filament_usage,
+)
+
+
+class TestParseGcodeLayerFilamentUsage:
+    """Tests for parse_gcode_layer_filament_usage()."""
+
+    def test_single_filament_single_layer(self):
+        """Single filament extruding on one layer."""
+        gcode = """
+M620 S0
+G1 X10 Y10 E5.0
+G1 X20 Y20 E3.0
+"""
+        result = parse_gcode_layer_filament_usage(gcode)
+        assert result == {0: {0: 8.0}}
+
+    def test_multi_layer_single_filament(self):
+        """Single filament across multiple layers."""
+        gcode = """
+M620 S0
+G1 X10 Y10 E10.0
+M73 L1
+G1 X20 Y20 E5.0
+M73 L2
+G1 X30 Y30 E7.0
+"""
+        result = parse_gcode_layer_filament_usage(gcode)
+        assert result[0] == {0: 10.0}
+        assert result[1] == {0: 15.0}
+        assert result[2] == {0: 22.0}
+
+    def test_multi_material(self):
+        """Multiple filaments switching via M620."""
+        gcode = """
+M620 S0
+G1 E10.0
+M73 L1
+M620 S1
+G1 E5.0
+M620 S0
+G1 E3.0
+M73 L2
+G1 E2.0
+"""
+        result = parse_gcode_layer_filament_usage(gcode)
+        # Layer 0: filament 0 = 10mm
+        assert result[0] == {0: 10.0}
+        # Layer 1: filament 0 = 13mm (10+3), filament 1 = 5mm
+        assert result[1] == {0: 13.0, 1: 5.0}
+        # Layer 2: filament 0 = 15mm (13+2)
+        assert result[2] == {0: 15.0, 1: 5.0}
+
+    def test_retractions_ignored(self):
+        """Negative E values (retractions) should be ignored."""
+        gcode = """
+M620 S0
+G1 E10.0
+G1 E-2.0
+G1 E5.0
+"""
+        result = parse_gcode_layer_filament_usage(gcode)
+        assert result == {0: {0: 15.0}}
+
+    def test_m620_s255_unloads(self):
+        """M620 S255 means unload - extrusion after should be ignored."""
+        gcode = """
+M620 S0
+G1 E10.0
+M620 S255
+G1 E5.0
+"""
+        result = parse_gcode_layer_filament_usage(gcode)
+        assert result == {0: {0: 10.0}}
+
+    def test_m620_with_suffix(self):
+        """M620 S0A format (filament ID with suffix letter)."""
+        gcode = """
+M620 S0A
+G1 E10.0
+M620 S1A
+G1 E5.0
+"""
+        result = parse_gcode_layer_filament_usage(gcode)
+        assert result == {0: {0: 10.0, 1: 5.0}}
+
+    def test_comments_ignored(self):
+        """Comment lines and inline comments are ignored."""
+        gcode = """
+; This is a comment
+M620 S0
+G1 X10 E5.0 ; inline comment with E value
+G1 E3.0
+"""
+        result = parse_gcode_layer_filament_usage(gcode)
+        assert result == {0: {0: 8.0}}
+
+    def test_empty_gcode(self):
+        """Empty G-code returns empty dict."""
+        assert parse_gcode_layer_filament_usage("") == {}
+        assert parse_gcode_layer_filament_usage("\n\n\n") == {}
+
+    def test_no_extrusion(self):
+        """G-code with moves but no extrusion."""
+        gcode = """
+G1 X10 Y10
+G1 X20 Y20
+"""
+        assert parse_gcode_layer_filament_usage(gcode) == {}
+
+    def test_no_active_filament_extrusion_ignored(self):
+        """Extrusion before any M620 is ignored (no active filament)."""
+        gcode = """
+G1 E10.0
+M620 S0
+G1 E5.0
+"""
+        result = parse_gcode_layer_filament_usage(gcode)
+        assert result == {0: {0: 5.0}}
+
+    def test_g0_g2_g3_extrusion(self):
+        """G0, G2, G3 with E parameter are also tracked."""
+        gcode = """
+M620 S0
+G0 E1.0
+G1 E2.0
+G2 E3.0
+G3 E4.0
+"""
+        result = parse_gcode_layer_filament_usage(gcode)
+        assert result == {0: {0: 10.0}}
+
+    def test_cumulative_across_layers(self):
+        """Values are cumulative, not per-layer."""
+        gcode = """
+M620 S0
+G1 E100.0
+M73 L1
+G1 E100.0
+M73 L2
+G1 E100.0
+"""
+        result = parse_gcode_layer_filament_usage(gcode)
+        assert result[0] == {0: 100.0}
+        assert result[1] == {0: 200.0}
+        assert result[2] == {0: 300.0}
+
+
+class TestMmToGrams:
+    """Tests for mm_to_grams()."""
+
+    def test_default_pla_175(self):
+        """Default PLA 1.75mm conversion."""
+        # 1000mm of 1.75mm PLA at 1.24 g/cm³
+        # Volume = π × (0.0875cm)² × 100cm = 2.405cm³
+        # Weight = 2.405 × 1.24 = 2.982g
+        result = mm_to_grams(1000.0)
+        expected = math.pi * (0.0875**2) * 100 * 1.24
+        assert abs(result - expected) < 0.001
+
+    def test_zero_length(self):
+        """Zero length returns zero weight."""
+        assert mm_to_grams(0.0) == 0.0
+
+    def test_custom_diameter(self):
+        """Custom diameter (2.85mm) changes result."""
+        result_175 = mm_to_grams(1000.0, diameter_mm=1.75)
+        result_285 = mm_to_grams(1000.0, diameter_mm=2.85)
+        # 2.85mm filament has more volume per mm
+        assert result_285 > result_175
+        ratio = (2.85 / 1.75) ** 2  # Volume scales with diameter²
+        assert abs(result_285 / result_175 - ratio) < 0.001
+
+    def test_custom_density(self):
+        """Different density (ABS vs PLA)."""
+        pla = mm_to_grams(1000.0, density_g_cm3=1.24)
+        abs_ = mm_to_grams(1000.0, density_g_cm3=1.04)
+        assert pla > abs_
+        assert abs(pla / abs_ - 1.24 / 1.04) < 0.001
+
+    def test_known_value(self):
+        """Verify against a known calculation.
+
+        1m (1000mm) of 1.75mm PLA at 1.24 g/cm³:
+        r = 0.0875 cm, L = 100 cm
+        V = π × 0.0875² × 100 = 2.4053 cm³
+        m = 2.4053 × 1.24 = 2.9826 g
+        """
+        result = mm_to_grams(1000.0, 1.75, 1.24)
+        assert abs(result - 2.9826) < 0.01
+
+
+class TestGetCumulativeUsageAtLayer:
+    """Tests for get_cumulative_usage_at_layer()."""
+
+    def test_exact_layer_match(self):
+        """Target layer exists exactly in the data."""
+        data = {0: {0: 100.0}, 5: {0: 500.0}, 10: {0: 1000.0}}
+        assert get_cumulative_usage_at_layer(data, 5) == {0: 500.0}
+
+    def test_between_layers(self):
+        """Target is between recorded layers - uses the closest lower one."""
+        data = {0: {0: 100.0}, 5: {0: 500.0}, 10: {0: 1000.0}}
+        # Layer 7 is between 5 and 10, should return layer 5's data
+        assert get_cumulative_usage_at_layer(data, 7) == {0: 500.0}
+
+    def test_beyond_last_layer(self):
+        """Target is beyond the last recorded layer."""
+        data = {0: {0: 100.0}, 5: {0: 500.0}}
+        assert get_cumulative_usage_at_layer(data, 100) == {0: 500.0}
+
+    def test_before_first_layer(self):
+        """Target is before any recorded data."""
+        data = {5: {0: 500.0}, 10: {0: 1000.0}}
+        assert get_cumulative_usage_at_layer(data, 3) == {}
+
+    def test_empty_data(self):
+        """Empty layer_usage returns empty dict."""
+        assert get_cumulative_usage_at_layer({}, 5) == {}
+
+    def test_none_data(self):
+        """None layer_usage returns empty dict."""
+        assert get_cumulative_usage_at_layer(None, 5) == {}
+
+    def test_multi_filament(self):
+        """Multi-filament data at target layer."""
+        data = {
+            0: {0: 50.0},
+            5: {0: 200.0, 1: 100.0},
+            10: {0: 400.0, 1: 250.0, 2: 50.0},
+        }
+        result = get_cumulative_usage_at_layer(data, 8)
+        assert result == {0: 200.0, 1: 100.0}
+
+    def test_layer_zero(self):
+        """Target layer 0."""
+        data = {0: {0: 10.0}, 1: {0: 20.0}}
+        assert get_cumulative_usage_at_layer(data, 0) == {0: 10.0}

frontend/src/__tests__/components/SpoolmanSettings.test.tsx

@@ -41,11 +41,15 @@ describe('SpoolmanSettings', () => {
       spoolman_enabled: 'false',
       spoolman_url: '',
       spoolman_sync_mode: 'auto',
+      spoolman_disable_weight_sync: 'false',
+      spoolman_report_partial_usage: 'true',
     });
     vi.mocked(api.updateSpoolmanSettings).mockResolvedValue({
       spoolman_enabled: 'false',
       spoolman_url: '',
       spoolman_sync_mode: 'auto',
+      spoolman_disable_weight_sync: 'false',
+      spoolman_report_partial_usage: 'true',
     });
     vi.mocked(api.getSpoolmanStatus).mockResolvedValue({
       enabled: false,
@@ -155,11 +159,15 @@ describe('SpoolmanSettings', () => {
         spoolman_enabled: 'true',
         spoolman_url: 'http://localhost:7912',
         spoolman_sync_mode: 'auto',
+        spoolman_disable_weight_sync: 'false',
+        spoolman_report_partial_usage: 'true',
       });
       vi.mocked(api.updateSpoolmanSettings).mockResolvedValue({
         spoolman_enabled: 'true',
         spoolman_url: 'http://localhost:7912',
         spoolman_sync_mode: 'auto',
+        spoolman_disable_weight_sync: 'false',
+        spoolman_report_partial_usage: 'true',
       });
     });
 
@@ -253,6 +261,87 @@ describe('SpoolmanSettings', () => {
     });
   });
 
+  describe('weight sync toggle', () => {
+    it('shows weight sync toggle when sync mode is auto and enabled', async () => {
+      vi.mocked(api.getSpoolmanSettings).mockResolvedValue({
+        spoolman_enabled: 'true',
+        spoolman_url: 'http://localhost:7912',
+        spoolman_sync_mode: 'auto',
+        spoolman_disable_weight_sync: 'false',
+        spoolman_report_partial_usage: 'true',
+      });
+
+      render(<SpoolmanSettings />);
+
+      await waitFor(() => {
+        expect(screen.getByText('Disable AMS Estimated Weight Sync')).toBeInTheDocument();
+      });
+    });
+
+    it('does not show weight sync toggle when sync mode is manual', async () => {
+      vi.mocked(api.getSpoolmanSettings).mockResolvedValue({
+        spoolman_enabled: 'true',
+        spoolman_url: 'http://localhost:7912',
+        spoolman_sync_mode: 'manual',
+        spoolman_disable_weight_sync: 'false',
+        spoolman_report_partial_usage: 'true',
+      });
+
+      render(<SpoolmanSettings />);
+
+      await waitFor(() => {
+        expect(screen.getByText('Spoolman Integration')).toBeInTheDocument();
+      });
+
+      expect(screen.queryByText('Disable AMS Estimated Weight Sync')).not.toBeInTheDocument();
+    });
+
+    it('shows weight sync toggle in disabled state when sync mode is auto', async () => {
+      render(<SpoolmanSettings />);
+
+      await waitFor(() => {
+        // Toggle label is visible since sync mode defaults to auto
+        expect(screen.getByText('Disable AMS Estimated Weight Sync')).toBeInTheDocument();
+      });
+    });
+  });
+
+  describe('partial usage toggle', () => {
+    it('shows partial usage toggle when weight sync is disabled', async () => {
+      vi.mocked(api.getSpoolmanSettings).mockResolvedValue({
+        spoolman_enabled: 'true',
+        spoolman_url: 'http://localhost:7912',
+        spoolman_sync_mode: 'auto',
+        spoolman_disable_weight_sync: 'true',
+        spoolman_report_partial_usage: 'true',
+      });
+
+      render(<SpoolmanSettings />);
+
+      await waitFor(() => {
+        expect(screen.getByText('Report Partial Usage for Failed Prints')).toBeInTheDocument();
+      });
+    });
+
+    it('does not show partial usage toggle when weight sync is enabled', async () => {
+      vi.mocked(api.getSpoolmanSettings).mockResolvedValue({
+        spoolman_enabled: 'true',
+        spoolman_url: 'http://localhost:7912',
+        spoolman_sync_mode: 'auto',
+        spoolman_disable_weight_sync: 'false',
+        spoolman_report_partial_usage: 'true',
+      });
+
+      render(<SpoolmanSettings />);
+
+      await waitFor(() => {
+        expect(screen.getByText('Spoolman Integration')).toBeInTheDocument();
+      });
+
+      expect(screen.queryByText('Report Partial Usage for Failed Prints')).not.toBeInTheDocument();
+    });
+  });
+
   describe('sync mode options', () => {
     it('shows Automatic option', async () => {
       render(<SpoolmanSettings />);

frontend/src/components/SpoolmanSettings.tsx

@@ -1,4 +1,5 @@
 import { useState, useEffect } from 'react';
+import { useTranslation } from 'react-i18next';
 import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query';
 import { Loader2, Check, X, RefreshCw, Link2, Link2Off, Database, ChevronDown, Info, AlertTriangle } from 'lucide-react';
 import { api } from '../api/client';
@@ -7,6 +8,7 @@ import { Card, CardContent, CardHeader } from './Card';
 import { Button } from './Button';
 
 export function SpoolmanSettings() {
+  const { t } = useTranslation();
   const queryClient = useQueryClient();
   const [localEnabled, setLocalEnabled] = useState(false);
   const [localUrl, setLocalUrl] = useState('');
@@ -258,11 +260,9 @@ export function SpoolmanSettings() {
         {localSyncMode === 'auto' && (
           <div className="flex items-center justify-between">
             <div>
-              <p className="text-white">Disable AMS Estimated Weight Sync</p>
+              <p className="text-white">{t('spoolman.disableWeightSync')}</p>
               <p className="text-sm text-bambu-gray">
-                Don't update remaining capacity from AMS estimates. Use this if you prefer
-                Spoolman's usage tracking over AMS percentage-based estimates. New spools
-                will still use the AMS estimate as their initial weight.
+                {t('spoolman.disableWeightSyncDesc')}
               </p>
             </div>
             <label className="relative inline-flex items-center cursor-pointer">
@@ -282,10 +282,9 @@ export function SpoolmanSettings() {
         {localDisableWeightSync && (
           <div className="flex items-center justify-between">
             <div>
-              <p className="text-white">Report Partial Usage for Failed Prints</p>
+              <p className="text-white">{t('spoolman.reportPartialUsage')}</p>
               <p className="text-sm text-bambu-gray">
-                When a print fails or is cancelled, report the estimated filament used
-                up to that point based on layer progress.
+                {t('spoolman.reportPartialUsageDesc')}
               </p>
             </div>
             <label className="relative inline-flex items-center cursor-pointer">

frontend/src/i18n/locales/de.ts

@@ -2181,6 +2181,10 @@ export default {
     spoolId: 'Spulen-ID',
     weight: 'Gewicht',
     remaining: 'Verbleibend',
+    disableWeightSync: 'AMS-Gewichtsschätzung deaktivieren',
+    disableWeightSyncDesc: 'Verbleibende Kapazität nicht aus AMS-Schätzungen aktualisieren. Verwenden Sie dies, wenn Sie die Verbrauchserfassung von Spoolman gegenüber den prozentualen AMS-Schätzungen bevorzugen. Neue Spulen verwenden weiterhin die AMS-Schätzung als Anfangsgewicht.',
+    reportPartialUsage: 'Teilverbrauch bei fehlgeschlagenen Drucken melden',
+    reportPartialUsageDesc: 'Wenn ein Druck fehlschlägt oder abgebrochen wird, den geschätzten Filamentverbrauch bis zu diesem Zeitpunkt basierend auf dem Schichtfortschritt melden.',
   },
 
   // Timelapse

frontend/src/i18n/locales/en.ts

@@ -2181,6 +2181,10 @@ export default {
     spoolId: 'Spool ID',
     weight: 'Weight',
     remaining: 'Remaining',
+    disableWeightSync: 'Disable AMS Estimated Weight Sync',
+    disableWeightSyncDesc: "Don't update remaining capacity from AMS estimates. Use this if you prefer Spoolman's usage tracking over AMS percentage-based estimates. New spools will still use the AMS estimate as their initial weight.",
+    reportPartialUsage: 'Report Partial Usage for Failed Prints',
+    reportPartialUsageDesc: 'When a print fails or is cancelled, report the estimated filament used up to that point based on layer progress.',
   },
 
   // Timelapse

frontend/src/i18n/locales/ja.ts

@@ -1119,6 +1119,10 @@ export default {
       linkSpool: 'スプールを連携',
       linkTooltip: 'このスプールをSpoolmanスプールに連携',
       noUnlinked: '未連携のスプールがありません',
+      disableWeightSync: 'AMS推定重量同期を無効化',
+      disableWeightSyncDesc: 'AMS推定値から残量を更新しません。AMSの割合ベースの推定よりもSpoolmanの使用量追跡を優先する場合に使用してください。新しいスプールは引き続きAMS推定値を初期重量として使用します。',
+      reportPartialUsage: '失敗した印刷の部分使用量を報告',
+      reportPartialUsageDesc: '印刷が失敗またはキャンセルされた場合、レイヤー進捗に基づいてその時点までの推定フィラメント使用量を報告します。',
     },
 
     // Page

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-DJ147FSl.js"></script>
+    <script type="module" crossorigin src="/assets/index-AXQRHtw2.js"></script>
     <link rel="stylesheet" crossorigin href="/assets/index-togsBDt6.css">
   </head>
   <body>