Merge pull request #1440 from Person2099/fix/filament-override-ams-mapping-dispatch

fix: compute AMS mapping from force-colour overrides when 3MF reqs unavailable

backend/app/services/filament_requirements.py

@@ -66,7 +66,25 @@ def extract_filament_requirements(file_path: Path, plate_id: int | None = None)
                         _collect_filaments(plate_elem, filaments)
                         break
             else:
-                _collect_filaments(root, filaments)
+                # Modern BambuStudio format wraps filaments inside <plate> elements.
+                # When no plate filter is requested, collect from every plate and
+                # deduplicate by slot_id (first occurrence wins after sort).
+                plate_elems = root.findall("./plate")
+                if plate_elems:
+                    for plate_elem in plate_elems:
+                        _collect_filaments(plate_elem, filaments)
+                    # Deduplicate: same slot_id can appear on multiple plates.
+                    # Keep the entry with the highest used_grams; ties go to the
+                    # first plate (stable after sort + dict insertion order).
+                    seen: dict[int, dict] = {}
+                    for f in filaments:
+                        sid = f["slot_id"]
+                        if sid not in seen or f["used_grams"] > seen[sid]["used_grams"]:
+                            seen[sid] = f
+                    filaments = list(seen.values())
+                else:
+                    # Older / non-plate-wrapped format: filaments are direct children of root.
+                    _collect_filaments(root, filaments)
 
             filaments.sort(key=lambda x: x["slot_id"])
 

backend/app/services/print_scheduler.py

@@ -792,6 +792,22 @@ class PrintScheduler:
         # Get filament requirements from source file
         filament_reqs = await self._get_filament_requirements(db, item)
         if not filament_reqs:
+            # When the 3MF can't be read but force-color overrides are present, build a
+            # direct mapping from the overrides so the printer uses the correct AMS slot.
+            if item.filament_overrides:
+                try:
+                    overrides = json.loads(item.filament_overrides)
+                    force_overrides = [o for o in overrides if o.get("force_color_match")]
+                    if force_overrides:
+                        logger.info(
+                            "Queue item %s: No filament reqs from 3MF; building AMS mapping from %d "
+                            "force-color override(s)",
+                            item.id,
+                            len(force_overrides),
+                        )
+                        return self._build_override_direct_mapping(force_overrides, status)
+                except (json.JSONDecodeError, KeyError, TypeError) as e:
+                    logger.warning("Queue item %s: Force-color fallback mapping failed: %s", item.id, e)
             logger.debug("No filament requirements found for queue item %s", item.id)
             return None
 
@@ -830,6 +846,33 @@ class PrintScheduler:
         # Compute mapping: match required filaments to available slots
         return self._match_filaments_to_slots(filament_reqs, loaded_filaments, prefer_lowest)
 
