mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-21 10:15:04 +02:00
The alert compares against ams_temp_fair, the same threshold that colours the printer card, which defaults to 35C. Drying deliberately runs at 45C for PLA, 65C for PETG and up to 85C on an AMS-HT, and the alert repeats once an hour for as long as the condition holds, so a twelve-hour dry sent twelve notifications about a temperature the user chose. It then kept sending them while the unit cooled back down, which is the half the reporter confirmed on an AMS 2 Pro and an H2C. Dispatch now consults the drying state the firmware already reports. dry_time alone is not enough: it reads 0 through the cooling phase that closes a cycle, so dry_status -- info bits 4-7, already parsed for the drying-complete edge -- carries the rest. That constant moves out of bambu_mqtt into a leaf util rather than being duplicated; drying_preflight would have been the natural home, but it imports printer_manager, which imports bambu_mqtt, and bambu_mqtt is one of the callers. The cool-down afterwards is held by a latch released as soon as the unit reads back at or below the threshold, rather than after a fixed delay, so a 65C cycle in a cold basement and a 45C one in a warm room each get the time they actually need. A two-hour cap bounds the one case the latch cannot resolve on its own -- a unit that never returns below the threshold -- and since such a unit would have been alarming with no drying involved, releasing there restores the ordinary behaviour instead of inventing a new alert. Two exclusions are deliberate. Humidity is untouched, because during drying that reading falling is the whole point. And dry_status 6, HeatOutOfControl, is kept out of the active set: an AMS that has lost thermal control is exactly when the alert should still arrive, so it must never read as expected heat. A cycle plus its cool-down outlasts a restart, so the latch is a settings row rather than a dict beside _ams_alarm_cooldown -- the internal timestamp-row pattern support.py already uses. It is read once per pass and written back only when a unit changed it. Stamps ahead of now are clamped on read, since a box whose clock jumps backwards writes them and suppression is measured as now minus the stamp; without the clamp the cap would measure from a moment that has not happened yet and hold the alert quiet for the skew on top of it. No new setting. The reporter was offered the opt-out checkbox they asked for and said they would not want it if the alert simply never fired during drying.
273 lines
12 KiB
Python
273 lines
12 KiB
Python
"""Tests for the gates that hold back AMS humidity / temperature alarms.
|
|
|
|
Two independent gates, both sitting in ``record_ams_history``'s dispatch: the
|
|
empty-AMS gate (#1619) documented below, and the drying gate (#1802) that stops
|
|
the temperature alarm firing throughout a drying cycle and the cool-down after
|
|
it.
|
|
|
|
Empty-AMS alarm gate (#1619).
|
|
|
|
Empty AMS units still emit humidity/temperature sensor readings, but those
|
|
readings are ambient and not actionable — there's no filament to dry. Without
|
|
the gate every empty AMS spammed an hourly alarm. ``_ams_has_filament``
|
|
inspects the firmware-reported ``tray_exist_bits`` bitmap (fallback: ``tray``
|
|
array's ``tray_type`` strings) so the alarm dispatch in ``record_ams_history``
|
|
can skip empty units while still alarming on loaded ones in the same printer.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from backend.app.main import _ams_has_filament
|
|
from backend.app.utils.ams_drying import is_drying_active, temperature_alarm_suppressed
|
|
|
|
|
|
class TestAmsHasFilament:
|
|
def test_tray_exist_bits_zero_means_empty(self):
|
|
assert _ams_has_filament({"tray_exist_bits": "0"}) is False
|
|
# Real firmware sometimes pads with extra zeros or prefixes; all
|
|
# parseable forms of zero should resolve to "empty".
|
|
assert _ams_has_filament({"tray_exist_bits": "00"}) is False
|
|
assert _ams_has_filament({"tray_exist_bits": "0x0"}) is False
|
|
|
|
def test_tray_exist_bits_nonzero_means_loaded(self):
|
|
# Single tray loaded — e.g. AMS-Lite or AMS-HT.
|
|
assert _ams_has_filament({"tray_exist_bits": "1"}) is True
|
|
# Four-slot AMS with all slots full (bitmap 0xf == 0b1111).
|
|
assert _ams_has_filament({"tray_exist_bits": "f"}) is True
|
|
# Mixed — 0xa == 0b1010, two slots loaded.
|
|
assert _ams_has_filament({"tray_exist_bits": "a"}) is True
|
|
# The exact bitmap seen in #1622 / #1602 logs.
|
|
assert _ams_has_filament({"tray_exist_bits": "ed"}) is True
|
|
|
|
def test_falls_back_to_tray_array_when_bits_missing(self):
|
|
# Empty tray_type strings across the whole tray array → empty AMS.
|
|
ams_empty = {
|
|
"tray": [
|
|
{"id": 0, "tray_type": ""},
|
|
{"id": 1, "tray_type": ""},
|
|
]
|
|
}
|
|
assert _ams_has_filament(ams_empty) is False
|
|
# Any non-empty tray_type → loaded AMS.
|
|
ams_loaded = {
|
|
"tray": [
|
|
{"id": 0, "tray_type": ""},
|
|
{"id": 1, "tray_type": "PLA"},
|
|
]
|
|
}
|
|
assert _ams_has_filament(ams_loaded) is True
|
|
|
|
def test_missing_both_signals_returns_false(self):
|
|
# No tray_exist_bits AND no tray array — early-pushall shape; we
|
|
# treat it as "no info → don't alarm" rather than guessing loaded.
|
|
assert _ams_has_filament({}) is False
|
|
|
|
def test_unparseable_bitmap_falls_back_to_tray_array(self):
|
|
# Garbage in tray_exist_bits — must not raise and must fall through
|
|
# to the tray array check.
|
|
loaded = {"tray_exist_bits": "garbage", "tray": [{"id": 0, "tray_type": "PETG"}]}
|
|
assert _ams_has_filament(loaded) is True
|
|
empty = {"tray_exist_bits": "garbage", "tray": []}
|
|
assert _ams_has_filament(empty) is False
|
|
|
|
def test_empty_bits_string_falls_back_to_tray_array(self):
|
|
# Some pre-handshake pushall shapes set the field but leave it blank.
|
|
loaded = {"tray_exist_bits": "", "tray": [{"id": 0, "tray_type": "ABS"}]}
|
|
assert _ams_has_filament(loaded) is True
|
|
|
|
def test_whitespace_tray_type_is_not_loaded(self):
|
|
# A tray_type that's all whitespace doesn't count as a real material.
|
|
assert _ams_has_filament({"tray": [{"id": 0, "tray_type": " "}]}) is False
|
|
|
|
def test_non_dict_tray_entries_are_skipped(self):
|
|
# Defensive: malformed tray array shouldn't crash the helper.
|
|
assert _ams_has_filament({"tray": [None, "junk", 42]}) is False
|
|
|
|
def test_non_string_bits_falls_back(self):
|
|
# Some MQTT shapes send tray_exist_bits as int; we only parse strings,
|
|
# so an int falls through to the tray array.
|
|
loaded = {"tray_exist_bits": 0xED, "tray": [{"id": 0, "tray_type": "PLA"}]}
|
|
assert _ams_has_filament(loaded) is True
|
|
empty_int = {"tray_exist_bits": 0xED} # no tray array, int ignored
|
|
assert _ams_has_filament(empty_int) is False
|
|
|
|
|
|
class TestIsDryingActive:
|
|
"""The two firmware signals that mean "a drying cycle is running" (#1802)."""
|
|
|
|
def test_countdown_running_is_active(self):
|
|
assert is_drying_active({"dry_time": 720}) is True
|
|
# Strings appear in some payload shapes.
|
|
assert is_drying_active({"dry_time": "45"}) is True
|
|
|
|
def test_idle_unit_is_not_active(self):
|
|
assert is_drying_active({"dry_time": 0, "dry_status": 0}) is False
|
|
assert is_drying_active({}) is False
|
|
|
|
def test_cooling_phase_counts_as_active(self):
|
|
# The reason dry_time alone is not enough: the cycle's own cooling phase
|
|
# runs with the countdown already at 0.
|
|
assert is_drying_active({"dry_time": 0, "dry_status": 3}) is True
|
|
|
|
def test_checking_and_drying_phases_count_as_active(self):
|
|
assert is_drying_active({"dry_time": 0, "dry_status": 1}) is True
|
|
assert is_drying_active({"dry_time": 0, "dry_status": 2}) is True
|
|
|
|
def test_ending_phases_do_not_count_as_active(self):
|
|
# 4=Stopping, 5=Error — the cycle is over or aborting.
|
|
assert is_drying_active({"dry_time": 0, "dry_status": 4}) is False
|
|
assert is_drying_active({"dry_time": 0, "dry_status": 5}) is False
|
|
|
|
def test_heat_out_of_control_is_not_active(self):
|
|
# 6=HeatOutOfControl is the one phase where a high-temperature alarm is
|
|
# exactly what the user needs, so it must never read as expected heat.
|
|
assert is_drying_active({"dry_time": 0, "dry_status": 6}) is False
|
|
|
|
def test_missing_dry_status_falls_back_to_countdown(self):
|
|
# Firmware that never sends a parseable `info` has no dry_status at all.
|
|
assert is_drying_active({"dry_time": 30}) is True
|
|
assert is_drying_active({"dry_time": 0}) is False
|
|
|
|
def test_unparseable_values_do_not_raise(self):
|
|
assert is_drying_active({"dry_time": "junk", "dry_status": 2}) is True
|
|
assert is_drying_active({"dry_time": None, "dry_status": None}) is False
|
|
assert is_drying_active({"dry_time": "junk", "dry_status": "junk"}) is False
|
|
|
|
def test_non_mapping_input_is_not_active(self):
|
|
assert is_drying_active(None) is False
|
|
assert is_drying_active("drying") is False
|
|
assert is_drying_active(42) is False
|
|
|
|
|
|
class TestTemperatureAlarmSuppressed:
|
|
"""Latch behaviour for the AMS high-temperature alarm during drying (#1802)."""
|
|
|
|
NOW = datetime(2026, 8, 16, 12, 0, tzinfo=timezone.utc)
|
|
GRACE = 120
|
|
|
|
def _call(self, **overrides):
|
|
kwargs = {
|
|
"drying_active": False,
|
|
"temperature": 50.0,
|
|
"threshold": 35.0,
|
|
"latched_at": None,
|
|
"now": self.NOW,
|
|
"grace_minutes": self.GRACE,
|
|
}
|
|
kwargs.update(overrides)
|
|
return temperature_alarm_suppressed(**kwargs)
|
|
|
|
def test_no_drying_no_latch_alarms_normally(self):
|
|
# The pre-#1802 behaviour has to survive untouched for units that never dry.
|
|
suppress, latch = self._call(temperature=40.0)
|
|
assert suppress is False
|
|
assert latch is None
|
|
|
|
def test_drying_suppresses_and_sets_latch(self):
|
|
suppress, latch = self._call(drying_active=True, temperature=65.0)
|
|
assert suppress is True
|
|
assert latch == self.NOW
|
|
|
|
def test_drying_latches_even_when_below_threshold(self):
|
|
# Early in a cycle the unit is still heating up. The latch has to be set
|
|
# then too, or the cool-down afterwards starts unprotected.
|
|
suppress, latch = self._call(drying_active=True, temperature=28.0)
|
|
assert suppress is True
|
|
assert latch == self.NOW
|
|
|
|
def test_still_hot_after_cycle_stays_suppressed(self):
|
|
# The reported symptom: alarms kept arriving while the unit cooled.
|
|
suppress, latch = self._call(
|
|
temperature=52.0,
|
|
latched_at=self.NOW - timedelta(minutes=20),
|
|
)
|
|
assert suppress is True
|
|
assert latch == self.NOW - timedelta(minutes=20)
|
|
|
|
def test_cooled_back_to_normal_clears_latch(self):
|
|
suppress, latch = self._call(
|
|
temperature=34.0,
|
|
latched_at=self.NOW - timedelta(minutes=40),
|
|
)
|
|
assert suppress is False
|
|
assert latch is None
|
|
|
|
def test_exactly_at_threshold_counts_as_cooled(self):
|
|
# The alarm itself fires on `> threshold`, so `== threshold` is not hot.
|
|
suppress, latch = self._call(
|
|
temperature=35.0,
|
|
latched_at=self.NOW - timedelta(minutes=40),
|
|
)
|
|
assert suppress is False
|
|
assert latch is None
|
|
|
|
def test_alarms_again_after_the_latch_is_cleared(self):
|
|
# Having cooled once, a later genuine overheat is not swallowed.
|
|
_, latch = self._call(temperature=34.0, latched_at=self.NOW - timedelta(minutes=40))
|
|
suppress, latch = self._call(temperature=48.0, latched_at=latch)
|
|
assert suppress is False
|
|
assert latch is None
|
|
|
|
def test_grace_cap_releases_a_unit_that_never_cools(self):
|
|
# A unit stuck above the threshold would have alarmed with no drying
|
|
# involved, so the cap restores that rather than inventing an alert.
|
|
suppress, latch = self._call(
|
|
temperature=45.0,
|
|
latched_at=self.NOW - timedelta(minutes=self.GRACE + 1),
|
|
)
|
|
assert suppress is False
|
|
assert latch is None
|
|
|
|
def test_grace_cap_boundary_releases(self):
|
|
suppress, _ = self._call(
|
|
temperature=45.0,
|
|
latched_at=self.NOW - timedelta(minutes=self.GRACE),
|
|
)
|
|
assert suppress is False
|
|
|
|
def test_just_inside_the_grace_cap_still_suppresses(self):
|
|
suppress, _ = self._call(
|
|
temperature=45.0,
|
|
latched_at=self.NOW - timedelta(minutes=self.GRACE - 1),
|
|
)
|
|
assert suppress is True
|
|
|
|
def test_a_new_cycle_refreshes_the_latch(self):
|
|
# Starting a second dry inside the grace window must restart the clock,
|
|
# otherwise the cap could expire midway through the new cycle.
|
|
suppress, latch = self._call(
|
|
drying_active=True,
|
|
temperature=60.0,
|
|
latched_at=self.NOW - timedelta(minutes=self.GRACE - 5),
|
|
)
|
|
assert suppress is True
|
|
assert latch == self.NOW
|
|
|
|
def test_unreadable_temperature_holds_the_latch(self):
|
|
# A dropped reading is not evidence the unit cooled, and there is no
|
|
# alarm to fire on this pass anyway.
|
|
suppress, latch = self._call(
|
|
temperature=None,
|
|
latched_at=self.NOW - timedelta(minutes=10),
|
|
)
|
|
assert suppress is True
|
|
assert latch == self.NOW - timedelta(minutes=10)
|
|
|
|
def test_the_cap_is_measured_from_the_latch(self):
|
|
# Guards the precondition the loader's clamp exists to maintain: with a
|
|
# non-future latch, suppression expires exactly one cap after it, so the
|
|
# cap is a real bound rather than a floor. A future latch would push the
|
|
# release out by the skew as well, which is why the clamp is at the read
|
|
# — see _load_ams_drying_latch and its persistence tests.
|
|
latched = self.NOW - timedelta(minutes=self.GRACE)
|
|
suppress, latch = temperature_alarm_suppressed(
|
|
drying_active=False,
|
|
temperature=45.0,
|
|
threshold=35.0,
|
|
latched_at=latched,
|
|
now=self.NOW,
|
|
grace_minutes=self.GRACE,
|
|
)
|
|
assert suppress is False
|
|
assert latch is None
|