tfaour
3fdda096a9
A 1M-ohm divider (way over the ~10k source impedance the ESP32 ADC's sample-and-hold expects) doesn't always misfire in isolation -- short bursts of a few consecutive bad readings, and multi-reading drifts, both slip past the existing step-level MAD outlier rejection since the steps between two bad readings in the same burst look ordinary. Add a Hampel-filter smoothing pass (local-neighborhood MAD, same statistical approach as the existing outlier rejection) ahead of it.
105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
"""_reject_outlier_drops and _smooth_percents -- the two outlier-rejection
|
|
passes in the battery remaining-time estimate (see routers/common.py's
|
|
battery_estimate_s). Pure functions, no DB/HTTP."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.routers.common import _reject_outlier_drops, _smooth_percents
|
|
|
|
|
|
def _steps(drops: list[float]) -> list[tuple[int, float]]:
|
|
return [(i + 1, d) for i, d in enumerate(drops)]
|
|
|
|
|
|
def _avg(steps: list[tuple[int, float]]) -> float:
|
|
total_weight = sum(w for w, _ in steps)
|
|
return sum(w * d for w, d in steps) / total_weight
|
|
|
|
|
|
def test_no_outlier_keeps_every_step():
|
|
steps = _steps([1, 1, 2, 1, 1, 2, 1])
|
|
assert _reject_outlier_drops(steps) == steps
|
|
|
|
|
|
def test_single_glitch_dip_is_rejected():
|
|
"""The exact shape reported in production: 18 ordinary 1%-per-wake
|
|
steps and one spliced-in 26% glitch -- the naive median-based MAD
|
|
degenerates to 0 here (more than half the steps tie at the median),
|
|
which used to let the glitch sail straight through untouched."""
|
|
normal = [1] * 18
|
|
glitchy = normal[:9] + [26] + normal[9:]
|
|
|
|
kept = _reject_outlier_drops(_steps(glitchy))
|
|
kept_drops = [d for _, d in kept]
|
|
assert 26 not in kept_drops
|
|
assert len(kept) == 18
|
|
|
|
# the whole point: the estimate should come out the same as if the
|
|
# glitch had never been recorded at all
|
|
baseline_avg = _avg(_steps(normal))
|
|
filtered_avg = _avg(kept)
|
|
assert abs(filtered_avg - baseline_avg) < 1e-9
|
|
|
|
|
|
def test_single_glitch_spike_is_rejected():
|
|
normal = [2] * 18
|
|
glitchy = normal[:5] + [40] + normal[5:]
|
|
|
|
kept = _reject_outlier_drops(_steps(glitchy))
|
|
kept_drops = [d for _, d in kept]
|
|
assert 40 not in kept_drops
|
|
assert len(kept) == 18
|
|
|
|
|
|
def test_identical_steps_reject_nothing():
|
|
"""Every step tied at the exact same value -- both the median MAD
|
|
and the mean-absolute-deviation fallback are 0 here, which is the
|
|
one case _reject_outlier_drops explicitly bails out of rather than
|
|
filtering down to nothing."""
|
|
steps = _steps([1] * 10)
|
|
assert _reject_outlier_drops(steps) == steps
|
|
|
|
|
|
def test_never_filters_down_to_nothing():
|
|
"""Even a genuinely bimodal series (half the wakes cheap, half
|
|
expensive -- not a single-glitch shape at all) shouldn't empty the
|
|
list; a battery_estimate_s caller treats an empty result as
|
|
"insufficient data," which a merely-noisy history isn't."""
|
|
steps = _steps([1, 1, 1, 1, 10, 10, 10, 10])
|
|
kept = _reject_outlier_drops(steps)
|
|
assert len(kept) > 0
|
|
|
|
|
|
def test_smooth_corrects_isolated_spike():
|
|
percents = [70, 70, 70, 70, 70, 90, 70, 70, 70, 70, 70]
|
|
smoothed = _smooth_percents(percents)
|
|
assert smoothed[5] == 70
|
|
assert smoothed[:5] == percents[:5]
|
|
assert smoothed[6:] == percents[6:]
|
|
|
|
|
|
def test_smooth_corrects_short_burst():
|
|
"""The shape seen in production: several consecutive corrupted
|
|
reports (a 1M-ohm divider glitching for a few reports in a row, not
|
|
just one) spliced into an otherwise flat run. A step computed
|
|
between two of these looks like an ordinary small change, which is
|
|
exactly why _reject_outlier_drops alone can't catch this shape."""
|
|
percents = [53, 53, 53, 53, 41, 40, 40, 42, 53, 53, 53, 53]
|
|
smoothed = _smooth_percents(percents)
|
|
assert smoothed[4:8] == [53, 53, 53, 53]
|
|
assert smoothed[:4] == percents[:4]
|
|
assert smoothed[8:] == percents[8:]
|
|
|
|
|
|
def test_smooth_leaves_gradual_legitimate_trend_alone():
|
|
"""A slow, steady climb (recharge) or decline spread over many
|
|
reports is a real trend, not a local glitch -- each reading is close
|
|
to its own neighborhood's median, so nothing should be flagged."""
|
|
percents = list(range(80, 60, -2)) # 80, 78, 76, ... steady discharge
|
|
assert _smooth_percents(percents) == percents
|
|
|
|
|
|
def test_smooth_identical_readings_untouched():
|
|
percents = [50] * 12
|
|
assert _smooth_percents(percents) == percents
|