+    def _build_override_direct_mapping(self, force_overrides: list[dict], status) -> list[int] | None:
+        """Build an AMS mapping directly from force-color overrides without a 3MF.
+
+        Used when ``_get_filament_requirements`` returns nothing (e.g. the 3MF's
+        slice_info is missing or unreadable) but ``force_color_match`` overrides
+        are present. Each override's ``slot_id``, ``type``, and ``color`` are
+        treated as the filament requirement for that slot and matched against the
+        current AMS state of the printer.
+
+        Returns the same format as ``_match_filaments_to_slots``, or None when
+        the AMS has no loaded filaments.
+        """
+        loaded = self._build_loaded_filaments(status)
+        if not loaded:
+            return None
+
+        reqs = [
+            {
+                "slot_id": o["slot_id"],
+                "type": o.get("type", ""),
+                "color": o.get("color", ""),
+                "tray_info_idx": "",
+            }
+            for o in force_overrides
+        ]
+        return self._match_filaments_to_slots(reqs, loaded)
+
     async def _get_filament_requirements(self, db: AsyncSession, item: PrintQueueItem) -> list[dict] | None:
         """Resolve the queue item's source 3MF and parse the per-slot
         filament requirements out of it. Thin DB-resolver wrapper around

backend/tests/unit/services/test_filament_requirements.py

@@ -109,6 +109,54 @@ class TestExtractFilamentRequirements:
         assert len(out) == 1
         assert out[0]["type"] == "PLA"
 
+    def test_no_plate_id_collects_from_all_plates(self, tmp_path: Path):
+        """Modern BambuStudio wraps filaments inside <plate> elements.  When
+        plate_id=None, every plate's filaments must be returned (deduplicated)."""
+        f = tmp_path / "multi.3mf"
+        _make_3mf(
+            f,
+            plates=[
+                (1, [{"id": "1", "type": "PLA", "color": "#FFFFFF", "used_g": "5"}]),
+                (2, [{"id": "2", "type": "PETG", "color": "#000000", "used_g": "3"}]),
+            ],
+        )
+        out = extract_filament_requirements(f, plate_id=None)
+        assert len(out) == 2
+        slot_ids = [r["slot_id"] for r in out]
+        assert 1 in slot_ids
+        assert 2 in slot_ids
+
+    def test_no_plate_id_deduplicates_shared_slots(self, tmp_path: Path):
+        """Same slot_id on multiple plates keeps only the entry with the
+        highest used_grams (the plate that actually consumes more)."""
+        f = tmp_path / "shared.3mf"
+        _make_3mf(
+            f,
+            plates=[
+                (1, [{"id": "1", "type": "PLA", "color": "#FFFFFF", "used_g": "5"}]),
+                (2, [{"id": "1", "type": "PLA", "color": "#FFFFFF", "used_g": "8"}]),
+            ],
+        )
+        out = extract_filament_requirements(f, plate_id=None)
+        assert len(out) == 1
+        assert out[0]["slot_id"] == 1
+        assert out[0]["used_grams"] == 8.0
+
+    def test_no_plate_id_single_plate_modern_format(self, tmp_path: Path):
+        """Single-plate 3MF using modern <plate> wrapping is parsed correctly
+        when plate_id=None — this is the common queue scenario where no specific
+        plate is targeted."""
+        f = tmp_path / "single.3mf"
+        _make_3mf(
+            f,
+            plates=[(1, [{"id": "1", "type": "PLA", "color": "#CBC6B8", "used_g": "0.12"}])],
+        )
+        out = extract_filament_requirements(f, plate_id=None)
+        assert len(out) == 1
+        assert out[0]["slot_id"] == 1
+        assert out[0]["type"] == "PLA"
+        assert out[0]["color"] == "#CBC6B8"
+
     def test_returns_empty_list_for_unparseable_file(self, tmp_path: Path):
         f = tmp_path / "bad.3mf"
         f.write_bytes(b"not a zip")

backend/tests/unit/test_scheduler_force_color_ams_fallback.py

