6FPS/micromesh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enhance testing and error handling; update documentation and examples

Browse Source
This commit is contained in:
pdxlocations
2026-08-01 01:14:46 -07:00
parent d242a82d1d
commit dbaa3b38b8
21 changed files with 228 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/README.md
+7 -4
View File
@@ -1,8 +1,9 @@
# Board examples
These examples connect a MicroPython board to the hardware UART exposed by a
Meshtastic radio. The radio must have its serial module enabled in `PROTO` mode at
115200 baud.
Meshtastic radio. The radio must have its serial module enabled in `PROTO` mode. The
examples choose 115200 baud, so configure the radio to match; Meshtastic's serial
module defaults to 38400 baud.
| Board | Example | MicroPython UART | TX pin | RX pin |
| --- | --- | --- | --- | --- |
@@ -21,9 +22,11 @@ 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:
For a radio wired to a XIAO RP2040, install MicroMesh and then install the bridge as
the XIAO's boot script:
```sh
mpremote connect /dev/cu.usbmodem2101 mip install package.json
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
@@ -37,7 +40,7 @@ Use `mpremote connect list`, `python -m serial.tools.list_ports`, or your operat
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`.
a port such as `COM4`. A `message queued` status is not a delivery receipt.
## Wiring
examples/curses_dashboard.py
+33 -7
View File
@@ -21,6 +21,7 @@ from micromesh import PortNum, SerialInterface
EVENT_PREFIX = "MMEVT "
COMMAND_PREFIX = "MMCMD "
MAX_BRIDGE_BUFFER = 8192
class PySerialUART:
@@ -85,7 +86,10 @@ class XiaoBridge:
command = {"type": kind}
command.update(values)
line = COMMAND_PREFIX + json.dumps(command, separators=(",", ":")) + "\n"
self.port.write(line.encode("utf-8"))
data = line.encode("utf-8")
written = self.port.write(data)
if written is not None and written != len(data):
raise OSError("short serial write")
def connect(self):
self._command("refresh")
@@ -101,6 +105,11 @@ class XiaoBridge:
waiting = self.port.in_waiting
if waiting:
self._buffer.extend(self.port.read(waiting))
if len(self._buffer) > MAX_BRIDGE_BUFFER and b"\n" not in self._buffer:
self._buffer = bytearray()
self.state.decode_errors += 1
self.state.status = "discarded oversized bridge line"
return
while True:
newline = self._buffer.find(b"\n")
if newline < 0:
@@ -112,8 +121,11 @@ class XiaoBridge:
if marker < 0:
continue
try:
self._event(json.loads(line[marker + len(EVENT_PREFIX):]))
except (ValueError, TypeError) as error:
event = json.loads(line[marker + len(EVENT_PREFIX):])
if not isinstance(event, dict):
raise ValueError("event is not an object")
self._event(event)
except (KeyError, ValueError, TypeError) as error:
self.state.decode_errors += 1
self.state.status = "bad bridge event: %s" % error
@@ -206,6 +218,17 @@ def node_values(info):
return node_id(info.num), name, short, snr, battery, hops, heard
def config_status(mesh, state):
"""Describe useful progress even when the completion ID never matches."""
if mesh.config_complete:
return "complete"
if mesh.my_info is not None and mesh.nodes:
return "ready"
if state.frames or mesh.my_info is not None or mesh.nodes:
return "receiving"
return "waiting"
def add_line(screen, row, text, style=0):
height, width = screen.getmaxyx()
if 0 <= row < height and width > 1:
@@ -225,13 +248,13 @@ def draw(screen, mesh, state):
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"
config = config_status(mesh, state)
uptime = int(time.monotonic() - state.started)
add_line(
screen,
1,
"Local: %s Config: %s Uptime: %ds Nodes: %d" % (
local, connected, uptime, len(mesh.nodes)
local, config, uptime, len(mesh.nodes)
),
)
add_line(
@@ -348,10 +371,10 @@ def run(screen, port_name, baudrate, xiao_bridge):
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.draft = ""
state.text_sent += 1
state.add_message("local", "all", text)
state.status = "message queued"
@@ -372,7 +395,10 @@ def run(screen, port_name, baudrate, xiao_bridge):
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(
"--baud", type=int, default=115200,
help="UART baud rate (default: 115200)",
)
parser.add_argument(
"--xiao-bridge",
action="store_true",
examples/esp32.py
+10 -4
View File
@@ -18,13 +18,19 @@ def on_packet(packet):
if packet.WhichOneof("payload_variant") != "decoded":
return
if packet.decoded.portnum == PortNum.TEXT_MESSAGE_APP:
print("from !%08x: %s" % (
packet.from_, packet.decoded.payload.decode("utf-8", "replace")
))
try:
text = packet.decoded.payload.decode("utf-8")
except UnicodeError:
text = repr(packet.decoded.payload)
print("from !%08x: %s" % (packet.from_, text))
def on_error(error, payload):
print("skipping undecodable radio frame:", error, "(%d bytes)" % len(payload))
uart = UART(2, baudrate=115200, tx=Pin(17), rx=Pin(16), timeout=0, rxbuf=1024)
mesh = SerialInterface(uart, on_packet=on_packet, on_log=print)
mesh = SerialInterface(uart, on_packet=on_packet, on_log=print, on_error=on_error)
mesh.connect()
hello_sent = False
examples/raspberry_pi_pico.py
+10 -4
View File
@@ -16,13 +16,19 @@ def on_packet(packet):
if packet.WhichOneof("payload_variant") != "decoded":
return
if packet.decoded.portnum == PortNum.TEXT_MESSAGE_APP:
print("from !%08x: %s" % (
packet.from_, packet.decoded.payload.decode("utf-8", "replace")
))
try:
text = packet.decoded.payload.decode("utf-8")
except UnicodeError:
text = repr(packet.decoded.payload)
print("from !%08x: %s" % (packet.from_, text))
def on_error(error, payload):
print("skipping undecodable radio frame:", error, "(%d bytes)" % len(payload))
uart = UART(0, 115200, tx=Pin(0), rx=Pin(1), timeout=0, rxbuf=1024)
mesh = SerialInterface(uart, on_packet=on_packet, on_log=print)
mesh = SerialInterface(uart, on_packet=on_packet, on_log=print, on_error=on_error)
mesh.connect()
hello_sent = False