From 3fdda096a93a93c7d938974f5c17df2756f7acef Mon Sep 17 00:00:00 2001 From: Thomas Faour Date: Tue, 28 Jul 2026 02:09:12 +0000 Subject: [PATCH] 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 | 78 ++++++++++++++++++++++----- server/tests/test_battery_estimate.py | 43 +++++++++++++-- 2 files changed, 104 insertions(+), 17 deletions(-) diff --git a/server/app/routers/common.py b/server/app/routers/common.py index 3b3450d..b82c203 100644 --- a/server/app/routers/common.py +++ b/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)): diff --git a/server/tests/test_battery_estimate.py b/server/tests/test_battery_estimate.py index 3781944..c7f76f8 100644 --- a/server/tests/test_battery_estimate.py +++ b/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