diff --git a/README.md b/README.md index e0b0df2..e448f31 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,25 @@ messages to any reachable node, and local message history that survives restarts β€” Meshtastic radios don't store message history themselves, so OrcMesh keeps its own. +### βš™ USB radio controls + +When connected over USB/Serial, the Device page provides typed controls for +radio and module configuration, owner identity, channels, fixed position, +reboot/shutdown, NodeDB reset, and guarded factory resets. Passwords, PINs, +PSKs, and cryptographic key material are never read into editable controls or +written to logs; secret replacements are write-only. + +The Firmware tab discovers official `meshtastic/firmware` releases for the +target reported by the connected radio. It verifies the release SHA-256, +target/hardware metadata, and image hashes before flashing. Update mode writes +only the application image and preserves settings. Full install requires a +typed confirmation, erases the device, and restores the verified factory, +OTA, and filesystem images at the release-declared offsets. A chip-ID preflight +must identify the expected ESP32 family before either path can write anything. + +Firmware flashing is inherently interruption-sensitive. Use a known-good USB +cable and do not unplug or power off the radio until OrcMesh reports completion. + ### πŸ“Ά Spectrum *(optional β€” needs an RTL-SDR)* A waterfall view of raw RF energy in the LoRa ISM bands. This shows *that* @@ -152,8 +171,6 @@ before reaching the UI or the SQLite store. The map is Leaflet in a ## Roadmap - MeshCore network support alongside Meshtastic -- Editing LoRa region/preset from the app (with appropriate guard rails β€” - region selection has real RF-regulatory meaning) - Compact image transmission over the mesh β€” see [docs/mcoreimg-integration.md](docs/mcoreimg-integration.md) diff --git a/ROADMAP.md b/ROADMAP.md index 35a32d7..de90334 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -46,6 +46,11 @@ below β€” they're cited here for provenance, not as a dependency. but never made it into a downloadable build until this one; also the first release with a proper Windows installer (`MeshChat-Setup-*.exe`, Inno Setup) instead of a raw `dist/` folder +- USB device controls for typed radio/module configuration, owner identity, + channels, fixed position, maintenance actions, and guarded resets +- Official Meshtastic firmware discovery and verified ESP32-S3 update/full + flashing, with target matching, chip-ID preflight, secret-safe logging, and + explicit destructive confirmation ## In progress @@ -64,7 +69,6 @@ Not yet built, in no particular priority order: - Multiple profiles - Multiple simultaneously connected local radios - Message search -- Channel configuration (from the app) - QR/channel URL import (needs explicit security warnings per the spec β€” channel URLs embed the PSK) - Windows installer is unsigned (`packaging/installer.iss`, Inno Setup) β€” @@ -86,14 +90,11 @@ Not yet built, in no particular priority order: first step: an offline IQ-capture-then-decode PoC using `lora-phy` to validate feasibility before any UI is built around it β€” the Spectrum page currently shows *that* RF activity exists, not decoded packet content. -- **Editing LoRa region/preset from the app.** Needs real guard rails β€” - region selection has RF-regulatory meaning, not just a UI preference. - **Compact image transmission over the mesh** β€” see [docs/mcoreimg-integration.md](docs/mcoreimg-integration.md). ## Explicitly out of scope (per the original build spec, Β§1/Β§26) -- Firmware flashing - Reimplementing Meshtastic protobuf framing (use the official `meshtastic` package unless an upstream defect makes that unavoidable) - Storing/logging channel PSKs, Bluetooth PINs, or Wi-Fi credentials diff --git a/THIRD_PARTY_LICENSES.md b/THIRD_PARTY_LICENSES.md index d8dfe77..13f2560 100644 --- a/THIRD_PARTY_LICENSES.md +++ b/THIRD_PARTY_LICENSES.md @@ -14,6 +14,7 @@ package, which is GPL-3.0-only copyleft. | [numpy](https://numpy.org/) | BSD-3-Clause | Numerics | | [platformdirs](https://github.com/platformdirs/platformdirs) | MIT | Per-user data/log paths | | [PyPubSub](https://github.com/schollii/pypubsub) | BSD-2-Clause | Event distribution | +| [esptool](https://github.com/espressif/esptool) | GPL-2.0-or-later | ESP32 firmware flashing | | [pyrtlsdr](https://github.com/pyrtlsdr/pyrtlsdr) | GPL-3.0 | Optional β€” RTL-SDR spectrum capture | ## Bundled web assets diff --git a/packaging/orcmesh.spec b/packaging/orcmesh.spec index ad9ebac..e8e298c 100644 --- a/packaging/orcmesh.spec +++ b/packaging/orcmesh.spec @@ -29,7 +29,10 @@ datas += [(str(ROOT / "src/meshchat/ui/map/web"), "meshchat/ui/map/web")] # 2. Meshtastic protobuf definitions (*.proto files shipped with the package) datas += collect_data_files("meshtastic", includes=["*.proto", "*.pyi"]) -# 3. PySide6 WebEngine resources (icudtl.dat, qtwebengine_resources.pak, etc.) +# 3. esptool flasher stubs (JSON payloads loaded at runtime) +datas += collect_data_files("esptool") + +# 4. PySide6 WebEngine resources (icudtl.dat, qtwebengine_resources.pak, etc.) # PyInstaller's PySide6 hook usually handles these, but be explicit for safety. try: import PySide6 @@ -80,6 +83,9 @@ hiddenimports += ["pubsub.core", "pubsub.utils"] hiddenimports += collect_submodules("pyqtgraph") hiddenimports += ["pyqtgraph.graphicsItems.PlotItem.PlotItem"] +# Official Meshtastic ESP32 firmware flashing +hiddenimports += collect_submodules("esptool") + # platformdirs hiddenimports += ["platformdirs", "platformdirs.windows"] diff --git a/pyproject.toml b/pyproject.toml index ff14b1c..2082293 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,7 @@ dependencies = [ "platformdirs>=4,<5", "pyqtgraph>=0.13", "numpy>=1.26", + "esptool>=5,<6", ] [project.scripts] diff --git a/requirements.txt b/requirements.txt index 3fd136b..3f5a28e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,6 +9,7 @@ PyPubSub>=4.0 platformdirs>=4,<5 pyqtgraph>=0.13 numpy>=1.26 +esptool>=5,<6 # Optional β€” RTL-SDR spectrum waterfall (Spectrum page). # Also needs the native librtlsdr driver; on Windows install the WinUSB diff --git a/src/meshchat/controllers/firmware_controller.py b/src/meshchat/controllers/firmware_controller.py new file mode 100644 index 0000000..1696aef --- /dev/null +++ b/src/meshchat/controllers/firmware_controller.py @@ -0,0 +1,96 @@ +"""Threaded facade for firmware network I/O and flashing.""" +from __future__ import annotations + +import logging + +from PySide6.QtCore import QObject, QThread, Signal, Slot + +from meshchat.services.firmware import discover_release, flash_bundle, prepare_bundle + + +log = logging.getLogger(__name__) + + +class _FirmwareWorker(QObject): + release_found = Signal(object) + progress = Signal(int, int) + bundle_ready = Signal(object) + log = Signal(str) + completed = Signal(str, bool, str) + + @Slot(str, bool) + def discover(self, pio_env: str, include_prerelease: bool) -> None: + try: + release = discover_release(pio_env, include_prerelease) + self.release_found.emit(release) + self.completed.emit("discover", True, f"Found Meshtastic {release.version}") + except Exception as exc: + log.exception("Firmware release discovery failed") + self.completed.emit("discover", False, str(exc)) + + @Slot(object, str, str) + def prepare(self, release, pio_env: str, hw_model: str) -> None: + try: + bundle = prepare_bundle( + release, pio_env, hw_model, + lambda done, total: self.progress.emit(done, total), + ) + self.bundle_ready.emit(bundle) + self.completed.emit("prepare", True, "Firmware downloaded and verified") + except Exception as exc: + log.exception("Firmware preparation failed") + self.completed.emit("prepare", False, str(exc)) + + @Slot(object, str, bool, object) + def flash(self, bundle, port: str, full_install: bool, expected_usb) -> None: + try: + flash_bundle( + bundle, port, full_install, expected_usb, + lambda line: self.log.emit(line), + ) + self.completed.emit("flash", True, "Firmware flash completed") + except Exception as exc: + log.exception("Firmware flash failed") + self.completed.emit("flash", False, str(exc)) + + +class FirmwareController(QObject): + release_found = Signal(object) + progress = Signal(int, int) + bundle_ready = Signal(object) + log = Signal(str) + completed = Signal(str, bool, str) + + _discover_requested = Signal(str, bool) + _prepare_requested = Signal(object, str, str) + _flash_requested = Signal(object, str, bool, object) + + def __init__(self, parent=None): + super().__init__(parent) + self._worker = _FirmwareWorker() + self._thread = QThread(self) + self._worker.moveToThread(self._thread) + self._worker.release_found.connect(self.release_found) + self._worker.progress.connect(self.progress) + self._worker.bundle_ready.connect(self.bundle_ready) + self._worker.log.connect(self.log) + self._worker.completed.connect(self.completed) + self._discover_requested.connect(self._worker.discover) + self._prepare_requested.connect(self._worker.prepare) + self._flash_requested.connect(self._worker.flash) + self._thread.start() + + def discover(self, pio_env: str, include_prerelease: bool = False) -> None: + self._discover_requested.emit(pio_env, include_prerelease) + + def prepare(self, release, pio_env: str, hw_model: str) -> None: + self._prepare_requested.emit(release, pio_env, hw_model) + + def flash(self, bundle, port: str, full_install: bool, expected_usb) -> None: + self._flash_requested.emit(bundle, port, full_install, expected_usb) + + def shutdown(self) -> None: + self._thread.quit() + if not self._thread.wait(5000): + log.warning("Firmware operation still active; waiting for it to finish") + self._thread.wait() diff --git a/src/meshchat/controllers/meshtastic_controller.py b/src/meshchat/controllers/meshtastic_controller.py index 1ac7598..38c9578 100644 --- a/src/meshchat/controllers/meshtastic_controller.py +++ b/src/meshchat/controllers/meshtastic_controller.py @@ -46,6 +46,7 @@ class ErrorCode(Enum): CONNECTION_LOST = "connection_lost" SEND_FAILED = "send_failed" INVALID_MESSAGE = "invalid_message" + DEVICE_CONTROL_FAILED = "device_control_failed" INTERNAL_ERROR = "internal_error" @@ -112,6 +113,9 @@ class DeviceSummary: hw_model: str | None firmware_version: str | None node_num: int | None = None + pio_env: str | None = None + serial_port: str | None = None + can_shutdown: bool = False @dataclass(frozen=True) @@ -216,7 +220,7 @@ def _extract_channels(interface) -> list[ChannelSummary]: return sorted(result, key=lambda c: c.index) -def _device_summary(interface) -> DeviceSummary: +def _device_summary(interface, serial_port: str | None = None) -> DeviceSummary: try: my_info = getattr(interface, "myInfo", None) or {} metadata = getattr(interface, "metadata", None) @@ -235,6 +239,9 @@ def _device_summary(interface) -> DeviceSummary: hw_model=user.get("hwModel"), firmware_version=fw, node_num=local_num, + pio_env=getattr(my_info, "pio_env", None), + serial_port=serial_port, + can_shutdown=bool(getattr(metadata, "can_shutdown", False)), ) except Exception as exc: log.warning("DeviceSummary extraction failed: %s", exc) @@ -316,12 +323,15 @@ class MeshtasticWorker(QObject): error_occurred = Signal(object) # UserFacingError diagnostic_log = Signal(str) raw_packet = Signal(dict) # for monitor ingestion + device_controls_updated = Signal(object) + device_operation_completed = Signal(str, str) def __init__(self, parent=None): super().__init__(parent) self._interface = None self._state = ConnectionState.DISCONNECTED self._subscribed = False + self._serial_port: str | None = None # ----------------------------------------------------------------------- # State helpers @@ -388,13 +398,14 @@ def _on_connection_established(self, interface=None, topics=None) -> None: if not self._is_active_interface(interface): return try: - summary = _device_summary(interface) + summary = _device_summary(interface, self._serial_port) channels = _extract_channels(interface) lora_config = _extract_lora_config(interface) self._set_state(ConnectionState.CONNECTED, summary.long_name or "") self.connected.emit(summary) self.channels_updated.emit(channels) self.lora_config_updated.emit(lora_config) + self._emit_device_controls() # The meshtastic library has already downloaded the radio's full # NodeDB by the time this event fires β€” push it out so the UI can @@ -571,6 +582,7 @@ def connect_ble(self, address: str) -> None: if self._state not in (ConnectionState.DISCONNECTED, ConnectionState.ERROR, ConnectionState.RECONNECTING): return self._close_interface() + self._serial_port = None self._subscribe() self._set_state(ConnectionState.CONNECTING, address) try: @@ -626,6 +638,7 @@ def connect_tcp(self, host: str, port: int = 4403) -> None: return self._close_interface() + self._serial_port = None self._subscribe() self._set_state(ConnectionState.CONNECTING, f"{host}:{port}") try: @@ -692,6 +705,7 @@ def connect_serial(self, port: str) -> None: return self._close_interface() + self._serial_port = port self._subscribe() self._set_state(ConnectionState.CONNECTING, port) try: @@ -897,6 +911,124 @@ def remove_node(self, node_num: int) -> None: self._emit_error(ErrorCode.SEND_FAILED, "Action Failed", f"Could not remove node: {exc}", True) + # ----------------------------------------------------------------------- + # Slots: connected-radio configuration and maintenance + # ----------------------------------------------------------------------- + + def _emit_device_controls(self) -> None: + if self._interface is None: + return + try: + from meshchat.services.device_config import build_snapshot + self.device_controls_updated.emit(build_snapshot(self._interface, self._serial_port)) + except Exception as exc: + log.exception("Device control snapshot failed") + self._emit_error( + ErrorCode.DEVICE_CONTROL_FAILED, "Device Read Failed", + "Could not read the connected radio controls.", True, str(exc), + ) + + @Slot() + def refresh_device_controls(self) -> None: + if self._require_connection("read device settings"): + self._emit_device_controls() + + @Slot(str, object) + def apply_device_section(self, section: str, changes: dict) -> None: + if not self._require_connection("change device settings"): + return + try: + from meshchat.services.device_config import apply_section + apply_section(self._interface.localNode, section, changes) + self.device_operation_completed.emit("config", f"Saved {section.replace('_', ' ')} settings") + self._emit_device_controls() + except Exception as exc: + log.exception("Device setting write failed for %s", section) + self._emit_error( + ErrorCode.DEVICE_CONTROL_FAILED, "Settings Write Failed", + f"Could not save {section.replace('_', ' ')} settings.", True, str(exc), + ) + + @Slot(str, str) + def set_owner(self, long_name: str, short_name: str) -> None: + if not self._require_connection("change device identity"): + return + try: + self._interface.localNode.setOwner(long_name=long_name, short_name=short_name) + self.device_operation_completed.emit("owner", "Device identity saved") + self._emit_device_controls() + except Exception as exc: + log.exception("Owner write failed") + self._emit_error(ErrorCode.DEVICE_CONTROL_FAILED, "Identity Write Failed", + "Could not save the device identity.", True, str(exc)) + + @Slot(object) + def update_channel(self, changes: dict) -> None: + if not self._require_connection("change channel settings"): + return + try: + from meshtastic.util import fromPSK + node = self._interface.localNode + index = int(changes["index"]) + if index < 0 or index >= len(node.channels): + raise ValueError("Invalid channel index") + channel = node.channels[index] + channel.role = int(changes["role"]) + channel.settings.name = str(changes.get("name", "")).strip() + channel.settings.uplink_enabled = bool(changes.get("uplink_enabled", False)) + channel.settings.downlink_enabled = bool(changes.get("downlink_enabled", False)) + channel.settings.module_settings.position_precision = int( + changes.get("position_precision", 0) + ) + replacement_psk = str(changes.get("psk", "")).strip() + if replacement_psk: + channel.settings.psk = fromPSK(replacement_psk) + node.writeChannel(index) + self.device_operation_completed.emit("channel", f"Saved channel {index}") + self._emit_device_controls() + except Exception as exc: + log.exception("Channel write failed") + self._emit_error(ErrorCode.DEVICE_CONTROL_FAILED, "Channel Write Failed", + "Could not save the channel.", True, str(exc)) + + def _run_local_node_action(self, action: str, callback) -> None: + if not self._require_connection(action): + return + try: + callback(self._interface.localNode) + self.device_operation_completed.emit(action, f"{action.title()} command sent") + except Exception as exc: + log.exception("Device action failed: %s", action) + self._emit_error(ErrorCode.DEVICE_CONTROL_FAILED, "Device Command Failed", + f"Could not {action} the radio.", True, str(exc)) + + @Slot(int) + def reboot_device(self, seconds: int = 2) -> None: + self._run_local_node_action("reboot", lambda node: node.reboot(max(0, seconds))) + + @Slot(int) + def shutdown_device(self, seconds: int = 2) -> None: + self._run_local_node_action("shutdown", lambda node: node.shutdown(max(0, seconds))) + + @Slot() + def reset_nodedb(self) -> None: + self._run_local_node_action("reset node database", lambda node: node.resetNodeDb()) + + @Slot(bool) + def factory_reset(self, full: bool = False) -> None: + self._run_local_node_action("factory reset", lambda node: node.factoryReset(full=full)) + + @Slot(float, float, int) + def set_fixed_position(self, latitude: float, longitude: float, altitude: int) -> None: + self._run_local_node_action( + "set fixed position", + lambda node: node.setFixedPosition(latitude, longitude, altitude), + ) + + @Slot() + def remove_fixed_position(self) -> None: + self._run_local_node_action("remove fixed position", lambda node: node.removeFixedPosition()) + # ----------------------------------------------------------------------- # Slot: Shutdown # ----------------------------------------------------------------------- @@ -971,6 +1103,16 @@ class MeshtasticController(QObject): error_occurred = Signal(object) diagnostic_log = Signal(str) raw_packet = Signal(dict) + device_controls_updated = Signal(object) + device_operation_completed = Signal(str, str) + + _apply_section_requested = Signal(str, object) + _set_owner_requested = Signal(str, str) + _update_channel_requested = Signal(object) + _reboot_requested = Signal(int) + _shutdown_requested = Signal(int) + _factory_reset_requested = Signal(bool) + _fixed_position_requested = Signal(float, float, int) def __init__(self, parent=None): super().__init__(parent) @@ -996,6 +1138,16 @@ def __init__(self, parent=None): w.error_occurred.connect(self.error_occurred) w.diagnostic_log.connect(self.diagnostic_log) w.raw_packet.connect(self.raw_packet) + w.device_controls_updated.connect(self.device_controls_updated) + w.device_operation_completed.connect(self.device_operation_completed) + + self._apply_section_requested.connect(w.apply_device_section) + self._set_owner_requested.connect(w.set_owner) + self._update_channel_requested.connect(w.update_channel) + self._reboot_requested.connect(w.reboot_device) + self._shutdown_requested.connect(w.shutdown_device) + self._factory_reset_requested.connect(w.factory_reset) + self._fixed_position_requested.connect(w.set_fixed_position) self._thread.start() @@ -1072,6 +1224,43 @@ def remove_node(self, node_num: int) -> None: QMetaObject.invokeMethod(self._worker, "remove_node", Qt.ConnectionType.QueuedConnection, Q_ARG("qlonglong", node_num)) + def refresh_device_controls(self) -> None: + from PySide6.QtCore import QMetaObject, Qt + QMetaObject.invokeMethod( + self._worker, "refresh_device_controls", Qt.ConnectionType.QueuedConnection + ) + + def apply_device_section(self, section: str, changes: dict) -> None: + self._apply_section_requested.emit(section, changes) + + def set_owner(self, long_name: str, short_name: str) -> None: + self._set_owner_requested.emit(long_name, short_name) + + def update_channel(self, changes: dict) -> None: + self._update_channel_requested.emit(changes) + + def reboot_device(self, seconds: int = 2) -> None: + self._reboot_requested.emit(seconds) + + def shutdown_device(self, seconds: int = 2) -> None: + self._shutdown_requested.emit(seconds) + + def reset_nodedb(self) -> None: + from PySide6.QtCore import QMetaObject, Qt + QMetaObject.invokeMethod(self._worker, "reset_nodedb", Qt.ConnectionType.QueuedConnection) + + def factory_reset(self, full: bool = False) -> None: + self._factory_reset_requested.emit(full) + + def set_fixed_position(self, latitude: float, longitude: float, altitude: int) -> None: + self._fixed_position_requested.emit(latitude, longitude, altitude) + + def remove_fixed_position(self) -> None: + from PySide6.QtCore import QMetaObject, Qt + QMetaObject.invokeMethod( + self._worker, "remove_fixed_position", Qt.ConnectionType.QueuedConnection + ) + def shutdown(self) -> None: from PySide6.QtCore import QMetaObject, Qt QMetaObject.invokeMethod(self._worker, "shutdown", Qt.ConnectionType.QueuedConnection) diff --git a/src/meshchat/models/device_control.py b/src/meshchat/models/device_control.py new file mode 100644 index 0000000..376ce0e --- /dev/null +++ b/src/meshchat/models/device_control.py @@ -0,0 +1,60 @@ +"""Typed, secret-safe models for connected-radio controls.""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass(frozen=True) +class ConfigChoice: + label: str + value: int + + +@dataclass(frozen=True) +class ConfigField: + name: str + label: str + kind: str + value: Any + choices: tuple[ConfigChoice, ...] = () + repeated: bool = False + write_only: bool = False + read_only: bool = False + + +@dataclass(frozen=True) +class ConfigSection: + name: str + label: str + fields: tuple[ConfigField, ...] + + +@dataclass(frozen=True) +class ChannelControl: + index: int + role: int + role_name: str + name: str + uplink_enabled: bool + downlink_enabled: bool + position_precision: int + + +@dataclass(frozen=True) +class DeviceControlSnapshot: + node_id: str | None + long_name: str + short_name: str + hw_model: str + firmware_version: str + pio_env: str + serial_port: str | None + usb_vid: int | None + usb_pid: int | None + usb_serial: str | None + can_shutdown: bool + has_wifi: bool + has_bluetooth: bool + sections: tuple[ConfigSection, ...] = field(default_factory=tuple) + channels: tuple[ChannelControl, ...] = field(default_factory=tuple) diff --git a/src/meshchat/services/device_config.py b/src/meshchat/services/device_config.py new file mode 100644 index 0000000..38156f8 --- /dev/null +++ b/src/meshchat/services/device_config.py @@ -0,0 +1,213 @@ +"""Build and apply secret-safe Meshtastic configuration snapshots.""" +from __future__ import annotations + +import re +from typing import Any + +from google.protobuf.descriptor import FieldDescriptor + +from meshchat.models.device_control import ( + ChannelControl, + ConfigChoice, + ConfigField, + ConfigSection, + DeviceControlSnapshot, +) + + +_WRITE_ONLY = { + ("network", "wifi_psk"), + ("mqtt", "password"), + ("bluetooth", "fixed_pin"), +} +_READ_ONLY = { + ("security", "public_key"), + ("security", "private_key"), + ("security", "admin_key"), +} +_SKIP_SECTIONS = {"version"} +_SKIP_FIELDS = {"version"} +_WRITABLE_SECTIONS = { + "device", "position", "power", "network", "display", "lora", "bluetooth", "security", + "mqtt", "serial", "external_notification", "store_forward", "range_test", "telemetry", + "canned_message", "audio", "remote_hardware", "neighbor_info", "ambient_lighting", + "detection_sensor", "paxcounter", "traffic_management", +} + +_INT_TYPES = { + FieldDescriptor.TYPE_INT32, + FieldDescriptor.TYPE_INT64, + FieldDescriptor.TYPE_UINT32, + FieldDescriptor.TYPE_UINT64, + FieldDescriptor.TYPE_SINT32, + FieldDescriptor.TYPE_SINT64, + FieldDescriptor.TYPE_FIXED32, + FieldDescriptor.TYPE_FIXED64, + FieldDescriptor.TYPE_SFIXED32, + FieldDescriptor.TYPE_SFIXED64, +} +_FLOAT_TYPES = {FieldDescriptor.TYPE_FLOAT, FieldDescriptor.TYPE_DOUBLE} + + +def _label(name: str) -> str: + return re.sub(r"\s+", " ", name.replace("_", " ")).strip().title() + + +def _field_kind(field: FieldDescriptor) -> str | None: + if field.type == FieldDescriptor.TYPE_BOOL: + return "bool" + if field.type == FieldDescriptor.TYPE_ENUM: + return "enum" + if field.type in _INT_TYPES: + return "int" + if field.type in _FLOAT_TYPES: + return "float" + if field.type == FieldDescriptor.TYPE_STRING: + return "string" + if field.type == FieldDescriptor.TYPE_BYTES: + return "bytes" + return None + + +def _section(name: str, message) -> ConfigSection: + fields: list[ConfigField] = [] + for descriptor in message.DESCRIPTOR.fields: + if descriptor.name in _SKIP_FIELDS: + continue + kind = _field_kind(descriptor) + if kind is None: + continue + write_only = (name, descriptor.name) in _WRITE_ONLY + read_only = (name, descriptor.name) in _READ_ONLY or kind == "bytes" + raw = getattr(message, descriptor.name) + if write_only or read_only: + value: Any = "" + elif descriptor.is_repeated: + value = list(raw) + else: + value = raw + choices: tuple[ConfigChoice, ...] = () + if descriptor.enum_type is not None: + choices = tuple(ConfigChoice(v.name, v.number) for v in descriptor.enum_type.values) + fields.append(ConfigField( + name=descriptor.name, + label=_label(descriptor.name), + kind=kind, + value=value, + choices=choices, + repeated=descriptor.is_repeated, + write_only=write_only, + read_only=read_only, + )) + return ConfigSection(name=name, label=_label(name), fields=tuple(fields)) + + +def build_snapshot(interface, serial_port: str | None) -> DeviceControlSnapshot: + node = interface.localNode + my_info = getattr(interface, "myInfo", None) + metadata = getattr(interface, "metadata", None) + local_num = getattr(my_info, "my_node_num", None) + local = (getattr(interface, "nodesByNum", {}) or {}).get(local_num, {}) + user = local.get("user", {}) if isinstance(local, dict) else {} + + sections: list[ConfigSection] = [] + for parent in (getattr(node, "localConfig", None), getattr(node, "moduleConfig", None)): + if parent is None: + continue + for descriptor in parent.DESCRIPTOR.fields: + if ( + descriptor.name in _SKIP_SECTIONS + or descriptor.name not in _WRITABLE_SECTIONS + or descriptor.type != FieldDescriptor.TYPE_MESSAGE + ): + continue + sections.append(_section(descriptor.name, getattr(parent, descriptor.name))) + + channels: list[ChannelControl] = [] + from meshtastic.protobuf import channel_pb2 + for channel in getattr(node, "channels", None) or []: + try: + role_name = channel_pb2.Channel.Role.Name(channel.role) + except ValueError: + role_name = "UNKNOWN" + settings = channel.settings + channels.append(ChannelControl( + index=int(channel.index), + role=int(channel.role), + role_name=role_name, + name=settings.name, + uplink_enabled=bool(settings.uplink_enabled), + downlink_enabled=bool(settings.downlink_enabled), + position_precision=int(settings.module_settings.position_precision), + )) + + usb_port = None + if serial_port: + from serial.tools import list_ports + usb_port = next((port for port in list_ports.comports() if port.device == serial_port), None) + return DeviceControlSnapshot( + node_id=user.get("id"), + long_name=user.get("longName") or "", + short_name=user.get("shortName") or "", + hw_model=user.get("hwModel") or getattr(metadata, "hw_model", "") or "", + firmware_version=getattr(metadata, "firmware_version", "") or "", + pio_env=getattr(my_info, "pio_env", "") or "", + serial_port=serial_port, + usb_vid=getattr(usb_port, "vid", None), + usb_pid=getattr(usb_port, "pid", None), + usb_serial=getattr(usb_port, "serial_number", None), + can_shutdown=bool(getattr(metadata, "can_shutdown", False)), + has_wifi=bool(getattr(metadata, "has_wifi", False)), + has_bluetooth=bool(getattr(metadata, "has_bluetooth", False)), + sections=tuple(sections), + channels=tuple(channels), + ) + + +def _coerce(field: FieldDescriptor, value: Any) -> Any: + if field.is_repeated: + if not isinstance(value, list): + raise ValueError(f"{field.name} must be a list") + return [_coerce_scalar(field, item) for item in value] + return _coerce_scalar(field, value) + + +def _coerce_scalar(field: FieldDescriptor, value: Any) -> Any: + if field.type == FieldDescriptor.TYPE_BOOL: + return bool(value) + if field.type == FieldDescriptor.TYPE_ENUM: + number = int(value) + if field.enum_type.values_by_number.get(number) is None: + raise ValueError(f"invalid value for {field.name}") + return number + if field.type in _INT_TYPES: + return int(value) + if field.type in _FLOAT_TYPES: + return float(value) + if field.type == FieldDescriptor.TYPE_STRING: + return str(value) + raise ValueError(f"{field.name} is not editable") + + +def apply_section(node, section_name: str, changes: dict[str, Any]) -> None: + if section_name not in _WRITABLE_SECTIONS: + raise ValueError(f"Unknown configuration section: {section_name}") + parent = node.localConfig if hasattr(node.localConfig, section_name) else node.moduleConfig + if not hasattr(parent, section_name): + raise ValueError(f"Unknown configuration section: {section_name}") + section = getattr(parent, section_name) + fields = section.DESCRIPTOR.fields_by_name + for name, value in changes.items(): + field = fields.get(name) + if field is None or (section_name, name) in _READ_ONLY or field.type == FieldDescriptor.TYPE_BYTES: + raise ValueError(f"{name} is not editable") + if (section_name, name) in _WRITE_ONLY and value in (None, ""): + continue + coerced = _coerce(field, value) + if field.is_repeated: + target = getattr(section, name) + del target[:] + target.extend(coerced) + else: + setattr(section, name, coerced) + node.writeConfig(section_name) diff --git a/src/meshchat/services/firmware.py b/src/meshchat/services/firmware.py new file mode 100644 index 0000000..22630e8 --- /dev/null +++ b/src/meshchat/services/firmware.py @@ -0,0 +1,358 @@ +"""Official Meshtastic firmware discovery, validation, and ESP32 flashing.""" +from __future__ import annotations + +import hashlib +import json +import os +import re +import time +import urllib.request +import zipfile +from dataclasses import dataclass +from pathlib import Path +from typing import Callable +from urllib.parse import urlsplit + +import platformdirs + + +_RELEASES_API = "https://api.github.com/repos/meshtastic/firmware/releases?per_page=20" +_USER_AGENT = "OrcMesh-firmware/0.2" +_MAX_ASSET_BYTES = 300 * 1024 * 1024 +_PORT_RE = re.compile(r"^COM\d+$", re.IGNORECASE) +_GITHUB_HOSTS = { + "api.github.com", + "github.com", + "objects.githubusercontent.com", + "release-assets.githubusercontent.com", +} +_CHIP_BY_PLATFORM = {"esp32s3": "esp32s3"} + + +class FirmwareError(RuntimeError): + pass + + +@dataclass(frozen=True) +class FirmwareRelease: + tag: str + version: str + prerelease: bool + published_at: str + asset_name: str + asset_url: str + asset_size: int + asset_sha256: str + platform: str + + +@dataclass(frozen=True) +class FirmwareBundle: + release: FirmwareRelease + root: Path + pio_env: str + hw_model: str + requires_dfu: bool + update_image: Path + factory_image: Path + ota_image: Path + filesystem_image: Path + ota_offset: str + filesystem_offset: str + file_md5: dict[str, str] + + +def _validate_url(url: str) -> None: + parsed = urlsplit(url) + if parsed.scheme != "https" or parsed.hostname not in _GITHUB_HOSTS: + raise FirmwareError("Firmware network access is restricted to official GitHub HTTPS hosts.") + + +def _json(url: str): + _validate_url(url) + request = urllib.request.Request(url, headers={ + "Accept": "application/vnd.github+json", + "User-Agent": _USER_AGENT, + "X-GitHub-Api-Version": "2022-11-28", + }) + with urllib.request.urlopen(request, timeout=30) as response: + _validate_url(response.geturl()) + return json.load(response) + + +def discover_release(pio_env: str, include_prerelease: bool = False) -> FirmwareRelease: + if not pio_env or not re.fullmatch(r"[A-Za-z0-9_.-]+", pio_env): + raise FirmwareError("The connected radio did not report a valid firmware target.") + releases = _json(_RELEASES_API) + release = next(( + item for item in releases + if not item.get("draft") and (include_prerelease or not item.get("prerelease")) + ), None) + if release is None: + raise FirmwareError("No matching Meshtastic firmware release is available.") + version = str(release["tag_name"]).removeprefix("v") + manifest_name = f"firmware-{version}.json" + manifest_asset = next((a for a in release["assets"] if a["name"] == manifest_name), None) + if manifest_asset is None: + raise FirmwareError("The release does not contain a firmware target manifest.") + manifest = _json(manifest_asset["browser_download_url"]) + target = next((t for t in manifest.get("targets", []) if t.get("board") == pio_env), None) + if target is None: + raise FirmwareError(f"Release {version} does not support {pio_env}.") + platform = target["platform"] + if platform not in _CHIP_BY_PLATFORM: + raise FirmwareError(f"OrcMesh firmware flashing does not support platform {platform}.") + asset_name = f"firmware-{platform}-{version}.zip" + asset = next((a for a in release["assets"] if a["name"] == asset_name), None) + if asset is None: + raise FirmwareError(f"Release {version} is missing {asset_name}.") + size = int(asset.get("size") or 0) + if not 0 < size <= _MAX_ASSET_BYTES: + raise FirmwareError("The firmware asset has an invalid size.") + digest = str(asset.get("digest") or "") + if not digest.startswith("sha256:"): + raise FirmwareError("The official release asset does not provide a SHA-256 digest.") + return FirmwareRelease( + tag=release["tag_name"], + version=version, + prerelease=bool(release.get("prerelease")), + published_at=release.get("published_at") or "", + asset_name=asset_name, + asset_url=asset["browser_download_url"], + asset_size=size, + asset_sha256=digest.removeprefix("sha256:"), + platform=platform, + ) + + +def _cache_root() -> Path: + root = Path(platformdirs.user_cache_dir("OrcMesh", appauthor=False)) / "firmware" + root.mkdir(parents=True, exist_ok=True) + return root + + +def _hash(path: Path, algorithm: str) -> str: + digest = hashlib.new(algorithm) + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def _download(release: FirmwareRelease, progress: Callable[[int, int], None]) -> Path: + destination = _cache_root() / release.asset_name + if destination.exists() and destination.stat().st_size == release.asset_size: + if _hash(destination, "sha256") == release.asset_sha256: + progress(release.asset_size, release.asset_size) + return destination + destination.unlink() + partial = destination.with_suffix(destination.suffix + ".partial") + _validate_url(release.asset_url) + request = urllib.request.Request(release.asset_url, headers={"User-Agent": _USER_AGENT}) + received = 0 + with urllib.request.urlopen(request, timeout=60) as response, partial.open("wb") as output: + _validate_url(response.geturl()) + while block := response.read(1024 * 1024): + output.write(block) + received += len(block) + if received > _MAX_ASSET_BYTES: + raise FirmwareError("Firmware download exceeded the safety limit.") + progress(received, release.asset_size) + if received != release.asset_size or _hash(partial, "sha256") != release.asset_sha256: + partial.unlink(missing_ok=True) + raise FirmwareError("Firmware download failed its size or SHA-256 verification.") + os.replace(partial, destination) + return destination + + +def prepare_bundle( + release: FirmwareRelease, + pio_env: str, + hw_model: str, + progress: Callable[[int, int], None] = lambda _done, _total: None, +) -> FirmwareBundle: + archive = _download(release, progress) + destination = _cache_root() / release.version / pio_env + destination.mkdir(parents=True, exist_ok=True) + metadata_name = f"firmware-{pio_env}-{release.version}.mt.json" + wanted_prefixes = ( + f"firmware-{pio_env}-{release.version}.bin", + f"firmware-{pio_env}-{release.version}.factory.bin", + f"littlefs-{pio_env}-{release.version}.bin", + f"mt-{release.platform}-ota.bin", + metadata_name, + ) + with zipfile.ZipFile(archive) as bundle_zip: + members = {Path(info.filename).name: info for info in bundle_zip.infolist()} + missing = [name for name in wanted_prefixes if name not in members] + if missing: + raise FirmwareError("Firmware bundle is missing: " + ", ".join(missing)) + for name in wanted_prefixes: + info = members[name] + if info.file_size > 32 * 1024 * 1024: + raise FirmwareError(f"Unexpectedly large firmware member: {name}") + target = destination / name + with bundle_zip.open(info) as source, target.open("wb") as output: + while block := source.read(1024 * 1024): + output.write(block) + + metadata = json.loads((destination / metadata_name).read_text(encoding="utf-8")) + if metadata.get("platformioTarget") != pio_env: + raise FirmwareError("Firmware target metadata does not match the connected radio.") + expected_hw = metadata.get("hwModelSlug") or "" + if hw_model and expected_hw and expected_hw != hw_model: + raise FirmwareError(f"Firmware hardware model {expected_hw} does not match {hw_model}.") + if metadata.get("activelySupported") is False: + raise FirmwareError("This firmware target is no longer actively supported.") + hashes = {item["name"]: item["md5"] for item in metadata.get("files", []) if item.get("md5")} + for name in wanted_prefixes[:-1]: + expected = hashes.get(name) + if not expected or not re.fullmatch(r"[0-9a-fA-F]{32}", expected): + raise FirmwareError(f"Firmware metadata is missing a valid image hash: {name}") + if _hash(destination / name, "md5") != expected.lower(): + raise FirmwareError(f"Firmware member failed verification: {name}") + parts = {part["subtype"]: part["offset"] for part in metadata.get("part", [])} + if "ota_1" not in parts or "spiffs" not in parts: + raise FirmwareError("Firmware partition metadata is incomplete.") + for subtype in ("ota_1", "spiffs"): + offset = str(parts[subtype]) + if not re.fullmatch(r"0x[0-9a-fA-F]+", offset) or not 0 < int(offset, 16) < 32 * 1024 * 1024: + raise FirmwareError(f"Firmware partition offset is invalid: {subtype}") + return FirmwareBundle( + release=release, + root=destination, + pio_env=pio_env, + hw_model=expected_hw or hw_model, + requires_dfu=bool(metadata.get("requiresDfu")), + update_image=destination / wanted_prefixes[0], + factory_image=destination / wanted_prefixes[1], + filesystem_image=destination / wanted_prefixes[2], + ota_image=destination / wanted_prefixes[3], + ota_offset=parts["ota_1"], + filesystem_offset=parts["spiffs"], + file_md5=hashes, + ) + + +def validate_bundle(bundle: FirmwareBundle) -> None: + for path in ( + bundle.update_image, bundle.factory_image, + bundle.ota_image, bundle.filesystem_image, + ): + expected = bundle.file_md5.get(path.name) + if ( + not path.is_file() + or not expected + or not re.fullmatch(r"[0-9a-fA-F]{32}", expected) + or _hash(path, "md5") != expected.lower() + ): + raise FirmwareError(f"Firmware file is missing or changed: {path.name}") + + +def _automatic_bootloader_port( + port: str, + expected_usb: tuple[int | None, int | None, str | None] | None, + output: Callable[[str], None], +) -> str: + from serial import Serial, SerialException + from serial.tools import list_ports + + output("Normal reset failed; trying automatic 1200-bps bootloader entry.") + try: + with Serial(port=port, baudrate=1200, timeout=1): + time.sleep(0.5) + except SerialException: + # The normal reset may already have started USB re-enumeration. + pass + + deadline = time.monotonic() + 8 + while time.monotonic() < deadline: + candidates = list(list_ports.comports()) + if expected_usb is None: + match = next((item for item in candidates if item.device == port), None) + else: + expected_vid, expected_pid, expected_serial = expected_usb + matches = [ + item for item in candidates + if (expected_vid is None or item.vid == expected_vid) + and (expected_pid is None or item.pid == expected_pid) + and (not expected_serial or item.serial_number == expected_serial) + ] + match = next((item for item in matches if item.device == port), None) + if match is None and len(matches) == 1: + match = matches[0] + if match is not None: + output(f"Bootloader available on {match.device}.") + return str(match.device) + time.sleep(0.25) + raise FirmwareError("The radio did not reappear in bootloader mode.") + + +def flash_bundle( + bundle: FirmwareBundle, + port: str, + full_install: bool, + expected_usb: tuple[int | None, int | None, str | None] | None = None, + output: Callable[[str], None] = lambda _line: None, +) -> None: + if os.name == "nt" and not _PORT_RE.fullmatch(port): + raise FirmwareError("Select one explicit Windows COM port before flashing.") + if expected_usb is not None: + from serial.tools import list_ports + current = next((item for item in list_ports.comports() if item.device == port), None) + if current is None: + raise FirmwareError(f"The verified radio is no longer present on {port}.") + expected_vid, expected_pid, expected_serial = expected_usb + if expected_vid is not None and current.vid != expected_vid: + raise FirmwareError("The USB vendor changed after disconnect; refusing to flash.") + if expected_pid is not None and current.pid != expected_pid: + raise FirmwareError("The USB product changed after disconnect; refusing to flash.") + if expected_serial and current.serial_number != expected_serial: + raise FirmwareError("A different USB device now owns the selected COM port.") + validate_bundle(bundle) + chip = _CHIP_BY_PLATFORM.get(bundle.release.platform) + if chip is None: + raise FirmwareError( + f"OrcMesh cannot verify platform {bundle.release.platform}; refusing to flash." + ) + try: + import esptool + except ImportError as exc: + raise FirmwareError("esptool is not installed in this OrcMesh build.") from exc + + def run(args: list[str], active_port: str | None = None) -> None: + active_port = active_port or port + output("esptool " + " ".join(Path(arg).name if "firmware" in arg else arg for arg in args)) + try: + esptool.main([ + "--chip", chip, "--port", active_port, "--baud", "115200", *args, + ]) + except SystemExit as exc: + if exc.code not in (None, 0): + raise FirmwareError(f"esptool failed with exit code {exc.code}") from exc + except Exception as exc: + raise FirmwareError(str(exc)) from exc + + try: + run(["chip-id"]) + except FirmwareError: + if bundle.requires_dfu: + try: + port = _automatic_bootloader_port(port, expected_usb, output) + run(["chip-id"], port) + except FirmwareError as automatic_exc: + raise FirmwareError( + "Automatic bootloader entry failed. Hold BOOT, tap RESET, release BOOT, " + "then retry with the radio's COM port." + ) from automatic_exc + else: + raise + + if full_install: + run(["erase-flash"]) + run(["write-flash", "0x0", str(bundle.factory_image)]) + run(["write-flash", bundle.ota_offset, str(bundle.ota_image)]) + run(["write-flash", bundle.filesystem_offset, str(bundle.filesystem_image)]) + else: + run(["write-flash", "0x10000", str(bundle.update_image)]) + output("Firmware flash completed; waiting for the radio to reboot.") diff --git a/src/meshchat/ui/device/__init__.py b/src/meshchat/ui/device/__init__.py new file mode 100644 index 0000000..7c8d298 --- /dev/null +++ b/src/meshchat/ui/device/__init__.py @@ -0,0 +1 @@ +"""Connected-radio controls UI.""" diff --git a/src/meshchat/ui/device/device_page.py b/src/meshchat/ui/device/device_page.py new file mode 100644 index 0000000..8accd03 --- /dev/null +++ b/src/meshchat/ui/device/device_page.py @@ -0,0 +1,513 @@ +"""Connected Meshtastic radio configuration and maintenance page.""" +from __future__ import annotations + +from typing import Any + +from PySide6.QtCore import Signal +from PySide6.QtWidgets import ( + QCheckBox, + QComboBox, + QFormLayout, + QGroupBox, + QHBoxLayout, + QInputDialog, + QLabel, + QLineEdit, + QMessageBox, + QPushButton, + QProgressBar, + QScrollArea, + QSpinBox, + QStackedWidget, + QTabWidget, + QTextEdit, + QVBoxLayout, + QWidget, +) + + +class DevicePage(QWidget): + save_section_requested = Signal(str, object) + owner_requested = Signal(str, str) + channel_requested = Signal(object) + reboot_requested = Signal(int) + shutdown_requested = Signal(int) + reset_nodedb_requested = Signal() + factory_reset_requested = Signal(bool) + fixed_position_requested = Signal(float, float, int) + remove_fixed_position_requested = Signal() + refresh_requested = Signal() + firmware_discover_requested = Signal(str, bool) + firmware_prepare_requested = Signal(object, str, str) + firmware_flash_requested = Signal(object, bool, object) + + def __init__(self, parent=None): + super().__init__(parent) + self._snapshot = None + self._section_widgets: dict[str, dict[str, tuple[Any, Any]]] = {} + self._channel_by_index = {} + self._firmware_release = None + self._firmware_bundle = None + + root = QVBoxLayout(self) + root.setContentsMargins(16, 12, 16, 12) + root.setSpacing(10) + + header = QHBoxLayout() + title = QLabel("USB Device Controls") + title.setStyleSheet("font-size: 18px; font-weight: 700;") + header.addWidget(title) + header.addStretch() + self._refresh = QPushButton("Refresh") + self._refresh.clicked.connect(self.refresh_requested) + header.addWidget(self._refresh) + root.addLayout(header) + + self._summary = QLabel("Connect a Meshtastic radio over USB / Serial to manage it.") + self._summary.setWordWrap(True) + root.addWidget(self._summary) + + self._tabs = QTabWidget() + root.addWidget(self._tabs, 1) + self._build_overview_tab() + self._build_settings_tab() + self._build_channels_tab() + self._build_firmware_tab() + + self.set_connected(False) + + def _build_overview_tab(self) -> None: + page = QWidget() + layout = QVBoxLayout(page) + + identity = QGroupBox("Identity") + form = QFormLayout(identity) + self._long_name = QLineEdit() + self._short_name = QLineEdit() + self._short_name.setMaxLength(4) + form.addRow("Long name", self._long_name) + form.addRow("Short name", self._short_name) + save_owner = QPushButton("Save Identity") + save_owner.clicked.connect( + lambda: self.owner_requested.emit( + self._long_name.text().strip(), self._short_name.text().strip() + ) + ) + form.addRow("", save_owner) + layout.addWidget(identity) + + position = QGroupBox("Fixed Position") + pos_form = QFormLayout(position) + self._latitude = QLineEdit() + self._longitude = QLineEdit() + self._altitude = QSpinBox() + self._altitude.setRange(-1000, 100000) + pos_form.addRow("Latitude", self._latitude) + pos_form.addRow("Longitude", self._longitude) + pos_form.addRow("Altitude (m)", self._altitude) + pos_buttons = QHBoxLayout() + set_pos = QPushButton("Set Fixed Position") + set_pos.clicked.connect(self._set_fixed_position) + remove_pos = QPushButton("Use GPS / Remove Fixed") + remove_pos.clicked.connect(self.remove_fixed_position_requested) + pos_buttons.addWidget(set_pos) + pos_buttons.addWidget(remove_pos) + pos_form.addRow("", pos_buttons) + layout.addWidget(position) + + maintenance = QGroupBox("Maintenance") + buttons = QHBoxLayout(maintenance) + for label, handler in ( + ("Reboot", self._confirm_reboot), + ("Shutdown", self._confirm_shutdown), + ("Reset NodeDB", self._confirm_nodedb_reset), + ("Factory Reset", lambda _checked=False: self._confirm_factory_reset(full=False)), + ("Full Factory Reset", lambda _checked=False: self._confirm_factory_reset(full=True)), + ): + button = QPushButton(label) + if "Reset" in label: + button.setObjectName("dangerBtn") + button.clicked.connect(handler) + buttons.addWidget(button) + layout.addWidget(maintenance) + layout.addStretch() + self._tabs.addTab(page, "Overview") + + def _build_settings_tab(self) -> None: + page = QWidget() + layout = QVBoxLayout(page) + selector_row = QHBoxLayout() + selector_row.addWidget(QLabel("Section")) + self._section_selector = QComboBox() + self._section_selector.currentIndexChanged.connect(self._on_section_changed) + selector_row.addWidget(self._section_selector, 1) + self._save_section = QPushButton("Save Section") + self._save_section.clicked.connect(self._save_current_section) + selector_row.addWidget(self._save_section) + layout.addLayout(selector_row) + self._section_stack = QStackedWidget() + layout.addWidget(self._section_stack, 1) + self._tabs.addTab(page, "Configuration") + + def _build_channels_tab(self) -> None: + page = QWidget() + layout = QFormLayout(page) + self._channel_selector = QComboBox() + self._channel_selector.currentIndexChanged.connect(self._load_channel) + self._channel_name = QLineEdit() + self._channel_role = QComboBox() + from meshtastic.protobuf import channel_pb2 + for value in channel_pb2.Channel.Role.values(): + self._channel_role.addItem(channel_pb2.Channel.Role.Name(value), value) + self._channel_uplink = QCheckBox() + self._channel_downlink = QCheckBox() + self._position_precision = QSpinBox() + self._position_precision.setRange(0, 32) + self._channel_psk = QLineEdit() + self._channel_psk.setEchoMode(QLineEdit.EchoMode.Password) + self._channel_psk.setPlaceholderText("Unchanged; enter default, random, none, or base64 PSK") + save = QPushButton("Save Channel") + save.clicked.connect(self._save_channel) + layout.addRow("Channel", self._channel_selector) + layout.addRow("Name", self._channel_name) + layout.addRow("Role", self._channel_role) + layout.addRow("Uplink", self._channel_uplink) + layout.addRow("Downlink", self._channel_downlink) + layout.addRow("Position precision", self._position_precision) + layout.addRow("Replacement PSK", self._channel_psk) + layout.addRow("", save) + self._tabs.addTab(page, "Channels") + + def _build_firmware_tab(self) -> None: + page = QWidget() + layout = QVBoxLayout(page) + warning = QLabel( + "Firmware is downloaded only from official meshtastic/firmware releases. " + "OrcMesh verifies the release SHA-256, target metadata, hardware model, " + "and every extracted firmware image before enabling Flash." + ) + warning.setWordWrap(True) + layout.addWidget(warning) + row = QHBoxLayout() + self._release_channel = QComboBox() + self._release_channel.addItem("Stable", False) + self._release_channel.addItem("Include prerelease", True) + self._check_firmware = QPushButton("Check Official Release") + self._check_firmware.clicked.connect(self._discover_firmware) + self._download_firmware = QPushButton("Download & Verify") + self._download_firmware.setEnabled(False) + self._download_firmware.clicked.connect(self._prepare_firmware) + row.addWidget(self._release_channel) + row.addWidget(self._check_firmware) + row.addWidget(self._download_firmware) + row.addStretch() + layout.addLayout(row) + self._firmware_status = QLabel("No release checked.") + self._firmware_status.setWordWrap(True) + layout.addWidget(self._firmware_status) + self._firmware_progress = QProgressBar() + self._firmware_progress.setRange(0, 1000) + self._firmware_progress.setValue(0) + layout.addWidget(self._firmware_progress) + flash_row = QHBoxLayout() + self._flash_update = QPushButton("Flash Update (Preserve Settings)") + self._flash_update.setEnabled(False) + self._flash_update.clicked.connect(lambda: self._confirm_flash(False)) + self._flash_full = QPushButton("Full Erase & Install") + self._flash_full.setObjectName("dangerBtn") + self._flash_full.setEnabled(False) + self._flash_full.clicked.connect(lambda: self._confirm_flash(True)) + flash_row.addWidget(self._flash_update) + flash_row.addWidget(self._flash_full) + flash_row.addStretch() + layout.addLayout(flash_row) + self._firmware_log = QTextEdit() + self._firmware_log.setReadOnly(True) + self._firmware_log.setPlaceholderText("Firmware progress will appear here. Secrets are never logged.") + layout.addWidget(self._firmware_log, 1) + self._tabs.addTab(page, "Firmware") + + def set_connected(self, connected: bool) -> None: + self._tabs.setEnabled(connected) + self._refresh.setEnabled(connected) + if not connected: + self._snapshot = None + self._summary.setText("Connect a Meshtastic radio over USB / Serial to manage it.") + self._flash_update.setEnabled(False) + self._flash_full.setEnabled(False) + + def set_snapshot(self, snapshot) -> None: + self._snapshot = snapshot + self.set_connected(snapshot.serial_port is not None) + transport = snapshot.serial_port or "non-USB connection" + self._summary.setText( + f"{snapshot.long_name or snapshot.node_id or 'Radio'} Β· {snapshot.hw_model} Β· " + f"Firmware {snapshot.firmware_version} Β· {snapshot.pio_env} Β· {transport}" + ) + self._long_name.setText(snapshot.long_name) + self._short_name.setText(snapshot.short_name) + self._rebuild_sections(snapshot.sections) + self._rebuild_channels(snapshot.channels) + self._firmware_release = None + self._firmware_bundle = None + self._download_firmware.setEnabled(False) + self._flash_update.setEnabled(False) + self._flash_full.setEnabled(False) + + def show_operation(self, detail: str) -> None: + self._summary.setText(detail) + + def set_firmware_release(self, release) -> None: + self._firmware_release = release + channel = "prerelease" if release.prerelease else "stable" + self._firmware_status.setText( + f"Meshtastic {release.version} ({channel}) Β· {release.asset_name} Β· " + f"{release.asset_size / 1024 / 1024:.1f} MB" + ) + self._download_firmware.setEnabled(True) + + def set_firmware_bundle(self, bundle) -> None: + self._firmware_bundle = bundle + self._firmware_status.setText( + f"Verified Meshtastic {bundle.release.version} for {bundle.hw_model} " + f"({bundle.pio_env}). Ready to flash." + ) + self._flash_update.setEnabled(True) + self._flash_full.setEnabled(True) + + def set_firmware_progress(self, done: int, total: int) -> None: + self._firmware_progress.setValue(0 if total <= 0 else min(1000, int(done * 1000 / total))) + + def append_firmware_log(self, line: str) -> None: + self._firmware_log.append(line) + + def firmware_completed(self, operation: str, success: bool, detail: str) -> None: + self._firmware_status.setText(detail) + if operation == "discover": + self._check_firmware.setEnabled(True) + elif operation == "prepare" and not success: + self._download_firmware.setEnabled(self._firmware_release is not None) + if operation == "flash": + self._flash_update.setEnabled(success is False and self._firmware_bundle is not None) + self._flash_full.setEnabled(success is False and self._firmware_bundle is not None) + if not success: + self._firmware_log.append("ERROR: " + detail) + + def _discover_firmware(self) -> None: + if self._snapshot is None or not self._snapshot.serial_port: + return + self._check_firmware.setEnabled(False) + self._firmware_status.setText("Checking official Meshtastic releases…") + self.firmware_discover_requested.emit( + self._snapshot.pio_env, bool(self._release_channel.currentData()) + ) + + def _prepare_firmware(self) -> None: + if self._snapshot is None or self._firmware_release is None: + return + self._download_firmware.setEnabled(False) + self._firmware_status.setText("Downloading and verifying firmware…") + self.firmware_prepare_requested.emit( + self._firmware_release, self._snapshot.pio_env, self._snapshot.hw_model + ) + + def _confirm_flash(self, full: bool) -> None: + if self._snapshot is None or self._firmware_bundle is None: + return + if full: + phrase = f"ERASE {self._snapshot.pio_env}" + text, ok = QInputDialog.getText( + self, "Full Firmware Install", + "This erases firmware, settings, channels, keys, and the NodeDB. " + f"Type {phrase} to continue:", + ) + if not ok or text != phrase: + return + elif not self._yes( + "Flash Firmware Update", + f"Flash Meshtastic {self._firmware_bundle.release.version} to " + f"{self._snapshot.serial_port}? Settings should be preserved, but power or cable " + "loss can require a full recovery flash.", + ): + return + expected_usb = ( + self._snapshot.usb_vid, self._snapshot.usb_pid, self._snapshot.usb_serial + ) + self._flash_update.setEnabled(False) + self._flash_full.setEnabled(False) + self._firmware_log.clear() + self.firmware_flash_requested.emit(self._firmware_bundle, full, expected_usb) + + def _rebuild_sections(self, sections) -> None: + while self._section_stack.count(): + widget = self._section_stack.widget(0) + if widget is None: + break + self._section_stack.removeWidget(widget) + widget.deleteLater() + self._section_selector.clear() + self._section_widgets.clear() + for section in sections: + container = QWidget() + form = QFormLayout(container) + widgets = {} + for field in section.fields: + widget = self._widget_for_field(field) + form.addRow(field.label, widget) + widgets[field.name] = (field, widget) + scroll = QScrollArea() + scroll.setWidgetResizable(True) + scroll.setWidget(container) + self._section_stack.addWidget(scroll) + self._section_selector.addItem(section.label, section.name) + self._section_widgets[section.name] = widgets + + @staticmethod + def _widget_for_field(field): + if field.kind == "bool" and not field.repeated: + widget = QCheckBox() + widget.setChecked(bool(field.value)) + elif field.kind == "enum" and not field.repeated: + widget = QComboBox() + for choice in field.choices: + widget.addItem(choice.label, choice.value) + index = widget.findData(int(field.value)) + if index < 0: + widget.addItem(f"Unknown ({int(field.value)})", int(field.value)) + index = widget.count() - 1 + widget.setCurrentIndex(index) + else: + widget = QLineEdit() + if field.repeated: + widget.setText(", ".join(str(value) for value in field.value)) + elif not field.write_only and not field.read_only: + widget.setText(str(field.value)) + if field.write_only: + widget.setEchoMode(QLineEdit.EchoMode.Password) + widget.setPlaceholderText("Stored on radio β€” enter to replace") + widget.setEnabled(not field.read_only) + if field.read_only: + widget.setToolTip("Cryptographic key material is intentionally not displayed or edited.") + return widget + + def _on_section_changed(self, index: int) -> None: + if index >= 0: + self._section_stack.setCurrentIndex(index) + + def _save_current_section(self) -> None: + section = self._section_selector.currentData() + if not section: + return + try: + changes = { + name: self._read_widget(field, widget) + for name, (field, widget) in self._section_widgets[section].items() + if not field.read_only + } + except ValueError as exc: + QMessageBox.warning(self, "Invalid Setting", str(exc)) + return + self.save_section_requested.emit(section, changes) + + @staticmethod + def _read_widget(field, widget): + if field.kind == "bool" and not field.repeated: + return widget.isChecked() + if field.kind == "enum" and not field.repeated: + return widget.currentData() + text = widget.text().strip() + if field.write_only and not text: + return "" + if field.repeated: + if not text: + return [] + parts = [part.strip() for part in text.split(",")] + if field.kind == "int": + return [int(part, 0) for part in parts] + if field.kind == "float": + return [float(part) for part in parts] + return parts + if field.kind == "int": + return int(text, 0) + if field.kind == "float": + return float(text) + return text + + def _rebuild_channels(self, channels) -> None: + self._channel_by_index = {channel.index: channel for channel in channels} + self._channel_selector.clear() + for channel in channels: + self._channel_selector.addItem( + f"{channel.index}: {channel.name or channel.role_name}", channel.index + ) + self._load_channel(0) + + def _load_channel(self, _index: int) -> None: + channel = self._channel_by_index.get(self._channel_selector.currentData()) + if channel is None: + return + self._channel_name.setText(channel.name) + role_index = self._channel_role.findData(channel.role) + if role_index < 0: + self._channel_role.addItem(f"UNKNOWN ({channel.role})", channel.role) + role_index = self._channel_role.count() - 1 + self._channel_role.setCurrentIndex(role_index) + self._channel_uplink.setChecked(channel.uplink_enabled) + self._channel_downlink.setChecked(channel.downlink_enabled) + self._position_precision.setValue(channel.position_precision) + self._channel_psk.clear() + + def _save_channel(self) -> None: + index = self._channel_selector.currentData() + if index is None: + return + self.channel_requested.emit({ + "index": index, + "name": self._channel_name.text(), + "role": self._channel_role.currentData(), + "uplink_enabled": self._channel_uplink.isChecked(), + "downlink_enabled": self._channel_downlink.isChecked(), + "position_precision": self._position_precision.value(), + "psk": self._channel_psk.text(), + }) + + def _set_fixed_position(self) -> None: + try: + latitude = float(self._latitude.text()) + longitude = float(self._longitude.text()) + except ValueError: + QMessageBox.warning(self, "Invalid Position", "Latitude and longitude must be numbers.") + return + if not -90 <= latitude <= 90 or not -180 <= longitude <= 180: + QMessageBox.warning(self, "Invalid Position", "Coordinates are outside valid ranges.") + return + self.fixed_position_requested.emit(latitude, longitude, self._altitude.value()) + + def _confirm_reboot(self) -> None: + if self._yes("Reboot Radio", "Reboot the connected radio now?"): + self.reboot_requested.emit(2) + + def _confirm_shutdown(self) -> None: + if self._yes("Shutdown Radio", "Shut down the connected radio?"): + self.shutdown_requested.emit(2) + + def _confirm_nodedb_reset(self) -> None: + if self._yes("Reset Node Database", "Delete the radio's learned NodeDB? It will repopulate over time."): + self.reset_nodedb_requested.emit() + + def _confirm_factory_reset(self, full: bool = False) -> None: + word = "ERASE" if full else "RESET" + text, ok = QInputDialog.getText( + self, + "Full Factory Reset" if full else "Factory Reset", + f"This removes radio configuration. Type {word} to continue:", + ) + if ok and text == word: + self.factory_reset_requested.emit(full) + + def _yes(self, title: str, text: str) -> bool: + return QMessageBox.question( + self, title, text, + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.Cancel, + QMessageBox.StandardButton.Cancel, + ) == QMessageBox.StandardButton.Yes diff --git a/src/meshchat/ui/main_window.py b/src/meshchat/ui/main_window.py index b9aa061..80706f1 100644 --- a/src/meshchat/ui/main_window.py +++ b/src/meshchat/ui/main_window.py @@ -101,12 +101,17 @@ def __init__(self): # ── Services ────────────────────────────────────────────────── self._controller = MeshtasticController(self) + from meshchat.controllers.firmware_controller import FirmwareController + self._firmware_controller = FirmwareController(self) self._session = NetworkSession.new() self._store = MonitorStore() self._ingestor = PacketIngestor(self._session, self._store) self._supervisor = ConnectionSupervisor(self._controller, self._store, parent=self) self._export_thread: QThread | None = None self._export_worker: _PacketExportWorker | None = None + self._device_snapshot = None + self._pending_flash = None + self._flash_port: str | None = None # ── Central layout ──────────────────────────────────────────── central = QWidget() @@ -127,9 +132,13 @@ def __init__(self): self._nav_monitor = _NavButton("πŸ“‘", "Monitor") self._nav_nodes = _NavButton("πŸ”΅", "Nodes") self._nav_spectrum = _NavButton("πŸ“Ά", "Spectrum") + self._nav_device = _NavButton("βš™", "Device") self._nav_chat.setChecked(True) - for btn in (self._nav_chat, self._nav_monitor, self._nav_nodes, self._nav_spectrum): + for btn in ( + self._nav_chat, self._nav_monitor, self._nav_nodes, + self._nav_spectrum, self._nav_device, + ): btn.setAutoExclusive(True) nav_layout.addWidget(btn) @@ -201,12 +210,37 @@ def __init__(self): from meshchat.ui.spectrum.spectrum_page import SpectrumPage self._spectrum_page = SpectrumPage() + from meshchat.ui.device.device_page import DevicePage + self._device_page = DevicePage() + self._device_page.refresh_requested.connect(self._controller.refresh_device_controls) + self._device_page.save_section_requested.connect(self._controller.apply_device_section) + self._device_page.owner_requested.connect(self._controller.set_owner) + self._device_page.channel_requested.connect(self._controller.update_channel) + self._device_page.reboot_requested.connect(self._controller.reboot_device) + self._device_page.shutdown_requested.connect(self._controller.shutdown_device) + self._device_page.reset_nodedb_requested.connect(self._controller.reset_nodedb) + self._device_page.factory_reset_requested.connect(self._controller.factory_reset) + self._device_page.fixed_position_requested.connect(self._controller.set_fixed_position) + self._device_page.remove_fixed_position_requested.connect( + self._controller.remove_fixed_position + ) + self._device_page.firmware_discover_requested.connect( + self._firmware_controller.discover + ) + self._device_page.firmware_prepare_requested.connect( + self._firmware_controller.prepare + ) + self._device_page.firmware_flash_requested.connect( + self._on_firmware_flash_requested + ) + # Stacked widget self._stack = QStackedWidget() self._stack.addWidget(chat_container) # index 0 self._stack.addWidget(self._monitor_page) # index 1 self._stack.addWidget(self._nodes_page) # index 2 self._stack.addWidget(self._spectrum_page) # index 3 + self._stack.addWidget(self._device_page) # index 4 content_layout.addWidget(self._stack, 1) root.addWidget(content, 1) @@ -216,6 +250,7 @@ def __init__(self): self._nav_monitor.clicked.connect(lambda: self._stack.setCurrentIndex(1)) self._nav_nodes.clicked.connect(lambda: self._stack.setCurrentIndex(2)) self._nav_spectrum.clicked.connect(lambda: self._stack.setCurrentIndex(3)) + self._nav_device.clicked.connect(lambda: self._stack.setCurrentIndex(4)) # ── Controller signals ──────────────────────────────────────── ctrl = self._controller @@ -234,6 +269,15 @@ def __init__(self): ctrl.node_action_completed.connect(self._on_node_action_completed) ctrl.error_occurred.connect(self._on_error) ctrl.raw_packet.connect(self._ingestor.ingest_raw) + ctrl.device_controls_updated.connect(self._on_device_controls_updated) + ctrl.device_operation_completed.connect(self._on_device_operation) + + firmware = self._firmware_controller + firmware.release_found.connect(self._device_page.set_firmware_release) + firmware.progress.connect(self._device_page.set_firmware_progress) + firmware.bundle_ready.connect(self._device_page.set_firmware_bundle) + firmware.log.connect(self._device_page.append_firmware_log) + firmware.completed.connect(self._on_firmware_completed) # Ingestor signals self._ingestor.packet_ingested.connect(self._monitor_page.on_packet_ingested) @@ -543,7 +587,45 @@ def _on_disconnected(self, reason: str) -> None: self._monitor_page.set_local_node(None) self._monitor_page.on_local_telemetry(None, None, None) self._local_node_num = None + self._device_page.set_connected(False) + self._device_snapshot = None self._status_bar.showMessage(f"Disconnected β€” {reason}") + if self._pending_flash is not None: + bundle, full_install, expected_usb = self._pending_flash + self._pending_flash = None + QTimer.singleShot( + 500, + lambda: self._firmware_controller.flash( + bundle, self._flash_port or "", full_install, expected_usb + ), + ) + + def _on_device_operation(self, _operation: str, detail: str) -> None: + self._device_page.show_operation(detail) + self._status_bar.showMessage(detail, 6000) + + def _on_device_controls_updated(self, snapshot) -> None: + self._device_snapshot = snapshot + self._device_page.set_snapshot(snapshot) + + def _on_firmware_flash_requested(self, bundle, full_install: bool, expected_usb) -> None: + snapshot = self._device_snapshot + if not self._is_connected or snapshot is None or not snapshot.serial_port: + QMessageBox.warning(self, "USB Radio Required", "Reconnect the radio over USB before flashing.") + return + self._flash_port = snapshot.serial_port + self._pending_flash = (bundle, full_install, expected_usb) + self._supervisor.cancel() + self._status_bar.showMessage("Releasing the USB port for firmware flashing…") + self._controller.disconnect() + + def _on_firmware_completed(self, operation: str, success: bool, detail: str) -> None: + self._device_page.firmware_completed(operation, success, detail) + self._status_bar.showMessage(detail, 10000) + if operation == "flash" and self._flash_port: + port = self._flash_port + self._flash_port = None + QTimer.singleShot(7000 if success else 1000, lambda: self._controller.connect_serial(port)) def _on_telemetry(self, sample) -> None: # The dashboard header shows the connected radio's own environment @@ -763,6 +845,7 @@ def closeEvent(self, event) -> None: settings = QSettings() settings.setValue(f"{_SETTINGS_KEY}/geometry", self.saveGeometry()) self._spectrum_page.shutdown() + self._firmware_controller.shutdown() self._controller.shutdown() self._store.shutdown() if self._export_thread is not None: diff --git a/tests/test_device_config.py b/tests/test_device_config.py new file mode 100644 index 0000000..49de377 --- /dev/null +++ b/tests/test_device_config.py @@ -0,0 +1,84 @@ +from types import SimpleNamespace + +import pytest + +from meshtastic.protobuf import channel_pb2, localonly_pb2 + +from meshchat.services.device_config import apply_section, build_snapshot + + +class _Node: + def __init__(self): + self.localConfig = localonly_pb2.LocalConfig() + self.moduleConfig = localonly_pb2.LocalModuleConfig() + self.channels = [channel_pb2.Channel(index=0, role=channel_pb2.Channel.Role.PRIMARY)] + self.writes = [] + + def writeConfig(self, name): + self.writes.append(name) + + +def _interface(): + node = _Node() + node.localConfig.network.wifi_ssid = "mesh-lan" + node.localConfig.network.wifi_psk = "never expose this" + node.localConfig.lora.ignore_incoming.extend([1, 2]) + node.moduleConfig.mqtt.password = "also secret" + return SimpleNamespace( + localNode=node, + myInfo=SimpleNamespace(my_node_num=123, pio_env="tbeam-s3-core"), + metadata=SimpleNamespace( + firmware_version="2.7.26.54e0d8d", + can_shutdown=True, + has_wifi=True, + has_bluetooth=True, + ), + nodesByNum={123: {"user": { + "id": "!0000007b", "longName": "Test", "shortName": "TST", + "hwModel": "LILYGO_TBEAM_S3_CORE", + }}}, + ) + + +def test_snapshot_redacts_credentials_and_preserves_editable_values(): + snapshot = build_snapshot(_interface(), "COM8") + assert snapshot.pio_env == "tbeam-s3-core" + assert snapshot.serial_port == "COM8" + network = next(section for section in snapshot.sections if section.name == "network") + values = {field.name: field for field in network.fields} + assert values["wifi_ssid"].value == "mesh-lan" + assert values["wifi_psk"].value == "" + assert values["wifi_psk"].write_only + assert "never expose this" not in repr(snapshot) + assert "also secret" not in repr(snapshot) + + +def test_apply_section_coerces_values_and_writes_only_requested_section(): + node = _Node() + apply_section(node, "lora", { + "hop_limit": "5", + "tx_enabled": False, + "ignore_incoming": [7, "9"], + }) + assert node.localConfig.lora.hop_limit == 5 + assert node.localConfig.lora.tx_enabled is False + assert list(node.localConfig.lora.ignore_incoming) == [7, 9] + assert node.writes == ["lora"] + + +def test_apply_section_does_not_clear_write_only_field_when_left_blank(): + node = _Node() + node.localConfig.network.wifi_psk = "keep-me" + apply_section(node, "network", {"wifi_psk": ""}) + assert node.localConfig.network.wifi_psk == "keep-me" + + +def test_cryptographic_key_fields_are_not_editable(): + node = _Node() + with pytest.raises(ValueError, match="not editable"): + apply_section(node, "security", {"private_key": b"bad"}) + + +def test_apply_section_rejects_names_outside_allowlist(): + with pytest.raises(ValueError, match="Unknown configuration section"): + apply_section(_Node(), "version", {}) diff --git a/tests/test_device_page.py b/tests/test_device_page.py new file mode 100644 index 0000000..0d3737d --- /dev/null +++ b/tests/test_device_page.py @@ -0,0 +1,71 @@ +from types import SimpleNamespace + +from PySide6.QtWidgets import QInputDialog, QPushButton +from PySide6.QtWidgets import QMessageBox + +from meshchat.models.device_control import ConfigChoice, ConfigField, DeviceControlSnapshot +from meshchat.ui.device.device_page import DevicePage +from meshchat.ui.main_window import MainWindow + + +def _snapshot() -> DeviceControlSnapshot: + return DeviceControlSnapshot( + node_id="!12345678", + long_name="OrcMesh Radio", + short_name="ORC", + hw_model="LILYGO_TBEAM_S3_CORE", + firmware_version="2.7.10", + pio_env="tbeam-s3-core", + serial_port="COM8", + usb_vid=0x303A, + usb_pid=0x1001, + usb_serial=None, + can_shutdown=True, + has_wifi=True, + has_bluetooth=True, + ) + + +def test_device_page_exposes_all_control_tabs(): + page = DevicePage() + assert [page._tabs.tabText(i) for i in range(page._tabs.count())] == [ + "Overview", "Configuration", "Channels", "Firmware", + ] + + +def test_device_page_enables_usb_controls_for_serial_snapshot(): + page = DevicePage() + assert not page._tabs.isEnabled() + page.set_snapshot(_snapshot()) + assert page._tabs.isEnabled() + assert "COM8" in page._summary.text() + assert "tbeam-s3-core" in page._summary.text() + + +def test_unknown_enum_value_is_preserved(): + field = ConfigField( + name="mode", label="Mode", kind="enum", value=99, + choices=(ConfigChoice("KNOWN", 1),), + ) + widget = DevicePage._widget_for_field(field) + assert widget.currentData() == 99 + assert DevicePage._read_widget(field, widget) == 99 + + +def test_factory_reset_button_emits_non_full_reset(monkeypatch): + page = DevicePage() + page.set_snapshot(_snapshot()) + monkeypatch.setattr(QInputDialog, "getText", lambda *_args: ("RESET", True)) + requested = [] + page.factory_reset_requested.connect(requested.append) + button = next(button for button in page.findChildren(QPushButton) if button.text() == "Factory Reset") + button.click() + assert requested == [False] + + +def test_flash_handoff_rejects_disconnected_snapshot(monkeypatch): + warnings = [] + monkeypatch.setattr(QMessageBox, "warning", lambda *_args: warnings.append(True)) + window = SimpleNamespace(_is_connected=False, _device_snapshot=_snapshot()) + MainWindow._on_firmware_flash_requested(window, object(), False, None) + assert warnings == [True] diff --git a/tests/test_firmware_service.py b/tests/test_firmware_service.py new file mode 100644 index 0000000..df1bfc0 --- /dev/null +++ b/tests/test_firmware_service.py @@ -0,0 +1,319 @@ +import hashlib +import json +import sys +import zipfile +from dataclasses import replace +from importlib.metadata import version +from types import SimpleNamespace + +import pytest + +from meshchat.services import firmware +from meshchat.controllers.firmware_controller import FirmwareController + + +def test_discovery_uses_manifest_target_and_official_digest(monkeypatch): + release = { + "tag_name": "v2.7.26.abc", + "draft": False, + "prerelease": False, + "published_at": "2026-01-01T00:00:00Z", + "assets": [ + {"name": "firmware-2.7.26.abc.json", "browser_download_url": "manifest"}, + { + "name": "firmware-esp32s3-2.7.26.abc.zip", + "browser_download_url": "bundle", + "size": 123, + "digest": "sha256:" + "a" * 64, + }, + ], + } + monkeypatch.setattr( + firmware, "_json", + lambda url: [release] if url == firmware._RELEASES_API else { + "targets": [{"board": "tbeam-s3-core", "platform": "esp32s3"}] + }, + ) + found = firmware.discover_release("tbeam-s3-core") + assert found.platform == "esp32s3" + assert found.asset_sha256 == "a" * 64 + + +def test_discovery_rejects_target_not_in_manifest(monkeypatch): + monkeypatch.setattr(firmware, "_json", lambda _url: [{ + "tag_name": "v1", "draft": False, "prerelease": False, + "assets": [{"name": "firmware-1.json", "browser_download_url": "manifest"}], + }] if _url == firmware._RELEASES_API else {"targets": []}) + with pytest.raises(firmware.FirmwareError, match="does not support"): + firmware.discover_release("tbeam-s3-core") + + +def test_discovery_rejects_unsupported_platform(monkeypatch): + release = { + "tag_name": "v1", "draft": False, "prerelease": False, + "assets": [{"name": "firmware-1.json", "browser_download_url": "manifest"}], + } + monkeypatch.setattr( + firmware, + "_json", + lambda url: [release] if url == firmware._RELEASES_API else { + "targets": [{"board": "other", "platform": "esp32"}] + }, + ) + with pytest.raises(firmware.FirmwareError, match="does not support platform"): + firmware.discover_release("other") + + +@pytest.mark.parametrize("url", ["http://github.com/file", "https://example.com/file"]) +def test_firmware_urls_are_restricted_to_github_https(url): + with pytest.raises(firmware.FirmwareError, match="official GitHub HTTPS"): + firmware._validate_url(url) + + +def _release(tmp_path): + return firmware.FirmwareRelease( + tag="v1", version="1", prerelease=False, published_at="", + asset_name="bundle.zip", asset_url="", asset_size=1, + asset_sha256="a" * 64, platform="esp32s3", + ) + + +def test_prepare_bundle_checks_target_hardware_and_member_hashes(tmp_path, monkeypatch): + release = _release(tmp_path) + names = { + "firmware-tbeam-s3-core-1.bin": b"update", + "firmware-tbeam-s3-core-1.factory.bin": b"factory", + "littlefs-tbeam-s3-core-1.bin": b"filesystem", + "mt-esp32s3-ota.bin": b"ota", + } + metadata = { + "platformioTarget": "tbeam-s3-core", + "hwModelSlug": "LILYGO_TBEAM_S3_CORE", + "activelySupported": True, + "requiresDfu": True, + "files": [ + {"name": name, "md5": hashlib.md5(data).hexdigest()} # noqa: S324 - release format + for name, data in names.items() + ], + "part": [ + {"subtype": "ota_1", "offset": "0x340000"}, + {"subtype": "spiffs", "offset": "0x670000"}, + ], + } + archive = tmp_path / "bundle.zip" + with zipfile.ZipFile(archive, "w") as output: + for name, data in names.items(): + output.writestr(name, data) + output.writestr("firmware-tbeam-s3-core-1.mt.json", json.dumps(metadata)) + monkeypatch.setattr(firmware, "_download", lambda _release, _progress: archive) + monkeypatch.setattr(firmware, "_cache_root", lambda: tmp_path / "cache") + bundle = firmware.prepare_bundle(release, "tbeam-s3-core", "LILYGO_TBEAM_S3_CORE") + assert bundle.requires_dfu + assert bundle.ota_offset == "0x340000" + firmware.validate_bundle(bundle) + + +def test_prepare_bundle_rejects_missing_image_hash(tmp_path, monkeypatch): + release = _release(tmp_path) + names = { + "firmware-tbeam-s3-core-1.bin": b"update", + "firmware-tbeam-s3-core-1.factory.bin": b"factory", + "littlefs-tbeam-s3-core-1.bin": b"filesystem", + "mt-esp32s3-ota.bin": b"ota", + } + metadata = { + "platformioTarget": "tbeam-s3-core", + "hwModelSlug": "LILYGO_TBEAM_S3_CORE", + "files": [], + "part": [ + {"subtype": "ota_1", "offset": "0x340000"}, + {"subtype": "spiffs", "offset": "0x670000"}, + ], + } + archive = tmp_path / "bundle.zip" + with zipfile.ZipFile(archive, "w") as output: + for name, data in names.items(): + output.writestr(name, data) + output.writestr("firmware-tbeam-s3-core-1.mt.json", json.dumps(metadata)) + monkeypatch.setattr(firmware, "_download", lambda _release, _progress: archive) + monkeypatch.setattr(firmware, "_cache_root", lambda: tmp_path / "cache") + with pytest.raises(firmware.FirmwareError, match="missing a valid image hash"): + firmware.prepare_bundle(release, "tbeam-s3-core", "LILYGO_TBEAM_S3_CORE") + + +def test_flash_update_uses_only_verified_update_offset(tmp_path, monkeypatch): + files = {} + for name in ("update.bin", "factory.bin", "ota.bin", "fs.bin"): + path = tmp_path / name + path.write_bytes(name.encode()) + files[name] = path + calls = [] + def esptool_main(args): + calls.append(args) + + monkeypatch.setitem(sys.modules, "esptool", SimpleNamespace(main=esptool_main)) + bundle = firmware.FirmwareBundle( + release=_release(tmp_path), root=tmp_path, pio_env="target", hw_model="model", + requires_dfu=False, update_image=files["update.bin"], factory_image=files["factory.bin"], + ota_image=files["ota.bin"], filesystem_image=files["fs.bin"], + ota_offset="0x340000", filesystem_offset="0x670000", + file_md5={name: hashlib.md5(path.read_bytes()).hexdigest() for name, path in files.items()}, # noqa: S324 + ) + firmware.flash_bundle(bundle, "COM8", False) + prefix = ["--chip", "esp32s3", "--port", "COM8", "--baud", "115200"] + assert calls == [ + [*prefix, "chip-id"], + [*prefix, "write-flash", "0x10000", str(files["update.bin"])], + ] + + +def test_flash_refuses_unsupported_platform(tmp_path, monkeypatch): + files = {} + for name in ("update.bin", "factory.bin", "ota.bin", "fs.bin"): + path = tmp_path / name + path.write_bytes(name.encode()) + files[name] = path + bundle = firmware.FirmwareBundle( + release=replace(_release(tmp_path), platform="esp32"), + root=tmp_path, pio_env="target", hw_model="model", + requires_dfu=False, update_image=files["update.bin"], factory_image=files["factory.bin"], + ota_image=files["ota.bin"], filesystem_image=files["fs.bin"], + ota_offset="0x340000", filesystem_offset="0x670000", + file_md5={name: hashlib.md5(path.read_bytes()).hexdigest() for name, path in files.items()}, # noqa: S324 + ) + with pytest.raises(firmware.FirmwareError, match="cannot verify platform"): + firmware.flash_bundle(bundle, "COM8", False) + + +def test_full_install_erases_then_writes_verified_partition_offsets(tmp_path, monkeypatch): + files = {} + for name in ("update.bin", "factory.bin", "ota.bin", "fs.bin"): + path = tmp_path / name + path.write_bytes(name.encode()) + files[name] = path + calls = [] + + def esptool_main(args): + calls.append(args) + + monkeypatch.setitem(sys.modules, "esptool", SimpleNamespace(main=esptool_main)) + bundle = firmware.FirmwareBundle( + release=_release(tmp_path), root=tmp_path, pio_env="target", hw_model="model", + requires_dfu=True, update_image=files["update.bin"], factory_image=files["factory.bin"], + ota_image=files["ota.bin"], filesystem_image=files["fs.bin"], + ota_offset="0x340000", filesystem_offset="0x670000", + file_md5={name: hashlib.md5(path.read_bytes()).hexdigest() for name, path in files.items()}, # noqa: S324 + ) + firmware.flash_bundle(bundle, "COM8", True) + prefix = ["--chip", "esp32s3", "--port", "COM8", "--baud", "115200"] + assert calls == [ + [*prefix, "chip-id"], + [*prefix, "erase-flash"], + [*prefix, "write-flash", "0x0", str(files["factory.bin"])], + [*prefix, "write-flash", "0x340000", str(files["ota.bin"])], + [*prefix, "write-flash", "0x670000", str(files["fs.bin"])], + ] + + +def test_dfu_failure_returns_recovery_instructions(tmp_path, monkeypatch): + files = {} + for name in ("update.bin", "factory.bin", "ota.bin", "fs.bin"): + path = tmp_path / name + path.write_bytes(name.encode()) + files[name] = path + monkeypatch.setitem( + sys.modules, + "esptool", + SimpleNamespace(main=lambda _args: (_ for _ in ()).throw(RuntimeError("no sync"))), + ) + monkeypatch.setattr( + firmware, + "_automatic_bootloader_port", + lambda *_args: (_ for _ in ()).throw(firmware.FirmwareError("no DFU port")), + ) + bundle = firmware.FirmwareBundle( + release=_release(tmp_path), root=tmp_path, pio_env="target", hw_model="model", + requires_dfu=True, update_image=files["update.bin"], factory_image=files["factory.bin"], + ota_image=files["ota.bin"], filesystem_image=files["fs.bin"], + ota_offset="0x340000", filesystem_offset="0x670000", + file_md5={name: hashlib.md5(path.read_bytes()).hexdigest() for name, path in files.items()}, # noqa: S324 + ) + with pytest.raises(firmware.FirmwareError, match="Hold BOOT, tap RESET"): + firmware.flash_bundle(bundle, "COM8", False) + + +def test_dfu_failure_retries_on_automatic_bootloader_port(tmp_path, monkeypatch): + files = {} + for name in ("update.bin", "factory.bin", "ota.bin", "fs.bin"): + path = tmp_path / name + path.write_bytes(name.encode()) + files[name] = path + calls = [] + + def esptool_main(args): + calls.append(args) + if args[-1] == "chip-id" and "COM8" in args: + raise RuntimeError("no sync") + + monkeypatch.setitem(sys.modules, "esptool", SimpleNamespace(main=esptool_main)) + monkeypatch.setattr(firmware, "_automatic_bootloader_port", lambda *_args: "COM9") + bundle = firmware.FirmwareBundle( + release=_release(tmp_path), root=tmp_path, pio_env="target", hw_model="model", + requires_dfu=True, update_image=files["update.bin"], factory_image=files["factory.bin"], + ota_image=files["ota.bin"], filesystem_image=files["fs.bin"], + ota_offset="0x340000", filesystem_offset="0x670000", + file_md5={name: hashlib.md5(path.read_bytes()).hexdigest() for name, path in files.items()}, # noqa: S324 + ) + + firmware.flash_bundle(bundle, "COM8", False) + + assert [call[3] for call in calls] == ["COM8", "COM9", "COM9"] + + +def test_automatic_bootloader_uses_1200_baud_and_follows_new_port(monkeypatch): + opened = [] + + class FakeSerial: + def __init__(self, **kwargs): + opened.append(kwargs) + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + import serial + from serial.tools import list_ports + + monkeypatch.setattr(serial, "Serial", FakeSerial) + monkeypatch.setattr(firmware.time, "sleep", lambda _seconds: None) + monkeypatch.setattr( + list_ports, + "comports", + lambda: [SimpleNamespace(device="COM9", vid=0x303A, pid=0x1001, serial_number="radio")], + ) + + port = firmware._automatic_bootloader_port( + "COM8", (0x303A, 0x1001, "radio"), lambda _line: None + ) + + assert port == "COM9" + assert opened == [{"port": "COM8", "baudrate": 1200, "timeout": 1}] + + +def test_installed_esptool_accepts_commands_used_by_flasher(): + esptool = pytest.importorskip("esptool") + assert int(version("esptool").split(".", 1)[0]) >= 5 + for command in ("chip-id", "erase-flash", "write-flash"): + esptool.main([command, "--help"]) + + +def test_firmware_shutdown_waits_for_active_worker(): + waits = [] + thread = SimpleNamespace( + quit=lambda: None, + wait=lambda timeout=None: waits.append(timeout) or timeout is None, + ) + FirmwareController.shutdown(SimpleNamespace(_thread=thread)) + assert waits == [5000, None]