Smooth battery percent readings before computing drop-rate steps

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.

server/app/routers/common.py

@@ -59,6 +59,12 @@ MIN_ESTIMATE_SAMPLES = 5  # discharge steps needed before trusting the average
 # single noisy reading needs rejecting at the per-wake-drop level, not
 # just at the recharge-detection level.
 OUTLIER_MODIFIED_Z_THRESHOLD = 3.5
+# Readings considered on each side of a given reading when
+# _smooth_percents looks for local outliers. Needs to be at least half
+# the length of the longest bad-reading burst a noisy divider produces
+# (observed up to ~4 consecutive corrupted reports on one frame) so the
+# good neighbors still outnumber the bad ones in the window.
+BATTERY_SMOOTHING_WINDOW = 4
 
 # "Overdue" threshold multiplier: the device should check in roughly every
 # refresh_interval_s; give it half again as long before flagging it.
@@ -179,6 +185,50 @@ def _reject_outlier_drops(steps: list[tuple[int, float]]) -> list[tuple[int, flo
     return kept or steps  # never filter down to nothing
 
 
+def _smooth_percents(percents: list[int]) -> list[float]:
+    """Replaces any reading that's a wild outlier against its own local
+    neighborhood with that neighborhood's median, before per-wake drop
+    steps are ever built from the series.
+
+    _reject_outlier_drops (above) only catches a bad reading by how much
+    it distorts the *steps* immediately on either side of it -- which is
+    exactly what one isolated glitch does, but a 1M-ohm divider (see
+    firmware/main/battery.c) doesn't always misfire in isolation: several
+    consecutive reports can drift or glitch together (a multi-minute
+    crawl from 68 up into the high 70s with nothing charging, or a run of
+    several ~40 reports spliced into an otherwise flat ~53 run). A step
+    computed *between* two bad readings in the same burst looks like an
+    ordinary small change, not an outlier, so it sails through
+    _reject_outlier_drops untouched.
+
+    A Hampel identifier catches that instead: each reading is compared to
+    the median of its own local window (not the whole series), using the
+    same MAD-based modified z-score as _reject_outlier_drops so this
+    adapts to how noisy a given frame's sensor actually is rather than a
+    fixed percent-point cutoff. A window of BATTERY_SMOOTHING_WINDOW
+    reports on each side tolerates a bad burst up to that long while
+    still being outvoted by the surrounding good readings."""
+    n = len(percents)
+    smoothed = list(percents)
+    for i in range(n):
+        lo = max(0, i - BATTERY_SMOOTHING_WINDOW)
+        hi = min(n, i + BATTERY_SMOOTHING_WINDOW + 1)
+        neighborhood = percents[lo:hi]
+        median = statistics.median(neighborhood)
+        abs_devs = [abs(v - median) for v in neighborhood]
+        # Unlike _reject_outlier_drops, no mean-of-abs-devs fallback here:
+        # a burst can be a big enough share of this small a window that
+        # the mean itself gets dragged up by the very values being
+        # tested, hiding them. A flat 1-percentage-point floor -- this
+        # project's smallest real unit of noise -- keeps the test from
+        # dividing by zero without being skewed by the burst it's
+        # checking.
+        mad = statistics.median(abs_devs) or 1
+        if abs(0.6745 * (percents[i] - median) / mad) > OUTLIER_MODIFIED_Z_THRESHOLD:
+            smoothed[i] = median
+    return smoothed
+
+
 def battery_estimate_s(frame: Frame, db: Session) -> int | None:
     """Remaining-time estimate from a recency-weighted average of the
     *per-wake* percent drop, over the last BATTERY_ESTIMATE_SAMPLE_COUNT
@@ -189,18 +239,21 @@ def battery_estimate_s(frame: Frame, db: Session) -> int | None:
 
     Consecutive reports are assumed to be consecutive wakes (firmware
     reports battery on every wake while on battery), so each step's
-    (prev_percent - next_percent) is that wake's cost. A step where
-    percent went *up* is a recharge, not negative drain, and is skipped
-    entirely rather than folded in as a weird outlier; a flat step
-    (0% change) still counts as a real, cheap wake -- excluding those
-    would systematically overstate the per-wake cost by only counting
-    the wakes that happened to tick the percentage down. The remaining
-    steps then get one more pass, _reject_outlier_drops, to catch the
-    single-noisy-reading case that "percent went up" alone can't (see
-    that function's docstring). Steps are weighted linearly by recency
-    (step i of n gets weight i, 1-indexed) so a recent change in usage
-    pattern shows up quickly instead of being washed out by a long flat
-    history.
+    (prev_percent - next_percent) is that wake's cost. Raw percents go
+    through _smooth_percents first, which corrects readings (including
+    short bursts of them) that are wild outliers against their own local
+    neighborhood -- see that function's docstring for why that catches
+    noise shapes _reject_outlier_drops can't. A step where percent went
+    *up* is a recharge, not negative drain, and is skipped entirely
+    rather than folded in as a weird outlier; a flat step (0% change)
+    still counts as a real, cheap wake -- excluding those would
+    systematically overstate the per-wake cost by only counting the
+    wakes that happened to tick the percentage down. The remaining steps
+    then get one more pass, _reject_outlier_drops, to catch whatever
+    single-noisy-reading shape survives smoothing (see that function's
+    docstring). Steps are weighted linearly by recency (step i of n gets
+    weight i, 1-indexed) so a recent change in usage pattern shows up
+    quickly instead of being washed out by a long flat history.
 
     The resulting %/wake rate is then converted to wall-clock time using
     the frame's *current* refresh_interval_s and quiet-hours settings
@@ -220,6 +273,7 @@ def battery_estimate_s(frame: Frame, db: Session) -> int | None:
     if len(rows) < MIN_ESTIMATE_SAMPLES + 1:
         return None
     percents = list(reversed(rows))  # chronological order
+    percents = _smooth_percents(percents)
 
     steps: list[tuple[int, float]] = []  # (recency_weight, drop_pct)
     for i in range(1, len(percents)):

server/tests/test_battery_estimate.py

@@ -1,11 +1,10 @@
-"""_reject_outlier_drops -- the outlier-rejection pass in the battery
-remaining-time estimate (see routers/common.py's battery_estimate_s).
-Pure function, no DB/HTTP -- (recency_weight, drop_pct) pairs in,
-filtered pairs out."""
+"""_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
+from app.routers.common import _reject_outlier_drops, _smooth_percents
 
 
 def _steps(drops: list[float]) -> list[tuple[int, float]]:
@@ -69,3 +68,37 @@ def test_never_filters_down_to_nothing():
     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