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

A channel guide lists the public channels heard on the mesh

Browse Source 
MeshTerm already stores every channel packet it hears, including those
on channels the device doesn't have. A packet names its channel only by
a one-byte hash, but a public channel's key is derived from its name, so
a candidate name can be tested: derive the key and check the packet's
MAC with it, exactly as the firmware does before decoding. The guide
tries the device's channels, Public, every #name mentioned in a stored
message, and every known region. On one desktop history that names 25
public channels.

A guessed name has to match at least two messages before the guide
shows it. The check compares three bytes, and with thousands of names
against thousands of packets roughly one false match is expected. The
candidates are grouped by their hash byte, so each packet only checks
the ~1/256 that could match; the naive loop took 40 s, this takes 1 s
on 31,000 messages, in a worker thread under a busy card, and the result
is kept for the session while its inputs are unchanged.

"Channel guide…" is the first row of "Add a channel" on the Channels
page, and it stays there when every slot is full. Channels not on the
device come first, and Enter adds one to the next free slot. Channels
the device already has follow with a ✓ the highlight doesn't walk, and a
closing line counts the messages no name matched. Each row shows the
message count, the age of the last message and seven days of activity.
`channels guide` lists the same channels on the command line.

Refs: #25
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-10-07 01:27:11 -04:00
1 parent 2d30e8e392
commit 73be9692b4
9 files changed
+881 -8
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/cli/README.md
+24
View File
@@ -1283,6 +1283,7 @@ the room).
| `import INDEX URL` | Import a `meshcore://channel/add` link. |
| `import --file FILE [--dry-run]` | Add the channels of a file from `export` to the device. |
| `export FILE` | Write each channel to a file, with its key, its send scope, and its mute. Another device can import the file. |
| `guide` | The public channels that MeshTerm heard, with the number of messages and the age of the last one. |
| `share INDEX` | Print the share link of a slot. |
| `clear INDEX --yes` | Clear a slot. This removes the channel from the device. |
| `scope INDEX [REGION] [--clear]` | Show or set the region into which the channel's messages are flooded. |
@@ -1391,6 +1392,29 @@ $ meshterm channels export channels.toml --json
{"path":null,"channels":0,"private":0}
```
**The guide.** `guide` lists each public channel that MeshTerm heard, the most recently
heard first. MeshTerm stores each channel packet that it hears, also on a channel that is
not on your device. A packet gives its channel only as a hash, so MeshTerm tries names:
the channels of your device, `Public`, each `#name` in a stored message, and each region
that it knows. The key of a public channel comes from its name, and MeshTerm checks the MAC
of each packet with that key, as the firmware does. A guessed name must match two messages
or more, because one match of three bytes is not proof. A private channel, or a public
channel whose name nobody wrote, has no name in the guide. stderr gives the number of
these messages.
```console
$ meshterm channels guide
NAME HASH MSGS LAST DEVICE
#news 03 2 now yes
#harbour 26 3 now no
```
`MSGS` counts different messages: the copies that repeaters relayed count one time.
`DEVICE` says whether the device has the channel. To add a channel from the guide, give
its name to `add` (`channels add 2 "#harbour"`). When MeshTerm heard no public channel,
`guide` exits with `5`. Under `--json`, each record is
`{"name":"#news","hash":"03","messages":2,"last_heard":"2026-10-07T05:23:09Z","on_device":true}`.
#### `meshterm courier`
A store-and-forward outbox. Queue a message for a contact that you cannot reach now. The
docs/guide/features.md
+1 -1
View File
@@ -16,7 +16,7 @@ This page lists each interactive screen with its scripted equivalent, where one
| Feature | What it does | Scripted |
| --- | --- | --- |
| **💬 Chat** | Live full-screen messages on channels and to contacts. The transcript scrolls, the input line stays at the bottom, and sent and received messages appear together. The rooms that you joined are also here. Each post shows under the name of its author. MeshTerm stores each message. Unread counts show in the menu header. | `meshterm chat send / history / list / listen` |
| **📻 Channels** | Create, join, reorder, mute, and share mesh channels. You can share a channel with a QR code or a `meshcore://` link. If you give a channel a **send scope**, its messages are flooded into one region. Only the repeaters that carry that region relay them. When you change to a new device, export all the channels to a file, then import the file on the new device. | `meshterm channels list / add / join / import / share / clear / scope / export` |
| **📻 Channels** | Create, join, reorder, mute, and share mesh channels. You can share a channel with a QR code or a `meshcore://` link. If you give a channel a **send scope**, its messages are flooded into one region. Only the repeaters that carry that region relay them. When you change to a new device, export all the channels to a file, then import the file on the new device. The **channel guide** lists the public channels that MeshTerm heard, also the channels that are not on your device, and adds one with Enter. | `meshterm channels list / add / join / import / share / clear / scope / export / guide` |
| **📌 Rooms** | Each room server that your radio knows. A room server is a message board on a radio. It keeps the latest posts and sends you the posts that you missed. Join a room with its password. If a room does not answer, MeshTerm tells you what it knows and offers the next step to try. For example, it can send again by flood when the route that your radio learned is no longer good. | `meshterm rooms list / join / forget` |
| **📨 Courier** | A store-and-forward outbox for contacts that you cannot reach yet. Queue a message. MeshTerm sends it when the contact is next heard, or at a time that you set. It tracks the acks, and it waits longer between each try (exponential backoff). | `meshterm courier queue / list / send / cancel / clear` |
| **👥 Contacts** | This node and its known contacts. The screen shows a heat map of how recently each contact was heard, the counts of overheard packets, and the full public keys. The hash of each key is highlighted. The scripted listing shows only the contacts. For this node, run `meshterm info`, which gives much more detail. | `meshterm contacts` |
meshterm/core/channel_guide.py
+248
View File
@@ -0,0 +1,248 @@
# SPDX-License-Identifier: Apache-2.0
"""The channel guide: the public channels that MeshTerm heard, and their traffic.
MeshTerm stores each channel packet that it hears, also on a channel for which it has no
key (the ``chan_hash``, ``cipher_mac``, and ``crypted`` columns of ``observations``). A
packet names its channel only by a one-byte hash, so the packet alone does not give the
name. But the key of a public channel comes from its name (``sha256("#name")``). Thus
MeshTerm can test a candidate name: it derives the key, and it checks the MAC of the packet
with that key, exactly as the firmware does before it decodes a packet. A name whose key
gives the MAC of the packet is the name of the channel.
The candidate names come from what MeshTerm already knows:
- The channels with a key that MeshTerm has: the slots of the device, and the ``Public``
channel of the firmware.
- Each ``#name`` that a stored message mentions.
- Each region that MeshTerm knows, as ``#region``.
A match of a guessed name on one message is not proof. The check compares three bytes (the
hash and the two-byte MAC). With thousands of names against thousands of packets,
approximately one false match is probable. Thus the guide names a guessed channel only when
:data:`MIN_MATCHES` messages or more match. A channel with a key that MeshTerm has needs one
match, because its name is not a guess.
The check is fast because the candidates are in groups by their hash byte. Each packet
checks only the candidates with its own hash, which is approximately 1/256 of them. A loop
over all the candidates for each packet took 40 s on a desktop for one history
(2026-10-06).
This module is pure logic. The caller reads the packets from the database and gives them
here. Thus the matching can run in a worker thread, and the database stays on its own
thread.
"""
from __future__ import annotations
import hmac
import re
from collections.abc import Iterable
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from .channels import (
DEFAULT_PUBLIC_SECRET,
channel_hash,
channel_identity,
derive_secret,
effective_secret,
is_public_channel,
)
#: The number of different messages that a *guessed* name must match before the guide
#: names the channel. Refer to the module docstring for the reason.
MIN_MATCHES = 2
#: The time window of the activity chart of each row: the last seven days.
GUIDE_WINDOW = timedelta(days=7)
#: The number of buckets in the window, newest first. Each bucket is six hours, and a
#: braille cell draws two buckets, so the full chart is 14 cells for 7 days.
GUIDE_BUCKETS = 28
#: A ``#name`` in the text of a message: a ``#`` that does not follow a letter, a digit, or
#: another ``#``, then 2 to 31 letters, digits, ``_``, or ``-``. Thus ``C#`` and ``##`` are
#: not names, and ``#10`` is one.
MENTION = re.compile(r"(?<![\w#])#[A-Za-z0-9][A-Za-z0-9_-]{1,30}")
def mention_names(texts: Iterable[str | None]) -> set[str]:
"""Each ``#name`` that the texts mention, as written and in lowercase.
The key comes from the exact bytes of the name, so ``#Ottawa`` and ``#ottawa`` are two
channels. A person who mentions a channel often writes it with a capital letter that
the channel does not have. Thus the function gives both forms.
"""
names: set[str] = set()
for text in texts:
for found in MENTION.findall(text or ""):
names.add(found)
names.add(found.lower())
return names
@dataclass(frozen=True, slots=True)
class HeardChannel:
"""One public channel that MeshTerm heard, with its traffic.
Attributes:
name: The name of the channel.
key: The 16-byte key that matched the MAC of its packets.
messages: The number of different messages heard (copies that repeaters relayed
count one time).
first_heard: When MeshTerm first heard a message of the channel.
last_heard: When MeshTerm last heard a message of the channel.
histogram: The messages in each bucket of :data:`GUIDE_WINDOW`, newest first.
"""
name: str
key: bytes
messages: int
first_heard: datetime | None
last_heard: datetime | None
histogram: tuple[int, ...]
@property
def identity(self) -> str:
"""The identity of the channel, the same value that a slot of the device gives."""
return channel_identity(self.name, self.key)
@property
def hash(self) -> str:
"""The channel hash (two hex digits) that the packets of the channel carry."""
return channel_hash(self.key)
@property
def write_secret(self) -> bytes | None:
"""The secret to send when the channel is added: ``None`` when the name gives it."""
return None if self.key == derive_secret(self.name) else self.key
@dataclass(frozen=True, slots=True)
class Guide:
"""What the guide found in the stored packets.
Attributes:
channels: The public channels that it named, the most recently heard first.
unnamed: The number of different messages that no name matched: messages on
private channels, or on public channels whose name nobody wrote.
"""
channels: tuple[HeardChannel, ...] = ()
unnamed: int = 0
@dataclass(slots=True)
class _Tally:
"""The counts of one candidate while the guide reads the packets."""
name: str
key: bytes
certain: bool
messages: int = 0
first: datetime | None = None
last: datetime | None = None
histogram: list[int] = field(default_factory=lambda: [0] * GUIDE_BUCKETS)
def build_guide(
packets: Iterable[tuple[str, str, str, str | None, str | None]],
known: Iterable[tuple[str, bytes]],
guesses: Iterable[str],
*,
now: datetime,
) -> Guide:
"""Name the channels of the stored packets, and count their traffic.
Args:
packets: One row for each different message: its channel hash, its MAC, and its
ciphertext (all in hex), then when it was first and last heard (ISO 8601).
known: The channels with a key that MeshTerm has, as ``(name, key)``. The
``Public`` channel of the firmware is always added. A private channel here
names its own packets, so that they do not count as unnamed, but the guide does
not show it, because it is not public.
guesses: The candidate names (``#name``) to test.
now: The present time, for the activity chart.
Returns:
The guide.
"""
tallies: dict[str, _Tally] = {}
for name, secret in [("Public", DEFAULT_PUBLIC_SECRET), *known]:
key = effective_secret(name, secret)
tallies.setdefault(channel_identity(name, key), _Tally(name, key, certain=True))
for name in guesses:
key = derive_secret(name)
tallies.setdefault(channel_identity(name, key), _Tally(name, key, certain=False))
by_hash: dict[str, list[_Tally]] = {}
for tally in tallies.values(): # the known channels first, in each group
by_hash.setdefault(channel_hash(tally.key), []).append(tally)
bucket_span = GUIDE_WINDOW / GUIDE_BUCKETS
unnamed = 0
for chan_hash, cipher_mac, crypted, first_at, last_at in packets:
tally = _match(by_hash.get((chan_hash or "").lower(), ()), cipher_mac, crypted)
if tally is None:
unnamed += 1
continue
first, last = _when(first_at), _when(last_at)
tally.messages += 1
if first is not None:
tally.first = first if tally.first is None else min(tally.first, first)
age = now - first
if timedelta(0) <= age < GUIDE_WINDOW:
tally.histogram[int(age / bucket_span)] += 1
if last is not None:
tally.last = last if tally.last is None else max(tally.last, last)
channels: list[HeardChannel] = []
for tally in tallies.values():
if tally.messages < (1 if tally.certain else MIN_MATCHES):
unnamed += tally.messages # too few to name it, so it stays unnamed
continue
if not is_public_channel(tally.name, tally.key):
continue # a private channel of the device: named, but not for the guide
channels.append(
HeardChannel(
name=tally.name,
key=tally.key,
messages=tally.messages,
first_heard=tally.first,
last_heard=tally.last,
histogram=tuple(tally.histogram),
)
)
epoch = datetime.min.replace(tzinfo=timezone.utc)
channels.sort(key=lambda c: (c.last_heard or epoch, c.messages), reverse=True)
return Guide(channels=tuple(channels), unnamed=unnamed)
def _match(candidates: Iterable[_Tally], cipher_mac: str, crypted: str) -> _Tally | None:
"""The candidate whose key gives the MAC of the packet, or ``None``.
This is the check of :func:`~meshterm.core.channels.identify_channel`, with the
``hmac`` module of the standard library. That module does the whole HMAC in C, and it
needs no import of PyCryptodome on the path of the guide.
"""
try:
mac = bytes.fromhex(cipher_mac or "")
body = bytes.fromhex(crypted or "")
except ValueError:
return None
if not mac or not body:
return None
for tally in candidates:
if hmac.digest(tally.key, body, "sha256")[: len(mac)] == mac:
return tally
return None
def _when(text: str | None) -> datetime | None:
"""Parse a stored time. A time with no zone is UTC, as MeshTerm stores it."""
if not text:
return None
try:
when = datetime.fromisoformat(text)
except ValueError:
return None
return when if when.tzinfo is not None else when.replace(tzinfo=timezone.utc)
meshterm/persistence/repository.py
+33
View File
@@ -2335,6 +2335,39 @@ class Repository:
)
return stats
def channel_packets(self) -> list[tuple[str, str, str, str | None, str | None]]:
"""Each different channel packet that MeshTerm stored, for the channel guide.
A repeater relays a message, so MeshTerm often hears one message several times. The
copies have the same hash, MAC, and ciphertext, and the query groups them into one
row. Thus each row is one message, with the time when MeshTerm first and last heard
it. A packet with no MAC or no ciphertext cannot be checked against a key, and the
query does not return it.
Returns:
``(chan_hash, cipher_mac, crypted, first_heard, last_heard)`` for each message.
The times are the stored ISO 8601 text.
"""
rows = self._conn.execute(
"SELECT chan_hash, cipher_mac, crypted, "
"MIN(observed_at) AS first_at, MAX(observed_at) AS last_at FROM observations "
"WHERE chan_hash IS NOT NULL AND cipher_mac IS NOT NULL AND crypted IS NOT NULL "
"GROUP BY chan_hash, cipher_mac, crypted"
).fetchall()
return [
(r["chan_hash"], r["cipher_mac"], r["crypted"], r["first_at"], r["last_at"])
for r in rows
]
def message_texts(self) -> list[str]:
"""The text of each stored chat message, for the names that the channel guide tests.
Returns:
The texts, in no specific order. An empty text is not returned.
"""
rows = self._conn.execute("SELECT text FROM messages WHERE text IS NOT NULL AND text != ''")
return [row["text"] for row in rows]
def backfill_channel_ids(self, mapping: dict[int, str]) -> int:
"""Give old channel messages an identity, keyed by the slot on which they were stored.
meshterm/tools/channels.py
+62 -1
View File
@@ -6,7 +6,7 @@ In the menu, the tool opens a full-screen channel manager (refer to
With it, the user can create a private channel, add a public ``#`` channel, join a channel
with a key, import a scanned ``meshcore://`` link, and share any channel as a QR code. On
the CLI, the tool gives the ``list``, ``add``, ``join``, ``import``, ``share``, ``clear``,
``scope``, and ``export`` subcommands for use in scripts. ``export`` writes all the
``scope``, ``export``, and ``guide`` subcommands for use in scripts. ``export`` writes all the
channels to a file, and ``import --file`` adds the channels of that file to another device
(refer to :mod:`meshterm.core.channel_file`).
@@ -92,6 +92,8 @@ class ChannelsTool(Tool):
return await self._cli_export(ctx, params)
if action == "import_file":
return await self._cli_import_file(ctx, params)
if action == "guide":
return await self._cli_guide(ctx)
return await self._cli_list(ctx)
async def _cli_list(self, ctx: AppContext) -> ToolResult:
@@ -338,6 +340,30 @@ class ChannelsTool(Tool):
report=_imported(plan, dry_run),
)
async def _cli_guide(self, ctx: AppContext) -> ToolResult:
"""List the public channels that MeshTerm heard, as the channel guide of the menu does.
One record for each channel, the most recently heard first: its name, its hash, the
number of different messages, the age of the last one, and whether the device has
the channel. The count of the messages that no name matched goes to stderr, because
it is a note about the listing and not a record in it. No public channel heard is
exit 5.
"""
from ..ui.channel_guide import read_guide
slots = list(await ctx.devstate.channel_slots())
guide = await read_guide(ctx, slots)
on_device = {slot.identity for slot in slots}
if guide.unnamed:
ctx.ui.ack(
f"[muted]{guide.unnamed} more messages on channels with no known name[/muted]"
)
return ToolResult(
summary={"channels": len(guide.channels), "unnamed": guide.unnamed},
report=(_guide_listing(guide, on_device),),
exit_code=exitcodes.OK if guide.channels else exitcodes.NO_RESULT,
)
@staticmethod
async def _show_qr(ctx: AppContext, name: str, url: str) -> None:
"""Draw the share link of a channel as a QR code, only in the menu.
@@ -416,6 +442,10 @@ class ChannelsTool(Tool):
raise typer.BadParameter("--dry-run works only with --file.")
run_tool_command(self, {"cli_action": "import", "index": index, "url": url})
@channels_app.command("guide", help="List the public channels heard on the mesh")
def _guide_cmd() -> None:
run_tool_command(self, {"cli_action": "guide"})
@channels_app.command("export", help="Write every channel to a file, for another device")
def _export_cmd(
path: Path = typer.Argument(..., help="Destination file"),
@@ -491,6 +521,37 @@ async def _complete_slots(ctx: AppContext) -> tuple[Any, list]:
return device, slots
def _guide_listing(guide: Any, on_device: set[str]) -> Listing:
"""The channels of the guide, one record for each channel.
``DEVICE`` says whether the device has the channel. ``LAST`` is an age, as each time in
a listing is. The document has the time itself, in UTC.
"""
from ..ui import fields
from ..ui.report import Listing
return Listing(
key="channels",
columns=(
fields.name("name", "NAME"),
fields.hexid("hash", "HASH"),
fields.integer("messages", "MSGS"),
fields.when("last_heard", "LAST"),
fields.flag("on_device", "DEVICE"),
),
rows=[
{
"name": channel.name,
"hash": channel.hash,
"messages": channel.messages,
"last_heard": channel.last_heard,
"on_device": channel.identity in on_device,
}
for channel in guide.channels
],
)
def _exported(path: Path | None, channels: int, private: int) -> Facts:
"""What ``channels export`` wrote. The plain face prints only the path."""
from ..ui import fields
meshterm/ui/channel_guide.py
+219
View File
@@ -0,0 +1,219 @@
# SPDX-License-Identifier: Apache-2.0
"""The channel guide screen: the public channels heard on the mesh, and an add for each.
The guide opens from the "Add a channel" section of the Channels page. It lists each public
channel that MeshTerm heard (refer to :mod:`meshterm.core.channel_guide`), in two sections.
The channels that are not on the device come first, because the user can add them: Enter
adds the highlighted channel to the next free slot at once, as the other add rows do. The
channels that the device already has come after them, each with a ``✓`` in the pointer
column. The highlight does not walk these rows, so the list is complete and Enter never does
nothing. A muted line at the end counts the messages that no name matched.
Each row shows the name, the number of different messages, the age of the last message, and
the activity of the last seven days. The chart is the last lane, and it becomes shorter to
fit the row, as on the Channels page.
The match of the stored packets takes a moment on a large history (one second on a desktop
for 31,000 messages, more on a handheld). Thus it runs in a worker thread under a busy card,
and the result stays for the session while the history and the candidate names are the same.
"""
from __future__ import annotations
import asyncio
from collections.abc import Callable
from datetime import datetime, timezone
from typing import TYPE_CHECKING
from rich.cells import cell_len
from rich.text import Text
from ..core.channel_guide import GUIDE_BUCKETS, Guide, HeardChannel, build_guide, mention_names
from ..core.channel_probe import ChannelSlot
from .braillechart import activity_peak, activity_sparkline
from .channels import _pick_free_slot, write_channel
from .menus import Lane, column_header, fit_cells, section_heading
from .tui import CANCEL, Choice, SelectScreen, Separator
from .widgets import age_seconds, channel_glyph, format_age
if TYPE_CHECKING:
from ..context import AppContext
from ..core.connection import Device
#: The footer of the guide: navigation, the filter, the action, and Esc last.
_GUIDE_HINT = "↑↓ move · type to filter · Enter add · Esc back"
#: The widest name lane. A longer name ends in an ellipsis, as on the Channels page.
_NAME_WIDTH_MAX = 20
#: The width of the message-count lane, which is aligned to the right.
_COUNT_WIDTH = 5
#: The width of the age lane, which is aligned to the right. It fits ``never``.
_AGE_WIDTH = 5
#: The cells of the activity chart at full width: two buckets of six hours for each cell.
_CHART_CELLS = GUIDE_BUCKETS // 2
#: The lower limit of the shared chart peak, so that one message stays a small mark.
_ACTIVITY_FLOOR = 2.0
#: The last guide that was built, and what it was built from. Refer to :func:`read_guide`.
_cache: tuple[tuple, Guide] | None = None
async def channel_guide(ctx: AppContext, device: Device, capacity: int) -> int:
"""Show the guide until the user leaves it. Return the number of channels added.
Args:
ctx: The shared application context.
device: The connected device, to add a channel to.
capacity: The number of channel slots of the device.
Returns:
The number of channels that the user added from the guide.
"""
async with ctx.ui.busy_dialog("reading the channels heard…", title="Channel guide"):
slots = list(await ctx.devstate.channel_slots())
guide = await read_guide(ctx, slots)
by_identity = {channel.identity: channel for channel in guide.channels}
added = 0
async def add(choice: object) -> bool:
"""Add the selected channel. Return whether the device changed."""
nonlocal slots, added
channel = by_identity.get(choice) if isinstance(choice, str) else None
if channel is None:
return False
idx = await _pick_free_slot(ctx, device, slots, capacity)
if idx is None:
return False
await write_channel(ctx, device, idx, channel.name, channel.write_secret)
slots = list(await ctx.devstate.channel_slots())
added += 1
return True
title, items = guide_items(guide, slots)
session = getattr(ctx.ui, "session", None)
if session is None: # a UI with no stack of screens: one round at a time
while True:
choice = await ctx.ui.select(title, items)
if choice is None:
return added
if await add(choice):
title, items = guide_items(guide, slots)
screen = SelectScreen(title, items, footer_hint=_GUIDE_HINT)
async with session.stay(screen) as visit:
while True:
choice = await visit.result()
if choice is CANCEL:
return added
if await add(choice):
# The rows are data, so the list changes in place: the added channel moves
# to the second section, and the filter stays.
title, items = guide_items(guide, slots)
screen.replace_items(items, title=title)
async def read_guide(ctx: AppContext, slots: list[ChannelSlot]) -> Guide:
"""Build the guide from the stored packets, or return the guide that is still correct.
The two reads of the database run here, on the thread of the database. Only the match,
which reads no database, goes to a worker thread. The guide stays for the session while
its inputs are the same: the same number of stored messages with the same latest time,
the same candidate names, and the same channels on the device.
"""
global _cache
packets = ctx.repo.channel_packets()
regions = ctx.region_store.names() if ctx.region_store is not None else []
guesses = mention_names(ctx.repo.message_texts()) | {f"#{name}" for name in regions}
known = [(slot.name, slot.secret) for slot in slots]
latest = max((row[4] or "" for row in packets), default="")
key = (len(packets), latest, frozenset(guesses), tuple(slot.identity for slot in slots))
if _cache is not None and _cache[0] == key:
return _cache[1]
now = datetime.now(timezone.utc)
guide = await asyncio.to_thread(build_guide, packets, known, guesses, now=now)
_cache = (key, guide)
return guide
def guide_items(guide: Guide, slots: list[ChannelSlot]) -> tuple[str, list]:
"""The title and the rows of the guide, for the channels that are on the device now.
Args:
guide: What the guide found.
slots: The channels of the device.
Returns:
``(title, items)`` for a :class:`SelectScreen`.
"""
on_device = {slot.identity for slot in slots}
absent = [c for c in guide.channels if c.identity not in on_device]
present = [c for c in guide.channels if c.identity in on_device]
title = f"Channel guide · {len(guide.channels)} heard"
items: list = []
if not guide.channels:
items.append(Separator(" no public channel heard yet — keep MeshTerm listening"))
else:
name_w = min(
_NAME_WIDTH_MAX, max(len("CHANNEL"), *(cell_len(c.name) for c in guide.channels))
)
peak = activity_peak(*(c.histogram for c in guide.channels), floor=_ACTIVITY_FLOOR)
items.append(Separator(lambda w: _header(name_w, w), heading=True))
items.append(section_heading("Not on this device"))
if not absent:
items.append(Separator(" each public channel heard is on this device"))
for channel in absent:
row = _row(channel, name_w, peak)
items.append(Choice(title=row, value=channel.identity, fitted=True))
if present:
items.append(section_heading("On this device"))
for channel in present:
items.append(Separator(_row(channel, name_w, peak, mark="✓")))
if guide.unnamed:
count = f"{guide.unnamed} message" if guide.unnamed == 1 else f"{guide.unnamed} messages"
items.append(Separator(" "))
items.append(Separator(f" {count} on channels with no known name"))
return title, items
def _header(name_w: int, width: int) -> str:
"""The column headers over the lanes of the guide (refer to :func:`_row_text`)."""
return column_header(
[
Lane("CHANNEL", cell_len(channel_glyph("#x", None)) + 1 + name_w + 2),
Lane(f"{'MSGS':>{_COUNT_WIDTH}}", _COUNT_WIDTH + 2),
Lane(f"{'LAST':>{_AGE_WIDTH}}", _AGE_WIDTH + 2),
Lane(("ACTIVITY", "ACT", "")), # it goes with its chart
],
width,
)
def _row(
channel: HeardChannel, name_w: int, peak: float, *, mark: str = ""
) -> Callable[[int], Text]:
"""A row title that the screen renders again at each paint, at the width of the row."""
return lambda width: _row_text(channel, name_w, peak, width, mark=mark)
def _row_text(channel: HeardChannel, name_w: int, peak: float, width: int, *, mark: str) -> Text:
"""Build one row of the guide, ``width`` cells at most.
A row that the highlight walks starts at the name, after the pointer column that the
screen draws. A row of a channel that is on the device draws that column itself: two
cells, with ``mark`` (``✓``) in the first one. Thus the two kinds of row align.
"""
text = Text(no_wrap=True, overflow="ellipsis")
if mark:
text.append(mark, style="ok")
text.append(" " * (2 - cell_len(mark)))
text.append(f"{channel_glyph(channel.name, channel.key)} ")
text.append(fit_cells(channel.name, name_w), style="muted" if mark else "")
text.append(" ")
text.append(f"{min(channel.messages, 99999):>{_COUNT_WIDTH}}")
text.append(" ")
age = format_age(age_seconds(channel.last_heard)) if channel.last_heard else "never"
text.append(f"{age:>{_AGE_WIDTH}}", style="muted")
cells = min(_CHART_CELLS, width - text.cell_len - 2)
if cells > 0:
text.append(" ")
text.append_text(activity_sparkline(channel.histogram, 2 * cells, peak=peak))
return text
meshterm/ui/channels.py
+17 -6
View File
@@ -10,6 +10,9 @@ comes from the name), joining with a key that the user pastes, or importing a sc
``meshcore://`` link. Each change is written to the device at once, as in a phone app.
Thus the list always shows the state of the radio.
The channel guide (:mod:`meshterm.ui.channel_guide`) is the first row of "Add a channel". It
lists the public channels that MeshTerm heard, and it adds one with Enter.
The section "Export and import" moves all the channels to another device. The export writes
each channel, with its key, its send scope, and its mute, to one TOML file. The import adds
the channels of such a file to the device, in the order of the file, and it never removes a
@@ -121,6 +124,7 @@ _IMPORT = "__import__"
_REORDER = "__reorder__"
_EXPORT_FILE = "__export_file__"
_IMPORT_FILE = "__import_file__"
_GUIDE = "__guide__"
# The sentinels of the actions of the channel detail screen.
_QR = "qr"
@@ -243,6 +247,10 @@ async def manage_channels(ctx: AppContext) -> int:
await _export_channels(ctx, slots) # it writes a file, and the device stays the same
elif choice == _IMPORT_FILE:
changes += await _import_channels(ctx, device, capacity)
elif choice == _GUIDE:
from .channel_guide import channel_guide # it imports this module
changes += await channel_guide(ctx, device, capacity)
else: # an existing slot index
slot = next((s for s in slots if s.idx == choice), None)
if slot is not None:
@@ -767,7 +775,7 @@ def _menu_items(
# cells (＋ ＃ 🌐 🔑 🔗), and not one column early. The Organize row once corrected this
# by hand, with two spaces and a comment of four lines. The column is empty where the
# platform draws no icons, and the labels use the cells.
lane = icon_lane(("↕", "💾", "📂", "🌐", "＋", "＃", "🔑", "🔗"))
lane = icon_lane(("↕", "💾", "📂", "⚡", "🌐", "＋", "＃", "🔑", "🔗"))
if len(slots) > 1:
items.append(section_heading("Organize"))
reorder = marked_label("↕", "Reorder channels", "", lane=lane)
@@ -782,8 +790,12 @@ def _menu_items(
import_file = marked_label("📂", "Import channels…", "", lane=lane)
transfer.append((import_file, "Add them from a file", _IMPORT_FILE))
adds = []
if _next_free_slot(slots, capacity) is not None:
# The guide is first, and it stays when each slot is full: its list of the channels heard
# is worth a read then too, and an add from it says why it cannot add.
guide = marked_label("⚡", "Channel guide…", "", lane=lane)
adds = [(guide, "Public channels heard on the mesh", _GUIDE)]
free = _next_free_slot(slots, capacity) is not None
if free:
# The Public channel of the firmware has a fixed key. It has one well-known secret, so
# it is the same channel on each slot. Offer to restore it only if no slot has it
# already.
@@ -825,9 +837,8 @@ def _menu_items(
items.append(section_heading("Export and import"))
items.extend(aligned[: len(transfer)])
items.append(section_heading("Add a channel"))
if adds:
items.extend(aligned[len(transfer) :])
else:
items.extend(aligned[len(transfer) :])
if not free:
# The list already knows that there is no free slot. It says so here, and it does not
# offer four rows that all end in the same refusal.
items.append(Separator(" every slot is full — clear one first"))
tests/test_channel_guide.py
+242
View File
@@ -0,0 +1,242 @@
# SPDX-License-Identifier: Apache-2.0
"""Tests for the channel guide: the candidate names, the match, the screen, and the CLI.
Each packet here is built as the firmware builds it (PyCryptodome, as the decryption tests
do), so that the match of the guide, which uses the ``hmac`` module, is checked against the
same MAC that the firmware writes. The channel names are invented.
"""
from __future__ import annotations
import io
from contextlib import asynccontextmanager
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace
from rich.console import Console
from meshterm.context import AppContext
from meshterm.core import exitcodes
from meshterm.core.admin_store import AdminStore
from meshterm.core.channel_guide import (
GUIDE_BUCKETS,
GUIDE_WINDOW,
MIN_MATCHES,
build_guide,
mention_names,
)
from meshterm.core.channel_probe import read_channel_slots
from meshterm.core.channels import (
DEFAULT_PUBLIC_SECRET,
channel_hash,
derive_secret,
identify_channel,
)
from meshterm.core.config import Settings
from meshterm.core.device_store import DeviceStore
from meshterm.core.models import Observation
from meshterm.persistence.repository import Repository
from meshterm.tools.channels import ChannelsTool
from meshterm.ui import channel_guide as guide_screen
_NOW = datetime(2026, 10, 7, 12, 0, tzinfo=timezone.utc)
def _packet(name: str, text: str, *, key: bytes | None = None) -> tuple[str, str, str]:
"""One channel packet as the firmware builds it: ``(chan_hash, cipher_mac, crypted)``."""
from Crypto.Cipher import AES
from Crypto.Hash import HMAC, SHA256
key = key or derive_secret(name)
plain = (1_700_000_000).to_bytes(4, "little") + b"\x00" + text.encode("utf-8")
plain += b"\x00" * (-len(plain) % 16)
crypted = AES.new(key, AES.MODE_ECB).encrypt(plain)
mac = HMAC.new(key, digestmod=SHA256)
mac.update(crypted)
return channel_hash(key), mac.digest()[:2].hex(), crypted.hex()
def _row(name: str, text: str, *, hours: float = 1, key: bytes | None = None) -> tuple:
"""A row as :meth:`Repository.channel_packets` gives it, heard ``hours`` before now."""
when = (_NOW - timedelta(hours=hours)).isoformat()
return (*_packet(name, text, key=key), when, when)
# -- the names --------------------------------------------------------------------------
def test_mentions_give_each_name_as_written_and_in_lowercase() -> None:
"""A ``#name`` gives two candidates, because the key comes from the exact bytes."""
names = mention_names(["join #Harbour or #ops-2, not C# or ##x", None, "net #10"])
assert names == {"#Harbour", "#harbour", "#ops-2", "#10"}
def test_the_match_of_the_guide_is_the_mac_that_the_firmware_writes() -> None:
"""The ``hmac`` check names the same channel as :func:`identify_channel` does."""
packet = _packet("#harbour", "hello")
assert identify_channel(*packet, [("#harbour", derive_secret("#harbour"))]) is not None
rows = [(*packet, None, None), (*_packet("#harbour", "again"), None, None)]
guide = build_guide(rows, [], {"#harbour"}, now=_NOW)
assert [c.name for c in guide.channels] == ["#harbour"]
# -- the guide --------------------------------------------------------------------------
def test_a_guessed_name_needs_two_messages() -> None:
"""One match of a guessed name is not proof, so the message stays unnamed."""
assert MIN_MATCHES == 2
rows = [_row("#harbour", "one"), _row("#ops", "one"), _row("#ops", "two")]
guide = build_guide(rows, [], {"#harbour", "#ops"}, now=_NOW)
assert [c.name for c in guide.channels] == ["#ops"]
assert guide.unnamed == 1
def test_a_channel_with_a_known_key_needs_one_message() -> None:
"""A channel of the device is not a guess. Public is always known."""
rows = [_row("Public", "hi", key=DEFAULT_PUBLIC_SECRET), _row("#harbour", "hi")]
guide = build_guide(rows, [("#harbour", derive_secret("#harbour"))], set(), now=_NOW)
assert sorted(c.name for c in guide.channels) == ["#harbour", "Public"]
assert guide.unnamed == 0
def test_a_private_channel_names_its_packets_but_is_not_in_the_guide() -> None:
"""The guide lists public channels. A private channel of the device is not unnamed."""
family = bytes(range(16))
rows = [_row("Family", "hi", key=family), _row("#elsewhere", "hi")]
guide = build_guide(rows, [("Family", family)], set(), now=_NOW)
assert guide.channels == ()
assert guide.unnamed == 1 # only the message on the channel that no key matched
def test_the_guide_counts_and_charts_each_channel() -> None:
"""The counts, the times, and the chart of a channel, the most recent channel first."""
rows = [
_row("#ops", "a", hours=1),
_row("#ops", "b", hours=7),
_row("#ops", "c", hours=24 * 30), # outside the chart, still counted
_row("#harbour", "a", hours=0.5),
_row("#harbour", "b", hours=2),
]
guide = build_guide(rows, [], {"#ops", "#harbour"}, now=_NOW)
harbour, ops = guide.channels
assert (harbour.name, ops.name) == ("#harbour", "#ops") # the most recent first
assert ops.messages == 3
assert ops.last_heard == _NOW - timedelta(hours=1)
assert ops.first_heard == _NOW - timedelta(days=30)
span = GUIDE_WINDOW / GUIDE_BUCKETS # six hours
assert len(ops.histogram) == GUIDE_BUCKETS
assert ops.histogram[0] == 1 and ops.histogram[int(timedelta(hours=7) / span)] == 1
assert sum(ops.histogram) == 2
def test_the_repository_gives_one_row_for_each_message(tmp_path: Path) -> None:
"""Copies that repeaters relayed are one message, heard first and last at two times."""
repo = Repository(tmp_path / "chan.db")
try:
run = repo.start_run("monitor", {}, None)
chan_hash, mac, crypted = _packet("#harbour", "hi")
raw = {"payload_typename": "GRP_TXT", "chan_hash": chan_hash}
raw |= {"cipher_mac": mac, "crypted": crypted}
for _ in range(3):
repo.record_observation(run, Observation(node="src", kind="packet", raw=raw))
rows = repo.channel_packets()
assert len(rows) == 1 and rows[0][:3] == (chan_hash, mac, crypted)
finally:
repo.close()
# -- the screen and the CLI -------------------------------------------------------------
def _ctx(config_dir: Path) -> AppContext:
"""A context on the mock device, with its own configuration directory and database."""
config_dir.mkdir(parents=True, exist_ok=True)
settings = Settings(config_dir=config_dir, db_path=config_dir / "chan.db")
return AppContext(
console=Console(file=io.StringIO()),
settings=settings,
repo=Repository(settings.db_path),
device_store=DeviceStore(config_dir / "devices.json"),
admin_store=AdminStore(config_dir / "admin.json"),
mock=True,
)
def _hear(ctx: AppContext, name: str, *texts: str) -> None:
"""Store one packet for each text, and one message that mentions the channel."""
run = ctx.repo.start_run("monitor", {}, None)
for text in texts:
chan_hash, mac, crypted = _packet(name, text)
raw = {"payload_typename": "GRP_TXT", "chan_hash": chan_hash}
raw |= {"cipher_mac": mac, "crypted": crypted}
ctx.repo.record_observation(run, Observation(node="src", kind="packet", raw=raw))
from meshterm.core.models import ChatMessage
ctx.repo.record_chat_message(ChatMessage(text=f"Alice: try {name}", is_channel=True))
class _ScriptedUi:
"""A UI that replays the selects, to drive the guide with no screen."""
def __init__(self, selects: list) -> None:
self._selects = list(selects)
self.titles: list[str] = []
@asynccontextmanager
async def busy_dialog(self, message: str = "", *, title: str = ""): # noqa: ANN201
yield SimpleNamespace(message=message)
def note(self, markup: str) -> None:
pass
async def select(self, title: str, items: list, *, default=None): # noqa: ANN001, ANN201
self.titles.append(title)
return self._selects.pop(0)
async def test_enter_on_a_heard_channel_adds_it_to_the_device(tmp_path: Path) -> None:
"""The guide adds the highlighted channel to the next free slot, with its derived key."""
ctx = _ctx(tmp_path / "home")
guide_screen._cache = None
try:
_hear(ctx, "#harbour", "one", "two")
target = derive_secret("#harbour")
identity = next(
c.identity for c in (await guide_screen.read_guide(ctx, [])).channels if c.key == target
)
ctx.ui = _ScriptedUi(selects=[identity, None])
added = await guide_screen.channel_guide(ctx, await ctx.device(), 8)
assert added == 1
slots = await read_channel_slots(await ctx.device())
assert [s.name for s in slots] == ["#harbour"] and slots[0].identity == identity
finally:
ctx.repo.close()
async def test_cli_guide_lists_the_heard_channels(tmp_path: Path) -> None:
"""``channels guide`` gives one record for each public channel heard."""
ctx = _ctx(tmp_path / "home")
guide_screen._cache = None
try:
_hear(ctx, "#harbour", "one", "two")
result = await ChannelsTool().run(ctx, {"cli_action": "guide"})
assert result.summary == {"channels": 1, "unnamed": 0}
(listing,) = result.report
assert [row["name"] for row in listing.rows] == ["#harbour"]
assert listing.rows[0]["messages"] == 2 and listing.rows[0]["on_device"] is False
finally:
ctx.repo.close()
async def test_cli_guide_with_nothing_heard_is_exit_5(tmp_path: Path) -> None:
"""No public channel heard is nothing to report."""
ctx = _ctx(tmp_path / "home")
guide_screen._cache = None
try:
result = await ChannelsTool().run(ctx, {"cli_action": "guide"})
assert result.exit_code == exitcodes.NO_RESULT
finally:
ctx.repo.close()
tests/test_gallery.py
+35
View File
@@ -877,6 +877,40 @@ def _channels_manager(cols: int, rows: int) -> Screen:
return SelectScreen(title, items, footer_hint=_MANAGER_HINT)
def _channel_guide(cols: int, rows: int) -> Screen:
"""The channel guide when it is widest: a name at the maximum of its lane, two sections.
One channel is on the device, so the second section and its ``✓`` rows are drawn. One
channel was heard long ago, so its chart is flat, and the closing note counts the
messages that no name matched.
"""
from datetime import timedelta
from meshterm.core.channel_guide import Guide, HeardChannel
from meshterm.core.channel_probe import ChannelSlot
from meshterm.ui.channel_guide import _GUIDE_HINT, guide_items
now = datetime.now(timezone.utc)
def heard(name: str, messages: int, hours: float, busy: bool = True) -> HeardChannel:
histogram = tuple((i * 3 + messages) % 4 if busy else 0 for i in range(28))
last = now - timedelta(hours=hours)
return HeardChannel(name, derive_secret(name), messages, last, last, histogram)
guide = Guide(
channels=(
heard("#harbour", 7578, 0.1),
heard("#lakeside-watchkeeper", 49, 2),
heard("#ops", 129, 30),
heard("#old-net", 3, 2000, busy=False),
),
unnamed=3438,
)
slots = [ChannelSlot(idx=0, name="#harbour", secret=derive_secret("#harbour"))]
title, items = guide_items(guide, slots)
return SelectScreen(title, items, footer_hint=_GUIDE_HINT)
def _channel_detail(cols: int, rows: int, *, scope: str | None = None) -> Screen:
"""The action page of one channel, with its summary line above the rows."""
from meshterm.core.channel_probe import ChannelSlot
@@ -1710,6 +1744,7 @@ _ENTRIES: list[_Entry] = [
_Entry("room_page", _room_page),
_Entry("room_login_card", _room_login_card),
_Entry("channels_manager", _channels_manager),
_Entry("channel_guide", _channel_guide),
_Entry("channel_detail", _channel_detail),
_Entry("channel_detail_scoped", _channel_detail_scoped),
_Entry("scope_picker", _scope_picker),