Battery: read the charging trend robustly so noise stops faking it

The firmware exposes no charging flag, so charging is inferred from the
terminal-voltage trend. The old test compared the first and last samples
in the window against a 15 mV threshold — but a single-cell LiPo sags
tens of millivolts under each LoRa transmission and springs back when the
radio idles, so a lone sagged sample at either endpoint read as a
+tens-of-mV "rise" and the gauge falsely showed charging on a discharging
or resting pack.

Read the trend from the median of the window's older half against the
median of its newer half (discarding the transient sag/recovery spikes)
and require a large, sustained rise, with hysteresis: hard to start
(25 mV), held while still clearly climbing (8 mV), and dropped the moment
the pack goes flat — topped off or unplugged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

meshterm/services/battery_service.py

@@ -14,9 +14,13 @@ Two device facts shape what it can report:
   discharge curve (:func:`battery_percent`). Devices with no battery gauge answer with no
   usable level; those are reported as *absent* so the header shows nothing for them.
 * Firmware exposes **no charging flag at all**. Charging is therefore *inferred* from the
-  terminal voltage trending upward across the recent sample window — the only dynamic
-  signal available. It is an estimate, held steady through the flat stretches (a charger's
-  constant-voltage phase, or an idle pack) so the gauge doesn't flap.
+  terminal voltage — but a single-cell LiPo sags tens of millivolts under each LoRa
+  transmission and springs back when the radio idles, so the raw sample trend is mostly load
+  noise, not charge. The inference reads it robustly (the median of the sample window's older
+  half against its newer half, which discards those transient spikes) and calls charge only
+  on a large, *sustained* rise, dropping straight back the moment the pack goes flat. It is
+  still an estimate — the best the firmware allows — but one a discharging or resting pack no
+  longer trips.
 """
 
 from __future__ import annotations
@@ -25,6 +29,7 @@ import asyncio
 import time
 from collections import deque
 from dataclasses import dataclass
+from statistics import median
 from typing import TYPE_CHECKING, Optional
 
 if TYPE_CHECKING:
@@ -40,12 +45,22 @@ POLL_S = 20.0
 #: single cell is treated as *no battery present* and the gauge is hidden.
 _BATTERY_PRESENT_FLOOR_MV = 1500
 
-#: Recent-samples window (seconds) the charging trend is measured over, and the rise across
-#: it that flips the estimate to *charging* (or, as a fall, to *not charging*). Between the
-#: two the last verdict is held, so a flat voltage — charger in constant-voltage phase, or a
-#: resting pack — doesn't make the gauge flicker between the two states.
+#: Recent-samples window (seconds) the charging trend is measured over.
 _TREND_WINDOW_S = 180.0
-_CHARGING_RISE_MV = 15
+
+#: Fewest samples the trend needs before it will call charging either way — about two minutes
+#: at :data:`POLL_S`, enough to span the window and give each half a meaningful median. Below
+#: this the verdict is simply *not charging*.
+_TREND_MIN_SAMPLES = 6
+
+#: Smoothed rise (millivolts: the newer half's median terminal voltage minus the older half's)
+#: that *starts* a charging verdict, and the smaller rise it must stay above to *hold* one.
+#: The start threshold clears the median-smoothed load-noise floor by a wide margin, so a
+#: discharging or resting pack never trips it; the hold threshold lets a genuine, still-climbing
+#: charge ride through the poll jitter without flicker, while a pack gone flat — topped off, or
+#: unplugged — falls below it and drops straight back to *not charging*.
+_CHARGING_RISE_ON_MV = 25
+_CHARGING_RISE_HOLD_MV = 8
 
 #: How many samples to retain — the trend window plus a little slack at the poll cadence.
 _TREND_SAMPLES = int(_TREND_WINDOW_S / POLL_S) + 2
@@ -94,8 +109,8 @@ class BatteryReading:
     Attributes:
         millivolts: The pack's terminal voltage, as the firmware reported it.
         percent: The state-of-charge estimate (see :func:`battery_percent`).
-        charging: Whether the pack appears to be taking charge (inferred from a rising
-            terminal voltage — the firmware exposes no charging flag).
+        charging: Whether the pack appears to be taking charge (inferred from a sustained
+            rise in terminal voltage — the firmware exposes no charging flag).
     """
 
     millivolts: int
@@ -191,10 +206,16 @@ class BatteryService:
     def _charging(self, now: float) -> bool:
         """Infer whether the pack is charging from its recent terminal-voltage trend.
 
-        The firmware exposes no charging flag, so this is the best available signal: a
-        clear rise across the window means charge is going in, a clear fall means it is
-        coming out, and a flat stretch holds the previous verdict (so a charger's
-        constant-voltage phase or a resting pack doesn't make the gauge flap).
+        The firmware exposes no charging flag, so the trend is the only signal — but a
+        single-cell LiPo's terminal voltage sags tens of millivolts under each transmission
+        and springs back when the radio idles, so the raw trend is mostly load noise.
+        Charging is read *robustly*: the median voltage of the window's older half is compared
+        against the median of its newer half, which discards those transient sag/recovery
+        spikes, and only a large, *sustained* rise counts as charge going in. The verdict
+        carries hysteresis — hard to start (:data:`_CHARGING_RISE_ON_MV`), then held while the
+        pack is still clearly climbing (:data:`_CHARGING_RISE_HOLD_MV`) — so a genuine charge
+        doesn't flicker, while a pack gone flat (topped off, or unplugged) drops straight back
+        to *not charging*.
 
         Args:
             now: The current monotonic time (the just-taken sample's timestamp).
@@ -202,13 +223,10 @@ class BatteryService:
         Returns:
             The charging estimate for this reading.
         """