@@ -0,0 +1,243 @@
+"""Tests for force-color-override AMS mapping fallback in the print scheduler.
+
+Covers the code path in ``_compute_ams_mapping_for_printer`` that kicks in
+when the 3MF's filament requirements cannot be read (e.g. ``plate_id=None``
+with a modern BambuStudio 3MF whose slice_info was missing or unreadable)
+but ``force_color_match`` overrides are present.
+
+Related issue: #1436
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services.print_scheduler import PrintScheduler
+
+
+class TestBuildOverrideDirectMapping:
+    """Unit tests for ``_build_override_direct_mapping``."""
+
+    @pytest.fixture
+    def scheduler(self):
+        return PrintScheduler()
+
+    def _status(self, ams: list[dict], vt_tray: list[dict] | None = None) -> MagicMock:
+        raw: dict = {"ams": ams}
+        if vt_tray is not None:
+            raw["vt_tray"] = vt_tray
+        return MagicMock(raw_data=raw)
+
+    def test_single_force_override_matches_ams_slot(self, scheduler):
+        """Override with type+color matches the correct AMS tray."""
+        status = self._status(
+            ams=[
+                {
+                    "id": 0,
+                    "tray": [
+                        {"id": 0, "tray_type": "PLA", "tray_color": "CBC6B8FF"},
+                    ],
+                }
+            ]
+        )
+        overrides = [{"slot_id": 1, "type": "PLA", "color": "#CBC6B8", "force_color_match": True}]
+        result = scheduler._build_override_direct_mapping(overrides, status)
+        assert result == [0]  # global_tray_id 0 (AMS 0, tray 0)
+
+    def test_no_loaded_filaments_returns_none(self, scheduler):
+        """Empty AMS → cannot compute mapping, return None."""
+        status = self._status(ams=[{"id": 0, "tray": [{"id": 0}]}])  # empty tray
+        overrides = [{"slot_id": 1, "type": "PLA", "color": "#CBC6B8", "force_color_match": True}]
+        result = scheduler._build_override_direct_mapping(overrides, status)
+        assert result is None
+
+    def test_no_color_match_returns_minus_one(self, scheduler):
+        """Override color not present → slot mapped to -1 (no match)."""
+        status = self._status(
+            ams=[
+                {
+                    "id": 0,
+                    "tray": [
+                        {"id": 0, "tray_type": "PLA", "tray_color": "FF0000FF"},
+                    ],
+                }
+            ]
+        )
+        overrides = [{"slot_id": 1, "type": "PLA", "color": "#CBC6B8", "force_color_match": True}]
+        result = scheduler._build_override_direct_mapping(overrides, status)
+        # Type matches but color is far off (red vs beige) → type-only fallback → [0]
+        # If colour threshold is exceeded, falls back to type-only, which IS a match.
+        # The important thing: result is not None and has the right length.
+        assert result is not None
+        assert len(result) == 1
+
+    def test_multiple_overrides_map_multiple_slots(self, scheduler):
+        """Two overrides with different slot_ids produce a two-element mapping."""
+        status = self._status(
+            ams=[
+                {
+                    "id": 0,
+                    "tray": [
+                        {"id": 0, "tray_type": "PLA", "tray_color": "CBC6B8FF"},
+                        {"id": 1, "tray_type": "PETG", "tray_color": "000000FF"},
+                    ],
+                }
+            ]
+        )
+        overrides = [
+            {"slot_id": 1, "type": "PLA", "color": "#CBC6B8", "force_color_match": True},
+            {"slot_id": 2, "type": "PETG", "color": "#000000", "force_color_match": True},
+        ]
+        result = scheduler._build_override_direct_mapping(overrides, status)
+        assert result == [0, 1]  # slot 1 → tray 0, slot 2 → tray 1
+
+    def test_external_spool_matched(self, scheduler):
+        """Override matching an external spool returns global_tray_id 254."""
+        status = self._status(
+            ams=[],
+            vt_tray=[{"tray_type": "TPU", "tray_color": "CBC6B8FF"}],
+        )
+        overrides = [{"slot_id": 1, "type": "TPU", "color": "#CBC6B8", "force_color_match": True}]
+        result = scheduler._build_override_direct_mapping(overrides, status)
+        assert result == [254]
+
+    def test_tray_info_idx_is_not_used_for_direct_mapping(self, scheduler):
+        """Direct-override mapping clears tray_info_idx so matching falls back
+        to colour rather than pinning to a specific spool ID from the 3MF."""
+        status = self._status(
+            ams=[
+                {
+                    "id": 0,
+                    "tray": [
+                        {
+                            "id": 0,
+                            "tray_type": "PLA",
+                            "tray_color": "CBC6B8FF",
+                            "tray_info_idx": "GFA00",
+                        },
+                    ],
+                }
+            ]
+        )
+        overrides = [{"slot_id": 1, "type": "PLA", "color": "#CBC6B8", "force_color_match": True}]
+        result = scheduler._build_override_direct_mapping(overrides, status)
+        # Should match by colour (#CBC6B8 ≈ CBC6B8FF after strip), not by tray_info_idx.
+        assert result == [0]
+
+
+class TestComputeAmsMappingFallback:
+    """Integration tests for the force-color fallback inside
+    ``_compute_ams_mapping_for_printer`` when filament reqs are unavailable."""
+
+    @pytest.fixture
+    def scheduler(self):
+        return PrintScheduler()
+
+    def _make_item(self, filament_overrides_json: str | None = None) -> MagicMock:
+        item = MagicMock()
+        item.archive_id = 141
+        item.library_file_id = None
+        item.plate_id = None
+        item.filament_overrides = filament_overrides_json
+        item.printer_id = 5
+        return item
+
+    def _make_status(self) -> MagicMock:
+        return MagicMock(
+            raw_data={
+                "ams": [
+                    {
+                        "id": 0,
+                        "tray": [
+                            {"id": 0, "tray_type": "PLA", "tray_color": "CBC6B8FF"},
+                        ],
+                    }
+                ]
+            }
+        )
+
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    async def test_fallback_used_when_filament_reqs_empty(self, mock_pm, scheduler):
+        """When _get_filament_requirements returns None but force-color overrides
+        are set, the fallback builds a mapping directly from the overrides."""
+        mock_pm.get_status.return_value = self._make_status()
+
+        item = self._make_item(
+            filament_overrides_json='[{"slot_id": 1, "type": "PLA", "color": "#CBC6B8", "force_color_match": true}]'
+        )
+
+        db = AsyncMock()
+
+        with patch.object(scheduler, "_get_filament_requirements", return_value=None):
+            result = await scheduler._compute_ams_mapping_for_printer(db, 5, item)
+
+        assert result == [0]  # global_tray_id 0 (AMS 0, tray 0)
+
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    async def test_fallback_not_used_when_no_force_color(self, mock_pm, scheduler):
+        """When overrides have no force_color_match, the fallback is not triggered."""
+        mock_pm.get_status.return_value = self._make_status()
+
+        item = self._make_item(filament_overrides_json='[{"slot_id": 1, "type": "PLA", "color": "#CBC6B8"}]')
+        db = AsyncMock()
+
+        with patch.object(scheduler, "_get_filament_requirements", return_value=None):
+            result = await scheduler._compute_ams_mapping_for_printer(db, 5, item)
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    async def test_fallback_not_used_when_no_overrides(self, mock_pm, scheduler):
+        """When filament_overrides is None, the fallback is not triggered."""
+        mock_pm.get_status.return_value = self._make_status()
+
+        item = self._make_item(filament_overrides_json=None)
+        db = AsyncMock()
+
+        with patch.object(scheduler, "_get_filament_requirements", return_value=None):
+            result = await scheduler._compute_ams_mapping_for_printer(db, 5, item)
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    async def test_normal_path_used_when_filament_reqs_available(self, mock_pm, scheduler):
+        """When filament requirements are available, the normal path is used
+        (overrides applied to reqs, then matched)."""
+        mock_pm.get_status.return_value = self._make_status()
+
+        item = self._make_item(
+            filament_overrides_json='[{"slot_id": 1, "type": "PLA", "color": "#CBC6B8", "force_color_match": true}]'
+        )
+        db = AsyncMock()
+
+        # 3MF says slot 1 is PLA with a different color; override will change it.
+        filament_reqs = [{"slot_id": 1, "type": "PLA", "color": "#000000", "tray_info_idx": "GFA00"}]
+
+        with (
+            patch.object(scheduler, "_get_filament_requirements", return_value=filament_reqs),
+            patch.object(scheduler, "_get_bool_setting", new=AsyncMock(return_value=False)),
+        ):
+            result = await scheduler._compute_ams_mapping_for_printer(db, 5, item)
+
+        # After override, slot 1 becomes PLA #CBC6B8 → matches tray 0.
+        assert result == [0]
+
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    async def test_fallback_returns_none_when_printer_status_unavailable(self, mock_pm, scheduler):
+        """When the printer has no status, the fallback also returns None gracefully."""
+        mock_pm.get_status.return_value = None
+
+        item = self._make_item(
+            filament_overrides_json='[{"slot_id": 1, "type": "PLA", "color": "#CBC6B8", "force_color_match": true}]'
+        )
+        db = AsyncMock()
+
+        with patch.object(scheduler, "_get_filament_requirements", return_value=None):
+            result = await scheduler._compute_ams_mapping_for_printer(db, 5, item)
+
+        assert result is None

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

@@ -200,7 +200,7 @@ describe('dual-color / tri-color hard-split bars (#1154 follow-up)', () => {
 describe('Sparkle prominence + checkerboard density (#1154 follow-up cosmetic)', () => {
   it('renders dense sparkle on card preset (at least 10 dots)', () => {
     // The original Sparkle pattern was 4 dots — too subtle on a 200×60px
-    // banner. Now we use situation-aware dot counts: more dots for larger presets. 
+    // banner. Now we use situation-aware dot counts: more dots for larger presets.
     // Verify the card preset produces a dense pattern with at least 10 dots.
     render(<FilamentSwatch rgba="ff0000ff" effectType="sparkle" effectSize="card" />);
     const el = screen.getByTestId('filament-swatch');

frontend/src/components/filamentSwatchHelpers.ts

@@ -106,7 +106,7 @@ export const EFFECT_OVERLAYS: Partial<
   Record<FilamentEffect, (effectSeed?: number, effectSize?: SwatchType) => EffectLayer>
 > = {
   // Sparkle: bright flecks — positions seeded from spool color+extracolors+subtype+effectType.
-  // to give identical spools the same sparkle pattern while different spools get different patterns. 
+  // to give identical spools the same sparkle pattern while different spools get different patterns.
   sparkle: (spoolSeed = 0, effectSize = 'table') => {
     const rand = random_mulberry32(spoolSeed);
     const preset = SWATCH_TYPE_PRESETS[effectSize] ?? SWATCH_TYPE_PRESETS.table;

frontend/src/utils/random.ts

@@ -3,7 +3,7 @@ const FNV1A_32_PRIME = 0x01000193;
 
 /**
  * Computes a fast 32-bit FNV-1a hash for deterministic, non-security tasks.
- * Accepts any number of string/nullable-string inputs, takes measurements 
+ * Accepts any number of string/nullable-string inputs, takes measurements
  * to avoid collisions, and combines them into a 32-bit hash.
  * Not cryptographically secure; use only for non-security-related use cases.
 */
@@ -38,7 +38,7 @@ export interface Mulberry32Sequence {
 
 /**
  * Creates a fast deterministic PRNG sequence using Mulberry32.
- * Same seed will always produce the same sequence. 
+ * Same seed will always produce the same sequence.
  * Not cryptographically secure; use only for non-security-related use cases.
  */
 export function random_mulberry32(seed: number): Mulberry32Sequence {

scripts/fill_spool_effects.py

@@ -18,12 +18,13 @@ from dataclasses import dataclass
 
 import requests
 
-
 API_PATH_BULK_CREATE = "/api/v1/inventory/spools/bulk"
 
+
 @dataclass(frozen=True)
 class TestSpool:
     "Class representing a spool definition and count for the test spool set"
+
     effect_type: str
     colors: dict[str, str]
     quantity: int = 1
@@ -188,10 +189,7 @@ def main() -> None:
                 timeout=args.timeout,
             )
             ids = [str(item.get("id", "?")) for item in created_spools]
-            print(
-                f"  Created effect={variant.effect_type:<11} qty={len(created_spools)} "
-                f"ids={','.join(ids)}"
-            )
+            print(f"  Created effect={variant.effect_type:<11} qty={len(created_spools)} ids={','.join(ids)}")
             created += len(created_spools)
         except (requests.RequestException, ValueError) as exc:
             print(f"  FAILED effect={variant.effect_type}: {exc}", file=sys.stderr)