bambuddy/backend/tests/unit/test_obico_smoothing.py

"""Unit tests for Obico detection smoothing math.

The decision is Obico's own is_failing() (#3459): the score is the EWM minus
the printer's long-term mean, so what the camera always sees is baseline.
"""

import pytest

from backend.app.services.obico_smoothing import (
    BASE_HIGH,
    BASE_LOW,
    ESCALATING_FACTOR,
    WARMUP_FRAMES,
    Baseline,
    PrintState,
    Reading,
    classify,
    score_from_detections,
    thresholds,
    triggers_action,
)


class TestThresholds:
    def test_medium_matches_base(self):
        low, high = thresholds("medium")
        assert low == pytest.approx(BASE_LOW)
        assert high == pytest.approx(BASE_HIGH)

    def test_low_sensitivity_is_stricter(self):
        low, high = thresholds("low")
        assert low > BASE_LOW
        assert high > BASE_HIGH

    def test_high_sensitivity_is_looser(self):
        low, high = thresholds("high")
        assert low < BASE_LOW
        assert high < BASE_HIGH

    def test_unknown_falls_back_to_medium(self):
        assert thresholds("bogus") == thresholds("medium")


class TestScoreFromDetections:
    def test_empty(self):
        assert score_from_detections([]) == 0.0
        assert score_from_detections(None) == 0.0

    def test_sums_confidences(self):
        dets = [["failure", 0.3, [0, 0, 10, 10]], ["failure", 0.5, [0, 0, 10, 10]]]
        assert score_from_detections(dets) == pytest.approx(0.8)

    def test_ignores_malformed(self):
        dets = [["failure", 0.4, []], ["bad"], ["failure", "nan", []]]
        assert score_from_detections(dets) == pytest.approx(0.4)


def _feed(state: PrintState, value: float, frames: int) -> Reading:
    reading = Reading()
    for _ in range(frames):
        reading = state.update(value)
    return reading


class TestPrintState:
    def test_warmup_is_always_safe(self):
        state = PrintState()
        for _ in range(WARMUP_FRAMES):
            assert classify(state.update(5.0), "high") == "safe"

    def test_sustained_zero_stays_safe(self):
        state = PrintState()
        assert classify(_feed(state, 0.0, WARMUP_FRAMES + 50), "high") == "safe"

    def test_a_steady_background_from_the_start_is_baseline(self):
        # The plate artwork in #3459: 5 boxes summing to 0.82 in every frame.
        # The raw EWM sat above HIGH; minus the baseline it is nothing.
        state = PrintState()
        reading = _feed(state, 0.82, WARMUP_FRAMES + 2000)
        assert state.ewm_mean > BASE_HIGH
        assert reading.score == pytest.approx(0.0, abs=1e-9)
        assert classify(reading, "high") == "safe"

    def test_the_baseline_outlives_the_print(self):
        # Seen on every print of this printer: the next print starts with it.
        baseline = Baseline()
        _feed(PrintState(baseline=baseline), 0.82, 3000)
        reading = _feed(PrintState(baseline=baseline), 0.82, WARMUP_FRAMES + 5)
        assert classify(reading, "medium") == "safe"
        # Without it, the same frames appearing on a clean printer are a
        # sudden sustained rise, which is what Obico pauses for.
        fresh = PrintState()
        _feed(fresh, 0.0, 600)
        assert classify(_feed(fresh, 0.82, 40), "medium") == "failure"

    def test_a_sudden_sustained_rise_pauses(self):
        state = PrintState()
        _feed(state, 0.0, 600)
        reading = _feed(state, 3.0, 40)
        assert reading.score / ESCALATING_FACTOR > BASE_HIGH
        assert classify(reading, "medium") == "failure"

    def test_a_late_sudden_background_is_a_rise_then_settles(self):
        # #3459: ~4 h of nothing, then boxes (~0.8) in every frame. That is a
        # sudden sustained rise, which Obico pauses for, like spaghetti; once
        # the last ~52 min account for it, it is no longer failing.
        state = PrintState()
        _feed(state, 0.0, 1550)
        verdicts = [classify(state.update(0.8), "medium") for _ in range(200)]
        assert "failure" in verdicts[:40]
        assert verdicts[-40:] == ["safe"] * 40

    def test_a_slow_build_up_warns_without_pausing(self):
        # The same 0.8 reached over ~25 minutes: Obico alerts, never pauses.
        state = PrintState()
        _feed(state, 0.0, 1550)
        verdicts = {classify(state.update(0.8 * min(1.0, i / 150)), "medium") for i in range(1, 300)}
        assert verdicts == {"safe", "warning"}

    def test_in_the_band_the_short_window_rise_must_be_beaten(self):
        # Between LOW and HIGH, a score that a slow build-up over the last
        # ~52 min accounts for (x3.8) is not failing.
        between = (BASE_LOW + BASE_HIGH) / 2
        assert classify(Reading(score=between, short_rise=between / 3.8 + 0.01, warmup=False), "medium") == "safe"
        assert classify(Reading(score=between, short_rise=between / 3.8 - 0.01, warmup=False), "medium") == "warning"

    def test_an_isolated_spike_does_not_alert(self):
        state = PrintState()
        _feed(state, 0.0, WARMUP_FRAMES + 100)
        assert classify(state.update(1.0), "medium") == "safe"


class TestClassify:
    @staticmethod
    def _r(score: float, short_rise: float = 0.0) -> Reading:
        return Reading(score=score, short_rise=short_rise, warmup=False)

    def test_safe_below_low(self):
        assert classify(self._r(BASE_LOW - 0.01), "medium") == "safe"

    def test_warning_above_high(self):
        assert classify(self._r(BASE_HIGH + 0.01, short_rise=1.0), "medium") == "warning"

    def test_failure_needs_the_escalated_score(self):
        assert classify(self._r(BASE_HIGH * ESCALATING_FACTOR - 0.01, short_rise=1.0), "medium") == "warning"
        assert classify(self._r(BASE_HIGH * ESCALATING_FACTOR + 0.01), "medium") == "failure"

    def test_sensitivity_scales_the_short_window_check_too(self):
        # Obico multiplies the score by the sensitivity; we scale thresholds.
        # Medium in the band, 0.5 vs 0.13 * 3.8 = 0.494: failing. Low (x1.25
        # thresholds, 0.475..0.975): 0.5 vs 0.494 * 1.25 = 0.6175: not.
        r = self._r(0.5, short_rise=0.13)
        assert classify(r, "medium") == "warning"
        assert classify(r, "low") == "safe"

    def test_warmup_is_safe_whatever_the_score(self):
        assert classify(Reading(score=10.0, short_rise=0.0, warmup=True), "high") == "safe"


class TestTriggersAction:
    def test_notify_fires_at_the_alert_level(self):
        assert triggers_action("warning", "notify")
        assert triggers_action("failure", "notify")
        assert not triggers_action("safe", "notify")

    @pytest.mark.parametrize("action", ["pause", "pause_and_off"])
    def test_pausing_waits_for_the_pause_level(self, action):
        assert not triggers_action("warning", action)
        assert triggers_action("failure", action)