|
|
@@ -0,0 +1,230 @@
|
|
|
+"""K-profiles follow an AMS when it moves between Filament Track Switch inlets.
|
|
|
+
|
|
|
+Measured on the maintainer's H2C, 2026-08-16. Spool 75, "HF Bambu PLA Matte
|
|
|
+Black", is calibrated on both hotends and Bambuddy stores both:
|
|
|
+
|
|
|
+ extruder 1 (left) K 0.018 cali_idx 16
|
|
|
+ extruder 0 (right) K 0.020 cali_idx 15
|
|
|
+
|
|
|
+A tray holds exactly one ``cali_idx``, and the printer's calibration table is
|
|
|
+numbered per nozzle — so index 16 exists on both hotends and means a different
|
|
|
+profile on each. Moving that AMS from In-A to In-B therefore leaves the slot
|
|
|
+bound to the *left* profile while it now feeds the right nozzle. The printer
|
|
|
+does not re-resolve it, and an RFID re-read only re-asserts the same wrong
|
|
|
+index. That is the state the live printer was in when this was written:
|
|
|
+
|
|
|
+ AMS 1 info=10002E03 inlet=IN-B tray 0: PLA 000000 cali_idx=16
|
|
|
+"""
|
|
|
+
|
|
|
+from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
+
|
|
|
+import pytest
|
|
|
+
|
|
|
+from backend.app.utils.fts_routing import FTS_INLET_EXTRUDER, extruder_for_inlet, slot_extruder
|
|
|
+
|
|
|
+
|
|
|
+class TestInletToNozzle:
|
|
|
+ def test_in_a_is_the_left_hotend(self):
|
|
|
+ assert extruder_for_inlet("A") == 1
|
|
|
+
|
|
|
+ def test_in_b_is_the_right_hotend(self):
|
|
|
+ assert extruder_for_inlet("B") == 0
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("value", [None, "", "C", "left"])
|
|
|
+ def test_anything_else_is_unknown(self, value):
|
|
|
+ assert extruder_for_inlet(value) is None
|
|
|
+
|
|
|
+ def test_the_two_inlets_do_not_share_a_nozzle(self):
|
|
|
+ """A table that mapped both inlets to one hotend would silently bind
|
|
|
+ every slot to the same K-profile."""
|
|
|
+ assert sorted(FTS_INLET_EXTRUDER.values()) == [0, 1]
|
|
|
+
|
|
|
+
|
|
|
+class TestSlotExtruder:
|
|
|
+ def test_a_real_extruder_id_wins(self):
|
|
|
+ """A non-0xE binding is authoritative even on a machine with a switch,
|
|
|
+ which is how BambuStudio treats it too."""
|
|
|
+ assert slot_extruder(2, 0, {"2": 1}, {"2": "B"}) == 1
|
|
|
+
|
|
|
+ def test_falls_back_to_the_switch_inlet(self):
|
|
|
+ """The FTS case: every AMS reports 0xE, so the map is empty."""
|
|
|
+ assert slot_extruder(1, 0, {}, {"1": "B"}) == 0
|
|
|
+ assert slot_extruder(2, 3, {}, {"2": "A"}) == 1
|
|
|
+
|
|
|
+ def test_unknown_is_none_not_zero(self):
|
|
|
+ """The bug this replaced: three call sites ended in `else 0`, which on a
|
|
|
+ switch machine filed every profile under the right-hand nozzle."""
|
|
|
+ assert slot_extruder(1, 0, {}, {}) is None
|
|
|
+ assert slot_extruder(1, 0, None, None) is None
|
|
|
+
|
|
|
+ def test_external_slots_name_their_own_side(self):
|
|
|
+ assert slot_extruder(255, 0, {}, {}) == 1 # Ext-L
|
|
|
+ assert slot_extruder(255, 1, {}, {}) == 0 # Ext-R
|
|
|
+
|
|
|
+ def test_an_unmapped_ams_does_not_borrow_another_units_inlet(self):
|
|
|
+ assert slot_extruder(3, 0, {}, {"1": "A", "2": "B"}) is None
|
|
|
+
|
|
|
+ def test_the_maintainers_h2c(self):
|
|
|
+ """The live layout: AMS 0/1/128 on In-B, AMS 2 on In-A, empty map."""
|
|
|
+ inlets = {"0": "B", "1": "B", "2": "A", "128": "B"}
|
|
|
+ assert slot_extruder(0, 0, {}, inlets) == 0
|
|
|
+ assert slot_extruder(1, 0, {}, inlets) == 0
|
|
|
+ assert slot_extruder(2, 1, {}, inlets) == 1
|
|
|
+ assert slot_extruder(128, 0, {}, inlets) == 0
|
|
|
+
|
|
|
+
|
|
|
+def _profile(cali_idx: int, extruder: int, k: float):
|
|
|
+ return MagicMock(
|
|
|
+ cali_idx=cali_idx, extruder=extruder, k_value=k, name="HF Bambu PLA Matte Black", filament_id="GFA01"
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+class TestReSelectOnInletMove:
|
|
|
+ """The callback that re-points a moved AMS's slots."""
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _run(inlet, tray, profile, *, connected=True):
|
|
|
+ """Drive on_fts_inlet_change for one AMS holding one tray."""
|
|
|
+ from backend.app import main as main_module
|
|
|
+
|
|
|
+ client = MagicMock()
|
|
|
+ client.extrusion_cali_sel = MagicMock(return_value=True)
|
|
|
+ state = MagicMock()
|
|
|
+ state.raw_data = {"ams": [{"id": "1", "tray": [tray]}]}
|
|
|
+ state.nozzles = [MagicMock(nozzle_diameter="0.4")]
|
|
|
+
|
|
|
+ pm = MagicMock()
|
|
|
+ pm.get_client.return_value = client if connected else None
|
|
|
+ pm.get_status.return_value = state
|
|
|
+
|
|
|
+ session = AsyncMock()
|
|
|
+ session.__aenter__ = AsyncMock(return_value=session)
|
|
|
+ session.__aexit__ = AsyncMock()
|
|
|
+
|
|
|
+ with (
|
|
|
+ patch("backend.app.main.printer_manager", pm),
|
|
|
+ patch("backend.app.main.async_session", return_value=session),
|
|
|
+ patch(
|
|
|
+ "backend.app.main.find_slot_kprofile_for_extruder",
|
|
|
+ new=AsyncMock(return_value=profile),
|
|
|
+ ) as lookup,
|
|
|
+ ):
|
|
|
+ import asyncio
|
|
|
+
|
|
|
+ asyncio.run(main_module.on_fts_inlet_change(7, 1, inlet))
|
|
|
+ return client, lookup
|
|
|
+
|
|
|
+ def test_moving_to_in_b_selects_the_right_hotends_profile(self):
|
|
|
+ """The reported case, with the maintainer's real numbers."""
|
|
|
+ tray = {"id": "0", "tray_type": "PLA", "cali_idx": 16, "tray_info_idx": "GFA01"}
|
|
|
+ client, lookup = self._run("B", tray, _profile(15, 0, 0.020))
|
|
|
+
|
|
|
+ assert lookup.await_args.args[4] == 0, "must look up the RIGHT hotend's profile"
|
|
|
+ client.extrusion_cali_sel.assert_called_once()
|
|
|
+ assert client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 15
|
|
|
+
|
|
|
+ def test_moving_to_in_a_selects_the_left_hotends_profile(self):
|
|
|
+ tray = {"id": "0", "tray_type": "PLA", "cali_idx": 15, "tray_info_idx": "GFA01"}
|
|
|
+ client, lookup = self._run("A", tray, _profile(16, 1, 0.018))
|
|
|
+
|
|
|
+ assert lookup.await_args.args[4] == 1
|
|
|
+ assert client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 16
|
|
|
+
|
|
|
+ def test_an_already_correct_slot_is_left_alone(self):
|
|
|
+ """No point re-sending a binding the printer already holds, and every
|
|
|
+ avoided write is one less chance of the firmware mislinking it."""
|
|
|
+ tray = {"id": "0", "tray_type": "PLA", "cali_idx": 15, "tray_info_idx": "GFA01"}
|
|
|
+ client, _ = self._run("B", tray, _profile(15, 0, 0.020))
|
|
|
+
|
|
|
+ client.extrusion_cali_sel.assert_not_called()
|
|
|
+
|
|
|
+ def test_a_spool_with_no_profile_for_that_nozzle_is_untouched(self):
|
|
|
+ """Calibrated on one hotend only: keep whatever the operator set by hand
|
|
|
+ rather than swapping it for a guess."""
|
|
|
+ tray = {"id": "0", "tray_type": "PLA", "cali_idx": 16, "tray_info_idx": "GFA01"}
|
|
|
+ client, _ = self._run("B", tray, None)
|
|
|
+
|
|
|
+ client.extrusion_cali_sel.assert_not_called()
|
|
|
+
|
|
|
+ def test_an_empty_slot_is_skipped(self):
|
|
|
+ tray = {"id": "0", "tray_type": "", "cali_idx": -1}
|
|
|
+ client, lookup = self._run("B", tray, _profile(15, 0, 0.020))
|
|
|
+
|
|
|
+ lookup.assert_not_awaited()
|
|
|
+ client.extrusion_cali_sel.assert_not_called()
|
|
|
+
|
|
|
+ def test_a_disconnected_printer_is_a_no_op(self):
|
|
|
+ tray = {"id": "0", "tray_type": "PLA", "cali_idx": 16, "tray_info_idx": "GFA01"}
|
|
|
+ client, _ = self._run("B", tray, _profile(15, 0, 0.020), connected=False)
|
|
|
+
|
|
|
+ client.extrusion_cali_sel.assert_not_called()
|
|
|
+
|
|
|
+ def test_an_unknown_inlet_is_a_no_op(self):
|
|
|
+ tray = {"id": "0", "tray_type": "PLA", "cali_idx": 16, "tray_info_idx": "GFA01"}
|
|
|
+ client, lookup = self._run("C", tray, _profile(15, 0, 0.020))
|
|
|
+
|
|
|
+ lookup.assert_not_awaited()
|
|
|
+ client.extrusion_cali_sel.assert_not_called()
|
|
|
+
|
|
|
+
|
|
|
+class TestTheMoveIsDetected:
|
|
|
+ """The MQTT side: only a genuine move fires the callback."""
|
|
|
+
|
|
|
+ @pytest.fixture
|
|
|
+ def mqtt_client(self):
|
|
|
+ from backend.app.services.bambu_mqtt import BambuMQTTClient
|
|
|
+
|
|
|
+ return BambuMQTTClient(ip_address="192.168.1.100", serial_number="TEST123", access_code="12345678")
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _info(inlet_bits: int) -> str:
|
|
|
+ return f"{(inlet_bits << 24) | (0xE << 8) | 1:08X}"
|
|
|
+
|
|
|
+ def _push(self, client, inlet_bits):
|
|
|
+ client._process_message(
|
|
|
+ {
|
|
|
+ "print": {
|
|
|
+ "gcode_state": "IDLE",
|
|
|
+ "device": {"fila_switch": {"in": [-1, -1], "out": [1, 1], "stat": 1, "info": 0}},
|
|
|
+ "ams": {"ams": [{"id": "1", "info": self._info(inlet_bits), "tray": []}]},
|
|
|
+ }
|
|
|
+ }
|
|
|
+ )
|
|
|
+
|
|
|
+ def test_a_move_fires_once(self, mqtt_client):
|
|
|
+ seen = []
|
|
|
+ mqtt_client.on_fts_inlet_change = lambda ams_id, inlet: seen.append((ams_id, inlet))
|
|
|
+
|
|
|
+ self._push(mqtt_client, 1) # In-A
|
|
|
+ self._push(mqtt_client, 0) # In-B
|
|
|
+
|
|
|
+ assert seen == [(1, "B")]
|
|
|
+
|
|
|
+ def test_the_first_sighting_does_not_fire(self, mqtt_client):
|
|
|
+ """Every reconnect learns the bindings afresh. Re-applying K-profiles
|
|
|
+ there would fight a binding the operator set deliberately."""
|
|
|
+ seen = []
|
|
|
+ mqtt_client.on_fts_inlet_change = lambda ams_id, inlet: seen.append((ams_id, inlet))
|
|
|
+
|
|
|
+ self._push(mqtt_client, 1)
|
|
|
+
|
|
|
+ assert seen == []
|
|
|
+
|
|
|
+ def test_repeated_frames_do_not_fire(self, mqtt_client):
|
|
|
+ seen = []
|
|
|
+ mqtt_client.on_fts_inlet_change = lambda ams_id, inlet: seen.append((ams_id, inlet))
|
|
|
+
|
|
|
+ for _ in range(4):
|
|
|
+ self._push(mqtt_client, 1)
|
|
|
+
|
|
|
+ assert seen == []
|
|
|
+
|
|
|
+ def test_moving_back_fires_again(self, mqtt_client):
|
|
|
+ seen = []
|
|
|
+ mqtt_client.on_fts_inlet_change = lambda ams_id, inlet: seen.append((ams_id, inlet))
|
|
|
+
|
|
|
+ self._push(mqtt_client, 1)
|
|
|
+ self._push(mqtt_client, 0)
|
|
|
+ self._push(mqtt_client, 1)
|
|
|
+
|
|
|
+ assert seen == [(1, "B"), (1, "A")]
|