mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-04 21:21:54 +02:00
400 lines
18 KiB
Python
400 lines
18 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, _resolve_temp_alarm_threshold
|
|
from backend.app.utils.ams_drying import is_countdown_parked, 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_parked_countdown_is_not_active(self):
|
|
# #2896: the printer took the command but the AMS is not heating, so a
|
|
# high temperature is not "expected heat" and must not be suppressed.
|
|
assert is_drying_active({"dry_time": 720, "dry_countdown_stalled": True}) is False
|
|
assert is_drying_active({"dry_time": 720, "dry_countdown_stalled": False}) is True
|
|
|
|
|
|
class TestIsCountdownParked:
|
|
def test_stalled_timer_is_parked(self):
|
|
assert is_countdown_parked({"dry_time": 720, "dry_countdown_stalled": True}) is True
|
|
assert is_countdown_parked({"dry_time": "720", "dry_countdown_stalled": True}) is True
|
|
|
|
def test_running_or_absent_timer_is_not_parked(self):
|
|
assert is_countdown_parked({"dry_time": 720}) is False
|
|
assert is_countdown_parked({"dry_time": 720, "dry_countdown_stalled": False}) is False
|
|
# A flag left behind on a finished cycle does not make it parked.
|
|
assert is_countdown_parked({"dry_time": 0, "dry_countdown_stalled": True}) is False
|
|
|
|
def test_malformed_input_is_not_parked(self):
|
|
assert is_countdown_parked(None) is False
|
|
assert is_countdown_parked({"dry_time": "x", "dry_countdown_stalled": True}) 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
|
|
|
|
|
|
class TestTempAlarmThresholdResolution:
|
|
"""The alarm gets its own threshold, falling back to the display band (#2905).
|
|
|
|
``ams_temp_fair`` decides when the AMS card turns amber and used to decide
|
|
when a notification was sent as well. 35 C is a reasonable place to change a
|
|
colour and not a reasonable place to page someone: a room above it made the
|
|
alarm fire once an hour for as long as the weather lasted, and the only way
|
|
to stop it was to raise the display band and lose the colour that says the
|
|
unit is warm.
|
|
"""
|
|
|
|
def test_unset_falls_back_to_the_fair_threshold(self):
|
|
"""Every install that has never set one keeps behaving exactly as it does
|
|
now — that is what makes this safe to ship without a migration."""
|
|
assert _resolve_temp_alarm_threshold(35.0, None) == 35.0
|
|
|
|
def test_the_literal_none_string_falls_back_too(self):
|
|
"""Settings storage stringifies None, so "not set" arrives as a string.
|
|
Handled by the same branch as any other unparseable value rather than by
|
|
a sentinel that has to be kept in sync."""
|
|
assert _resolve_temp_alarm_threshold(35.0, "None") == 35.0
|
|
|
|
def test_a_set_value_wins(self):
|
|
assert _resolve_temp_alarm_threshold(35.0, "45") == 45.0
|
|
assert _resolve_temp_alarm_threshold(35.0, "45.5") == 45.5
|
|
|
|
def test_it_tracks_an_edited_fair_threshold_while_unset(self):
|
|
"""Seeded from the resolved fair value, not from a hardcoded 35."""
|
|
assert _resolve_temp_alarm_threshold(40.0, None) == 40.0
|
|
|
|
def test_a_value_below_the_display_band_is_honoured(self):
|
|
"""Nothing requires the alarm to sit above the amber band. Someone who
|
|
wants to be told before the card even changes colour may say so."""
|
|
assert _resolve_temp_alarm_threshold(35.0, "30") == 30.0
|
|
|
|
def test_garbage_falls_back_rather_than_raising(self):
|
|
assert _resolve_temp_alarm_threshold(35.0, "") == 35.0
|
|
assert _resolve_temp_alarm_threshold(35.0, "warm") == 35.0
|
|
|
|
def test_zero_and_negative_are_refused(self):
|
|
"""Zero would alarm permanently, and is far more likely to be a cleared
|
|
field than a deliberate choice."""
|
|
assert _resolve_temp_alarm_threshold(35.0, "0") == 35.0
|
|
assert _resolve_temp_alarm_threshold(35.0, "-5") == 35.0
|
|
|
|
def test_nan_and_infinity_are_refused(self):
|
|
assert _resolve_temp_alarm_threshold(35.0, "nan") == 35.0
|
|
assert _resolve_temp_alarm_threshold(35.0, "inf") == 35.0
|
|
|
|
|
|
class TestLatchReleasesAtTheAlarmThreshold:
|
|
"""The latch's release check takes the alarm threshold, not the display band.
|
|
|
|
``temperature_alarm_suppressed`` releases once the unit reads at or below
|
|
``threshold``. Handing it the display band strands the latch on any unit that
|
|
settles back above it — which is not hypothetical: the AMS this was reported
|
|
from rests at 37.7 C in a warm room and never returns under a 35 C band, so
|
|
the latch could only ever expire on the grace cap.
|
|
"""
|
|
|
|
def test_a_unit_resting_above_the_display_band_releases_on_the_alarm_threshold(self):
|
|
"""37.7 C after a cycle, alarm threshold 45: released promptly."""
|
|
latched = datetime.now(timezone.utc) - timedelta(minutes=5)
|
|
suppress, latch = temperature_alarm_suppressed(
|
|
drying_active=False,
|
|
temperature=37.7,
|
|
threshold=45.0,
|
|
latched_at=latched,
|
|
now=datetime.now(timezone.utc),
|
|
grace_minutes=120,
|
|
)
|
|
assert suppress is False
|
|
assert latch is None
|
|
|
|
def test_the_same_reading_stays_latched_against_the_display_band(self):
|
|
"""The behaviour before this change, kept as the contrast: 37.7 C never
|
|
drops under 35, so the latch survives and can only expire on the cap."""
|
|
latched = datetime.now(timezone.utc) - timedelta(minutes=5)
|
|
suppress, latch = temperature_alarm_suppressed(
|
|
drying_active=False,
|
|
temperature=37.7,
|
|
threshold=35.0,
|
|
latched_at=latched,
|
|
now=datetime.now(timezone.utc),
|
|
grace_minutes=120,
|
|
)
|
|
assert suppress is True
|
|
assert latch == latched
|
|
|
|
def test_a_genuinely_hot_unit_still_alarms_after_the_cap(self):
|
|
"""Releasing on the cap is not a new alert — a unit that stays that hot
|
|
would have been alarming with no drying involved."""
|
|
latched = datetime.now(timezone.utc) - timedelta(minutes=121)
|
|
suppress, latch = temperature_alarm_suppressed(
|
|
drying_active=False,
|
|
temperature=70.0,
|
|
threshold=45.0,
|
|
latched_at=latched,
|
|
now=datetime.now(timezone.utc),
|
|
grace_minutes=120,
|
|
)
|
|
assert suppress is False
|
|
assert latch is None
|