bambuddy/backend/app/utils/ams_drying.py

"""Shared reading of the firmware's own AMS drying state.

Kept as a leaf module on purpose. ``drying_preflight`` would be the natural
home, but it imports ``printer_manager``, which imports ``bambu_mqtt`` — and
``bambu_mqtt`` is one of the callers here, so putting these there would close an
import cycle. Nothing in this module imports from the app.
"""

from collections.abc import Mapping
from datetime import datetime, timedelta
from typing import Any

# ``dry_status`` is bits 4-7 of the per-AMS ``info`` hex string (BambuStudio
# DevFilaSystem.cpp): 0=Off, 1=Checking, 2=Drying, 3=Cooling, 4=Stopping,
# 5=Error, 6=HeatOutOfControl, 7=PrdTesting. Only the first three mean a cycle
# is still live.
#
# 4 (Stopping) and 5 (Error) are excluded because the cycle is over or ending.
# 6 (HeatOutOfControl) is excluded deliberately and for a different reason: an
# AMS that has lost thermal control is exactly when a high-temperature alarm
# should still reach the user, so it must never read as "expected heat".
ACTIVE_DRY_STATUSES = frozenset({1, 2, 3})  # Checking, Drying, Cooling

# A live drying countdown ticks down once per minute. A ``dry_time`` that has
# not changed for this long — with no active ``dry_status`` phase vouching for
# the cycle — is a timer with no cycle running behind it. Usually one that never
# started (seen on an H2D mid-print with two AMS-HT cycles already active: the
# third unit's timer froze at its full duration and no heating ever began), but a
# cycle paused partway reads the same. 150 s allows two full missed ticks plus
# jitter before the countdown is called stalled.
DRY_COUNTDOWN_STALL_SECONDS = 150


def is_drying_active(ams_data: Any) -> bool:
    """True when this AMS unit reports a drying cycle in progress.

    Two independent signals, because neither alone is sufficient. ``dry_time``
    is minutes remaining and reads 0 through the cooling phase that closes a
    cycle; ``dry_status`` covers that phase but is only present when the
    firmware sent a parseable ``info`` field.

    A parked countdown (``dry_countdown_stalled``, see is_countdown_parked) is
    not a cycle in progress: the AMS is not heating, so nothing it reports is
    "expected heat" from drying.
    """
    if not isinstance(ams_data, Mapping):
        return False
    if ams_data.get("dry_countdown_stalled"):
        return False
    try:
        if int(ams_data.get("dry_time") or 0) > 0:
            return True
    except (TypeError, ValueError):
        pass  # Unparseable countdown — fall through to the phase field
    try:
        return int(ams_data["dry_status"]) in ACTIVE_DRY_STATUSES
    except (KeyError, TypeError, ValueError):
        return False


def is_countdown_parked(ams_data: Any) -> bool:
    """True when this AMS unit holds a drying timer that is not running.

    The MQTT layer sets ``dry_countdown_stalled`` once ``dry_time`` has stayed
    unchanged past DRY_COUNTDOWN_STALL_SECONDS with no active ``dry_status``
    phase. Such a timer never reaches 0 on its own, so anything that waits for
    drying to finish must not wait on it.
    """
    if not isinstance(ams_data, Mapping) or not ams_data.get("dry_countdown_stalled"):
        return False
    try:
        return int(ams_data.get("dry_time") or 0) > 0
    except (TypeError, ValueError):
        return False


def temperature_alarm_suppressed(
    *,
    drying_active: bool,
    temperature: float | None,
    threshold: float,
    latched_at: datetime | None,
    now: datetime,
    grace_minutes: int,
) -> tuple[bool, datetime | None]:
    """Decide whether to hold back the AMS high-temperature alarm (#1802).

    Drying heats an AMS far past the alarm threshold by design — 45 C for PLA,
    65 C for PETG, up to 85 C on an AMS-HT, against a default threshold of
    35 C — so without this the alarm fires once an hour for the length of the
    cycle and keeps going while the unit cools back down.

    Returns ``(suppress, latched_at)``. The second element is the latch to
    persist: a timestamp while suppression is in force, ``None`` to clear it.

    Suppression is released as soon as the unit reads back at or below the
    threshold rather than after a fixed delay, so a 65 C cycle in a cold
    basement and a 45 C one in a warm room each get exactly the cool-down they
    need. ``grace_minutes`` only bounds the case where the unit never returns
    below the threshold at all — and a unit that stays that hot would have been
    alarming with no drying involved, so releasing there restores the ordinary
    behaviour instead of inventing a new alert.
    """
    if drying_active:
        return True, now
    if latched_at is None:
        return False, None
    # Back at a normal storage temperature: the cool-down is over. Note this is
    # also the only path that can clear the latch promptly, so it is checked
    # before the cap.
    if temperature is not None and temperature <= threshold:
        return False, None
    # ``latched_at`` is never in the future: the caller either just stamped it
    # with this ``now`` or read it back through a loader that clamps. A future
    # stamp would make this difference negative and hold suppression for the
    # skew on top of the cap, which is why the clamp lives at the read.
    if now - latched_at >= timedelta(minutes=grace_minutes):
        return False, None
    return True, latched_at