fix(ams): read the firmware presence bit, not the tray state (issue #3084)

Swapping a Bambu spool for one the AMS cannot read left Assign Spool
    publishing no ams_filament_setting at all. The printer kept showing "?"
    on its screen and in the slicer, and only Configure, which publishes
    unconditionally, put anything there.

    Four places asked the tray's `state` field whether a spool was in the
    slot. It cannot answer that. An AMS-HT reports its LOADED tray as 9
    rather than 11, because it does not feed into a shared buffer the way a
    4-slot AMS does -- the merge has skipped its own state heuristic for HT
    units since #2594 for exactly this reason. And the field is partly our
    own writing: apply_tray_exist_bits stamps state=9 on every slot whose
    tray_exist_bits bit is 0, and when the bit comes back it refreshes only
    the `exists` annotation beside it. Either way the slot sits at
    exists=True, state=9 until something configures it.

    That 9 also kept the deferred-configuration replay from firing -- its
    own "has a spool appeared" test was the same heuristic -- which is the
    deadlock #1322 removed from the assign path, still in place one step
    further along. And it is what deleted the assignments in #3100: with the
    replay never firing, the row kept the empty fingerprint it was stored
    with, and the first tray report naming a filament was read as a swap.

    All four now read tray_exist_bits first, which is the mask firmware
    answers this question with and the one the printer card has drawn its
    "?" from since #2527. The bit is allowed to overrule an "empty" state
    and nothing else: a bit reading empty deliberately does not start
    suppressing pushes that go out today, because the cost of computing a
    bit position wrong is a slot that silently stops configuring, against a
    saving of one message firmware would have dropped.

    A blank tray report from a slot the bit calls occupied no longer unlinks
    anything, off a print as well as during one, in both inventory modes --
    Spoolman's parse_ams_tray calls a tray with no type empty, so a tag-less
    spool assigned through the UI had its row deleted by the first idle push
    after it went in. A filament the AMS cannot identify is not a filament
    that was removed.
This commit is contained in:
maziggy
2026-09-20 13:40:30 +02:00
5 changed files with 536 additions and 5 deletions
+30 -1
View File
@@ -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:
+41 -2
View File
@@ -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
+60
View File
@@ -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
@@ -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(
@@ -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