Add per-filament Spoolman usage tracking with G-code parsing

Implement accurate per-filament usage tracking for Spoolman integration,
similar to OpenSpoolman v0.3.0. This replaces the previous single-spool
reporting with multi-material aware tracking.

Features:
- Parse G-code from 3MF files at print start to build per-layer,
  per-filament cumulative extrusion maps
- Store tracking data in new `active_print_spoolman` database table
  (survives server restarts for long prints)
- Report accurate partial usage when prints fail/cancel based on
  actual layer progress and G-code data
- Add "Disable AMS Weight Sync" setting to prevent AMS percentage-based
  weight estimates from overwriting Spoolman's granular tracking
- Add "Report Partial Usage for Failed Prints" toggle (only shown when
  weight sync is disabled)
- Use Spoolman's filament density instead of defaults for mm-to-grams
  conversion
- Prefer tray_uuid over tag_uid for spool identification

backend/app/api/routes/settings.py

@@ -64,6 +64,8 @@ async def get_settings(
                 "save_thumbnails",
                 "capture_finish_photo",
                 "spoolman_enabled",
+                "spoolman_disable_weight_sync",
+                "spoolman_report_partial_usage",
                 "check_updates",
                 "check_printer_firmware",
                 "virtual_printer_enabled",
@@ -207,11 +209,15 @@ async def get_spoolman_settings(
     spoolman_enabled = await get_setting(db, "spoolman_enabled") or "false"
     spoolman_url = await get_setting(db, "spoolman_url") or ""
     spoolman_sync_mode = await get_setting(db, "spoolman_sync_mode") or "auto"
+    spoolman_disable_weight_sync = await get_setting(db, "spoolman_disable_weight_sync") or "false"
+    spoolman_report_partial_usage = await get_setting(db, "spoolman_report_partial_usage") or "true"
 
     return {
         "spoolman_enabled": spoolman_enabled,
         "spoolman_url": spoolman_url,
         "spoolman_sync_mode": spoolman_sync_mode,
+        "spoolman_disable_weight_sync": spoolman_disable_weight_sync,
+        "spoolman_report_partial_usage": spoolman_report_partial_usage,
     }
 
 
@@ -228,6 +234,10 @@ async def update_spoolman_settings(
         await set_setting(db, "spoolman_url", settings["spoolman_url"])
     if "spoolman_sync_mode" in settings:
         await set_setting(db, "spoolman_sync_mode", settings["spoolman_sync_mode"])
+    if "spoolman_disable_weight_sync" in settings:
+        await set_setting(db, "spoolman_disable_weight_sync", settings["spoolman_disable_weight_sync"])
+    if "spoolman_report_partial_usage" in settings:
+        await set_setting(db, "spoolman_report_partial_usage", settings["spoolman_report_partial_usage"])
 
     await db.commit()
     db.expire_all()

backend/app/api/routes/spoolman.py

@@ -52,15 +52,16 @@ class SyncResult(BaseModel):
     errors: list[str]
 
 
-async def get_spoolman_settings(db: AsyncSession) -> tuple[bool, str, str]:
+async def get_spoolman_settings(db: AsyncSession) -> tuple[bool, str, str, bool]:
     """Get Spoolman settings from database.
 
     Returns:
-        Tuple of (enabled, url, sync_mode)
+        Tuple of (enabled, url, sync_mode, disable_weight_sync)
     """
     enabled = False
     url = ""
     sync_mode = "auto"
+    disable_weight_sync = False
 
     result = await db.execute(select(Settings))
     for setting in result.scalars().all():
@@ -70,8 +71,10 @@ async def get_spoolman_settings(db: AsyncSession) -> tuple[bool, str, str]:
             url = setting.value
         elif setting.key == "spoolman_sync_mode":
             sync_mode = setting.value
+        elif setting.key == "spoolman_disable_weight_sync":
+            disable_weight_sync = setting.value.lower() == "true"
 
-    return enabled, url, sync_mode
+    return enabled, url, sync_mode, disable_weight_sync
 
 
 @router.get("/status", response_model=SpoolmanStatus)
@@ -80,7 +83,7 @@ async def get_spoolman_status(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """Get Spoolman integration status."""
-    enabled, url, _ = await get_spoolman_settings(db)
+    enabled, url, _, _ = await get_spoolman_settings(db)
 
     client = await get_spoolman_client()
     connected = False
@@ -100,7 +103,7 @@ async def connect_spoolman(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
 ):
     """Connect to Spoolman server using configured URL."""
-    enabled, url, _ = await get_spoolman_settings(db)
+    enabled, url, _, _ = await get_spoolman_settings(db)
 
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
@@ -144,7 +147,7 @@ async def sync_printer_ams(
 ):
     """Sync AMS data from a specific printer to Spoolman."""
     # Check if Spoolman is enabled and connected
-    enabled, url, _ = await get_spoolman_settings(db)
+    enabled, url, _, disable_weight_sync = await get_spoolman_settings(db)
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -249,7 +252,7 @@ async def sync_printer_ams(
                 current_tray_uuids.add(spool_tag.upper())
 
             try:
-                sync_result = await client.sync_ams_tray(tray, printer.name)
+                sync_result = await client.sync_ams_tray(tray, printer.name, disable_weight_sync=disable_weight_sync)
                 if sync_result:
                     synced += 1
                     logger.info(f"Synced {tray.tray_sub_brands} from {printer.name} AMS {ams_id} tray {tray.tray_id}")
@@ -285,7 +288,7 @@ async def sync_all_printers(
 ):
     """Sync AMS data from all connected printers to Spoolman."""
     # Check if Spoolman is enabled
-    enabled, url, _ = await get_spoolman_settings(db)
+    enabled, url, _, disable_weight_sync = await get_spoolman_settings(db)
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -382,7 +385,7 @@ async def sync_all_printers(
                     printer_tray_uuids[printer.name].add(spool_tag.upper())
 
                 try:
-                    sync_result = await client.sync_ams_tray(tray, printer.name)
+                    sync_result = await client.sync_ams_tray(tray, printer.name, disable_weight_sync=disable_weight_sync)
                     if sync_result:
                         total_synced += 1
                 except Exception as e:
@@ -412,7 +415,7 @@ async def get_spools(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """Get all spools from Spoolman."""
-    enabled, url, _ = await get_spoolman_settings(db)
+    enabled, url, _, _ = await get_spoolman_settings(db)
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -436,7 +439,7 @@ async def get_filaments(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """Get all filaments from Spoolman."""
-    enabled, url, _ = await get_spoolman_settings(db)
+    enabled, url, _, _ = await get_spoolman_settings(db)
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -471,7 +474,7 @@ 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)
+    enabled, url, _, _ = await get_spoolman_settings(db)
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -516,7 +519,7 @@ 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)
+    enabled, url, _, _ = await get_spoolman_settings(db)
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 
@@ -560,7 +563,7 @@ 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)
+    enabled, url, _, _ = await get_spoolman_settings(db)
     if not enabled:
         raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
 

backend/app/core/database.py

@@ -54,6 +54,7 @@ async def get_db() -> AsyncSession:
 async def init_db():
     # Import models to register them with SQLAlchemy
     from backend.app.models import (  # noqa: F401
+        active_print_spoolman,
         ams_history,
         api_key,
         archive,
@@ -1080,6 +1081,25 @@ async def run_migrations(conn):
     except Exception:
         pass
 
+    # Create active_print_spoolman table for Spoolman per-filament tracking
+    try:
+        await conn.execute(
+            text("""
+            CREATE TABLE IF NOT EXISTS active_print_spoolman (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                printer_id INTEGER NOT NULL REFERENCES printers(id) ON DELETE CASCADE,
+                archive_id INTEGER NOT NULL REFERENCES print_archives(id) ON DELETE CASCADE,
+                filament_usage TEXT NOT NULL,
+                ams_trays TEXT NOT NULL,
+                slot_to_tray TEXT,
+                layer_usage TEXT,
+                filament_properties TEXT,
+                UNIQUE(printer_id, archive_id)
+            )
+        """)
+        )
+    except Exception:
+        pass
 
 async def seed_notification_templates():
     """Seed default notification templates if they don't exist."""

backend/app/main.py

@@ -3,6 +3,7 @@ import logging
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta, timezone
 from logging.handlers import RotatingFileHandler
+from pathlib import Path
 
 
 # =============================================================================
@@ -294,85 +295,475 @@ _last_status_broadcast: dict[int, str] = {}
 _nozzle_count_updated: set[int] = set()  # Track printers where we've updated nozzle_count
 
 
-async def _report_spoolman_usage(printer_id: int, archive_id: int, logger):
-    """Report filament usage to Spoolman after print completion.
+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(f"[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}")
+
 
-    This finds the spool by RFID tag_uid from current AMS state and reports
-    the filament_used_grams from the archive metadata.
+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:
-        from backend.app.api.routes.settings import get_setting
-        from backend.app.models.archive import PrintArchive
+        # 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 Spoolman URL
-        spoolman_url = await get_setting(db, "spoolman_url")
-        if not spoolman_url:
-            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
 
-        # Get or create Spoolman client
+    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:
-            client = await init_spoolman_client(spoolman_url)
+            spoolman_url = await get_setting(db, "spoolman_url")
+            if spoolman_url:
+                client = await init_spoolman_client(spoolman_url)
 
-        # Check if Spoolman is reachable
-        if not await client.health_check():
-            logger.warning("Spoolman not reachable for usage reporting")
+        if not client or not await client.health_check():
+            logger.warning("[SPOOLMAN] Not reachable for partial usage reporting")
             return
 
-        # Get archive to find filament usage
-        result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-        archive = result.scalar_one_or_none()
-        if not archive or not archive.filament_used_grams:
-            logger.debug(f"No filament usage data for archive {archive_id}")
-            return
+        spools_updated = 0
 
-        filament_used = archive.filament_used_grams
-        logger.info(f"[SPOOLMAN] Archive {archive_id} used {filament_used}g of filament")
+        # 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 current AMS state from printer to find the active spool
-        state = printer_manager.get_status(printer_id)
-        if not state or not state.raw_data:
-            logger.debug("No printer state available for usage reporting")
-            return
+            # Get cumulative usage at current layer
+            usage_mm = get_cumulative_usage_at_layer(layer_usage_int, current_layer)
 
-        ams_data = state.raw_data.get("ams")
-        if not ams_data:
-            logger.debug("No AMS data available for usage reporting")
-            return
+            if usage_mm:
+                logger.info(f"[SPOOLMAN] Using G-code parsed data for layer {current_layer}")
 
-        # Find spools with RFID tags in Spoolman and report usage
-        # For now, we report usage to the first spool found with a matching tag
-        # TODO: In future, track which specific trays were used during the print
-        spools_updated = 0
-        for ams_unit in ams_data:
-            trays = ams_unit.get("tray", [])
+                # 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
 
-            for tray_data in trays:
-                tag_uid = tray_data.get("tag_uid")
-                if not tag_uid:
-                    continue
+                    # 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
 
-                # Find spool in Spoolman by tag
-                spool = await client.find_spool_by_tag(tag_uid)
-                if spool:
                     # Report usage to Spoolman
-                    result = await client.use_spool(spool["id"], filament_used)
+                    result = await client.use_spool(spool["id"], grams_used)
                     if result:
                         logger.info(
-                            f"[SPOOLMAN] Reported {filament_used}g usage to spool {spool['id']} (tag: {tag_uid})"
+                            f"[SPOOLMAN] Partial (G-code): slot {slot_id}: {grams_used}g ({mm_used:.1f}mm, d={density}) -> spool {spool['id']}"
                         )
                         spools_updated += 1
-                        # Only report to one spool for single-material prints
-                        # Multi-material prints would need more sophisticated tracking
-                        return
+
+                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.debug(f"No matching Spoolman spools found for printer {printer_id}")
+            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):
@@ -611,6 +1002,10 @@ async def on_ams_change(printer_id: int, ams_data: list):
             if sync_mode and sync_mode != "auto":
                 return  # Only sync on auto mode
 
+            # Check if weight sync is disabled
+            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"
+
             # Get Spoolman URL
             spoolman_url = await get_setting(db, "spoolman_url")
             if not spoolman_url:
@@ -643,7 +1038,7 @@ async def on_ams_change(printer_id: int, ams_data: list):
                         continue  # Empty tray
 
                     try:
-                        result = await client.sync_ams_tray(tray, printer_name)
+                        result = await client.sync_ams_tray(tray, printer_name, disable_weight_sync=disable_weight_sync)
                         if result:
                             synced += 1
                     except Exception as e:
@@ -1008,6 +1403,12 @@ async def on_print_start(printer_id: int, data: dict):
                 # Extract printable objects from the archived 3MF file
                 _load_objects_from_archive(archive, printer_id, logger)
 
+                # Store Spoolman tracking data for per-filament usage reporting
+                try:
+                    await _store_spoolman_print_data(printer_id, archive.id, archive.file_path, db, logger)
+                except Exception as e:
+                    logger.warning(f"[SPOOLMAN] Failed to store tracking data: {e}")
+
             return  # Skip creating a new archive
 
         # Check if there's already a "printing" archive for this printer/file
@@ -1302,6 +1703,14 @@ async def on_print_start(printer_id: int, data: dict):
                 except Exception:
                     pass  # Don't fail if MQTT fails
 
+                # 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
+                    )
+                except Exception as e:
+                    logger.debug(f"[SPOOLMAN] Could not store tracking for fallback archive: {e}")
+
                 # Send notification without archive data (file not found)
                 if not notification_sent:
                     await _send_print_start_notification(printer_id, data, logger=logger)
@@ -1413,6 +1822,12 @@ async def on_print_start(printer_id: int, data: dict):
                             logger.info(f"Loaded {len(printable_objects)} printable objects for printer {printer_id}")
                 except Exception as e:
                     logger.debug(f"Failed to extract printable objects: {e}")
+
+                # Store Spoolman tracking data for per-filament usage reporting
+                try:
+                    await _store_spoolman_print_data(printer_id, archive.id, archive.file_path, db, logger)
+                except Exception as e:
+                    logger.warning(f"[SPOOLMAN] Failed to store tracking data: {e}")
         finally:
             if temp_path and temp_path.exists():
                 temp_path.unlink()
@@ -1764,6 +2179,13 @@ async def on_print_complete(printer_id: int, data: dict):
             log_timing("Spoolman usage report")
         except Exception as e:
             logger.warning(f"Spoolman usage reporting failed: {e}")
+    else:
+        # 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)
+        except Exception as e:
+            logger.debug(f"[SPOOLMAN] Cleanup failed: {e}")
 
     # Run slow operations as background tasks to avoid blocking the event loop
     # These operations can take 5-10+ seconds and would freeze the UI if awaited

backend/app/models/active_print_spoolman.py

@@ -0,0 +1,42 @@
+"""Track Spoolman data for active prints."""
+
+from sqlalchemy import JSON, ForeignKey, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.app.core.database import Base
+
+
+class ActivePrintSpoolman(Base):
+    """Stores Spoolman tracking data for active prints.
+
+    This data is captured at print start and used at print completion
+    to report per-filament usage to the correct Spoolman spools.
+    Rows are deleted after print completes.
+
+    Key: (printer_id, archive_id) - allows same archive on different printers
+    """
+
+    __tablename__ = "active_print_spoolman"
+    __table_args__ = (UniqueConstraint("printer_id", "archive_id", name="uq_printer_archive"),)
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    printer_id: Mapped[int] = mapped_column(ForeignKey("printers.id", ondelete="CASCADE"))
+    archive_id: Mapped[int] = mapped_column(ForeignKey("print_archives.id", ondelete="CASCADE"))
+
+    # Per-filament usage from 3MF: [{"slot_id": 1, "used_g": 50.5, "type": "PLA"}, ...]
+    filament_usage: Mapped[list] = mapped_column(JSON)
+
+    # AMS tray state at print start: {0: {"tray_uuid": "...", "tag_uid": "..."}, ...}
+    ams_trays: Mapped[dict] = mapped_column(JSON)
+
+    # Custom slot-to-tray mapping from queue (optional): [5, -1, 2, -1]
+    slot_to_tray: Mapped[list | None] = mapped_column(JSON, nullable=True)
+
+    # Per-layer cumulative usage from G-code parsing (for accurate partial usage)
+    # Format: {"0": {0: 125.5}, "1": {0: 250.0, 1: 50.0}, ...}
+    # Keys are layer numbers (as strings for JSON), values are filament_id -> mm
+    layer_usage: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+
+    # Filament properties (density, diameter per filament slot)
+    # Format: {1: {"density": 1.24, "diameter": 1.75, "type": "PLA"}, ...}
+    filament_properties: Mapped[dict | None] = mapped_column(JSON, nullable=True)

backend/app/schemas/settings.py

@@ -23,6 +23,14 @@ class AppSettings(BaseModel):
     spoolman_sync_mode: str = Field(
         default="auto", description="Sync mode: 'auto' syncs immediately, 'manual' requires button press"
     )
+    spoolman_disable_weight_sync: bool = Field(
+        default=False,
+        description="Disable remaining_weight sync. When enabled, only location is updated for existing spools.",
+    )
+    spoolman_report_partial_usage: bool = Field(
+        default=True,
+        description="Report Partial Usage for Failed Prints. When a print fails or is cancelled, report the estimated filament used up to that point based on layer progress.",
+    )
 
     # Updates
     check_updates: bool = Field(default=True, description="Automatically check for updates on startup")
@@ -134,6 +142,8 @@ class AppSettingsUpdate(BaseModel):
     spoolman_enabled: bool | None = None
     spoolman_url: str | None = None
     spoolman_sync_mode: str | None = None
+    spoolman_disable_weight_sync: bool | None = None
+    spoolman_report_partial_usage: bool | None = None
     check_updates: bool | None = None
     check_printer_firmware: bool | None = None
     notification_language: str | None = None

backend/app/services/spoolman.py

@@ -662,7 +662,7 @@ class SpoolmanClient:
         """
         return (remain_percent / 100.0) * spool_weight
 
-    async def sync_ams_tray(self, tray: AMSTray, printer_name: str) -> dict | None:
+    async def sync_ams_tray(self, tray: AMSTray, printer_name: str, disable_weight_sync: bool = False) -> dict | None:
         """Sync a single AMS tray to Spoolman.
 
         Only syncs trays with valid Bambu Lab tray_uuid (32 hex characters).
@@ -674,6 +674,8 @@ class SpoolmanClient:
         Args:
             tray: The AMSTray to sync
             printer_name: Name of the printer for location
+            disable_weight_sync: If True, skip updating remaining_weight for existing spools.
+                This allows Spoolman's granular usage tracking to maintain accurate weights.
 
         Returns:
             Synced spool dictionary or None if skipped or failed.
@@ -720,7 +722,7 @@ class SpoolmanClient:
             logger.info(f"Updating existing spool {existing['id']} for tag {spool_tag[:16]}...")
             return await self.update_spool(
                 spool_id=existing["id"],
-                remaining_weight=remaining,
+                remaining_weight=None if disable_weight_sync else remaining,
                 location=location,
             )
 

backend/app/utils/threemf_tools.py

@@ -0,0 +1,307 @@
+"""3MF file parsing utilities for filament tracking.
+
+This module provides functions to parse Bambu Lab 3MF files and extract
+per-layer filament usage data from the embedded G-code. This enables
+accurate partial usage reporting for multi-material prints.
+"""
+
+import json
+import math
+import re
+import zipfile
+from pathlib import Path
+from xml.etree import ElementTree as ET
+
+# Default filament properties
+DEFAULT_FILAMENT_DIAMETER = 1.75  # mm
+DEFAULT_FILAMENT_DENSITY = 1.24  # g/cm³ (PLA)
+
+
+def parse_gcode_layer_filament_usage(gcode_content: str) -> dict[int, dict[int, float]]:
+    """Parse G-code to extract per-layer, per-filament cumulative extrusion in mm.
+
+    This function tracks filament extrusion across layers and tool changes,
+    building a cumulative usage map that can be used to calculate partial
+    usage at any layer.
+
+    Args:
+        gcode_content: The raw G-code content as a string
+
+    Returns:
+        A nested dictionary mapping layer numbers to filament usage:
+        {layer: {filament_id: cumulative_mm}, ...}
+
+    Example:
+        {0: {0: 125.5}, 1: {0: 250.0, 1: 50.0}, 2: {0: 375.0, 1: 150.0}}
+
+        This shows:
+        - Layer 0: filament 0 used 125.5mm cumulative
+        - Layer 1: filament 0 used 250mm cumulative, filament 1 used 50mm
+        - Layer 2: filament 0 used 375mm cumulative, filament 1 used 150mm
+
+    G-code commands parsed:
+        - M73 L<layer>: Layer change marker
+        - M620 S<filament>: Filament/tool change (S255 = unload)
+        - G0/G1/G2/G3 E<amount>: Extrusion moves
+    """
+    layer_filaments: dict[int, dict[int, float]] = {}
+    current_layer = 0
+    active_filament: int | None = None
+    cumulative_extrusion: dict[int, float] = {}  # filament_id -> total mm
+
+    for line in gcode_content.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+
+        # Handle comments - skip but check for layer markers
+        if line.startswith(";"):
+            # Some slicers use comment-based layer markers
+            # e.g., "; CHANGE_LAYER" or ";LAYER_CHANGE"
+            continue
+
+        # Split line into command and inline comment
+        if ";" in line:
+            line = line.split(";")[0].strip()
+
+        # Extract command and parameters
+        parts = line.split()
+        if not parts:
+            continue
+        cmd = parts[0].upper()
+
+        # Layer change: M73 L<layer>
+        # Bambu printers use M73 with L parameter for layer indication
+        if cmd == "M73":
+            for part in parts[1:]:
+                part_upper = part.upper()
+                if part_upper.startswith("L"):
+                    try:
+                        new_layer = int(part[1:])
+                        # Save current state before layer change
+                        if cumulative_extrusion:
+                            layer_filaments[current_layer] = cumulative_extrusion.copy()
+                        current_layer = new_layer
+                    except ValueError:
+                        pass
+
+        # Filament change: M620 S<filament>
+        # Bambu uses M620 for AMS filament switching
+        # S255 means full unload (no active filament)
+        elif cmd == "M620":
+            for part in parts[1:]:
+                part_upper = part.upper()
+                if part_upper.startswith("S"):
+                    filament_str = part[1:]
+                    if filament_str == "255":
+                        # Full unload - no active filament
+                        active_filament = None
+                    else:
+                        try:
+                            # Extract digits (e.g., "0A" -> 0, "1" -> 1)
+                            match = re.match(r"(\d+)", filament_str)
+                            if match:
+                                active_filament = int(match.group(1))
+                        except (ValueError, AttributeError):
+                            pass
+
+        # Extrusion moves: G0/G1/G2/G3 with E parameter
+        # Only G1 typically has extrusion, but check all for safety
+        elif cmd in ("G0", "G1", "G2", "G3"):
+            if active_filament is None:
+                continue
+            for part in parts[1:]:
+                part_upper = part.upper()
+                if part_upper.startswith("E"):
+                    try:
+                        extrusion = float(part[1:])
+                        # Only count positive extrusion (not retractions)
+                        if extrusion > 0:
+                            current = cumulative_extrusion.get(active_filament, 0)
+                            cumulative_extrusion[active_filament] = current + extrusion
+                    except ValueError:
+                        pass
+
+    # Save final layer state
+    if cumulative_extrusion:
+        layer_filaments[current_layer] = cumulative_extrusion.copy()
+
+    return layer_filaments
+
+
+def mm_to_grams(
+    length_mm: float,
+    diameter_mm: float = DEFAULT_FILAMENT_DIAMETER,
+    density_g_cm3: float = DEFAULT_FILAMENT_DENSITY,
+) -> float:
+    """Convert filament length in mm to weight in grams.
+
+    Uses the formula: mass = volume × density
+    where volume = π × r² × length
+
+    Args:
+        length_mm: Length of filament in millimeters
+        diameter_mm: Filament diameter in millimeters (default: 1.75)
+        density_g_cm3: Material density in g/cm³ (default: 1.24 for PLA)
+
+    Returns:
+        Weight in grams
+    """
+    radius_cm = (diameter_mm / 2) / 10  # Convert mm to cm
+    length_cm = length_mm / 10  # Convert mm to cm
+    volume_cm3 = math.pi * radius_cm * radius_cm * length_cm
+    return volume_cm3 * density_g_cm3
+
+
+def extract_layer_filament_usage_from_3mf(file_path: Path) -> dict[int, dict[int, float]] | None:
+    """Extract per-layer filament usage from a 3MF file's embedded G-code.
+
+    Args:
+        file_path: Path to the 3MF file
+
+    Returns:
+        Dictionary mapping layers to filament usage, or None if parsing fails.
+        Format: {layer: {filament_id: cumulative_mm}, ...}
+    """
+    try:
+        with zipfile.ZipFile(file_path, "r") as zf:
+            # Find G-code file(s) - usually plate_1.gcode or Metadata/plate_1.gcode
+            gcode_files = [f for f in zf.namelist() if f.endswith(".gcode")]
+            if not gcode_files:
+                return None
+
+            # Use the first G-code file (typically only one per 3MF export)
+            gcode_path = gcode_files[0]
+            gcode_content = zf.read(gcode_path).decode("utf-8", errors="ignore")
+
+            return parse_gcode_layer_filament_usage(gcode_content)
+    except Exception:
+        return None
+
+
+def get_cumulative_usage_at_layer(
+    layer_usage: dict[int, dict[int, float]],
+    target_layer: int,
+) -> dict[int, float]:
+    """Get cumulative filament usage (in mm) up to and including target_layer.
+
+    Args:
+        layer_usage: The output from parse_gcode_layer_filament_usage()
+        target_layer: The layer number to get usage for
+
+    Returns:
+        Dictionary of {filament_id: cumulative_mm} for each filament used
+        up to target_layer. Returns empty dict if no data available.
+    """
+    if not layer_usage:
+        return {}
+
+    # Find the highest recorded layer <= target_layer
+    # (we store snapshots at layer changes, so we need the closest one)
+    relevant_layers = [layer for layer in layer_usage.keys() if layer <= target_layer]
+    if not relevant_layers:
+        return {}
+
+    max_layer = max(relevant_layers)
+    return layer_usage.get(max_layer, {})
+
+
+def extract_filament_properties_from_3mf(file_path: Path) -> dict[int, dict]:
+    """Extract filament properties (density, diameter, type) from 3MF metadata.
+
+    Args:
+        file_path: Path to the 3MF file
+
+    Returns:
+        Dictionary mapping filament IDs to their properties:
+        {filament_id: {"diameter": 1.75, "density": 1.24, "type": "PLA"}, ...}
+
+        Note: filament_id is 1-based (matches slot_id in slice_info.config)
+    """
+    properties: dict[int, dict] = {}
+    try:
+        with zipfile.ZipFile(file_path, "r") as zf:
+            # Try slice_info.config first for filament types
+            if "Metadata/slice_info.config" in zf.namelist():
+                content = zf.read("Metadata/slice_info.config").decode()
+                root = ET.fromstring(content)
+                for f in root.findall(".//filament"):
+                    try:
+                        # id is 1-based in slice_info.config
+                        fid = int(f.get("id", 0))
+                        properties[fid] = {
+                            "type": f.get("type", "PLA"),
+                            "diameter": DEFAULT_FILAMENT_DIAMETER,
+                            "density": DEFAULT_FILAMENT_DENSITY,
+                        }
+                    except ValueError:
+                        pass
+
+            # Try project_settings.config for density values
+            if "Metadata/project_settings.config" in zf.namelist():
+                content = zf.read("Metadata/project_settings.config").decode()
+                try:
+                    data = json.loads(content)
+                    densities = data.get("filament_density", [])
+                    for i, density in enumerate(densities):
+                        # project_settings uses 0-based indexing, convert to 1-based
+                        fid = i + 1
+                        if fid not in properties:
+                            properties[fid] = {
+                                "type": "",
+                                "diameter": DEFAULT_FILAMENT_DIAMETER,
+                            }
+                        try:
+                            properties[fid]["density"] = float(density)
+                        except (ValueError, TypeError):
+                            properties[fid]["density"] = DEFAULT_FILAMENT_DENSITY
+                except json.JSONDecodeError:
+                    pass
+    except Exception:
+        pass
+
+    return properties
+
+
+def extract_filament_usage_from_3mf(file_path: Path) -> list[dict]:
+    """Extract per-filament total usage from 3MF slice_info.config.
+
+    This extracts the slicer-estimated total usage per filament slot,
+    not the per-layer breakdown.
+
+    Args:
+        file_path: Path to the 3MF file
+
+    Returns:
+        List of filament usage dictionaries:
+        [{"slot_id": 1, "used_g": 50.5, "type": "PLA", "color": "#FF0000"}, ...]
+    """
+    filament_usage = []
+    try:
+        with zipfile.ZipFile(file_path, "r") as zf:
+            if "Metadata/slice_info.config" not in zf.namelist():
+                return []
+
+            content = zf.read("Metadata/slice_info.config").decode()
+            root = ET.fromstring(content)
+
+            for f in root.findall(".//filament"):
+                filament_id = f.get("id")
+                used_g = f.get("used_g", "0")
+                try:
+                    used_amount = float(used_g)
+                    if filament_id:
+                        filament_usage.append(
+                            {
+                                "slot_id": int(filament_id),
+                                "used_g": used_amount,
+                                "type": f.get("type", ""),
+                                "color": f.get("color", ""),
+                            }
+                        )
+                except (ValueError, TypeError):
+                    pass
+    except Exception:
+        pass
+
+    return filament_usage

backend/tests/integration/test_spoolman_api.py

@@ -499,3 +499,209 @@ class TestSpoolmanAPI:
         assert isinstance(data["filaments"], list)
         assert len(data["filaments"]) == 1
         assert data["filaments"][0]["name"] == "PLA Basic"
+
+    # =========================================================================
+    # Disable Weight Sync Tests
+    # =========================================================================
+
+    @pytest.fixture
+    async def spoolman_settings_weight_sync_disabled(self, db_session):
+        """Create Spoolman settings with weight sync disabled."""
+        from backend.app.models.settings import Settings
+
+        enabled_setting = Settings(key="spoolman_enabled", value="true")
+        url_setting = Settings(key="spoolman_url", value="http://localhost:7912")
+        disable_weight_setting = Settings(key="spoolman_disable_weight_sync", value="true")
+        partial_usage_setting = Settings(key="spoolman_report_partial_usage", value="true")
+        db_session.add(enabled_setting)
+        db_session.add(url_setting)
+        db_session.add(disable_weight_setting)
+        db_session.add(partial_usage_setting)
+        await db_session.commit()
+        return {
+            "enabled": enabled_setting,
+            "url": url_setting,
+            "disable_weight": disable_weight_setting,
+            "partial_usage": partial_usage_setting,
+        }
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_settings_returns_disable_weight_sync(
+        self, async_client: AsyncClient, spoolman_settings_weight_sync_disabled
+    ):
+        """Verify settings endpoint returns the disable_weight_sync setting."""
+        response = await async_client.get("/api/v1/settings/spoolman")
+        assert response.status_code == 200
+        data = response.json()
+        assert "spoolman_disable_weight_sync" in data
+        assert data["spoolman_disable_weight_sync"] == "true"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_settings_update_disable_weight_sync(self, async_client: AsyncClient, spoolman_settings):
+        """Verify settings endpoint can update the disable_weight_sync setting."""
+        # First verify it's false by default
+        response = await async_client.get("/api/v1/settings/spoolman")
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get("spoolman_disable_weight_sync", "false") == "false"
+
+        # Update the setting
+        response = await async_client.put(
+            "/api/v1/settings/spoolman",
+            json={"spoolman_disable_weight_sync": "true"},
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["spoolman_disable_weight_sync"] == "true"
+
+        # Verify it persisted
+        response = await async_client.get("/api/v1/settings/spoolman")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["spoolman_disable_weight_sync"] == "true"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_sync_with_weight_sync_disabled_updates_location_only(
+        self,
+        async_client: AsyncClient,
+        spoolman_settings_weight_sync_disabled,
+        mock_spoolman_client,
+        printer_factory,
+    ):
+        """Verify sync only updates location when disable_weight_sync is enabled."""
+        printer = await printer_factory()
+
+        # Mock existing spool
+        mock_existing_spool = {
+            "id": 42,
+            "remaining_weight": 800,
+            "extra": {"tag": '"A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4"'},
+            "filament": {"id": 1, "name": "PLA Red", "material": "PLA"},
+        }
+        mock_spoolman_client.find_spool_by_tag = AsyncMock(return_value=mock_existing_spool)
+        mock_spoolman_client.parse_ams_tray = MagicMock()
+
+        # Create mock AMSTray
+        from backend.app.services.spoolman import AMSTray
+
+        mock_tray = AMSTray(
+            ams_id=0,
+            tray_id=0,
+            tray_type="PLA",
+            tray_sub_brands="PLA Basic",
+            tray_color="FF0000FF",
+            remain=50,
+            tag_uid="",
+            tray_uuid="A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4",
+            tray_info_idx="GFA00",
+            tray_weight=1000,
+        )
+        mock_spoolman_client.parse_ams_tray.return_value = mock_tray
+        mock_spoolman_client.is_bambu_lab_spool = MagicMock(return_value=True)
+        mock_spoolman_client.convert_ams_slot_to_location = MagicMock(return_value="AMS A1")
+        mock_spoolman_client.sync_ams_tray = AsyncMock(return_value={"id": 42})
+        mock_spoolman_client.clear_location_for_removed_spools = AsyncMock(return_value=0)
+
+        with patch("backend.app.api.routes.spoolman.printer_manager") as pm_mock:
+            mock_state = MagicMock()
+            mock_state.raw_data = {
+                "ams": [
+                    {
+                        "id": 0,
+                        "tray": [
+                            {
+                                "id": 0,
+                                "tray_type": "PLA",
+                                "tray_sub_brands": "PLA Basic",
+                                "tray_color": "FF0000FF",
+                                "remain": 50,
+                                "tag_uid": "",
+                                "tray_uuid": "A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4",
+                                "tray_info_idx": "GFA00",
+                                "tray_weight": 1000,
+                            }
+                        ],
+                    }
+                ]
+            }
+            pm_mock.get_status = MagicMock(return_value=mock_state)
+
+            response = await async_client.post(f"/api/v1/spoolman/sync/{printer.id}")
+            assert response.status_code == 200
+
+            # Verify sync_ams_tray was called with disable_weight_sync=True
+            mock_spoolman_client.sync_ams_tray.assert_called()
+            call_kwargs = mock_spoolman_client.sync_ams_tray.call_args.kwargs
+            assert call_kwargs.get("disable_weight_sync") is True
+
+    # =========================================================================
+    # Report Partial Usage Tests
+    # =========================================================================
+
+    @pytest.fixture
+    async def spoolman_settings_partial_usage_disabled(self, db_session):
+        """Create Spoolman settings with partial usage reporting disabled."""
+        from backend.app.models.settings import Settings
+
+        enabled_setting = Settings(key="spoolman_enabled", value="true")
+        url_setting = Settings(key="spoolman_url", value="http://localhost:7912")
+        partial_usage_setting = Settings(key="spoolman_report_partial_usage", value="false")
+        db_session.add(enabled_setting)
+        db_session.add(url_setting)
+        db_session.add(partial_usage_setting)
+        await db_session.commit()
+        return {
+            "enabled": enabled_setting,
+            "url": url_setting,
+            "partial_usage": partial_usage_setting,
+        }
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_settings_returns_report_partial_usage(
+        self, async_client: AsyncClient, spoolman_settings_partial_usage_disabled
+    ):
+        """Verify settings endpoint returns the report_partial_usage setting."""
+        response = await async_client.get("/api/v1/settings/spoolman")
+        assert response.status_code == 200
+        data = response.json()
+        assert "spoolman_report_partial_usage" in data
+        assert data["spoolman_report_partial_usage"] == "false"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_settings_update_report_partial_usage(self, async_client: AsyncClient, spoolman_settings):
+        """Verify settings endpoint can update the report_partial_usage setting."""
+        # First verify it's true by default
+        response = await async_client.get("/api/v1/settings/spoolman")
+        assert response.status_code == 200
+        data = response.json()
+        assert data.get("spoolman_report_partial_usage", "true") == "true"
+
+        # Update the setting to false
+        response = await async_client.put(
+            "/api/v1/settings/spoolman",
+            json={"spoolman_report_partial_usage": "false"},
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["spoolman_report_partial_usage"] == "false"
+
+        # Verify it persisted
+        response = await async_client.get("/api/v1/settings/spoolman")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["spoolman_report_partial_usage"] == "false"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_settings_report_partial_usage_defaults_to_true(self, async_client: AsyncClient, spoolman_settings):
+        """Verify report_partial_usage defaults to true (unlike disable_weight_sync which defaults to false)."""
+        response = await async_client.get("/api/v1/settings/spoolman")
+        assert response.status_code == 200
+        data = response.json()
+        # Should default to "true"
+        assert data["spoolman_report_partial_usage"] == "true"

backend/tests/unit/services/test_spoolman_service.py

@@ -0,0 +1,166 @@
+"""Unit tests for Spoolman service.
+
+These tests specifically target the sync_ams_tray method's disable_weight_sync
+functionality that controls whether remaining_weight is updated.
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from backend.app.services.spoolman import AMSTray, SpoolmanClient
+
+
+class TestSpoolmanClient:
+    """Tests for SpoolmanClient class."""
+
+    @pytest.fixture
+    def client(self):
+        """Create a SpoolmanClient instance."""
+        return SpoolmanClient("http://localhost:7912")
+
+    @pytest.fixture
+    def sample_tray(self):
+        """Create a sample AMSTray for testing."""
+        return AMSTray(
+            ams_id=0,
+            tray_id=0,
+            tray_type="PLA",
+            tray_sub_brands="PLA Basic",
+            tray_color="FF0000FF",
+            remain=50,
+            tag_uid="",
+            tray_uuid="A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4",
+            tray_info_idx="GFA00",
+            tray_weight=1000,
+        )
+
+    @pytest.fixture
+    def existing_spool(self):
+        """Create a mock existing spool response."""
+        return {
+            "id": 42,
+            "remaining_weight": 800,
+            "extra": {"tag": '"A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4"'},
+            "filament": {"id": 1, "name": "PLA Red", "material": "PLA"},
+        }
+
+    @pytest.fixture
+    def mock_filament(self):
+        """Create a mock filament response."""
+        return {"id": 1, "name": "PLA Basic", "material": "PLA"}
+
+    # ========================================================================
+    # Tests for sync_ams_tray with disable_weight_sync
+    # ========================================================================
+
+    @pytest.mark.asyncio
+    async def test_sync_ams_tray_updates_weight_by_default(self, client, sample_tray, existing_spool):
+        """Verify sync_ams_tray updates remaining_weight by default."""
+        with patch.object(client, "find_spool_by_tag", AsyncMock(return_value=existing_spool)):
+            with patch.object(client, "update_spool", AsyncMock(return_value={"id": 42})) as mock_update:
+                await client.sync_ams_tray(sample_tray, "TestPrinter")
+
+                mock_update.assert_called_once()
+                call_kwargs = mock_update.call_args.kwargs
+                assert "remaining_weight" in call_kwargs
+                assert call_kwargs["remaining_weight"] == 500.0  # 50% of 1000g
+                assert "location" in call_kwargs
+
+    @pytest.mark.asyncio
+    async def test_sync_ams_tray_skips_weight_when_disabled(self, client, sample_tray, existing_spool):
+        """Verify sync_ams_tray skips remaining_weight when disable_weight_sync=True."""
+        with patch.object(client, "find_spool_by_tag", AsyncMock(return_value=existing_spool)):
+            with patch.object(client, "update_spool", AsyncMock(return_value={"id": 42})) as mock_update:
+                await client.sync_ams_tray(sample_tray, "TestPrinter", disable_weight_sync=True)
+
+                mock_update.assert_called_once()
+                call_kwargs = mock_update.call_args.kwargs
+                # remaining_weight should be None (not updated)
+                assert call_kwargs.get("remaining_weight") is None
+                # location should still be updated
+                assert "location" in call_kwargs
+                assert "TestPrinter" in call_kwargs["location"]
+
+    @pytest.mark.asyncio
+    async def test_sync_ams_tray_new_spool_always_includes_weight(
+        self, client, sample_tray, mock_filament
+    ):
+        """Verify new spool creation always includes remaining_weight even when disabled."""
+        with patch.object(client, "find_spool_by_tag", AsyncMock(return_value=None)):
+            with patch.object(client, "_find_or_create_filament", AsyncMock(return_value=mock_filament)):
+                with patch.object(client, "create_spool", AsyncMock(return_value={"id": 99})) as mock_create:
+                    await client.sync_ams_tray(sample_tray, "TestPrinter", disable_weight_sync=True)
+
+                    mock_create.assert_called_once()
+                    call_kwargs = mock_create.call_args.kwargs
+                    # New spools should ALWAYS include remaining_weight
+                    assert "remaining_weight" in call_kwargs
+                    assert call_kwargs["remaining_weight"] == 500.0  # 50% of 1000g
+
+    @pytest.mark.asyncio
+    async def test_sync_ams_tray_location_format(self, client, sample_tray, existing_spool):
+        """Verify location format is correct when updating spool."""
+        with patch.object(client, "find_spool_by_tag", AsyncMock(return_value=existing_spool)):
+            with patch.object(client, "update_spool", AsyncMock(return_value={"id": 42})) as mock_update:
+                await client.sync_ams_tray(sample_tray, "My Printer", disable_weight_sync=True)
+
+                call_kwargs = mock_update.call_args.kwargs
+                # Location should follow pattern: "PrinterName - AMS A1"
+                assert "location" in call_kwargs
+                assert "My Printer" in call_kwargs["location"]
+                assert "AMS" in call_kwargs["location"]
+
+    @pytest.mark.asyncio
+    async def test_sync_ams_tray_skips_non_bambu_spool(self, client):
+        """Verify non-Bambu Lab spools are skipped."""
+        # Third-party spool without proper identifiers
+        tray = AMSTray(
+            ams_id=0,
+            tray_id=0,
+            tray_type="PLA",
+            tray_sub_brands="Third Party PLA",
+            tray_color="FF0000FF",
+            remain=50,
+            tag_uid="",
+            tray_uuid="",
+            tray_info_idx="",  # No Bambu Lab preset ID
+            tray_weight=1000,
+        )
+
+        result = await client.sync_ams_tray(tray, "TestPrinter")
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_sync_ams_tray_weight_calculation(self, client, existing_spool):
+        """Verify remaining weight is calculated correctly for various percentages."""
+        test_cases = [
+            (100, 1000, 1000.0),  # Full spool
+            (50, 1000, 500.0),  # Half spool
+            (25, 1000, 250.0),  # Quarter spool
+            (0, 1000, 0.0),  # Empty spool
+            (75, 500, 375.0),  # Different spool weight
+        ]
+
+        for remain, weight, expected in test_cases:
+            tray = AMSTray(
+                ams_id=0,
+                tray_id=0,
+                tray_type="PLA",
+                tray_sub_brands="PLA Basic",
+                tray_color="FF0000FF",
+                remain=remain,
+                tag_uid="",
+                tray_uuid="A1B2C3D4E5F6A1B2C3D4E5F6A1B2C3D4",
+                tray_info_idx="GFA00",
+                tray_weight=weight,
+            )
+
+            with patch.object(client, "find_spool_by_tag", AsyncMock(return_value=existing_spool)):
+                with patch.object(client, "update_spool", AsyncMock(return_value={"id": 42})) as mock_update:
+                    await client.sync_ams_tray(tray, "TestPrinter", disable_weight_sync=False)
+
+                    call_kwargs = mock_update.call_args.kwargs
+                    assert call_kwargs["remaining_weight"] == expected, (
+                        f"Expected {expected}g for {remain}% of {weight}g, got {call_kwargs['remaining_weight']}"
+                    )

frontend/src/api/client.ts

@@ -3089,9 +3089,9 @@ export const api = {
       body: JSON.stringify({ tray_uuid: trayUuid }),
     }),
   getSpoolmanSettings: () =>
-    request<{ spoolman_enabled: string; spoolman_url: string; spoolman_sync_mode: string }>('/settings/spoolman'),
-  updateSpoolmanSettings: (data: { spoolman_enabled?: string; spoolman_url?: string; spoolman_sync_mode?: string }) =>
-    request<{ spoolman_enabled: string; spoolman_url: string; spoolman_sync_mode: string }>('/settings/spoolman', {
+    request<{ spoolman_enabled: string; spoolman_url: string; spoolman_sync_mode: string; spoolman_disable_weight_sync: string; spoolman_report_partial_usage: string; }>('/settings/spoolman'),
+  updateSpoolmanSettings: (data: { spoolman_enabled?: string; spoolman_url?: string; spoolman_sync_mode?: string; spoolman_disable_weight_sync?: string; spoolman_report_partial_usage?: string; }) =>
+    request<{ spoolman_enabled: string; spoolman_url: string; spoolman_sync_mode: string; spoolman_disable_weight_sync: string; spoolman_report_partial_usage: string; }>('/settings/spoolman', {
       method: 'PUT',
       body: JSON.stringify(data),
     }),

frontend/src/components/SpoolmanSettings.tsx

@@ -11,6 +11,8 @@ export function SpoolmanSettings() {
   const [localEnabled, setLocalEnabled] = useState(false);
   const [localUrl, setLocalUrl] = useState('');
   const [localSyncMode, setLocalSyncMode] = useState('auto');
+  const [localDisableWeightSync, setLocalDisableWeightSync] = useState(false);
+  const [localReportPartialUsage, setLocalReportPartialUsage] = useState(true);
   const [showSaved, setShowSaved] = useState(false);
   const [selectedPrinterId, setSelectedPrinterId] = useState<number | 'all'>('all');
   const [isInitialized, setIsInitialized] = useState(false);
@@ -41,6 +43,8 @@ export function SpoolmanSettings() {
       setLocalEnabled(settings.spoolman_enabled === 'true');
       setLocalUrl(settings.spoolman_url || '');
       setLocalSyncMode(settings.spoolman_sync_mode || 'auto');
+      setLocalDisableWeightSync(settings.spoolman_disable_weight_sync === 'true');
+      setLocalReportPartialUsage(settings.spoolman_report_partial_usage !== 'false');
       setIsInitialized(true);
     }
   }, [settings]);
@@ -53,7 +57,9 @@ export function SpoolmanSettings() {
     const hasChanges =
       (settings.spoolman_enabled === 'true') !== localEnabled ||
       (settings.spoolman_url || '') !== localUrl ||
-      (settings.spoolman_sync_mode || 'auto') !== localSyncMode;
+      (settings.spoolman_sync_mode || 'auto') !== localSyncMode ||
+      (settings.spoolman_disable_weight_sync === 'true') !== localDisableWeightSync ||
+      (settings.spoolman_report_partial_usage !== 'false') !== localReportPartialUsage;
 
     if (hasChanges) {
       const timeoutId = setTimeout(() => {
@@ -62,7 +68,7 @@ export function SpoolmanSettings() {
       return () => clearTimeout(timeoutId);
     }
     // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, [localEnabled, localUrl, localSyncMode, isInitialized]);
+  }, [localEnabled, localUrl, localSyncMode, localDisableWeightSync, localReportPartialUsage, isInitialized]);
 
   // Save mutation
   const saveMutation = useMutation({
@@ -71,6 +77,8 @@ export function SpoolmanSettings() {
         spoolman_enabled: localEnabled ? 'true' : 'false',
         spoolman_url: localUrl,
         spoolman_sync_mode: localSyncMode,
+        spoolman_disable_weight_sync: localDisableWeightSync ? 'true' : 'false',
+        spoolman_report_partial_usage: localReportPartialUsage ? 'true' : 'false',
       }),
     onSuccess: () => {
       queryClient.invalidateQueries({ queryKey: ['spoolman-settings'] });
@@ -246,6 +254,53 @@ export function SpoolmanSettings() {
           </p>
         </div>
 
+        {/* Disable Weight Sync toggle - only show when sync mode is auto */}
+        {localSyncMode === 'auto' && (
+          <div className="flex items-center justify-between">
+            <div>
+              <p className="text-white">Disable AMS Estimated Weight Sync</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.
+              </p>
+            </div>
+            <label className="relative inline-flex items-center cursor-pointer">
+              <input
+                type="checkbox"
+                checked={localDisableWeightSync}
+                onChange={(e) => setLocalDisableWeightSync(e.target.checked)}
+                disabled={!localEnabled}
+                className="sr-only peer"
+              />
+              <div className="w-11 h-6 bg-bambu-dark-tertiary peer-focus:outline-none rounded-full peer peer-checked:after:translate-x-full peer-checked:after:border-white after:content-[''] after:absolute after:top-[2px] after:left-[2px] after:bg-white after:rounded-full after:h-5 after:w-5 after:transition-all peer-checked:bg-bambu-green"></div>
+            </label>
+          </div>
+        )}
+
+        {/* Report Partial Usage toggle - only show when weight sync is disabled */}
+        {localDisableWeightSync && (
+          <div className="flex items-center justify-between">
+            <div>
+              <p className="text-white">Report Partial Usage for Failed Prints</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.
+              </p>
+            </div>
+            <label className="relative inline-flex items-center cursor-pointer">
+              <input
+                type="checkbox"
+                checked={localReportPartialUsage}
+                onChange={(e) => setLocalReportPartialUsage(e.target.checked)}
+                disabled={!localEnabled}
+                className="sr-only peer"
+              />
+              <div className="w-11 h-6 bg-bambu-dark-tertiary peer-focus:outline-none rounded-full peer peer-checked:after:translate-x-full peer-checked:after:border-white after:content-[''] after:absolute after:top-[2px] after:left-[2px] after:bg-white after:rounded-full after:h-5 after:w-5 after:transition-all peer-checked:bg-bambu-green"></div>
+            </label>
+          </div>
+        )}
+
         {/* Connection status */}
         {localEnabled && (
           <div className="pt-2 border-t border-bambu-dark-tertiary">