diff --git a/meshterm/context.py b/meshterm/context.py index fb2a63c..c7c09ac 100644 --- a/meshterm/context.py +++ b/meshterm/context.py @@ -32,6 +32,7 @@ if TYPE_CHECKING: from .services.courier import CourierService from .services.monitor_service import MonitorService from .services.watchtower import WatchtowerService + from .services.battery_service import BatteryService from .ui.surface import Ui @@ -103,6 +104,7 @@ class AppContext: _adverts: "Optional[AdvertScheduler]" = field(default=None, init=False, repr=False) _watchtower: "Optional[WatchtowerService]" = field(default=None, init=False, repr=False) _courier: "Optional[CourierService]" = field(default=None, init=False, repr=False) + _battery: "Optional[BatteryService]" = field(default=None, init=False, repr=False) _ui: "Optional[Ui]" = field(default=None, init=False, repr=False) def __post_init__(self) -> None: @@ -285,6 +287,20 @@ class AppContext: self._courier = CourierService(self) return self._courier + @property + def battery(self) -> "BatteryService": + """Return the session's battery poller, creating it on first use. + + Created idle here; the interactive session starts it alongside the other + always-on services. It only ever reads the pack (best-effort, skipping quietly + without a device), so scripted CLI runs simply never start the loop. + """ + if self._battery is None: + from .services.battery_service import BatteryService + + self._battery = BatteryService(self) + return self._battery + @property def log(self): # type: ignore[no-untyped-def] """The application logger.""" @@ -500,6 +516,8 @@ class AppContext: """Stop monitoring and chat, stop the event hub, disconnect, and close the repo.""" if self._courier is not None: await self._courier.aclose() + if self._battery is not None: + await self._battery.aclose() if self._watchtower is not None: await self._watchtower.aclose() if self._adverts is not None: diff --git a/meshterm/services/battery_service.py b/meshterm/services/battery_service.py new file mode 100644 index 0000000..64835b4 --- /dev/null +++ b/meshterm/services/battery_service.py @@ -0,0 +1,214 @@ +"""The battery poller: keep the status bar's fuel gauge current, quietly. + +A session-long task that reads the connected companion's battery voltage on a slow +cadence and caches a small :class:`BatteryReading` for the header to draw (see +:func:`~meshterm.ui.widgets.battery_cell`). Like the advert scheduler and the Watchtower +it holds no device subscription — each pass just checks whether a device is connected and +skips quietly otherwise — so it is indifferent to reconnects, ticking across a link drop +and resuming once the radio is back. + +Two device facts shape what it can report: + +* Firmware exposes the pack as a **terminal voltage in millivolts**, not a percentage, so + the reading is turned into a state-of-charge estimate against a single-cell LiPo + 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. +""" + +from __future__ import annotations + +import asyncio +import time +from collections import deque +from dataclasses import dataclass +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from ..context import AppContext + +#: Seconds between battery reads. A pack drifts slowly, so a coarse tick is plenty; it is +#: also fine enough that the charging trend (measured over :data:`_TREND_WINDOW_S`) has +#: several samples to work with. +POLL_S = 20.0 + +#: Millivolts at or above which a reading counts as a real battery. A device with no fuel +#: gauge answers with an empty frame or a zero level; anything below a badly-depleted +#: 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. +_TREND_WINDOW_S = 180.0 +_CHARGING_RISE_MV = 15 + +#: 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 + +#: Single-cell LiPo terminal-voltage → state-of-charge lookup ``(millivolts, percent)``, +#: high to low. The discharge curve is far from a straight line — most of the usable charge +#: sits in a narrow band around 3.7–3.9 V — so a piecewise table read with linear +#: interpolation tracks it far better than a flat voltage-to-percent line. These are the +#: widely-used community/Battery-University rungs. +_LIPO_SOC: tuple[tuple[int, int], ...] = ( + (4200, 100), (4150, 95), (4110, 90), (4080, 85), (4020, 80), (3980, 75), + (3950, 70), (3910, 65), (3870, 60), (3850, 55), (3840, 50), (3820, 45), + (3800, 40), (3790, 35), (3770, 30), (3750, 25), (3730, 20), (3710, 15), + (3690, 10), (3610, 5), (3270, 0), +) + + +def battery_percent(millivolts: int) -> int: + """Estimate a single-cell LiPo's state of charge (0–100%) from its terminal mV. + + Reads the :data:`_LIPO_SOC` curve, interpolating linearly between the two rungs the + voltage falls between and clamping past either end. + + Args: + millivolts: The pack's terminal voltage in millivolts. + + Returns: + The estimated charge as a whole percent in ``[0, 100]``. + """ + mv = int(millivolts) + if mv >= _LIPO_SOC[0][0]: + return 100 + if mv <= _LIPO_SOC[-1][0]: + return 0 + for (v_hi, p_hi), (v_lo, p_lo) in zip(_LIPO_SOC, _LIPO_SOC[1:], strict=False): + if v_lo <= mv <= v_hi: + frac = (mv - v_lo) / (v_hi - v_lo) + return round(p_lo + (p_hi - p_lo) * frac) + return 0 # pragma: no cover - the clamps above cover the whole range + + +@dataclass(frozen=True) +class BatteryReading: + """One cached battery snapshot for the header to draw. + + 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). + """ + + millivolts: int + percent: int + charging: bool + + +class BatteryService: + """Polls the connected companion's battery and caches a reading for the header. + + Interact through the async lifecycle methods (:meth:`start`, :meth:`stop`, + :meth:`aclose`) and read the latest snapshot with :meth:`reading`; everything else + happens on the loop's own schedule. + """ + + def __init__(self, ctx: "AppContext") -> None: + """Initialize an idle poller bound to an application context. + + Args: + ctx: The shared application context, read each pass for the connected device. + Nothing is touched until :meth:`start`. + """ + self._ctx = ctx + self._task: Optional[asyncio.Task] = None + self._reading: Optional[BatteryReading] = None + #: Recent ``(monotonic_time, millivolts)`` samples, for the charging trend. + self._history: deque[tuple[float, int]] = deque(maxlen=_TREND_SAMPLES) + + @property + def active(self) -> bool: + """Whether the poll loop is running.""" + return self._task is not None and not self._task.done() + + def reading(self) -> Optional[BatteryReading]: + """The latest battery snapshot, or ``None`` when unknown or no battery is present.""" + return self._reading + + async def start(self) -> None: + """Start the poll loop. Idempotent; never blocks on the first read.""" + if self.active: + return + self._task = asyncio.ensure_future(self._run()) + self._ctx.log.info("battery poller started") + + async def stop(self) -> None: + """Stop the poll loop. Idempotent.""" + task, self._task = self._task, None + if task is not None: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self._ctx.log.info("battery poller stopped") + + async def aclose(self) -> None: + """Stop the loop at session end (an alias for :meth:`stop`).""" + await self.stop() + + async def _run(self) -> None: + """Read once at once, then tick forever: a best-effort pass, then sleep.""" + while True: + try: + await self._poll() + except asyncio.CancelledError: + 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 asyncio.sleep(POLL_S) + + async def _poll(self) -> None: + """One pass: read the pack, estimate charge, and re-derive the charging trend.""" + ctx = self._ctx + if not ctx.is_connected: + return + device = await ctx.device() + info = await device.get_battery() + mv = int(info.get("level") or 0) + if mv < _BATTERY_PRESENT_FLOOR_MV: + # No usable level — this companion has no battery gauge; report absence so the + # header draws nothing, and forget any stale trend from a different device. + self._reading = None + self._history.clear() + return + now = time.monotonic() + self._history.append((now, mv)) + self._reading = BatteryReading( + millivolts=mv, + percent=battery_percent(mv), + charging=self._charging(now), + ) + + 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). + + Args: + now: The current monotonic time (the just-taken sample's timestamp). + + 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: + return False + return last diff --git a/meshterm/ui/menu.py b/meshterm/ui/menu.py index 932cd71..2551ccf 100644 --- a/meshterm/ui/menu.py +++ b/meshterm/ui/menu.py @@ -29,6 +29,7 @@ from ..tools import all_tools from .surface import TuiUi from .braillechart import activity_sparkline from .theme import make_console +from .widgets import battery_cell from .tui import ( CANCEL, Choice, @@ -133,9 +134,11 @@ def _header(ctx: AppContext, cache: dict, width: int) -> Text: (``(COM5)`` / ``(BLE)``), unread-message and Watchtower badges (shown only when something is waiting), and the braille activity pulse counting every packet the hub hears at one minute per dot column — newest at the right edge, like every MeshTerm - timeline — stretched to fill every remaining cell of the row, so a wider terminal - simply shows deeper history. Separators are roomy by default and drop to a compact - ``·`` when that would squeeze the pulse below :data:`_SPARK_MIN_CELLS`. + timeline. The pulse fills the middle of the row; a battery gauge (only when the + companion reports a pack) is pinned to the right edge and the pulse cedes exactly the + cells it needs, so a wider terminal simply shows deeper history. Separators are roomy + by default and drop to a compact ``·`` when that would squeeze the pulse below + :data:`_SPARK_MIN_CELLS`. Args: ctx: The shared application context, read live on every repaint. @@ -149,11 +152,19 @@ def _header(ctx: AppContext, cache: dict, width: int) -> Text: single line (see ``frame.compose_base``), so a too-narrow terminal chops the tail rather than wrapping. """ - header = _header_segments(ctx, cache, " · ") - if width - header.cell_len < _SPARK_MIN_CELLS: - header = _header_segments(ctx, cache, " · ") - # Two dot columns per cell: every cell left of the row's edge shows two minutes. - room = width - header.cell_len + # The battery gauge is pinned to the row's right edge, so reserve its width (plus a + # leading separator) before the pulse claims the rest; a wider separator estimate for + # the fit decision is harmless slack. + battery = _battery_segment(ctx) + reserve = (cell_len(" · ") + battery.cell_len) if battery.cell_len else 0 + sep = " · " + header = _header_segments(ctx, cache, sep) + if width - header.cell_len - reserve < _SPARK_MIN_CELLS: + sep = " · " + header = _header_segments(ctx, cache, sep) + reserve = (cell_len(sep) + battery.cell_len) if battery.cell_len else 0 + # Two dot columns per cell: every cell left of the reserved tail shows two minutes. + room = width - header.cell_len - reserve if room > 0: # Buckets seeded from a previous session's stored history draw grey; only # traffic this session actually heard pulses green. @@ -168,9 +179,46 @@ def _header(ctx: AppContext, cache: dict, width: int) -> Text: column_styles=styles, ) ) + if battery.cell_len: + # Pin the gauge flush right: once the pulse has filled `room`, the separator and + # gauge land against the edge. With no room for a pulse (a very narrow terminal), + # pad instead so the gauge still sits in the corner rather than trailing the text. + if room <= 0: + header.append(" " * max(0, width - header.cell_len - reserve)) + header.append(sep) + header.append_text(battery) return header +#: Seconds per animation step of the header battery gauge (the charging sweep and the +#: low-battery blink). Matched to the session's ~1 Hz idle repaint (see the app's +#: ``refresh_interval``) so each repaint advances the animation by one clean step rather +#: than aliasing across skipped frames. +_BATTERY_ANIM_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. + + Reads the poller's cached snapshot (:meth:`~meshterm.services.battery_service. + BatteryService.reading`) — never the radio — so it is safe on the render path. The + animation frame is derived from the wall clock, so the charging sweep and low-battery + blink advance on their own without the header threading a counter through. + + Args: + ctx: The shared application context, read for the latest battery snapshot. + + Returns: + The gauge as a Rich :class:`Text` (glyph + ``%``), or an empty one when the + companion reports no battery. + """ + reading = ctx.battery.reading() + if reading is None: + return Text() + frame = int(time.monotonic() / _BATTERY_ANIM_S) + return battery_cell(reading.percent, charging=reading.charging, frame=frame) + + def _header_segments(ctx: AppContext, cache: dict, sep: str) -> Text: """The header's fixed segments — everything left of the pulse — at one separator width. @@ -548,6 +596,9 @@ async def _resume_monitor(ctx: AppContext) -> None: # The courier drains the outbox on its own paced schedule; each pass checks for a # connected device and skips quietly without one, exactly like the advert scheduler. await ctx.courier.start() + # The battery poller reads the pack for the header's fuel gauge; it only ever reads, + # and skips quietly without a device, so it too is safe to run from launch. + await ctx.battery.start() if not ctx.settings.connect_on_start: return try: diff --git a/meshterm/ui/theme.py b/meshterm/ui/theme.py index d823a6b..d9528f6 100644 --- a/meshterm/ui/theme.py +++ b/meshterm/ui/theme.py @@ -61,6 +61,13 @@ MESH_THEME = Theme( "snr.good": "bold #4ade80", "snr.ok": "bold #fbbf24", "snr.bad": "bold #f87171", + # The status-bar battery gauge, as a fuel-gauge palette: green healthy, orange at + # a quarter, red near empty. ``batt.dim`` is the off-beat of the critically-low + # blink — a dark slate the red glyph flickers to, so it pulses without vanishing. + "batt.high": "bold #4ade80", + "batt.mid": "bold #fb923c", + "batt.low": "bold #f87171", + "batt.dim": "#475569", } ) diff --git a/meshterm/ui/widgets.py b/meshterm/ui/widgets.py index ca44a80..295cc12 100644 --- a/meshterm/ui/widgets.py +++ b/meshterm/ui/widgets.py @@ -844,3 +844,85 @@ def tx_opt_summary(result: TxOptResult) -> Panel: ("status ", "muted"), Text.from_markup(applied), ) return Panel(body, title="[accent]TX optimization[/accent]", border_style="accent", expand=False) + + +# -- battery gauge -------------------------------------------------------------------------- + +#: Unicode braille pattern base; add an 8-bit dot mask (a 2×4 cell) to get the glyph. +_BRAILLE_BASE = 0x2800 + +#: Dot bit per (column, row) within a braille cell — the Unicode standard layout, the same +#: mapping the map canvas rasters with. Kept here so the one-cell battery glyph needn't reach +#: into the canvas module for two constants. +_BRAILLE_DOTS = ((0x01, 0x02, 0x04, 0x40), (0x08, 0x10, 0x20, 0x80)) + +#: Battery charge tiers: ``(percent floor, braille rows lit from the bottom, colour)``, +#: highest first. The glyph is one braille cell (2×4 dots) filled from its bottom row up, so +#: four coarse levels read as a shrinking fuel gauge: the whole cell above three-quarters, +#: six dots above half, four above a quarter, two below it — green while healthy, orange at a +#: quarter, red near empty (the app's ok/warn/err semantics as a gauge). +_BATTERY_TIERS: tuple[tuple[int, int, str], ...] = ( + (75, 4, "batt.high"), + (50, 3, "batt.high"), + (25, 2, "batt.mid"), + (0, 1, "batt.low"), +) + +#: At or below this charge the red glyph blinks with a dim frame — a can't-miss low warning. +_BATTERY_CRITICAL = 10 + +#: Frames in the charging sweep: empty → four fills → round again (bottom-to-full loop). +_CHARGE_FRAMES = 5 + + +def _braille_fill(rows: int) -> str: + """A single braille cell with its bottom ``rows`` (0–4) dot rows lit.""" + rows = max(0, min(4, rows)) + bits = 0 + for row in range(4 - rows, 4): + bits |= _BRAILLE_DOTS[0][row] | _BRAILLE_DOTS[1][row] + return chr(_BRAILLE_BASE + bits) + + +def _battery_tier(percent: int) -> tuple[int, str]: + """The ``(rows lit, colour)`` a charge level draws at (see :data:`_BATTERY_TIERS`).""" + for floor, rows, color in _BATTERY_TIERS: + if percent >= floor: + return rows, color + return 1, "batt.low" + + +def battery_cell(percent: int, *, charging: bool = False, frame: int = 0) -> Text: + """The status-bar battery gauge: one braille cell coloured by charge, then its ``%``. + + The braille cell fills from the bottom up in four coarse steps (see + :data:`_BATTERY_TIERS`), coloured green while healthy, orange at a quarter, red near + empty. Two live states animate off the caller's ``frame`` counter (advanced one step per + repaint tick), so the gauge moves without any per-frame plumbing: + + * **Charging** overrides the fill: the cell sweeps empty-to-full on a loop, the charge + colour held, so a plugged-in pack visibly climbs — while the number beside it stays the + *true* charge, never the animation's. + * **Critically low** (≤ :data:`_BATTERY_CRITICAL`%, not charging) blinks the red cell to a + dim slate on alternate frames — an unmissable pulse in the corner. + + Args: + percent: State of charge, 0–100 (clamped). + charging: Whether the pack is taking charge (drives the fill sweep). + frame: A monotonically advancing tick; only its phase is read, so any + steadily-incrementing integer animates the two live states. + + Returns: + A Rich :class:`Text`: the coloured glyph, a space, and ``NN%`` in muted text. + """ + pct = max(0, min(100, int(percent))) + rows, color = _battery_tier(pct) + if 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 + elif pct <= _BATTERY_CRITICAL and frame % 2: + color = "batt.dim" + out = Text(_braille_fill(rows), style=color) + out.append(f" {pct}%", style="muted") + return out diff --git a/tests/test_battery.py b/tests/test_battery.py new file mode 100644 index 0000000..615aa89 --- /dev/null +++ b/tests/test_battery.py @@ -0,0 +1,140 @@ +"""Battery gauge tests: the LiPo curve, the one-cell braille glyph, and the poller. + +The widget is pure (percent + flags → a coloured glyph) and the poller runs against a +fake context, so charge estimation, the charging sweep, the low-battery blink, and the +charging inference are all assertable without a device or a terminal. +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + +from meshterm.services.battery_service import ( + _BATTERY_PRESENT_FLOOR_MV, + _CHARGING_RISE_MV, + BatteryService, + battery_percent, +) +from meshterm.ui.widgets import _BATTERY_CRITICAL, battery_cell + +# --- the LiPo state-of-charge curve ------------------------------------------------------- + + +def test_battery_percent_clamps_past_either_end() -> None: + """Above the top rung reads 100%, below the bottom rung reads 0%.""" + assert battery_percent(4300) == 100 + assert battery_percent(4200) == 100 + assert battery_percent(3270) == 0 + assert battery_percent(3000) == 0 + + +def test_battery_percent_interpolates_and_stays_monotonic() -> None: + """A rung reads exactly; between rungs interpolates; more volts is never less charge.""" + assert battery_percent(3840) == 50 # an exact rung + assert 40 < battery_percent(3810) < 45 # between the 3820/45 and 3800/40 rungs + pcts = [battery_percent(mv) for mv in range(3270, 4201, 10)] + assert pcts == sorted(pcts) + + +# --- the one-cell braille glyph ----------------------------------------------------------- + + +def test_battery_cell_fills_from_the_bottom_by_tier() -> None: + """Each tier drops dot rows off the top: full, six, four, two dots lit.""" + assert battery_cell(100).plain[0] == "⣿" # all 8 dots + assert battery_cell(90).plain[0] == "⣿" + assert battery_cell(60).plain[0] == chr(0x2800 | 0xF6) # bottom 6 dots + assert battery_cell(30).plain[0] == chr(0x2800 | 0xE4) # bottom 4 dots + assert battery_cell(10).plain[0] == chr(0x2800 | 0xC0) # bottom 2 dots + + +def test_battery_cell_shows_the_true_percent() -> None: + """The number beside the glyph is the actual charge.""" + assert battery_cell(42).plain == chr(0x2800 | 0xE4) + " 42%" + + +def test_battery_cell_colours_by_tier() -> None: + """Green healthy, orange at a quarter, red near empty — the fuel-gauge palette.""" + assert str(battery_cell(90).style) == "batt.high" # green + assert str(battery_cell(60).style) == "batt.high" # green + assert str(battery_cell(30).style) == "batt.mid" # orange + assert str(battery_cell(5).style) == "batt.low" # red + + +def test_battery_cell_blinks_when_critically_low() -> None: + """Below the critical line the red glyph flickers to a dim slate on alternate frames.""" + assert 5 <= _BATTERY_CRITICAL + assert str(battery_cell(5, frame=0).style) == "batt.low" # the on-beat, red + assert str(battery_cell(5, frame=1).style) == "batt.dim" # the off-beat, dim + # Above the critical line the glyph never blinks. + assert {str(battery_cell(40, frame=f).style) for f in range(4)} == {"batt.mid"} + + +def test_battery_cell_charging_sweeps_bottom_to_full_holding_the_percent() -> None: + """Charging animates the fill empty→full on a loop while the % stays the true charge.""" + fills = [battery_cell(20, charging=True, frame=f).plain[0] for f in range(6)] + assert fills[0] == chr(0x2800) # empty + assert fills[4] == "⣿" # full + assert fills[5] == fills[0] # loops back + assert battery_cell(20, charging=True, frame=4).plain.endswith(" 20%") + # The charge colour tracks the real state of charge (20% → red), held across the sweep. + assert {str(battery_cell(20, charging=True, frame=f).style) for f in range(6)} == {"batt.low"} + + +# --- the poller --------------------------------------------------------------------------- + + +def _service(levels) -> BatteryService: + """A ``BatteryService`` over a fake context whose device answers ``levels`` in turn.""" + seq = list(levels) + + async def get_battery() -> dict: + return {"level": seq.pop(0)} if seq else {} + + async def device() -> SimpleNamespace: + return SimpleNamespace(get_battery=get_battery) + + ctx = SimpleNamespace( + is_connected=True, + device=device, + log=SimpleNamespace(info=lambda *a, **k: None, debug=lambda *a, **k: None), + ) + return BatteryService(ctx) + + +def test_battery_poller_reads_and_estimates() -> None: + """One pass reads the pack and caches a charge estimate; charging is off with one sample.""" + svc = _service([3840]) # 50% + asyncio.run(svc._poll()) + reading = svc.reading() + assert reading is not None + assert reading.millivolts == 3840 and reading.percent == 50 and reading.charging is False + + +def test_battery_poller_hides_a_device_with_no_pack() -> None: + """An empty or sub-floor level reads as *no battery*, so the header shows nothing.""" + svc = _service([0]) + asyncio.run(svc._poll()) + assert svc.reading() is None + svc = _service([_BATTERY_PRESENT_FLOOR_MV - 1]) + asyncio.run(svc._poll()) + 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.""" + svc = _service([]) + now = 1000.0 + svc._history.extend([(now - 120, 3700), (now, 3700 + _CHARGING_RISE_MV)]) + assert svc._charging(now) is True # rising → charging + + 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._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 diff --git a/tests/test_header.py b/tests/test_header.py index 567cb53..3dda37f 100644 --- a/tests/test_header.py +++ b/tests/test_header.py @@ -7,8 +7,13 @@ from types import SimpleNamespace from meshterm.ui.menu import _HEADER_ACTIVITY_LEVELS, _header -def _ctx(histogram=(), unread=0, alerts=0) -> SimpleNamespace: - """A minimal stand-in exposing only what the header reads (simulator path).""" +def _ctx(histogram=(), unread=0, alerts=0, battery=None) -> SimpleNamespace: + """A minimal stand-in exposing only what the header reads (simulator path). + + ``battery`` is the poller's cached reading (a ``percent``/``charging`` object) or + ``None`` for a companion with no pack — the default, so most tests see the old + battery-free header. + """ return SimpleNamespace( mock=True, chat=SimpleNamespace(unread_total=lambda: unread), @@ -17,6 +22,7 @@ def _ctx(histogram=(), unread=0, alerts=0) -> SimpleNamespace: activity_session_flags=lambda: (True,) * len(tuple(histogram)), ), watchtower=SimpleNamespace(unacked_count=lambda: alerts), + battery=SimpleNamespace(reading=lambda: battery), ) @@ -89,4 +95,28 @@ def test_header_survives_a_sliver_terminal() -> None: """When the fixed segments already overflow the width, the sparkline just stays out.""" header = _header(_ctx(histogram=(9,) * 360), {}, 10) # No room left: no braille at all, nothing stretched past the edge. - assert all(not (0x2800 <= ord(ch) <= 0x28FF) for ch in header.plain) \ No newline at end of file + assert all(not (0x2800 <= ord(ch) <= 0x28FF) for ch in header.plain) + + +def _spark(row: str) -> list[str]: + return [ch for ch in row if 0x2800 <= ord(ch) <= 0x28FF] + + +def test_header_pins_the_battery_gauge_to_the_right_edge() -> None: + """A companion with a pack shows its charge in the corner; the pulse cedes the room.""" + reading = SimpleNamespace(percent=87, charging=False) + header = _header(_ctx(histogram=(21,) * 360, battery=reading), {}, 80) + assert header.cell_len == 80 # still fills the row exactly, gauge included + assert header.plain.rstrip().endswith("87%") # the gauge sits flush right + # The pulse gave up real cells to make room (fewer sparkline cells than a bare row, + # even counting the gauge's own full-cell glyph). + with_batt = len(_spark(header.plain)) + bare = len(_spark(_header(_ctx(histogram=(21,) * 360), {}, 80).plain)) + assert with_batt < bare + + +def test_header_hides_the_gauge_without_a_battery() -> None: + """No pack, no gauge — the pulse keeps the whole row it always had.""" + header = _header(_ctx(histogram=(21,) * 360, battery=None), {}, 80) + assert "%" not in header.plain + assert header.cell_len == 80 \ No newline at end of file