micromesh/examples/curses_dashboard.py

#!/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 "
MAX_BRIDGE_BUFFER = 8192


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"
        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")

    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))
        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:
                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:
                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

    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 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:
        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"
    config = config_status(mesh, state)
    uptime = int(time.monotonic() - state.started)
    add_line(
        screen,
        1,
        "Local: %s  Config: %s  Uptime: %ds  Nodes: %d" % (
            local, config, 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()
                if text:
                    try:
                        mesh.sendText(text)
                        state.draft = ""
                        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()