diff --git a/backend/app/api/routes/inventory.py b/backend/app/api/routes/inventory.py index 18eadec7a..8354e9fd5 100644 --- a/backend/app/api/routes/inventory.py +++ b/backend/app/api/routes/inventory.py @@ -45,6 +45,7 @@ from backend.app.schemas.spool import ( normalize_extra_colors, ) from backend.app.schemas.spool_usage import SpoolUsageHistoryResponse +from backend.app.services.ams_slot_presence import spool_present from backend.app.services.location_service import ( DUPLICATE_LOCATION_NAME, assign_location_name, @@ -1850,6 +1851,9 @@ async def assign_spool( fingerprint_type = None current_tray_info_idx = "" tray_state: int | None = None + # Firmware's tray_exist_bits answer for this slot, when the payload carries + # one. Outranks tray_state below — see services/ams_slot_presence.py. + tray_has_spool: bool | None = None state = printer_manager.get_status(data.printer_id) if state and state.raw_data: if data.ams_id == 255: @@ -1864,6 +1868,7 @@ async def assign_spool( raw_state = vt.get("state") if isinstance(raw_state, int): tray_state = raw_state + tray_has_spool = spool_present(vt) break else: ams_data = state.raw_data.get("ams", {}) @@ -1886,6 +1891,7 @@ async def assign_spool( raw_state = tray.get("state") if isinstance(raw_state, int): tray_state = raw_state + tray_has_spool = spool_present(tray) # 3. Upsert assignment (replace if same printer+ams+tray) existing = await db.execute( @@ -1945,7 +1951,30 @@ async def assign_spool( # a doomed MQTT push when the firmware has positively confirmed "no # spool" — and to keep the on_ams_change replay path as the single # source of truth for those slots. - slot_is_definitely_empty = tray_state == 9 or tray_state == 10 + # + # ...except that `state` cannot carry that meaning. Two independent ways + # a loaded slot reads 9 here: + # + # - an AMS-HT reports its LOADED tray as 9, not 11, because it does not + # feed into a shared buffer the way a 4-slot AMS does (#2594, and the + # merge above skips its own state heuristic for HT units for exactly + # this reason). So this branch called every HT slot empty on sight. + # - apply_tray_exist_bits stamps state=9 on any slot whose tray_exist_bits + # bit is 0 and never takes it back when the bit returns, so a slot that + # was briefly emptied keeps the 9 until something configures it. + # + # Either way the slot sits at exists=True, state=9, this branch took the + # pending path, nothing was published, and the printer kept showing "?" + # (#3084 — reported against an H2C's AMS-HT, where both apply). Firmware's + # presence bit is what actually answers "is a spool in this slot", and the + # printer card has read it ahead of `state` since #2527. + # + # It is allowed to overrule the 9 and nothing else. A bit reading *empty* + # deliberately does NOT start suppressing pushes that go out today: the + # cost of being wrong there is a slot that silently stops configuring, on + # whichever AMS variant we compute the bit position wrong for, against a + # saving of one MQTT message the firmware would have dropped anyway. + slot_is_definitely_empty = tray_has_spool is not True and (tray_state == 9 or tray_state == 10) configured = False if not slot_is_definitely_empty: try: diff --git a/backend/app/main.py b/backend/app/main.py index 6491dc76b..fb6cc3ab8 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -2022,6 +2022,7 @@ async def on_ams_change(printer_id: int, ams_data: list): from backend.app.api.routes.inventory import _find_tray_in_ams_data from backend.app.models.spool import Spool as _Spool from backend.app.models.spool_assignment import SpoolAssignment as SA + from backend.app.services.ams_slot_presence import spool_present from backend.app.services.inventory_mode import spoolman_owns_assignments # Built-in assignments only. Since #2812 they survive a switch to @@ -2138,7 +2139,18 @@ async def on_ams_change(printer_id: int, ams_data: list): # (#1322). The state ∉ {9,10} guard keeps the firmware's # explicit "empty" signals authoritative over any stale # tray_type that might survive the relay's auto-clearing. - loaded = cur_state == 11 or (cur_state not in (9, 10) and cur_type.strip()) + # + # tray_exist_bits comes first because that guard cannot tell + # a firmware "empty" from Bambuddy's own: apply_tray_exist_bits + # writes state=9 when the bit is 0 and leaves it there when the + # bit returns. A non-RFID spool inserted into a pre-assigned + # slot brings no tray_type with it, so the stale 9 made this + # expression false forever and the deferred config never fired + # — the deadlock #1322 removed from the assign path, still in + # place here (#3084, #3100). + loaded = spool_present(current_tray) is True or ( + cur_state == 11 or (cur_state not in (9, 10) and cur_type.strip()) + ) if not fp_type.strip() and loaded and assignment.spool: try: from backend.app.api.routes.inventory import ( @@ -2188,6 +2200,23 @@ async def on_ams_change(printer_id: int, ams_data: list): assignment.tray_id, ) continue + # Same reasoning off the print, on firmware's own say-so: + # a blank tray report from a slot whose tray_exist_bits + # bit is set describes a spool the AMS cannot identify — + # a non-RFID one, or one whose slot was reset — not a + # spool that was taken out. Deleting the assignment there + # threw away the identity the user had supplied, which is + # the only place it existed (#3100). A slot the bit calls + # empty, or one that carries no bit at all, still unlinks. + if spool_present(current_tray) is True and not cur_color.strip() and not cur_type.strip(): + logger.info( + "Auto-unlink skipped: spool %d AMS%d-T%d — slot still occupied, " + "tray reports no filament data yet", + assignment.spool_id, + assignment.ams_id, + assignment.tray_id, + ) + continue # Fingerprint mismatch — but check if tray now matches the # assigned spool (e.g. auto-configure changed the tray). # Both sides are reduced to the type the slot can carry @@ -2594,6 +2623,7 @@ async def on_ams_change(printer_id: int, ams_data: list): from backend.app.models.spool_assignment import SpoolAssignment from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment + from backend.app.services.ams_slot_presence import spool_present from backend.app.services.inventory_mode import spoolman_owns_assignments # Built-in remaining weight, used by sync_ams_tray only when the @@ -2662,7 +2692,16 @@ async def on_ams_change(printer_id: int, ams_data: list): # completion (#1459), so deleting the row mid-print # loses the runout segment's usage — the same failure # the internal inventory's auto-unlink had. - if not printing_now: + # + # Nor when firmware's presence bit says the slot is + # occupied. parse_ams_tray calls a tray with no type or + # no colour empty, and a spool the AMS cannot read has + # neither until something configures it — so a tag-less + # spool assigned through the UI had its row deleted by + # the first idle push after it was inserted. Same + # deletion as the internal inventory's in #3100, same + # answer, so the two modes stay in step. + if not printing_now and spool_present(tray_data) is not True: empty_slots.append((ams_id, tray_id_raw)) _clear_unknown_tag_dedup(printer_id, ams_id, tray_id_raw) continue diff --git a/backend/app/services/ams_slot_presence.py b/backend/app/services/ams_slot_presence.py new file mode 100644 index 000000000..26d27cc71 --- /dev/null +++ b/backend/app/services/ams_slot_presence.py @@ -0,0 +1,60 @@ +"""Is there a spool in this AMS slot? + +Three backend decisions turn on the answer -- whether to push +``ams_filament_setting`` when a spool is assigned, whether a pre-assigned slot +has just been filled, and whether an assignment has gone stale -- and all three +used to read it off the tray's ``state`` field. That field cannot carry it. + +## Why ``state`` is the wrong source + +``state`` is firmware-variant. The A1 Mini BMCU and the P1S Standard AMS report +3 for a loaded slot and never emit 11; the AMS-HT's codes differ again (#2670). +Bambuddy already works around that on the printer card, where +``getEmptySlotKind`` (``PrintersPage.tsx``) reads the presence bit first and +only falls back to the 9/10 heuristic when there is no bit to read. + +Worse, ``state`` is partly Bambuddy's own writing. ``apply_tray_exist_bits`` +sets ``state = 9`` on every slot whose presence bit is 0 -- and when the bit +comes back it leaves the 9 exactly where it was, because the "slot occupied" +branch only annotates ``exists`` and moves on. So a slot the firmware says is +full can sit in the cache reading ``exists=True, state=9`` indefinitely. That +is #3084: a non-Bambu spool is swapped in, Assign Spool reads the stale 9, +calls the slot empty, sends no MQTT, and the printer keeps showing ``?``. The +deferred-configuration replay could not rescue it either, because its own +"loaded" test was the same 9/10 heuristic -- the exact deadlock #1322 removed +elsewhere. #3100 is the same stale 9 one step further on: the replay does not +fire, the assignment keeps the empty fingerprint it was stored with, and the +first real tray report is read as a spool swap and deleted. + +## What this module answers + +``tray_exist_bits`` is the firmware's own "which slots have a spool" bitmask -- +the one BambuStudio draws its ``?`` from -- and ``apply_tray_exist_bits`` +records it per tray as ``exists``. That bit is authoritative where it exists, +and absent otherwise; it is never a guess. Callers that want a decision for a +payload carrying no bit at all keep their own fallback, because the right +fallback differs per caller: the assign path wants to know whether the push is +doomed, the unlink pass wants to know whether a spool was removed, and a state +of 26 ("unloaded", mid-runout) answers those two questions differently. +""" + +from collections.abc import Mapping +from typing import Any + + +def spool_present(tray: Mapping[str, Any] | None) -> bool | None: + """Does firmware's presence bit say a spool is in this slot? + + ``True`` / ``False`` straight from ``tray_exist_bits``; ``None`` when the + tray carries no presence annotation, which means the caller has to decide + on its own terms rather than assume either way. + + Only the internal AMS path annotates ``exists`` (``apply_tray_exist_bits`` + is called with ``annotate_exists=True`` there and False for the VP bridge), + so the external spool's ``vt_tray`` entries answer ``None`` -- they have no + bit in the mask. + """ + if not isinstance(tray, Mapping): + return None + exists = tray.get("exists") + return exists if isinstance(exists, bool) else None diff --git a/backend/tests/integration/test_inventory_assign.py b/backend/tests/integration/test_inventory_assign.py index b620f70c1..58059443e 100644 --- a/backend/tests/integration/test_inventory_assign.py +++ b/backend/tests/integration/test_inventory_assign.py @@ -1226,6 +1226,221 @@ class TestAssignSpoolEmptyDetection: assert body["configured"] is True +class TestAssignSpoolPresenceBit: + """#3084: the slot the firmware says is full, and the cache says is empty. + + ``apply_tray_exist_bits`` stamps ``state = 9`` on every slot whose + ``tray_exist_bits`` bit is 0 and annotates ``exists`` on every slot it + looks at — but when the bit comes back it only refreshes ``exists`` and + leaves the 9 where it was. Swap a Bambu spool for a non-RFID one and the + slot sits at ``exists=True, state=9`` until something configures it. + + Reported on an H2D/H2C AMS-HT: remove the Bambu spool (bits ``f``), insert + a third-party one 9 seconds later (bits ``1000f``), then Assign Spool 28 + seconds after that — and no ``ams_filament_setting`` was published at all. + Configure worked, because it publishes unconditionally. + """ + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_the_bit_overrules_a_stale_empty_state( + self, async_client: AsyncClient, printer_factory, spool_factory + ): + """exists=True with a leftover state=9 — MQTT must fire.""" + printer = await printer_factory(name="H2D") + spool = await spool_factory(slicer_filament="GFL05", material="PLA") + + mock_client = MagicMock() + mock_client.ams_set_filament_setting.return_value = True + mock_client.extrusion_cali_sel.return_value = True + + tray_data = {"id": 0, "state": 9, "exists": True, "tray_type": "", "tray_color": "", "tray_info_idx": ""} + status = _make_mock_status(ams_data=[{"id": 128, "tray": [tray_data]}]) + + with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: + mock_pm.get_client.return_value = mock_client + mock_pm.get_status.return_value = status + + response = await async_client.post( + "/api/v1/inventory/assignments", + json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 128, "tray_id": 0}, + ) + + assert response.status_code == 200 + mock_client.ams_set_filament_setting.assert_called_once() + body = response.json() + assert body["configured"] is True + assert body["pending_config"] is False + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_an_empty_bit_does_not_start_suppressing_pushes( + self, async_client: AsyncClient, printer_factory, spool_factory + ): + """The bit overrules the 9 and nothing else. + + Reading it the other way too would be tidier — skip the push firmware + is going to drop — but it also means a slot that silently stops + configuring on whichever AMS variant we compute the bit position + wrong for. The saving is one MQTT message; the failure is the bug + this commit is fixing, inverted. So a state that does not say "empty" + still publishes, exactly as it did before. + """ + printer = await printer_factory(name="H2D") + spool = await spool_factory(slicer_filament="GFL05", material="PLA") + + mock_client = MagicMock() + mock_client.ams_set_filament_setting.return_value = True + mock_client.extrusion_cali_sel.return_value = True + + tray_data = {"id": 3, "state": 11, "exists": False, "tray_type": "", "tray_color": ""} + status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}]) + + with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: + mock_pm.get_client.return_value = mock_client + mock_pm.get_status.return_value = status + + response = await async_client.post( + "/api/v1/inventory/assignments", + json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, + ) + + assert response.status_code == 200 + mock_client.ams_set_filament_setting.assert_called_once() + assert response.json()["pending_config"] is False + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_the_pre_assign_workflow_still_skips_a_genuinely_empty_slot( + self, async_client: AsyncClient, printer_factory, spool_factory + ): + """SpoolBuddy weighs a spool and assigns it before it goes in. Bit + clear and state 9 agree that the slot is empty, so the push is still + deferred to the replay — unchanged.""" + printer = await printer_factory(name="H2D") + spool = await spool_factory(slicer_filament="GFL05", material="PLA") + + mock_client = MagicMock() + mock_client.ams_set_filament_setting.return_value = True + + tray_data = {"id": 3, "state": 9, "exists": False, "tray_type": "", "tray_color": ""} + status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}]) + + with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: + mock_pm.get_client.return_value = mock_client + mock_pm.get_status.return_value = status + + response = await async_client.post( + "/api/v1/inventory/assignments", + json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, + ) + + assert response.status_code == 200 + mock_client.ams_set_filament_setting.assert_not_called() + body = response.json() + assert body["configured"] is False + assert body["pending_config"] is True + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_an_unannotated_tray_still_reads_the_state( + self, async_client: AsyncClient, printer_factory, spool_factory + ): + """No presence bit in the payload → the 9/10 heuristic still decides. + + The external spool's ``vt_tray`` has no bit in the mask, and neither do + the hand-built payloads every other test in this file uses. + """ + printer = await printer_factory(name="H2D") + spool = await spool_factory(slicer_filament="GFL05", material="PLA") + + mock_client = MagicMock() + mock_client.ams_set_filament_setting.return_value = True + + status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "state": 9, "tray_type": ""}]}]) + + with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: + mock_pm.get_client.return_value = mock_client + mock_pm.get_status.return_value = status + + response = await async_client.post( + "/api/v1/inventory/assignments", + json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, + ) + + assert response.status_code == 200 + mock_client.ams_set_filament_setting.assert_not_called() + assert response.json()["pending_config"] is True + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_deferred_config_fires_for_a_spool_the_ams_cannot_name( + self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession + ): + """The pre-assign workflow's half of the same bug. + + A non-RFID spool inserted into a pre-assigned slot brings no + ``tray_type`` with it, and the stale 9 kept the replay's "loaded" test + false, so the deferred configuration never fired for it either. The + presence bit is the only thing in the payload that changed. + """ + from unittest.mock import AsyncMock + + from backend.app.main import on_ams_change + from backend.app.models.spool_assignment import SpoolAssignment + + printer = await printer_factory(name="H2D") + spool = await spool_factory(slicer_filament="GFL05", material="PLA") + + pre_assignment = SpoolAssignment( + spool_id=spool.id, + printer_id=printer.id, + ams_id=2, + tray_id=3, + fingerprint_color=None, + fingerprint_type=None, + ) + db_session.add(pre_assignment) + await db_session.commit() + + ams_data = [{"id": 2, "tray": [{"id": 3, "state": 9, "exists": True, "tray_type": "", "tray_color": ""}]}] + + mock_client = MagicMock() + mock_client.ams_set_filament_setting.return_value = True + mock_client.extrusion_cali_sel.return_value = True + + status = _make_mock_status(ams_data=ams_data) + + with ( + patch("backend.app.main.printer_manager") as mock_pm_main, + patch("backend.app.services.printer_manager.printer_manager") as mock_pm_inv, + patch("backend.app.main.mqtt_relay") as mock_relay, + patch("backend.app.main.ws_manager") as mock_ws, + ): + mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427") + mock_pm_main.get_status.return_value = status + mock_pm_main.get_client.return_value = mock_client + mock_pm_main.get_model.return_value = "H2D" + mock_pm_inv.get_client.return_value = mock_client + mock_pm_inv.get_status.return_value = status + mock_relay.on_ams_change = AsyncMock() + mock_ws.send_printer_status = AsyncMock() + mock_ws.broadcast = AsyncMock() + + await on_ams_change(printer.id, ams_data) + + mock_client.ams_set_filament_setting.assert_called_once() + call_kwargs = mock_client.ams_set_filament_setting.call_args.kwargs + assert call_kwargs["ams_id"] == 2 + assert call_kwargs["tray_id"] == 3 + assert call_kwargs["tray_info_idx"] == "GFL05" + + # The assignment is still there — the pass that fires the config is the + # same pass that deletes stale ones (#3100). + db_session.expunge_all() + assert await db_session.get(SpoolAssignment, pre_assignment.id) is not None + + class TestAssignSpoolPfcnCloudPreset: """Assign path for PFCN-prefix cloud presets (#1648). @@ -1561,6 +1776,118 @@ class TestAutoUnlinkDuringRunout: assert await db_session.get(SpoolAssignment, assignment_id) is not None +class TestAutoUnlinkOccupiedSlot: + """#3100: six saved assignments deleted across three X1 Carbons. + + Each one had an explicit earlier assignment and then an ``Auto-unlink ... + fingerprint mismatch``, and the reporter recovered the mappings from logs + because they were gone from the inventory API, not merely hidden. Two + shapes, one cause: a slot the presence bit calls occupied while the tray + reports nothing about what is in it. + + ``cur=/ fp=BCBCBCFF/PLA spool=8A8F92FF/PLA`` — an established assignment, + a blank idle report, and the row deleted. The spool never went anywhere; + the AMS simply had nothing to say about a filament it cannot read. + """ + + @staticmethod + async def _run(printer_id, ams_data, status): + from unittest.mock import AsyncMock + + from backend.app.main import on_ams_change + + with ( + patch("backend.app.main.printer_manager") as mock_pm_main, + patch("backend.app.main.mqtt_relay") as mock_relay, + patch("backend.app.main.ws_manager") as mock_ws, + ): + mock_pm_main.get_printer.return_value = MagicMock(name="X1C", serial_number="0948BB540200427") + mock_pm_main.get_status.return_value = status + mock_pm_main.get_model.return_value = "X1C" + mock_pm_main.get_client.return_value = None + mock_relay.on_ams_change = AsyncMock() + mock_ws.send_printer_status = AsyncMock() + mock_ws.broadcast = AsyncMock() + + await on_ams_change(printer_id, ams_data) + + @staticmethod + async def _assignment(db_session, printer, spool): + from backend.app.models.spool_assignment import SpoolAssignment + + assignment = SpoolAssignment( + spool_id=spool.id, + printer_id=printer.id, + ams_id=1, + tray_id=1, + fingerprint_color="BCBCBCFF", + fingerprint_type="PLA", + ) + db_session.add(assignment) + await db_session.commit() + return assignment.id + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_a_blank_report_from_an_occupied_slot_keeps_the_assignment( + self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession + ): + from backend.app.models.spool_assignment import SpoolAssignment + + printer = await printer_factory(name="X1C") + spool = await spool_factory(material="PLA", rgba="8A8F92FF") + assignment_id = await self._assignment(db_session, printer, spool) + + ams_data = [{"id": 1, "tray": [{"id": 1, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}] + await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE")) + + db_session.expunge_all() + assert await db_session.get(SpoolAssignment, assignment_id) is not None, ( + "a spool the AMS cannot identify is not a spool that was removed" + ) + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_a_blank_report_from_an_empty_slot_still_unlinks( + self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession + ): + """The guard reads the bit, not the blankness — take the spool out and + the assignment still goes, which is what makes the test above a + distinction rather than a blanket reprieve.""" + from backend.app.models.spool_assignment import SpoolAssignment + + printer = await printer_factory(name="X1C") + spool = await spool_factory(material="PLA", rgba="8A8F92FF") + assignment_id = await self._assignment(db_session, printer, spool) + + ams_data = [{"id": 1, "tray": [{"id": 1, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}] + await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE")) + + db_session.expunge_all() + assert await db_session.get(SpoolAssignment, assignment_id) is None + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_a_genuinely_different_filament_in_an_occupied_slot_still_unlinks( + self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession + ): + """The guard is for a blank report only. A slot that names a filament + which is not the assigned spool is a swap, bit set or not.""" + from backend.app.models.spool_assignment import SpoolAssignment + + printer = await printer_factory(name="X1C") + spool = await spool_factory(material="PLA", rgba="8A8F92FF") + assignment_id = await self._assignment(db_session, printer, spool) + + ams_data = [ + {"id": 1, "tray": [{"id": 1, "exists": True, "tray_type": "PETG", "tray_color": "00FF00FF", "state": 11}]} + ] + await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE")) + + db_session.expunge_all() + assert await db_session.get(SpoolAssignment, assignment_id) is None + + class TestSpoolmanSlotAssignmentDuringRunout: """`spoolman_slot_assignments` is how a tag-less spool assigned through the Bambuddy UI is resolved at completion (#1459). Deleting the row when a slot @@ -1578,14 +1905,14 @@ class TestSpoolmanSlotAssignmentDuringRunout: db_session.add(Settings(key=key, value=value)) await db_session.commit() - async def _run(self, printer_id, state): + async def _run(self, printer_id, state, tray=None): from unittest.mock import AsyncMock from backend.app.main import on_ams_change # A tray the firmware has cleared: parse_ams_tray returns None, which # is what marks the slot empty for the cleanup pass. - ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}] + ams_data = [{"id": 0, "tray": [tray or {"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}] spoolman_client = MagicMock() spoolman_client.health_check = AsyncMock(return_value=True) @@ -1628,6 +1955,37 @@ class TestSpoolmanSlotAssignmentDuringRunout: db_session.expunge_all() assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None + @pytest.mark.asyncio + @pytest.mark.integration + async def test_an_occupied_slot_the_ams_cannot_read_keeps_its_row( + self, async_client: AsyncClient, printer_factory, db_session: AsyncSession + ): + """Spoolman mode's half of #3100. + + parse_ams_tray calls a tray with no type empty, and a tag-less spool + has none until something configures it — so the row assigned through + the UI was deleted by the first idle push after the spool went in. + Firmware's presence bit is the same answer the built-in inventory + uses, so the two modes stay in step. + """ + from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment + + await self._enable_spoolman(db_session) + printer = await printer_factory(name="H2D") + row = SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=2, spoolman_spool_id=41) + db_session.add(row) + await db_session.commit() + row_id = row.id + + await self._run( + printer.id, + _make_printing_status(None, state="IDLE"), + tray={"id": 2, "exists": True, "tray_type": "", "tray_color": "", "state": 9}, + ) + + db_session.expunge_all() + assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None + @pytest.mark.asyncio @pytest.mark.integration async def test_the_slot_row_is_still_cleaned_up_when_idle( diff --git a/backend/tests/unit/services/test_ams_slot_presence.py b/backend/tests/unit/services/test_ams_slot_presence.py new file mode 100644 index 000000000..c2c189ac0 --- /dev/null +++ b/backend/tests/unit/services/test_ams_slot_presence.py @@ -0,0 +1,45 @@ +"""Unit tests for the firmware presence bit helper. + +#3084: a slot can read ``exists=True, state=9`` — the bit says a spool is in +it, the state says the opposite — because ``apply_tray_exist_bits`` writes the +9 itself and never takes it back. Everything downstream has to know which of +the two to believe, and this helper is the single place that says so. +""" + +from backend.app.services.ams_slot_presence import spool_present + + +class TestSpoolPresent: + def test_the_bit_is_reported_as_it_stands(self): + assert spool_present({"id": 0, "exists": True}) is True + assert spool_present({"id": 0, "exists": False}) is False + + def test_the_bit_answers_over_a_contradicting_state(self): + # The #3084 slot: non-Bambu spool swapped in, so the bit is set, while + # the 9 Bambuddy stamped on the slot when it was briefly empty is still + # sitting there. + assert spool_present({"id": 0, "exists": True, "state": 9}) is True + # And the converse — a spool pulled from a slot the firmware last + # described as loaded. + assert spool_present({"id": 0, "exists": False, "state": 11}) is False + + def test_a_tray_with_no_annotation_answers_nothing(self): + # vt_tray entries and the VP bridge's cache carry no presence bit, so + # callers have to fall back to their own reading rather than be handed + # a guess dressed up as firmware's answer. + assert spool_present({"id": 0, "state": 11, "tray_type": "PLA"}) is None + assert spool_present({}) is None + + def test_a_non_bool_exists_is_not_a_presence_bit(self): + # Only apply_tray_exist_bits writes this key, and it writes a bool. + # Anything else reached the dict some other way and is not firmware's + # answer — 0 and "" would otherwise read as a confident "empty". + assert spool_present({"exists": 0}) is None + assert spool_present({"exists": ""}) is None + assert spool_present({"exists": "true"}) is None + assert spool_present({"exists": None}) is None + + def test_a_missing_tray_answers_nothing(self): + assert spool_present(None) is None + assert spool_present([]) is None + assert spool_present("tray") is None