fix(queue): scope chamber preheat to mapped trays

backend/app/services/preheat_chamber.py

@@ -0,0 +1,101 @@
+"""Resolve chamber preheat targets from the trays a print actually uses."""
+
+import json
+from collections.abc import Mapping
+
+_EXTERNAL_TRAY_ID_MIN = 254
+
+
+def normalize_filament_type(tray_type: str | None) -> str:
+    """Return the filament-map key reported by Bambu tray telemetry."""
+    return tray_type.split()[0].upper() if tray_type else ""
+
+
+def _int_or(value: object, default: int) -> int:
+    try:
+        return int(value)  # type: ignore[arg-type]
+    except (TypeError, ValueError):
+        return default
+
+
+def _global_tray_id(ams_id: int, tray_id: int) -> int:
+    """Mirror the global IDs produced by scheduler AMS mapping."""
+    return ams_id if ams_id >= 128 else ams_id * 4 + tray_id
+
+
+def _used_global_tray_ids(mapping_raw: object) -> set[int] | None:
+    """Return mapped tray IDs, or ``None`` when the mapping is unusable."""
+    if not mapping_raw:
+        return None
+    if isinstance(mapping_raw, str):
+        try:
+            mapping = json.loads(mapping_raw)
+        except (json.JSONDecodeError, TypeError):
+            return None
+    else:
+        mapping = mapping_raw
+    if not isinstance(mapping, list):
+        return None
+    used = {tray for tray in mapping if isinstance(tray, int) and not isinstance(tray, bool) and tray >= 0}
+    return used or None
+
+
+def _target_for_type(tray_type: object, targets: Mapping[str, int]) -> int:
+    normalized = normalize_filament_type(tray_type if isinstance(tray_type, str) else None)
+    if not normalized:
+        return 0
+    return targets.get(normalized, targets.get("DEFAULT", targets.get("default", 0)))
+
+
+def derive_chamber_target(
+    raw_data: object,
+    targets: Mapping[str, int],
+    mapping_raw: object = None,
+) -> int:
+    """Find the maximum target among the mapped AMS or external trays.
+
+    Missing or unusable mappings conservatively retain the previous behavior:
+    all loaded AMS trays are considered, while external trays are considered
+    only when the print mapping names them explicitly.
+    """
+    if not isinstance(raw_data, dict):
+        return 0
+
+    used = _used_global_tray_ids(mapping_raw)
+    ams_entries = raw_data.get("ams")
+    if isinstance(ams_entries, dict):
+        ams_entries = ams_entries.get("ams") or []
+    if not isinstance(ams_entries, list):
+        ams_entries = []
+
+    best = 0
+    for ams in ams_entries:
+        if not isinstance(ams, dict):
+            continue
+        ams_id = _int_or(ams.get("id"), 0)
+        trays = ams.get("tray") or []
+        if not isinstance(trays, list):
+            continue
+        for tray in trays:
+            if not isinstance(tray, dict):
+                continue
+            tray_id = _global_tray_id(ams_id, _int_or(tray.get("id"), 0))
+            if used is not None and tray_id not in used:
+                continue
+            best = max(best, _target_for_type(tray.get("tray_type"), targets))
+
+    if used is None or not any(tray_id >= _EXTERNAL_TRAY_ID_MIN for tray_id in used):
+        return best
+
+    virtual_trays = raw_data.get("vt_tray") or []
+    if isinstance(virtual_trays, dict):
+        virtual_trays = [virtual_trays]
+    if not isinstance(virtual_trays, list):
+        return best
+    for tray in virtual_trays:
+        if not isinstance(tray, dict):
+            continue
+        if _int_or(tray.get("id"), _EXTERNAL_TRAY_ID_MIN) not in used:
+            continue
+        best = max(best, _target_for_type(tray.get("tray_type"), targets))
+    return best

backend/app/services/print_scheduler.py

@@ -42,6 +42,7 @@
     fail_missing_library_source,
 )
 from backend.app.services.notification_service import notification_service
