Battery: a frame a second, bought where the tick won't give it
The sweep read a frame per idle repaint, so it climbed at half speed on the handheld — a platform that ticks every 2 s because a frame costs 7-11% of a core there. Pacing it off the tick was solving the wrong problem: the gauge should climb at a second a frame everywhere, and the repaints that takes are worth buying, since a pack only sweeps while it is plugged in. So the poller now buys them, and only while all three hold: the gauge is actually sweeping (a pack sitting full on the charger is not — which is where a forgotten handheld spends the night), the platform idles slower than a frame, and a surface is listening. Everything else sleeps out the interval as before. BATTERY_ANIM_S moves to the widget as the one cadence both read, and Ui.invalidate gives a service its first channel to repaint live chrome. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
2731af12cb
commit
385654ab5f
6 files changed
+147
-35
No files matched your search
@@ -189,7 +189,7 @@ class BatteryService:
|
||||
await self.stop()
|
||||
|
||||
async def _run(self) -> None:
|
||||
"""Read once at once, then tick forever: a best-effort pass, then sleep."""
|
||||
"""Read once at once, then tick forever: a best-effort pass, then sleep it out."""
|
||||
while True:
|
||||
try:
|
||||
await self._poll()
|
||||
@@ -197,7 +197,38 @@ class BatteryService:
|
||||
raise
|
||||
except Exception as exc: # noqa: BLE001 - a bad read must not kill the loop
|
||||
self._ctx.log.debug("battery poller: read failed: %s", exc)
|
||||
await self._idle()
|
||||
|
||||
async def _idle(self) -> None:
|
||||
"""Sleep out the poll interval — stepping the header's charging sweep through it.
|
||||
|
||||
The gauge says *charging* by sweeping its fill, one frame per
|
||||
:data:`~meshterm.ui.widgets.BATTERY_ANIM_S`, and a frame only advances when something
|
||||
repaints. The desktop's own idle tick is already that fast, but a platform where a
|
||||
frame is expensive idles slower (``Platform.tick_s``: the PicoCalc's 2 s), and at that
|
||||
rate the sweep would show every other frame — a cell jumping two rows at a time rather
|
||||
than climbing. So the poller buys the missing repaints itself, and only where all
|
||||
three things are true: the gauge is actually sweeping (:func:`~meshterm.ui.widgets.
|
||||
battery_sweeps` — a pack left full on the charger is *not*, which is where a forgotten
|
||||
handheld spends the night), the platform idles slower than a frame, and there is a
|
||||
surface listening. Every other case is one plain sleep, exactly as before.
|
||||
"""
|
||||
from ..ui.widgets import BATTERY_ANIM_S, battery_sweeps # deferred: services < ui
|
||||
|
||||
reading = self._reading
|
||||
sweeping = reading is not None and battery_sweeps(reading.percent, reading.charging)
|
||||
if not sweeping or get_platform().tick_s <= BATTERY_ANIM_S:
|
||||
await asyncio.sleep(POLL_S)
|
||||
return
|
||||
deadline = time.monotonic() + POLL_S
|
||||
while (remaining := deadline - time.monotonic()) > 0:
|
||||
await asyncio.sleep(min(BATTERY_ANIM_S, remaining))
|
||||
try:
|
||||
self._ctx.ui.invalidate()
|
||||
except Exception as exc: # noqa: BLE001 - a dead surface must not kill the loop
|
||||
self._ctx.log.debug("battery poller: repaint failed: %s", exc)
|
||||
await asyncio.sleep(max(0.0, deadline - time.monotonic()))
|
||||
return
|
||||
|
||||
async def _poll(self) -> None:
|
||||
"""One pass: read the pack, estimate charge, and re-derive the charging trend.
|
||||
|
||||
+7
-17
@@ -31,7 +31,7 @@ from .surface import TuiUi
|
||||
from .braillechart import activity_peak, activity_sparkline
|
||||
from .menus import SEP_COMPACT, SEP_ROOMY, command_icon, command_label, section_heading
|
||||
from .theme import make_console
|
||||
from .widgets import battery_cell
|
||||
from .widgets import BATTERY_ANIM_S, battery_cell
|
||||
from .tui import (
|
||||
CANCEL,
|
||||
Choice,
|
||||
@@ -213,15 +213,6 @@ def _header(ctx: AppContext, cache: dict, width: int) -> Text:
|
||||
return header
|
||||
|
||||
|
||||
#: Floor on the seconds per animation step of the header battery gauge (the charging sweep
|
||||
#: and the low-battery blink). The step is the platform's own idle repaint interval
|
||||
#: (``Platform.tick_s``, the app's ``refresh_interval``) so each repaint advances the
|
||||
#: animation by one clean step rather than aliasing across skipped frames — 1 s on the
|
||||
#: desktop, one every 2 s on the PicoCalc's slower tick, where a full sweep therefore takes
|
||||
#: :data:`~meshterm.ui.widgets._CHARGE_FRAMES` × 2 s to climb.
|
||||
_BATTERY_ANIM_MIN_S = 1.0
|
||||
|
||||
|
||||
def _battery_segment(ctx: AppContext) -> Text:
|
||||
"""The header's right-anchored battery gauge, or an empty Text when there is no pack.
|
||||
|
||||
@@ -240,17 +231,16 @@ def _battery_segment(ctx: AppContext) -> Text:
|
||||
reading = ctx.battery.reading()
|
||||
if reading is None:
|
||||
return Text()
|
||||
platform = get_platform()
|
||||
# The sweep is how the gauge says "charging" at all, so its clock runs on every platform;
|
||||
# only the step widens to whatever that platform actually repaints at. The low-battery
|
||||
# blink is decoration over a charge the cell already shows, so it stays behind `effects`
|
||||
# — not worth a forced repaint, nor a colour swap on a 16-slot console.
|
||||
frame = int(time.monotonic() / max(_BATTERY_ANIM_MIN_S, platform.tick_s))
|
||||
# The sweep is how the gauge says "charging" at all, so its clock runs at the same second
|
||||
# a step on every platform — the poller buys a slow-ticking one the repaints to show it.
|
||||
# The low-battery blink is decoration over a charge the cell already shows, so it stays
|
||||
# behind `effects`: not worth a forced repaint, nor a colour swap on a 16-slot console.
|
||||
frame = int(time.monotonic() / BATTERY_ANIM_S)
|
||||
return battery_cell(
|
||||
reading.percent,
|
||||
charging=reading.charging,
|
||||
frame=frame,
|
||||
animate=platform.effects,
|
||||
animate=get_platform().effects,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -105,6 +105,16 @@ class Ui:
|
||||
def discard(self) -> None:
|
||||
"""Drop any buffered-but-unshown output (a no-op when output is immediate)."""
|
||||
|
||||
def invalidate(self) -> None:
|
||||
"""Ask for a repaint (a no-op where there is no persistent frame to repaint).
|
||||
|
||||
The one channel a background service has to move the *chrome* — a header the user
|
||||
is looking at whose data changed on nobody's keystroke. Screens invalidate their own
|
||||
session directly; this exists for the services, which hold no session (the battery
|
||||
poller's charging sweep, at :data:`~meshterm.ui.widgets.BATTERY_ANIM_S` a frame, is
|
||||
the first). Cheap and safe to call when nothing is listening.
|
||||
"""
|
||||
|
||||
def progress(self, title: str = "Working"): # noqa: ANN201 - context manager, varies by backend
|
||||
"""Return a progress context manager exposing ``add_task``/``advance``/``update``."""
|
||||
raise NotImplementedError
|
||||
@@ -590,6 +600,10 @@ class TuiUi(Ui):
|
||||
self.session = session
|
||||
self._buffer: list[RenderableType] = []
|
||||
|
||||
def invalidate(self) -> None:
|
||||
"""Repaint the live frame, so a service's change to the chrome actually shows."""
|
||||
self.session.invalidate()
|
||||
|
||||
# --- output --------------------------------------------------------------
|
||||
|
||||
def show(self, *renderables: RenderableType) -> None:
|
||||
|
||||
+32
-5
@@ -1361,6 +1361,32 @@ _BATTERY_CRITICAL = 10
|
||||
#: Frames in the charging sweep: empty → four fills → round again (bottom-to-full loop).
|
||||
_CHARGE_FRAMES = 5
|
||||
|
||||
#: Seconds per animation step of the gauge (the charging sweep and the low-battery blink) —
|
||||
#: a second a step, so a full sweep climbs in :data:`_CHARGE_FRAMES` seconds. THE cadence for
|
||||
#: the gauge: the header derives its frame from it, and the battery poller paces the repaints
|
||||
#: a sweep needs by it (see :meth:`~meshterm.services.battery_service.BatteryService._idle`),
|
||||
#: since a platform idling slower than this would otherwise show every other frame.
|
||||
BATTERY_ANIM_S = 1.0
|
||||
|
||||
|
||||
def battery_sweeps(percent: int, charging: bool) -> bool:
|
||||
"""Whether the gauge animates for this reading — i.e. whether the charging sweep runs.
|
||||
|
||||
The rule the sweep is drawn by, exposed because the cost of animating is paid somewhere
|
||||
else: the poller reads this to decide whether the header is worth repainting faster than
|
||||
the platform's idle tick. A pack at 100% doesn't sweep (see :func:`battery_cell`), and a
|
||||
device left on the charger overnight sits exactly there, so asking first is the difference
|
||||
between an idle handheld and one repainting all night for a cell that never moves.
|
||||
|
||||
Args:
|
||||
percent: State of charge, 0–100.
|
||||
charging: Whether the pack is taking charge.
|
||||
|
||||
Returns:
|
||||
Whether the gauge's fill moves between frames.
|
||||
"""
|
||||
return charging and int(percent) < 100
|
||||
|
||||
|
||||
def _braille_fill(rows: int) -> str:
|
||||
"""A single braille cell with its bottom ``rows`` (0–4) dot rows lit."""
|
||||
@@ -1396,10 +1422,11 @@ def battery_cell(
|
||||
dim slate on alternate frames — an unmissable pulse in the corner.
|
||||
|
||||
The two differ in what they're *for*, which is what ``animate`` sorts out. The sweep is
|
||||
the only thing that says charging at all, so it runs everywhere — a platform whose frames
|
||||
are dear (the PicoCalc: see ``Platform.effects``) just steps it on that platform's own
|
||||
slower repaint cadence. The blink is an alarm laid over a charge the gauge already shows,
|
||||
so it is genuinely decorative and ``animate=False`` drops it, leaving the red cell steady.
|
||||
the only thing that says charging at all, so it runs everywhere, at :data:`BATTERY_ANIM_S`
|
||||
a step — a platform whose frames are dear (the PicoCalc: see ``Platform.effects``) buys
|
||||
the repaints for it while a sweep is live rather than going without. The blink is an alarm
|
||||
laid over a charge the gauge already shows, so it is genuinely decorative and
|
||||
``animate=False`` drops it, leaving the red cell steady.
|
||||
|
||||
A pack at 100% is drawn as **not charging** whichever way the flag reads: a full cell
|
||||
resting full is the honest picture, and a topped-off charger left plugged in shouldn't
|
||||
@@ -1418,7 +1445,7 @@ def battery_cell(
|
||||
"""
|
||||
pct = max(0, min(100, int(percent)))
|
||||
rows, color = _battery_tier(pct)
|
||||
if charging and pct < 100:
|
||||
if battery_sweeps(pct, charging):
|
||||
# The charge colour is held; only the fill sweeps, so it reads as "filling", not
|
||||
# as the charge itself jumping around.
|
||||
rows = frame % _CHARGE_FRAMES
|
||||
|
||||
+53
-1
@@ -10,16 +10,18 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
from unittest import mock
|
||||
|
||||
from meshterm.core.connection import charging_from_battery_level_status
|
||||
from meshterm.services import battery_service
|
||||
from meshterm.services.battery_service import (
|
||||
_BATTERY_PRESENT_FLOOR_MV,
|
||||
POLL_S,
|
||||
BatteryReading,
|
||||
BatteryService,
|
||||
battery_percent,
|
||||
)
|
||||
from meshterm.ui.widgets import _BATTERY_CRITICAL, battery_cell
|
||||
from meshterm.ui.widgets import _BATTERY_CRITICAL, BATTERY_ANIM_S, battery_cell
|
||||
|
||||
# --- the LiPo state-of-charge curve -------------------------------------------------------
|
||||
|
||||
@@ -204,6 +206,56 @@ def test_battery_poller_charging_ignores_load_sag_and_flat_or_falling_packs() ->
|
||||
assert svc._charging(now) is False
|
||||
|
||||
|
||||
# --- the repaints a live sweep needs -------------------------------------------------------
|
||||
|
||||
|
||||
def _idle_repaints(svc: BatteryService, tick_s: float) -> int:
|
||||
"""Run one poll interval on a platform ticking every ``tick_s``, counting repaints."""
|
||||
painted = 0
|
||||
|
||||
def invalidate() -> None:
|
||||
nonlocal painted
|
||||
painted += 1
|
||||
|
||||
svc._ctx.ui = SimpleNamespace(invalidate=invalidate)
|
||||
slept = 0.0
|
||||
|
||||
async def sleep(seconds: float) -> None:
|
||||
"""Advance a fake clock instead of actually waiting."""
|
||||
nonlocal slept
|
||||
slept += seconds
|
||||
|
||||
with_platform = SimpleNamespace(tick_s=tick_s)
|
||||
with mock.patch.object(battery_service.asyncio, "sleep", sleep), mock.patch.object(
|
||||
battery_service.time, "monotonic", lambda: slept
|
||||
), mock.patch.object(battery_service, "get_platform", lambda: with_platform):
|
||||
asyncio.run(svc._idle())
|
||||
assert slept >= POLL_S # however it waits, it waits the whole poll interval
|
||||
return painted
|
||||
|
||||
|
||||
def test_a_live_sweep_buys_the_repaints_a_slow_platform_wont_give_it() -> None:
|
||||
"""Where the idle tick is slower than a frame, the poller paces the sweep itself."""
|
||||
svc = _service([])
|
||||
svc._reading = BatteryReading(millivolts=3800, percent=40, charging=True)
|
||||
# A 2 s idle tick would show every other frame, so the poller fills in a frame a second.
|
||||
assert _idle_repaints(svc, tick_s=2.0) == POLL_S / BATTERY_ANIM_S
|
||||
# A platform already ticking that fast needs nothing bought for it.
|
||||
assert _idle_repaints(svc, tick_s=BATTERY_ANIM_S) == 0
|
||||
|
||||
|
||||
def test_a_gauge_that_isnt_sweeping_costs_nothing() -> None:
|
||||
"""No sweep, no repaints — a pack left full on the charger doesn't paint all night."""
|
||||
svc = _service([])
|
||||
for reading in (
|
||||
None,
|
||||
BatteryReading(millivolts=3800, percent=40, charging=False), # discharging
|
||||
BatteryReading(millivolts=4200, percent=100, charging=True), # full: draws at rest
|
||||
):
|
||||
svc._reading = reading
|
||||
assert _idle_repaints(svc, tick_s=2.0) == 0
|
||||
|
||||
|
||||
# --- the host pack (the PicoCalc's own power_supply) ---------------------------------------
|
||||
|
||||
|
||||
|
||||
@@ -194,11 +194,12 @@ def test_an_explicit_spinner_cycle_still_wins() -> None:
|
||||
assert Spinner("ab").frames == "ab"
|
||||
|
||||
|
||||
def test_battery_gauge_steps_the_sweep_at_the_platforms_own_cadence(monkeypatch) -> None:
|
||||
"""The charging sweep runs on the PicoCalc too — a step per repaint, not per second."""
|
||||
def test_battery_gauge_sweeps_a_frame_a_second_on_both_platforms(monkeypatch) -> None:
|
||||
"""One second a frame everywhere — the PicoCalc buys the repaints rather than slowing."""
|
||||
from meshterm.ui import menu
|
||||
from meshterm.services.battery_service import BatteryReading
|
||||
from meshterm.ui.menu import _battery_segment
|
||||
from meshterm.ui.widgets import BATTERY_ANIM_S
|
||||
|
||||
class _Ctx:
|
||||
class battery: # noqa: N801 - a stand-in for the service, not a real class name
|
||||
@@ -210,21 +211,18 @@ def test_battery_gauge_steps_the_sweep_at_the_platforms_own_cadence(monkeypatch)
|
||||
clock = 0.0
|
||||
monkeypatch.setattr(menu.time, "monotonic", lambda: clock)
|
||||
|
||||
def _sweep(seconds: float) -> list[str]:
|
||||
"""The gauge's fills over five ticks of ``seconds`` each."""
|
||||
def _sweep() -> list[str]:
|
||||
"""The gauge's fills over five one-second steps."""
|
||||
nonlocal clock
|
||||
out = []
|
||||
for step in range(5):
|
||||
clock = step * seconds
|
||||
clock = step * BATTERY_ANIM_S
|
||||
out.append(_battery_segment(ctx).plain[0]) # type: ignore[arg-type]
|
||||
return out
|
||||
|
||||
set_platform(PICOCALC)
|
||||
# A step per 2 s idle repaint: the sweep climbs, and does not alias across skipped frames.
|
||||
assert len(set(_sweep(PICOCALC.tick_s))) == 5
|
||||
assert len(set(_sweep(PICOCALC.tick_s / 2))) < 5 # half a tick apart, some frames repeat
|
||||
set_platform(REGULAR)
|
||||
assert len(set(_sweep(REGULAR.tick_s))) == 5
|
||||
for platform in (PICOCALC, REGULAR):
|
||||
set_platform(platform)
|
||||
assert len(set(_sweep())) == 5 # a distinct frame each second, no held frames
|
||||
assert _battery_segment(ctx).plain.endswith("5%") # type: ignore[arg-type]
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user