Add MCP server for interacting with meshtastic devices and testing framework / TUI (#10194)

* Start of MCP server and test suite

* Add MCP server for interacting with meshtastic devices and testing framework / TUI

* Update mcp-server/README.md

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* fix mcp-server review feedback from thread

Agent-Logs-Url: https://github.com/meshtastic/firmware/sessions/91dc128a-ed50-4d07-8bb2-3dc6623a05f7

Co-authored-by: thebentern <9000580+thebentern@users.noreply.github.com>

* Enhance StreamAPI and PhoneAPI for improved log record handling and concurrency control

* Semgrep fixes

* Trunk and semgrep fixes

* optimize pio streaming tee file writes

Agent-Logs-Url: https://github.com/meshtastic/firmware/sessions/04e26c6b-6a2b-45be-bbeb-79ae4d0be633

Co-authored-by: thebentern <9000580+thebentern@users.noreply.github.com>

* chore: remove redundant log handle assignment

Agent-Logs-Url: https://github.com/meshtastic/firmware/sessions/04e26c6b-6a2b-45be-bbeb-79ae4d0be633

Co-authored-by: thebentern <9000580+thebentern@users.noreply.github.com>

* Consolidate type imports and remove placeholder test files

* Add tests for config persistence and more exchange messages

* Refactor position test to validate on-demand request/reply behavior

* Remove  position request/reply test and update README for telemetry behavior

* Fix transmit history file to get removed on factory reset

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
This commit is contained in:
Ben Meadors
2026-04-18 11:29:02 -05:00
co-authored by Copilot thebentern copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
parent 8fd0a7f283
commit 6b15571e14
77 changed files with 10701 additions and 13 deletions
@@ -0,0 +1,3 @@
"""Meshtastic MCP server — device discovery, PlatformIO tooling, and device admin."""
__version__ = "0.1.0"
+11
View File
@@ -0,0 +1,11 @@
"""Entry point for `python -m meshtastic_mcp`."""
from meshtastic_mcp.server import app
def main() -> None:
app.run()
if __name__ == "__main__":
main()
+377
View File
@@ -0,0 +1,377 @@
"""Device administration: owner, config, channels, messaging, admin actions.
All operations use the same `connect()` context manager so port selection,
port-busy detection, and cleanup are handled uniformly.
Config writes use a dot-path: the first segment names a section (e.g.
`"lora"` in LocalConfig or `"mqtt"` in LocalModuleConfig), remaining segments
walk protobuf fields. Enum fields accept their string names (`"US"` for
`lora.region`) so callers don't need to know the numeric values.
"""
from __future__ import annotations
from typing import Any
from google.protobuf import descriptor as pb_descriptor
from google.protobuf import json_format
from meshtastic.protobuf import localonly_pb2
from .connection import connect
class AdminError(RuntimeError):
pass
LOCAL_CONFIG_SECTIONS = {f.name for f in localonly_pb2.LocalConfig.DESCRIPTOR.fields}
MODULE_CONFIG_SECTIONS = {
f.name for f in localonly_pb2.LocalModuleConfig.DESCRIPTOR.fields
}
def _require_confirm(confirm: bool, operation: str) -> None:
if not confirm:
raise AdminError(f"{operation} is destructive and requires confirm=True.")
def _message_to_dict(msg: Any) -> dict[str, Any]:
# `including_default_value_fields` was renamed to
# `always_print_fields_with_no_presence` in protobuf 5.26+. Pick whichever
# kwarg the installed version accepts so we work against both.
kwargs: dict[str, Any] = {"preserving_proto_field_name": True}
import inspect
sig = inspect.signature(json_format.MessageToDict)
if "always_print_fields_with_no_presence" in sig.parameters:
kwargs["always_print_fields_with_no_presence"] = False
elif "including_default_value_fields" in sig.parameters:
kwargs["including_default_value_fields"] = False
return json_format.MessageToDict(msg, **kwargs)
# ---------- owner ----------------------------------------------------------
def set_owner(
long_name: str,
short_name: str | None = None,
port: str | None = None,
) -> dict[str, Any]:
if short_name is not None and len(short_name) > 4:
raise AdminError("short_name must be 4 characters or fewer")
with connect(port=port) as iface:
iface.localNode.setOwner(long_name=long_name, short_name=short_name)
return {
"ok": True,
"long_name": long_name,
"short_name": short_name,
}
# ---------- config reads ---------------------------------------------------
def _section_container(node, section: str) -> tuple[Any, str]:
"""Return (container_message, parent_name) for a section name.
Parent is 'localConfig' or 'moduleConfig' so callers know where to call
writeConfig() after mutating.
"""
if section in LOCAL_CONFIG_SECTIONS:
return getattr(node.localConfig, section), "localConfig"
if section in MODULE_CONFIG_SECTIONS:
return getattr(node.moduleConfig, section), "moduleConfig"
raise AdminError(
f"Unknown config section: {section!r}. "
f"Valid sections: {sorted(LOCAL_CONFIG_SECTIONS | MODULE_CONFIG_SECTIONS)}"
)
def get_config(section: str | None = None, port: str | None = None) -> dict[str, Any]:
"""Read one or all config sections.
`section` may be any name in LocalConfig (device, lora, position, power,
network, display, bluetooth, security) or LocalModuleConfig (mqtt, serial,
telemetry, ...). Omit `section` or pass `"all"` for everything.
"""
with connect(port=port) as iface:
node = iface.localNode
if section in (None, "all"):
lc = _message_to_dict(node.localConfig)
mc = _message_to_dict(node.moduleConfig)
return {
"config": {
"localConfig": lc,
"moduleConfig": mc,
}
}
container, _parent = _section_container(node, section)
return {"config": {section: _message_to_dict(container)}}
# ---------- config writes --------------------------------------------------
def _coerce_enum(field: pb_descriptor.FieldDescriptor, value: Any) -> int:
"""Accept an enum value as either its int or its string name."""
enum_type = field.enum_type
if isinstance(value, bool):
raise AdminError(f"{field.name}: expected enum {enum_type.name}, got bool")
if isinstance(value, int):
if enum_type.values_by_number.get(value) is None:
raise AdminError(
f"{field.name}: {value} is not a valid {enum_type.name} value"
)
return value
if isinstance(value, str):
upper = value.upper()
ev = enum_type.values_by_name.get(upper)
if ev is None:
valid = sorted(enum_type.values_by_name.keys())
raise AdminError(
f"{field.name}: {value!r} is not a valid {enum_type.name}. "
f"Valid: {valid}"
)
return ev.number
raise AdminError(
f"{field.name}: expected enum {enum_type.name}, got {type(value).__name__}"
)
def _coerce_scalar(field: pb_descriptor.FieldDescriptor, value: Any) -> Any:
t = field.type
FT = pb_descriptor.FieldDescriptor
if t == FT.TYPE_ENUM:
return _coerce_enum(field, value)
if t == FT.TYPE_BOOL:
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.strip().lower() in ("true", "yes", "1", "on")
if isinstance(value, int):
return bool(value)
if t in (
FT.TYPE_INT32,
FT.TYPE_INT64,
FT.TYPE_UINT32,
FT.TYPE_UINT64,
FT.TYPE_SINT32,
FT.TYPE_SINT64,
FT.TYPE_FIXED32,
FT.TYPE_FIXED64,
):
return int(value)
if t in (FT.TYPE_FLOAT, FT.TYPE_DOUBLE):
return float(value)
if t == FT.TYPE_STRING:
return str(value)
if t == FT.TYPE_BYTES:
if isinstance(value, (bytes, bytearray)):
return bytes(value)
return str(value).encode("utf-8")
raise AdminError(
f"{field.name}: unsupported field type {t}. Use raw protobuf for this field."
)
def _walk_to_field(
root_msg: Any, path_segments: list[str]
) -> tuple[Any, pb_descriptor.FieldDescriptor]:
"""Walk `root_msg` by field names until the leaf; return (parent_msg, leaf_field_descriptor)."""
msg = root_msg
for i, name in enumerate(path_segments):
desc = msg.DESCRIPTOR
field = desc.fields_by_name.get(name)
if field is None:
trail = ".".join(path_segments[:i] or ["<root>"])
valid = [f.name for f in desc.fields]
raise AdminError(f"No field {name!r} in {trail}. Valid: {valid}")
is_last = i == len(path_segments) - 1
if is_last:
return msg, field
if field.type != pb_descriptor.FieldDescriptor.TYPE_MESSAGE:
raise AdminError(
f"{'.'.join(path_segments[:i+1])} is a scalar; cannot descend into it"
)
msg = getattr(msg, name)
# path_segments was empty
raise AdminError("Empty config path")
def set_config(path: str, value: Any, port: str | None = None) -> dict[str, Any]:
"""Set a single config field by dot-path and write it to the device.
Examples:
set_config("lora.region", "US")
set_config("lora.modem_preset", "LONG_FAST")
set_config("device.role", "ROUTER")
set_config("mqtt.enabled", True)
set_config("mqtt.address", "mqtt.example.com")
"""
segments = [s for s in path.split(".") if s]
if not segments:
raise AdminError("path cannot be empty")
section = segments[0]
with connect(port=port) as iface:
node = iface.localNode
container, parent_name = _section_container(node, section)
# Treat the section as the root; the rest of the path walks into it.
leaf_parent, field = _walk_to_field(container, segments[1:] or [])
# Use `is_repeated` (modern upb protobuf API) rather than the
# deprecated `label == LABEL_REPEATED` check — the C-extension
# FieldDescriptor in protobuf >= 5.x doesn't expose `.label` at
# all, and `is_repeated` is the supported replacement that works
# across both the pure-python and upb backends.
if field.is_repeated:
raise AdminError(
f"{path!r} is a repeated field; v1 only supports scalar sets. "
"Use the raw meshtastic CLI for now."
)
old_raw = getattr(leaf_parent, field.name)
coerced = _coerce_scalar(field, value)
try:
setattr(leaf_parent, field.name, coerced)
except (TypeError, ValueError) as exc:
raise AdminError(f"{path}: {exc}") from exc
node.writeConfig(section)
# Stringify enums for the response (so the caller can see the change in
# the same vocabulary they used to set it).
if field.type == pb_descriptor.FieldDescriptor.TYPE_ENUM:
try:
old_display = field.enum_type.values_by_number[old_raw].name
new_display = field.enum_type.values_by_number[coerced].name
except Exception:
old_display, new_display = old_raw, coerced
else:
old_display, new_display = old_raw, coerced
return {
"ok": True,
"path": path,
"section": section,
"parent": parent_name,
"old_value": old_display,
"new_value": new_display,
}
# ---------- channels -------------------------------------------------------
def get_channel_url(
include_all: bool = False, port: str | None = None
) -> dict[str, Any]:
with connect(port=port) as iface:
url = iface.localNode.getURL(includeAll=include_all)
return {"url": url}
def set_channel_url(url: str, port: str | None = None) -> dict[str, Any]:
with connect(port=port) as iface:
# setURL replaces the channel set from the URL's contents. It does not
# return a count; we infer by counting non-DISABLED channels after.
iface.localNode.setURL(url)
channels = iface.localNode.channels or []
active = sum(1 for c in channels if getattr(c, "role", 0) != 0)
return {"ok": True, "channels_imported": active}
# ---------- messaging ------------------------------------------------------
def send_text(
text: str,
to: str | int | None = None,
channel_index: int = 0,
want_ack: bool = False,
port: str | None = None,
) -> dict[str, Any]:
destination = to if to is not None else "^all"
with connect(port=port) as iface:
packet = iface.sendText(
text,
destinationId=destination,
wantAck=want_ack,
channelIndex=channel_index,
)
packet_id = getattr(packet, "id", None)
return {"ok": True, "packet_id": packet_id, "destination": destination}
# ---------- diagnostics ----------------------------------------------------
def set_debug_log_api(enabled: bool, port: str | None = None) -> dict[str, Any]:
"""Toggle `config.security.debug_log_api_enabled` on the local node.
When enabled, firmware emits log lines as protobuf `LogRecord` messages
over the StreamAPI instead of raw text. meshtastic-python surfaces them
on pubsub topic `meshtastic.log.line`, which flows through the SAME
SerialInterface our tests already hold open — no `pio device monitor`
needed, no port-contention with admin/info calls.
Firmware gate: `src/SerialConsole.cpp` (`usingProtobufs &&
config.security.debug_log_api_enabled`). Setting persists in NVS; it
survives reboot. `factory_reset(full=False)` clears it unless it's
re-applied after reset.
Previously-documented concurrency hazard (emitLogRecord sharing the
main packet-emission buffers) has been fixed — see `StreamAPI.h`
where the log path now owns dedicated `fromRadioScratchLog` /
`txBufLog` buffers, and `StreamAPI::emitTxBuffer` +
`StreamAPI::emitLogRecord` both serialize their `stream->write`
calls via `streamLock`. Leaving the flag on under traffic is safe.
"""
with connect(port=port) as iface:
sec = iface.localNode.localConfig.security
sec.debug_log_api_enabled = bool(enabled)
iface.localNode.writeConfig("security")
return {"ok": True, "debug_log_api_enabled": bool(enabled)}
# ---------- admin actions --------------------------------------------------
def reboot(
port: str | None = None, confirm: bool = False, seconds: int = 10
) -> dict[str, Any]:
_require_confirm(confirm, "reboot")
with connect(port=port) as iface:
iface.localNode.reboot(secs=seconds)
return {"ok": True, "rebooting_in_s": seconds}
def shutdown(
port: str | None = None, confirm: bool = False, seconds: int = 10
) -> dict[str, Any]:
_require_confirm(confirm, "shutdown")
with connect(port=port) as iface:
iface.localNode.shutdown(secs=seconds)
return {"ok": True, "shutting_down_in_s": seconds}
def factory_reset(
port: str | None = None, confirm: bool = False, full: bool = False
) -> dict[str, Any]:
"""Tell the node to factory-reset its config.
Works around a meshtastic-python 2.7.8 bug: `Node.factoryReset(full=True)`
internally does `p.factory_reset_config = True` where the field is
int32. protobuf 5.x rejects bool→int assignment as a TypeError. We build
the AdminMessage directly with int values (1=non-full, 2=full) and call
`_sendAdmin` to sidestep the SDK bug entirely.
"""
_require_confirm(confirm, "factory_reset")
from meshtastic.protobuf import admin_pb2 # type: ignore[import-untyped]
with connect(port=port) as iface:
msg = admin_pb2.AdminMessage()
msg.factory_reset_config = 2 if full else 1
iface.localNode._sendAdmin(msg)
return {"ok": True, "full": full}
+159
View File
@@ -0,0 +1,159 @@
"""Board / PlatformIO env enumeration.
Parses `pio project config --json-output` — a nested list of
`[section_name, [[key, value], ...]]` pairs — into a dict keyed by env name,
extracting the `custom_meshtastic_*` metadata the firmware variants expose.
The parsed config is cached and invalidated when `platformio.ini`'s mtime
changes, so subsequent calls don't pay the 12s pio startup cost.
"""
from __future__ import annotations
import threading
from typing import Any
from . import config, pio
_CACHE_LOCK = threading.Lock()
_CACHE: dict[str, Any] = {"mtime": None, "envs": None}
def _parse_bool(value: Any) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.strip().lower() in ("true", "yes", "1", "on")
return bool(value)
def _parse_int(value: Any) -> int | None:
try:
return int(value)
except (TypeError, ValueError):
return None
def _parse_tags(value: Any) -> list[str]:
if value is None:
return []
if isinstance(value, list):
return [str(v).strip() for v in value if str(v).strip()]
return [t.strip() for t in str(value).replace(",", " ").split() if t.strip()]
def _env_record(env_name: str, items: list[list[Any]]) -> dict[str, Any]:
"""Build a normalized dict for one env section."""
d = dict(items)
return {
"env": env_name,
"architecture": d.get("custom_meshtastic_architecture"),
"hw_model": _parse_int(d.get("custom_meshtastic_hw_model")),
"hw_model_slug": d.get("custom_meshtastic_hw_model_slug"),
"display_name": d.get("custom_meshtastic_display_name"),
"actively_supported": _parse_bool(
d.get("custom_meshtastic_actively_supported")
),
"support_level": _parse_int(d.get("custom_meshtastic_support_level")),
"board_level": d.get("board_level"), # "pr", "extra", or None
"tags": _parse_tags(d.get("custom_meshtastic_tags")),
"images": _parse_tags(d.get("custom_meshtastic_images")),
"board": d.get("board"),
"upload_speed": _parse_int(d.get("upload_speed")),
"upload_protocol": d.get("upload_protocol"),
"monitor_speed": _parse_int(d.get("monitor_speed")),
"monitor_filters": d.get("monitor_filters") or [],
"_raw": d, # Full dict for get_board
}
def _load_all() -> dict[str, dict[str, Any]]:
"""Parse `pio project config` into `{env_name: record}`."""
raw = pio.run_json(["project", "config"], timeout=pio.TIMEOUT_PROJECT_CONFIG)
result: dict[str, dict[str, Any]] = {}
for section_name, items in raw:
if not isinstance(section_name, str) or not section_name.startswith("env:"):
continue
env_name = section_name.split(":", 1)[1]
result[env_name] = _env_record(env_name, items)
return result
def _get_cached() -> dict[str, dict[str, Any]]:
root = config.firmware_root()
platformio_ini = root / "platformio.ini"
try:
mtime = platformio_ini.stat().st_mtime
except FileNotFoundError:
mtime = None
with _CACHE_LOCK:
if _CACHE["envs"] is not None and _CACHE["mtime"] == mtime:
return _CACHE["envs"]
envs = _load_all()
_CACHE["envs"] = envs
_CACHE["mtime"] = mtime
return envs
def invalidate_cache() -> None:
with _CACHE_LOCK:
_CACHE["envs"] = None
_CACHE["mtime"] = None
def _public_record(rec: dict[str, Any]) -> dict[str, Any]:
"""Strip the `_raw` field for list outputs."""
return {k: v for k, v in rec.items() if not k.startswith("_")}
def list_boards(
architecture: str | None = None,
actively_supported_only: bool = False,
query: str | None = None,
board_level: str | None = None, # "release" | "pr" | "extra"
) -> list[dict[str, Any]]:
"""Enumerate PlatformIO envs with Meshtastic metadata.
Filters are cumulative (AND). `board_level="release"` means envs with no
explicit `board_level` set (the default release targets).
"""
envs = _get_cached()
q = query.lower().strip() if query else None
out = []
for rec in envs.values():
if architecture and rec.get("architecture") != architecture:
continue
if actively_supported_only and not rec.get("actively_supported"):
continue
if board_level is not None:
rec_level = rec.get("board_level")
if board_level == "release":
if rec_level not in (None, ""):
continue
elif rec_level != board_level:
continue
if q:
display = (rec.get("display_name") or "").lower()
env_name = rec.get("env", "").lower()
slug = (rec.get("hw_model_slug") or "").lower()
if q not in display and q not in env_name and q not in slug:
continue
out.append(_public_record(rec))
out.sort(key=lambda r: (r.get("architecture") or "", r.get("env")))
return out
def get_board(env: str) -> dict[str, Any]:
"""Full metadata for one env, including the raw pio config dict."""
envs = _get_cached()
rec = envs.get(env)
if rec is None:
raise KeyError(
f"Unknown env: {env!r}. Use list_boards() to see available envs."
)
public = _public_record(rec)
public["raw_config"] = rec["_raw"]
return public
@@ -0,0 +1,6 @@
"""Command-line entry points that sit alongside the MCP server.
Modules here are loaded on-demand by `[project.scripts]` entries in
`pyproject.toml`. They are NOT imported by `meshtastic_mcp.server` or the
admin/info tool surface — the MCP server stays pure stdio JSON-RPC.
"""
@@ -0,0 +1,73 @@
"""Flash progress log tailer for ``meshtastic-mcp-test-tui``.
``pio.py`` / ``hw_tools.py`` tee subprocess output (``pio run -t upload``,
``esptool erase_flash``, ``nrfutil dfu``, etc.) to ``tests/flash.log``
line-by-line as it arrives — controlled by the ``MESHTASTIC_MCP_FLASH_LOG``
env var that ``run-tests.sh`` sets. The TUI tails that file so the operator
sees live flash progress in the pytest pane instead of 3 minutes of silence
during ``test_00_bake``.
Separate from ``_fwlog.py`` because that one parses JSONL, this one
streams plain text lines. Same daemon-thread + EOF-backoff structure.
"""
from __future__ import annotations
import pathlib
import threading
import time
from typing import Callable
class FlashLogTailer(threading.Thread):
"""Tail a plain-text log file, publish each stripped line via ``post``.
``post`` is invoked with a single ``str`` for every new line. Lines are
stripped of trailing newlines; empty lines after stripping are dropped.
The file may not exist yet when this thread starts — it's truncated by
``run-tests.sh`` at session start, but if the tailer races the shell,
we tolerate FileNotFoundError for up to ``wait_s`` seconds.
"""
def __init__(
self,
path: pathlib.Path,
post: Callable[[str], None],
stop: threading.Event,
*,
wait_s: float = 30.0,
) -> None:
super().__init__(daemon=True, name="flashlog-tail")
self._path = path
self._post = post
self._stop = stop
self._wait_s = wait_s
def run(self) -> None:
deadline = time.monotonic() + self._wait_s
while not self._path.is_file():
if self._stop.is_set() or time.monotonic() > deadline:
return
time.sleep(0.1)
try:
fh = self._path.open("r", encoding="utf-8", errors="replace")
except OSError:
return
try:
while not self._stop.is_set():
line = fh.readline()
if not line:
time.sleep(0.05)
continue
line = line.rstrip("\r\n")
if not line:
continue
try:
self._post(line)
except Exception:
# A post failure (e.g. closed app) is terminal for this
# thread but we still want to close the file handle.
return
finally:
fh.close()
@@ -0,0 +1,96 @@
"""Firmware log tail worker for ``meshtastic-mcp-test-tui``.
Complements v1's reportlog-tail worker. ``tests/conftest.py`` owns a
session-scoped autouse fixture (``_firmware_log_stream``) that mirrors
every ``meshtastic.log.line`` pubsub event to ``tests/fwlog.jsonl`` —
one JSON object per line:
{"ts": 1729100000.123, "port": "/dev/cu.usbmodem1101", "line": "..."}
The TUI tails that file from a worker thread; each new line becomes a
:class:`FirmwareLogLine` message posted to the App. Same pattern as the
reportlog tail worker — truncate on launch, tolerate missing file for
30 s, back off at EOF.
Kept in its own module so the (large) ``test_tui.py`` stays focused on
the Textual App shell.
"""
from __future__ import annotations
import json
import pathlib
import threading
import time
from typing import Any, Callable
class FirmwareLogTailer(threading.Thread):
"""Tail ``tests/fwlog.jsonl``, publish parsed records via ``post``.
``post`` is the App's ``post_message`` (or any callable that accepts a
single payload arg). We pass parsed dicts rather than constructing
Textual Message objects here — keeps this module free of the
textual dependency so it's unit-testable in a bare venv.
Parameters
----------
path:
Path to ``tests/fwlog.jsonl``. The file may not exist yet at
startup — pytest only creates it once the session fixture runs.
post:
Callable invoked with a dict ``{"ts", "port", "line"}`` for every
new line parsed from the file.
stop:
An event the App sets to signal shutdown.
wait_s:
How long to poll for the file's creation before giving up. Default
30 s; pytest collection on a cold cache can be slow.
"""
def __init__(
self,
path: pathlib.Path,
post: Callable[[dict[str, Any]], None],
stop: threading.Event,
*,
wait_s: float = 30.0,
) -> None:
super().__init__(daemon=True, name="fwlog-tail")
self._path = path
self._post = post
self._stop = stop
self._wait_s = wait_s
def run(self) -> None:
deadline = time.monotonic() + self._wait_s
while not self._path.is_file():
if self._stop.is_set() or time.monotonic() > deadline:
return
time.sleep(0.1)
try:
fh = self._path.open("r", encoding="utf-8")
except OSError:
return
try:
while not self._stop.is_set():
line = fh.readline()
if not line:
time.sleep(0.05)
continue
line = line.strip()
if not line:
continue
try:
record = json.loads(line)
except json.JSONDecodeError:
continue
# Defensive: require the three fields we rely on.
if not isinstance(record, dict):
continue
if "line" not in record:
continue
self._post(record)
finally:
fh.close()
@@ -0,0 +1,127 @@
"""Cross-run history for ``meshtastic-mcp-test-tui``.
Persists one JSON object per pytest run to
``mcp-server/tests/.history/runs.jsonl``. The TUI reads the last N
entries on launch to render a duration sparkline in the header — a
quick read on whether the suite is slowing down over time.
Schema (keep small; the file can grow for months):
{"run": 42, "ts": 1729100000.0, "duration_s": 387.2,
"passed": 52, "failed": 0, "skipped": 23, "exit_code": 0,
"seed": "mcp-user-host"}
"""
from __future__ import annotations
import json
import pathlib
import time
from dataclasses import asdict, dataclass
from typing import Iterable
# Sparkline glyphs, low → high. 8 levels is the Unicode convention.
_SPARK_BLOCKS = "▁▂▃▄▅▆▇█"
@dataclass
class RunRecord:
run: int
ts: float
duration_s: float
passed: int
failed: int
skipped: int
exit_code: int
seed: str
class HistoryStore:
"""Append-only JSONL store with bounded read.
Writes are fsynced after each append (the file is tiny; fsync cost
is negligible and protects against truncation on a crash).
"""
def __init__(self, path: pathlib.Path, *, keep_last: int = 50) -> None:
self._path = path
self._keep_last = keep_last
def append(self, record: RunRecord) -> None:
try:
self._path.parent.mkdir(parents=True, exist_ok=True)
with self._path.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(asdict(record)) + "\n")
fh.flush()
except Exception:
# Non-fatal: history is cosmetic.
pass
def read_recent(self) -> list[RunRecord]:
"""Return the last ``keep_last`` records in chronological order."""
if not self._path.is_file():
return []
try:
lines = self._path.read_text(encoding="utf-8").splitlines()
except OSError:
return []
out: list[RunRecord] = []
# Parse tail-first so we don't waste work on a huge history.
for line in lines[-self._keep_last :]:
line = line.strip()
if not line:
continue
try:
raw = json.loads(line)
except json.JSONDecodeError:
continue
try:
out.append(RunRecord(**raw))
except TypeError:
# Schema drift; skip the record rather than crash.
continue
return out
def record_run(
self,
*,
run: int,
duration_s: float,
passed: int,
failed: int,
skipped: int,
exit_code: int,
seed: str,
) -> RunRecord:
rec = RunRecord(
run=run,
ts=time.time(),
duration_s=float(duration_s),
passed=int(passed),
failed=int(failed),
skipped=int(skipped),
exit_code=int(exit_code),
seed=seed,
)
self.append(rec)
return rec
def sparkline(values: Iterable[float], *, width: int = 20) -> str:
"""Render a Unicode block-character sparkline from the last ``width`` values.
Returns an empty string for empty input so the header handles
"no history yet" gracefully.
"""
buf = [v for v in values if v >= 0][-width:]
if not buf:
return ""
lo, hi = min(buf), max(buf)
if hi - lo < 1e-9:
return _SPARK_BLOCKS[len(_SPARK_BLOCKS) // 2] * len(buf)
n = len(_SPARK_BLOCKS) - 1
out = []
for v in buf:
idx = int(round((v - lo) / (hi - lo) * n))
out.append(_SPARK_BLOCKS[max(0, min(n, idx))])
return "".join(out)
@@ -0,0 +1,214 @@
"""Reproducer bundle builder for ``meshtastic-mcp-test-tui``.
When the operator presses ``x`` on a failed test leaf, we package the
minimum viable failure context into a tarball under
``mcp-server/tests/reproducers/``:
::
repro-<ts>-<short_nodeid>.tar.gz
├── README.md human-readable overview
├── test_report.json the failing TestReport event from reportlog
├── fwlog.jsonl firmware log filtered to the failure window
├── devices.json per-device device_info + lora config snapshot
└── env.json seed, run #, pytest version, platform, hostname
Separate module so the logic can be unit-tested without Textual. The
TUI glue is thin — one key binding calls :func:`build_reproducer_bundle`
with the focused test's state and shows the path in a modal.
"""
from __future__ import annotations
import io
import json
import pathlib
import platform
import re
import socket
import tarfile
import time
from dataclasses import dataclass
from typing import Any, Iterable
@dataclass
class ReproContext:
"""Everything :func:`build_reproducer_bundle` needs. Shaped to map
cleanly onto the state the TUI already tracks — no extra data
collection required at export time."""
nodeid: str
longrepr: str
sections: list[tuple[str, str]]
start_ts: float | None
stop_ts: float | None
seed: str
run_number: int
exit_code: int | None
fwlog_path: pathlib.Path
output_dir: pathlib.Path
extra_device_rows: list[dict[str, Any]] # [{role, port, info, ...}, ...]
def _short_nodeid(nodeid: str) -> str:
"""Collapse a pytest nodeid into a filename-safe slug (<= 60 chars)."""
# Drop the file path prefix; keep test name + parametrization.
tail = nodeid.split("::", 1)[-1] if "::" in nodeid else nodeid
slug = re.sub(r"[^A-Za-z0-9_.\-]", "_", tail)
return slug[:60].strip("_.-") or "test"
def _filtered_fwlog(
fwlog_path: pathlib.Path,
start_ts: float | None,
stop_ts: float | None,
*,
pad_s: float = 5.0,
) -> bytes:
"""Return fwlog.jsonl lines whose ``ts`` lies in [start-pad, stop+pad]."""
if not fwlog_path.is_file():
return b""
if start_ts is None or stop_ts is None:
# Without a time window, include the whole file — rare; happens
# when a test fails in setup before pytest emitted a start ts.
try:
return fwlog_path.read_bytes()
except OSError:
return b""
lo, hi = start_ts - pad_s, stop_ts + pad_s
out = io.BytesIO()
try:
with fwlog_path.open("r", encoding="utf-8") as fh:
for line in fh:
stripped = line.strip()
if not stripped:
continue
try:
record = json.loads(stripped)
except json.JSONDecodeError:
continue
ts = record.get("ts")
if not isinstance(ts, (int, float)):
continue
if lo <= ts <= hi:
out.write(line.encode("utf-8"))
except OSError:
return b""
return out.getvalue()
def _readme(ctx: ReproContext) -> str:
t = time.strftime("%Y-%m-%d %H:%M:%S %Z", time.localtime())
return f"""# Reproducer bundle
Exported by `meshtastic-mcp-test-tui` on {t}.
## Failing test
- **nodeid:** `{ctx.nodeid}`
- **seed:** `{ctx.seed}`
- **run #:** {ctx.run_number}
- **suite exit code (at export time):** {ctx.exit_code if ctx.exit_code is not None else "in progress"}
## Files in this archive
| File | Contents |
|---|---|
| `test_report.json` | The pytest-reportlog `TestReport` event for the failing test — includes `longrepr`, captured `sections` (stdout/stderr/log), `duration`, `location`, `keywords`. |
| `fwlog.jsonl` | Firmware log lines (from `meshtastic.log.line` pubsub) filtered to [start5s, stop+5s] around the test's run window. Each line is `{{ts, port, line}}`. |
| `devices.json` | Per-device snapshot at export time: `device_info` + `lora` config per detected role. |
| `env.json` | Python version, platform, hostname, seed, run number. |
## How to triage
1. Open `test_report.json` and read `longrepr` + `sections` — most failures explain themselves there.
2. If the failure is a mesh/telemetry assertion, `fwlog.jsonl` is where the answer usually lives. Grep for `Error=`, `NAK`, `PKI_UNKNOWN_PUBKEY`, `Skip send`, `Guru Meditation`, or the uptime timestamps around the assertion event.
3. Compare `devices.json` against the expected state (e.g. `num_nodes >= 2`, `primary_channel == "McpTest"`, `region == "US"`). If fields disagree with the seed-derived USERPREFS profile, the device probably wasn't baked with this session's profile.
## Reproducing locally
```bash
cd mcp-server
MESHTASTIC_MCP_SEED='{ctx.seed}' .venv/bin/pytest '{ctx.nodeid}' --tb=long -v
```
"""
def build_reproducer_bundle(ctx: ReproContext) -> pathlib.Path:
"""Build a tarball under ``ctx.output_dir`` and return its path.
Parent dirs are created as needed. Errors during optional sections
(devices, env) are swallowed — the bundle is still useful without
them; refusing to export because the device poller had a hiccup
would be worse than the export missing a file.
"""
ctx.output_dir.mkdir(parents=True, exist_ok=True)
ts = int(time.time())
slug = _short_nodeid(ctx.nodeid)
archive_path = ctx.output_dir / f"repro-{ts}-{slug}.tar.gz"
with tarfile.open(archive_path, "w:gz") as tar:
def _add(name: str, data: bytes) -> None:
info = tarfile.TarInfo(name=name)
info.size = len(data)
info.mtime = ts
tar.addfile(info, io.BytesIO(data))
# README
_add("README.md", _readme(ctx).encode("utf-8"))
# test_report.json — reconstruct from the fields the TUI stashes.
test_report = {
"nodeid": ctx.nodeid,
"outcome": "failed",
"longrepr": ctx.longrepr,
"sections": [list(s) for s in ctx.sections],
"start": ctx.start_ts,
"stop": ctx.stop_ts,
}
_add(
"test_report.json",
json.dumps(test_report, indent=2, default=str).encode("utf-8"),
)
# fwlog.jsonl (filtered)
_add("fwlog.jsonl", _filtered_fwlog(ctx.fwlog_path, ctx.start_ts, ctx.stop_ts))
# devices.json
try:
devices_payload = json.dumps(
ctx.extra_device_rows or [], indent=2, default=str
)
except Exception:
devices_payload = "[]"
_add("devices.json", devices_payload.encode("utf-8"))
# env.json
try:
from importlib.metadata import version as _pkg_version
pytest_version = _pkg_version("pytest")
except Exception:
pytest_version = "unknown"
env_payload = {
"seed": ctx.seed,
"run": ctx.run_number,
"exit_code": ctx.exit_code,
"export_ts": ts,
"python": platform.python_version(),
"pytest": pytest_version,
"platform": f"{platform.system()} {platform.release()} {platform.machine()}",
"hostname": socket.gethostname(),
}
_add("env.json", json.dumps(env_payload, indent=2).encode("utf-8"))
return archive_path
def iter_entries(archive_path: pathlib.Path) -> Iterable[str]:
"""Yield member names — used by callers that want to confirm the bundle shape."""
with tarfile.open(archive_path, "r:gz") as tar:
for m in tar.getmembers():
yield m.name
File diff suppressed because it is too large Load Diff
+137
View File
@@ -0,0 +1,137 @@
"""Resolves the firmware repo root and the binaries we invoke.
Everything that needs a path (the firmware root, `pio`, `esptool`, etc.) goes
through this module so the rest of the package never calls `shutil.which` or
parses environment variables directly.
"""
from __future__ import annotations
import os
import shutil
from pathlib import Path
from typing import Iterable
class ConfigError(RuntimeError):
"""Raised when a required path or binary cannot be resolved."""
def firmware_root() -> Path:
"""Resolve the root of the Meshtastic firmware repo.
Resolution order:
1. `MESHTASTIC_FIRMWARE_ROOT` env var.
2. Walk up from `cwd` looking for a directory with `platformio.ini`.
"""
env = os.environ.get("MESHTASTIC_FIRMWARE_ROOT")
if env:
root = Path(env).expanduser().resolve()
if not (root / "platformio.ini").is_file():
raise ConfigError(
f"MESHTASTIC_FIRMWARE_ROOT={env!r} does not contain platformio.ini"
)
return root
cur = Path.cwd().resolve()
for candidate in (cur, *cur.parents):
if (candidate / "platformio.ini").is_file():
return candidate
raise ConfigError(
"Could not locate Meshtastic firmware root. Set MESHTASTIC_FIRMWARE_ROOT "
"to the directory containing platformio.ini."
)
def _first_existing(paths: Iterable[Path]) -> Path | None:
for p in paths:
if p and p.is_file() and os.access(p, os.X_OK):
return p
return None
def pio_bin() -> Path:
"""Resolve the `pio` binary.
Order: MESHTASTIC_PIO_BIN → ~/.platformio/penv/bin/pio (PlatformIO keeps
this one current) → `pio` on PATH → `platformio` on PATH.
"""
env = os.environ.get("MESHTASTIC_PIO_BIN")
if env:
p = Path(env).expanduser()
if p.is_file() and os.access(p, os.X_OK):
return p
raise ConfigError(f"MESHTASTIC_PIO_BIN={env!r} is not an executable file")
penv = Path.home() / ".platformio" / "penv" / "bin" / "pio"
if penv.is_file() and os.access(penv, os.X_OK):
return penv
for name in ("pio", "platformio"):
w = shutil.which(name)
if w:
return Path(w)
raise ConfigError(
"Could not find `pio`. Install PlatformIO (https://platformio.org/install/cli) "
"or set MESHTASTIC_PIO_BIN."
)
def _hw_tool(env_var: str, names: tuple[str, ...], install_hint: str) -> Path:
"""Shared resolver for esptool / nrfutil / picotool.
Prefers the firmware repo's own `.venv/bin/<name>` (esptool lives there),
then PATH.
"""
env = os.environ.get(env_var)
if env:
p = Path(env).expanduser()
if p.is_file() and os.access(p, os.X_OK):
return p
raise ConfigError(f"{env_var}={env!r} is not an executable file")
try:
venv_bin = firmware_root() / ".venv" / "bin"
except ConfigError:
venv_bin = None
for name in names:
if venv_bin is not None:
p = venv_bin / name
if p.is_file() and os.access(p, os.X_OK):
return p
for name in names:
w = shutil.which(name)
if w:
return Path(w)
raise ConfigError(
f"Could not find `{names[0]}`. {install_hint} "
f"Or set {env_var} to an absolute path."
)
def esptool_bin() -> Path:
return _hw_tool(
"MESHTASTIC_ESPTOOL_BIN",
("esptool", "esptool.py"),
"Install via `pip install esptool`.",
)
def nrfutil_bin() -> Path:
return _hw_tool(
"MESHTASTIC_NRFUTIL_BIN",
("nrfutil", "adafruit-nrfutil"),
"Install via `pip install adafruit-nrfutil` or download Nordic nRF Util.",
)
def picotool_bin() -> Path:
return _hw_tool(
"MESHTASTIC_PICOTOOL_BIN",
("picotool",),
"Install via `brew install picotool` or build from https://github.com/raspberrypi/picotool.",
)
@@ -0,0 +1,84 @@
"""Context manager for meshtastic.SerialInterface connections.
Every info/admin tool goes through `connect(port)` so we have a single place
that:
- auto-selects the port when one likely_meshtastic device is present,
- fails fast if a serial_session is already holding the port,
- guarantees `.close()` is called, even on exception.
The `SerialInterface` blocks on construction waiting for the node database;
that's fine for v1 since every tool is a short-lived request.
"""
from __future__ import annotations
from contextlib import contextmanager
from typing import Iterator
from . import devices, registry
class ConnectionError(RuntimeError):
pass
def resolve_port(port: str | None) -> str:
"""Pick a port: explicit > sole likely_meshtastic candidate > error."""
if port:
return port
candidates = [d for d in devices.list_devices() if d["likely_meshtastic"]]
if not candidates:
raise ConnectionError(
"No Meshtastic devices detected. Plug one in or pass `port` explicitly. "
"Run `list_devices` with include_unknown=True to see all serial ports."
)
if len(candidates) > 1:
ports = ", ".join(c["port"] for c in candidates)
raise ConnectionError(
f"Multiple Meshtastic devices detected ({ports}). "
"Specify `port` explicitly."
)
return candidates[0]["port"]
@contextmanager
def connect(port: str | None = None, timeout_s: float = 8.0) -> Iterator:
"""Open a `meshtastic.SerialInterface` and always close it.
Raises `ConnectionError` immediately if another serial session holds the
port (a `pio device monitor` in `serial_sessions/`, for instance).
"""
from meshtastic.serial_interface import (
SerialInterface, # type: ignore[import-untyped]
)
resolved = resolve_port(port)
active = registry.active_session_for_port(resolved)
if active is not None:
raise ConnectionError(
f"Port {resolved} is held by serial session {active.id}. "
"Call `serial_close` first."
)
lock = registry.port_lock(resolved)
if not lock.acquire(blocking=False):
raise ConnectionError(
f"Port {resolved} is busy — another device operation is in flight. "
"Retry shortly."
)
iface = None
try:
iface = SerialInterface(devPath=resolved, connectNow=True, noProto=False)
yield iface
finally:
if iface is not None:
try:
iface.close()
except Exception:
pass
try:
lock.release()
except RuntimeError:
pass
+75
View File
@@ -0,0 +1,75 @@
"""USB/serial device discovery.
Combines the canonical `meshtastic.util.findPorts()` allowlist/blocklist with
the richer metadata (`serial.tools.list_ports.comports()`) so callers see
VID/PID, descriptions, and manufacturer strings alongside the "is this likely
a Meshtastic device" signal.
"""
from __future__ import annotations
from typing import Any
from serial.tools import list_ports
def _to_hex(value: int | None) -> str | None:
if value is None:
return None
return f"0x{value:04x}"
def list_devices(include_unknown: bool = False) -> list[dict[str, Any]]:
"""Return enriched info for serial ports, flagging Meshtastic candidates.
`likely_meshtastic` is True when the port's USB VID matches the Meshtastic
allowlist (`0x239a` Adafruit/RAK, `0x303a` Espressif). When no allowlisted
ports are present, ports whose VID is NOT in the blocklist (J-Link, ST-LINK,
PPK2, etc.) are surfaced as `likely_meshtastic=False` candidates.
With `include_unknown=False` (default), we return only ports that are
plausibly Meshtastic. With `include_unknown=True`, every serial port the
OS knows about is returned (useful for debugging "why isn't my board
detected").
"""
# Import lazily so the module loads even without the `meshtastic` package
# (useful for introspection / schema generation).
from meshtastic import util as mt_util # type: ignore[import-untyped]
meshtastic_ports: set[str] = set(mt_util.findPorts(eliminate_duplicates=True))
whitelist = getattr(mt_util, "whitelistVids", {})
blacklist = getattr(mt_util, "blacklistVids", {})
results: list[dict[str, Any]] = []
for info in list_ports.comports():
port_path = info.device
vid = info.vid
in_whitelist = vid is not None and vid in whitelist
in_blacklist = vid is not None and vid in blacklist
likely = port_path in meshtastic_ports and in_whitelist
# If no allowlisted ports were found, findPorts falls back to
# everything-not-in-blacklist; treat those as plausible candidates
# but not "likely".
fallback_candidate = port_path in meshtastic_ports and not in_whitelist
if not likely and not fallback_candidate and not include_unknown:
continue
results.append(
{
"port": port_path,
"vid": _to_hex(vid),
"pid": _to_hex(info.pid),
"description": info.description or None,
"manufacturer": info.manufacturer or None,
"product": info.product or None,
"serial_number": info.serial_number or None,
"likely_meshtastic": likely,
"blacklisted": in_blacklist,
}
)
# Stable ordering: likely_meshtastic first, then by port path
results.sort(key=lambda r: (not r["likely_meshtastic"], r["port"]))
return results
+447
View File
@@ -0,0 +1,447 @@
"""Build, clean, flash, and bootloader-entry operations.
Design: pio is the preferred path for every architecture via `flash()`. For
ESP32 factory flashes we shell out to `bin/device-install.sh` (which knows
about partition offsets and the OTA/littlefs partitions); for ESP32 OTA
updates we use `bin/device-update.sh`. Both scripts require the build
artifacts to exist, so these tools build first if needed.
"""
from __future__ import annotations
import subprocess
import threading
import time
from pathlib import Path
from typing import Any
import serial
from . import boards, config, devices, pio, userprefs
# Meshtastic variants use both `esp32s3` and `esp32-s3` style names across
# variants/*/platformio.ini (no consistency enforced). Accept both spellings.
ESP32_ARCHES = {
"esp32",
"esp32s2",
"esp32-s2",
"esp32s3",
"esp32-s3",
"esp32c3",
"esp32-c3",
"esp32c6",
"esp32-c6",
}
class FlashError(RuntimeError):
pass
def _require_confirm(confirm: bool, operation: str) -> None:
if not confirm:
raise FlashError(
f"{operation} is destructive and requires confirm=True. "
"This will overwrite firmware on the device."
)
def _artifacts_for(env: str) -> list[Path]:
build_dir = config.firmware_root() / ".pio" / "build" / env
if not build_dir.is_dir():
return []
patterns = (
"firmware*.bin",
"firmware*.uf2",
"firmware*.hex",
"firmware*.zip",
"firmware*.elf",
"*.mt.json",
"littlefs-*.bin",
)
out: list[Path] = []
for pat in patterns:
out.extend(sorted(build_dir.glob(pat)))
return out
def _factory_bin_for(env: str) -> Path | None:
build_dir = config.firmware_root() / ".pio" / "build" / env
if not build_dir.is_dir():
return None
matches = sorted(build_dir.glob("firmware-*.factory.bin"))
return matches[0] if matches else None
def _firmware_bin_for(env: str) -> Path | None:
"""Return the OTA-update firmware binary (app partition only)."""
build_dir = config.firmware_root() / ".pio" / "build" / env
if not build_dir.is_dir():
return None
# device-update.sh expects firmware-<env>-<version>.bin (not .factory.bin)
matches = sorted(
p
for p in build_dir.glob("firmware-*.bin")
if not p.name.endswith(".factory.bin")
)
return matches[0] if matches else None
def _userprefs_summary(active: dict[str, str]) -> dict[str, Any]:
"""Compact summary of which USERPREFS_* are baked into the build."""
return {"count": len(active), "keys": sorted(active.keys())}
def build(
env: str,
with_manifest: bool = True,
userprefs_overrides: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Run `pio run -e <env>` and return artifact paths.
`userprefs_overrides` (optional): dict of `USERPREFS_<KEY>: value` to inject
into userPrefs.jsonc for this build only. File is restored byte-for-byte
on exit. Use `userprefs_set()` for persistent changes.
"""
args = ["run", "-e", env]
if with_manifest:
args.extend(["-t", "mtjson"])
with userprefs.temporary_overrides(userprefs_overrides) as effective:
result = pio.run(args, timeout=pio.TIMEOUT_BUILD, check=False)
return {
"exit_code": result.returncode,
"artifacts": [str(p) for p in _artifacts_for(env)],
"stdout_tail": pio.tail_lines(result.stdout, 200),
"stderr_tail": pio.tail_lines(result.stderr, 200),
"duration_s": round(result.duration_s, 2),
"userprefs": _userprefs_summary(effective),
}
def clean(env: str) -> dict[str, Any]:
"""Run `pio run -e <env> -t clean`."""
result = pio.run(["run", "-e", env, "-t", "clean"], timeout=120, check=False)
return {
"exit_code": result.returncode,
"stdout_tail": pio.tail_lines(result.stdout, 200),
"stderr_tail": pio.tail_lines(result.stderr, 200),
"duration_s": round(result.duration_s, 2),
}
def flash(
env: str,
port: str,
confirm: bool = False,
userprefs_overrides: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""`pio run -e <env> -t upload --upload-port <port>`. All architectures.
`userprefs_overrides` (optional): see `build()` — the rebuild-before-upload
that pio performs will pick up the injected values.
"""
_require_confirm(confirm, "flash")
with userprefs.temporary_overrides(userprefs_overrides) as effective:
result = pio.run(
["run", "-e", env, "-t", "upload", "--upload-port", port],
timeout=pio.TIMEOUT_UPLOAD,
check=False,
)
return {
"exit_code": result.returncode,
"stdout_tail": pio.tail_lines(result.stdout, 200),
"stderr_tail": pio.tail_lines(result.stderr, 200),
"duration_s": round(result.duration_s, 2),
"userprefs": _userprefs_summary(effective),
}
def _check_esp32_env(env: str) -> str:
rec = boards.get_board(env)
arch = rec.get("architecture")
if arch not in ESP32_ARCHES:
raise FlashError(
f"Env {env!r} has architecture {arch!r}, not ESP32. "
"Use `flash` for non-ESP32 boards."
)
return arch
def _run_install_script(script: Path, port: str, binary: Path) -> dict[str, Any]:
"""Invoke bin/device-install.sh or bin/device-update.sh."""
t0 = time.monotonic()
proc = subprocess.run(
[str(script), "-p", port, "-f", str(binary)],
cwd=str(config.firmware_root()),
capture_output=True,
text=True,
timeout=pio.TIMEOUT_UPLOAD,
)
duration = time.monotonic() - t0
return {
"exit_code": proc.returncode,
"stdout_tail": pio.tail_lines(proc.stdout, 200),
"stderr_tail": pio.tail_lines(proc.stderr, 200),
"duration_s": round(duration, 2),
}
def erase_and_flash(
env: str,
port: str,
confirm: bool = False,
skip_build: bool = False,
userprefs_overrides: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""ESP32-only: full erase + factory flash via bin/device-install.sh.
`userprefs_overrides`: baked into the factory.bin via a fresh build. If
overrides are provided we always force a rebuild (skip_build=True errors
in that case) since a cached factory.bin would not reflect the new prefs.
"""
_require_confirm(confirm, "erase_and_flash")
_check_esp32_env(env)
if userprefs_overrides and skip_build:
raise FlashError(
"userprefs_overrides forces a rebuild so the factory.bin reflects "
"the new values; skip_build=True is incompatible."
)
with userprefs.temporary_overrides(userprefs_overrides) as effective:
# If overrides were provided, always build; otherwise only build if
# no factory.bin is present.
factory = _factory_bin_for(env)
if factory is None or userprefs_overrides:
if skip_build:
raise FlashError(
f"No factory.bin found for env {env!r} and skip_build=True. "
"Run `build` first or set skip_build=False."
)
build_args = ["run", "-e", env, "-t", "mtjson"]
build_result = pio.run(build_args, timeout=pio.TIMEOUT_BUILD, check=False)
if build_result.returncode != 0:
return {
"exit_code": build_result.returncode,
"stdout_tail": pio.tail_lines(build_result.stdout, 200),
"stderr_tail": pio.tail_lines(build_result.stderr, 200),
"duration_s": round(build_result.duration_s, 2),
"error": "build failed before erase_and_flash could run",
"userprefs": _userprefs_summary(effective),
}
factory = _factory_bin_for(env)
if factory is None:
raise FlashError(
f"Build succeeded but no factory.bin appeared in .pio/build/{env}/"
)
script = config.firmware_root() / "bin" / "device-install.sh"
if not script.is_file():
raise FlashError(f"device-install.sh not found at {script}")
result = _run_install_script(script, port, factory)
result["userprefs"] = _userprefs_summary(effective)
return result
def update_flash(
env: str,
port: str,
confirm: bool = False,
skip_build: bool = False,
userprefs_overrides: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""ESP32-only: OTA app-partition update via bin/device-update.sh.
`userprefs_overrides`: baked into the firmware.bin via a fresh build. If
overrides are provided we always force a rebuild.
"""
_require_confirm(confirm, "update_flash")
_check_esp32_env(env)
if userprefs_overrides and skip_build:
raise FlashError(
"userprefs_overrides forces a rebuild so the firmware.bin reflects "
"the new values; skip_build=True is incompatible."
)
with userprefs.temporary_overrides(userprefs_overrides) as effective:
firmware = _firmware_bin_for(env)
if firmware is None or userprefs_overrides:
if skip_build:
raise FlashError(
f"No firmware.bin found for env {env!r} and skip_build=True. "
"Run `build` first or set skip_build=False."
)
build_args = ["run", "-e", env, "-t", "mtjson"]
build_result = pio.run(build_args, timeout=pio.TIMEOUT_BUILD, check=False)
if build_result.returncode != 0:
return {
"exit_code": build_result.returncode,
"stdout_tail": pio.tail_lines(build_result.stdout, 200),
"stderr_tail": pio.tail_lines(build_result.stderr, 200),
"duration_s": round(build_result.duration_s, 2),
"error": "build failed before update_flash could run",
"userprefs": _userprefs_summary(effective),
}
firmware = _firmware_bin_for(env)
if firmware is None:
raise FlashError(
f"Build succeeded but no firmware.bin appeared in .pio/build/{env}/"
)
script = config.firmware_root() / "bin" / "device-update.sh"
if not script.is_file():
raise FlashError(f"device-update.sh not found at {script}")
result = _run_install_script(script, port, firmware)
result["userprefs"] = _userprefs_summary(effective)
return result
def _do_1200bps_touch(port: str, settle_ms: int, touch_timeout_s: float = 3.0) -> None:
"""Open port at 1200 baud and close, bounded by a worker thread.
Both the open and the close can block on a busy CDC device — we wrap the
whole thing in a worker so the caller returns in at most `touch_timeout_s`
regardless. The touch is signal-only: the USB configuration change to
1200 baud alone is enough to trip the Adafruit bootloader's reset, so a
worker that's still blocked in the background after timeout has already
delivered the signal.
"""
errors: list[BaseException] = []
def _inner() -> None:
try:
s = serial.Serial(port, 1200)
except serial.SerialException as exc:
if "No such file" in str(exc) or "could not open" in str(exc).lower():
raise
return # other serial errors mid-open are expected during DFU entry
try:
time.sleep(settle_ms / 1000.0)
finally:
try:
s.close()
except Exception:
pass
def _runner() -> None:
try:
_inner()
except BaseException as exc: # re-raised on caller thread after join
errors.append(exc)
worker = threading.Thread(target=_runner, daemon=True)
worker.start()
worker.join(timeout=touch_timeout_s)
if worker.is_alive():
return # signal already delivered; allow daemon worker to finish/exit
if errors:
raise errors[0]
# Adafruit nRF52 bootloader VID/PID (BOTH RAK4631 and most Feather nRF52 boards).
# See https://github.com/adafruit/Adafruit_nRF52_Bootloader
_NRF52_BOOTLOADER_VID = 0x239A
_NRF52_BOOTLOADER_PIDS = {
0x0029, # Adafruit nRF52 bootloader (generic, used by RAK4631)
0x002A, # Adafruit Feather Express bootloader variant
0x4029, # alt seen on some boards
}
def _find_nrf52_bootloader_port() -> dict[str, Any] | None:
"""Return a dict for any currently-enumerated nRF52 bootloader port, or None."""
for d in devices.list_devices(include_unknown=True):
vid_str = d.get("vid")
pid_str = d.get("pid")
if vid_str is None or pid_str is None:
continue
try:
vid = int(vid_str, 16) if isinstance(vid_str, str) else int(vid_str)
pid = int(pid_str, 16) if isinstance(pid_str, str) else int(pid_str)
except ValueError:
continue
if vid == _NRF52_BOOTLOADER_VID and pid in _NRF52_BOOTLOADER_PIDS:
return d
return None
def touch_1200bps(
port: str,
settle_ms: int = 250,
poll_timeout_s: float = 8.0,
retries: int = 2,
) -> dict[str, Any]:
"""Open port at 1200 baud, close immediately — triggers USB CDC bootloader.
Works for: nRF52840 (Adafruit bootloader), ESP32-S3 (native USB download
mode), RP2040 (when built with 1200bps-reset stdio), Arduino Leonardo/Micro.
For nRF52 specifically: after the touch, polls for the Adafruit bootloader
VID/PID (0x239A / 0x0029) for up to `poll_timeout_s` seconds. Adafruit's
bootloader docs note a touch sometimes needs to be repeated, so this
retries up to `retries` times. The returned `new_port` is the bootloader
port (distinct from the app port) — exactly what's needed for `pio run
-t upload` to drive nrfutil.
For non-nRF52 devices (ESP32-S3, RP2040, Arduino), falls back to
"any-new-port appeared" detection.
Returns `{ok, former_port, new_port, new_port_vid_pid, attempts}`.
"""
before_list = devices.list_devices(include_unknown=True)
before_ports = {d["port"] for d in before_list}
attempts = 0
new_port_info: dict[str, Any] | None = None
for attempt in range(1, retries + 1):
attempts = attempt
_do_1200bps_touch(port, settle_ms=settle_ms, touch_timeout_s=3.0)
# Poll for either (a) the nRF52 bootloader VID/PID appearing, or
# (b) a brand-new port appearing that wasn't there before.
deadline = time.monotonic() + poll_timeout_s
while time.monotonic() < deadline:
time.sleep(0.2)
bootloader = _find_nrf52_bootloader_port()
if bootloader is not None:
new_port_info = bootloader
break
current = devices.list_devices(include_unknown=True)
current_paths = {d["port"] for d in current}
added = current_paths - before_ports
if added:
added_record = next((d for d in current if d["port"] in added), None)
if added_record:
new_port_info = added_record
break
if new_port_info is not None:
break
# No bootloader appeared; try touching again (Adafruit recommends
# sometimes requiring two touches for reliability).
if new_port_info is not None:
return {
"ok": True,
"former_port": port,
"new_port": new_port_info["port"],
"new_port_vid_pid": (
new_port_info.get("vid"),
new_port_info.get("pid"),
),
"attempts": attempts,
}
return {
"ok": False,
"former_port": port,
"new_port": None,
"new_port_vid_pid": (None, None),
"attempts": attempts,
}
+243
View File
@@ -0,0 +1,243 @@
"""Direct wrappers around vendor flashing tools: esptool, nrfutil, picotool.
These are escape hatches. Prefer the pio-based tools in flash.py when they
cover the operation — pio knows the correct offsets, protocols, and filters
for every supported board. Use these when pio doesn't: to erase a bricked
ESP32, DFU-flash an nRF52 zip package, or inspect an RP2040's bootloader.
Every destructive `*_raw` subcommand is gated by `confirm=True` so callers
can't accidentally `--write-flash` from freeform args.
"""
from __future__ import annotations
import re
import subprocess
from pathlib import Path
from typing import Any, Sequence
from . import config, pio
_TIMEOUT_SHORT = 30
_TIMEOUT_LONG = 600
class ToolError(RuntimeError):
pass
def _run(
binary: Path,
args: Sequence[str],
*,
timeout: float = _TIMEOUT_LONG,
cwd: Path | None = None,
) -> dict[str, Any]:
# Shared with pio.run(): if `MESHTASTIC_MCP_FLASH_LOG` is set, each line
# of output is tee'd to that file as it arrives so the TUI can show live
# esptool/nrfutil/picotool progress instead of 3 minutes of silence.
full = [str(binary), *args]
try:
rc, stdout, stderr, duration = pio._run_capturing(
full,
cwd=cwd,
timeout=timeout,
tee_header=f"{binary.name} {' '.join(args)}",
)
except subprocess.TimeoutExpired as exc:
raise ToolError(
f"{binary.name} {' '.join(args)} timed out after {timeout}s"
) from exc
return {
"exit_code": rc,
"stdout": stdout,
"stderr": stderr,
"stdout_tail": pio.tail_lines(stdout, 200),
"stderr_tail": pio.tail_lines(stderr, 200),
"duration_s": round(duration, 2),
}
def _require_confirm(confirm: bool, what: str) -> None:
if not confirm:
raise ToolError(f"{what} is destructive and requires confirm=True.")
# ---------- esptool --------------------------------------------------------
ESPTOOL_DESTRUCTIVE = {
"write_flash",
"write-flash",
"erase_flash",
"erase-flash",
"erase_region",
"erase-region",
"merge_bin",
"merge-bin",
}
def _parse_esptool_chip_info(stdout: str) -> dict[str, Any]:
"""Parse `esptool chip_id` / `flash_id` output into structured fields."""
result: dict[str, Any] = {
"chip": None,
"mac": None,
"crystal_mhz": None,
"flash_size": None,
"features": [],
}
for line in stdout.splitlines():
line = line.strip()
if m := re.match(r"Chip is (.+)", line):
result["chip"] = m.group(1).strip()
elif m := re.match(r"MAC: ([0-9a-fA-F:]+)", line):
result["mac"] = m.group(1)
elif m := re.match(r"Crystal is (\d+)MHz", line):
result["crystal_mhz"] = int(m.group(1))
elif m := re.match(r"Detected flash size: (\S+)", line):
result["flash_size"] = m.group(1)
elif m := re.match(r"Features: (.+)", line):
result["features"] = [f.strip() for f in m.group(1).split(",") if f.strip()]
return result
def esptool_chip_info(port: str) -> dict[str, Any]:
binary = config.esptool_bin()
# `chip_id` prints chip + mac + crystal + features. `flash_id` adds flash.
combined = _run(binary, ["--port", port, "flash_id"], timeout=_TIMEOUT_SHORT)
if combined["exit_code"] != 0:
raise ToolError(
f"esptool failed (exit {combined['exit_code']}):\n{combined['stderr_tail']}"
)
parsed = _parse_esptool_chip_info(combined["stdout"])
return {**parsed, "raw_stdout_tail": combined["stdout_tail"]}
def esptool_erase_flash(port: str, confirm: bool = False) -> dict[str, Any]:
"""Full-chip erase. Leaves the device unbootable until reflashed."""
_require_confirm(confirm, "esptool_erase_flash")
binary = config.esptool_bin()
# esptool v5 uses `erase-flash`, older uses `erase_flash`. Try the new name
# first; if it fails with unknown command, retry old.
res = _run(binary, ["--port", port, "erase-flash"], timeout=_TIMEOUT_LONG)
if (
res["exit_code"] != 0
and "unrecognized" in (res["stderr"] or res["stdout"]).lower()
):
res = _run(binary, ["--port", port, "erase_flash"], timeout=_TIMEOUT_LONG)
return res
def esptool_raw(
args: list[str], port: str | None = None, confirm: bool = False
) -> dict[str, Any]:
"""Raw esptool passthrough. Destructive subcommands require confirm=True."""
if not args:
raise ToolError("args must not be empty")
# Find the first non-flag arg (the subcommand).
subcommand = next((a for a in args if not a.startswith("-")), None)
if subcommand and subcommand.replace("-", "_") in {
s.replace("-", "_") for s in ESPTOOL_DESTRUCTIVE
}:
_require_confirm(confirm, f"esptool {subcommand}")
binary = config.esptool_bin()
full_args: list[str] = []
if port:
full_args.extend(["--port", port])
full_args.extend(args)
return _run(binary, full_args, timeout=_TIMEOUT_LONG)
# ---------- nrfutil --------------------------------------------------------
NRFUTIL_DESTRUCTIVE = {"dfu", "settings"}
def nrfutil_dfu(port: str, package_path: str, confirm: bool = False) -> dict[str, Any]:
_require_confirm(confirm, "nrfutil_dfu")
pkg = Path(package_path).expanduser()
if not pkg.is_file():
raise ToolError(f"Package not found: {pkg}")
binary = config.nrfutil_bin()
return _run(
binary,
["dfu", "serial", "--package", str(pkg), "--port", port, "-b", "115200"],
timeout=_TIMEOUT_LONG,
)
def nrfutil_raw(args: list[str], confirm: bool = False) -> dict[str, Any]:
if not args:
raise ToolError("args must not be empty")
subcommand = next((a for a in args if not a.startswith("-")), None)
if subcommand in NRFUTIL_DESTRUCTIVE:
_require_confirm(confirm, f"nrfutil {subcommand}")
binary = config.nrfutil_bin()
return _run(binary, args, timeout=_TIMEOUT_LONG)
# ---------- picotool -------------------------------------------------------
PICOTOOL_DESTRUCTIVE = {"load", "reboot", "save", "erase"}
def _parse_picotool_info(stdout: str) -> dict[str, Any]:
result: dict[str, Any] = {
"vendor": None,
"product": None,
"serial": None,
"flash_size": None,
"program_name": None,
"program_version": None,
}
for line in stdout.splitlines():
line = line.strip()
if m := re.match(r"Program information:", line):
continue
if m := re.match(r"name:\s*(.+)", line):
result["program_name"] = m.group(1).strip()
elif m := re.match(r"version:\s*(.+)", line):
result["program_version"] = m.group(1).strip()
elif m := re.match(r"vendor:\s*(.+)", line):
result["vendor"] = m.group(1).strip()
elif m := re.match(r"product:\s*(.+)", line):
result["product"] = m.group(1).strip()
elif m := re.match(r"serial number:\s*(.+)", line):
result["serial"] = m.group(1).strip()
elif m := re.match(r"flash size:\s*(.+)", line):
result["flash_size"] = m.group(1).strip()
return result
def picotool_info(port: str | None = None) -> dict[str, Any]:
"""Read device info from a Pico in BOOTSEL mode. `port` is informational
only — picotool auto-detects."""
binary = config.picotool_bin()
res = _run(binary, ["info", "-a"], timeout=_TIMEOUT_SHORT)
if res["exit_code"] != 0:
raise ToolError(
f"picotool info failed (exit {res['exit_code']}): "
"is the Pico in BOOTSEL mode?\n" + res["stderr_tail"]
)
parsed = _parse_picotool_info(res["stdout"])
return {**parsed, "raw_stdout_tail": res["stdout_tail"]}
def picotool_load(uf2_path: str, confirm: bool = False) -> dict[str, Any]:
_require_confirm(confirm, "picotool_load")
uf2 = Path(uf2_path).expanduser()
if not uf2.is_file():
raise ToolError(f"UF2 not found: {uf2}")
binary = config.picotool_bin()
return _run(binary, ["load", "-x", "-t", "uf2", str(uf2)], timeout=_TIMEOUT_LONG)
def picotool_raw(args: list[str], confirm: bool = False) -> dict[str, Any]:
if not args:
raise ToolError("args must not be empty")
subcommand = next((a for a in args if not a.startswith("-")), None)
if subcommand in PICOTOOL_DESTRUCTIVE:
_require_confirm(confirm, f"picotool {subcommand}")
binary = config.picotool_bin()
return _run(binary, args, timeout=_TIMEOUT_LONG)
+103
View File
@@ -0,0 +1,103 @@
"""Read-only device queries via meshtastic.SerialInterface."""
from __future__ import annotations
from typing import Any
from .connection import connect
def _primary_channel_name(iface) -> str | None:
try:
channels = iface.localNode.channels or []
except AttributeError:
return None
for ch in channels:
role = getattr(ch, "role", None)
# Role enum: 0 DISABLED, 1 PRIMARY, 2 SECONDARY
if role == 1:
name = getattr(getattr(ch, "settings", None), "name", None)
return name or "(default)"
return None
def device_info(port: str | None = None, timeout_s: float = 8.0) -> dict[str, Any]:
"""Return summary info for the connected device."""
with connect(port=port, timeout_s=timeout_s) as iface:
my = iface.myInfo
meta = iface.metadata
local = iface.localNode
# Owner (long/short name) is on the local node's user record
long_name: str | None = None
short_name: str | None = None
hw_model: str | int | None = None
if iface.nodesByNum and my is not None:
local_rec = iface.nodesByNum.get(my.my_node_num, {})
user = local_rec.get("user") or {}
long_name = user.get("longName")
short_name = user.get("shortName")
hw_model = user.get("hwModel")
region = None
if local is not None and local.localConfig is not None:
try:
lora = local.localConfig.lora
# region is an enum; get its string name
region = (
lora.DESCRIPTOR.fields_by_name["region"]
.enum_type.values_by_number[lora.region]
.name
)
except Exception:
region = None
return {
"port": iface.devPath if hasattr(iface, "devPath") else port,
"my_node_num": getattr(my, "my_node_num", None),
"long_name": long_name,
"short_name": short_name,
"firmware_version": getattr(meta, "firmware_version", None),
"hw_model": hw_model,
"region": region,
"num_nodes": len(iface.nodesByNum) if iface.nodesByNum else 0,
"primary_channel": _primary_channel_name(iface),
}
def _node_record(node_dict: dict[str, Any]) -> dict[str, Any]:
user = node_dict.get("user") or {}
position = node_dict.get("position") or None
device_metrics = node_dict.get("deviceMetrics") or {}
return {
"node_num": node_dict.get("num"),
"user": {
"long_name": user.get("longName"),
"short_name": user.get("shortName"),
"hw_model": user.get("hwModel"),
"role": user.get("role"),
},
"position": (
{
"latitude": position.get("latitude"),
"longitude": position.get("longitude"),
"altitude": position.get("altitude"),
"time": position.get("time"),
}
if position
else None
),
"snr": node_dict.get("snr"),
"rssi": node_dict.get("rssi"),
"last_heard": node_dict.get("lastHeard"),
"battery_level": device_metrics.get("batteryLevel"),
"is_favorite": bool(node_dict.get("isFavorite", False)),
}
def list_nodes(port: str | None = None, timeout_s: float = 8.0) -> list[dict[str, Any]]:
"""Return the device's node database."""
with connect(port=port, timeout_s=timeout_s) as iface:
if not iface.nodesByNum:
return []
return [_node_record(n) for n in iface.nodesByNum.values()]
+295
View File
@@ -0,0 +1,295 @@
"""Subprocess wrappers around the `pio` CLI.
Every PlatformIO interaction in this package funnels through `run()` so we
have a single place that owns timeouts, buffer sizes, JSON parsing, and the
"stderr on exit-0 is informational" convention.
`run()` has two execution paths:
* Fast path (default): `subprocess.run(capture_output=True)` — buffered, one
return; fine for sub-second pio calls like `pio --version` or
`pio project config --json-output`.
* Streaming path: when the `MESHTASTIC_MCP_FLASH_LOG` env var is set, each
output line is tee'd to that file as it arrives via a threaded reader.
The TUI tails the file to give live flash progress — otherwise a 3-minute
`pio run -t upload` is completely silent to the operator.
`hw_tools.py` shares the streaming helper via `pio._run_capturing()` so
esptool/nrfutil/picotool output also streams when the env var is set.
"""
from __future__ import annotations
import json
import os
import subprocess
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Sequence, TextIO
from . import config
# 10 MB matches the reference impl (jl-codes/platformio-mcp). Build output can
# be hundreds of KB; we'd rather keep it in memory than truncate.
_MAX_BUFFER = 10 * 1024 * 1024
# Per-operation defaults (seconds). None = no timeout.
TIMEOUT_DEFAULT = 120
TIMEOUT_PROJECT_CONFIG = 60
TIMEOUT_DEVICE_LIST = 15
TIMEOUT_BUILD = 900
TIMEOUT_UPLOAD = 600
class PioError(RuntimeError):
"""pio exited non-zero."""
def __init__(self, args: Sequence[str], returncode: int, stdout: str, stderr: str):
self.args = list(args)
self.returncode = returncode
self.stdout = stdout
self.stderr = stderr
tail = (stderr or stdout).strip().splitlines()[-20:]
super().__init__(
f"pio {' '.join(args)} failed with exit {returncode}:\n" + "\n".join(tail)
)
class PioTimeout(RuntimeError):
"""pio did not return within the timeout."""
@dataclass
class PioResult:
args: list[str]
returncode: int
stdout: str
stderr: str
duration_s: float
_FLASH_LOG_ENV = "MESHTASTIC_MCP_FLASH_LOG"
def _flash_log_path() -> Path | None:
"""Return the path to tee subprocess output to, or None if streaming off.
Controlled by `MESHTASTIC_MCP_FLASH_LOG`. `run-tests.sh` sets this to
`tests/flash.log`; the TUI tails that file so `pio run -t upload` shows
live progress in the pytest pane.
"""
raw = os.environ.get(_FLASH_LOG_ENV)
if not raw:
return None
return Path(raw)
def _run_capturing(
argv: Sequence[str],
*,
cwd: Path | None = None,
timeout: float | None = None,
tee_header: str | None = None,
) -> tuple[int, str, str, float]:
"""Run a subprocess, capture stdout+stderr, optionally tee to the flash log.
Returns `(returncode, stdout_str, stderr_str, duration_s)`. Raises
`subprocess.TimeoutExpired` on timeout (callers map this to their own
domain-specific error).
Fast path: `subprocess.run(capture_output=True)` when no flash log is
configured (unchanged behavior).
Streaming path: `Popen` with line-buffered stdout+stderr pipes; two
reader threads accumulate into result strings AND append each line to
the flash log file. Stdout and stderr stay separate in the return value
(so `stderr_tail` still means stderr), but are interleaved in the log
file in the order they arrived — that's what a human wants to read.
"""
log_path = _flash_log_path()
t0 = time.monotonic()
if log_path is None:
# Fast path — unchanged.
proc = subprocess.run(
list(argv),
cwd=str(cwd) if cwd else None,
capture_output=True,
text=True,
timeout=timeout,
)
return (
proc.returncode,
proc.stdout or "",
proc.stderr or "",
time.monotonic() - t0,
)
# Streaming path: line-buffered Popen, threaded readers, tee to file.
# Ensure parent directory exists so the first tee write doesn't fail.
log_path.parent.mkdir(parents=True, exist_ok=True)
log_fh: TextIO | None = None
try:
log_fh = log_path.open("a", encoding="utf-8")
except OSError:
pass
# Append mode: the TUI truncates on startup, the session may produce
# many tee'd commands (erase + flash + factory-reset response), and
# we want all of them chronologically in one log.
proc = subprocess.Popen( # noqa: S603
list(argv),
cwd=str(cwd) if cwd else None,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1, # line-buffered
)
stdout_chunks: list[str] = []
stderr_chunks: list[str] = []
log_lock = threading.Lock()
def _append_log(line: str) -> None:
# Hold the lock briefly to serialize interleaved stdout/stderr writes
# so a half-written line from one stream doesn't get garbled by the
# other.
nonlocal log_fh
with log_lock:
if log_fh is None:
return
try:
log_fh.write(line)
log_fh.flush()
except OSError:
# Log file disappeared (umount, operator deleted the dir).
# Don't let that bubble up — the subprocess output is still
# collected in-memory for the return value.
try:
log_fh.close()
except OSError:
pass
log_fh = None
def _tee(stream, sink: list[str]) -> None:
try:
for line in stream:
sink.append(line)
_append_log(line)
except Exception:
pass
# Header line so the operator can tell commands apart in the log.
if tee_header:
_append_log(f"\n--- {tee_header} (start)\n")
assert proc.stdout is not None and proc.stderr is not None
t_out = threading.Thread(
target=_tee, args=(proc.stdout, stdout_chunks), daemon=True
)
t_err = threading.Thread(
target=_tee, args=(proc.stderr, stderr_chunks), daemon=True
)
t_out.start()
t_err.start()
# `Popen.wait` with a timeout is the cleanest way to get TimeoutExpired.
try:
proc.wait(timeout=timeout)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait()
# Drain readers before re-raising so we don't leave threads behind.
t_out.join(timeout=2)
t_err.join(timeout=2)
raise
t_out.join()
t_err.join()
duration = time.monotonic() - t0
if tee_header:
_append_log(f"--- {tee_header} (exit {proc.returncode} in {duration:.1f}s)\n")
try:
return (
proc.returncode,
"".join(stdout_chunks),
"".join(stderr_chunks),
duration,
)
finally:
if log_fh is not None:
try:
log_fh.close()
except OSError:
pass
def run(
args: Sequence[str],
*,
cwd: Path | None = None,
timeout: float | None = TIMEOUT_DEFAULT,
check: bool = True,
) -> PioResult:
"""Invoke `pio <args>` and return captured output.
`cwd` defaults to the firmware root. `check=True` raises `PioError` on
non-zero exit; set `check=False` to inspect `returncode` manually.
If `MESHTASTIC_MCP_FLASH_LOG` is set, output is also tee'd to that file
line-by-line as it arrives (for live flash progress in the TUI).
"""
binary = str(config.pio_bin())
work_dir = cwd or config.firmware_root()
full = [binary, *args]
try:
rc, stdout, stderr, duration = _run_capturing(
full,
cwd=work_dir,
timeout=timeout,
tee_header=f"pio {' '.join(args)}",
)
except subprocess.TimeoutExpired as exc:
raise PioTimeout(f"pio {' '.join(args)} timed out after {timeout}s") from exc
result = PioResult(
args=list(args),
returncode=rc,
stdout=stdout,
stderr=stderr,
duration_s=duration,
)
if check and result.returncode != 0:
raise PioError(args, result.returncode, result.stdout, result.stderr)
return result
def run_json(
args: Sequence[str],
*,
cwd: Path | None = None,
timeout: float | None = TIMEOUT_DEFAULT,
):
"""Run pio with `--json-output` appended and parse the result."""
full = list(args)
if "--json-output" not in full:
full.append("--json-output")
res = run(full, cwd=cwd, timeout=timeout, check=True)
if not res.stdout.strip():
raise PioError(args, 0, res.stdout, res.stderr or "pio returned empty output")
try:
return json.loads(res.stdout)
except json.JSONDecodeError as exc:
raise PioError(
args, 0, res.stdout[:2000], f"invalid JSON from pio: {exc}"
) from exc
def tail_lines(text: str, n: int = 200) -> str:
"""Last `n` lines of `text`, joined with newlines. Empty string stays empty."""
if not text:
return ""
lines = text.splitlines()
return "\n".join(lines[-n:])
+98
View File
@@ -0,0 +1,98 @@
"""In-memory registry of active serial monitor sessions and port locks.
Two things live here so the rest of the package has a single place to reach
them:
1. `sessions`: `{session_id: SerialSession}` for pio device monitor subprocs.
2. `port_locks`: `{port: threading.Lock}` so admin/info tools can fail fast
when a serial monitor or another meshtastic client already owns a port.
"""
from __future__ import annotations
import threading
from typing import Any
from .serial_session import SerialSession, close_session
_LOCK = threading.Lock()
_sessions: dict[str, SerialSession] = {}
_port_locks: dict[str, threading.Lock] = {}
def register_session(session: SerialSession) -> None:
with _LOCK:
_sessions[session.id] = session
def get_session(session_id: str) -> SerialSession:
with _LOCK:
session = _sessions.get(session_id)
if session is None:
raise KeyError(f"Unknown session_id: {session_id!r}")
return session
def remove_session(session_id: str) -> SerialSession | None:
with _LOCK:
return _sessions.pop(session_id, None)
def active_session_for_port(port: str) -> SerialSession | None:
"""Find any active (non-eof) session owning `port`."""
sweep_dead()
with _LOCK:
for s in _sessions.values():
if s.port == port and s.proc.poll() is None:
return s
return None
def all_sessions() -> list[SerialSession]:
with _LOCK:
return list(_sessions.values())
def sweep_dead() -> int:
"""Remove sessions whose subprocess has exited. Returns count removed."""
removed_sessions: list[SerialSession] = []
with _LOCK:
for sid, s in list(_sessions.items()):
if s.proc.poll() is not None:
removed_sessions.append(_sessions.pop(sid))
for session in removed_sessions:
try:
close_session(session)
except Exception:
pass
return len(removed_sessions)
def shutdown_all() -> None:
"""Close every live session (called on server exit)."""
with _LOCK:
items = list(_sessions.items())
_sessions.clear()
for _sid, session in items:
try:
close_session(session)
except Exception:
pass
def port_lock(port: str) -> threading.Lock:
"""Per-port lock for SerialInterface / admin tool serialization."""
with _LOCK:
lock = _port_locks.get(port)
if lock is None:
lock = threading.Lock()
_port_locks[port] = lock
return lock
def snapshot() -> dict[str, Any]:
"""Debug dump: session count, port lock count."""
with _LOCK:
return {
"sessions": len(_sessions),
"port_locks": len(_port_locks),
}
@@ -0,0 +1,216 @@
"""Long-running serial monitor sessions via `pio device monitor`.
Why pio instead of raw pyserial: pio applies the board's monitor_filters —
`esp32_exception_decoder` symbolicates crash stacks, `time` adds timestamps,
etc. Raw pyserial would give us bytes; pio gives us developer-grade logs.
Each session runs `pio device monitor` in a subprocess, with a daemon reader
thread draining stdout into a bounded ring buffer. Callers pull lines via
`serial_read` using a cursor that survives across calls.
"""
from __future__ import annotations
import collections
import subprocess
import threading
import time
import uuid
from dataclasses import dataclass, field
from typing import Any
from . import boards, config
_BUFFER_MAX_LINES = 10_000
_POLL_NEW_PORT_TIMEOUT_S = 3.0
@dataclass
class SerialSession:
id: str
port: str
baud: int
filters: list[str]
env: str | None
proc: subprocess.Popen
buffer: collections.deque = field(
default_factory=lambda: collections.deque(maxlen=_BUFFER_MAX_LINES)
)
# Total lines seen (not bounded by buffer maxlen). `dropped = total - len(buffer)`
# if the reader has advanced past buffer head.
total_lines: int = 0
started_at: float = field(default_factory=time.time)
stopped_at: float | None = None
lock: threading.Lock = field(default_factory=threading.Lock)
_thread: threading.Thread | None = None
def _drain(session: SerialSession) -> None:
"""Reader thread: line-by-line pull stdout into buffer."""
assert session.proc.stdout is not None
try:
for line in session.proc.stdout:
line_stripped = line.rstrip("\r\n")
with session.lock:
session.buffer.append(line_stripped)
session.total_lines += 1
except Exception: # pragma: no cover - defensive
pass
finally:
session.stopped_at = time.time()
def open_session(
port: str,
baud: int = 115200,
env: str | None = None,
filters: list[str] | None = None,
) -> SerialSession:
"""Spawn `pio device monitor` and return a SerialSession.
If `env` is supplied, pio resolves baud and filters from platformio.ini.
Otherwise uses the supplied `baud` and `filters` (default `['direct']`).
"""
args = ["device", "monitor", "--port", port, "--no-reconnect"]
effective_filters: list[str]
effective_baud: int = baud
if env is not None:
args.extend(["-e", env])
raw_config: dict[str, Any] = {}
try:
raw = boards.get_board(env).get("raw_config")
if isinstance(raw, dict):
raw_config = raw
except Exception:
raw_config = {}
monitor_speed = raw_config.get("monitor_speed")
has_board_speed = False
if monitor_speed is not None:
try:
effective_baud = int(str(monitor_speed).strip())
has_board_speed = True
except (TypeError, ValueError):
pass
monitor_filters_raw = raw_config.get("monitor_filters")
parsed_board_filters: list[str] = []
if isinstance(monitor_filters_raw, str):
for token in monitor_filters_raw.replace("\n", ",").split(","):
item = token.strip()
if item:
parsed_board_filters.append(item)
elif isinstance(monitor_filters_raw, list):
parsed_board_filters = [
str(item).strip() for item in monitor_filters_raw if str(item).strip()
]
has_board_filters = len(parsed_board_filters) > 0
effective_filters = (
parsed_board_filters if has_board_filters else (filters or [])
)
if not has_board_speed:
args.extend(["--baud", str(effective_baud)])
if not has_board_filters:
for f in effective_filters:
args.extend(["--filter", f])
else:
args.extend(["--baud", str(baud)])
effective_filters = filters or ["direct"]
for f in effective_filters:
args.extend(["--filter", f])
binary = str(config.pio_bin())
work_dir = str(config.firmware_root())
proc = subprocess.Popen(
[binary, *args],
cwd=work_dir,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1, # line-buffered
)
session = SerialSession(
id=uuid.uuid4().hex,
port=port,
baud=effective_baud,
filters=effective_filters,
env=env,
proc=proc,
)
t = threading.Thread(target=_drain, args=(session,), daemon=True)
t.start()
session._thread = t
return session
def read_session(
session: SerialSession, max_lines: int = 200, since_cursor: int | None = None
) -> dict[str, Any]:
"""Snapshot recent lines from the buffer.
Cursor semantics: the global cursor is `total_lines` at read time. Pass
`since_cursor` from a previous response's `new_cursor` to page forward.
`since_cursor=0` reads everything still in the ring buffer.
"""
with session.lock:
total = session.total_lines
buf_len = len(session.buffer)
head_cursor = total - buf_len # cursor value at buffer[0]
current_buffer = list(session.buffer)
if since_cursor is None:
since_cursor = head_cursor
# Clamp: never read what's aged out of the buffer.
effective_start = max(since_cursor, head_cursor)
# Number of lines skipped because they aged out between reads.
dropped = max(0, head_cursor - since_cursor) if since_cursor < head_cursor else 0
start_idx = effective_start - head_cursor
end_idx = min(start_idx + max_lines, buf_len)
lines = current_buffer[start_idx:end_idx]
new_cursor = effective_start + len(lines)
eof = session.proc.poll() is not None
return {
"lines": lines,
"new_cursor": new_cursor,
"eof": eof,
"dropped": dropped,
}
def close_session(session: SerialSession) -> bool:
"""Terminate the subprocess and join the reader thread."""
proc = session.proc
if proc.poll() is None:
try:
proc.terminate()
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait(timeout=3)
if session._thread is not None:
session._thread.join(timeout=3)
session.stopped_at = session.stopped_at or time.time()
return True
def session_summary(session: SerialSession) -> dict[str, Any]:
with session.lock:
line_count = session.total_lines
return {
"session_id": session.id,
"port": session.port,
"baud": session.baud,
"filters": session.filters,
"env": session.env,
"started_at": session.started_at,
"stopped_at": session.stopped_at,
"line_count": line_count,
"eof": session.proc.poll() is not None,
}
+590
View File
@@ -0,0 +1,590 @@
"""FastMCP server wiring — 38 tools across 7 categories.
Each tool handler is a thin delegation to a named module (pio.py, admin.py,
etc.). Business logic does not live here.
"""
from __future__ import annotations
from typing import Any
from mcp.server.fastmcp import FastMCP
from . import (
admin,
boards,
devices,
flash,
hw_tools,
info,
registry,
serial_session,
)
from . import userprefs as userprefs_mod
app = FastMCP("meshtastic-mcp")
# ---------- Discovery & metadata ------------------------------------------
@app.tool()
def list_devices(include_unknown: bool = False) -> list[dict[str, Any]]:
"""List USB/serial ports, flagging those likely to be Meshtastic devices.
With include_unknown=True, returns every serial port the OS knows about
(useful for debugging when a device isn't detected). Otherwise returns
only likely-Meshtastic candidates.
"""
return devices.list_devices(include_unknown=include_unknown)
@app.tool()
def list_boards(
architecture: str | None = None,
actively_supported_only: bool = False,
query: str | None = None,
board_level: str | None = None,
) -> list[dict[str, Any]]:
"""Enumerate PlatformIO envs (boards) with Meshtastic metadata.
architecture: filter to one arch ("esp32", "esp32s3", "nrf52840", "rp2040", "stm32", "native").
actively_supported_only: filter to boards marked custom_meshtastic_actively_supported=true.
query: substring match on display_name, env name, or hw_model_slug (case-insensitive).
board_level: "release" (default-track release boards), "pr" (PR CI), or "extra" (opt-in extras).
"""
return boards.list_boards(
architecture=architecture,
actively_supported_only=actively_supported_only,
query=query,
board_level=board_level,
)
@app.tool()
def get_board(env: str) -> dict[str, Any]:
"""Full metadata for one PlatformIO env, including raw pio config fields."""
return boards.get_board(env)
# ---------- Build & flash -------------------------------------------------
@app.tool()
def build(
env: str,
with_manifest: bool = True,
userprefs: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build firmware for one env via `pio run -e <env>`.
Returns exit_code, duration, artifact paths under .pio/build/<env>/, and
tails of stdout/stderr (last 200 lines each). with_manifest=True adds the
mtjson target which produces an .mt.json manifest alongside the firmware.
`userprefs` (optional): dict of `USERPREFS_<KEY>: value` baked into this
build via userPrefs.jsonc injection. The file is restored after the build
completes. Use `userprefs_manifest` to discover available keys. Use
`userprefs_set` for persistent changes.
"""
return flash.build(env, with_manifest=with_manifest, userprefs_overrides=userprefs)
@app.tool()
def clean(env: str) -> dict[str, Any]:
"""Clean one env's build output via `pio run -e <env> -t clean`.
Useful when switching branches or debugging a stale-cache build failure.
"""
return flash.clean(env)
@app.tool()
def pio_flash(
env: str,
port: str,
confirm: bool = False,
userprefs: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Flash firmware via `pio run -e <env> -t upload --upload-port <port>`.
Works for any architecture (ESP32/nRF52/RP2040/STM32). Requires confirm=True.
For first-time flashing a blank ESP32 board (erase + bootloader + app + fs),
prefer `erase_and_flash`. For ESP32 OTA updates, prefer `update_flash`.
`userprefs` (optional): dict of `USERPREFS_<KEY>: value` baked into this
build via userPrefs.jsonc injection; restored after upload.
"""
return flash.flash(env, port, confirm=confirm, userprefs_overrides=userprefs)
@app.tool()
def erase_and_flash(
env: str,
port: str,
confirm: bool = False,
skip_build: bool = False,
userprefs: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""ESP32-only: full erase + factory flash via bin/device-install.sh.
Wipes the entire flash and writes bootloader, app, OTA, and LittleFS
partitions from the factory.bin. Requires confirm=True. Runs `build` first
if no factory.bin is present (set skip_build=True to require a prior build).
`userprefs` (optional): dict of `USERPREFS_<KEY>: value` baked into the
factory.bin via userPrefs.jsonc injection. When provided, forces a rebuild
(skip_build=True is incompatible). File is restored after upload.
"""
return flash.erase_and_flash(
env, port, confirm=confirm, skip_build=skip_build, userprefs_overrides=userprefs
)
@app.tool()
def update_flash(
env: str,
port: str,
confirm: bool = False,
skip_build: bool = False,
userprefs: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""ESP32-only: OTA app-partition update via bin/device-update.sh.
Updates only the application partition, preserving device config and node
database. Faster than erase_and_flash but won't recover a broken bootloader.
Requires confirm=True. Builds first if needed.
`userprefs` (optional): dict of `USERPREFS_<KEY>: value` baked into the
firmware.bin via userPrefs.jsonc injection. When provided, forces a rebuild.
"""
return flash.update_flash(
env, port, confirm=confirm, skip_build=skip_build, userprefs_overrides=userprefs
)
# ---------- USERPREFS discovery & persistence -----------------------------
@app.tool()
def userprefs_manifest() -> dict[str, Any]:
"""Full manifest of USERPREFS_* keys the firmware knows about.
Combines `userPrefs.jsonc` (active + commented examples) with a scan of
`src/**` for `USERPREFS_<KEY>` references — so every key the firmware
actually consumes shows up, even if undocumented in the jsonc.
Each entry has: key, active (is it uncommented), value (current), example
(jsonc commented default), declared_in_jsonc, consumed_by (list of src
files), inferred_type (brace|number|bool|enum|string|unknown).
`inferred_type` mirrors how platformio-custom.py wraps values at build
time: `brace` = byte array `{ 0x01, ... }`, `number` = decimal, `bool` =
true/false, `enum` = `meshtastic_*` constant, `string` = wrapped in quotes
via StringifyMacro.
"""
return userprefs_mod.build_manifest()
@app.tool()
def userprefs_get() -> dict[str, Any]:
"""Return the current userPrefs.jsonc state.
`active` is the dict of uncommented `USERPREFS_*` → value that will be
baked into the next build. `commented` is the dict of commented example
defaults (shown for reference).
"""
state = userprefs_mod.read_state()
# Drop `order` (internal for round-trip rendering) from the public payload.
return {
"path": state["path"],
"active": state["active"],
"commented": state["commented"],
}
@app.tool()
def userprefs_set(prefs: dict[str, Any]) -> dict[str, Any]:
"""Merge `prefs` into userPrefs.jsonc persistently (uncommenting keys).
Existing active values not in `prefs` are kept. To remove a key from the
active set, call `userprefs_reset` (restores the MCP backup if present)
or edit the jsonc manually. Values are stringified the way
platformio-custom.py expects (bool → "true"/"false", int → "42", etc.).
"""
return userprefs_mod.merge_active(prefs)
@app.tool()
def userprefs_reset() -> dict[str, Any]:
"""Restore userPrefs.jsonc from the most recent MCP backup (if any).
The backup is only created by the legacy `userprefs_set` workflow (not
currently written automatically). Returns `{restored: bool, ...}` — false
when no backup is present, in which case the caller should edit the
jsonc directly.
"""
return userprefs_mod.reset()
@app.tool()
def userprefs_testing_profile(
psk_seed: str | None = None,
channel_name: str = "McpTest",
channel_num: int = 88,
region: str = "US",
modem_preset: str = "LONG_FAST",
short_name: str | None = None,
long_name: str | None = None,
disable_mqtt: bool = True,
disable_position: bool = False,
) -> dict[str, Any]:
"""Generate a USERPREFS dict for provisioning an isolated test-mesh device.
Baking this into firmware produces devices that:
- Run on a deterministic non-default LoRa slot (default 88 on US LONG_FAST,
well off the `hash("LongFast")` slot a stock production device uses)
- Join a private channel with a name and PSK that differ from public
defaults — so no accidental mesh-with-production-devices
- Have MQTT disabled (no uplink/downlink bridge), so test traffic never
leaks to a public broker
- Optionally disable GPS for bench-test conditions
For a multi-device test cluster, pass the same `psk_seed` to every call so
every device shares the same PSK and lands on the same isolated mesh.
Returned dict is ready to pass straight to `build`, `pio_flash`,
`erase_and_flash`, or `update_flash` via their `userprefs` parameter.
Example:
profile = userprefs_testing_profile(psk_seed="ci-run-2026-04-16")
erase_and_flash(env="tbeam", port="/dev/cu.usbmodem...", confirm=True,
userprefs=profile)
Args:
psk_seed: seed for deterministic 32-byte PSK via SHA-256. None = random
(fine one-off, useless for multi-device clusters).
channel_name: primary channel name (≤11 chars). Default "McpTest".
channel_num: 1-indexed LoRa slot (0 = fall back to name-hash). Default
88 — mid-upper US band, unlikely to collide with production slots.
region: short code — one of US, EU_433, EU_868, CN, JP, ANZ, KR, TW,
RU, IN, NZ_865, TH, UA_433, UA_868, MY_433, MY_919, SG_923, LORA_24.
modem_preset: one of LONG_FAST, LONG_SLOW, LONG_MODERATE, VERY_LONG_SLOW,
MEDIUM_SLOW, MEDIUM_FAST, SHORT_SLOW, SHORT_FAST, SHORT_TURBO.
short_name: optional owner short name (≤4 chars) stamped into the build.
long_name: optional owner long name stamped into the build.
disable_mqtt: disable MQTT module + uplink/downlink (default True).
disable_position: disable GPS + smart-position broadcasts (default False).
"""
return userprefs_mod.build_testing_profile(
psk_seed=psk_seed,
channel_name=channel_name,
channel_num=channel_num,
region=region,
modem_preset=modem_preset,
short_name=short_name,
long_name=long_name,
disable_mqtt=disable_mqtt,
disable_position=disable_position,
)
@app.tool()
def touch_1200bps(port: str, settle_ms: int = 250) -> dict[str, Any]:
"""Open `port` at 1200 baud and immediately close, triggering USB CDC
bootloader entry on nRF52840, ESP32-S3 (native USB), RP2040, etc.
After the touch, polls serial devices for up to 3 seconds and reports any
new port that appeared (the bootloader often enumerates as a different
device). Not destructive — this is just a reset signal.
"""
return flash.touch_1200bps(port, settle_ms=settle_ms)
# ---------- Serial log sessions -------------------------------------------
@app.tool()
def serial_open(
port: str,
baud: int = 115200,
env: str | None = None,
filters: list[str] | None = None,
) -> dict[str, Any]:
"""Open a `pio device monitor` session reading from `port`.
If `env` is set, pio picks up monitor_speed and monitor_filters from
platformio.ini — recommended for firmware debugging since it enables
esp32_exception_decoder / esp32_c3_exception_decoder for ESP32 envs.
Without `env`, uses the supplied baud and filters (default ["direct"]).
Common filters: direct, time, hexlify, esp32_exception_decoder,
esp32_c3_exception_decoder, log2file.
Returns a session_id for use with serial_read / serial_close, plus the
resolved baud and filters so callers can confirm what pio selected.
"""
session = serial_session.open_session(
port=port, baud=baud, env=env, filters=filters
)
registry.register_session(session)
return {
"session_id": session.id,
"resolved_baud": session.baud,
"resolved_filters": session.filters,
"env": session.env,
}
@app.tool()
def serial_read(
session_id: str,
max_lines: int = 200,
since_cursor: int | None = None,
) -> dict[str, Any]:
"""Read buffered lines from a serial monitor session.
Default: returns everything since your last call to serial_read (uses an
advancing cursor). Pass `since_cursor=N` to re-read from a specific point,
or `since_cursor=0` to read from the start of the in-memory buffer.
Returns `dropped` = count of lines that aged out of the 10k-line ring
buffer between reads — so a value > 0 means you missed data.
"""
session = registry.get_session(session_id)
return serial_session.read_session(
session, max_lines=max_lines, since_cursor=since_cursor
)
@app.tool()
def serial_list() -> list[dict[str, Any]]:
"""List all active serial monitor sessions."""
return [serial_session.session_summary(s) for s in registry.all_sessions()]
@app.tool()
def serial_close(session_id: str) -> dict[str, Any]:
"""Terminate a serial monitor session and free its port."""
session = registry.remove_session(session_id)
if session is None:
return {"ok": False, "reason": f"Unknown session_id {session_id!r}"}
serial_session.close_session(session)
return {"ok": True}
# ---------- Device interaction: reads -------------------------------------
@app.tool()
def device_info(port: str | None = None, timeout_s: float = 8.0) -> dict[str, Any]:
"""Connect via meshtastic.SerialInterface and return a summary of the node.
If `port` is omitted and exactly one likely-Meshtastic device is connected,
it's auto-selected; otherwise the tool errors with the candidate list.
"""
return info.device_info(port=port, timeout_s=timeout_s)
@app.tool()
def list_nodes(port: str | None = None, timeout_s: float = 8.0) -> list[dict[str, Any]]:
"""Return the device's current node database (local node + all known peers)."""
return info.list_nodes(port=port, timeout_s=timeout_s)
# ---------- Device interaction: writes ------------------------------------
@app.tool()
def set_owner(
long_name: str, short_name: str | None = None, port: str | None = None
) -> dict[str, Any]:
"""Set the device's owner long name and (optional) short name (≤4 chars)."""
return admin.set_owner(long_name=long_name, short_name=short_name, port=port)
@app.tool()
def get_config(section: str | None = None, port: str | None = None) -> dict[str, Any]:
"""Read one or all config sections.
`section` may be any LocalConfig section (device, position, power, network,
display, lora, bluetooth, security) or LocalModuleConfig section (mqtt,
serial, telemetry, external_notification, canned_message, range_test,
store_forward, neighbor_info, ambient_lighting, detection_sensor,
paxcounter, audio, remote_hardware, statusmessage, traffic_management).
Omit or pass "all" for every section.
"""
return admin.get_config(section=section, port=port)
@app.tool()
def set_config(path: str, value: Any, port: str | None = None) -> dict[str, Any]:
"""Set one config field via dot-path and write it to the device.
Examples: "lora.region"="US", "lora.modem_preset"="LONG_FAST",
"device.role"="ROUTER", "mqtt.enabled"=True, "mqtt.address"="host".
Enum fields accept their name (case-insensitive) or int.
"""
return admin.set_config(path=path, value=value, port=port)
@app.tool()
def get_channel_url(
include_all: bool = False, port: str | None = None
) -> dict[str, Any]:
"""Get the shareable channel URL (QR-code content).
include_all=True returns the admin URL including all secondary channels;
False returns only the primary channel (what users typically share).
"""
return admin.get_channel_url(include_all=include_all, port=port)
@app.tool()
def set_channel_url(url: str, port: str | None = None) -> dict[str, Any]:
"""Import channels from a Meshtastic channel URL."""
return admin.set_channel_url(url=url, port=port)
@app.tool()
def set_debug_log_api(enabled: bool, port: str | None = None) -> dict[str, Any]:
"""Toggle security.debug_log_api_enabled on the local node.
When true, firmware streams log lines as protobuf `LogRecord` messages
over the StreamAPI (topic `meshtastic.log.line` in meshtastic-python)
instead of raw text. Lets diagnostic clients capture firmware-side logs
through the SAME SerialInterface used for admin/info calls — no
separate `pio device monitor` session needed, no exclusive-port-lock
conflict. Persists across reboot via NVS; wiped by factory_reset
unless re-applied.
The earlier emitLogRecord race (shared tx buffer) is fixed at the
firmware level — the log path has a dedicated scratch + txBuf and
both emission paths serialize via a mutex. Safe to leave on under
traffic.
"""
return admin.set_debug_log_api(enabled=enabled, port=port)
@app.tool()
def send_text(
text: str,
to: str | int | None = None,
channel_index: int = 0,
want_ack: bool = False,
port: str | None = None,
) -> dict[str, Any]:
"""Send a text message over the mesh.
`to` defaults to broadcast ("^all"). Pass a node ID (hex string like
"!abcdef01") or node number (int) to direct-message a specific node.
channel_index picks which configured channel to send on.
"""
return admin.send_text(
text=text, to=to, channel_index=channel_index, want_ack=want_ack, port=port
)
@app.tool()
def reboot(
port: str | None = None, confirm: bool = False, seconds: int = 10
) -> dict[str, Any]:
"""Reboot the connected node in `seconds` seconds. Requires confirm=True."""
return admin.reboot(port=port, confirm=confirm, seconds=seconds)
@app.tool()
def shutdown(
port: str | None = None, confirm: bool = False, seconds: int = 10
) -> dict[str, Any]:
"""Shut down the connected node in `seconds` seconds. Requires confirm=True."""
return admin.shutdown(port=port, confirm=confirm, seconds=seconds)
@app.tool()
def factory_reset(
port: str | None = None, confirm: bool = False, full: bool = False
) -> dict[str, Any]:
"""Factory-reset the connected node. Requires confirm=True.
`full=True` also wipes device identity/keys (not just config).
"""
return admin.factory_reset(port=port, confirm=confirm, full=full)
# ---------- Direct hardware tools -----------------------------------------
@app.tool()
def esptool_chip_info(port: str) -> dict[str, Any]:
"""Run `esptool flash_id` and return chip, MAC, crystal, and flash size.
Read-only — no confirm required. Prefer this over parsing pio upload logs
when you just want to identify the chip.
"""
return hw_tools.esptool_chip_info(port)
@app.tool()
def esptool_erase_flash(port: str, confirm: bool = False) -> dict[str, Any]:
"""Full-chip erase via `esptool erase_flash`. Leaves the device unbootable.
Prefer `erase_and_flash` which also writes firmware. Use this only for
recovery when a device is in a weird state. Requires confirm=True.
"""
return hw_tools.esptool_erase_flash(port, confirm=confirm)
@app.tool()
def esptool_raw(
args: list[str], port: str | None = None, confirm: bool = False
) -> dict[str, Any]:
"""Pass-through to `esptool`. Destructive subcommands (write_flash,
erase_flash, erase_region, merge_bin) require confirm=True.
Prefer the high-level `pio_flash` / `erase_and_flash` / `update_flash`
tools where possible — they know board-specific offsets and protocols.
"""
return hw_tools.esptool_raw(args, port=port, confirm=confirm)
@app.tool()
def nrfutil_dfu(port: str, package_path: str, confirm: bool = False) -> dict[str, Any]:
"""DFU-flash a .zip package to an nRF52840 via `nrfutil dfu serial`.
Prefer `pio_flash` for flashing firmware built from this repo — pio handles
the DFU invocation automatically. Use this tool when flashing a pre-built
release zip or a custom bootloader. Requires confirm=True.
"""
return hw_tools.nrfutil_dfu(port, package_path, confirm=confirm)
@app.tool()
def nrfutil_raw(args: list[str], confirm: bool = False) -> dict[str, Any]:
"""Pass-through to `nrfutil`. dfu/settings subcommands require confirm=True."""
return hw_tools.nrfutil_raw(args, confirm=confirm)
@app.tool()
def picotool_info(port: str | None = None) -> dict[str, Any]:
"""Run `picotool info -a`. Requires the RP2040 to be in BOOTSEL mode
(hold BOOTSEL button while plugging in, or call `touch_1200bps` if the
firmware supports 1200bps-reset)."""
return hw_tools.picotool_info(port=port)
@app.tool()
def picotool_load(uf2_path: str, confirm: bool = False) -> dict[str, Any]:
"""Load a UF2 to a Pico in BOOTSEL mode via `picotool load -x -t uf2`.
Prefer `pio_flash` for flashing firmware built from this repo.
Requires confirm=True.
"""
return hw_tools.picotool_load(uf2_path, confirm=confirm)
@app.tool()
def picotool_raw(args: list[str], confirm: bool = False) -> dict[str, Any]:
"""Pass-through to `picotool`. load/reboot/save/erase require confirm=True."""
return hw_tools.picotool_raw(args, confirm=confirm)
+532
View File
@@ -0,0 +1,532 @@
"""USERPREFS: build-time constants baked into the firmware binary.
The firmware repo has `userPrefs.jsonc` at its root — a JSONC file with every
available USERPREFS_* key listed, most commented out. At build time,
`bin/platformio-custom.py` reads it, strips comments, and emits
`-DUSERPREFS_<KEY>=<value>` build flags into the compile step. Firmware code
uses `#ifdef USERPREFS_<KEY>` to pick up the baked-in defaults for channels,
owner name, LoRa region, OEM branding, MQTT credentials, etc.
This module:
1. Parses `userPrefs.jsonc` (preserving which keys are active vs commented)
2. Greps `src/` for the set of keys the firmware actually consumes (the
real discovery manifest — anything here that isn't in the jsonc is still
a valid override)
3. Provides a context manager for temporarily swapping in overrides during
a build/flash, then restoring the original file
4. Provides persistent `set` / `reset` for when the caller wants the change
to stick across multiple builds
The firmware's platformio-custom.py value-type detection mirrors what we need
for serialization: dict-like `{...}` (byte arrays, enum lists), digit-like
(ints and floats), `true`/`false`, `meshtastic_*` enum constants, and
everything else gets string-wrapped via `env.StringifyMacro`. We store the
raw string values exactly as they'd appear in the jsonc to avoid round-trip
surprises.
"""
from __future__ import annotations
import json
import re
import shutil
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Iterator
from . import config
USERPREFS_FILE = "userPrefs.jsonc"
BACKUP_SUFFIX = ".mcp.bak"
# Pattern for lines like `// "USERPREFS_FOO": "value",` or `"USERPREFS_FOO": "v"`
_ACTIVE_LINE = re.compile(r'^\s*"(USERPREFS_[A-Z0-9_]+)"\s*:\s*"((?:[^"\\]|\\.)*)"')
_COMMENTED_LINE = re.compile(
r'^\s*//\s*"(USERPREFS_[A-Z0-9_]+)"\s*:\s*"((?:[^"\\]|\\.)*)"'
)
# Inline comment stripper (matches platformio-custom.py:219)
_LINE_COMMENT = re.compile(r"//.*")
# USERPREFS_* usage in firmware source (#ifdef, #if defined, direct refs)
_USAGE_PATTERN = re.compile(r"\bUSERPREFS_[A-Z0-9_]+\b")
def jsonc_path() -> Path:
return config.firmware_root() / USERPREFS_FILE
def _read_file(path: Path) -> str:
return path.read_text(encoding="utf-8")
def _parse_jsonc_state(text: str) -> dict[str, Any]:
"""Parse userPrefs.jsonc while preserving comment state per key.
Returns:
{
"active": {key: string_value, ...}, # uncommented
"commented": {key: string_value, ...}, # commented examples
"order": [key, ...] # source order for round-trip
}
"""
active: dict[str, str] = {}
commented: dict[str, str] = {}
order: list[str] = []
for line in text.splitlines():
if m := _COMMENTED_LINE.match(line):
key, val = m.group(1), m.group(2)
commented[key] = val
order.append(key)
elif m := _ACTIVE_LINE.match(line):
key, val = m.group(1), m.group(2)
active[key] = val
order.append(key)
return {"active": active, "commented": commented, "order": order}
def _parse_jsonc_active(text: str) -> dict[str, str]:
"""Parse active-only values by stripping line comments + feeding to json."""
stripped = "\n".join(_LINE_COMMENT.sub("", line) for line in text.splitlines())
try:
return {k: str(v) for k, v in json.loads(stripped).items()}
except json.JSONDecodeError as exc:
raise ValueError(f"userPrefs.jsonc is not valid JSONC: {exc}") from exc
def read_state() -> dict[str, Any]:
"""Return {active, commented, order, path}."""
path = jsonc_path()
if not path.is_file():
return {"active": {}, "commented": {}, "order": [], "path": str(path)}
state = _parse_jsonc_state(_read_file(path))
state["path"] = str(path)
return state
# ---------- Manifest ------------------------------------------------------
def _scan_consumed_keys() -> dict[str, list[str]]:
"""Grep firmware src/ for USERPREFS_* references.
Returns {key: [relative_file_paths]} — only includes files under `src/`.
"""
src_dir = config.firmware_root() / "src"
if not src_dir.is_dir():
return {}
out: dict[str, set[str]] = {}
for path in src_dir.rglob("*"):
if not path.is_file() or path.suffix.lower() not in {
".c",
".cc",
".cpp",
".h",
".hpp",
".ipp",
".inl",
}:
continue
try:
text = path.read_text(encoding="utf-8", errors="ignore")
except Exception:
continue
for m in _USAGE_PATTERN.finditer(text):
key = m.group(0)
# Skip our own "_USERPREFS_" artifacts (reserve-word guard from build-userprefs-json.py)
if key.startswith("_USERPREFS_"):
continue
out.setdefault(key, set()).add(
str(path.relative_to(config.firmware_root()))
)
return {k: sorted(v) for k, v in sorted(out.items())}
def build_manifest() -> dict[str, Any]:
"""Build the discovery manifest.
Every known USERPREFS_* key appears exactly once with:
- `value` (current active value, if any)
- `example` (commented default from jsonc, if any)
- `active` bool
- `declared_in_jsonc` bool (key appears anywhere in userPrefs.jsonc)
- `consumed_by` list of source files that reference it
- `inferred_type`: one of "brace", "number", "bool", "enum", "string"
— matches platformio-custom.py's value-wrapping switch
"""
state = read_state()
consumed = _scan_consumed_keys()
all_keys = set(state["active"]) | set(state["commented"]) | set(consumed)
records = []
for key in sorted(all_keys):
example = state["commented"].get(key)
value = state["active"].get(key)
records.append(
{
"key": key,
"active": key in state["active"],
"value": value,
"example": example,
"declared_in_jsonc": key in state["active"]
or key in state["commented"],
"consumed_by": consumed.get(key, []),
"inferred_type": infer_type(value if value is not None else example),
}
)
return {
"path": state["path"],
"active_count": len(state["active"]),
"commented_count": len(state["commented"]),
"consumed_key_count": len(consumed),
"total_keys": len(records),
"entries": records,
}
def infer_type(value: str | None) -> str:
"""Classify a raw value string the way platformio-custom.py does.
Mirrors the branch order in `bin/platformio-custom.py:222-235`.
"""
if value is None:
return "unknown"
v = value.strip()
if v.startswith("{"):
return "brace" # byte array / enum init list
if v.lstrip("-").replace(".", "", 1).isdigit():
return "number"
if v in ("true", "false"):
return "bool"
if v.startswith("meshtastic_"):
return "enum"
return "string"
# ---------- Writing -------------------------------------------------------
def _format_jsonc_line(key: str, value: str, commented: bool) -> str:
prefix = " // " if commented else " "
# Escape backslashes and quotes inside value the way platformio-custom.py
# expects — the original jsonc uses raw strings for most content. Keep it
# literal; callers are responsible for correct escaping if they pass
# dict/enum-init values that contain quotes.
return f'{prefix}"{key}": "{value}",'
def _render_jsonc(
active: dict[str, str], commented: dict[str, str], order: list[str]
) -> str:
"""Render userPrefs.jsonc preserving source order and comment state."""
seen: set[str] = set()
lines = ["{"]
for key in order:
if key in seen:
continue
seen.add(key)
if key in active:
lines.append(_format_jsonc_line(key, active[key], commented=False))
elif key in commented:
lines.append(_format_jsonc_line(key, commented[key], commented=True))
# Append any newly-added keys (not in original order) at the end, active.
for key, value in active.items():
if key in seen:
continue
seen.add(key)
lines.append(_format_jsonc_line(key, value, commented=False))
# Strip trailing comma on the last data line (valid JSONC allows it, but
# strict `json.loads` after comment-stripping does not; the loader in
# platformio-custom.py uses json.loads).
if len(lines) > 1 and lines[-1].endswith(","):
lines[-1] = lines[-1].rstrip(",")
lines.append("}")
lines.append("") # trailing newline
return "\n".join(lines)
def _validate_after_write(text: str) -> None:
"""Ensure the rendered text still parses the way platformio-custom.py does."""
stripped = "\n".join(_LINE_COMMENT.sub("", line) for line in text.splitlines())
json.loads(stripped) # raises on any error
def write_state(
active: dict[str, str], commented: dict[str, str], order: list[str]
) -> None:
path = jsonc_path()
text = _render_jsonc(active, commented, order)
_validate_after_write(text)
path.write_text(text, encoding="utf-8")
def merge_active(overrides: dict[str, Any]) -> dict[str, Any]:
"""Merge `overrides` into the active set and persist.
Existing active values not in `overrides` are kept. Example/commented
values are preserved. Returns {before_active, after_active, path}.
"""
state = read_state()
before = dict(state["active"])
after = dict(before)
commented = dict(state["commented"])
order = list(state["order"])
for key, raw in overrides.items():
if not key.startswith("USERPREFS_"):
raise ValueError(f"key {key!r} must start with USERPREFS_")
after[key] = _stringify(raw)
# If the key was commented, uncommenting it means removing from commented set.
commented.pop(key, None)
if key not in order:
order.append(key)
write_state(after, commented, order)
return {"before_active": before, "after_active": after, "path": str(jsonc_path())}
def _stringify(value: Any) -> str:
"""Convert a Python value to the string form userPrefs.jsonc expects.
bool → "true" / "false"; int/float → str(); anything else → str(value).
Callers passing brace-init strings (`"{ 0x01, 0x02, ... }"`) must format
them themselves — this function doesn't try to synthesize them.
"""
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, (int, float)):
return str(value)
return str(value)
def reset() -> dict[str, Any]:
"""Restore userPrefs.jsonc from the MCP backup if present.
Returns {restored: bool, path, backup_path}.
"""
path = jsonc_path()
backup = path.with_suffix(path.suffix + BACKUP_SUFFIX)
if backup.is_file():
shutil.copy2(backup, path)
backup.unlink()
return {"restored": True, "path": str(path), "backup_path": str(backup)}
return {"restored": False, "path": str(path), "backup_path": str(backup)}
# ---------- Transient override (for build/flash) --------------------------
# ---------- Pre-baked profiles --------------------------------------------
def _psk_from_bytes(data: bytes) -> str:
"""Format 32 bytes as a C-style brace-init list for USERPREFS_CHANNEL_*_PSK.
Matches the exact format used in userPrefs.jsonc:
{ 0x38, 0x4b, 0xbc, ... }
"""
if len(data) != 32:
raise ValueError(f"PSK must be exactly 32 bytes, got {len(data)}")
return "{ " + ", ".join(f"0x{b:02x}" for b in data) + " }"
def generate_psk(seed: str | None = None) -> str:
"""Generate a 32-byte PSK as a brace-init string.
If `seed` is provided, the PSK is deterministic (derived via SHA-256 of
the seed); otherwise it's cryptographically random. Use a seed for
automated testing so every device in a test run shares the same key.
"""
if seed is None:
import secrets
raw = secrets.token_bytes(32)
else:
import hashlib
raw = hashlib.sha256(seed.encode("utf-8")).digest()
return _psk_from_bytes(raw)
# Meshtastic region enum name → short description (for the manifest tool).
# Not exhaustive; these are the regions a US-based test lab is likely to pick.
KNOWN_REGIONS = {
"US": "meshtastic_Config_LoRaConfig_RegionCode_US",
"EU_433": "meshtastic_Config_LoRaConfig_RegionCode_EU_433",
"EU_868": "meshtastic_Config_LoRaConfig_RegionCode_EU_868",
"CN": "meshtastic_Config_LoRaConfig_RegionCode_CN",
"JP": "meshtastic_Config_LoRaConfig_RegionCode_JP",
"ANZ": "meshtastic_Config_LoRaConfig_RegionCode_ANZ",
"KR": "meshtastic_Config_LoRaConfig_RegionCode_KR",
"TW": "meshtastic_Config_LoRaConfig_RegionCode_TW",
"RU": "meshtastic_Config_LoRaConfig_RegionCode_RU",
"IN": "meshtastic_Config_LoRaConfig_RegionCode_IN",
"NZ_865": "meshtastic_Config_LoRaConfig_RegionCode_NZ_865",
"TH": "meshtastic_Config_LoRaConfig_RegionCode_TH",
"UA_433": "meshtastic_Config_LoRaConfig_RegionCode_UA_433",
"UA_868": "meshtastic_Config_LoRaConfig_RegionCode_UA_868",
"MY_433": "meshtastic_Config_LoRaConfig_RegionCode_MY_433",
"MY_919": "meshtastic_Config_LoRaConfig_RegionCode_MY_919",
"SG_923": "meshtastic_Config_LoRaConfig_RegionCode_SG_923",
"LORA_24": "meshtastic_Config_LoRaConfig_RegionCode_LORA_24",
}
KNOWN_MODEM_PRESETS = {
"LONG_FAST": "meshtastic_Config_LoRaConfig_ModemPreset_LONG_FAST",
"LONG_SLOW": "meshtastic_Config_LoRaConfig_ModemPreset_LONG_SLOW",
"LONG_MODERATE": "meshtastic_Config_LoRaConfig_ModemPreset_LONG_MODERATE",
"VERY_LONG_SLOW": "meshtastic_Config_LoRaConfig_ModemPreset_VERY_LONG_SLOW",
"MEDIUM_SLOW": "meshtastic_Config_LoRaConfig_ModemPreset_MEDIUM_SLOW",
"MEDIUM_FAST": "meshtastic_Config_LoRaConfig_ModemPreset_MEDIUM_FAST",
"SHORT_SLOW": "meshtastic_Config_LoRaConfig_ModemPreset_SHORT_SLOW",
"SHORT_FAST": "meshtastic_Config_LoRaConfig_ModemPreset_SHORT_FAST",
"SHORT_TURBO": "meshtastic_Config_LoRaConfig_ModemPreset_SHORT_TURBO",
}
def build_testing_profile(
psk_seed: str | None = None,
channel_name: str = "McpTest",
channel_num: int = 88,
region: str = "US",
modem_preset: str = "LONG_FAST",
short_name: str | None = None,
long_name: str | None = None,
disable_mqtt: bool = True,
disable_position: bool = False,
) -> dict[str, Any]:
"""Build a USERPREFS dict for an isolated test-mesh device.
Defaults: US region, LONG_FAST modem, channel slot 88 (well away from the
default `hash("LongFast") % numChannels` slot that production devices use),
and a private PSK. Devices baked with the same `psk_seed` land on the same
isolated mesh.
See `src/mesh/RadioInterface.cpp:849` for the slot-selection math:
`slot = (channel_num ? channel_num - 1 : hash(name)) % numChannels`.
Setting `channel_num` explicitly (non-zero) forces a deterministic slot.
Args:
psk_seed: seed for deterministic PSK generation. `None` = random (fine
for one-off bakes, useless for multi-device test clusters).
channel_name: primary channel name. Must differ from defaults
("LongFast", "MediumFast", etc.) so production devices don't
accidentally match after the PSK check.
channel_num: 1-indexed LoRa slot (1..numChannels). 88 is mid-upper US
band. Set to 0 to fall back to name-hash (not recommended for
isolation).
region: short code from `KNOWN_REGIONS`.
modem_preset: short code from `KNOWN_MODEM_PRESETS`.
short_name: optional owner short-name stamp (≤4 chars). None = unset.
long_name: optional owner long-name stamp. None = unset.
disable_mqtt: if True (default), disables the MQTT module and the
uplink/downlink bridge on the primary channel — so private test
traffic never leaks to a public broker.
disable_position: if True, disables GPS + position broadcasts — useful
when test devices sit on a bench without antennas.
"""
if region not in KNOWN_REGIONS:
raise ValueError(
f"Unknown region {region!r}. Known: {sorted(KNOWN_REGIONS.keys())}"
)
if modem_preset not in KNOWN_MODEM_PRESETS:
raise ValueError(
f"Unknown modem_preset {modem_preset!r}. Known: {sorted(KNOWN_MODEM_PRESETS.keys())}"
)
if not (0 <= channel_num <= 255):
raise ValueError(f"channel_num must be 0..255, got {channel_num}")
if len(channel_name) > 11:
raise ValueError(
f"channel_name {channel_name!r} exceeds Meshtastic's 11-char max"
)
if short_name is not None and len(short_name) > 4:
raise ValueError(f"short_name must be ≤4 chars, got {len(short_name)}")
psk = generate_psk(seed=psk_seed)
prefs: dict[str, Any] = {
# --- LoRa ---
"USERPREFS_CONFIG_LORA_REGION": KNOWN_REGIONS[region],
"USERPREFS_LORACONFIG_MODEM_PRESET": KNOWN_MODEM_PRESETS[modem_preset],
"USERPREFS_LORACONFIG_CHANNEL_NUM": channel_num,
# --- Primary channel (isolated from public default) ---
"USERPREFS_CHANNELS_TO_WRITE": 1,
"USERPREFS_CHANNEL_0_NAME": channel_name,
"USERPREFS_CHANNEL_0_PSK": psk,
"USERPREFS_CHANNEL_0_PRECISION": 14,
}
if disable_mqtt:
prefs.update(
{
"USERPREFS_CONFIG_LORA_IGNORE_MQTT": True,
"USERPREFS_MQTT_ENABLED": 0,
"USERPREFS_CHANNEL_0_UPLINK_ENABLED": False,
"USERPREFS_CHANNEL_0_DOWNLINK_ENABLED": False,
}
)
if disable_position:
prefs.update(
{
"USERPREFS_CONFIG_GPS_MODE": "meshtastic_Config_PositionConfig_GpsMode_DISABLED",
"USERPREFS_CONFIG_SMART_POSITION_ENABLED": False,
}
)
if long_name is not None:
prefs["USERPREFS_CONFIG_OWNER_LONG_NAME"] = long_name
if short_name is not None:
prefs["USERPREFS_CONFIG_OWNER_SHORT_NAME"] = short_name
return prefs
@contextmanager
def temporary_overrides(overrides: dict[str, Any] | None) -> Iterator[dict[str, str]]:
"""Apply `overrides` to userPrefs.jsonc for the duration of the context.
Yields a dict of the *effective* active values (original active merged
with overrides). Always restores the original file on exit, even on
exception. If `overrides` is None or empty, this is a no-op.
The restore writes the original file content byte-for-byte, so there's no
round-trip ambiguity even if the file had unusual whitespace.
"""
if not overrides:
state = read_state()
yield dict(state["active"])
return
path = jsonc_path()
if not path.is_file():
raise FileNotFoundError(f"userPrefs.jsonc not found at {path}")
original_bytes = path.read_bytes()
original_stat = path.stat()
# Merge and write
state = _parse_jsonc_state(original_bytes.decode("utf-8"))
effective = dict(state["active"])
commented = dict(state["commented"])
order = list(state["order"])
for key, raw in overrides.items():
if not key.startswith("USERPREFS_"):
raise ValueError(f"key {key!r} must start with USERPREFS_")
effective[key] = _stringify(raw)
commented.pop(key, None)
if key not in order:
order.append(key)
rendered = _render_jsonc(effective, commented, order)
_validate_after_write(rendered)
path.write_text(rendered, encoding="utf-8")
# pio watches file mtimes to invalidate build cache; force the modification
# time to now so a pre-existing `.pio/build/<env>/` cache is discarded.
now = time.time()
import os
os.utime(path, (now, now))
try:
yield effective
finally:
path.write_bytes(original_bytes)
os.utime(path, (original_stat.st_atime, original_stat.st_mtime))