Jean-Pierre MartineauandClaude Opus 5.5
45697db856
A plain border, the one with no caution tier, was the indigo accent: one more colour on every screen, and it meant nothing. It is now its own theme style, border, in the grey of muted, with title.border and hint.border beside it; the amber and red tiers keep their colours. That covers the base frame, every dialog, the handhelds' title bar, the tab strip, and the CLI's panels. The title bar's clip arrow for a direction with nothing more now dims to faint, since the border and muted are the same grey. On the PicoCalc the border takes the light grey slot. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
352 lines
14 KiB
Python
352 lines
14 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
"""The ``info`` tool: show the identity and the radio settings of the connected device."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections.abc import Callable
|
|
from datetime import datetime, timezone
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import typer
|
|
from rich.panel import Panel
|
|
from rich.text import Text
|
|
|
|
from ..context import AppContext
|
|
from ..core.connection import Device
|
|
from ..core.models import node_type_label
|
|
from .base import Tool, ToolResult, register
|
|
|
|
if TYPE_CHECKING: # pragma: no cover - typing only
|
|
from ..ui.report import Column, Facts
|
|
|
|
|
|
@register
|
|
class InfoTool(Tool):
|
|
"""Show the identity and the radio settings of the device.
|
|
|
|
The list of known contacts is in the separate ``nodes`` tool.
|
|
"""
|
|
|
|
name = "info"
|
|
title = "Device info"
|
|
icon = "📋"
|
|
help = "Show device status, identity, and radio config"
|
|
category = "This node"
|
|
order = 10 # what it is, before the tools that change it
|
|
|
|
async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
|
|
"""Query the device, and render its live status and all its settings.
|
|
|
|
Args:
|
|
ctx: The shared application context.
|
|
params: Not used.
|
|
|
|
Returns:
|
|
A :class:`ToolResult` with the device name in its summary.
|
|
"""
|
|
from rich.console import Group
|
|
|
|
from ..ui.config_editor import cached_snapshot, config_table, has_pin
|
|
|
|
device = await ctx.device()
|
|
snapshot = await cached_snapshot(ctx, device)
|
|
|
|
# On the CLI, this is only the status report: the current condition of the device,
|
|
# one fact on each line. The settings are the answer of `config show`, with the
|
|
# same keys that `config get` and `config set` take. If this command also prints
|
|
# the settings, the same table shows two times, with two different spellings.
|
|
session = getattr(ctx.ui, "session", None)
|
|
if session is None:
|
|
return ToolResult(
|
|
summary={"name": snapshot.get("name")},
|
|
report=(await status_facts(device, snapshot),),
|
|
)
|
|
|
|
from ..ui.device_info_screen import DeviceInfoScreen
|
|
|
|
custom = await device.get_custom_vars()
|
|
# A live status panel (role, firmware, battery, clock, radio and packet statistics)
|
|
# above all the current settings, each with a short explanation. The panel is read
|
|
# one time. Only the table is different between the two states, and it is fast to
|
|
# build again.
|
|
panel = await _status_panel(device, snapshot)
|
|
|
|
def page(reveal_pin: bool) -> Group:
|
|
return Group(panel, Text(""), config_table(snapshot, custom, reveal_pin=reveal_pin))
|
|
|
|
# The page is a separate screen, because it has its own keyboard key: the pairing
|
|
# PIN is masked until ``p`` shows it.
|
|
await session.run_screen(
|
|
DeviceInfoScreen(session, page, title=self.title, conceals=has_pin(snapshot))
|
|
)
|
|
return ToolResult(summary={"name": snapshot.get("name")})
|
|
|
|
def register_cli(self, app: typer.Typer) -> None:
|
|
"""Register the ``info`` subcommand.
|
|
|
|
Args:
|
|
app: The Typer application.
|
|
"""
|
|
from ..cli import run_tool_command
|
|
|
|
@app.command(name=self.name, help=self.help)
|
|
def _info() -> None:
|
|
run_tool_command(self, {})
|
|
|
|
|
|
# -- the status panel -----------------------------------------------------------
|
|
|
|
|
|
async def _read_status(device: Device) -> dict[str, Any]:
|
|
"""Read each live status value that the firmware gives, as a best effort.
|
|
|
|
This is the only place that asks the device. Both faces use what it returns: the panel
|
|
of the menu makes prose rows from these values, and the CLI prints them one fact on
|
|
each line (refer to :func:`_status_panel` and :func:`status_facts`). Each read is
|
|
optional, because firmware that is older than a query gives nothing for it. Thus this
|
|
function works on any device.
|
|
|
|
Args:
|
|
device: The connected device to query.
|
|
|
|
Returns:
|
|
The merged raw readings: ``info``, ``battery``, ``stats``, and ``clock``.
|
|
"""
|
|
return {
|
|
"info": await _try(device.get_device_info) or {},
|
|
"battery": await _try(device.get_battery) or {},
|
|
"stats": await _try(device.get_stats) or {},
|
|
"clock": await _try(device.get_time),
|
|
}
|
|
|
|
|
|
async def _status_panel(device: Device, snapshot: dict) -> Panel:
|
|
"""Build the live status panel: role, firmware, battery, clock, and statistics.
|
|
|
|
This is the menu face of :func:`_read_status`. Each row is a short phrase, not an
|
|
atomic value, because a person reads "-110 dBm noise floor · last RSSI -62 dBm" in one
|
|
step. A script wants the three numbers separately.
|
|
|
|
Args:
|
|
device: The connected device to query.
|
|
snapshot: The settings snapshot that is already built (for the role and identity
|
|
fields).
|
|
|
|
Returns:
|
|
A Rich :class:`Panel` of status rows, each with a label and a value.
|
|
"""
|
|
rows: list[tuple[str, Text]] = []
|
|
readings = await _read_status(device)
|
|
info, battery, stats = readings["info"], readings["battery"], readings["stats"]
|
|
|
|
adv_type = snapshot.get("adv_type")
|
|
if adv_type is not None:
|
|
role = node_type_label(adv_type)
|
|
rows.append(("role", Text(role, style="brand")))
|
|
|
|
if info.get("model"):
|
|
rows.append(("model", Text(str(info["model"]))))
|
|
firmware = _firmware(info)
|
|
if firmware:
|
|
rows.append(("firmware", Text(firmware)))
|
|
|
|
level = battery.get("level")
|
|
if level:
|
|
rows.append(("battery", Text(f"{int(level) / 1000:.2f} V")))
|
|
if battery.get("total_kb"):
|
|
rows.append(("storage", Text(f"{battery.get('used_kb', 0)} / {battery['total_kb']} kB")))
|
|
|
|
clock = readings["clock"]
|
|
if clock:
|
|
stamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(clock))
|
|
rows.append(("clock", Text.assemble(stamp, (f" {_drift(clock)}", "muted"))))
|
|
|
|
if stats.get("uptime_secs") is not None:
|
|
rows.append(("uptime", Text(_uptime(int(stats["uptime_secs"])))))
|
|
if stats.get("noise_floor") is not None:
|
|
parts = Text(f"{stats['noise_floor']} dBm noise floor")
|
|
if stats.get("last_rssi") is not None:
|
|
parts.append(f" · last RSSI {stats['last_rssi']} dBm", style="muted")
|
|
if stats.get("last_snr") is not None:
|
|
parts.append(f" · last SNR {stats['last_snr']:+.1f} dB", style="muted")
|
|
rows.append(("radio", parts))
|
|
if stats.get("tx_air_secs") is not None:
|
|
rows.append(
|
|
("airtime", Text(f"TX {stats['tx_air_secs']} s · RX {stats.get('rx_air_secs', 0)} s"))
|
|
)
|
|
if stats.get("recv") is not None:
|
|
packets = Text(f"{stats.get('sent', 0)} sent · {stats['recv']} received")
|
|
if stats.get("recv_errors"):
|
|
packets.append(f" · {stats['recv_errors']} receive errors", style="warn")
|
|
rows.append(("packets", packets))
|
|
|
|
width = max((len(label) for label, _ in rows), default=0)
|
|
body = Text()
|
|
for i, (label, value) in enumerate(rows):
|
|
if i:
|
|
body.append("\n")
|
|
body.append(f"{label.ljust(width)} ", style="muted")
|
|
body.append_text(value)
|
|
if not rows:
|
|
body = Text("no status reported", style="muted")
|
|
return Panel(body, title="[title.border]Status[/]", border_style="border", expand=False)
|
|
|
|
|
|
def _firmware(info: dict) -> str:
|
|
"""The firmware version and build, joined, or ``""`` if the device reported neither."""
|
|
return " ".join(str(info[k]) for k in ("ver", "fw_build") if info.get(k))
|
|
|
|
|
|
async def status_facts(device: Device, snapshot: dict) -> Facts:
|
|
"""The live status of the device, as one set of facts.
|
|
|
|
This is the CLI face of :func:`_read_status`. The panel writes a phrase in each row,
|
|
but this function writes one value for each key. It puts the unit in the key
|
|
(``battery_v``, ``uptime_s``, ``noise_floor_dbm``), so that no program must get the
|
|
number out of prose. On both faces, a number never has a unit in its value.
|
|
|
|
The three readings in seconds and the clock drift have a **gloss** next to the number:
|
|
``uptime_s 93784 (1d 2h)``. This is the only key/value block that can have a gloss.
|
|
The reason is that it is the only block with no ``get`` and no ``set`` behind it. Thus
|
|
there is no round trip to protect, the key still tells what the number counts, and
|
|
``93784`` is a value that no person can easily understand. The document holds only the
|
|
number, because it has no useful place for a phrase.
|
|
|
|
If the firmware gave no answer for a reading, the plain block **leaves out** that
|
|
reading, and it does not show a dash. These are optional queries. A missing key says
|
|
"this device does not report it", but a dash says that it reported nothing. The
|
|
document keeps the key and writes ``null``. Refer to
|
|
:attr:`~meshterm.ui.report.Facts.omit_absent` for the reason why the two faces are
|
|
different here on purpose.
|
|
|
|
Args:
|
|
device: The connected device to query.
|
|
snapshot: The settings snapshot that is already built (for the role and identity).
|
|
|
|
Returns:
|
|
The facts block.
|
|
"""
|
|
from ..ui import fields
|
|
from ..ui.report import Facts
|
|
|
|
readings = await _read_status(device)
|
|
info, battery, stats = readings["info"], readings["battery"], readings["stats"]
|
|
|
|
level = battery.get("level")
|
|
adv_type = snapshot.get("adv_type")
|
|
clock = readings["clock"]
|
|
values: dict[str, Any] = {
|
|
# The identity is first: which device this is. All the values that it can be set
|
|
# to are the answer of `config show`, not of this command.
|
|
"name": snapshot.get("name"),
|
|
"public_key": str(snapshot.get("public_key") or "").lower() or None,
|
|
"role": node_type_label(adv_type),
|
|
"model": info.get("model"),
|
|
"firmware": _firmware(info) or None,
|
|
"battery_v": round(int(level) / 1000, 2) if level else None,
|
|
"storage_used_kb": battery.get("used_kb", 0) if battery.get("total_kb") else None,
|
|
"storage_total_kb": battery.get("total_kb"),
|
|
"clock_at": (
|
|
datetime.fromtimestamp(clock, tz=timezone.utc).astimezone() if clock else None
|
|
),
|
|
"clock_drift_s": (clock - int(time.time())) if clock else None,
|
|
"uptime_s": stats.get("uptime_secs"),
|
|
"noise_floor_dbm": stats.get("noise_floor"),
|
|
"last_rssi_dbm": stats.get("last_rssi"),
|
|
"last_snr_db": stats.get("last_snr"),
|
|
"tx_air_s": stats.get("tx_air_secs"),
|
|
"rx_air_s": stats.get("rx_air_secs"),
|
|
"packets_sent": stats.get("sent"),
|
|
"packets_received": stats.get("recv"),
|
|
"receive_errors": stats.get("recv_errors"),
|
|
}
|
|
return Facts(
|
|
key="info",
|
|
fields=(
|
|
fields.word("name", "name"),
|
|
fields.hexid("public_key", "public_key"),
|
|
fields.word("role", "role"),
|
|
fields.word("model", "model"),
|
|
fields.word("firmware", "firmware"),
|
|
fields.decimal("battery_v", "battery_v", ".2f"),
|
|
fields.integer("storage_used_kb", "storage_used_kb"),
|
|
fields.integer("storage_total_kb", "storage_total_kb"),
|
|
fields.instant("clock_at", "clock_at"),
|
|
_glossed("clock_drift_s", _drift_gloss),
|
|
_glossed("uptime_s", _span_gloss),
|
|
fields.integer("noise_floor_dbm", "noise_floor_dbm"),
|
|
fields.integer("last_rssi_dbm", "last_rssi_dbm"),
|
|
fields.snr("last_snr_db", "last_snr_db"),
|
|
_glossed("tx_air_s", _span_gloss),
|
|
_glossed("rx_air_s", _span_gloss),
|
|
fields.integer("packets_sent", "packets_sent"),
|
|
fields.integer("packets_received", "packets_received"),
|
|
fields.integer("receive_errors", "receive_errors"),
|
|
),
|
|
values=values,
|
|
omit_absent=True,
|
|
)
|
|
|
|
|
|
def _glossed(key: str, gloss: Callable[[int], str]) -> Column:
|
|
"""A reading in seconds, printed as its number with a readable duration next to it."""
|
|
from ..ui import script
|
|
from ..ui.report import Column, Lane
|
|
|
|
def cell(value: Any) -> str:
|
|
if value is None:
|
|
return script.NONE
|
|
return f"{value} ({gloss(int(value))})"
|
|
|
|
return Column(key=key, lanes=(Lane(header=key, render=cell),))
|
|
|
|
|
|
def _span_gloss(seconds: int) -> str:
|
|
"""``93784`` → ``1d 2h``: a value that a person can easily understand."""
|
|
from ..ui import script
|
|
|
|
return script.duration(seconds)
|
|
|
|
|
|
def _drift_gloss(seconds: int) -> str:
|
|
"""The error of the device clock, in the words that a person uses (fast or slow)."""
|
|
from ..ui import script
|
|
|
|
if abs(seconds) < 2:
|
|
return "in sync"
|
|
return f"{script.duration(seconds)} {'fast' if seconds > 0 else 'slow'}"
|
|
|
|
|
|
async def _try(read) -> Any | None:
|
|
"""Await a device read. Return ``None`` if the firmware does not support it."""
|
|
try:
|
|
return await read()
|
|
except Exception: # noqa: BLE001 - an optional read. If it is absent, that is acceptable.
|
|
return None
|
|
|
|
|
|
def _drift(device_epoch: int) -> str:
|
|
"""Describe the drift of the device clock, compared with the clock of this computer."""
|
|
drift = device_epoch - int(time.time())
|
|
if abs(drift) < 2:
|
|
return "(in sync)"
|
|
direction = "ahead" if drift > 0 else "behind"
|
|
return f"({abs(drift)} s {direction})"
|
|
|
|
|
|
def _uptime(secs: int) -> str:
|
|
"""Render an uptime as ``3d 2h 41m`` (the seconds show only below one minute)."""
|
|
if secs < 60:
|
|
return f"{secs} s"
|
|
days, rem = divmod(secs, 86400)
|
|
hours, rem = divmod(rem, 3600)
|
|
minutes = rem // 60
|
|
parts = []
|
|
if days:
|
|
parts.append(f"{days}d")
|
|
if hours:
|
|
parts.append(f"{hours}h")
|
|
parts.append(f"{minutes}m")
|
|
return " ".join(parts)
|