|
@@ -424,3 +424,184 @@ class TestRealisticMessageFlow:
|
|
|
|
|
|
|
|
assert complete_data["timelapse_was_active"] is True
|
|
assert complete_data["timelapse_was_active"] is True
|
|
|
assert complete_data["status"] == "failed"
|
|
assert complete_data["status"] == "failed"
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+class TestAMSDataMerging:
|
|
|
|
|
+ """Tests for AMS data merging, particularly handling empty slots."""
|
|
|
|
|
+
|
|
|
|
|
+ @pytest.fixture
|
|
|
|
|
+ def mqtt_client(self):
|
|
|
|
|
+ """Create a BambuMQTTClient instance for testing."""
|
|
|
|
|
+ from backend.app.services.bambu_mqtt import BambuMQTTClient
|
|
|
|
|
+
|
|
|
|
|
+ client = BambuMQTTClient(
|
|
|
|
|
+ ip_address="192.168.1.100",
|
|
|
|
|
+ serial_number="TEST123",
|
|
|
|
|
+ access_code="12345678",
|
|
|
|
|
+ )
|
|
|
|
|
+ return client
|
|
|
|
|
+
|
|
|
|
|
+ def test_empty_slot_clears_tray_type(self, mqtt_client):
|
|
|
|
|
+ """Test that empty slot update clears tray_type (Issue #147).
|
|
|
|
|
+
|
|
|
|
|
+ When a spool is removed from an old AMS, the printer sends empty values.
|
|
|
|
|
+ These must overwrite the previous values to show the slot as empty.
|
|
|
|
|
+ """
|
|
|
|
|
+ # Initial state: AMS unit with a loaded spool
|
|
|
|
|
+ initial_ams = {
|
|
|
|
|
+ "ams": [
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": 0,
|
|
|
|
|
+ "tray": [
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": 0,
|
|
|
|
|
+ "tray_type": "PLA",
|
|
|
|
|
+ "tray_sub_brands": "Bambu PLA Basic",
|
|
|
|
|
+ "tray_color": "FF0000",
|
|
|
|
|
+ "tag_uid": "1234567890ABCDEF",
|
|
|
|
|
+ "remain": 80,
|
|
|
|
|
+ }
|
|
|
|
|
+ ],
|
|
|
|
|
+ }
|
|
|
|
|
+ ]
|
|
|
|
|
+ }
|
|
|
|
|
+ mqtt_client._handle_ams_data(initial_ams)
|
|
|
|
|
+
|
|
|
|
|
+ # Verify initial state
|
|
|
|
|
+ ams_data = mqtt_client.state.raw_data.get("ams", [])
|
|
|
|
|
+ assert len(ams_data) == 1
|
|
|
|
|
+ tray = ams_data[0]["tray"][0]
|
|
|
|
|
+ assert tray["tray_type"] == "PLA"
|
|
|
|
|
+ assert tray["tray_color"] == "FF0000"
|
|
|
|
|
+
|
|
|
|
|
+ # Now simulate spool removal - printer sends empty values
|
|
|
|
|
+ empty_update = {
|
|
|
|
|
+ "ams": [
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": 0,
|
|
|
|
|
+ "tray": [
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": 0,
|
|
|
|
|
+ "tray_type": "", # Empty = slot is empty
|
|
|
|
|
+ "tray_sub_brands": "",
|
|
|
|
|
+ "tray_color": "",
|
|
|
|
|
+ "tag_uid": "0000000000000000", # Zero UID
|
|
|
|
|
+ "remain": 0,
|
|
|
|
|
+ }
|
|
|
|
|
+ ],
|
|
|
|
|
+ }
|
|
|
|
|
+ ]
|
|
|
|
|
+ }
|
|
|
|
|
+ mqtt_client._handle_ams_data(empty_update)
|
|
|
|
|
+
|
|
|
|
|
+ # Verify empty values were applied (not ignored by merge logic)
|
|
|
|
|
+ ams_data = mqtt_client.state.raw_data.get("ams", [])
|
|
|
|
|
+ tray = ams_data[0]["tray"][0]
|
|
|
|
|
+ assert tray["tray_type"] == "", "tray_type should be cleared when slot is empty"
|
|
|
|
|
+ assert tray["tray_color"] == "", "tray_color should be cleared when slot is empty"
|
|
|
|
|
+ assert tray["tray_sub_brands"] == "", "tray_sub_brands should be cleared"
|
|
|
|
|
+ assert tray["tag_uid"] == "0000000000000000", "tag_uid should be cleared"
|
|
|
|
|
+
|
|
|
|
|
+ def test_partial_update_preserves_other_fields(self, mqtt_client):
|
|
|
|
|
+ """Test that partial updates still preserve non-slot-status fields."""
|
|
|
|
|
+ # Initial state with full data
|
|
|
|
|
+ initial_ams = {
|
|
|
|
|
+ "ams": [
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": 0,
|
|
|
|
|
+ "humidity": "3",
|
|
|
|
|
+ "temp": "25.5",
|
|
|
|
|
+ "tray": [
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": 0,
|
|
|
|
|
+ "tray_type": "PLA",
|
|
|
|
|
+ "tray_color": "00FF00",
|
|
|
|
|
+ "remain": 90,
|
|
|
|
|
+ "k": 0.02,
|
|
|
|
|
+ }
|
|
|
|
|
+ ],
|
|
|
|
|
+ }
|
|
|
|
|
+ ]
|
|
|
|
|
+ }
|
|
|
|
|
+ mqtt_client._handle_ams_data(initial_ams)
|
|
|
|
|
+
|
|
|
|
|
+ # Partial update - only remain changes
|
|
|
|
|
+ partial_update = {
|
|
|
|
|
+ "ams": [
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": 0,
|
|
|
|
|
+ "tray": [
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": 0,
|
|
|
|
|
+ "remain": 85, # Only this changed
|
|
|
|
|
+ }
|
|
|
|
|
+ ],
|
|
|
|
|
+ }
|
|
|
|
|
+ ]
|
|
|
|
|
+ }
|
|
|
|
|
+ mqtt_client._handle_ams_data(partial_update)
|
|
|
|
|
+
|
|
|
|
|
+ # Verify remain was updated but other fields preserved
|
|
|
|
|
+ ams_data = mqtt_client.state.raw_data.get("ams", [])
|
|
|
|
|
+ tray = ams_data[0]["tray"][0]
|
|
|
|
|
+ assert tray["remain"] == 85, "remain should be updated"
|
|
|
|
|
+ assert tray["tray_type"] == "PLA", "tray_type should be preserved"
|
|
|
|
|
+ assert tray["tray_color"] == "00FF00", "tray_color should be preserved"
|
|
|
|
|
+ assert tray["k"] == 0.02, "k should be preserved"
|
|
|
|
|
+
|
|
|
|
|
+ def test_tray_exist_bits_clears_empty_slots(self, mqtt_client):
|
|
|
|
|
+ """Test that tray_exist_bits clears slots marked as empty (Issue #147).
|
|
|
|
|
+
|
|
|
|
|
+ New AMS models (AMS 2 Pro) don't send empty tray data when a spool is removed.
|
|
|
|
|
+ Instead, they update tray_exist_bits to indicate which slots have spools.
|
|
|
|
|
+ """
|
|
|
|
|
+ # Initial state: AMS 0 and AMS 1 with loaded spools
|
|
|
|
|
+ initial_ams = {
|
|
|
|
|
+ "ams": [
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": 0,
|
|
|
|
|
+ "tray": [
|
|
|
|
|
+ {"id": 0, "tray_type": "PLA", "tray_color": "FF0000", "remain": 80},
|
|
|
|
|
+ {"id": 1, "tray_type": "PETG", "tray_color": "00FF00", "remain": 60},
|
|
|
|
|
+ {"id": 2, "tray_type": "ABS", "tray_color": "0000FF", "remain": 40},
|
|
|
|
|
+ {"id": 3, "tray_type": "TPU", "tray_color": "FFFF00", "remain": 20},
|
|
|
|
|
+ ],
|
|
|
|
|
+ },
|
|
|
|
|
+ {
|
|
|
|
|
+ "id": 1,
|
|
|
|
|
+ "tray": [
|
|
|
|
|
+ {"id": 0, "tray_type": "PLA", "tray_color": "FFFFFF", "remain": 90},
|
|
|
|
|
+ {"id": 1, "tray_type": "PLA", "tray_color": "000000", "remain": 70},
|
|
|
|
|
+ {"id": 2, "tray_type": "PLA", "tray_color": "FF00FF", "remain": 50},
|
|
|
|
|
+ {"id": 3, "tray_type": "PLA", "tray_color": "00FFFF", "remain": 30},
|
|
|
|
|
+ ],
|
|
|
|
|
+ },
|
|
|
|
|
+ ],
|
|
|
|
|
+ "tray_exist_bits": "ff", # All 8 slots have spools (0xFF = 11111111)
|
|
|
|
|
+ }
|
|
|
|
|
+ mqtt_client._handle_ams_data(initial_ams)
|
|
|
|
|
+
|
|
|
|
|
+ # Verify initial state
|
|
|
|
|
+ ams_data = mqtt_client.state.raw_data.get("ams", [])
|
|
|
|
|
+ assert ams_data[1]["tray"][3]["tray_type"] == "PLA" # AMS 1 slot 3 (B4) has spool
|
|
|
|
|
+
|
|
|
|
|
+ # Now simulate spool removal from AMS 1 slot 3 (B4)
|
|
|
|
|
+ # tray_exist_bits: 0x7f = 01111111 (bit 7 = 0 means AMS 1 slot 3 is empty)
|
|
|
|
|
+ update_ams = {
|
|
|
|
|
+ "ams": [
|
|
|
|
|
+ {"id": 0, "tray": [{"id": 0}, {"id": 1}, {"id": 2}, {"id": 3}]},
|
|
|
|
|
+ {"id": 1, "tray": [{"id": 0}, {"id": 1}, {"id": 2}, {"id": 3}]},
|
|
|
|
|
+ ],
|
|
|
|
|
+ "tray_exist_bits": "7f", # Bit 7 = 0 -> AMS 1 slot 3 is empty
|
|
|
|
|
+ }
|
|
|
|
|
+ mqtt_client._handle_ams_data(update_ams)
|
|
|
|
|
+
|
|
|
|
|
+ # Verify AMS 1 slot 3 was cleared
|
|
|
|
|
+ ams_data = mqtt_client.state.raw_data.get("ams", [])
|
|
|
|
|
+ b4_tray = ams_data[1]["tray"][3]
|
|
|
|
|
+ assert b4_tray["tray_type"] == "", "tray_type should be cleared for empty slot"
|
|
|
|
|
+ assert b4_tray["remain"] == 0, "remain should be 0 for empty slot"
|
|
|
|
|
+
|
|
|
|
|
+ # Verify other slots are preserved
|
|
|
|
|
+ assert ams_data[0]["tray"][0]["tray_type"] == "PLA", "A1 should still have PLA"
|
|
|
|
|
+ assert ams_data[1]["tray"][0]["tray_type"] == "PLA", "B1 should still have PLA"
|