rich/MeshTerm
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Live feed: the top of the list is two stops, watching and reading

Browse Source 
The newest row now carries a pin — the stop above row 0, which holds the
topmost position rather than a packet. Every arrival takes the highlight
with it, so the cursor stays on the traffic instead of being pushed down
into history by it, and it draws ^ rather than ❯: the mark points past
the top of the list, at the packets still to come, because that is what
it is attached to.

One ↓ off the pin does not move a row. It lands on that same newest
packet, now normally selected under a plain ❯, and from there the
highlight belongs to that packet and rides down as newer ones arrive. So
the pin follows the stream and a selection follows a packet, and the
cursor lane says which you are doing. ↑ from the newest packet re-pins,
Home jumps back to the pin from anywhere, and the F-lane's Top chip is
how the PicoCalc reaches it with no hint line to name it.

The pin is a real cursor stop, not a flag beside one, so every move
stays a clamp on the single index _cursor already was — which is what
makes ↓ off it land on the newest packet rather than the second. Opening
a packet drops the pin: Enter names *this* packet, and arrivals during a
long read would otherwise walk the cursor off it and land the closing
viewer somewhere the reader never chose.

^ is plain ASCII, so it needs no console-font glyph. It is deliberately
not ↑, which this screen already spends on the window's "↑ n more" and
would read here as "there is more above" — the one thing the top of a
feed can never mean; nor ▲, which is the repeater in the node-type set.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Opus 5 committed 2026-08-13 15:05:48 -04:00
1 parent 3c714e4c84
commit ec4836f63d
2 files changed
+190 -28
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

