How MeshTerm behaves gets a page of its own

Three kinds of "setting" were tangled into two files and one of them had no
screen at all. The radio's settings live on the radio and are edited on Device
config. Machine setup -- where the database is, which companion to talk to --
stays in config.toml, a text editor's job. Everything in between was app
behaviour: six keys sitting in config.toml with nothing to open them, and a
dozen more that were only module constants a reader could never reach.

Those are preferences now, and they have a page. core/preferences.py is a
PrefSpec registry in the shape of the device-config one -- group, type, bounds,
help, and a default -- and the page is built from the same parts as the device
editor, deliberately: one grouped lane-aligned list, current -> new staged in
warn, Apply/Back appearing only once something is staged, the shared discard
confirm on the way out. Two editors that stage values should not be two screens
to learn. What it adds is a Defaults section holding one row that stages every
changed preference back to its built-in value; it is drawn only while a reset
would do something, and it stages rather than writes, so it is as reviewable and
as Esc-discardable as any other change and the file is still only ever written
by the one action at the bottom. Its rows scroll sideways where the device
editor's don't: the explanation is the widest thing on the row and the only part
that runs past the edge, so <-> slide it under a pinned setting and value.

The file is preferences.yaml and holds only the values you disagree with the
code about -- grouped and commented, each entry above its help and its default.
That is what lets a later change to a default reach an install that never asked
to be exempt; delete a line and the code takes over again. A hand edit gone
wrong costs that line, never the session.

Every preference is wired to something that reads it. ctx.preferences is the
explicit path; preferences.current() is for the render layer, which has no
context to reach through, the same reason get_platform() exists.

The About pages come along into a This app section, the third scope after this
node and other nodes, with Preferences leading it -- the one row there that
changes MeshTerm rather than describing it. A section of one would have broken
the menu's own no-bucket-of-one rule, and the app's scope is the honest home for
both halves anyway.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Opus 5 committed 2026-09-05 15:36:40 -04:00
1 parent c6892f9b6c
commit cef7a764af
30 files changed
+2052 -172

No files matched your search

