fix(printers): an unplugged AMS leaves the printer card (issue #3356)

- The AMS merge kept every unit a push didn't list, so an AMS unplugged
  while Bambuddy ran stayed on the card, and the no-AMS rule for
  use_ams never fired until a restart. A unit is now dropped when the
  same push's ams_exist_bits say it is gone: an AMS 0-3 whose bit is
  clear, or every unit when the list is empty and no bit is set. The
  bit layout is measured on an X1C (unit 1 -> "2", unplugged -> "0"
  with an empty list) and an H2C (units 0-2 and HT 128 -> "17").
  Partial pushes, pushes without the bits, an AMS HT or AMS Lite next
  to other units, and the shutdown pattern keep their units as before.
- The virtual printer's slicer-facing merge uses the same rule, so the
  slicer no longer sees an AMS the printer card has dropped.
- Spools assigned to an AMS that stays unplugged for more than two
  minutes are unassigned, as for a spool taken out of a slot.

CHANGELOG.md

@@ -176,7 +176,7 @@ All notable changes to Bambuddy will be documented in this file.
 
 ### Fixed
 - **The MQTT Debug log shows what the printer sent** — Each logged message was the same object Bambuddy then processed, and processing writes Bambuddy's own state into it. An incoming push therefore showed Bambuddy's merged AMS list instead of the printer's AMS block, without fields such as `ams_exist_bits` and still listing an AMS that had been unplugged, and later messages could gain an AMS list they never carried. The log now keeps every incoming message exactly as it arrived.
-- **Printers without an AMS print from the spool holder (#3356, reported by @dogod3814t)** — A queue print to a printer with no AMS attached, such as a P1S on its spool holder alone, failed with "Failed to get AMS mapping table" when nothing was set on the spool holder. With nothing to match, the job went out without a filament mapping and asking the printer to use its AMS, which it refuses. Bambuddy now tells a printer that reports no AMS to print from the spool holder, as the slicer does. Dual-nozzle printers are unchanged.
+- **Printers without an AMS print from the spool holder (#3356, reported by @dogod3814t)** — A queue print to a printer with no AMS attached, such as a P1S on its spool holder alone, failed with "Failed to get AMS mapping table" when nothing was set on the spool holder. With nothing to match, the job went out without a filament mapping and asking the printer to use its AMS, which it refuses. Bambuddy now tells a printer that reports no AMS to print from the spool holder, as the slicer does. Dual-nozzle printers are unchanged. An AMS unplugged while Bambuddy runs also disappears from the printer card and from slicers connected through a virtual printer now, instead of staying until a restart, so this works straight after removing one. As with a spool taken out of a slot, spools assigned to an AMS that stays unplugged for more than two minutes are unassigned; RFID spools are assigned again when the AMS is back, other spools need assigning again.
 - **A personal cost center's budget can be changed (#3358, reported by @airforceixi)** — Saving the **Edit cost center** dialog on a user's personal cost center failed with "Private cost centers cannot be deactivated or renamed; set their budget to 0 to prevent printing", even when only the budget was changed, so a budget (or 0, to stop that user printing) could not be set. The dialog always sent the name as well, which a personal cost center refuses, and the budget was never saved. It now saves only the budget for a personal cost center, and its name field is greyed out, as the name is the user's.
 - **Slicing a painted multi-colour model keeps its colours (#3355, reported by @cartman75)** — On a model coloured with the paint tool, slicing in Bambuddy could turn several filaments into one colour, for example black, white, green and green all coming out green. Before slicing, Bambuddy works out which filaments the plate doesn't use and swaps them for one it does, so the slicer's material checks pass. It misread the painted areas, so painted filaments counted as unused and got another slot's profile and colour, and the print dialog then mapped them to that slot's spool. Painted areas are now read correctly, including small ones. A filament that really is unused keeps its own colour when it is swapped.
 - **An AMS Lite shows its serial number and firmware version (#3348, reported by @Spegeli)** — The printer reports both in its version answer, where the AMS Lite calls itself `ams_f1`. Bambuddy only knew the names of the other AMS models, so it skipped the AMS Lite's entry. The AMS popup showed "—" for both, the status API returned them empty, and the log warned on every connection that they were missing. They now come through on the A1, A1 Mini and A2L, including the A2L's AMS Lite, which the printer numbers differently. A name given to an AMS Lite before this, which was kept by its position because it had no serial number, still shows on the printer card and in the inventory's location column. Saving it again moves it to the serial number, so it follows the unit, and **Clear** now removes it; before, the old name came back after clearing.

backend/app/services/bambu_mqtt.py

@@ -259,6 +259,29 @@ def resolve_external_spools_in_mapping(ams_mapping: object, ams_mapping2: object
     return resolved
 
 
+def unit_reported_absent(ams_id, ams_list: list, ams_exist_bits) -> bool:
+    """Whether a push that doesn't list AMS unit ``ams_id`` says it is unplugged.
+
+    ``ams_exist_bits`` (hex) has bit n set for AMS unit n, measured on an X1C
+    (unit 1 -> "2", none -> "0") and an H2C (units 0, 1, 2 and HT 128 -> "17").
+    So an AMS unit 0-3 whose bit is clear is gone. An empty unit list with no
+    bits set means no unit at all, which is what the X1C sends with its AMS
+    unplugged. Anything else is kept, as the merge always did: a partial push
+    lists only some units, and the bits of an AMS HT or AMS Lite next to other
+    units are not measured well enough to read.
+    """
+    if ams_exist_bits is None:
+        return False
+    try:
+        bits = int(ams_exist_bits, 16) if isinstance(ams_exist_bits, str) else int(ams_exist_bits)
+        unit = int(ams_id)
+    except (TypeError, ValueError):
+        return False
+    if bits == 0 and not ams_list:
+        return True
+    return 0 <= unit <= 3 and not bits >> unit & 1
+
+
 def apply_tray_exist_bits(
     units: list,
     tray_exist_bits_str: str | int | None,
@@ -3829,6 +3852,16 @@ class BambuMQTTClient:
                     ams_unit = {**existing_unit, **ams_unit}
                 existing_by_id[ams_id] = ams_unit
 
+        # The merge keeps a unit the push doesn't list, so an unplugged AMS used
+        # to stay until Bambuddy restarted (#3356). Drop one only on the
+        # printer's own word that it is gone.
+        if isinstance(ams_data, dict):
+            listed = {ams_unit.get("id") for ams_unit in ams_list}
+            for ams_id in [i for i in existing_by_id if i not in listed]:
+                if unit_reported_absent(ams_id, ams_list, ams_data.get("ams_exist_bits")):
+                    logger.info("[%s] AMS %s no longer attached, removing it", self.serial_number, ams_id)
+                    del existing_by_id[ams_id]
+
         # Convert back to list, sorted by ID for consistent ordering
         merged_ams = sorted(existing_by_id.values(), key=lambda x: x.get("id", 0))
 

backend/app/services/virtual_printer/mqtt_bridge.py

@@ -44,7 +44,7 @@ import os
 import socket
 from typing import TYPE_CHECKING
 
-from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+from backend.app.services.bambu_mqtt import apply_tray_exist_bits, unit_reported_absent
 from backend.app.services.virtual_printer._debug import append_event, dump_wire
 
 if TYPE_CHECKING:
@@ -210,7 +210,8 @@ def _merge_ams_dict(prev_ams: dict, new_ams: dict) -> dict:
       - Shallow-merge top-level scalars: keys in ``new`` win; keys only
         in ``prev`` are preserved.
       - For the ``ams`` array (list of units): match by ``id``. Units
-        only in ``prev`` survive. Units in ``new`` overlay onto their
+        only in ``prev`` survive unless ``ams_exist_bits`` reports them
+        unplugged. Units in ``new`` overlay onto their
         ``prev`` counterpart; same recursion applies to each unit's
         ``tray`` array by tray ``id``.
     """
@@ -277,9 +278,11 @@ def _merge_ams_dict(prev_ams: dict, new_ams: dict) -> dict:
         merged_units.append(merged_unit)
         if uid is not None:
             seen_ids.add(uid)
-    # Preserve prev units not mentioned in the incremental.
+    # Preserve prev units not mentioned in the incremental, unless the push's
+    # ams_exist_bits say the unit was unplugged; the printer card drops it on
+    # the same rule, so the slicer and Bambuddy agree (#3356).
     for uid, prev_unit in prev_by_id.items():
-        if uid not in seen_ids:
+        if uid not in seen_ids and not unit_reported_absent(uid, new_units, new_ams.get("ams_exist_bits")):
             merged_units.append(prev_unit)
     merged["ams"] = merged_units
     return merged

backend/tests/unit/test_use_ams_no_ams_3356.py

@@ -76,3 +76,86 @@ def test_dual_nozzle_is_left_alone(client):
     assert client.start_print("case_body.3mf", ams_mapping=None, use_ams=True)
 
     assert _use_ams(client) is True
+
+
+# The AMS blocks below are shaped like Martin's X1C and H2C captures: the unit
+# list and ams_exist_bits arrive together in every push.
+def _unit(unit_id, trays=4):
+    return {"id": str(unit_id), "tray": [{"id": str(t), "tray_type": "PLA"} for t in range(trays)]}
+
+
+def _push(units, bits):
+    return {"ams": units, "ams_exist_bits": bits, "tray_now": "255", "power_on_flag": False}
+
+
+def _unit_ids(client):
+    return sorted((u["id"] for u in client.state.raw_data["ams"]), key=int)
+
+
+class TestAnUnpluggedAms:
+    def test_the_x1c_unplug_and_replug(self, client):
+        client.model = "X1C"
+        client._handle_ams_data(_push([_unit(1)], "2"))
+
+        client._handle_ams_data(_push([], "0"))
+        assert _unit_ids(client) == []
+
+        client._handle_ams_data(_push([_unit(1)], "2"))
+        assert _unit_ids(client) == ["1"]
+
+    def test_printing_from_the_spool_holder_after_the_unplug(self, client):
+        """No restart needed: the no-AMS rule sees the unplug."""
+        client._handle_ams_data(_push([_unit(0)], "1"))
+        client._handle_ams_data(_push([], "0"))
+
+        assert client.start_print("case_body.3mf", ams_mapping=None, use_ams=True)
+
+        assert _use_ams(client) is False
+
+    def test_a_unit_whose_bit_is_cleared_goes(self, client):
+        client.model = "H2C"
+        client._handle_ams_data(_push([_unit(0), _unit(1), _unit(2), _unit(128, 1)], "17"))
+
+        client._handle_ams_data(_push([_unit(0), _unit(2), _unit(128, 1)], "15"))
+
+        assert _unit_ids(client) == ["0", "2", "128"]
+
+    def test_a_partial_list_keeps_the_units_it_leaves_out(self, client):
+        client.model = "H2C"
+        client._handle_ams_data(_push([_unit(0), _unit(1), _unit(2), _unit(128, 1)], "17"))
+
+        client._handle_ams_data(_push([_unit(0)], "17"))
+
+        assert _unit_ids(client) == ["0", "1", "2", "128"]
+
+    def test_an_ams_ht_next_to_other_units_is_kept(self, client):
+        """Its bit is not read: measured once, on one printer."""
+        client.model = "H2C"
+        client._handle_ams_data(_push([_unit(0), _unit(128, 1)], "11"))
+
+        client._handle_ams_data(_push([_unit(0)], "1"))
+
+        assert _unit_ids(client) == ["0", "128"]
+
+    def test_the_last_ams_ht_goes_with_everything_else(self, client):
+        client._handle_ams_data(_push([_unit(128, 1)], "10"))
+
+        client._handle_ams_data(_push([], "0"))
+
+        assert _unit_ids(client) == []
+
+    def test_a_push_without_bits_changes_nothing(self, client):
+        client._handle_ams_data(_push([_unit(0)], "1"))
+
+        client._handle_ams_data({"ams": []})
+
+        assert _unit_ids(client) == ["0"]
+
+    def test_no_bits_next_to_a_listed_unit_keeps_an_ams_ht(self, client):
+        """All-zero bits while units are still listed is the shutdown pattern
+        (#765), not a report that the AMS HT is gone."""
+        client._handle_ams_data(_push([_unit(0), _unit(128, 1)], "11"))
+
+        client._handle_ams_data(_push([_unit(0)], "0"))
+
+        assert _unit_ids(client) == ["0", "128"]

backend/tests/unit/test_vp_mqtt_bridge.py

@@ -713,6 +713,43 @@ class TestPushStatusCache:
 
         await bridge.stop()
 
+    @pytest.mark.asyncio
+    async def test_an_unplugged_ams_leaves_the_slicer_cache(self):
+        """#3356: the printer card drops an AMS the printer reports unplugged;
+        the slicer must not go on seeing it. Shapes from an X1C capture: unit 1
+        comes with ams_exist_bits "2", and with the AMS unplugged every push
+        carries an empty unit list and "0"."""
+        server = _make_server()
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        def push(units, bits):
+            bridge._on_printer_raw(
+                f"device/{H2D_SERIAL}/report",
+                json.dumps(
+                    {
+                        "print": {
+                            "command": "push_status",
+                            "ams": {
+                                "ams": units,
+                                "ams_exist_bits": bits,
+                                "tray_exist_bits": "0",
+                                "power_on_flag": False,
+                            },
+                        }
+                    }
+                ).encode(),
+            )
+
+        push([{"id": "1", "tray": [{"id": "0", "tray_type": "PLA"}]}], "2")
+        await asyncio.sleep(0.01)
+        push([], "0")
+        await asyncio.sleep(0.01)
+
+        assert bridge.get_latest_print_state()["ams"]["ams"] == []
+
+        await bridge.stop()
+
     @pytest.mark.asyncio
     async def test_tray_exist_bits_clears_empty_slots_in_slicer_cache(self):
         """#1726 (reported by @needo37): the bridge cache forwards the real