The Time Machine's whole-mesh page cycles through each scope, as the dashboard does

s (F2 on the PicoCalc) steps all → unscoped → each region heard in the window
A–Z → unknown scope, beside w's window cycle. A narrowed page charts that
scope's floods per day (per hour at 24 h), their time-of-day rhythm, and its
own ledger; nodes per day and arrivals leave, being read off adverts, which
carry no scope. A window's stored floods are read once (Repository.flood_frames)
and resolved against the known region names for the page's life.

The node and own-node pages offer no cycle: a node's page is its adverts and
telemetry, and what we send is traces, which go direct.

The ring's order, words and chip move to ui/scopering.py, shared with the
dashboard.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Opus 5.5 committed 2026-09-28 02:35:31 -04:00
1 parent ca194fbd30
commit 307e767570
6 files changed
+500 -93

No files matched your search

+38
View File
@@ -1401,6 +1401,44 @@ class Repository:
continue # a malformed stray timestamp simply isn't counted
return counts
def flood_frames(self, *, since: datetime | None = None) -> list[tuple[datetime, dict]]:
"""Every stored flood frame, oldest first, with what its scope is read from.
The Time Machine's scope views: a scoped frame's region is named by recomputing its
transport code under each known region's key, which no SQL can do, so the frames
come back as the raw mapping :meth:`~meshterm.core.region_store.RegionStore.scope_of`
reads (route, transport code, the bytes the code was computed over) for the caller
to resolve. Only floods: a direct frame has no scope, and a frame stored before its
route was kept cannot say which it was.
Args:
since: Only frames at or after this time, if given.
Returns:
``(observed_at, raw)`` per flood frame, oldest first.
"""
sql = (
"SELECT observed_at, route, transport_code, scope_body FROM observations "
"WHERE kind = 'packet' AND upper(route) IN ('FLOOD', 'TC_FLOOD')"
)
params: list[Any] = []
if since is not None:
sql += " AND observed_at >= ?"
params.append(since.isoformat())
sql += " ORDER BY observed_at"
frames: list[tuple[datetime, dict]] = []
for row in self._conn.execute(sql, params):
when = _as_when(row["observed_at"])
if when is None:
continue # a malformed stray simply isn't counted
raw = {"route_typename": row["route"]}
if row["transport_code"]:
raw["transport_code"] = row["transport_code"]
if row["scope_body"]:
raw["scope_body"] = row["scope_body"]
frames.append((when, raw))
return frames
def self_transmissions(self, *, since: datetime | None = None) -> list[datetime]:
"""Timestamps of everything this station put on the air, oldest first.
+12 -71
View File
@@ -55,8 +55,8 @@ from ..core.regions import Scope
from ..persistence.repository import OBSERVATION_WINDOW
from ..services.monitor_service import ACTIVITY_BUCKET_S, ACTIVITY_BUCKETS
from .braillechart import axis_chart, axis_chrome, axis_label_w, meter, timeline_rows
from .menus import fit_cells
from .packet_viewer import KIND_STYLES, payload_marks
from .scopering import ScopeKey, ring, scope_atom, scope_chip, scope_key, step
from .theme import snr_style
from .trace_screen import snr_bar
from .tui.render import render_lines
@@ -122,57 +122,10 @@ _TRAFFIC_METER_CELLS = 24
_GRID_LABEL_W = 9
#: A scope view: ``("unscoped",)``, ``("region", name)`` or ``("unknown",)``. ``None`` is
#: the unnarrowed view — every packet heard, direct ones included.
ScopeKey = tuple[str, ...]
#: The screen's title, before any scope atom.
_TITLE = "Dashboard — mesh overview"
def _scope_key(scope: Scope | None) -> ScopeKey | None:
"""The view a frame counts in, or ``None`` for one with no scope to state.
A direct frame has none, and neither does one stored before its route type was kept;
both are counted only in the unnarrowed view. Every scoped frame no known region name
reproduces shares one ``unknown`` view: its code changes with each packet, so it
cannot tell two unnamed regions apart.
"""
if scope is None:
return None
if not scope.scoped:
return ("unscoped",)
if scope.region:
return ("region", scope.region)
return ("unknown",)
def _ring_order(key: ScopeKey) -> tuple[int, str]:
"""Where a view sits in the cycle: unscoped, the regions A–Z, then unknown.
Alphabetical rather than busiest-first so the ring holds still while the counts move:
a cycle whose order shifted under the reader's thumb would skip or repeat a view.
"""
if key[0] == "unscoped":
return (0, "")
if key[0] == "region":
return (1, key[1].casefold())
return (2, "")
def _scope_atom(key: ScopeKey) -> str:
"""A view's title atom: ``scope yul``, ``unscoped`` or ``unknown scope``."""
if key[0] == "region":
return f"scope {key[1]}"
return "unscoped" if key[0] == "unscoped" else "unknown scope"
def _scope_chip(key: ScopeKey | None) -> str:
"""The F3 chip naming the view a press goes *to*, within the lane's 6 cells."""
word = "all" if key is None else key[-1]
return "▸ " + fit_cells(word, 4).rstrip()
class _ScopeDigest(NamedTuple):
"""One scope view's numbers over the window, in the shapes the sections draw.
@@ -230,10 +183,9 @@ class DashboardScreen(Screen):
from .tui.fkeys import FPair, default_lane
lane = list(default_lane(nav=self.content_overflows))
ring = self._ring()
if len(ring) > 1:
nxt = ring[(ring.index(self._scope) + 1) % len(ring)]
lane[2] = FPair(_scope_chip(nxt), "scope")
views = self._ring()
if len(views) > 1:
lane[2] = FPair(scope_chip(step(views, self._scope)), "scope")
return lane
def __init__(
@@ -339,27 +291,16 @@ class DashboardScreen(Screen):
# --- scope -----------------------------------------------------------------------
def _ring(self) -> list[ScopeKey | None]:
"""Every view on offer: all, then each scope heard in the window.
The view on screen stays in the ring even once its last frame ages out, so the
cycle never loses the reader's place; it simply leaves on the next press.
"""
"""Every view on offer: all, then each scope heard in the window."""
self._digest_window()
keys = {key for key, _ in self._win_scoped}
if self._scope is not None:
keys.add(self._scope)
return [None, *sorted(keys, key=_ring_order)]
return ring((key for key, _ in self._win_scoped), self._scope)
def _cycle_scope(self) -> None:
"""Step to the next view.
The cycle wraps: ``s`` is forward-only, with no reverse key of its own, so a ring
that stopped at its end would strand the reader there.
"""
ring = self._ring()
if len(ring) < 2:
"""Step to the next view (see :func:`~meshterm.ui.scopering.step`)."""
views = self._ring()
if len(views) < 2:
return
self._scope = ring[(ring.index(self._scope) + 1) % len(ring)]
self._scope = step(views, self._scope)
self._set_title()
self.scroll_to_top()
self._session.invalidate()
@@ -369,7 +310,7 @@ class DashboardScreen(Screen):
if self._scope is None:
self.title, self.short_title = _TITLE, ""
return
atom = _scope_atom(self._scope)
atom = scope_atom(self._scope)
self.title = f"{_TITLE} · {atom}"
self.short_title = f"Dashboard · {atom}"
@@ -445,7 +386,7 @@ class DashboardScreen(Screen):
repeaters.add(o.node)
if o.kind == "packet":
# The region store memoizes by frame, so a frame is checked once per visit.
key = _scope_key(self._scope_of(o.raw)) if self._scope_of else None
key = scope_key(self._scope_of(o.raw)) if self._scope_of else None
if key is not None:
scoped.append((key, o))
continue
+91
View File
@@ -0,0 +1,91 @@
# SPDX-License-Identifier: Apache-2.0
"""The scope ring: the views a screen of recorded traffic can narrow to, and their words.
A flood is either unscoped or scoped into a region, and a screen that counts floods can
show all of them or only one scope's. The dashboard and the Time Machine's mesh page both
cycle those views on ``s`` (F2/F3 on the PicoCalc), so the ring's order, the title atom a
view is named by, and the chip that names the next one are defined once, here.
A view is a :data:`ScopeKey` — ``("unscoped",)``, ``("region", name)`` or ``("unknown",)``
— and ``None`` is the unnarrowed view, every packet heard. A direct frame has no scope, so
it belongs to no view but that one.
"""
from __future__ import annotations
from collections.abc import Iterable
from ..core.regions import Scope
from .menus import fit_cells
#: A scope view: ``("unscoped",)``, ``("region", name)`` or ``("unknown",)``.
ScopeKey = tuple[str, ...]
def scope_key(scope: Scope | None) -> ScopeKey | None:
"""The view a frame counts in, or ``None`` for one with no scope to state.
A direct frame has none, and neither does one stored before its route type was kept;
both are counted only in the unnarrowed view. Every scoped frame no known region name
reproduces shares one ``unknown`` view: its code changes with each packet, so it
cannot tell two unnamed regions apart.
"""
if scope is None:
return None
if not scope.scoped:
return ("unscoped",)
if scope.region:
return ("region", scope.region)
return ("unknown",)
def _ring_order(key: ScopeKey) -> tuple[int, str]:
"""Where a view sits in the cycle: unscoped, the regions A–Z, then unknown.
Alphabetical rather than busiest-first so the ring holds still while the counts move:
a cycle whose order shifted under the reader's thumb would skip or repeat a view.
"""
if key[0] == "unscoped":
return (0, "")
if key[0] == "region":
return (1, key[1].casefold())
return (2, "")
def ring(heard: Iterable[ScopeKey], current: ScopeKey | None) -> list[ScopeKey | None]:
"""Every view on offer: all, then each scope heard, in cycle order.
The view on screen stays in the ring even once nothing in it is left to count (its
last frame aged out, or a narrower window holds none), so the cycle never loses the
reader's place; it simply leaves on the next press.
"""
keys = set(heard)
if current is not None:
keys.add(current)
return [None, *sorted(keys, key=_ring_order)]
def step(views: list[ScopeKey | None], current: ScopeKey | None) -> ScopeKey | None:
"""The view after ``current``.
The cycle wraps: ``s`` is forward-only, with no reverse key of its own, so a ring
that stopped at its end would strand the reader there.
"""
return views[(views.index(current) + 1) % len(views)]
def scope_atom(key: ScopeKey, *, bare: bool = False) -> str:
"""A view's title atom: ``scope yul``, ``unscoped`` or ``unknown scope``.
``bare`` drops the ``scope`` lead-in before a region's name, for a title too narrow
to spend the word on (the lexicon's region is shown bare anyway).
"""
if key[0] == "region":
return key[1] if bare else f"scope {key[1]}"
return "unscoped" if key[0] == "unscoped" else "unknown scope"
def scope_chip(key: ScopeKey | None) -> str:
"""The F-lane chip naming the view a press goes *to*, within the lane's 6 cells."""
word = "all" if key is None else key[-1]
return "▸ " + fit_cells(word, 4).rstrip()
+199 -19
View File
@@ -15,6 +15,14 @@ the PicoCalc's ring stops at 30 d, its F3 chip cycling the same three spans):
mesh-wide hour-of-day rhythm, the arrivals of the window (nodes heard for the first
time ever, in aligned name/hash/first-heard lanes), and the all-time totals.
The whole-mesh page also cycles its **scope** on ``s`` (F2 on the PicoCalc): all, unscoped,
each region heard in the window A–Z, unknown scope — the dashboard's ring over the recorded
history. A narrowed page charts that scope's floods per day (or hour) and their rhythm, and
its own ledger; the node counts and arrivals leave, being read off adverts, which carry no
scope. Floods stored before MeshTerm kept their route count only under all. The node and
own-node pages have no scope to narrow: a node's page is its adverts and telemetry, and
what we send is traces, which go direct.
Charts read chronologically — oldest at the left, now at the right, the app-wide
timeline direction — and draw through :mod:`~meshterm.ui.braillechart`, so the grey
baseline always marks zero: the SNR band's readings hang below it or rise above it
@@ -24,6 +32,7 @@ transmits.
from __future__ import annotations
from collections import Counter
from collections.abc import Callable
from datetime import date, datetime, timedelta
from statistics import median
@@ -45,6 +54,7 @@ from .braillechart import (
)
from .contactlist import SORT_COLUMNS, SORT_OPENS_ASCENDING, ContactListScreen, ContactRow
from .menus import command_label, fit_cells, section_heading
from .scopering import ScopeKey, ring, scope_atom, scope_chip, scope_key, step
from .theme import name_style, snr_style
from .tui.render import render_lines
from .tui.screen import CANCEL, Screen
@@ -214,10 +224,18 @@ def _quarter_axis(frac: float) -> str:
class TimeMachineScreen(Screen):
"""One subject's history page: scrollable sections, ``w`` cycles the window."""
"""One subject's history page: scrollable sections, ``w`` cycles the window.
A page whose history carries scope (the whole mesh's) also cycles it on ``s``.
"""
floating = False
footer_hint = "↑↓ PgUp/PgDn scroll · w window · Esc back"
@property
def footer_hint(self) -> str: # type: ignore[override]
"""The page's keys; ``s scope`` only where there is a scope to narrow to."""
scope = " · s scope" if len(self._ring()) > 1 else ""
return f"↑↓ PgUp/PgDn scroll · w window{scope} · Esc back"
@property
def fkey_lane(self):
@@ -241,6 +259,10 @@ class TimeMachineScreen(Screen):
lane = list(default_lane(nav=self.content_overflows))
nxt, _delta = _WINDOWS[(self._window_index + 1) % len(_WINDOWS)]
lane[2] = FPair(f"▸ {'all' if nxt == 'all time' else nxt}", "window")
# The scope cycle beside it, on the same terms: the chip names where a press goes.
views = self._ring()
if len(views) > 1:
lane[1] = FPair(scope_chip(step(views, self._scope)), "scope")
return lane
def __init__(
@@ -248,28 +270,45 @@ class TimeMachineScreen(Screen):
*,
session, # noqa: ANN001 - TuiSession, imported lazily to avoid a cycle
label: str,
build: Callable[[timedelta | None, int], list[RenderableType]],
build: Callable[[timedelta | None, int, ScopeKey | None], list[RenderableType]],
scopes: Callable[[timedelta | None], set[ScopeKey]] | None = None,
) -> None:
"""Create the page over its section builder.
Args:
session: The running TUI session (for repaints on window switch).
label: The subject's display name (titles the screen).
build: Renders the sections for ``(window, width)``; called once per
window/width combination and cached — the data is stored history,
so nothing needs re-querying per repaint.
build: Renders the sections for ``(window, width, scope)``; called once per
combination and cached — the data is stored history, so nothing needs
re-querying per repaint. ``scope`` is ``None`` for every packet.
scopes: The scope views a window's history holds, for a page that can narrow
to one; ``None`` offers no scope cycle.
"""
super().__init__()
self._session = session
self._label = label
self._build = build
self._scopes = scopes
self._window_index = 1 # open on 7 d: enough depth to see shape, still fast
self._cache: dict[tuple[int, int], list[str]] = {}
#: The scope view on screen; ``None`` is every packet.
self._scope: ScopeKey | None = None
self._cache: dict[tuple[int, int, ScopeKey | None], list[str]] = {}
self._set_title()
def _ring(self) -> list[ScopeKey | None]:
"""The scope views on offer in the current window (just ``[None]`` if none)."""
if self._scopes is None:
return [None]
_name, delta = _WINDOWS[self._window_index]
return ring(self._scopes(delta), self._scope)
def _set_title(self) -> None:
name, _delta = _WINDOWS[self._window_index]
self.title = f"{self._label} · {name}"
self.short_title = ""
if self._scope is not None:
self.title += f" · {scope_atom(self._scope)}"
self.short_title = f"{self._label} · {name} · {scope_atom(self._scope, bare=True)}"
def handle(self, action: str, data: str = "") -> None:
"""Scroll, cycle the window, or dismiss.
@@ -295,16 +334,23 @@ class TimeMachineScreen(Screen):
self._set_title()
self.scroll_to_top()
self._session.invalidate()
elif action == "scope" or (action == "text" and data.lower() == "s"):
views = self._ring()
if len(views) > 1:
self._scope = step(views, self._scope)
self._set_title()
self.scroll_to_top()
self._session.invalidate()
elif action == "escape":
self.resolve(None)
def render_body(self, width: int) -> list[str]:
"""Render (or reuse) the current window's sections."""
key = (self._window_index, width)
key = (self._window_index, width, self._scope)
lines = self._cache.get(key)
if lines is None:
_name, delta = _WINDOWS[self._window_index]
lines = render_lines(Group(*self._build(delta, width)), width)
lines = render_lines(Group(*self._build(delta, width, self._scope)), width)
self._cache[key] = lines
self._scroll_total = max(1, len(lines))
return lines
@@ -1110,6 +1156,133 @@ def _mesh_sections(
return out
class _MeshFloods:
"""The mesh page's stored floods per window, each with the scope view it counts in.
Naming a scoped flood's region recomputes its transport code under each known region's
key (see :meth:`~meshterm.core.region_store.RegionStore.scope_of`), so a window's
floods are read and resolved once and kept for the page's life: the ring asks which
views a window holds, and a narrowed page asks for one view's timestamps.
"""
def __init__(self, ctx: AppContext) -> None:
"""Bind to the context's history and region names."""
self._ctx = ctx
self._by_window: dict[timedelta | None, list[tuple[ScopeKey, datetime]]] = {}
def _frames(self, window: timedelta | None) -> list[tuple[ScopeKey, datetime]]:
if window not in self._by_window:
store = getattr(self._ctx, "region_store", None)
frames: list[tuple[ScopeKey, datetime]] = []
if store is not None:
since = utcnow() - window if window is not None else None
for when, raw in self._ctx.repo.flood_frames(since=since):
key = scope_key(store.scope_of(raw))
if key is not None:
frames.append((key, when))
self._by_window[window] = frames
return self._by_window[window]
def keys(self, window: timedelta | None) -> set[ScopeKey]:
"""The scope views ``window`` holds any flood in."""
return {key for key, _ in self._frames(window)}
def stamps(self, window: timedelta | None, scope: ScopeKey) -> list[datetime]:
"""When each of the window's floods in ``scope`` was heard, oldest first."""
return [when for key, when in self._frames(window) if key == scope]
def _floods(count: int) -> str:
"""``1 flood``, ``12 floods``."""
return f"{count} flood{'' if count == 1 else 's'}"
def _mesh_scope_sections(
stamps: list[datetime], window: timedelta | None, width: int
) -> list[RenderableType]:
"""The whole-mesh page narrowed to one scope: its floods per day, their rhythm, a ledger.
The unnarrowed page's day (or hour) bars and rhythm, over the scope's floods instead of
every observation. Its node counts and arrivals have no narrowed form — a node is known
by its adverts, and an advert says nothing about the scope other traffic was sent in —
so they leave rather than show the whole mesh's under a scope's title.
"""
now = utcnow()
since = now - window if window is not None else None
if not stamps:
return [
Text(),
Text("Nothing recorded in this scope in this window.", style="muted"),
Text("Press w to widen it, or s for another scope.", style="muted"),
]
hourly = since is not None and window is not None and window <= timedelta(days=1)
# The same local-time keys the repository's day and hour series group by, so the fill
# and tick machinery below reads them exactly as it reads the unnarrowed page's.
per = Counter(
when.astimezone().strftime("%Y-%m-%dT%H" if hourly else "%Y-%m-%d") for when in stamps
)
active = [(iso, count, 0) for iso, count in sorted(per.items())]
series = _fill_hours(active, since, now) if hourly and since else _fill_days(active, since, now)
base_w = axis_label_w(max(d[1] for d in series), _CHART_ROWS)
slots, slice_minutes, label_w = _fit_rhythm(
lambda minutes: _rhythm_slots(stamps, minutes), width, base_w
)
chars = max(20, width - axis_chrome(label_w))
shown = series[-chars * 2 :]
floods = [d[1] for d in shown]
out: list[RenderableType] = []
out.append(
body_heading(
"Floods per hour" if hourly else "Floods per day",
"local hours" if hourly else "local days",
)
)
out.extend(
axis_chart(
timeline_rows(_day_columns(floods, chars), rows=_CHART_ROWS),
max(floods),
chars,
label_w=label_w,
ticks=_hour_ticks(shown, chars) if hourly else _day_ticks(shown, chars),
)
)
out.append(Text())
out.append(
body_heading("Rhythm", f"floods by local time of day · {_slice_note(slice_minutes)}")
)
out.extend(
axis_chart(
timeline_rows(slots, rows=_CHART_ROWS),
max(slots),
len(slots) // 2,
_quarter_axis,
label_w=label_w,
)
)
days = Counter(when.astimezone().strftime("%Y-%m-%d") for when in stamps)
busiest = max(days.items(), key=lambda day: day[1])
out.append(Text())
out.append(body_heading("Ledger", "this window"))
out.append(
Text.assemble(
("history ", "muted"),
(_floods(len(stamps)), ""),
(f" across {len(days)} active day{'' if len(days) == 1 else 's'}", "muted"),
)
)
out.append(
Text.assemble(
("busiest ", "muted"),
(busiest[0], "brand"),
(f" {_floods(busiest[1])}", "muted"),
)
)
return out
# --- the picker loop ---------------------------------------------------------------------
#: Widest the mesh page's arrivals name lane grows (longer names ellipsize so the lanes
@@ -1336,24 +1509,29 @@ async def open_timemachine(ctx: AppContext) -> None:
return
if picked == SELF:
label = self_name or "you"
build = lambda window, width: _self_sections( # noqa: E731
build = lambda window, width, _scope: _self_sections( # noqa: E731
ctx, window, width
)
scopes = None
elif picked == MESH:
label = "the whole mesh"
build = ( # noqa: E731
lambda window, width, _pb=prefix_bytes: _mesh_sections(
ctx, window, width, _pb, resolve, resolve_key
)
)
floods = _MeshFloods(ctx)
def build(window, width, scope, _pb=prefix_bytes, _floods=floods): # noqa: ANN001, ANN202
if scope is None:
return _mesh_sections(ctx, window, width, _pb, resolve, resolve_key)
return _mesh_scope_sections(_floods.stamps(window, scope), window, width)
scopes = floods.keys
else:
node_id, label = picked
build = ( # noqa: E731 - a tiny binding closure beats a def here
lambda window, width, _id=node_id, _lb=label: _node_sections(
lambda window, width, _scope, _id=node_id, _lb=label: _node_sections(
ctx, _id, _lb, window, width
)
)
screen = TimeMachineScreen(session=session, label=label, build=build)
scopes = None
screen = TimeMachineScreen(session=session, label=label, build=build, scopes=scopes)
await session.run_screen(screen)
if {n.node for n in ctx.repo.heard_nodes() if n.node} != subjects:
break # the mesh spoke while the page was open; relist
@@ -1381,7 +1559,9 @@ async def open_timemachine_node(ctx: AppContext, node_id: str, label: str) -> No
raise RuntimeError("the time machine is only available in the menu")
session = ctx.ui.session
build = ( # noqa: E731 - a tiny binding closure reads better than a def here
lambda window, width, _id=node_id, _lb=label: _node_sections(ctx, _id, _lb, window, width)
lambda window, width, _scope, _id=node_id, _lb=label: _node_sections(
ctx, _id, _lb, window, width
)
)
await session.run_screen(TimeMachineScreen(session=session, label=label, build=build))
@@ -1407,5 +1587,5 @@ async def open_timemachine_self(ctx: AppContext) -> None:
session = ctx.ui.session
self_name, _ = await _self_identity(ctx)
label = self_name or "you"
build = lambda window, width: _self_sections(ctx, window, width) # noqa: E731
build = lambda window, width, _scope: _self_sections(ctx, window, width) # noqa: E731
await session.run_screen(TimeMachineScreen(session=session, label=label, build=build))
+28 -1
View File
@@ -812,6 +812,32 @@ def _walk_topo() -> tuple[MeshTopology, dict[str, Contact]]:
return topo, {hub_id: hub, far_id: far}
def _timemachine_scoped(cols: int, rows: int) -> Screen:
"""The whole-mesh page narrowed to one scope: its title, its F2 chip, its charts.
The region is named long on purpose, so the title needs its short form on the console.
"""
from datetime import timedelta
from meshterm.ui.timemachine_screen import _mesh_scope_sections
now = utcnow()
stamps = [now - timedelta(hours=h) for h in (150, 90, 60, 30, 20, 5, 2)]
def build(window, width, scope): # noqa: ANN001, ANN202
return _mesh_scope_sections(stamps, window, width) if scope else [Text("page")]
screen = TimeMachineScreen(
session=_GallerySession(cols, rows),
label="the whole mesh",
build=build,
scopes=lambda _window: {("unscoped",), ("region", "laurentides-nord")},
)
screen.handle("scope")
screen.handle("scope")
return screen
def _walk(cols: int, rows: int) -> Screen:
topo, contacts = _walk_topo()
return WalkScreen(
@@ -823,7 +849,7 @@ def _walk(cols: int, rows: int) -> Screen:
def _timemachine(cols: int, rows: int) -> Screen:
def build(window, width): # noqa: ANN001
def build(window, width, scope): # noqa: ANN001
return [Text("Hilltop-Repeater 5m ago advert"), Text("Alice 12m ago packet")]
return TimeMachineScreen(
@@ -1388,6 +1414,7 @@ _ENTRIES: list[_Entry] = [
_Entry("livefeed_scopes", _livefeed_scopes),
_Entry("walk", _walk),
_Entry("timemachine", _timemachine),
_Entry("timemachine-scoped", _timemachine_scoped),
_Entry("message_paths", _message_paths),
_Entry("message_paths_scoped", _message_paths_scoped),
_Entry("message_paths_unknown_scope", _message_paths_unknown_scope),
+132 -2
View File
@@ -13,6 +13,8 @@ from itertools import pairwise
from pathlib import Path
from types import SimpleNamespace
from rich.text import Text
from meshterm.core.models import Observation, utcnow
from meshterm.persistence.repository import Repository
from meshterm.ui.braillechart import GAP
@@ -954,7 +956,7 @@ def test_screen_cycles_windows_and_caches(tmp_path: Path) -> None:
"""`w` moves to the next window (retitling) and each window renders once."""
calls: list = []
def build(window, width):
def build(window, width, scope):
calls.append(window)
from rich.text import Text
@@ -988,7 +990,9 @@ def test_picocalc_window_ring_stops_at_30_days() -> None:
from meshterm.platforms import PICOCALC, set_platform
set_platform(PICOCALC)
screen = TimeMachineScreen(session=_FakeSession(), label="Hub", build=lambda window, width: [])
screen = TimeMachineScreen(
session=_FakeSession(), label="Hub", build=lambda window, width, scope: []
)
assert "7 d" in screen.title # opens on 7 d, as everywhere
screen.handle("text", "w")
assert "30 d" in screen.title
@@ -1212,3 +1216,129 @@ def test_picker_hash_lane_shows_a_captured_full_key_without_a_device() -> None:
screen = _picker([(node, "Rep")], width=100)
row = next(c for c in screen._choices() if c.value == (stored, "Rep"))
assert stored + "cd" * 10 in row.label.plain
# -- the scope cycle ----------------------------------------------------------------------
def _flood_repo(tmp_path: Path) -> Repository:
"""Stored floods under three scopes, plus a direct frame that has none.
Two ``yul`` floods, one unscoped, one scoped to a region nobody here knows (``qc``, left
out of :func:`_scope_ctx`'s names), and a direct message — each a real frame whose code
the repository must carry through storage for its region to be named again.
"""
from tests.test_scope_display import _DIRECT, _FLOOD, _scoped
repo = Repository(tmp_path / "floods.db")
run = repo.start_run("monitor", {}, None)
now = utcnow()
for hours_ago, raw in (
(30, _scoped("yul")),
(5, _scoped("yul")),
(4, _FLOOD),
(3, _scoped("qc")),
(2, _DIRECT),
):
repo.record_observation(
run,
Observation(
node=None,
kind="packet",
path="",
observed_at=now - timedelta(hours=hours_ago),
raw=dict(raw),
),
)
return repo
def _scope_ctx(repo: Repository) -> SimpleNamespace:
from tests.test_scope_display import _knows
return SimpleNamespace(repo=repo, region_store=SimpleNamespace(scope_of=_knows("yul", "mtl")))
def test_flood_frames_come_back_with_what_their_scope_is_read_from(tmp_path: Path) -> None:
"""Floods only, oldest first, each still resolvable to its region after storage."""
from tests.test_scope_display import _knows
repo = _flood_repo(tmp_path)
frames = repo.flood_frames()
assert [raw["route_typename"] for _when, raw in frames] == [
"TC_FLOOD",
"TC_FLOOD",
"FLOOD",
"TC_FLOOD",
]
assert _knows("yul")(frames[0][1]).region == "yul"
assert len(repo.flood_frames(since=utcnow() - timedelta(hours=24))) == 3
repo.close()
def test_the_mesh_page_offers_each_scope_its_window_holds(tmp_path: Path) -> None:
"""The ring is the window's: a 24 h window holds one yul flood, not the 30 h old one."""
from meshterm.ui.timemachine_screen import _MeshFloods
floods = _MeshFloods(_scope_ctx(_flood_repo(tmp_path)))
assert floods.keys(timedelta(days=1)) == {("unscoped",), ("region", "yul"), ("unknown",)}
assert len(floods.stamps(timedelta(days=7), ("region", "yul"))) == 2
assert len(floods.stamps(timedelta(days=1), ("region", "yul"))) == 1
def test_s_cycles_the_mesh_pages_scope_and_retitles_it(tmp_path: Path) -> None:
"""All → unscoped → yul → unknown → all, each view rendered for its own scope."""
from meshterm.ui.timemachine_screen import _MeshFloods
floods = _MeshFloods(_scope_ctx(_flood_repo(tmp_path)))
built: list = []
def build(window, width, scope): # noqa: ANN001, ANN202
built.append(scope)
return [Text("page")]
screen = TimeMachineScreen(
session=_FakeSession(), label="the whole mesh", build=build, scopes=floods.keys
)
assert "s scope" in screen.footer_hint
titles = []
for _ in range(4):
screen.handle("text", "s")
screen.render_body(80)
titles.append(screen.title.removeprefix("the whole mesh · 7 d"))
assert titles == [" · unscoped", " · scope yul", " · unknown scope", ""]
assert built == [("unscoped",), ("region", "yul"), ("unknown",), None]
screen.handle("scope") # the F2 chip's action is the same cycle
assert screen.short_title == "the whole mesh · 7 d · unscoped"
def test_a_page_with_no_scope_offers_no_cycle() -> None:
"""A node's page is its adverts, which carry no scope: no hint, no chip, no cycle."""
screen = TimeMachineScreen(
session=_FakeSession(), label="Hub", build=lambda window, width, scope: []
)
assert "scope" not in screen.footer_hint
screen.handle("text", "s")
assert screen.title == "Hub · 7 d"
def test_a_narrowed_mesh_page_charts_the_scopes_floods_and_drops_the_node_counts(
tmp_path: Path,
) -> None:
"""Floods per day, their rhythm, and a ledger of the scope's own; no nodes, no arrivals."""
from meshterm.ui.timemachine_screen import _mesh_scope_sections, _MeshFloods
floods = _MeshFloods(_scope_ctx(_flood_repo(tmp_path)))
week = timedelta(days=7)
text = _plain(_mesh_scope_sections(floods.stamps(week, ("region", "yul")), week, 80), 80)
assert "Floods per day" in text and "Rhythm" in text and "2 floods" in text
assert "Nodes per day" not in text and "Arrivals" not in text
day = _plain(
_mesh_scope_sections(
floods.stamps(timedelta(days=1), ("unscoped",)), timedelta(days=1), 80
),
80,
)
assert "Floods per hour" in day and "1 flood " in day
empty = _plain(_mesh_scope_sections([], week, 80), 80)
assert "Nothing recorded in this scope" in empty