meshterm/ui/livefeed_screen.py
+115 -24
View File
@@ -18,8 +18,21 @@ tag. Only a class that genuinely says nothing about itself leaves the lane empty
Seeded from stored history so the screen opens full, then streamed
live off the event hub. ``↑``/``↓`` walk the rows, PgUp/PgDn/Home/End page and jump
them, ``←``/``→`` scroll the highlighted row sideways on a terminal too narrow to hold
it whole, and Enter opens the highlighted packet in the shared
them, and ``←``/``→`` scroll the highlighted row sideways on a terminal too narrow to
hold it whole.
The top of the list is **two stops, not one** — the difference between watching the feed
and reading it. The first is the *pin*: it holds the very topmost position, so every
packet that arrives takes the highlight with it, and the cursor draws as ``^`` (pointing
past the list's top, at the traffic still to come) rather than the app's ``❯``. One
``↓`` off it does not move down a row — it lands on that same newest packet, now
*normally* selected under a plain ``❯``, and from there the highlight belongs to that
packet and rides down as newer ones push it along. So the pin follows the stream and a
selection follows a packet, and the mark in the cursor lane says which you are doing.
The screen opens pinned (a live feed's resting state is *watching*), ``↑`` from the newest
packet re-pins, and Home jumps straight back to the pin from anywhere.
Enter opens the highlighted packet in the shared
:class:`~meshterm.ui.packet_viewer.PacketViewer` — which then pages through the feed
itself with the same ``↑``/``↓``, and, for an overheard channel-text packet naming a
channel we hold the key for, decrypts it.
@@ -128,6 +141,20 @@ _HSHIFT_RESET = frozenset({
"up", "down", "pageup", "pagedown", "space", "home", "ctrl_home", "end", "ctrl_end",
})
#: The cursor over a normally selected row — the app-wide select-list pointer, pointing
#: *into* the list at the one row it marks.
_CURSOR = "❯ "
#: The cursor over the pinned newest row (see the module docstring). The same lane, turned
#: to point past the top of the list — at the traffic still to arrive — because that is
#: what the pin is attached to: not this packet, but whichever one is newest. It is
#: deliberately not ``↑``, which this screen already spends on the window's ``↑ n more``
#: marker and would read here as "there is more above" — the one thing the top of the feed
#: can never mean; nor ``▲``, which is the repeater in the app's node-type set. A caret is
#: unclaimed, reads as "topmost" on both platforms, and is plain ASCII, so it needs no
#: entry in the console font's glyph map (see :mod:`meshterm.ui.fontset`).
_PIN_CURSOR = "^ "
def _channel_sender(text: Optional[str]) -> Optional[str]:
"""The sender named by a channel message's ``Name: `` prefix, or ``None`` if absent.
@@ -152,8 +179,10 @@ class LiveFeedScreen(Screen):
Every nav key here moves the *cursor* over the feed rather than a scroll offset,
but the vocabulary is the same either way — the window follows the highlight — so
the shared labels stand. ``Top`` and ``Bottom`` stay literal on purpose: the top of
this list is the newest packet, and its bottom is the ``Back`` row one past the
oldest, which is where End actually lands.
this list is the pin riding the newest packet, and its bottom is the ``Back`` row
one past the oldest, which is where End actually lands. ``Top`` is how the pin is
reached where there is no hint line to name it — a feed read down into history
resumes following in one chip.
A feed with nothing in it still has that ``Back`` row, and nowhere to move to, so
all four chips go dim until the first packet arrives.
@@ -220,6 +249,13 @@ class LiveFeedScreen(Screen):
#: row past the oldest packet — the screen's other cursor stop, and the only one
#: an empty feed offers. Starts on the newest packet.
self._selected: Optional[int] = 0 if self._feed else None
#: Whether the cursor is on the *pin* — the stop above row 0, which holds the
#: topmost position rather than a packet (see the module docstring). Implies
#: ``_selected == 0``: the pin is a mode on the newest row, so everything that
#: reads the highlight (the window fit, the viewer, the ``←→`` scroll) needs no
#: special case; only the cursor mark drawn in the pointer lane differs. A feed
#: with nothing in it has no row to pin to, so it opens on ``Back`` instead.
self._pinned: bool = bool(self._feed)
#: The feed's window within the fixed screen (its rows scroll under the heading).
self._feed_window = ListWindow()
#: The highlighted row's horizontal scroll (cells shifted in, ``←/→``) and how
@@ -231,7 +267,16 @@ class LiveFeedScreen(Screen):
# --- live feed -----------------------------------------------------------------
def on_event(self, event: MeshEvent) -> None:
"""Fold one hub event into the feed and repaint."""
"""Fold one hub event into the feed and repaint.
The arrival is where the screen's two top stops part company. A *selection* is
attached to a packet, so it rides down with it as the newer row pushes in above.
The *pin* is attached to the position, so it stays at row 0 and the packet under
it changes — which is the whole point of it, and also why the highlighted row's
horizontal scroll is dropped here: the cursor has landed on a different packet
just as surely as if the reader had pressed ``↓``, and every other move onto a new
row starts that row at its own beginning (see :data:`_HSHIFT_RESET`).
"""
entry: Optional[PacketEntry] = None
obs = event.observation
if obs is not None:
@@ -252,8 +297,11 @@ class LiveFeedScreen(Screen):
if entry is None:
return # nothing new on screen — don't buy a full repaint for it
self._feed.appendleft(entry)
# Keep the highlight on the same packet as new rows push it down.
if self._selected is not None:
if self._pinned:
self._selected = 0 # the pin holds the top: the newest packet is now this one
self._hshift = 0
elif self._selected is not None:
# Keep the highlight on the same packet as new rows push it down.
self._selected = min(self._selected + 1, len(self._feed) - 1)
self._session.invalidate()
@@ -289,11 +337,13 @@ class LiveFeedScreen(Screen):
elif action in ("pagedown", "space"):
self._select_stop(self._cursor + self._feed_window.page)
elif action in ("home", "ctrl_home"):
# Home jumps to the newest packet; End to the list's last row, which — as on
# every list in the app — is the ``Back`` row (the oldest packet is one ↑ up).
# Home jumps to the pin — the top of a live feed is the *following* state, so
# "take me back to the top" resumes the stream rather than parking on whichever
# packet happens to be newest this instant. End goes to the list's last row,
# which — as on every list in the app — is ``Back`` (the oldest packet is one ↑).
self._select_stop(0)
elif action in ("end", "ctrl_end"):
self._select_stop(len(self._feed))
self._select_stop(len(self._feed) + 1)
elif action == "left":
self._scroll_line(-_HSCROLL_STEP)
elif action == "right":
@@ -310,24 +360,54 @@ class LiveFeedScreen(Screen):
@property
def _cursor(self) -> int:
"""The cursor as one index over the screen's stops: a feed row, else the ``Back`` row.
"""The cursor as one index over the screen's stops, from the pin down to ``Back``.
The feed's rows number ``0 .. len(feed) - 1`` and the exit row sits one past them,
so every move is a clamp on this one number — and an empty feed leaves the ``Back``
row as the only stop there is.
The stops are the pin (0), then the feed's rows (``1 .. len(feed)``), then the exit
row one past them — so every move stays a clamp on this one number, the pin
included. That the pin and row 0 draw on the *same line* is a rendering detail;
as cursor positions they are two, which is exactly what makes ``↓`` off the pin
land on the newest packet rather than the second one.
An empty feed has no row to pin to, so it leaves ``Back`` as the only stop there is.
"""
return len(self._feed) if self._selected is None else self._selected
if self._pinned:
return 0
return len(self._feed) + 1 if self._selected is None else self._selected + 1
def _move_selection(self, delta: int) -> None:
"""Step the cursor by ``delta`` rows (off the oldest packet lands it on ``Back``)."""
"""Step the cursor by ``delta`` stops (off the oldest packet lands it on ``Back``)."""
self._select_stop(self._cursor + delta)
def _select_stop(self, index: int) -> None:
"""Move the cursor to one stop — a feed row, or the ``Back`` row past them — and repaint."""
index = max(0, min(index, len(self._feed)))
self._selected = index if index < len(self._feed) else None
"""Move the cursor to one stop — the pin, a feed row, or ``Back`` — and repaint.
Args:
index: The stop to land on, clamped into range (see :attr:`_cursor`). Both ends
clamp rather than wrap, so a key held down settles at an end.
"""
if not self._feed: # nothing to pin or select: Back is the whole cursor space
self._pinned = False
self._selected = None
self._session.invalidate()
return
index = max(0, min(index, len(self._feed) + 1))
self._pinned = index == 0
if index <= len(self._feed):
self._selected = max(0, index - 1)
else:
self._selected = None
self._session.invalidate()
def _select_row(self, row: int) -> None:
"""Put the cursor on feed row ``row``, normally selected — never on the pin.
The way *something else* moves the highlight (the packet viewer paging through the
feed under it). Landing on a specific packet is by definition a selection and not a
pin, even when that packet happens to be the newest one: the reader is following
that packet, so the next arrival must push it down rather than steal the cursor.
"""
self._select_stop(row + 1)
def _commit(self) -> None:
"""Enter: open the highlighted packet, or leave when the cursor is on ``Back``."""
if self._selected is None:
@@ -338,17 +418,21 @@ class LiveFeedScreen(Screen):
def _open_packet(self) -> None:
"""Float the packet viewer over the highlighted feed row.
Paging inside the viewer walks the feed highlight in step (via
``on_navigate``), so closing it lands back on the packet last viewed.
Paging inside the viewer walks the feed highlight in step (via ``on_navigate``), so
closing it lands back on the packet last viewed — and opening one drops the pin for
the same reason. Enter names *this* packet; if the pin outlived it, the arrivals
during a long read would walk the cursor off it, and closing the viewer would land
somewhere the reader never chose.
"""
if self._selected is None or not self._feed:
return
self._select_row(self._selected)
def follow(entry: PacketEntry) -> None:
# The feed may have grown since the snapshot; find the entry itself.
for i, candidate in enumerate(self._feed):
if candidate is entry:
self._select_stop(i)
self._select_row(i)
return
viewer = PacketViewer(
@@ -377,6 +461,7 @@ class LiveFeedScreen(Screen):
self._selected = min(self._selected, len(self._feed) - 1) if self._feed else None
if self._selected is None:
self._hmax = 0 # nothing highlighted scrolls, so nothing advertises ←→
self._pinned = False # a feed with no rows has nothing to pin to
show_label = width >= _FEED_LABEL_MIN_WIDTH
lines = [render_to_ansi(self._heading(), width, no_wrap=True)]
if self._feed: # a header over nothing is noise; the empty note speaks for itself
@@ -396,7 +481,7 @@ class LiveFeedScreen(Screen):
brand row when it is the stop the cursor rests on.
"""
selected = self._selected is None
row = Text("❯ " if selected else " ", style="brand" if selected else "")
row = Text(_CURSOR if selected else " ", style="brand" if selected else "")
row.append("Back")
if selected:
row.style = "brand"
@@ -482,6 +567,11 @@ class LiveFeedScreen(Screen):
whole row, exactly as every select list draws its cursor. The lanes keep their own
colours over it — the SNR its quality hue, a name its palette hue — because those
colours are the content; the pointer and the base tint are what say "this row".
The one row that can wear a *different* pointer is the newest, which takes ``^``
while the cursor is pinned to the top of the stream rather than to the packet
currently sitting there (:data:`_PIN_CURSOR`, and the module docstring). Same lane,
same brand tint, same one cell — only the mark differs, because only the meaning
does.
A row that runs past the right edge — a terminal narrower than the lanes need —
is read by scrolling it, not by growing it: the *highlighted* row slides under
@@ -491,7 +581,8 @@ class LiveFeedScreen(Screen):
clamp it back into range.
"""
row = Text(no_wrap=True, overflow="ellipsis")
row.append("❯ " if selected else " ", style="brand" if selected else "")
cursor = _PIN_CURSOR if self._pinned else _CURSOR
row.append(cursor if selected else " ", style="brand" if selected else "")
avail = max(1, width - 2)
body = Text(no_wrap=True, overflow="ellipsis")
tests/test_livefeed.py
+75 -4
View File
@@ -280,6 +280,7 @@ def test_livefeed_column_header_is_pinned_and_carries_no_sort_cue() -> None:
def test_livefeed_highlight_uses_the_app_wide_cursor() -> None:
"""The feed marks its row like every other list: a ``❯`` pointer over a brand row."""
screen = _screen(seed=[_obs(node="n0", age_s=1), _obs(node="n1", age_s=0)])
screen.handle("down") # off the pin, onto the newest packet normally selected
rows = _rows(screen, 100)
assert rows[0].startswith("❯ ") and rows[1].startswith(" ")
assert "▸" not in "\n".join(rows) # the old odd-one-out mark is gone
@@ -287,6 +288,71 @@ def test_livefeed_highlight_uses_the_app_wide_cursor() -> None:
assert raw.startswith("\x1b[") # the pointer carries the brand style, not bare text
def test_livefeed_opens_pinned_and_the_pin_rides_the_newest_packet() -> None:
"""The top stop follows the stream: arrivals take the highlight with them.
A live feed's resting state is *watching*, so the screen opens on the pin — drawn
``^`` rather than ``❯`` — and every packet that lands keeps the cursor on the very
topmost row instead of being pushed down with the packet it was on.
"""
screen = _screen(seed=[_obs(node="n0", age_s=1)])
assert screen._pinned and screen._selected == 0
assert _rows(screen, 100)[0].startswith("^ ") # the pin's own mark
screen.on_event(MeshEvent.observation_event(_obs(node="a1b2")))
assert screen._selected == 0, "the pin let go of the top"
rows = _rows(screen, 100)
assert rows[0].startswith("^ ") and rows[1].startswith(" ")
assert "Alice" in rows[0] # …and it is the *newly arrived* packet under the cursor
def test_livefeed_down_off_the_pin_selects_the_newest_packet_itself() -> None:
"""``↓`` off the pin does not move a row — it changes what the cursor is attached to.
The two stops share the top line: the pin holds the position, a selection holds the
packet. So the first ``↓`` lands on that same newest packet under a plain ``❯``, and
only the *second* reaches the packet below it. ``↑`` re-pins.
"""
screen = _screen(seed=[_obs(node="n0", age_s=2), _obs(node="n1", age_s=1)])
screen.handle("down")
assert screen._selected == 0 and not screen._pinned, "↓ skipped past the newest packet"
assert _rows(screen, 100)[0].startswith("❯ ")
# Now the highlight belongs to that packet: an arrival pushes it down, as ever.
screen.on_event(MeshEvent.observation_event(_obs(node="a1b2")))
assert screen._selected == 1
screen.handle("down")
assert screen._selected == 2 and not screen._pinned # the next ↓ really does move
screen.handle("up")
screen.handle("up")
assert screen._selected == 0 and not screen._pinned # back on the newest packet…
screen.handle("up")
assert screen._pinned # …and one more ↑ re-pins
def test_livefeed_home_resumes_following_from_deep_in_the_history() -> None:
"""Home lands on the pin, not merely on whichever packet is newest right now."""
screen = _screen(seed=[_obs(node=f"n{i}", age_s=i) for i in range(10)])
screen.handle("end")
assert screen._selected is None and not screen._pinned
screen.handle("home")
assert screen._pinned and screen._selected == 0
screen.on_event(MeshEvent.observation_event(_obs(node="a1b2")))
assert screen._selected == 0 # following again, not parked on the old newest
def test_livefeed_opening_a_packet_drops_the_pin() -> None:
"""Enter names *this* packet, so arrivals during a long read can't walk off it."""
screen = _screen(seed=[_obs(node="n0", age_s=1)])
screen.future = _Fut()
assert screen._pinned
screen._select_row(0) # what _open_packet does before floating the viewer
assert not screen._pinned and screen._selected == 0
screen.on_event(MeshEvent.observation_event(_obs(node="a1b2")))
assert screen._selected == 1, "the cursor left the packet being viewed"
def test_livefeed_page_keys_move_the_feed_selection() -> None:
"""PgUp/PgDn walk the cursor a windowful at a time, clamped to the list's two ends."""
seed = [_obs(node=f"n{i}", age_s=i) for i in range(20)]
@@ -294,11 +360,13 @@ def test_livefeed_page_keys_move_the_feed_selection() -> None:
screen._feed_window.page = 5 # as if the last paint settled a five-row window
assert screen._selected == 0 # the newest packet is highlighted from the start
screen.handle("pagedown")
assert screen._selected == 5 # the selection travelled down a page, not just the view
# A page is five *stops*, and the pin is the first of them (see LiveFeedScreen._cursor),
# so paging off it lands on row 4 — the same five positions every other page travels.
assert screen._selected == 4 # the selection travelled down a page, not just the view
screen.handle("pageup")
assert screen._selected == 0
assert screen._selected == 0 and screen._pinned
screen.handle("pageup")
assert screen._selected == 0 # …and stops at the newest rather than wrapping
assert screen._selected == 0 # …and stops at the pin rather than wrapping
for _ in range(5): # paged past the oldest packet, the cursor lands on Back and stays
screen.handle("pagedown")
@@ -329,6 +397,7 @@ def test_livefeed_back_row_is_the_cursors_last_stop() -> None:
screen = _screen(seed=[_obs(node="n0", age_s=1), _obs(node="n1", age_s=0)])
screen.future = _Fut()
screen.handle("down") # off the pin, onto the newest
screen.handle("down")
screen.handle("down")
assert screen._selected is None # past the oldest packet: Back
@@ -391,7 +460,9 @@ def test_livefeed_highlighted_row_scrolls_sideways_to_its_tail() -> None:
screen.handle("right")
scrolled = _rows(screen, _CRAMPED)[0]
assert screen._hshift == screen._hmax
assert scrolled.startswith("❯ ") # the pointer lane stays pinned while the row slides
# The cursor lane holds its two cells while the row slides under it — here carrying
# the pin's ``^``, since a feed just opened is following the stream.
assert scrolled.startswith("^ ")
assert scrolled.rstrip().endswith("-90 dBm") # the tail is now readable
assert opening[2:10] not in scrolled # …at the cost of the time lane, slid off left