From d242a82d1d21fe830802b6f5fbae44c96b900221 Mon Sep 17 00:00:00 2001 From: pdxlocations Date: Sat, 1 Aug 2026 00:46:04 -0700 Subject: [PATCH] Add curses dashboard and XIAO bridge examples; enhance stream parser for UTF-8 logging --- README.md | 462 +++++++++++++++--- examples/README.md | 29 ++ examples/curses_dashboard.py | 386 +++++++++++++++ examples/xiao_dashboard_bridge.py | 162 ++++++ micromesh/__pycache__/stream.cpython-311.pyc | Bin 4116 -> 4392 bytes micromesh/stream.py | 8 +- .../test_micromesh.cpython-311.pyc | Bin 12971 -> 13498 bytes tests/test_micromesh.py | 6 + 8 files changed, 988 insertions(+), 65 deletions(-) create mode 100644 examples/curses_dashboard.py create mode 100644 examples/xiao_dashboard_bridge.py diff --git a/README.md b/README.md index a396d32..2b0dffa 100644 --- a/README.md +++ b/README.md @@ -1,128 +1,462 @@ -# MicroMesh +# MicroMesh πŸ“» -MicroMesh is a small, dependency-free Meshtastic client for MicroPython. It talks to a -Meshtastic device over its serial API and implements the protobuf wire format directly, -so `google.protobuf`, threads, and `pyserial` are not required. +**A tiny, dependency-free Meshtastic client for MicroPython.** -The initial release supports: +MicroMesh lets a small MicroPython board talk to a Meshtastic radio over UART. Use it +to receive messages, send text and position packets, inspect the node list, or power a +full-screen terminal dashboard. -- serial `ToRadio` / `FromRadio` framing and resynchronization; -- the configuration handshake and a small in-memory node list; -- text, arbitrary data, and position packets; -- routing acknowledgements, node/user data, waypoints, and device metrics; -- familiar generated-protobuf methods: `SerializeToString`, `ParseFromString`, - `CopyFrom`, `HasField`, `ClearField`, and `WhichOneof`; -- preservation of unknown protobuf fields for forward compatibility. +No protobuf compiler runs on the board. No threads. No `google.protobuf`. Just a UART, +three wires, and regular Python. -This is intentionally not a complete replacement for the desktop -[`meshtastic`](https://github.com/meshtastic/python) package. Configuration and admin -messages are currently returned as raw encoded bytes. +> [!NOTE] +> MicroMesh is young and intentionally small. It supports the most useful Meshtastic +> client operations, but it is not yet a complete replacement for the official desktop +> [`meshtastic`](https://github.com/meshtastic/python) package. -## Install +## What can it do? -From a checkout of this repository, install with a current `mpremote`: +- Connect to a Meshtastic radio using its framed serial API +- Download and track the radio's node database +- Receive and send text messages +- Send binary data and positions +- Read user, position, signal, battery, and hop information +- Preserve unknown protobuf fields for forward compatibility +- Recover from serial noise and skip newer messages it cannot decode +- Run a curses dashboard with live stats, nodes, and messages +- Work on MicroPython without third-party runtime dependencies -```sh -mpremote mip install package.json +## Pick your setup + +| I want to… | Use this | +| --- | --- | +| Run a simple listener on a XIAO RP2040 | [XIAO quick start](#xiao-rp2040-quick-start) | +| View nodes and messages through my XIAO | [Curses dashboard](#curses-dashboard-through-the-xiao) | +| Connect my computer directly to the Meshtastic radio | [Direct desktop dashboard](#direct-to-radio-dashboard) | +| Use a Pico, Pico W, or ESP32 | [Other boards](#other-boards) | +| Write my own program | [Python API](#python-api) | + +## How the XIAO setup works + +The XIAO is the computer running MicroMesh. The Meshtastic device is the LoRa radio. + +```text + Mac / PC XIAO RP2040 Meshtastic radio +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β” USB/REPL β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” UART β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ terminal│◀─────────────▢│ MicroPython + │◀────────▢│ Meshtastic │◀──▢ LoRa +β”‚ or UI β”‚ β”‚ MicroMesh β”‚ 3 wires β”‚ firmware β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` -Alternatively, copy the `micromesh` directory to `/lib` on the board. After this -repository is published on GitHub it can also be installed as -`mpremote mip install github:ORG/micromesh`. +The XIAO does not contain a LoRa radio. It controls a separate Meshtastic device over +UART. -It can also be installed on CPython for development: +## Before you begin -```sh -python -m pip install . +You need: + +- A Seeed Studio XIAO RP2040 with MicroPython installed +- A separate Meshtastic-compatible radio running Meshtastic firmware +- Three jumper wires for TX, RX, and GND +- A data-capable USB cable for the XIAO +- Python 3 on your Mac, Linux computer, or Windows PC + +Configure the Meshtastic radio's serial interface for: + +```text +Enabled: yes +Mode: PROTO +Baud: 115200 ``` -After the first PyPI release, install it by name with: +The radio's UART pin names depend on its model. Consult that board's pinout before +connecting wires. -```sh -python -m pip install micromesh +> [!CAUTION] +> Use **3.3 V UART logic** and always connect the grounds. Do not connect the boards' +> power pins unless you have verified their voltage and current requirements. + +## XIAO RP2040 quick start + +### 1. Wire the boards + +UART wires cross: TX goes to RX, and RX goes to TX. + +```text +XIAO RP2040 Meshtastic radio +──────────────────────────────────────────────────── +D6 / GPIO0 / UART0 TX ──────────▢ UART RX +D7 / GPIO1 / UART0 RX ◀────────── UART TX +GND ─────────── GND ``` -## Quick start +The XIAO D6/D7 mapping comes from the +[`XIAO RP2040 pinout`](https://wiki.seeedstudio.com/XIAO-RP2040/). + +### 2. Prepare a Python environment + +Run these commands from the MicroMesh repository directory: + +```sh +python3 -m venv .venv +source .venv/bin/activate +python -m pip install --upgrade mpremote +``` + +Windows PowerShell uses this activation command instead: + +```powershell +.venv\Scripts\Activate.ps1 +``` + +### 3. Find the XIAO serial port + +Connect the XIAO over USB, then run: + +```sh +mpremote connect list +``` + +On macOS it will usually look similar to `/dev/cu.usbmodem2101`. Linux commonly uses +`/dev/ttyACM0`; Windows uses a name such as `COM4`. + +The commands below use `/dev/cu.usbmodem2101`. Replace it with your port. + +### 4. Install MicroMesh on the XIAO + +```sh +mpremote connect /dev/cu.usbmodem2101 mip install package.json +``` + +This installs the `micromesh` package under `/lib` on the board. + +### 5. Run the example + +The example broadcasts one `hello from XIAO RP2040` message after each successful +configuration download. Comment out its `mesh.sendText(...)` line first if you want a +receive-only test. + +```sh +mpremote connect /dev/cu.usbmodem2101 run examples/xiao_rp2040.py +``` + +You should immediately see: + +```text +MicroMesh starting on XIAO RP2040 +UART0: D6/GPIO0 TX -> radio RX +UART0: D7/GPIO1 RX <- radio TX +UART0: 115200 baud; common GND required +configuration requested; id=... +``` + +When the radio finishes sending its configuration and node database: + +```text +radio configuration complete; sending greeting +connected; 17 nodes known +``` + +The number of nodes will be different for your radio. Press `Ctrl-C` to stop. + +### 6. Start it automatically + +Once the temporary run works, save it as the XIAO boot program: + +```sh +mpremote connect /dev/cu.usbmodem2101 fs cp examples/xiao_rp2040.py :main.py +mpremote connect /dev/cu.usbmodem2101 reset +``` + +To stop or replace an automatically running script, connect the XIAO, press `Ctrl-C` +in `mpremote repl`, and upload a different `main.py`. + +## Curses dashboard through the XIAO + +The desktop curses dashboard provides keyboard message composition and a resizable UI. +It displays live connection stats, known nodes, battery levels, SNR, hops, last-seen +times, and incoming messages. The XIAO runs a small bridge between the radio UART and +the dashboard over USB. + +The dashboard needs a small bridge program on the XIAO. The regular +`xiao_rp2040.py` example prints human-readable logs; it does not expose structured data +to desktop applications. + +### 1. Install the desktop dependencies + +```sh +source .venv/bin/activate +python -m pip install -e . pyserial +``` + +Windows users also need: + +```powershell +python -m pip install windows-curses +``` + +### 2. Install the bridge on the XIAO + +This replaces the current `main.py`: + +```sh +mpremote connect /dev/cu.usbmodem2101 fs cp examples/xiao_dashboard_bridge.py :main.py +mpremote connect /dev/cu.usbmodem2101 reset +``` + +### 3. Close `mpremote`, then launch the dashboard + +```sh +python examples/curses_dashboard.py --xiao-bridge /dev/cu.usbmodem2101 +``` + +Only one application can own a serial port. Close `mpremote`, Arduino Serial Monitor, +screen, minicom, and other serial tools before launching the dashboard. + +The message field is always active. Type a message and press Enterβ€”there is no separate +compose mode. + +| Key | Action | +| --- | --- | +| Type + Enter | Broadcast the text in the message field | +| Escape | Clear the message field | +| `F5` (or `Ctrl-R`) | Request the configuration and node database again | +| `F10` (or `Ctrl-Q`) | Quit cleanly | + +To switch back to the regular logging example: + +```sh +mpremote connect /dev/cu.usbmodem2101 fs cp examples/xiao_rp2040.py :main.py +mpremote connect /dev/cu.usbmodem2101 reset +``` + +## Direct-to-radio dashboard + +If the Meshtastic radio itself is connected to your computer over USB, the desktop can +talk directly to it. Do not use `--xiao-bridge` in this mode. + +```sh +python -m pip install -e . pyserial +python -m serial.tools.list_ports -v +python examples/curses_dashboard.py /dev/cu.YOUR_RADIO_PORT +``` + +The XIAO's USB port identifies as a MicroPython board. Direct mode will not work with +that port; use `--xiao-bridge` when the radio is wired through the XIAO. + +## Other boards + +Ready-to-run examples are included for common MicroPython boards: + +| Board | Example | UART | TX | RX | +| --- | --- | --- | --- | --- | +| Seeed Studio XIAO RP2040 | [`xiao_rp2040.py`](examples/xiao_rp2040.py) | UART0 | D6 / GPIO0 | D7 / GPIO1 | +| Raspberry Pi Pico / Pico W | [`raspberry_pi_pico.py`](examples/raspberry_pi_pico.py) | UART0 | GP0 | GP1 | +| Generic ESP32 | [`esp32.py`](examples/esp32.py) | UART2 | GPIO17 | GPIO16 | + +Change the UART and pin numbers if your board uses a different mapping. See the +complete [examples guide](examples/README.md) for wiring notes. + +## Troubleshooting + +### `could not enter raw repl` + +Use the exact port instead of `connect auto`: + +```sh +mpremote connect list +mpremote connect /dev/cu.usbmodem2101 exec "print('MicroPython REPL OK')" +``` + +If it still fails, close every other serial application, unplug and reconnect the +XIAO, and try again. + +### It repeatedly says `waiting for radio data` + +The XIAO is running, but no complete Meshtastic frame has arrived. Check: + +1. XIAO D6/TX goes to radio RXβ€”not radio TX. +2. XIAO D7/RX goes to radio TX. +3. Both boards share GND. +4. The radio UART is enabled in `PROTO` mode at 115200 baud. +5. The configured radio UART pins match the pins you physically connected. + +### The desktop curses dashboard stays at `configuration requested` + +If the radio is wired through the XIAO, install `xiao_dashboard_bridge.py` as +`main.py` and include `--xiao-bridge` in the dashboard command. Without the bridge, +the desktop sends Meshtastic data to the MicroPython REPL instead of UART0. + +Also confirm that `mpremote` or another serial monitor is not holding the port. + +### The node count appears, but no messages appear + +The node list is stored data downloaded from the radio. It is not a message history. +Send a **new** text from another Meshtastic node while MicroMesh is running. + +### A frame is skipped or reports a decode error + +Meshtastic's protobuf schema evolves. MicroMesh reports and skips newer messages that +its compact schema does not yet understand. Text and other supported packets continue +processing. Update the installed files after pulling a newer MicroMesh version: + +```sh +mpremote connect /dev/cu.usbmodem2101 mip install package.json +``` + +### The serial port is busy + +One port can have only one owner. Quit the dashboard before using `mpremote`, and close +`mpremote` before restarting the dashboard. + +## Python API + +Here is the smallest useful program: ```python -from machine import UART, Pin +from machine import Pin, UART +from time import sleep_ms + from micromesh import PortNum, SerialInterface -uart = UART(1, baudrate=115200, tx=Pin(17), rx=Pin(16), timeout=0) def received(packet): if packet.WhichOneof("payload_variant") != "decoded": return if packet.decoded.portnum == PortNum.TEXT_MESSAGE_APP: - print("from !%08x:" % packet.from_, packet.decoded.payload.decode()) + print("from !%08x: %s" % ( + packet.from_, + packet.decoded.payload.decode("utf-8"), + )) + +uart = UART(0, 115200, tx=Pin(0), rx=Pin(1), timeout=0, rxbuf=1024) mesh = SerialInterface(uart, on_packet=received) mesh.connect() -mesh.sendText("hello mesh") while True: mesh.poll() + sleep_ms(10) ``` -The UART baud rate and pins depend on the attached Meshtastic device. `poll()` is -non-blocking when the UART is configured with `timeout=0`. - -Ready-to-run wiring and code for the Seeed Studio XIAO RP2040, Raspberry Pi Pico, and -generic ESP32 boards are in [`examples/`](examples/README.md). - -Direct messages accept either an integer or Meshtastic's hexadecimal node ID: +### Send packets ```python +# Broadcast text on the primary channel +mesh.sendText("hello mesh") + +# Direct message with an acknowledgement mesh.sendText("hello", destinationId="!a1b2c3d4", wantAck=True) + +# Position packet mesh.sendPosition(45.5152, -122.6784, altitude=15) + +# Custom application data mesh.sendData(b"custom", portNum=PortNum.PRIVATE_APP) ``` -## Protobufs +Payloads may be at most 233 bytes. `poll()` is non-blocking when the UART uses +`timeout=0`, so call it frequently from your main loop. -Messages can be used without a radio: +### Inspect connection state + +```python +if mesh.config_complete: + print("connected to", mesh.my_info.my_node_num) + print("known nodes:", len(mesh.nodes)) + +for number, info in mesh.nodes.items(): + name = info.user.long_name if info.HasField("user") else "unknown" + print("!%08x %s" % (number, name)) +``` + +### Use the lightweight protobuf codec ```python from micromesh import Data, PortNum -data = Data(portnum=PortNum.TEXT_MESSAGE_APP, payload=b"hello") -encoded = data.SerializeToString() +message = Data(portnum=PortNum.TEXT_MESSAGE_APP, payload=b"hello") +encoded = message.SerializeToString() decoded = Data().ParseFromString(encoded) +print(decoded.payload) ``` Generated-module-style imports are available as `micromesh.mesh_pb2` and -`micromesh.portnums_pb2` for easier porting from the desktop library. +`micromesh.portnums_pb2`. A protobuf field named `from` is accessed as `from_` because +`from` is a Python keyword. -Only the commonly useful schemas are modeled. Unknown fields survive a decode/encode -round trip, but their contents are not interpreted. A field named `from` is accessed as -`packet.from_` because `from` is a Python keyword. +## Installation alternatives + +Install from a local checkout with MicroPython's package installer: + +```sh +mpremote mip install package.json +``` + +Or copy the package directory manually: + +```sh +mpremote fs cp -r micromesh :lib/ +``` + +For CPython development: + +```sh +python -m pip install -e . +``` + +After the first PyPI release: + +```sh +python -m pip install micromesh +``` + +## Supported scope + +MicroMesh currently models the most useful portions of the Meshtastic API: + +- `ToRadio` / `FromRadio` framing and configuration handshake +- `MeshPacket`, `Data`, `Position`, `User`, and `NodeInfo` +- Device metrics, waypoints, routing, queue status, and log records +- Unknown-field preservation and stream resynchronization + +Large configuration, admin, metadata, and module-configuration messages are retained +as encoded bytes rather than fully interpreted. + +The protocol definitions come from +[`meshtastic/protobufs`](https://github.com/meshtastic/protobufs). ## Development +Set up the project and run the tests: + ```sh +python3 -m venv .venv +source .venv/bin/activate +python -m pip install -e . python -m unittest discover -s tests ``` +The test suite covers encoding, decoding, framing, serial recovery, sending, and +configuration state. + ## Releasing to PyPI -PyPI publishing uses GitHub Trusted Publishing, so no API token is stored in the -repository. Configure a pending publisher at -[`pypi.org/manage/account/publishing`](https://pypi.org/manage/account/publishing/) -with these values: +Maintainers publish with GitHub Trusted Publishing; no long-lived PyPI token is stored +in the repository. Configure the pending publisher with: -- PyPI project name: `micromesh` -- GitHub owner: `pdxlocations` -- Repository: `micromesh` -- Workflow: `release.yaml` -- Environment: `pypi` +| Setting | Value | +| --- | --- | +| PyPI project | `micromesh` | +| GitHub owner | `pdxlocations` | +| Repository | `micromesh` | +| Workflow | `release.yaml` | +| Environment | `pypi` | Keep the version in `pyproject.toml`, `package.json`, and `micromesh/__init__.py` in -sync. Push the change, create a matching GitHub release such as `v0.1.0`, and publish -the release. The workflow builds, validates, and uploads both the wheel and source -distribution. +sync. Publishing a matching GitHub release, such as `v0.1.0`, builds, validates, and +uploads the wheel and source distribution. -The API protocol is defined by the -[`meshtastic/protobufs`](https://github.com/meshtastic/protobufs) project. Update the -field declarations in `micromesh/messages.py` when adopting newer schemas. +## License -`manifest.py` is also included for freezing the package into custom MicroPython firmware. +MicroMesh is released under the [GNU General Public License v3.0 or later](LICENSE). diff --git a/examples/README.md b/examples/README.md index c1ab628..d71e889 100644 --- a/examples/README.md +++ b/examples/README.md @@ -10,6 +10,35 @@ Meshtastic radio. The radio must have its serial module enabled in `PROTO` mode | Raspberry Pi Pico / Pico W | `raspberry_pi_pico.py` | UART0 | GP0 | GP1 | | Generic ESP32 | `esp32.py` | UART2 | GPIO17 | GPIO16 | +## Desktop curses dashboard + +`curses_dashboard.py` is a host-side dashboard for macOS or Linux. It displays +connection statistics, known nodes, and received messages, and it can send broadcast +messages. It can connect directly to a radio or through a XIAO RP2040 UART bridge. + +```sh +python -m pip install -e . pyserial +python examples/curses_dashboard.py /dev/cu.usbmodem123456 +``` + +For a radio wired to a XIAO RP2040, first install the bridge as the XIAO's boot script: + +```sh +mpremote connect /dev/cu.usbmodem2101 fs cp examples/xiao_dashboard_bridge.py :main.py +mpremote connect /dev/cu.usbmodem2101 reset +python examples/curses_dashboard.py --xiao-bridge /dev/cu.usbmodem2101 +``` + +Do not run `mpremote` at the same time as the dashboard; only one program can own the +XIAO USB serial port. To return to the regular example, replace `main.py` with +`xiao_rp2040.py`. + +Use `mpremote connect list`, `python -m serial.tools.list_ports`, or your operating +system's device list to find the radio port. The message field is always active: type +and press Enter to send. Use Escape to clear it, F5 to refresh, and F10 to quit. +`Ctrl-R` and `Ctrl-Q` are also supported. On Windows, install `windows-curses` and use +a port such as `COM4`. + ## Wiring UART signals cross between the two boards: diff --git a/examples/curses_dashboard.py b/examples/curses_dashboard.py new file mode 100644 index 0000000..32eebbb --- /dev/null +++ b/examples/curses_dashboard.py @@ -0,0 +1,386 @@ +#!/usr/bin/env python3 +"""Curses dashboard for a Meshtastic radio connected to a desktop computer. + +Install pyserial, then pass the radio's serial device: + + python -m pip install -e . pyserial + python examples/curses_dashboard.py /dev/cu.usbmodem123456 + +The message field is always active. Enter sends, Escape clears, F5 refreshes, and +F10 quits. Ctrl-R and Ctrl-Q are also supported. +""" + +import argparse +import curses +import json +import time +from collections import deque + +from micromesh import PortNum, SerialInterface + + +EVENT_PREFIX = "MMEVT " +COMMAND_PREFIX = "MMCMD " + + +class PySerialUART: + """Give pyserial the small UART interface expected by MicroMesh.""" + + def __init__(self, port): + self.port = port + + def any(self): + return self.port.in_waiting + + def read(self, count=None): + return self.port.read(count or 1) + + def write(self, data): + return self.port.write(data) + + +class DashboardState: + def __init__(self): + self.started = time.monotonic() + self.frames = 0 + self.packets = 0 + self.text_received = 0 + self.text_sent = 0 + self.decode_errors = 0 + self.messages = deque(maxlen=200) + self.status = "starting" + self.draft = "" + + def add_message(self, source, destination, text): + self.messages.append((time.strftime("%H:%M:%S"), source, destination, text)) + + +class BridgeRecord: + """Small generated-message lookalike used by bridge events.""" + + def __init__(self, **values): + self._present = set(values) + for name, value in values.items(): + setattr(self, name, value) + + def HasField(self, name): + return name in self._present + + def __getattr__(self, name): + return 0 + + +class XiaoBridge: + """Desktop side of the JSON-lines bridge running on the XIAO.""" + + def __init__(self, port, state): + self.port = port + self.state = state + self.my_info = None + self.nodes = {} + self.config_complete = False + self._buffer = bytearray() + + def _command(self, kind, **values): + command = {"type": kind} + command.update(values) + line = COMMAND_PREFIX + json.dumps(command, separators=(",", ":")) + "\n" + self.port.write(line.encode("utf-8")) + + def connect(self): + self._command("refresh") + + def request_config(self): + self.config_complete = False + self._command("refresh") + + def sendText(self, text): + self._command("send", text=text) + + def poll(self): + waiting = self.port.in_waiting + if waiting: + self._buffer.extend(self.port.read(waiting)) + while True: + newline = self._buffer.find(b"\n") + if newline < 0: + break + raw = bytes(self._buffer[:newline]) + del self._buffer[:newline + 1] + line = raw.decode("utf-8", "replace") + marker = line.find(EVENT_PREFIX) + if marker < 0: + continue + try: + self._event(json.loads(line[marker + len(EVENT_PREFIX):])) + except (ValueError, TypeError) as error: + self.state.decode_errors += 1 + self.state.status = "bad bridge event: %s" % error + + def _event(self, event): + kind = event.get("type") + if kind == "frame": + self.state.frames += 1 + self.state.status = "received %s" % event.get("variant", "frame") + elif kind == "local": + self.my_info = BridgeRecord(my_node_num=event["num"]) + elif kind == "config": + self.config_complete = bool(event.get("complete")) + self.state.status = "configuration complete" + elif kind == "node": + user = BridgeRecord( + id=event.get("user_id", ""), + long_name=event.get("name", "unknown"), + short_name=event.get("short", "--"), + ) + values = {"num": event["num"], "user": user} + for source, target in ( + ("snr", "snr"), ("last_heard", "last_heard"), + ("hops", "hops_away"), + ): + if source in event: + values[target] = event[source] + if "battery" in event: + values["device_metrics"] = BridgeRecord(battery_level=event["battery"]) + self.nodes[event["num"]] = BridgeRecord(**values) + elif kind == "packet": + self.state.packets += 1 + if event.get("is_text"): + self.state.text_received += 1 + self.state.add_message( + node_id(event.get("source", 0)), + node_id(event.get("destination", 0)), + event.get("text", "[packet]"), + ) + elif kind == "error": + self.state.decode_errors += 1 + self.state.status = event.get("text", "bridge error") + elif kind == "status": + self.state.status = event.get("text", "bridge status") + elif kind == "heartbeat": + self.config_complete = bool(event.get("connected")) + self.state.status = "bridge connected" if self.config_complete else "bridge waiting" + + +def node_id(number): + return "!%08x" % int(number) + + +def age(timestamp): + if not timestamp: + return "--" + seconds = max(0, int(time.time()) - int(timestamp)) + if seconds < 60: + return "%ds" % seconds + if seconds < 3600: + return "%dm" % (seconds // 60) + if seconds < 86400: + return "%dh" % (seconds // 3600) + return "%dd" % (seconds // 86400) + + +def clip(value, width): + value = str(value) + if width <= 0: + return "" + if len(value) <= width: + return value + if width == 1: + return value[:1] + return value[:width - 1] + "…" + + +def node_values(info): + name = "unknown" + short = "--" + if info.HasField("user"): + name = info.user.long_name or info.user.id or name + short = info.user.short_name or short + + snr = "%.1f" % info.snr if info.HasField("snr") else "--" + battery = "--" + if info.HasField("device_metrics") and info.device_metrics.HasField("battery_level"): + battery = "%d%%" % info.device_metrics.battery_level + hops = str(info.hops_away) if info.HasField("hops_away") else "--" + heard = age(info.last_heard) if info.HasField("last_heard") else "--" + return node_id(info.num), name, short, snr, battery, hops, heard + + +def add_line(screen, row, text, style=0): + height, width = screen.getmaxyx() + if 0 <= row < height and width > 1: + try: + screen.addnstr(row, 0, text, width - 1, style) + except curses.error: + pass + + +def draw(screen, mesh, state): + screen.erase() + height, width = screen.getmaxyx() + + title_style = curses.A_BOLD + if curses.has_colors(): + title_style |= curses.color_pair(1) + add_line(screen, 0, " MicroMesh dashboard ", title_style) + + local = node_id(mesh.my_info.my_node_num) if mesh.my_info else "unknown" + connected = "yes" if mesh.config_complete else "waiting" + uptime = int(time.monotonic() - state.started) + add_line( + screen, + 1, + "Local: %s Config: %s Uptime: %ds Nodes: %d" % ( + local, connected, uptime, len(mesh.nodes) + ), + ) + add_line( + screen, + 2, + "Frames: %d Packets: %d Text RX/TX: %d/%d Decode errors: %d" % ( + state.frames, + state.packets, + state.text_received, + state.text_sent, + state.decode_errors, + ), + ) + + nodes_top = 4 + messages_height = max(5, height // 3) + messages_top = max(nodes_top + 3, height - messages_height - 3) + node_rows = max(1, messages_top - nodes_top - 2) + + add_line(screen, nodes_top, "Nodes", title_style) + add_line(screen, nodes_top + 1, "ID Name Sh SNR Batt Hops Seen") + nodes = sorted(mesh.nodes.values(), key=lambda item: item.last_heard, reverse=True) + for offset, info in enumerate(nodes[:node_rows]): + values = node_values(info) + line = "%-10s %-20s %-4s %5s %5s %4s %4s" % ( + values[0], clip(values[1], 20), clip(values[2], 4), + values[3], values[4], values[5], values[6], + ) + add_line(screen, nodes_top + 2 + offset, line) + + add_line(screen, messages_top, "Messages", title_style) + visible = max(1, height - messages_top - 4) + for offset, message in enumerate(list(state.messages)[-visible:]): + when, source, destination, text = message + prefix = "%s %s>%s " % (when, source, destination) + add_line(screen, messages_top + 1 + offset, prefix + clip(text, width - len(prefix) - 1)) + + add_line(screen, height - 3, "Status: " + state.status) + add_line(screen, height - 2, "Message: " + state.draft + "_", curses.A_REVERSE) + add_line( + screen, + height - 1, + "Type + Enter: send | Esc: clear | F5: refresh | F10: quit", + ) + screen.refresh() + + +def run(screen, port_name, baudrate, xiao_bridge): + try: + import serial + except ImportError: + raise SystemExit("pyserial is required: python -m pip install pyserial") + + state = DashboardState() + screen.nodelay(True) + screen.keypad(True) + try: + curses.curs_set(0) + except curses.error: + pass + if curses.has_colors(): + curses.start_color() + curses.use_default_colors() + curses.init_pair(1, curses.COLOR_CYAN, -1) + + serial_port = serial.Serial(port_name, baudrate=baudrate, timeout=0) + + def on_receive(message): + state.frames += 1 + state.status = "received %s" % (message.WhichOneof("payload_variant") or "frame") + + def on_packet(packet): + state.packets += 1 + if packet.WhichOneof("payload_variant") != "decoded": + state.add_message(node_id(packet.from_), node_id(packet.to), "[encrypted]") + return + data = packet.decoded + if data.portnum == PortNum.TEXT_MESSAGE_APP: + try: + text = data.payload.decode("utf-8") + except UnicodeError: + text = repr(data.payload) + state.text_received += 1 + else: + text = "[port %d, %d bytes]" % (data.portnum, len(data.payload)) + state.add_message(node_id(packet.from_), node_id(packet.to), text) + + def on_error(error, payload): + state.decode_errors += 1 + state.status = "skipped frame: %s" % error + + if xiao_bridge: + mesh = XiaoBridge(serial_port, state) + else: + mesh = SerialInterface( + PySerialUART(serial_port), + on_receive=on_receive, + on_packet=on_packet, + on_log=lambda line: setattr(state, "status", "radio: " + line), + on_error=on_error, + ) + + try: + mesh.connect() + state.status = "configuration requested" + while True: + mesh.poll() + draw(screen, mesh, state) + key = screen.getch() + if key in (17, curses.KEY_F10): # Ctrl-Q or F10 + break + if key in (18, curses.KEY_F5): # Ctrl-R or F5 + mesh.request_config() + state.status = "configuration requested" + elif key in (10, 13, curses.KEY_ENTER): + text = state.draft.strip() + state.draft = "" + if text: + try: + mesh.sendText(text) + state.text_sent += 1 + state.add_message("local", "all", text) + state.status = "message queued" + except (ValueError, OSError) as error: + state.status = "send failed: %s" % error + elif key == 27: # Escape + state.draft = "" + state.status = "message cleared" + elif key in (8, 127, curses.KEY_BACKSPACE): + state.draft = state.draft[:-1] + elif 32 <= key < 127 and len(state.draft) < 200: + state.draft += chr(key) + time.sleep(0.05) + finally: + serial_port.close() + + +def main(): + parser = argparse.ArgumentParser(description="MicroMesh curses dashboard") + parser.add_argument("port", help="Meshtastic serial port, such as /dev/cu.usbmodem1234") + parser.add_argument("--baud", type=int, default=115200, help="UART baud rate (default: 115200)") + parser.add_argument( + "--xiao-bridge", + action="store_true", + help="connect through xiao_dashboard_bridge.py running on a XIAO RP2040", + ) + args = parser.parse_args() + curses.wrapper(run, args.port, args.baud, args.xiao_bridge) + + +if __name__ == "__main__": + main() diff --git a/examples/xiao_dashboard_bridge.py b/examples/xiao_dashboard_bridge.py new file mode 100644 index 0000000..5b4f892 --- /dev/null +++ b/examples/xiao_dashboard_bridge.py @@ -0,0 +1,162 @@ +"""USB-to-Meshtastic bridge for the XIAO RP2040 curses dashboard. + +Install this file as ``main.py`` on the XIAO. UART0 talks to the Meshtastic +radio on D6/D7, while newline-delimited JSON events and commands use USB stdio. +""" + +import sys + +try: + import ujson as json +except ImportError: + import json + +try: + import uselect as select +except ImportError: + import select + +from machine import Pin, UART +from time import sleep_ms, ticks_diff, ticks_ms + +from micromesh import PortNum, SerialInterface + + +EVENT_PREFIX = "MMEVT " +COMMAND_PREFIX = "MMCMD " + + +def emit(event): + print(EVENT_PREFIX + json.dumps(event)) + flush = getattr(sys.stdout, "flush", None) + if flush: + flush() + + +def on_packet(packet): + event = { + "type": "packet", + "source": packet.from_, + "destination": packet.to, + "portnum": 0, + "text": "[encrypted]", + "is_text": False, + } + if packet.WhichOneof("payload_variant") == "decoded": + data = packet.decoded + event["portnum"] = data.portnum + if data.portnum == PortNum.TEXT_MESSAGE_APP: + try: + event["text"] = data.payload.decode("utf-8") + except UnicodeError: + event["text"] = repr(data.payload) + event["is_text"] = True + else: + event["text"] = "[port %d, %d bytes]" % ( + data.portnum, len(data.payload) + ) + emit(event) + + +def node_event(info): + event = {"type": "node", "num": info.num} + if info.HasField("user"): + event["name"] = info.user.long_name or info.user.id or "unknown" + event["short"] = info.user.short_name or "--" + event["user_id"] = info.user.id + if info.HasField("snr"): + event["snr"] = info.snr + if info.HasField("last_heard"): + event["last_heard"] = info.last_heard + if info.HasField("hops_away"): + event["hops"] = info.hops_away + if info.HasField("device_metrics") and info.device_metrics.HasField("battery_level"): + event["battery"] = info.device_metrics.battery_level + return event + + +def on_receive(message): + variant = message.WhichOneof("payload_variant") or "unknown" + emit({"type": "frame", "variant": variant}) + if variant == "my_info": + emit({"type": "local", "num": message.my_info.my_node_num}) + elif variant == "node_info": + emit(node_event(message.node_info)) + elif variant == "config_complete_id": + emit({ + "type": "config", + "complete": message.config_complete_id == mesh.config_id, + "nodes": len(mesh.nodes), + }) + + +def on_error(error, payload): + emit({"type": "error", "text": str(error), "bytes": len(payload)}) + + +uart = UART( + 0, + baudrate=115200, + tx=Pin(0), # XIAO D6 + rx=Pin(1), # XIAO D7 + timeout=0, + rxbuf=2048, +) + +mesh = SerialInterface( + uart, + on_receive=on_receive, + on_packet=on_packet, + on_log=lambda line: emit({"type": "status", "text": "radio: " + line}), + on_error=on_error, +) + +command_poll = select.poll() +command_poll.register(sys.stdin, select.POLLIN) + + +def handle_commands(): + for _ in command_poll.poll(0): + first = sys.stdin.read(1) + if first == "\x03": + # Never consume the interrupt mpremote uses to regain the REPL. + raise KeyboardInterrupt + # Dashboard commands are always newline-terminated. Reading the rest + # as one line also drains bytes MicroPython buffered after the first + # character but no longer reports through select.poll(). + line = first + sys.stdin.readline() + line = line.strip() + if not line.startswith(COMMAND_PREFIX): + continue + try: + command = json.loads(line[len(COMMAND_PREFIX):]) + kind = command.get("type") + if kind == "send": + text = command.get("text", "") + if text: + mesh.sendText(text) + emit({"type": "status", "text": "message queued"}) + elif kind == "refresh": + mesh.request_config() + emit({"type": "status", "text": "configuration requested"}) + except Exception as error: + emit({"type": "error", "text": "command: " + str(error)}) + + +emit({"type": "status", "text": "XIAO bridge starting"}) +mesh.connect() +emit({"type": "status", "text": "configuration requested"}) + +last_status = ticks_ms() +while True: + mesh.poll() + handle_commands() + now = ticks_ms() + if ticks_diff(now, last_status) >= 5000: + emit({ + "type": "heartbeat", + "connected": mesh.config_complete, + "nodes": len(mesh.nodes), + }) + last_status = now + sleep_ms(10) diff --git a/micromesh/__pycache__/stream.cpython-311.pyc b/micromesh/__pycache__/stream.cpython-311.pyc index 1d911648ea0922d868c52d5c2c3bbc6d9dce7ceb..0404023ac90c003fa77980b5b87c201b311d839d 100644 GIT binary patch delta 810 zcmZvZ&ubGw6vyA}W|Pg@P12VBFa(Wl)3Aa;teOh7A+|jUu_YjSkg{gi{z^77*_O6c z+JiTR3L~`j2lgOZs0V*sy!s!A#|5D`n}e{`BEgeyHvNSX6~4(A-w=QmSRe+O0s0hIVM^dBbZG%5x#0~u zhdYei+flZBY~=X)PO#;DA3zt-Bmu*>E-@lCw}TR+#dlsox-bM>bgk0V`}1_PrsxXkins!wMnru-g#xp!uqC zRH`l_ev?d2BaT=Yr<&jhsk4qWqZg;m3U69GP9r=f*f(Zc=ImYCKd1u9!zC>#82l)wufzomRllHrwf0BT??n9mwBDM`g)_~zEatkE}H!` W!`8!@3%vvr|HTgX?f&P3C-nzXD8zmM delta 494 zcmZ3XG(~}LIWI340}ups=VnQ9Zse0^a`n>>Elw>e)=x^!%gHP#POa4UO)btSNh~hO zOxBN2&n(F(P0}w&smRGsPAtjH&nwo?%}g%J&jl*d2N7W2W^<--2RM=7&f*DG9VJ1nWux5kJD3U?8d;$Ap zM^@3vTY01y*(P7$k>Y1BN-fAqOiq2t0CbWj$7D-B)yXM*0gMkPZ{<^GVHR{}oP3{8 zRe%x5_`rZ5fr6q^J3KE+yIqlXo8SV1ADBVXKYsjpARsb1ls}Qv$kCknpc$j1`Q-il z{i0l~${!feiOI2KczG$)vm~F@)iMmQzJ&8Moi)(NbCz1 H39wNBV4{9% diff --git a/micromesh/stream.py b/micromesh/stream.py index 4d38cda..163ace4 100644 --- a/micromesh/stream.py +++ b/micromesh/stream.py @@ -60,7 +60,13 @@ class StreamParser: try: line = bytes(self._log).decode("utf-8") except UnicodeError: - line = bytes(self._log).decode("utf-8", "replace") + # Some MicroPython ports do not implement the optional + # ``errors`` argument to bytes.decode(). Preserve readable + # ASCII and replace other bytes without invoking a codec. + line = "".join( + chr(value) if 32 <= value < 127 else "?" + for value in self._log + ) self.on_log(line) self._log = bytearray() elif len(self._log) < 256: diff --git a/tests/__pycache__/test_micromesh.cpython-311.pyc b/tests/__pycache__/test_micromesh.cpython-311.pyc index c82f00852e206d09551275c30e026009506970b7..801fb5ad01c3634e914af1e628bd566657a070c9 100644 GIT binary patch delta 694 zcmaLU-D}fO6aettv}U2MYh%}~T_>iSol7U}+MyGvY(5Gim5R>F2rZ3i%lctbiw%WV zomxeh^$5>$kAe>(Lc0|<3jPmM^mX&zf*%j!>*|$t!6!Wqx5c zn<%`79%a(oZ_FzUio<(m)-+E=tqDtn60G`gSI&lB>M&rU=@kk%XAqU)bWcBBrlCIE z2Oq|)v`oQqyA2{ecBYIH1{p!vVy_raDRJCe?ZU)kiK9E)CXFG%)RbUU%s8vagdib_ z5e!N{iBp6$Mm1n~o>v_B+4dDAwIHemMT%v_d`wX#MO0(iR3W8Fs+h_aV{@s5c)4)# zh&Y#%Ra!d+S=UhIKP6#uM-)?~HZ5Gu_=VfMeCGLu1{Y{@ffg62F`I1n2J2X39gBTW zT}!Tp=Wx?G*>X<4+tXx2EjH9(Lw`zom)S+F$$DF?xAAK$aH)4de==A%p!4A8Pa+k{ zd=w>NgTI1G5O!xB_GOUgs zU)oq$=7^S&^SiVzTHraSM}NM&cFE)taqw=n|K8E)Hv|3KR)=W+?9gwt&{BdKak zmQ-Z{J_<#Y*QdR88ddbQgH?p?Lc~`=*J0iF7IARXKd~=M8s-Vt2zLk%2oDKMu&@*@(czr01P;iU-xA$gR_K~lh!|#8WMB!@C`xD%`%y<9* delta 329 zcmdm$xjL11IWI340}xD_lA9&Gk#`FlW6b13Y_TlC44V9#HQDop88tW0l;UFAJVS0G zGh^^%EhQhug30AdQH<%6FDX?pmQD6nu3`+>d`#JxnXz*6FEttN7NBuOwIHHwvZ8th zW76g>^|MTDjUbVx$+24DjA@g%YN<08PQI(<2Q*AaJ3<962Hk4Otb(yOktV; diff --git a/tests/test_micromesh.py b/tests/test_micromesh.py index f03c342..3a563e9 100644 --- a/tests/test_micromesh.py +++ b/tests/test_micromesh.py @@ -77,6 +77,12 @@ class MicroMeshTests(unittest.TestCase): self.assertEqual(packets, [b"one", b"two"]) self.assertEqual(logs, ["debug line"]) + def test_stream_parser_sanitizes_invalid_utf8_logs(self): + logs = [] + parser = StreamParser(on_log=logs.append) + self.assertEqual(parser.feed(b"bad:\xff\xfe\n"), []) + self.assertEqual(logs, ["bad:??"]) + def test_send_text_builds_to_radio_frame(self): import micromesh.interface as interface_module old_packet_id = interface_module._packet_id