-        last = self._reading.charging if self._reading is not None else False
-        recent = [(t, mv) for t, mv in self._history if now - t <= _TREND_WINDOW_S]
-        if len(recent) < 2:
-            return last
-        rise = recent[-1][1] - recent[0][1]
-        if rise >= _CHARGING_RISE_MV:
-            return True
-        if rise <= -_CHARGING_RISE_MV:
+        charging = self._reading.charging if self._reading is not None else False
+        recent = [mv for t, mv in self._history if now - t <= _TREND_WINDOW_S]
+        if len(recent) < _TREND_MIN_SAMPLES:
             return False
-        return last
+        mid = len(recent) // 2
+        rise = median(recent[mid:]) - median(recent[:mid])
+        return rise >= (_CHARGING_RISE_HOLD_MV if charging else _CHARGING_RISE_ON_MV)

tests/test_battery.py

@@ -12,7 +12,7 @@ from types import SimpleNamespace
 
 from meshterm.services.battery_service import (
     _BATTERY_PRESENT_FLOOR_MV,
-    _CHARGING_RISE_MV,
+    POLL_S,
     BatteryService,
     battery_percent,
 )
@@ -122,19 +122,56 @@ def test_battery_poller_hides_a_device_with_no_pack() -> None:
     assert svc.reading() is None
 
 
-def test_battery_poller_infers_charging_from_the_voltage_trend() -> None:
-    """A clear rise reads as charging, a clear fall as not, a flat stretch holds the verdict."""
+def _history_of(now: float, mvs: list[int]) -> list[tuple[float, int]]:
+    """``(time, mv)`` samples one poll apart, oldest first, the last landing at ``now``."""
+    n = len(mvs)
+    return [(now - (n - 1 - i) * POLL_S, mv) for i, mv in enumerate(mvs)]
+
+
+def test_battery_poller_calls_charging_only_on_a_strong_sustained_rise() -> None:
+    """A large, sustained climb reads as charging; a gentle one only holds a verdict already set."""
     svc = _service([])
     now = 1000.0
-    svc._history.extend([(now - 120, 3700), (now, 3700 + _CHARGING_RISE_MV)])
-    assert svc._charging(now) is True  # rising → charging
 
+    # A clear, sustained rise from a cold (not-charging) start crosses the ON threshold.
+    svc._history.extend(_history_of(now, [3700, 3712, 3724, 3736, 3748, 3760, 3772, 3784]))
+    assert svc._charging(now) is True
+
+    # A gentle climb (over the hold floor, under the start floor) will not *start* a verdict…
+    svc._reading = SimpleNamespace(charging=False)
+    svc._history.clear()
+    svc._history.extend(_history_of(now, [3900, 3902, 3904, 3915, 3917, 3919]))
+    assert svc._charging(now) is False
+    # …but it *holds* one already in flight, so a real charge doesn't flicker on poll jitter.
+    svc._reading = SimpleNamespace(charging=True)
+    assert svc._charging(now) is True
+
+
+def test_battery_poller_charging_ignores_load_sag_and_flat_or_falling_packs() -> None:
+    """A transient TX sag, a flat pack, and a falling pack all read as *not charging*."""
+    svc = _service([])
+    now = 1000.0
+
+    # A single deep TX sag at the window's start would fool a raw first-to-last diff into
+    # seeing a +60 mV "rise"; the median of each half discards the spike, so it does not.
+    svc._reading = SimpleNamespace(charging=False)
+    svc._history.extend(_history_of(now, [3740, 3802, 3798, 3801, 3800, 3799, 3802, 3800]))
+    assert svc._charging(now) is False
+
+    # A pack held flat (topped off, or unplugged) drops a charging verdict back off.
     svc._reading = SimpleNamespace(charging=True)
     svc._history.clear()
-    svc._history.extend([(now - 120, 3900), (now, 3900 - _CHARGING_RISE_MV)])
-    assert svc._charging(now) is False  # falling → not charging
+    svc._history.extend(_history_of(now, [3800, 3801, 3800, 3799, 3800, 3801]))
+    assert svc._charging(now) is False
 
+    # A clear discharge never reads as charging.
     svc._reading = SimpleNamespace(charging=True)
     svc._history.clear()
-    svc._history.extend([(now - 60, 3800), (now, 3803)])
-    assert svc._charging(now) is True  # flat → holds the last verdict
+    svc._history.extend(_history_of(now, [3900, 3890, 3880, 3870, 3860, 3850]))
+    assert svc._charging(now) is False
+
+    # Too little history to span the window is *not charging*, whatever the last verdict.
+    svc._reading = SimpleNamespace(charging=True)
+    svc._history.clear()
+    svc._history.extend(_history_of(now, [3800, 3900]))
+    assert svc._charging(now) is False