+9
View File
@@ -55,6 +55,15 @@ the courier recipient, anything you *address*. The reception/persistence layer
sortable list you pick from is `contactlist.py` (`ContactListScreen`/`ContactRow`/
`ContactsSort`/`contacts_table`). The `contacts` tool lists the device's added contacts.
Setting vs preference — the other boundary, three-way and never blurred: a **setting** is
the *radio's* (`core/device_config.py`, read live from the companion, edited on Device
config); a **preference** is *MeshTerm's own behaviour* (`core/preferences.py`, defaults in
code, overrides in `preferences.yaml`, edited on the Preferences page — `ctx.preferences`,
or `preferences.current()` where there is no context to reach through); and **config** is
machine setup — paths and device profiles (`core/config.py`, `config.toml`, a text editor
only). A behaviour value belongs in the preference registry, not as a module constant and
not in `config.toml`: a code constant a reader can't reach is a preference nobody has.
Relative ages: `format_ago` for prose ("now", "5m ago", "never" — never "now ago"),
`_format_age` for aligned columns ("now", "5m").
+25 -11
View File
@@ -186,19 +186,33 @@ Global options (before the subcommand): `--profile/-p`, `--port`, `--ble`, `--bl
## Configuration
MeshTerm runs with zero configuration — it discovers attached serial devices and nearby
Bluetooth companions, lets you pick one, and remembers the last good default. A network (TCP)
companion isn't discoverable, so reach it with `--tcp host:port`, a TCP profile, or the
picker's "add a network device" prompt. Define a config file only to give your hardware stable
aliases or tune behaviour.
Two files, two jobs — and you need neither to run.
**Preferences** are how MeshTerm behaves: the pause it leaves between transmissions, how
hard it retries a message, how far back it keeps history, how a map frames itself. Every one has a built-in default, so change them only where you disagree — from
the **Preferences** page in the menu (grouped, staged, saved by one action at the bottom),
or from a shell:
```console
$ meshterm preferences show # every preference, its value, its default
$ meshterm preferences set history_days 90
$ meshterm preferences reset --yes # back to the built-in defaults
```
They are kept in `~/.meshterm/preferences.yaml`, which lists only what you have changed;
delete a line and the default takes over again.
**Config** is where things live and which device to talk to. MeshTerm runs with none of it
— it discovers attached serial devices and nearby Bluetooth companions, lets you pick one,
and remembers the last good default. A network (TCP) companion isn't discoverable, so reach
it with `--tcp host:port`, a TCP profile, or the picker's "add a network device" prompt.
Write a config file only to give your hardware stable aliases.
Copy [`config.example.toml`](config.example.toml) to `~/.meshterm/config.toml`:
```toml
default_profile = "s3"
trace_cooldown_s = 1.0
direct_message_soft_retries = 2 # auto-resend an unacked DM up to N extra times (0–2)
history_days = 365 # days of overheard-packet history to keep (0 = forever)
connect_on_start = true # false opens the radio link lazily instead of at launch
[profiles.s3]
port = "COM5"
@@ -218,8 +232,8 @@ host = "192.168.1.50"
description = "Basestation over WiFi"
```
Timestamps are stored as UTC and rendered in your local time. History older than
`history_days` is pruned once at session start.
Timestamps are stored as UTC and rendered in your local time. History older than the
`history_days` preference is pruned once at session start.
## Architecture
@@ -228,7 +242,7 @@ MeshTerm is layered so a new feature is one file in `meshterm/tools/`:
```
cli.py Typer app; no subcommand -> interactive menu
context.py AppContext (console, config, repository, device) dependency container
core/ Domain: models, config + device profiles, connection abstraction (+ mock)
core/ Domain: models, preferences + device profiles, connection abstraction (+ mock)
tools/ Pluggable "menu options"; each self-registers and gets logging for free
services/ Background algorithms (monitor, trace, tx search, courier, watchtower) — no UI
persistence/ SQLite schema, repository, structured logging
+10 -9
View File
@@ -1,6 +1,12 @@
# MeshTerm configuration.
# MeshTerm configuration: where things live, and which device to talk to.
# Copy to ~/.meshterm/config.toml (or pass --db / --port on the CLI).
#
# How MeshTerm itself behaves — transmit cooldown, message retries, how much history it
# keeps, whether it connects at launch — is *not* here. Those are preferences: they live
# in ~/.meshterm/preferences.yaml, and are edited on the Preferences page (or with
# `meshterm preferences set <key> <value>`). Every one has a built-in default, so that
# file only ever lists what you have changed.
#
# Profiles are optional. With no profile, no --port/--ble/--tcp, MeshTerm discovers attached
# serial devices and nearby Bluetooth companions, lets you pick one (interactively) or uses the
# single one present, and remembers the last device that connected as the default for next time.
@@ -9,14 +15,9 @@
# stable aliases (e.g. --profile yagi, --profile handheld for Bluetooth, --profile wifi for TCP).
default_profile = "s3"
trace_cooldown_s = 1.0
# Automatic re-sends of a direct message that goes unacknowledged, before it's marked failed.
# 0, 1, or 2 (clamped): 0 = one shot, 2 (default) = up to three tries. Ctrl-R in chat is a
# further manual retry on top of this.
direct_message_soft_retries = 2
# Days of overheard-packet history the recorder keeps; older observations are pruned
# once per session start (0 = keep everything forever).
history_days = 365
# Open the radio link as soon as the menu appears. False opens it lazily instead — only
# once monitoring is turned on or a tool first needs the radio.
connect_on_start = true
# db_path = "C:/Users/you/meshterm/meshterm.db"
# One [profiles.<name>] block per physical companion device.
+6 -3
View File
@@ -20,6 +20,7 @@ from .core.admin_store import AdminStore
from .core.config import Settings
from .core.connection import DeviceCommandError, is_connection_lost
from .core.device_config import DeviceConfigError
from .core.preferences import PreferenceError
from .core.device_store import DeviceStore
from .core.selection import DeviceSelectionError
from .persistence.logging import configure_logging
@@ -211,9 +212,11 @@ def run_tool_command(tool: Tool, params: dict) -> None:
assert _state is not None # set by the callback before any subcommand runs
try:
asyncio.run(_drive(_execute_and_render(tool, params, _state), _state))
except (DeviceSelectionError, DeviceConfigError, DeviceCommandError) as exc:
# Expected user-facing error (ambiguous/absent device, bad config value, or a
# transient command failure): show the message, not a traceback.
except (
DeviceSelectionError, DeviceConfigError, DeviceCommandError, PreferenceError
) as exc:
# Expected user-facing error (ambiguous/absent device, a bad config or preference
# value, or a transient command failure): show the message, not a traceback.
_state.console.print(f"[err]✗[/err] {exc}")
raise typer.Exit(1) from exc
except Exception as exc:
+24 -3
View File
@@ -19,6 +19,7 @@ from .core.config import DeviceProfile, Settings
from .core.contact_store import ContactStore
from .core.courier_store import CourierStore
from .core.mute_store import MuteStore
from .core.preferences import Preferences, install as install_preferences
from .core.remote_store import RemoteStore
from .core.settings_store import SettingsStore
from .core.watch_store import WatchStore
@@ -48,7 +49,14 @@ class AppContext:
Attributes:
console: Rich console for all rendering.
settings: Loaded application settings.
settings: Loaded application settings — where things live and which device to
talk to (config directory, database, named profiles). *How MeshTerm behaves*
is ``preferences``, not this.
preferences: MeshTerm's own preferences, backed by
``<config_dir>/preferences.yaml`` and edited on the Preferences page. Defaults
to a set loaded from that file when not injected, and is installed as the
process-wide :func:`~meshterm.core.preferences.current` set so the render
layer — which has no context to reach through — reads the same values.
repo: Database repository.
profile: Active device profile, if one was resolved.
device_store: Store for the remembered "last known good" device.
@@ -94,6 +102,7 @@ class AppContext:
repo: Repository
device_store: DeviceStore
admin_store: AdminStore
preferences: Optional[Preferences] = None
advert_store: Optional[AdvertStore] = None
remote_store: Optional[RemoteStore] = None
watch_store: Optional[WatchStore] = None
@@ -142,7 +151,17 @@ class AppContext:
_ui: "Optional[Ui]" = field(default=None, init=False, repr=False)
def __post_init__(self) -> None:
"""Derive the JSON stores' default locations when they were not injected."""
"""Load the preferences and derive the stores' default locations, where not injected.
The preferences are also installed as the process-wide set (see
:func:`~meshterm.core.preferences.install`), because the session's renderer and the
stores built without a context still have to read them.
"""
if self.preferences is None:
self.preferences = Preferences.load(
self.settings.config_dir / "preferences.yaml"
)
install_preferences(self.preferences)
if self.advert_store is None:
self.advert_store = AdvertStore(self.settings.config_dir / "adverts.json")
if self.remote_store is None:
@@ -390,7 +409,9 @@ class AppContext:
from .ui.map_render import DRAWN_LAYERS
self._basemap_source = BasemapSource(
self.settings.config_dir / "tilecache", layers=DRAWN_LAYERS
self.settings.config_dir / "tilecache",
layers=DRAWN_LAYERS,
tilejson_url=self.preferences.basemap_tilejson_url,
)
return self._basemap_source
+12 -4
View File
@@ -30,7 +30,10 @@ DIRECT_CADENCE_HOURS = (1, 2, 3, 4)
FLOOD_CADENCE_HOURS = (3, 6, 12, 24, 48, 168)
#: Defaults applied when a device has no stored policy: announce to neighbours hourly,
#: flood the wider mesh daily.
#: flood the wider mesh daily. These are the *code's* defaults; the cadence a device
#: without a stored policy actually starts on is the ``advert_direct_hours`` /
#: ``advert_flood_hours`` preference, which defaults to these (see
#: :mod:`meshterm.core.preferences`).
DEFAULT_DIRECT_HOURS = 1
DEFAULT_FLOOD_HOURS = 24
@@ -130,12 +133,17 @@ class AdvertStore:
Returns:
The device's :class:`AdvertPolicy`; defaults apply for anything unset.
"""
from .preferences import current as current_preferences
preferences = current_preferences()
direct = preferences.advert_direct_hours
flood = preferences.advert_flood_hours
record = self._load_all().get(self._key(public_key))
if not isinstance(record, dict):
return AdvertPolicy()
return AdvertPolicy(direct_hours=direct, flood_hours=flood)
return AdvertPolicy(
direct_hours=_as_hours(record.get("direct_hours"), DEFAULT_DIRECT_HOURS),
flood_hours=_as_hours(record.get("flood_hours"), DEFAULT_FLOOD_HOURS),
direct_hours=_as_hours(record.get("direct_hours"), direct),
flood_hours=_as_hours(record.get("flood_hours"), flood),
last_direct=_as_time(record.get("last_direct")),
last_flood=_as_time(record.get("last_flood")),
)
+14 -45
View File
@@ -1,9 +1,16 @@
"""Configuration and named device profiles.
"""Machine setup and named device profiles.
Settings are read from a TOML file (default ``~/.meshterm/config.toml``) and may
be overridden per-invocation by CLI flags. Profiles let you alias your hardware
(``yagi`` repeater, ``local`` repeater, ``observer`` bot, ``s3`` serial companion) to a
serial port and defaults so commands can target them by name.
This file answers *where things are and which device to talk to* — nothing else. How
MeshTerm itself behaves (cooldowns, retry budgets, how much history it keeps, whether it
connects at launch) is a **preference**, lives in
:mod:`meshterm.core.preferences`, and is edited on the Preferences page rather than in a
text editor. The two were one thing until the page existed, which is why a few behaviour
keys used to sit in this file with no screen behind them.
"""
from __future__ import annotations
@@ -19,14 +26,6 @@ else: # pragma: no cover - exercised only on 3.10
import tomli as tomllib
#: Hard ceiling on a direct message's *soft retries* — the automatic re-sends that follow the
#: initial transmission when it goes unacknowledged. Capped at 2, so a message gets at most one
#: send plus two soft retries (three tries total) and is never re-broadcast more than twice on
#: the shared mesh before being called a failure. :data:`Settings.direct_message_soft_retries`
#: is clamped to ``0..DIRECT_MESSAGE_MAX_SOFT_RETRIES``.
DIRECT_MESSAGE_MAX_SOFT_RETRIES = 2
def default_config_dir() -> Path:
"""Return the directory MeshTerm uses for config and data.
@@ -95,47 +94,28 @@ class DeviceProfile:
@dataclass(slots=True)
class Settings:
"""Top-level application settings.
"""Where MeshTerm keeps its files, and which device to talk to.
Attributes:
config_dir: Directory holding the config file and database.
db_path: SQLite database location.
default_profile: Profile used when ``--profile`` is omitted.
profiles: Mapping of profile name to :class:`DeviceProfile`.
trace_cooldown_s: Minimum delay between transmit bursts (duty-cycle safety).
tx_opt_min: Lowest TX power explored by the remote-admin optimizer (dBm).
tx_opt_max: Highest TX power explored by the remote-admin optimizer (dBm).
direct_message_soft_retries: How many times a direct message is automatically
re-sent after its initial transmission goes unacknowledged, before it is reported
as failed. Each soft retry waits a full delivery-ack window (see
:meth:`Device.send_direct_message`) before firing, so a message only resends after
genuinely going unanswered rather than hammering the radio. Capped at 0, 1, or 2 —
clamped to ``0..DIRECT_MESSAGE_MAX_SOFT_RETRIES`` on load — so the send is tried at
most three times total. ``0`` disables soft retries (one shot); the default of
``2`` matches the mesh convention of a few tries before giving up. The manual
Ctrl-R resend in the chat screen is a further, user-driven retry layered on top of
these automatic ones, not a substitute for them.
connect_on_start: Whether the interactive session opens the companion connection
(and starts always-on background listening) immediately at launch. When
``False`` the connection is opened lazily — only once monitoring is turned on
or a tool first needs the radio — so launching the menu touches no serial port.
history_days: How many days of overheard-packet history the recorder retains.
Observations older than this are pruned once per session start (the
housekeeping sweep), keeping the database bounded while the dashboard and
Time Machine draw on everything inside the window. ``0`` disables pruning
entirely — history grows forever.
or a tool first needs the radio — so launching the menu touches no serial
port. It sits here rather than among the preferences because it is about
*which device this machine talks to and when*, alongside the profile that
names it, and because it decides its own question before the session that
would show a preferences page exists.
"""
config_dir: Path = field(default_factory=default_config_dir)
db_path: Optional[Path] = None
default_profile: Optional[str] = None
profiles: dict[str, DeviceProfile] = field(default_factory=dict)
trace_cooldown_s: float = 1.0
tx_opt_min: int = 12
tx_opt_max: int = 28
direct_message_soft_retries: int = 2
connect_on_start: bool = True
history_days: int = 365
def __post_init__(self) -> None:
"""Derive dependent paths that were not explicitly provided."""
@@ -203,16 +183,5 @@ class Settings:
db_path=Path(data["db_path"]) if data.get("db_path") else None,
default_profile=data.get("default_profile"),
profiles=profiles,
trace_cooldown_s=float(data.get("trace_cooldown_s", 1.0)),
tx_opt_min=int(data.get("tx_opt_min", 12)),
tx_opt_max=int(data.get("tx_opt_max", 28)),
direct_message_soft_retries=max(
0,
min(
DIRECT_MESSAGE_MAX_SOFT_RETRIES,
int(data.get("direct_message_soft_retries", DIRECT_MESSAGE_MAX_SOFT_RETRIES)),
),
),
connect_on_start=bool(data.get("connect_on_start", True)),
history_days=max(0, int(data.get("history_days", 365))),
)
+7 -3
View File
@@ -27,7 +27,8 @@ QUEUED = "queued"
DELIVERED = "delivered"
GAVE_UP = "gave-up"
#: How many finished (delivered / given-up) entries are kept for the screen's history.
#: How many finished (delivered / given-up) entries are kept for the screen's history —
#: the code's default behind the ``courier_history_kept`` preference.
DONE_CAP = 100
@@ -193,9 +194,12 @@ class CourierStore:
message.finished = when or utcnow()
messages = self._load()
done = [m for m in messages if m.status != QUEUED]
if len(done) > DONE_CAP:
from .preferences import current as current_preferences
cap = current_preferences().courier_history_kept
if len(done) > cap:
done.sort(key=lambda m: m.finished or m.created)
drop = {m.ident for m in done[: len(done) - DONE_CAP]}
drop = {m.ident for m in done[: len(done) - cap]}
self._messages = [m for m in messages if m.ident not in drop]
self._save()
+666
View File
@@ -0,0 +1,666 @@
"""Declarative registry of MeshTerm's own preferences, and the YAML file they live in.
The distinction this module draws is between the three kinds of "setting" the app already
had and never named apart:
* :mod:`meshterm.core.device_config` describes the **radio's** settings — they live in
the companion's firmware, are read live each session, and are edited through the Device
config screen.
* :mod:`meshterm.core.config` (``config.toml``) describes **where things are and which
device to talk to** — the config directory, the database, the named device profiles.
It is machine setup, edited in a text editor, and has no place in a full-screen page.
* This module describes **how MeshTerm itself behaves** — whether it opens the link at
launch, how long it waits between transmissions, how much history it keeps, how a map
frames itself. Those values were scattered through the code as module constants and a
handful of undocumented ``config.toml`` keys with no screen behind them; they are
gathered here, given one place to be changed (the Preferences page), and one file to be
remembered in.
Each :class:`PrefSpec` states a preference's group, type, bounds, and — the point of the
whole exercise — its **default**. A preference is only written to disk once it is changed
away from that default, so the file is a short list of your disagreements with the
built-in behaviour rather than a snapshot that silently pins every value forever. Delete a
key (or use *Reset to defaults*) and the code's default takes over again.
The file is ``<config_dir>/preferences.yaml``, written grouped and commented so it reads
the way the page does. Reads are tolerant: an unknown key, a malformed value, or an
unparseable file costs the override, never the session.
Two access paths, deliberately:
* :attr:`~meshterm.context.AppContext.preferences` is the explicit one, and what every
tool, service, and screen should use.
* :func:`current` is for the render and session layer, which has no context to reach
through — the same reason :func:`meshterm.platforms.get_platform` exists. The context
installs itself there as it is built.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional
from .advert_store import (
DEFAULT_DIRECT_HOURS,
DEFAULT_FLOOD_HOURS,
DIRECT_CADENCE_HOURS,
FLOOD_CADENCE_HOURS,
OFF,
cadence_label,
)
from .watch_store import DEFAULT_SILENCE_HOURS, SILENCE_CHOICES_H
#: Display groups, in the order the page and the file present them. The order is the
#: order a session happens in — what MeshTerm puts on the air, how loud, what it
#: announces, what it watches for, how it draws the world, what it keeps, and how it
#: paints. Nothing sorts alphabetically: a reader looking for "how long before I give up
#: on a message" should not have to know it starts with a D.
GROUPS: tuple[str, ...] = (
"Sending",
"Transmit power",
"Background adverts",
"Watchtower",
"Map",
"History kept",
"Display",
)
class PreferenceError(ValueError):
"""Raised when a preference value cannot be parsed or fails validation.
The message is user-facing — the CLI prints it, and the editor's typed prompt shows
it under the input as the reason the value was refused.
"""
@dataclass(frozen=True, slots=True)
class PrefSpec:
"""Specification for one preference.
Attributes:
key: Canonical key — the attribute name on :class:`Preferences`, and the key in
the YAML file.
label: Human-friendly name, as the page's SETTING lane shows it.
help: One-line description, as the page's DESCRIPTION lane shows it.
group: One of :data:`GROUPS`.
value_type: ``"str" | "int" | "float" | "bool" | "enum"``.
default: The built-in value, used whenever the file names no override.
choices: For ``enum``, a mapping of allowed value to label.
minimum: Inclusive lower bound for numeric types, if any.
maximum: Inclusive upper bound for numeric types, if any.
unit: Short suffix appended when the value is formatted (``"s"``, ``"dBm"``,
``"days"``) so a bare number in the VALUE lane still says what it counts.
relaunch: Whether the new value only takes hold next launch. Woven into the
description rather than shown as its own mark — it is a caveat about one
preference, not a status the reader scans a column for.
"""
key: str
label: str
help: str
group: str
value_type: str
default: Any
choices: Optional[dict[Any, str]] = None
minimum: Optional[float] = None
maximum: Optional[float] = None
unit: str = ""
relaunch: bool = False
@property
def description(self) -> str:
"""The DESCRIPTION lane's text: the help, plus the relaunch caveat where it applies."""
return f"{self.help} (next launch)" if self.relaunch else self.help
def _cadence_choices(hours: tuple[int, ...]) -> dict[int, str]:
"""Advert-cadence choices as ``hours -> label``, ``off`` last (the editor's own order)."""
return {h: cadence_label(h) for h in (*hours, OFF)}
#: The silence-rule choices, drawn from the Watchtower's own ring so the two never drift.
_SILENCE_CHOICES: dict[int, str] = {
**{h: f"{h} h" for h in SILENCE_CHOICES_H},
OFF: "off",
}
#: How the terminal-width reclaim can be asked for: follow the platform's own verdict, or
#: overrule it in either direction. See :func:`meshterm.ui.tui.session._reclaim_last_column`
#: for why a terminal may need to disagree with its platform.
_WIDTH_CHOICES: dict[str, str] = {
"auto": "follow the platform",
"on": "reclaim the column",
"off": "never reclaim",
}
#: Every preference MeshTerm has, in page order. Adding one here gives it a row on the
#: Preferences page, a key in the YAML file, a ``preferences get``/``set`` CLI face, and a
#: default — nothing else has to follow.
PREFERENCES: tuple[PrefSpec, ...] = (
# --- Sending -----------------------------------------------------------------
PrefSpec(
key="trace_cooldown_s",
label="Transmit cooldown",
help="Pause between our own transmissions, for duty-cycle safety",
group="Sending",
value_type="float",
default=1.0,
minimum=0.0,
maximum=60.0,
unit="s",
),
PrefSpec(
key="direct_message_soft_retries",
label="Message retries",
help="Automatic re-sends of an unacknowledged direct message",
group="Sending",
value_type="int",
default=0, # one shot: a re-send is a second transmission on a shared mesh
minimum=0,
maximum=2,
),
# --- Transmit power ----------------------------------------------------------
PrefSpec(
key="tx_opt_min",
label="Sweep floor",
help="Lowest power the transmit-power optimizer explores",
group="Transmit power",
value_type="int",
default=18,
minimum=1,
maximum=30,
unit="dBm",
),
PrefSpec(
key="tx_opt_max",
label="Sweep ceiling",
help="Highest power the transmit-power optimizer explores",
group="Transmit power",
value_type="int",
default=28,
minimum=1,
maximum=30,
unit="dBm",
),
PrefSpec(
key="tx_snr_tolerance_db",
label="Tie tolerance",
help="Within this much SNR of the best, the lower power wins",
group="Transmit power",
value_type="float",
default=1.0,
minimum=0.0,
maximum=10.0,
unit="dB",
),
# --- Background adverts ------------------------------------------------------
PrefSpec(
key="advert_direct_hours",
label="Direct advert",
help="Cadence a newly seen device starts on for zero-hop announces",
group="Background adverts",
value_type="enum",
default=DEFAULT_DIRECT_HOURS,
choices=_cadence_choices(DIRECT_CADENCE_HOURS),
),
PrefSpec(
key="advert_flood_hours",
label="Flood advert",
help="Cadence a newly seen device starts on for mesh-wide announces",
group="Background adverts",
value_type="enum",
default=DEFAULT_FLOOD_HOURS,
choices=_cadence_choices(FLOOD_CADENCE_HOURS),
),
# --- Watchtower --------------------------------------------------------------
PrefSpec(
key="watch_silence_hours",
label="Silence alarm",
help="Silence a newly watched node is allowed before the alarm",
group="Watchtower",
value_type="enum",
default=DEFAULT_SILENCE_HOURS,
choices=_SILENCE_CHOICES,
),
PrefSpec(
key="watch_alerts_kept",
label="Alerts kept",
help="Alerts the Watchtower log retains, oldest dropped first",
group="Watchtower",
value_type="int",
default=200,
minimum=10,
maximum=5000,
),
# --- Map ---------------------------------------------------------------------
PrefSpec(
key="map_view_fraction",
label="Opening frame",
help="Fraction of nodes a map frames, so outliers don't zoom it out",
group="Map",
value_type="float",
default=0.5,
minimum=0.1,
maximum=1.0,
),
PrefSpec(
key="basemap_tilejson_url",
label="Basemap source",
help="TileJSON the map's vector tiles are fetched from",
group="Map",
value_type="str",
default="https://tiles.openfreemap.org/planet",
relaunch=True,
),
# --- History kept ------------------------------------------------------------
PrefSpec(
key="history_days",
label="Packet history",
help="Days of overheard history kept; 0 keeps everything forever",
group="History kept",
value_type="int",
default=365,
minimum=0,
maximum=36500,
unit="days",
),
PrefSpec(
key="chat_history_limit",
label="Transcript depth",
help="Past messages loaded when a conversation opens",
group="History kept",
value_type="int",
default=200,
minimum=20,
maximum=5000,
),
PrefSpec(
key="courier_history_kept",
label="Courier history",
help="Delivered and given-up outbox entries kept for the screen",
group="History kept",
value_type="int",
default=100,
minimum=10,
maximum=5000,
),
# --- Display -----------------------------------------------------------------
PrefSpec(
key="fast_render",
label="Fast redraw",
help="Write rows straight to the terminal instead of through the stock renderer",
group="Display",
value_type="bool",
default=True,
relaunch=True,
),
PrefSpec(
key="full_width",
label="Last column",
help="Use the column a terminal that under-reports its width hides",
group="Display",
value_type="enum",
default="auto",
choices=_WIDTH_CHOICES,
),
)
_BY_KEY: dict[str, PrefSpec] = {spec.key: spec for spec in PREFERENCES}
def get_spec(key: str) -> PrefSpec:
"""Look up one preference's spec.
Args:
key: The preference key.
Returns:
Its :class:`PrefSpec`.
Raises:
PreferenceError: If no preference is registered under ``key``.
"""
try:
return _BY_KEY[key]
except KeyError:
raise PreferenceError(f"unknown preference: {key!r}") from None
def by_group() -> list[tuple[str, list[PrefSpec]]]:
"""Every preference grouped for display, in :data:`GROUPS` order.
Returns:
``(group, specs)`` pairs; a group holding no preferences is omitted.
"""
grouped = [(g, [s for s in PREFERENCES if s.group == g]) for g in GROUPS]
return [(g, specs) for g, specs in grouped if specs]
# --- value parsing / formatting -----------------------------------------------
_TRUE = {"1", "true", "yes", "on", "y"}
_FALSE = {"0", "false", "no", "off", "n"}
def parse_value(spec: PrefSpec, raw: Any) -> Any:
"""Parse and validate a raw value (a typed scalar, or text from the CLI/file/prompt).
Args:
spec: The target preference.
raw: The value to coerce.
Returns:
The typed, range-checked value.
Raises:
PreferenceError: If the value is the wrong type, out of range, or not a choice.
"""
text = str(raw).strip()
if spec.value_type == "bool":
low = text.lower()
if low in _TRUE:
return True
if low in _FALSE:
return False
raise PreferenceError(f"{spec.key}: expected on or off, got {raw!r}")
if spec.value_type == "str":
return text
if spec.value_type == "enum":
choices = spec.choices or {}
if raw in choices:
return raw
# Text arriving from the CLI or a hand-edited file: match a choice by its own
# spelling, so `preferences set advert_flood_hours 24` and a YAML `24` land alike.
for choice in choices:
if text == str(choice):
return choice
allowed = ", ".join(str(c) for c in choices)
raise PreferenceError(f"{spec.key}: must be one of {allowed}, got {raw!r}")
try:
value: Any = float(text) if spec.value_type == "float" else int(text, 0)
except (ValueError, TypeError) as exc:
raise PreferenceError(f"{spec.key}: invalid {spec.value_type} value {raw!r}") from exc
if spec.minimum is not None and value < spec.minimum:
raise PreferenceError(f"{spec.key}: must be >= {spec.minimum:g}, got {value:g}")
if spec.maximum is not None and value > spec.maximum:
raise PreferenceError(f"{spec.key}: must be <= {spec.maximum:g}, got {value:g}")
return value
def format_value(spec: PrefSpec, value: Any) -> str:
"""Render one preference's value for display.
Booleans read ``on``/``off`` (the words the editor's button pair offers), an enum
reads its own label, and a number carries its unit so a bare figure in the VALUE lane
still says what it counts.
Args:
spec: The preference the value belongs to.
value: The value to render.
Returns:
The display string.
"""
if spec.value_type == "bool":
return "on" if value else "off"
if spec.value_type == "enum" and spec.choices is not None:
return str(spec.choices.get(value, value))
if spec.value_type in ("int", "float"):
text = f"{value:g}"
return f"{text} {spec.unit}" if spec.unit else text
return str(value)
def range_hint(spec: PrefSpec) -> str:
"""A muted "allowed values" hint for a typed prompt, or ``""`` where nothing bounds it."""
unit = f" {spec.unit}" if spec.unit else ""
if spec.minimum is not None and spec.maximum is not None:
return f"Allowed: {spec.minimum:g} – {spec.maximum:g}{unit}"
if spec.minimum is not None:
return f"Allowed: >= {spec.minimum:g}{unit}"
if spec.maximum is not None:
return f"Allowed: <= {spec.maximum:g}{unit}"
return ""
# --- the store ----------------------------------------------------------------
#: The file's opening comment. It says the one thing a reader hand-editing the file needs
#: to know — that absence means "use the default", so deleting a line is how a preference
#: is undone — and points at the two neighbouring kinds of setting so nobody looks for a
#: device's radio parameters in here.
_HEADER = """\
# MeshTerm preferences -- how the app itself behaves.
#
# Every preference has a built-in default. A key is listed here only while you have
# changed it away from that default, so deleting a line (or "Reset to defaults" on the
# Preferences page) hands the value back to the code. `meshterm preferences show` prints
# every preference, what it is set to, and what it defaults to.
#
# Device profiles, the database location, and the rest of the machine setup stay in
# config.toml; the radio's own settings live on the radio, under Device config.
"""
class Preferences:
"""The effective preferences: the built-in defaults, with the file's overrides on top.
Read a preference as an attribute — ``preferences.trace_cooldown_s`` — which resolves
through the registry, so a typo raises :class:`AttributeError` at the call site
instead of quietly reading ``None``. :meth:`set` takes a raw value, validates it
through the key's spec, and records it only while it differs from the default;
:meth:`save` writes the result.
Nothing here writes on its own: the Preferences page stages changes and saves them on
*Apply*, which — with the CLI — is the only path that reaches :meth:`save`.
"""
def __init__(
self, path: Optional[Path] = None, values: Optional[Mapping[str, Any]] = None
) -> None:
"""Build a preferences view over an optional file and an optional set of overrides.
Args:
path: Where :meth:`save` writes. ``None`` for an in-memory set (tests, and
the fallback :func:`current` hands out before a context is built).
values: Overrides to start from; invalid ones are dropped, and a value equal
to its default is not recorded as an override at all.
"""
self._path = path
self._values: dict[str, Any] = {}
if values:
self.update(values)
@classmethod
def load(cls, path: Path) -> "Preferences":
"""Read ``path``, returning all-defaults for a missing, empty, or corrupt file.
Args:
path: The YAML file to read.
Returns:
The loaded preferences, bound to ``path`` for a later :meth:`save`.
"""
import yaml
try:
data = yaml.safe_load(path.read_text(encoding="utf-8"))
except (OSError, ValueError, yaml.YAMLError):
return cls(path)
return cls(path, data if isinstance(data, Mapping) else None)
@property
def path(self) -> Optional[Path]:
"""Where :meth:`save` writes, or ``None`` for an in-memory set."""
return self._path
def __getattr__(self, name: str) -> Any:
"""Resolve a registered preference key as an attribute.
Raises:
AttributeError: For any name that is not a preference key, so a mistyped
``preferences.trace_cooldwn_s`` fails where it is written rather than
quietly reading nothing.
"""
if name.startswith("_") or name not in _BY_KEY:
raise AttributeError(name)
return self.get(name)
def get(self, key: str) -> Any:
"""The effective value for ``key``: the override where there is one, else the default.
Raises:
PreferenceError: If ``key`` is not a registered preference.
"""
spec = get_spec(key)
return self._values.get(key, spec.default)
def is_overridden(self, key: str) -> bool:
"""Whether ``key`` currently differs from its built-in default."""
return key in self._values
def overrides(self) -> dict[str, Any]:
"""The values that differ from their defaults, in registry order (a copy)."""
return {s.key: self._values[s.key] for s in PREFERENCES if s.key in self._values}
def set(self, key: str, raw: Any) -> Any:
"""Validate ``raw`` for ``key`` and record it, dropping it when it *is* the default.
An override is only ever a disagreement with the default, so setting a value back
to its default clears the entry rather than pinning it. That is what lets a later
change to the code's default reach an install that never asked to be exempt.
Args:
key: The preference key.
raw: The new value, typed or as text.
Returns:
The parsed value now in force.
Raises:
PreferenceError: If ``key`` is unknown or ``raw`` fails its spec.
"""
spec = get_spec(key)
value = parse_value(spec, raw)
if value == spec.default:
self._values.pop(key, None)
else:
self._values[key] = value
return value
def update(self, values: Mapping[str, Any]) -> int:
"""Apply several changes at once, skipping any the registry or a spec refuses.
Tolerant by design: this is the path a hand-edited file arrives through, and one
bad line should cost that line, not the file.
Args:
values: ``key -> value`` pairs.
Returns:
How many were accepted.
"""
applied = 0
for key, value in values.items():
try:
self.set(str(key), value)
except PreferenceError:
continue
applied += 1
return applied
def reset(self) -> int:
"""Drop every override, returning how many there were."""
count = len(self._values)
self._values.clear()
return count
def as_yaml(self) -> str:
"""Render the current overrides as the file's text: header, then grouped entries.
Each group holding an override gets its own comment heading, and each entry its
help line and its default above it, so the file reads the way the page does
rather than as a bare map someone has to look up elsewhere.
"""
import yaml
lines = [_HEADER]
for group, specs in by_group():
present = [s for s in specs if s.key in self._values]
if not present:
continue
lines.append(f"# --- {group} ---")
for spec in present:
entry = yaml.safe_dump(
{spec.key: self._values[spec.key]},
default_flow_style=False,
sort_keys=False,
allow_unicode=True,
).rstrip("\n")
lines.append(f"# {spec.help}")
lines.append(f"# default: {format_value(spec, spec.default)}")
lines.append(entry)
lines.append("")
if len(lines) == 1:
lines.append("# (nothing overridden -- every preference is at its default)")
lines.append("")
return "\n".join(lines)
def save(self) -> None:
"""Write the overrides to :attr:`path` atomically (a crash mid-write keeps the old file).
Raises:
RuntimeError: If this set was built without a path (an in-memory set).
"""
if self._path is None:
raise RuntimeError("these preferences have no file to save to")
self._path.parent.mkdir(parents=True, exist_ok=True)
tmp = self._path.with_suffix(self._path.suffix + ".tmp")
tmp.write_text(self.as_yaml(), encoding="utf-8")
tmp.replace(self._path)
#: The set installed for this process. Starts as a defaults-only, file-less instance so
#: the render layer works before (and without) an :class:`~meshterm.context.AppContext` —
#: a specimen render, a unit test, the CLI's early boot.
_current: Preferences = Preferences()
def current() -> Preferences:
"""The preferences in force for this process.
For the layers that have no context to reach through — the session's renderer, the
stores constructed without one — mirroring
:func:`meshterm.platforms.get_platform`. Anything holding an
:class:`~meshterm.context.AppContext` reads
:attr:`~meshterm.context.AppContext.preferences` instead.
"""
return _current
def install(preferences: Preferences) -> None:
"""Make ``preferences`` the set :func:`current` returns (the context does this at boot)."""
global _current
_current = preferences
__all__ = [
"GROUPS",
"PREFERENCES",
"PrefSpec",
"PreferenceError",
"Preferences",
"by_group",
"current",
"format_value",
"get_spec",
"install",
"parse_value",
"range_hint",
]
+17 -5
View File
@@ -30,13 +30,16 @@ from .models import utcnow
SILENCE_CHOICES_H = (1, 3, 6, 12, 24, 48)
#: Default silence threshold: half a day covers nodes that only advertise daily-ish
#: without crying wolf over a quiet afternoon.
#: without crying wolf over a quiet afternoon. It is the *code's* default; what a newly
#: starred node actually gets is the ``watch_silence_hours`` preference, which defaults
#: to this (see :mod:`meshterm.core.preferences`).
DEFAULT_SILENCE_HOURS = 12
#: The stored value meaning "this rule is off".
OFF = 0
#: How many alerts the log retains (oldest dropped first).
#: How many alerts the log retains (oldest dropped first) — the code's default behind
#: the ``watch_alerts_kept`` preference.
ALERT_CAP = 200
#: Minimum seconds between disk flushes for the packet-driven last-heard updates.
@@ -202,8 +205,14 @@ class WatchStore:
last_seen: When the node was last heard, if known.
node_type: The node's advertised type, if known (for the watchlist glyph).
"""
from .preferences import current as current_preferences
self.state.watched[key] = WatchedNode(
key=key, name=name, last_heard=last_seen or utcnow(), node_type=node_type
key=key,
name=name,
silence_hours=current_preferences().watch_silence_hours,
last_heard=last_seen or utcnow(),
node_type=node_type,
)
self._save()
@@ -324,8 +333,11 @@ class WatchStore:
)
state.next_id += 1
state.alerts.append(alert)
if len(state.alerts) > ALERT_CAP:
del state.alerts[: len(state.alerts) - ALERT_CAP]
from .preferences import current as current_preferences
cap = current_preferences().watch_alerts_kept
if len(state.alerts) > cap:
del state.alerts[: len(state.alerts) - cap]
self._dirty = True
self.flush(only_if_due=True)
return alert
+7 -5
View File
@@ -427,8 +427,10 @@ class ChatService:
"""Transmit a direct message, softly retrying until it is acknowledged.
A single logical send makes one initial transmission plus up to
``settings.direct_message_soft_retries`` soft retries (two by default, three tries in
all): the message is re-sent only when an attempt goes unacknowledged, and each device
``preferences.direct_message_soft_retries`` soft retries (none by default — one shot,
since a re-send is a second transmission on a shared mesh; two is the ceiling, for
three tries in all): the message is re-sent only when an attempt goes unacknowledged,
and each device
call already blocks for a full delivery-ack window (see
:meth:`~meshterm.core.connection.Device.send_direct_message`) before returning, so the
retries are naturally spaced by that window rather than hammering the radio. The loop
@@ -449,7 +451,7 @@ class ChatService:
a rejection is not retried, since it is not a lost-in-the-mesh timeout.
"""
device = await self._ctx.device()
attempts = max(0, self._ctx.settings.direct_message_soft_retries) + 1
attempts = max(0, self._ctx.preferences.direct_message_soft_retries) + 1
ack = None
for attempt in range(1, attempts + 1):
ack = await device.send_direct_message(contact, text)
@@ -467,7 +469,7 @@ class ChatService:
async def send_direct(self, contact: Contact, text: str) -> ChatMessage:
"""Send a direct message to a contact and record it in history.
Delivery is attempted with up to ``settings.direct_message_soft_retries`` automatic
Delivery is attempted with up to ``preferences.direct_message_soft_retries`` automatic
soft retries (see :meth:`_deliver_direct`) before the message is recorded as
unacknowledged.
@@ -499,7 +501,7 @@ class ChatService:
is ``False``). The same stored row is reused — its delivery state is updated rather
than a duplicate transcript entry created — so the message simply flips to delivered
(or stays unacknowledged for another retry). Like an initial send, each manual retry
makes up to ``settings.direct_message_soft_retries`` soft retries of its own (see
makes up to ``preferences.direct_message_soft_retries`` soft retries of its own (see
:meth:`_deliver_direct`).
Args:
+2 -1
View File
@@ -81,7 +81,8 @@ class MonitorService:
self._kind_counts: dict[str, int] = {}
# Housekeeping: age out observations past the retention window once per
# session, so an always-recording database stays bounded (0 = keep forever).
days = getattr(getattr(ctx, "settings", None), "history_days", 0) or 0
preferences = getattr(ctx, "preferences", None)
days = (preferences.history_days if preferences is not None else 0) or 0
if days:
try:
pruned = ctx.repo.prune_observations(utcnow() - timedelta(days=days))
+5 -3
View File
@@ -1,8 +1,10 @@
"""The About MeshTerm tools: the four pages under the menu's last section.
"""The About MeshTerm tools: the four written pages in the menu's last section.
*About MeshTerm*, *About the author*, *Join Discord*, *Support MeshTerm* — in that
order, which is the order a stranger asks the questions in: what is this, who made it,
where is everyone, how do I help.
where is everyone, how do I help. They share the *This app* section with Preferences,
which leads it: the section is the app's own scope, and the one row that *changes*
MeshTerm sits above the four that describe it.
Each is a page rather than a feature: it reads nothing, transmits nothing, and needs no
device, so it opens straight to its screen (see :mod:`meshterm.ui.about`) instead of
prompting for anything first.
@@ -34,7 +36,7 @@ class _AboutTool(Tool):
subclass supplies :meth:`page`, and nothing more.
"""
category = "About MeshTerm"
category = "This app"
@staticmethod
def page() -> "MarkdownDoc":
+14 -10
View File
@@ -125,16 +125,19 @@ class Tool(ABC):
"""
from ..core.connection import DeviceCommandError
from ..core.device_config import DeviceConfigError
from ..core.preferences import PreferenceError
from ..core.selection import DeviceSelectionError
run_id = ctx.repo.start_run(self.name, params, ctx.profile_name)
ctx.log.debug("run %s start: tool=%s params=%s", run_id, self.name, params)
try:
result = await self.run(ctx, {**params, "_run_id": run_id})
except (DeviceSelectionError, DeviceConfigError, DeviceCommandError) as exc:
# Expected user-facing condition (no/ambiguous device, bad config value, or
# a transient command failure): record it but don't dump a traceback;
# callers print the message cleanly.
except (
DeviceSelectionError, DeviceConfigError, DeviceCommandError, PreferenceError
) as exc:
# Expected user-facing condition (no/ambiguous device, a bad config or
# preference value, or a transient command failure): record it but don't dump
# a traceback; callers print the message cleanly.
ctx.repo.finish_run(run_id, "error", {"error": str(exc)})
ctx.log.debug("run %s aborted: %s", run_id, exc)
raise
@@ -157,18 +160,19 @@ _REGISTRY: dict[str, Tool] = {}
#: that score the walking), then the two scopes a setting can belong to — This node, the
#: radio in your hand, and Other nodes, someone else's over the mesh.
#:
#: About MeshTerm closes the list and is the one section that names a *subject* instead:
#: "what is this thing, and who made it" is a question none of the five doings can hold,
#: and its pages act on nothing — so it sits last, where a reader who has run out of
#: things to do finds it. Categories not listed here sort last, alphabetically, so a new
#: category still appears.
#: This app closes the list as the third and last *scope*, after the radio in your hand and
#: someone else's over the mesh: the program in front of you. It holds the one thing that
#: changes how MeshTerm behaves (Preferences) and the four pages that say what MeshTerm is
#: — questions none of the doings above can hold, and which nothing on the mesh can answer.
#: It sits last because a reader who has run out of things to do is who goes looking for it.
#: Categories not listed here sort last, alphabetically, so a new category still appears.
_CATEGORY_ORDER = [
"Message",
"Watch",
"Explore",
"This node",
"Other nodes",
"About MeshTerm",
"This app",
]
+25 -8
View File
@@ -22,10 +22,28 @@ from rich.table import Table
from rich.text import Text
from ..context import AppContext
from ..core.geo import DEFAULT_VIEW_FRACTION, usable_fix
from ..core.geo import usable_fix
from ..core.models import NODE_TYPE_REPEATER
from .base import Tool, ToolResult, register
def _fraction(ctx: AppContext, params: dict[str, Any]) -> float:
"""How much of the mesh a map opens framed on: the flag if one was passed, else the preference.
``--fraction`` is resolved here rather than as the Typer option's default because the
options are declared at CLI *registration* time — before a context, and so before the
preferences file has been read.
Args:
ctx: Shared application context.
params: This invocation's parameters.
Returns:
The fraction to frame.
"""
fraction = params.get("fraction")
return ctx.preferences.map_view_fraction if fraction is None else float(fraction)
if TYPE_CHECKING:
from ..ui.map_render import MapMarker
@@ -66,9 +84,7 @@ class MapTool(Tool):
if interactive:
from ..ui.map_screen import open_map
await open_map(
ctx, markers, fraction=params.get("fraction", DEFAULT_VIEW_FRACTION)
)
await open_map(ctx, markers, fraction=_fraction(ctx, params))
else:
await self._render_static(ctx, markers, params)
@@ -123,7 +139,7 @@ class MapTool(Tool):
dot_w,
dot_h,
max_zoom=max_zoom,
fraction=params.get("fraction", DEFAULT_VIEW_FRACTION),
fraction=_fraction(ctx, params),
)
tiles = {}
@@ -151,14 +167,15 @@ class MapTool(Tool):
zoom: Optional[int] = typer.Option(
None, "--zoom", "-z", help="Fixed zoom level (omit to fit the nodes)"
),
fraction: float = typer.Option(
DEFAULT_VIEW_FRACTION,
fraction: Optional[float] = typer.Option(
None,
"--fraction",
"-f",
min=0.0,
max=1.0,
help="Fraction of nodes to frame: the densest that many, so distant "
"outliers don't zoom the view out. 1.0 fits every node. Ignored with --zoom",
"outliers don't zoom the view out. 1.0 fits every node. Ignored with "
"--zoom. Defaults to the 'Opening frame' preference",
),
basemap: bool = typer.Option(
True, "--basemap/--no-basemap", help="Draw the OpenStreetMap street basemap"
+159
View File
@@ -0,0 +1,159 @@
"""The ``preferences`` tool: how MeshTerm itself behaves.
Interactively it opens the Preferences page (see :mod:`meshterm.ui.preferences`), which
stages changes and hands them back here to write. On the CLI it exposes the same
preferences as ``show`` / ``get`` / ``set`` / ``reset``, so a value can be changed from a
shell — or from a script setting a machine up — without launching the full-screen session.
Everything funnels through :meth:`PreferencesTool.run`, which applies a list of operations
and saves the file once at the end: one write per invocation, whether it came from the page
staging eleven changes or from a single ``preferences set``.
This is deliberately the *app's* tool, not the radio's. It reads nothing from the
companion, transmits nothing, and works with no device attached at all — which is why it
leads the **This app** section (the third scope, after this node and other nodes) rather
than sitting beside Device config.
"""
from __future__ import annotations
from typing import Any, Optional
import typer
from ..context import AppContext
from ..core.preferences import PreferenceError, format_value, get_spec
from .base import Tool, ToolResult, register
@register
class PreferencesTool(Tool):
"""View and change MeshTerm's own preferences, saved to ``preferences.yaml``."""
name = "preferences"
title = "Preferences"
icon = "⚙"
help = "How MeshTerm behaves — startup, sending, history, …"
category = "This app"
order = 5 # leads the section: the one row there that changes MeshTerm, not describes it
async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]:
"""Open the Preferences page and collect the changes it staged.
Args:
ctx: Shared application context.
Returns:
``{"ops": [("set", key, value), …]}`` to write, or ``None`` if the reader left
with nothing staged.
"""
from ..ui.preferences import edit_preferences
staged = await edit_preferences(ctx)
if not staged:
return None
return {"ops": [("set", key, value) for key, value in staged.items()]}
async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Apply a list of preference operations, saving the file once if anything changed.
Operations are ``("show",)``, ``("get", key)``, ``("set", key, value)``, and
``("reset",)``. They run in order, so a scripted ``set`` after a ``reset`` lands on
top of the defaults rather than under them.
Args:
ctx: Shared application context.
params: ``ops`` — the operations to perform.
Returns:
A :class:`ToolResult` counting the preferences changed.
Raises:
PreferenceError: If a key is unknown or a value fails its spec — reported
cleanly by the CLI rather than as a traceback.
"""
from ..ui.preferences import preferences_table
prefs = ctx.preferences
ops: list[tuple] = list(params.get("ops") or [])
changes = 0
for op in ops:
action = op[0]
if action == "show":
ctx.ui.show(preferences_table(prefs, ctx.console.width))
elif action == "get":
spec = get_spec(op[1])
where = "changed" if prefs.is_overridden(spec.key) else "default"
ctx.ui.show(
f"[accent]{spec.key}[/accent] = "
f"[brand]{format_value(spec, prefs.get(spec.key))}[/brand] ({where})"
)
elif action == "set":
before = prefs.get(op[1])
after = prefs.set(op[1], op[2])
if after != before:
changes += 1
elif action == "reset":
changes += prefs.reset()
if changes:
prefs.save()
plural = "" if changes == 1 else "s"
message = (
f"[ok]✓[/ok] saved [brand]{changes}[/brand] preference{plural} to "
f"[accent]{prefs.path}[/accent]"
if changes
else None
)
return ToolResult(summary={"changes": changes}, message=message)
# -- CLI --------------------------------------------------------------------
def register_cli(self, app: typer.Typer) -> None:
"""Register the ``preferences`` subcommand group.
Args:
app: The Typer application.
"""
from ..cli import run_tool_command
prefs_app = typer.Typer(
help=self.help, no_args_is_help=False, rich_markup_mode="rich"
)
@prefs_app.callback(invoke_without_command=True)
def _root(ctx: typer.Context) -> None:
"""Show every preference when no subcommand is given."""
if ctx.invoked_subcommand is None:
run_tool_command(self, {"ops": [("show",)]})
@prefs_app.command("show", help="Show every preference, its value, and its default")
def _show_cmd() -> None:
run_tool_command(self, {"ops": [("show",)]})
@prefs_app.command("get", help="Print one preference's value")
def _get_cmd(key: str = typer.Argument(..., help="Preference key")) -> None:
run_tool_command(self, {"ops": [("get", key)]})
@prefs_app.command("set", help="Set one preference to a value")
def _set_cmd(
key: str = typer.Argument(..., help="Preference key"),
value: str = typer.Argument(..., help="New value"),
) -> None:
run_tool_command(self, {"ops": [("set", key, value)]})
@prefs_app.command("reset", help="Return every preference to its built-in default")
def _reset_cmd(
yes: bool = typer.Option(False, "--yes", help="Skip the confirmation"),
) -> None:
if not yes:
raise typer.BadParameter(
"resetting drops every preference you have changed; pass --yes to confirm"
)
run_tool_command(self, {"ops": [("reset",)]})
app.add_typer(prefs_app, name=self.name)
__all__ = ["PreferencesTool", "PreferenceError"]
+4 -3
View File
@@ -298,12 +298,13 @@ class TxOptimizeTool(Tool):
target_label,
admin_node,
path,
tx_min=int(params.get("tx_min", ctx.settings.tx_opt_min)),
tx_max=int(params.get("tx_max", ctx.settings.tx_opt_max)),
tx_min=int(params.get("tx_min", ctx.preferences.tx_opt_min)),
tx_max=int(params.get("tx_max", ctx.preferences.tx_opt_max)),
coarse_step=int(params.get("step", 3)),
samples_per_level=samples,
apply=bool(params.get("apply", True)),
cooldown_s=ctx.settings.trace_cooldown_s,
cooldown_s=ctx.preferences.trace_cooldown_s,
snr_tolerance=ctx.preferences.tx_snr_tolerance_db,
on_level=on_level,
persist_level=lambda lv: ctx.repo.record_tx_sample(run_id, lv),
persist_trace=lambda t: ctx.repo.record_trace(run_id, t),
+1 -3
View File
@@ -41,8 +41,6 @@ from .widgets import name_chip
if TYPE_CHECKING:
from ..context import AppContext
#: How many past messages to load into the transcript when a conversation opens.
_HISTORY_LIMIT = 200
#: Matches an ``@[Name]`` mention token, as the reply flow primes into the compose line (see
#: :meth:`ChatScreen._begin_reply`). The transcript renders each as a bare ``@Name`` colored
@@ -997,7 +995,7 @@ async def open_chat(ctx: "AppContext", conversation: Conversation) -> int:
is_channel=conversation.is_channel,
channel_id=conversation.channel_id,
peer=conversation.peer,
limit=_HISTORY_LIMIT,
limit=ctx.preferences.chat_history_limit,
)
contacts = await ctx.devstate.contacts()
names = _contact_names(contacts)
+395
View File
@@ -0,0 +1,395 @@
"""The Preferences page: MeshTerm's own behaviour, staged and saved.
The app-side twin of the device-configuration editor (:mod:`meshterm.ui.config_editor`),
built out of the same parts on purpose — one grouped, lane-aligned list; a row that stages
``current → new`` in the warn style; an Apply/Back pair that appears only once something
is staged; the shared discard confirm on the way out. Two editors that stage values should
not be two different screens to learn, so the only things that differ here are what is
being edited (:mod:`meshterm.core.preferences`, not the radio) and where Apply writes
(``preferences.yaml``, not the companion).
Its rows scroll sideways, which the device editor's do not: a preference's explanation is
the widest thing on its row and the only part that runs past the edge, so ←→ slide the
DESCRIPTION lane under a pinned setting and value rather than letting it die at the fold.
What the page adds is the **Defaults** section: one row that stages every preference back
to its built-in default. It follows the same rule the Apply/Back pair does — it is drawn
only while there is something to reset, because a row offering to undo nothing is a row
that does nothing. Resetting *stages*; it does not write. So the reset is as reviewable,
and as discardable with Esc, as any other change on the page, and the file is still only
ever written by the one action at the bottom.
:func:`preferences_table` is the same content as a printed table, for the CLI's
``preferences show``.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Optional
from rich import box
from rich.cells import cell_len
from rich.table import Table
from rich.text import Text
from ..core.preferences import (
PREFERENCES,
PreferenceError,
Preferences,
PrefSpec,
by_group,
format_value,
get_spec,
parse_value,
range_hint,
)
from .menus import (
confirm_discard,
exit_rows,
lane_header,
lane_row,
section_heading,
)
from .tui import Choice, Separator
if TYPE_CHECKING:
from ..context import AppContext
#: Menu action sentinels, distinct from the plain-string preference keys the rows carry.
_RESET = "__reset__"
_APPLY = "__apply__"
_CANCEL = "__cancel__"
async def edit_preferences(ctx: "AppContext") -> Optional[dict[str, Any]]:
"""Run the Preferences editor and return the values to save.
Nothing is written here: the staged map goes back to
:class:`~meshterm.tools.preferences.PreferencesTool` to apply and log, the same
division of labour the device-config editor keeps with its own tool.
Args:
ctx: Shared application context (provides the preferences and the UI surface).
Returns:
The staged ``key -> value`` map to write, or ``None`` if the reader left with
nothing staged (or chose to discard what was).
"""
from .tui import CANCEL, SelectScreen
session = getattr(ctx.ui, "session", None)
if session is None: # pragma: no cover - guarded by the menu-only caller
raise RuntimeError("the preferences editor is only available in the menu")
prefs = ctx.preferences
pending: dict[str, Any] = {} # preference key -> staged new value
# The rows *are* the data — each carries its own staged ``current → new`` value and
# the title counts what is staged — so they refresh in place after every round
# (``replace_items``) rather than being rebuilt as a new screen. One screen for the
# whole visit means the typed filter survives editing a preference, not just the cursor.
title, items = _menu_items(prefs, pending)
menu = SelectScreen(
title, items, wrap=False,
footer_hint="↑↓ move · type to filter · Enter select · Esc back",
)
async with session.stay(menu) as visit:
while True:
choice = await visit.result()
if choice is CANCEL: # Esc at the list
choice = _CANCEL
if choice in (None, _CANCEL):
if pending and not await confirm_discard(ctx, len(pending), verb="saving"):
continue # keep editing — the same list, the same place in it
return None
if choice == _APPLY:
return dict(pending) or None
if choice == _RESET:
await _stage_reset(ctx, prefs, pending)
else: # a preference key
await _stage_preference(ctx, prefs, choice, pending)
title, items = _menu_items(prefs, pending)
menu.replace_items(items, title=title)
# --- rendering ----------------------------------------------------------------
def _effective(prefs: Preferences, key: str, pending: dict[str, Any]) -> Any:
"""What ``key`` would be worth if the staged changes were applied right now."""
return pending[key] if key in pending else prefs.get(key)
def _resettable(prefs: Preferences, pending: dict[str, Any]) -> list[str]:
"""The keys whose *effective* value differs from their default — what a reset would touch.
Staged values count, so a preference edited this visit is resettable before it is ever
saved, and one staged back to its own default already isn't.
"""
return [
spec.key
for spec in _all_specs()
if _effective(prefs, spec.key, pending) != spec.default
]
def _all_specs() -> list[PrefSpec]:
"""Every spec in page order (the grouping flattened)."""
return [spec for _, specs in by_group() for spec in specs]
#: Cell cap on *each side* of a VALUE lane's ``current → staged``. The lane is sized to
#: its widest row, and the description lane gets whatever is left — so one long value (the
#: basemap URL, at 36 cells) would otherwise spend a third of the screen on itself and
#: shred every other row's prose down to an ellipsis. Capped, only that row is shortened,
#: and the row still opens to a prompt showing the value in full.
_VALUE_CAP = 20
def _capped(text: str) -> Text:
"""One side of a VALUE lane, shortened to :data:`_VALUE_CAP` with an ellipsis."""
value = Text(text)
value.truncate(_VALUE_CAP, overflow="ellipsis")
return value
def _preference_value(prefs: Preferences, spec: PrefSpec, pending: dict[str, Any]) -> Text:
"""One preference's VALUE lane: ``current [→ staged]``.
The staged arrow is drawn in the warn style so a dirty row stands out at a glance; the
span survives the select highlight, which only tints the row's base style. Both sides
are capped independently, so a staged change never hides behind the value it replaces.
"""
value = _capped(format_value(spec, prefs.get(spec.key)))
if spec.key in pending:
staged = _capped(format_value(spec, pending[spec.key]))
value.append(" → ", style="warn")
value.append_text(Text(staged.plain, style="warn"))
return value
def _menu_items(prefs: Preferences, pending: dict[str, Any]) -> tuple[str, list]:
"""Build the page's title and rows for the current values plus whatever is staged.
The rows sit in three aligned lanes — preference, current value (and any staged new
value), description — under one pinned header, so the page reads as the editor it is.
"""
sections: list[tuple[str, list[tuple[str, Text, str, Any]]]] = []
for group, specs in by_group():
sections.append((
group,
[
(spec.label, _preference_value(prefs, spec, pending), spec.description, spec.key)
for spec in specs
],
))
# The reset row is drawn only while a reset would do something, for the reason the
# Apply/Back pair is drawn only while something is staged (see menus.exit_rows): an
# affordance that undoes nothing is one the reader has to try to learn that.
resettable = _resettable(prefs, pending)
if resettable:
count = len(resettable)
sections.append((
"Defaults",
[(
"Reset to defaults…",
Text(),
f"Return {count} changed values to default"
if count > 1
else "Return the one changed value to default",
_RESET,
)],
))
label_w = max(cell_len(label) for _, rows in sections for label, _, _, _ in rows)
value_w = max(cell_len(value.plain) for _, rows in sections for _, value, _, _ in rows)
# The shared editor header (menus.lane_header), pinned for the whole list: the lanes
# mean the same in every group, so a scrolled row keeps both its column header and its
# section heading overhead.
items: list = [Separator(lambda w: lane_header(label_w, value_w, w), pinned=True)]
# Everything up to the DESCRIPTION lane is the row's identity — which preference, and
# what it is set to — and it always fits. Only the explanation runs past the edge, so
# ←→ slide *it* under a pinned setting and value (Choice.hscroll_from, the same
# arrangement menu_rows gives a command list). Without it the descriptions simply died
# at the fold, which on the PicoCalc's 53 columns is most of every one of them.
description_at = label_w + 2 + value_w + 2
for group, rows in sections:
items.append(section_heading(group))
for label, value, help_text, key in rows:
items.append(Choice(
title=lane_row(label, value, help_text, label_w, value_w),
value=key,
hscroll_from=description_at,
))
# Nothing at all while the page is clean — Esc leaves. With changes staged the pair
# appears below one blank line: Apply is the save action, and Back spells out what
# leaving instead would cost (see menus.exit_rows).
items.extend(exit_rows(len(pending), apply_value=_APPLY, back_value=_CANCEL))
title = "Preferences" + (f" — {len(pending)} staged" if pending else "")
return title, items
#: Widest terminal that still can't hold the DESCRIPTION lane. The three lanes the table
#: can't do without — the key you would type into ``preferences set``, what it is set to,
#: and what it would be if you hadn't — take about eighty cells between them, and the
#: descriptions are the lane that gives way, exactly as they do in the device-config table
#: on a narrow console. They are still on the page, and in the file's own comments.
_DESCRIBE_FROM = 100
def preferences_table(prefs: Preferences, width: int = _DESCRIBE_FROM) -> Table:
"""Build the full preferences table — every value, its default, and what it does.
The printed face of the page, for ``meshterm preferences show``. Unlike the screen it
lists *every* preference's default beside its value, because a terminal print has no
row to open and no reset row to infer the difference from — and it names each
preference by its **key**, since that is what ``preferences set`` takes.
Args:
prefs: The preferences to render.
width: The console's width, which decides whether the DESCRIPTION lane is drawn.
Returns:
A Rich :class:`Table` ready to hand to ``ctx.ui.show``.
"""
# Same frameless SIMPLE_HEAD shape as the device-config table, so the two printed
# settings views read as one family.
table = Table(
title="[accent]Preferences[/accent]",
title_justify="left",
box=box.SIMPLE_HEAD,
show_edge=False,
pad_edge=False,
header_style="muted",
expand=False,
padding=(0, 2, 0, 0),
)
describe = width >= _DESCRIBE_FROM
# The key lane is pinned to the longest key so no key is ever elided: a truncated
# `direct_message_s…` is not a thing anyone can type back. The two value lanes wrap
# instead (the basemap URL is 36 cells on its own), and DESCRIPTION takes the rest.
keys_w = max(len(spec.key) for spec in PREFERENCES) + 2
table.add_column("PREFERENCE", style="muted", no_wrap=True, min_width=keys_w)
table.add_column("VALUE")
table.add_column("DEFAULT", style="muted")
if describe:
table.add_column("DESCRIPTION", style="muted")
def row(cells: list[str]) -> list[str]:
"""Trim a row to the lanes this width actually draws."""
return cells if describe else cells[:-1]
for group, specs in by_group():
table.add_section()
table.add_row(*row([f"[accent]── {group} ──[/accent]", "", "", ""]))
for spec in specs:
value = format_value(spec, prefs.get(spec.key))
# An overridden value is the only thing on the row worth a colour: it is what
# this install disagrees with the code about.
table.add_row(*row([
f" {spec.key}",
f"[warn]{value}[/warn]" if prefs.is_overridden(spec.key) else value,
format_value(spec, spec.default),
spec.description,
]))
return table
# --- staging individual changes -----------------------------------------------
async def _stage_preference(
ctx: "AppContext", prefs: Preferences, key: str, pending: dict[str, Any]
) -> None:
"""Prompt for one preference's new value and stage it.
Staging the value already in force is a no-op, so a row set back to where it started
reads clean again rather than counting as a change that changes nothing.
"""
spec = get_spec(key)
value = await _prompt_value(ctx, spec, _effective(prefs, key, pending))
if value is None:
return
if value == prefs.get(key):
pending.pop(key, None)
else:
pending[key] = value
async def _prompt_value(ctx: "AppContext", spec: PrefSpec, current: Any) -> Any:
"""Prompt for a typed value for ``spec`` in the fitting dialog; ``None`` on cancel."""
if spec.value_type == "bool":
# A straight two-state choice reads best as a button pair, with the current state
# highlighted so Enter changes nothing by accident.
return await ctx.ui.dialog(
spec.description,
[("Off", False), ("On", True)],
title=spec.label,
default=1 if current else 0,
keys={"0": False, "1": True, "n": False, "y": True},
)
if spec.value_type == "enum" and spec.choices is not None:
items = [
Choice(
title=f"{label}" + (" (current)" if value == current else ""),
value=value,
)
for value, label in spec.choices.items()
]
return await ctx.ui.select(
spec.label, items, prompt=f"{spec.description}:", default=current
)
def validate(text: str) -> bool | str:
try:
parse_value(spec, text)
return True
except PreferenceError as exc:
return str(exc)
raw = await ctx.ui.text(
spec.label,
prompt=f"{spec.description}:",
default="" if current is None else str(current),
validate=validate,
help_text=range_hint(spec),
)
return None if raw is None else parse_value(spec, raw)
async def _stage_reset(
ctx: "AppContext", prefs: Preferences, pending: dict[str, Any]
) -> None:
"""Confirm, then stage every changed preference back to its built-in default.
Amber rather than red: nothing is written yet and Esc still discards it, but one press
undoing every preference you have ever changed deserves to be asked about first.
"""
keys = _resettable(prefs, pending)
if not keys: # pragma: no cover - the row is only drawn when there is something to reset
return
plural = "" if len(keys) == 1 else "s"
if await ctx.ui.dialog(
f"Stage {len(keys)} preference{plural} back to their built-in defaults?",
[("Cancel", False), ("Reset", True)],
title="Reset to defaults",
default=1,
danger=True,
) is not True:
return
for key in keys:
default = get_spec(key).default
# Same rule as a hand-edited row: stage the default only where the *saved* value
# isn't already it — where it is, the entry was a staged change and simply goes.
if prefs.get(key) == default:
pending.pop(key, None)
else:
pending[key] = default
__all__ = ["edit_preferences", "preferences_table"]
+2 -2
View File
@@ -416,7 +416,7 @@ async def _apply(
("✗ ", "err"), f"{spec.label} — {reply.strip()} (still staged)"
))
if i < total:
await asyncio.sleep(ctx.settings.trace_cooldown_s)
await asyncio.sleep(ctx.preferences.trace_cooldown_s)
aborted = await _run_under_dialog(ctx, f"Applying — {node.name}", work)
ctx.repo.finish_run(
@@ -461,7 +461,7 @@ async def _read_all(ctx: "AppContext", device: "Device", node: Contact) -> None:
ctx.remote_store.remember_setting(node, spec.key, value)
read += 1
if i < len(specs):
await asyncio.sleep(ctx.settings.trace_cooldown_s)
await asyncio.sleep(ctx.preferences.trace_cooldown_s)
aborted = await _run_under_dialog(ctx, f"Reading — {node.name}", work)
ctx.repo.finish_run(
+3 -3
View File
@@ -1822,7 +1822,7 @@ async def _open_session(
``None`` — the count is session state the screen reads back, not a spec.
"""
nonlocal sample_count
pace = ctx.settings.trace_cooldown_s
pace = ctx.preferences.trace_cooldown_s
items: list = []
for n in SAMPLE_CHOICES:
title = Text(f"{n} trace{'s' if n > 1 else ' '}")
@@ -1910,7 +1910,7 @@ async def _open_session(
device,
target,
candidates,
cooldown_s=ctx.settings.trace_cooldown_s,
cooldown_s=ctx.preferences.trace_cooldown_s,
on_result=on_result,
persist_trace=lambda t: ctx.repo.record_trace(run_id, t),
persist_candidate=lambda o: ctx.repo.record_path_candidate(
@@ -2187,7 +2187,7 @@ async def _open_session(
pick_samples=pick_samples,
width_bytes=lambda: width_bytes,
sample_count=lambda: sample_count,
pace_s=ctx.settings.trace_cooldown_s,
pace_s=ctx.preferences.trace_cooldown_s,
previous=previous,
auto_spec=auto_spec,
auto_source=auto_source,
+12 -4
View File
@@ -44,11 +44,19 @@ def enabled() -> bool:
"""Whether the direct row writer is active.
On by default — it is 2-3x on every navigation keystroke and its output is verified
byte-identical against the stock renderer by reading the console back. Set
``MESHTERM_FASTRENDER=0`` to fall back to prompt_toolkit's renderer, which is the
escape hatch if a terminal ever disagrees.
byte-identical against the stock renderer by reading the console back. The escape
hatch, for a terminal that ever disagrees, is the ``fast_render`` preference;
``MESHTERM_FASTRENDER=0``/``=1`` overrules it for one run without editing anything.
Read at session build, so a change made on the Preferences page takes hold at the
next launch — which is what that preference's description promises.
"""
return os.environ.get("MESHTERM_FASTRENDER", "1") != "0"
from ...core.preferences import current as current_preferences
override = os.environ.get("MESHTERM_FASTRENDER")
if override is not None:
return override != "0"
return bool(current_preferences().fast_render)
class FastRenderer(Renderer):
+11 -4
View File
@@ -228,18 +228,25 @@ def _reclaim_last_column() -> bool:
This is a gate, not a certainty: it is *correct* only when the probe under-reports. On a
terminal whose width probe is already right, the extra column falls off the real screen and
the frame would wrap and tear — which is exactly PicoCalc's exact-width console, so
:data:`~meshterm.platforms.PICOCALC` defaults this off. ``MESHTERM_FULL_WIDTH=0``/``=1``
remains an explicit override on top of the platform default for any terminal that needs
to disagree with its platform's usual verdict (TODO: thread through ``settings`` and the
config editor once confirmed further).
:data:`~meshterm.platforms.PICOCALC` defaults this off.
Three sources, in confidence order, because the person at the keyboard can see the
column and the platform can only guess at it: ``MESHTERM_FULL_WIDTH=0``/``=1`` for a
one-off, then the ``full_width`` preference (``auto`` — the default — steps aside),
then the platform's own verdict.
Read fresh on every call (never cached at import time) so it reflects whichever platform
:func:`~meshterm.platforms.set_platform` installed for this process — see that module's
docstring for why a cached/imported copy of the platform would go stale.
"""
from ...core.preferences import current as current_preferences
override = os.environ.get("MESHTERM_FULL_WIDTH")
if override is not None:
return override != "0"
preferred = current_preferences().full_width
if preferred != "auto":
return preferred == "on"
return get_platform().width_reclaim
#: How many stacked dialog layers the layout can float over the background at once. A fixed
+8 -7
View File
@@ -638,8 +638,8 @@ async def open_tx_optimize(
device_hash=device_hash,
resolve=resolve,
session=session,
tx_min=ctx.settings.tx_opt_min,
tx_max=ctx.settings.tx_opt_max,
tx_min=ctx.preferences.tx_opt_min,
tx_max=ctx.preferences.tx_opt_max,
admin_key=admin_hash,
target_key=target_hex,
)
@@ -690,11 +690,11 @@ async def open_tx_optimize(
async def pick_range() -> None:
"""Pick the sweep's TX window: the configured default, the full range, or typed."""
settings_lo, settings_hi = ctx.settings.tx_opt_min, ctx.settings.tx_opt_max
preferred_lo, preferred_hi = ctx.preferences.tx_opt_min, ctx.preferences.tx_opt_max
items = [
Choice(
title=f"TX {settings_lo}–{settings_hi} — your configured default",
value=(settings_lo, settings_hi),
title=f"TX {preferred_lo}–{preferred_hi} — your configured default",
value=(preferred_lo, preferred_hi),
),
Choice(
title=f"TX {REMOTE_TX_MIN}–{REMOTE_TX_MAX} — the full remote range",
@@ -757,7 +757,7 @@ async def open_tx_optimize(
async def pick_samples() -> None:
"""Pick how many traces each level measures (paced, like the trace tool's)."""
pace = ctx.settings.trace_cooldown_s
pace = ctx.preferences.trace_cooldown_s
items = [
Choice(
title=f"{n} trace{'s' if n > 1 else ' '}"
@@ -908,7 +908,8 @@ async def open_tx_optimize(
coarse_step=screen.step,
samples_per_level=screen.samples,
apply=False, # the decision moves to the dialog, after the evidence is in
cooldown_s=ctx.settings.trace_cooldown_s,
cooldown_s=ctx.preferences.trace_cooldown_s,
snr_tolerance=ctx.preferences.tx_snr_tolerance_db,
on_level=screen.on_level,
on_phase=screen.on_phase,
persist_level=lambda lv: ctx.repo.record_tx_sample(run_id, lv),
+1
View File
@@ -21,6 +21,7 @@ dependencies = [
"rich>=13.7", # modern console rendering (progress, tables, panels)
"markdown-it-py>=3.0", # CommonMark parser behind the written About pages
"prompt_toolkit>=3.0", # full-screen interactive TUI (terminal, input, resize)
"PyYAML>=6.0", # the preferences file (meshterm.core.preferences)
"segno>=1.6", # pure-Python QR codes (channel share links)
"tomli>=2.0; python_version < '3.11'", # TOML config reader on 3.10
"tomli-w>=1.0", # TOML writer for device-config backups
+5 -3
View File
@@ -16,6 +16,7 @@ from tests.conftest import plain as _strip_ansi # THE strip-and-join screen rea
from meshterm.core.channels import DEFAULT_PUBLIC_SECRET, derive_secret
from meshterm.core.config import Settings
from meshterm.core.preferences import Preferences
from meshterm.core.connection import MockDevice
from meshterm.core.events import MeshEvent
from meshterm.core.models import (
@@ -57,6 +58,7 @@ class _StubContext:
self.log = logging.getLogger("test.chat")
self.profile_name = None
self.settings = Settings()
self.preferences = Preferences()
self.events = EventHub(self)
async def device(self) -> MockDevice:
@@ -1586,7 +1588,7 @@ async def test_send_direct_soft_retries_until_acked(repo: Repository) -> None:
"""A DM unacked on the first tries is re-sent, and is recorded delivered once one lands."""
device = _ScriptedDevice([None, None, object()]) # ack only on the third try
ctx = _StubContext(device, repo)
ctx.settings.direct_message_soft_retries = 2 # 1 send + 2 retries = 3 tries
ctx.preferences.direct_message_soft_retries = 2 # 1 send + 2 retries = 3 tries
chat = ChatService(ctx)
sent = await chat.send_direct(_contact(), "hey")
@@ -1599,7 +1601,7 @@ async def test_send_direct_soft_retries_capped_by_setting(repo: Repository) -> N
"""The retry budget stops the resends: an always-unacked DM is tried retries+1 times."""
device = _ScriptedDevice([]) # never acknowledges
ctx = _StubContext(device, repo)
ctx.settings.direct_message_soft_retries = 2
ctx.preferences.direct_message_soft_retries = 2
chat = ChatService(ctx)
sent = await chat.send_direct(_contact(), "hey")
@@ -1612,7 +1614,7 @@ async def test_send_direct_zero_retries_is_one_shot(repo: Repository) -> None:
"""With soft retries disabled a DM is transmitted exactly once, acked or not."""
device = _ScriptedDevice([]) # never acknowledges
ctx = _StubContext(device, repo)
ctx.settings.direct_message_soft_retries = 0
ctx.preferences.direct_message_soft_retries = 0
chat = ChatService(ctx)
sent = await chat.send_direct(_contact(), "hey")
+7 -27
View File
@@ -1,10 +1,14 @@
"""Tests for application settings loading."""
"""Tests for machine-setup loading (where things live, and which device to talk to).
MeshTerm's *behaviour* settings are preferences and live in ``tests/test_preferences.py``;
this file is only what ``config.toml`` still answers for.
"""
from __future__ import annotations
from pathlib import Path
from meshterm.core.config import DIRECT_MESSAGE_MAX_SOFT_RETRIES, Settings
from meshterm.core.config import Settings
def test_connect_on_start_defaults_true() -> None:
@@ -13,36 +17,12 @@ def test_connect_on_start_defaults_true() -> None:
def test_connect_on_start_loads_from_toml(tmp_path: Path) -> None:
"""The connect_on_start preference round-trips from the config file."""
"""Whether the session opens the radio link at launch round-trips from the config file."""
config = tmp_path / "config.toml"
config.write_text("connect_on_start = false\n", encoding="utf-8")
assert Settings.load(config).connect_on_start is False
def test_direct_message_soft_retries_defaults_to_two() -> None:
"""Two soft retries (three tries total) is the out-of-the-box direct-message budget."""
assert Settings().direct_message_soft_retries == 2
assert DIRECT_MESSAGE_MAX_SOFT_RETRIES == 2
def test_direct_message_soft_retries_loads_from_toml(tmp_path: Path) -> None:
"""The soft-retry budget round-trips from the config file."""
config = tmp_path / "config.toml"
config.write_text("direct_message_soft_retries = 1\n", encoding="utf-8")
assert Settings.load(config).direct_message_soft_retries == 1
def test_direct_message_soft_retries_clamped_to_range(tmp_path: Path) -> None:
"""Out-of-range values are clamped to 0..2 so a DM never floods the mesh (or is dropped)."""
hot = tmp_path / "hot.toml"
hot.write_text("direct_message_soft_retries = 9\n", encoding="utf-8")
assert Settings.load(hot).direct_message_soft_retries == DIRECT_MESSAGE_MAX_SOFT_RETRIES
cold = tmp_path / "cold.toml"
cold.write_text("direct_message_soft_retries = -4\n", encoding="utf-8")
assert Settings.load(cold).direct_message_soft_retries == 0
def test_tcp_profile_inferred_and_loaded(tmp_path: Path) -> None:
"""A profile with a ``host`` loads as a TCP profile and exposes its host:port endpoint."""
config = tmp_path / "config.toml"
+15 -1
View File
@@ -45,6 +45,7 @@ from meshterm.core.models import (
TxLevelResult,
utcnow,
)
from meshterm.core.preferences import Preferences
from meshterm.core.watch_store import WatchStore
from meshterm.persistence.repository import DiscoveredPath
from meshterm.platforms import PICOCALC, REGULAR, Platform, set_platform
@@ -76,11 +77,12 @@ from meshterm.ui.message_paths_screen import MessagePathsScreen
from meshterm.ui.node_detail_screen import NodeDetailScreen, _Action, _RoutesView, _Tab
from meshterm.ui.packet_viewer import PacketEntry, PacketViewer
from meshterm.ui.path_composer import PathComposerScreen
from meshterm.ui.preferences import _menu_items as _preference_items
from meshterm.ui.records_screen import RecordDialog
from meshterm.ui.remote_cli import RemoteCliScreen
from meshterm.ui.timemachine_screen import TimeMachineScreen
from meshterm.ui.trace_screen import TraceScreen
from meshterm.ui.tui import Screen, frame
from meshterm.ui.tui import Screen, SelectScreen, frame
from meshterm.ui.tx_screen import TxSweepScreen
from meshterm.ui.walk_screen import WalkScreen
from meshterm.ui.theme import name_style
@@ -552,6 +554,17 @@ def _config_editor_revealed(cols: int, rows: int) -> Screen:
return screen
def _preferences(cols: int, rows: int) -> Screen:
"""The Preferences page at its widest: one saved override, one staged, reset row drawn."""
prefs = Preferences()
prefs.set("trace_cooldown_s", 2.5)
title, items = _preference_items(prefs, {"history_days": 90})
return SelectScreen(
title, items, wrap=False,
footer_hint="↑↓ move · type to filter · Enter select · Esc back",
)
def _about_meshterm(cols: int, rows: int) -> Screen:
return AboutPage("About MeshTerm", about_meshterm())
@@ -592,6 +605,7 @@ _ENTRIES: list[_Entry] = [
_Entry("device_info_revealed", _device_info_revealed),
_Entry("config_editor", _config_editor),
_Entry("config_editor_revealed", _config_editor_revealed),
_Entry("preferences", _preferences),
_Entry("about_meshterm", _about_meshterm),
_Entry("about_author", _about_author),
_Entry("join_discord", _join_discord),
+5 -5
View File
@@ -145,11 +145,11 @@ def test_main_menu_sections_answer_the_menus_own_question() -> None:
by subject: the address book sits with the features that message from it, the
recorded history with the live views it is the past tense of, the two walks with the
topology they walk over, and a setting lands under the scope it changes — this radio,
or someone else's over the mesh.
someone else's over the mesh, or MeshTerm itself.
About MeshTerm is the deliberate exception, and the reason it sorts last: its pages
do nothing at all, so it names a subject and stays out of the workflow the five
doings above it describe.
This app is the last of the three scopes and the reason it sorts last: it holds the
preferences that change the program and the pages that describe it, neither of which
is a thing to do *on the mesh*.
"""
from meshterm.tools import load_all_tools
from meshterm.tools.base import _CATEGORY_ORDER, all_tools
@@ -167,7 +167,7 @@ def test_main_menu_sections_answer_the_menus_own_question() -> None:
"Explore": ["map", "walk", "trace", "trace-path", "records"],
"This node": ["info", "config", "device-actions", "advert"],
"Other nodes": ["repeater-admin", "tx-optimize"],
"About MeshTerm": ["about", "about-author", "discord", "support"],
"This app": ["preferences", "about", "about-author", "discord", "support"],
}
# No section is so big it stops being a grouping (the old Mesh bucket held seven of
# nineteen rows), and none is a bucket of one.
+581
View File
@@ -0,0 +1,581 @@
"""Tests for MeshTerm's own preferences: the registry, the YAML file, and the page.
Three layers, in that order — what a preference *is* (spec, default, validation), where it
is kept (the file, and its tolerance for a hand edit gone wrong), and how it is changed
(the staged page, its reset row, and the gate on the way out) — plus the wiring checks that
keep a preference from being a value the page writes and nothing reads.
"""
from __future__ import annotations
import io
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any, Optional
import pytest
from rich.console import Console
from meshterm.context import AppContext
from meshterm.core.admin_store import AdminStore
from meshterm.core.advert_store import AdvertStore
from meshterm.core.config import Settings
from meshterm.core.device_store import DeviceStore
from meshterm.core.preferences import (
GROUPS,
PREFERENCES,
PreferenceError,
Preferences,
by_group,
format_value,
get_spec,
install,
parse_value,
)
from meshterm.core.watch_store import WatchStore
from meshterm.persistence.repository import Repository
from meshterm.ui.preferences import _menu_items, edit_preferences, preferences_table
from meshterm.ui.theme import active_theme
from tests.conftest import plain as _plain
# -- the registry ----------------------------------------------------------------
def test_every_preference_declares_a_usable_spec() -> None:
"""Keys are unique, groups are real, and every default survives its own validation."""
keys = [spec.key for spec in PREFERENCES]
assert len(keys) == len(set(keys))
for spec in PREFERENCES:
assert spec.group in GROUPS, f"{spec.key} is in an unlisted group"
assert spec.help and spec.label, f"{spec.key} has no label or help"
# The default is the value used unless overridden, so it has to be a value the
# spec would accept from the file or the CLI.
assert parse_value(spec, spec.default) == spec.default
def test_by_group_covers_every_preference_in_group_order() -> None:
"""The page's grouping is the whole registry, in the declared group order."""
grouped = by_group()
assert [group for group, _ in grouped] == [g for g in GROUPS if g in dict(grouped)]
assert sum(len(specs) for _, specs in grouped) == len(PREFERENCES)
def test_defaults_are_what_an_untouched_install_reads() -> None:
"""With nothing overridden, every attribute reads its spec's default."""
prefs = Preferences()
for spec in PREFERENCES:
assert getattr(prefs, spec.key) == spec.default
assert not prefs.is_overridden(spec.key)
assert prefs.overrides() == {}
def test_an_unknown_key_fails_where_it_is_written() -> None:
"""A mistyped preference raises rather than quietly reading nothing."""
prefs = Preferences()
with pytest.raises(AttributeError):
prefs.trace_cooldwn_s # noqa: B018 - the typo is the point
with pytest.raises(PreferenceError):
prefs.get("no_such_preference")
@pytest.mark.parametrize(
"key,raw,expected",
[
("fast_render", "off", False),
("fast_render", "yes", True),
("trace_cooldown_s", "2.5", 2.5),
("history_days", "0", 0),
("advert_flood_hours", "24", 24), # an enum named as text lands on its own value
("full_width", "on", "on"),
],
)
def test_values_parse_from_text(key: str, raw: str, expected: Any) -> None:
"""Text — from the CLI, the file, or a typed prompt — coerces to the spec's type."""
assert parse_value(get_spec(key), raw) == expected
@pytest.mark.parametrize(
"key,raw",
[
("direct_message_soft_retries", "9"), # over the maximum
("map_view_fraction", "0"), # under the minimum
("trace_cooldown_s", "soon"), # not a number at all
("advert_flood_hours", "7"), # not one of the cadences
("fast_render", "maybe"),
],
)
def test_bad_values_are_refused_with_a_readable_message(key: str, raw: str) -> None:
"""Validation failures carry the key and the reason — the text the prompt shows."""
with pytest.raises(PreferenceError) as exc:
parse_value(get_spec(key), raw)
assert key in str(exc.value)
def test_values_format_for_the_lane_they_are_drawn_in() -> None:
"""Booleans read as words, enums as their labels, numbers carry their unit."""
assert format_value(get_spec("fast_render"), True) == "on"
assert format_value(get_spec("fast_render"), False) == "off"
assert format_value(get_spec("advert_flood_hours"), 24) == "daily"
assert format_value(get_spec("advert_flood_hours"), 0) == "off"
assert format_value(get_spec("trace_cooldown_s"), 2.5) == "2.5 s"
assert format_value(get_spec("history_days"), 365) == "365 days"
# -- the file --------------------------------------------------------------------
def test_only_a_disagreement_with_the_default_is_recorded() -> None:
"""Setting a value back to its default clears the override rather than pinning it."""
prefs = Preferences()
prefs.set("history_days", 30)
assert prefs.overrides() == {"history_days": 30}
prefs.set("history_days", get_spec("history_days").default)
assert prefs.overrides() == {}
assert prefs.history_days == 365
def test_the_file_round_trips_and_holds_only_the_overrides(tmp_path: Path) -> None:
"""What is saved is what was changed; everything else comes back from the code."""
path = tmp_path / "preferences.yaml"
prefs = Preferences(path)
prefs.set("trace_cooldown_s", 2.5)
prefs.set("fast_render", False)
prefs.save()
text = path.read_text(encoding="utf-8")
assert "trace_cooldown_s: 2.5" in text
assert "fast_render: false" in text
assert "history_days" not in text # untouched, so not written
reloaded = Preferences.load(path)
assert reloaded.trace_cooldown_s == 2.5
assert reloaded.fast_render is False
assert reloaded.history_days == 365 # the code's default, not a stale copy
def test_the_file_reads_the_way_the_page_does(tmp_path: Path) -> None:
"""Grouped under comment headings, each entry above its help and its default."""
prefs = Preferences(tmp_path / "preferences.yaml")
prefs.set("history_days", 30)
text = prefs.as_yaml()
assert "# --- History kept ---" in text
assert "# Days of overheard history kept" in text
assert "# default: 365 days" in text
# A group with nothing changed in it earns no heading.
assert "# --- Display ---" not in text
def test_an_untouched_file_says_so(tmp_path: Path) -> None:
"""Saving with nothing overridden writes a file that explains its own emptiness."""
prefs = Preferences(tmp_path / "preferences.yaml")
prefs.save()
assert "every preference is at its default" in (tmp_path / "preferences.yaml").read_text(
encoding="utf-8"
)
@pytest.mark.parametrize(
"text",
[
"",
"not a mapping at all",
"history_days: [1, 2, 3]\n", # a container where a scalar belongs
"no_such_preference: 3\n", # a key the registry never had
"history_days: yesterday\n", # right key, unparseable value
"{{{ not yaml", # not even a document
],
)
def test_a_broken_file_costs_the_line_not_the_session(tmp_path: Path, text: str) -> None:
"""A hand edit gone wrong falls back to defaults instead of refusing to start."""
path = tmp_path / "preferences.yaml"
path.write_text(text, encoding="utf-8")
assert Preferences.load(path).history_days == 365
def test_a_good_line_survives_a_bad_one(tmp_path: Path) -> None:
"""One unusable entry is dropped; the rest of the file still applies."""
path = tmp_path / "preferences.yaml"
path.write_text("history_days: soon\ntrace_cooldown_s: 3.0\n", encoding="utf-8")
prefs = Preferences.load(path)
assert prefs.history_days == 365 and prefs.trace_cooldown_s == 3.0
def test_a_missing_file_is_simply_the_defaults(tmp_path: Path) -> None:
"""Nothing has to exist for the app to run: absence *is* the default state."""
prefs = Preferences.load(tmp_path / "never-written.yaml")
assert prefs.overrides() == {} and prefs.fast_render is True
def test_reset_drops_every_override() -> None:
"""Reset returns the whole set to the code's defaults and reports how many it undid."""
prefs = Preferences()
prefs.set("history_days", 30)
prefs.set("trace_cooldown_s", 2.5)
assert prefs.reset() == 2
assert prefs.overrides() == {} and prefs.history_days == 365
def test_an_in_memory_set_refuses_to_save() -> None:
"""A set with no file behind it says so rather than silently dropping the write."""
with pytest.raises(RuntimeError):
Preferences().save()
# -- the page --------------------------------------------------------------------
#: Wide enough that no lane is truncated, so an assertion about a row is about the row and
#: not about the width it was read at. The platform widths are the gallery's job.
_WIDE = 100
def _rows(prefs: Preferences, pending: Optional[dict] = None) -> tuple[str, list[str]]:
"""The page's title and the lines it draws, as the reader sees them."""
from meshterm.ui.tui import SelectScreen
title, items = _menu_items(prefs, pending or {})
screen = SelectScreen(title, items, wrap=False)
screen.note_viewport(len(items) + 4) # no paging: every row on screen at once
return title, [line.strip() for line in _plain(screen.render_body(_WIDE)).splitlines()]
def test_the_page_groups_every_preference_under_its_own_heading() -> None:
"""Each group is a section, and each of its preferences a row under it."""
_, items = _menu_items(Preferences(), {})
_, rows = _rows(Preferences())
values = [item.value for item in items if hasattr(item, "value")]
for group, specs in by_group():
assert any(f"── {group} ──" in row for row in rows), f"{group} has no heading"
for spec in specs:
assert spec.key in values, f"{spec.key} has no row"
assert any(spec.label in row for row in rows), f"{spec.label} is not drawn"
def test_a_clean_page_offers_neither_apply_nor_reset() -> None:
"""Nothing to save and nothing to undo means neither row is drawn (Esc just leaves)."""
title, rows = _rows(Preferences())
assert title == "Preferences"
assert not any("Apply" in row for row in rows)
assert not any("Reset to defaults" in row for row in rows)
assert not any("Defaults" in row for row in rows)
def test_a_changed_preference_brings_out_the_reset_row() -> None:
"""The reset row appears once there is something for it to undo, and counts it."""
prefs = Preferences()
prefs.set("history_days", 30)
_, rows = _rows(prefs)
assert any("Reset to defaults…" in row for row in rows)
assert any("Return the one changed value" in row for row in rows)
def test_a_staged_change_shows_its_arrow_and_the_save_action() -> None:
"""A staged row reads ``current → new``, and the Apply/Back pair appears below."""
title, rows = _rows(Preferences(), {"history_days": 90})
assert title == "Preferences — 1 staged"
assert any("365 days → 90 days" in row for row in rows)
assert any("✓ Apply 1 staged change" in row for row in rows)
assert any("✗ Back — discard staged changes" in row for row in rows)
def test_a_long_value_is_capped_so_the_descriptions_keep_their_lane() -> None:
"""The basemap URL is shortened in the lane rather than eating the prose column."""
_, rows = _rows(Preferences())
basemap = next(row for row in rows if "Basemap source" in row)
assert "https://tiles.openf…" in basemap
assert "TileJSON" in basemap # its description still made it onto the row
def _table_lines(prefs: Preferences, width: int) -> list[str]:
"""``preferences show``'s output at ``width`` columns."""
console = Console(width=width, file=io.StringIO(), theme=active_theme(), legacy_windows=False)
with console.capture() as capture:
console.print(preferences_table(prefs, width))
return _plain(capture.get()).splitlines()
def test_a_description_scrolls_under_a_pinned_setting_and_value() -> None:
"""A row too wide to read slides its DESCRIPTION lane with ←→; the lanes left of it stay."""
from meshterm.ui.tui import SelectScreen
title, items = _menu_items(Preferences(), {})
screen = SelectScreen(title, items, wrap=False)
screen.note_viewport(len(items) + 4)
before = _plain(screen.render_body(72)).splitlines()
cursor = next(line for line in before if line.lstrip().startswith("❯"))
assert cursor.rstrip().endswith("…") # the description runs past the edge
# ←→ is advertised exactly where it would act (SelectScreen folds the atom in itself).
assert "←→ scroll" in screen.footer_hint
for _ in range(4):
screen.handle("right")
scrolled = next(
line for line in _plain(screen.render_body(72)).splitlines()
if line.lstrip().startswith("❯")
)
# The setting and its value are the row's identity and have not moved; only the
# explanation slid, and it now carries the mark saying there is more to its left.
head = cursor.split(" ")[0]
assert scrolled.startswith(head)
assert "…" in scrolled and scrolled != cursor
def test_the_printed_table_names_every_value_and_its_default() -> None:
"""``preferences show`` prints the full value, the default beside it, and marks changes."""
prefs = Preferences()
prefs.set("history_days", 30)
body = "\n".join(_table_lines(prefs, 160))
assert "https://tiles.openfreemap.org/planet" in body # never capped here
assert "30 days" in body and "365 days" in body # value beside its default
assert "── History kept ──" in body
assert "Days of overheard history kept" in body # the description lane, at this width
def test_the_printed_table_never_elides_a_key() -> None:
"""The key is what ``preferences set`` takes, so a narrow console shortens prose instead."""
lines = _table_lines(Preferences(), 79)
assert any("direct_message_soft_retries" in line for line in lines)
assert not any("DESCRIPTION" in line for line in lines) # the lane that gave way
# -- driving the page ------------------------------------------------------------
class _FakeVisit:
"""One round of a visited screen: the script's next ``select`` answer."""
def __init__(self, ui: "_ScriptedUi") -> None:
self._ui = ui
async def result(self) -> Any:
from meshterm.ui.tui.screen import CANCEL
value = self._ui._answer("select")
return CANCEL if value is None else value
class _FakeSession:
"""Enough of :class:`TuiSession` for the page's ``stay`` loop."""
def __init__(self, ui: "_ScriptedUi") -> None:
self.ui = ui
self.pushed: list = []
@asynccontextmanager
async def stay(self, screen: Any) -> Any:
self.pushed.append(screen)
yield _FakeVisit(self.ui)
class _ScriptedUi:
"""A fake UI surface answering every prompt from a FIFO ``(method, answer)`` script."""
def __init__(self, script: list[tuple[str, Any]]) -> None:
self.script = list(script)
self.session = _FakeSession(self)
def _answer(self, method: str) -> Any:
assert self.script, f"unexpected prompt: {method}"
expected, value = self.script.pop(0)
assert expected == method, f"expected {expected} prompt, got {method}"
return value
async def select(self, title: str, items: list, **kwargs: Any) -> Any:
return self._answer("select")
async def dialog(self, prompt: str, buttons: list, **kwargs: Any) -> Any:
return self._answer("dialog")
async def text(self, title: str, **kwargs: Any) -> Optional[str]:
return self._answer("text")
def show(self, *renderables: Any) -> None:
pass
def note(self, markup: str) -> None:
pass
@pytest.fixture()
def ctx(tmp_path: Path) -> AppContext:
"""A mock-backed application context with its own preferences file."""
settings = Settings(config_dir=tmp_path, db_path=tmp_path / "prefs.db")
context = AppContext(
console=Console(file=io.StringIO()),
settings=settings,
repo=Repository(settings.db_path),
device_store=DeviceStore(tmp_path / "devices.json"),
admin_store=AdminStore(tmp_path / "admin.json"),
mock=True,
)
yield context
context.repo.close()
def _install(ctx: AppContext, script: list[tuple[str, Any]]) -> _ScriptedUi:
ui = _ScriptedUi(script)
ctx.ui = ui # type: ignore[assignment]
return ui
async def test_the_page_stages_a_typed_value_and_apply_returns_it(ctx: AppContext) -> None:
"""Editing a row stages it; the Apply row hands the map to the tool to write."""
_install(ctx, [
("select", "history_days"),
("text", "30"),
("select", "__apply__"),
])
assert await edit_preferences(ctx) == {"history_days": 30}
# Nothing reached disk: the page stages, the tool saves.
assert not (ctx.settings.config_dir / "preferences.yaml").exists()
async def test_the_page_unstages_a_value_set_back_to_where_it_started(ctx: AppContext) -> None:
"""Typing the value already in force clears the row instead of staging a no-op."""
_install(ctx, [
("select", "history_days"),
("text", "30"),
("select", "history_days"),
("text", "365"), # back to what is in force
("select", None), # nothing staged — Esc leaves with no discard dialog
])
assert await edit_preferences(ctx) is None
async def test_leaving_with_unsaved_changes_is_gated(ctx: AppContext) -> None:
"""Esc with something staged asks first; "keep editing" returns to the same page."""
_install(ctx, [
("select", "history_days"),
("text", "30"),
("select", None), # Esc
("dialog", "keep"), # ... and think better of it
("select", "__apply__"),
])
assert await edit_preferences(ctx) == {"history_days": 30}
async def test_discarding_at_the_gate_drops_the_changes(ctx: AppContext) -> None:
"""Confirming the discard leaves with nothing, staged values and all."""
_install(ctx, [
("select", "history_days"),
("text", "30"),
("select", None), # Esc
("dialog", "discard"),
])
assert await edit_preferences(ctx) is None
async def test_the_back_row_runs_the_same_gate_as_esc(ctx: AppContext) -> None:
"""``✗ Back — discard staged changes`` is Esc's twin, confirm and all."""
_install(ctx, [
("select", "history_days"),
("text", "30"),
("select", "__cancel__"),
("dialog", "discard"),
])
assert await edit_preferences(ctx) is None
async def test_reset_stages_the_defaults_rather_than_writing_them(ctx: AppContext) -> None:
"""Reset is a staged change like any other: confirmed, reviewable, and discardable."""
ctx.preferences.set("history_days", 30)
ctx.preferences.set("trace_cooldown_s", 2.5)
_install(ctx, [
("select", "__reset__"),
("dialog", True),
("select", "__apply__"),
])
assert await edit_preferences(ctx) == {"history_days": 365, "trace_cooldown_s": 1.0}
# Still only staged — the values in force are untouched until the tool applies them.
assert ctx.preferences.history_days == 30
async def test_cancelling_the_reset_confirm_stages_nothing(ctx: AppContext) -> None:
"""Backing out of the confirm leaves the page exactly as it was."""
ctx.preferences.set("history_days", 30)
_install(ctx, [
("select", "__reset__"),
("dialog", False),
("select", None), # nothing staged, so Esc leaves without a discard dialog
])
assert await edit_preferences(ctx) is None
# -- the tool, and what actually reads a preference ------------------------------
async def test_the_tool_writes_the_staged_values_once(ctx: AppContext) -> None:
"""Applying the page's ops saves the file, and the saved file reads back the same."""
from meshterm.tools.base import get_tool
_install(ctx, [])
tool = get_tool("preferences")
ops = [("set", "history_days", 30), ("set", "full_width", "on")]
result = await tool.run(ctx, {"ops": ops})
assert result.summary == {"changes": 2}
path = ctx.settings.config_dir / "preferences.yaml"
assert path.exists()
assert Preferences.load(path).history_days == 30
async def test_the_tool_reports_a_bad_value_rather_than_writing_it(ctx: AppContext) -> None:
"""A refused value never reaches the file."""
_install(ctx, [])
from meshterm.tools.base import get_tool
with pytest.raises(PreferenceError):
await get_tool("preferences").run(ctx, {"ops": [("set", "history_days", "soon")]})
assert not (ctx.settings.config_dir / "preferences.yaml").exists()
def test_the_context_installs_its_preferences_process_wide(ctx: AppContext) -> None:
"""The render layer, which has no context, still reads this session's values."""
from meshterm.core.preferences import current
assert current() is ctx.preferences
def test_a_newly_watched_node_takes_the_preferred_silence_rule(tmp_path: Path) -> None:
"""The Watchtower's silence preference is what a star actually arms the alarm with."""
prefs = Preferences()
prefs.set("watch_silence_hours", 3)
install(prefs)
try:
store = WatchStore(tmp_path / "watchtower.json")
store.watch("a1b2c3d4e5f6", "Alice")
assert store.watched()["a1b2c3d4e5f6"].silence_hours == 3
finally:
install(Preferences())
def test_a_device_with_no_policy_starts_on_the_preferred_cadence(tmp_path: Path) -> None:
"""The advert preferences are the cadence an unseen device is scheduled at."""
prefs = Preferences()
prefs.set("advert_flood_hours", 168)
install(prefs)
try:
policy = AdvertStore(tmp_path / "adverts.json").load("ab" * 32)
assert policy.flood_hours == 168
finally:
install(Preferences())
def test_the_last_column_preference_overrules_the_platform() -> None:
"""``full_width`` steps aside on ``auto`` and decides otherwise (env still wins over both)."""
from meshterm.platforms import PICOCALC, set_platform
from meshterm.ui.tui.session import _reclaim_last_column
set_platform(PICOCALC) # a platform that defaults the reclaim off
prefs = Preferences()
install(prefs)
try:
assert _reclaim_last_column() is False # auto: the platform's verdict stands
prefs.set("full_width", "on")
assert _reclaim_last_column() is True
prefs.set("full_width", "off")
assert _reclaim_last_column() is False
finally:
install(Preferences())