mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
e336eb53e7
Assigning a new filament to the external spool (e.g. generic ABS over generic TPU) left the previous inventory spool assigned. The reconciliation that unlinks a stale external-spool assignment lives in on_ams_change, but that callback only fired on regular AMS-unit changes: its change-hash never included the external spool (vt_tray/vir_slot), and the external-spool data is stored after the AMS handler runs. Detect external-spool identity changes (type, colour, tag, or reset to empty) and re-fire on_ams_change so the stale assignment is unlinked. The fill percentage (remain) is excluded from the fingerprint so a running print doesn't trigger it on every push.
93 lines
3.7 KiB
Python
93 lines
3.7 KiB
Python
"""External-spool (vt_tray) change detection (#2575).
|
|
|
|
The AMS change-hash in ``_handle_ams_data`` is built only from AMS units, so a
|
|
filament swap on the external spool alone (e.g. generic TPU -> generic ABS on
|
|
the printer) used to never re-trigger ``on_ams_change``. That left a stale
|
|
inventory assignment on the ``ams_id=255`` slot: Bambuddy kept showing the old
|
|
filament after the physical type had changed.
|
|
|
|
These tests drive full MQTT messages through ``_process_message`` and assert the
|
|
callback fires exactly when the external spool's *identity* changes — and not on
|
|
every push (e.g. a steadily-dropping ``remain`` percentage during a print).
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from backend.app.services.bambu_mqtt import BambuMQTTClient
|
|
|
|
|
|
def _ext_spool_msg(tray_type: str, remain: int = 100, color: str = "000000FF"):
|
|
"""A realistic print message carrying only external-spool (vt_tray) data."""
|
|
return {
|
|
"print": {
|
|
"vt_tray": {
|
|
"id": "254",
|
|
"tray_type": tray_type,
|
|
"tray_color": color,
|
|
"tray_info_idx": "",
|
|
"tag_uid": "0000000000000000",
|
|
"tray_uuid": "00000000000000000000000000000000",
|
|
"remain": remain,
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
class TestExternalSpoolChangeDetection:
|
|
@pytest.fixture
|
|
def mqtt_client(self):
|
|
return BambuMQTTClient(
|
|
ip_address="192.168.1.100",
|
|
serial_number="TEST123",
|
|
access_code="12345678",
|
|
)
|
|
|
|
def test_type_swap_fires_callback(self, mqtt_client):
|
|
"""Swapping the external filament type re-triggers the sync callback."""
|
|
calls: list = []
|
|
mqtt_client.on_ams_change = lambda ams_data: calls.append(ams_data)
|
|
|
|
# First observation of the external spool (TPU) — fires once.
|
|
mqtt_client._process_message(_ext_spool_msg("TPU"))
|
|
assert len(calls) == 1
|
|
|
|
# Physical filament changed to ABS — must fire again so the stale
|
|
# ams_id=255 assignment gets reconciled.
|
|
mqtt_client._process_message(_ext_spool_msg("ABS"))
|
|
assert len(calls) == 2
|
|
|
|
# The callback receives the merged AMS list (never None).
|
|
assert all(isinstance(c, list) for c in calls)
|
|
|
|
def test_identical_push_does_not_refire(self, mqtt_client):
|
|
"""Repeated identical vt_tray pushes fire the callback only once."""
|
|
calls: list = []
|
|
mqtt_client.on_ams_change = lambda ams_data: calls.append(ams_data)
|
|
|
|
mqtt_client._process_message(_ext_spool_msg("ABS"))
|
|
mqtt_client._process_message(_ext_spool_msg("ABS"))
|
|
mqtt_client._process_message(_ext_spool_msg("ABS"))
|
|
assert len(calls) == 1
|
|
|
|
def test_remain_only_change_does_not_refire(self, mqtt_client):
|
|
"""A dropping fill percentage must not spam the reconciliation callback."""
|
|
calls: list = []
|
|
mqtt_client.on_ams_change = lambda ams_data: calls.append(ams_data)
|
|
|
|
mqtt_client._process_message(_ext_spool_msg("PLA", remain=100))
|
|
assert len(calls) == 1
|
|
# remain drops during a print — identity unchanged, no refire.
|
|
mqtt_client._process_message(_ext_spool_msg("PLA", remain=87))
|
|
mqtt_client._process_message(_ext_spool_msg("PLA", remain=42))
|
|
assert len(calls) == 1
|
|
|
|
def test_reset_to_empty_fires_callback(self, mqtt_client):
|
|
"""Resetting the external spool (empty tray_type) is an identity change."""
|
|
calls: list = []
|
|
mqtt_client.on_ams_change = lambda ams_data: calls.append(ams_data)
|
|
|
|
mqtt_client._process_message(_ext_spool_msg("TPU"))
|
|
assert len(calls) == 1
|
|
mqtt_client._process_message(_ext_spool_msg("")) # reset / unloaded
|
|
assert len(calls) == 2
|