+from backend.app.services.preheat_chamber import derive_chamber_target, normalize_filament_type
 from backend.app.services.print_completion_identity import StrandedPrintRecovery, clear_stale_dispatch_claims
 from backend.app.services.printer_manager import (
     printer_manager,
@@ -2500,44 +2501,18 @@ async def _get_preheat_filament_targets(self, db: AsyncSession) -> dict[str, int
 
     @staticmethod
     def _normalize_filament_type(tray_type: str) -> str:
-        """Reduce the printer's tray_type to a preset-lookup key. Mirrors the
-        existing drying-preset normalisation (split-at-space, upper-case) so
-        the two maps share vocabulary — "PLA Basic" → "PLA", "PA-CF" stays
-        "PA-CF" (no space to split on)."""
-        return tray_type.split()[0].upper() if tray_type else ""
+        return normalize_filament_type(tray_type)
 
     def _derive_chamber_target(
         self,
         printer: Printer,
         targets: dict[str, int],
+        item: PrintQueueItem | None = None,
     ) -> int:
-        """Look up the chamber target for each loaded AMS tray and return the
-        max. Returns 0 when no AMS data is available (e.g. external-spool
-        prints) or when every loaded slot maps to 0 — the chamber phase then
-        short-circuits in the main loop.
-
-        Reads from `printer_manager.get_status(...).raw_data['ams']`, which is
-        the same source the dispatcher uses for AMS slot mapping. Empty / RFID-
-        less slots have empty `tray_type` and contribute nothing."""
+        """Return the target required by the trays this queue item maps."""
         state = printer_manager.get_status(printer.id)
-        if state is None:
-            return 0
-        ams_list = (state.raw_data or {}).get("ams") if state.raw_data else None
-        # Older Bambu firmware nests AMS as {"ams": {"ams": [...]}} — try both.
-        if isinstance(ams_list, dict):
-            ams_list = ams_list.get("ams") or []
-        if not isinstance(ams_list, list):
-            return 0
-        best = 0
-        for ams in ams_list:
-            for tray in (ams.get("tray") or []) if isinstance(ams, dict) else []:
-                normalised = self._normalize_filament_type(tray.get("tray_type") or "")
-                if not normalised:
-                    continue
-                target = targets.get(normalised, targets.get("DEFAULT", 0))
-                if target > best:
-                    best = target
-        return best
+        mapping = getattr(item, "ams_mapping", None)
+        return derive_chamber_target(getattr(state, "raw_data", None), targets, mapping)
 
     async def _preheat_and_soak(
         self,
@@ -2554,8 +2529,8 @@ async def _preheat_and_soak(
              even if the global is off.
           2. Chamber target — `item.preheat_chamber_target_override` if non-null;
              else max of `preheat_filament_targets[normalize(t.tray_type)]`
-             across loaded AMS slots; else 0 (skips chamber phase, keeps bed
-             phase + soak timer).
+             across the trays selected by this plate's AMS mapping; else 0
+             (skips chamber phase, keeps bed phase + soak timer).
           3. Three hardware tiers branch the wait loop:
              - Chamber heater (H2C/H2D/H2DPro/H2S/X2D/X1E via supports_chamber_heater):
                send M141 to the resolved target, then wait for the chamber sensor
@@ -2591,9 +2566,8 @@ async def _preheat_and_soak(
 
         # Chamber target resolution:
         #   1. Explicit per-item override beats everything (user knows best).
-        #   2. Otherwise derive from loaded AMS filament types via the per-
-        #      filament target map. PLA-only print derives 0 → chamber phase
-        #      auto-skips without the user touching anything.
+        #   2. Otherwise derive from the filament trays this plate maps via the
+        #      per-filament target map. A parked ASA spool cannot affect PLA.
         explicit_target = getattr(item, "preheat_chamber_target_override", None)
         if explicit_target is not None and explicit_target > 0:
             chamber_target = int(explicit_target)
@@ -2603,7 +2577,7 @@ async def _preheat_and_soak(
             chamber_source = "item-override-zero"
         else:
             targets = await self._get_preheat_filament_targets(db)
-            chamber_target = self._derive_chamber_target(printer, targets)
+            chamber_target = self._derive_chamber_target(printer, targets, item)
             chamber_source = "filament-map"
 
         plate_id = getattr(item, "plate_id", None)

backend/tests/unit/test_issue_139_chamber_preheat.py

@@ -0,0 +1,187 @@
+"""Plate-scoped chamber preheat regression coverage for issue #139."""
+
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services.print_scheduler import PrintScheduler
+
+TARGETS = {"PLA": 0, "PETG": 0, "ASA": 45, "PA": 50, "DEFAULT": 0}
+PLA_TRAY = 1
+ASA_TRAY = 2
+
+
+@pytest.fixture
+def scheduler():
+    return PrintScheduler()
+
+
+def _item(mapping=None, **overrides):
+    values = {
+        "id": 139,
+        "ams_mapping": mapping,
+        "preheat_override": "inherit",
+        "preheat_chamber_target_override": None,
+        "plate_id": 1,
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+def _state(*, ams=None, vt_tray=None, bed=60.0, chamber=50.0):
+    raw_data = {}
+    if ams is not None:
+        raw_data["ams"] = ams
+    if vt_tray is not None:
+        raw_data["vt_tray"] = vt_tray
+    return SimpleNamespace(
+        raw_data=raw_data,
+        temperatures={"bed": bed, "chamber": chamber},
+        airduct_mode=1,
+    )
+
+
+def _mixed_ams():
+    return [
+        {
+            "id": "0",
+            "tray": [
+                {"id": "0", "tray_type": "PETG Pro"},
+                {"id": "1", "tray_type": "PLA"},
+                {"id": "2", "tray_type": "ASA"},
+            ],
+        }
+    ]
+
+
+def _derive(scheduler, state, item, *, model="P2S"):
+    with patch("backend.app.services.print_scheduler.printer_manager") as manager:
+        manager.get_status.return_value = state
+        return scheduler._derive_chamber_target(SimpleNamespace(id=7, model=model), TARGETS, item)
+
+
+def test_fixed_printer_pla_job_ignores_parked_asa(scheduler):
+    mapping = json.dumps([-1, -1, -1, PLA_TRAY])
+    assert _derive(scheduler, _state(ams=_mixed_ams()), _item(mapping)) == 0
+
+
+@pytest.mark.asyncio
+async def test_preheat_stage_uses_the_assigned_items_mapping(scheduler):
+    item = _item("[1]", preheat_override="on", plate_id=None)
+    client = MagicMock()
+    client.set_bed_temperature = MagicMock(return_value=True)
+    client.set_chamber_temperature = MagicMock(return_value=True)
+    client.set_airduct_mode = MagicMock(return_value=True)
+
+    with (
+        patch.object(scheduler, "_get_int_setting", AsyncMock(return_value=0)),
+        patch.object(scheduler, "_get_setting", AsyncMock(return_value=None)),
+        patch("backend.app.services.print_scheduler.printer_manager") as manager,
+        patch("backend.app.services.print_scheduler.asyncio.sleep", AsyncMock()),
+    ):
+        manager.get_status.return_value = _state(ams=_mixed_ams())
+        manager.get_client.return_value = client
+        await scheduler._preheat_and_soak(
+            MagicMock(),
+            item,
+            SimpleNamespace(id=7, model="H2D"),
+            SimpleNamespace(bed_temperature=60),
+        )
+
+    client.set_bed_temperature.assert_called_once_with(60)
+    client.set_chamber_temperature.assert_not_called()
+
+
+def test_model_assigned_high_temperature_job_still_preheats(scheduler):
+    mapping = json.dumps([-1, ASA_TRAY])
+    assert _derive(scheduler, _state(ams=_mixed_ams()), _item(mapping), model="H2D") == 45
+
+
+def test_multi_material_job_uses_maximum_of_only_mapped_trays(scheduler):
+    mapping = json.dumps([PLA_TRAY, ASA_TRAY])
+    assert _derive(scheduler, _state(ams=_mixed_ams()), _item(mapping)) == 45
+
+
+@pytest.mark.parametrize("mapping", [None, "", "[-1, -1]", "[null]", "[]", "not-json", '{"tray": 1}'])
+def test_unknown_mapping_conservatively_scans_loaded_ams(scheduler, mapping):
+    assert _derive(scheduler, _state(ams=_mixed_ams()), _item(mapping)) == 45
+
+
+def test_regular_and_ams_ht_global_ids_match_dispatch_mapping(scheduler):
+    ams = _mixed_ams() + [
+        {"id": 1, "tray": [{"id": 2, "tray_type": "PA"}]},
+        {"id": 128, "tray": [{"id": 0, "tray_type": "ASA"}]},
+    ]
+    assert _derive(scheduler, _state(ams=ams), _item("[6]")) == 50
+    assert _derive(scheduler, _state(ams=ams), _item("[128]")) == 45
+    assert _derive(scheduler, _state(ams=ams), _item("[1]")) == 0
+
+
+def test_external_spool_is_used_only_when_mapping_names_it(scheduler):
+    state = _state(
+        ams=[{"id": 0, "tray": [{"id": 0, "tray_type": "PLA"}]}],
+        vt_tray=[{"id": 254, "tray_type": "ASA"}],
+    )
+    assert _derive(scheduler, state, _item("[254]")) == 45
+    assert _derive(scheduler, state, _item("[0]")) == 0
+    assert _derive(scheduler, state, _item(None)) == 0
+
+
+def test_selected_cross_model_variant_supplies_its_own_mapping(scheduler):
+    item = _item("[2]")
+    item.library_file_id = 1
+    item.library_file = SimpleNamespace(id=1)
+    item.archive_id = None
+    item.archive = None
+    item.target_model = "H2S"
+    item.nozzle_mapping = None
+    item.filament_overrides = None
+    item.required_filament_types = None
+    item.print_time_seconds = 10
+    variant = SimpleNamespace(
+        library_file_id=2,
+        library_file=SimpleNamespace(id=2),
+        target_model="H2C",
+        plate_id=3,
+        ams_mapping="[1]",
+        nozzle_mapping="[0]",
+        filament_overrides=None,
+        required_filament_types='["PLA"]',
+        print_time_seconds=20,
+    )
+
+    scheduler._resolve_variant(item, SimpleNamespace(variant=variant))
+
+    assert item.plate_id == 3
+    assert item.ams_mapping == "[1]"
+    assert _derive(scheduler, _state(ams=_mixed_ams()), item, model="H2C") == 0
+
+
+@pytest.mark.asyncio
+async def test_explicit_zero_override_still_disables_mapped_asa(scheduler):
+    item = _item("[2]", preheat_override="on", preheat_chamber_target_override=0)
+    client = MagicMock()
+    client.set_bed_temperature = MagicMock(return_value=True)
+    client.set_chamber_temperature = MagicMock(return_value=True)
+    client.set_airduct_mode = MagicMock(return_value=True)
+    state = _state(ams=_mixed_ams())
+
+    with (
+        patch.object(scheduler, "_get_int_setting", AsyncMock(return_value=0)),
+        patch("backend.app.services.print_scheduler.extract_bed_temperature_from_3mf", return_value=60),
+        patch("backend.app.services.print_scheduler.printer_manager") as manager,
+        patch("backend.app.services.print_scheduler.asyncio.sleep", AsyncMock()),
+    ):
+        manager.get_status.return_value = state
+        manager.get_client.return_value = client
+        await scheduler._preheat_and_soak(
+            MagicMock(),
+            item,
+            SimpleNamespace(id=7, model="H2D"),
+            SimpleNamespace(bed_temperature=60, file_path="archives/job.3mf"),
+        )
+
+    client.set_bed_temperature.assert_called_once_with(60)
+    client.set_chamber_temperature.assert_not_called()

tools/check_source_size_budget.py

@@ -88,7 +88,7 @@
     "backend/app/main.py": 6831,
     "backend/app/services/bambu_mqtt.py": 5928,
     "backend/app/services/notification_service.py": 2184,
-    "backend/app/services/print_scheduler.py": 3826,
+    "backend/app/services/print_scheduler.py": 3800,
     "backend/tests/integration/test_mfa_api.py": 5132,
     "backend/tests/integration/test_printers_api.py": 4100,
     "backend/tests/unit/services/test_bambu_mqtt.py": 6746,