diff --git a/backend/app/services/filament_requirements.py b/backend/app/services/filament_requirements.py index b03280fd2..b342bf7a8 100644 --- a/backend/app/services/filament_requirements.py +++ b/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 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"]) diff --git a/backend/app/services/print_scheduler.py b/backend/app/services/print_scheduler.py index 9d0cc3f8d..c58672214 100644 --- a/backend/app/services/print_scheduler.py +++ b/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 diff --git a/backend/tests/unit/services/test_filament_requirements.py b/backend/tests/unit/services/test_filament_requirements.py index 7397eca27..3ad046f43 100644 --- a/backend/tests/unit/services/test_filament_requirements.py +++ b/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 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 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") diff --git a/backend/tests/unit/test_scheduler_force_color_ams_fallback.py b/backend/tests/unit/test_scheduler_force_color_ams_fallback.py new file mode 100644 index 000000000..c3c1c8d86 --- /dev/null +++ b/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 diff --git a/frontend/src/__tests__/components/FilamentSwatch.test.tsx b/frontend/src/__tests__/components/FilamentSwatch.test.tsx index 7ec3222ba..e6b07e721 100644 --- a/frontend/src/__tests__/components/FilamentSwatch.test.tsx +++ b/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(); const el = screen.getByTestId('filament-swatch'); diff --git a/frontend/src/components/filamentSwatchHelpers.ts b/frontend/src/components/filamentSwatchHelpers.ts index 43dadd9ff..665f68572 100644 --- a/frontend/src/components/filamentSwatchHelpers.ts +++ b/frontend/src/components/filamentSwatchHelpers.ts @@ -106,7 +106,7 @@ export const EFFECT_OVERLAYS: Partial< Record 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; diff --git a/frontend/src/utils/random.ts b/frontend/src/utils/random.ts index c7f228f8c..966e86c29 100644 --- a/frontend/src/utils/random.ts +++ b/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 { diff --git a/scripts/fill_spool_effects.py b/scripts/fill_spool_effects.py index aa8848df3..8e1d331e0 100644 --- a/scripts/fill_spool_effects.py +++ b/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)