#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""
meshterm-spi-bridge: make an SPI-attached MeshCore radio available to MeshTerm.

Some boards (the hackergadgets AIO of the ClockworkPi uConsole, and Waveshare LoRa HATs)
connect the LoRa chip directly to the SPI bus of the host. They have no companion MCU and
no firmware to flash. The node runs in software on the host, with the `openhop_core`
library. The name of the library was `pymc_core` until its rename in mid-2026. The bridge
supports both names, and it prefers the newer one. But MeshTerm uses the MeshCore
companion protocol over serial, BLE, or TCP. It expects to talk to a node. It does not
drive a bare radio.

This tool connects the two. It runs the software node itself, and it puts the standard
companion protocol in front of the node, on a local TCP port. Then

    meshterm --tcp 127.0.0.1:5000

connects in the same way as if a real companion device is connected.

This file is two things: a setup menu and the bridge process. Run it with any Python 3.
It finds the node runtime for you, and it runs the bridge half again under that
interpreter automatically. You do not need to know about venvs.

    ./meshterm-spi-bridge            # interactive menu
    ./meshterm-spi-bridge --check    # run prerequisite checks and exit
    ./meshterm-spi-bridge --run      # run the bridge now in this terminal
"""
from __future__ import annotations

import argparse
import os
import shutil
import socket
import subprocess
import sys
from pathlib import Path

# --------------------------------------------------------------------------- #
# Configuration
# --------------------------------------------------------------------------- #

APP_NAME = "meshterm-spi-bridge"
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 5000
DEFAULT_NODE_NAME = "uConsole"

HOME = Path.home()

# The node runtimes that this bridge can drive, newest first. `pymc_core` became
# `openhop_core` in 2026. (PyPI `pymc-core` stops at 1.0.12, and `openhop-core` continues
# from 1.1.x.) The newer library has taken two of the three compat shims of this file
# upstream. Thus, when both are installed, the bridge uses the newer one.
RUNTIMES = ("openhop_core", "pymc_core")

# The interpreters that probably have a runtime, in order of preference. The first is the
# own venv of the bridge. This is the answer on a host where the distro package cannot be
# upgraded. (The meshcore-uconsole .deb for trixie is built against Python 3.13 and does
# not run on a bookworm uConsole. `python3 -m venv <this path> && pip install
# "openhop-core[hardware]"` gives the bridge the current runtime, and the GUI keeps its
# own.) The second is the private virtualenv of the hackergadgets uConsole .deb.
VENV_CANDIDATES = [
    str(HOME / ".local" / "share" / APP_NAME / "venv" / "bin" / "python"),
    "/opt/venvs/meshcore-uconsole/bin/python",
]
INSTALL_PATH = HOME / ".local" / "bin" / APP_NAME
UNIT_NAME = f"{APP_NAME}.service"
UNIT_PATH = HOME / ".config" / "systemd" / "user" / UNIT_NAME
MESHTERM_CONFIG = HOME / ".meshterm" / "config.toml"

RULE = "=" * 62
THIN = "-" * 62


# --------------------------------------------------------------------------- #
# Colour (restrained, and off when piped, on a dumb terminal, or when NO_COLOR is set)
# --------------------------------------------------------------------------- #

_USE_COLOR = (
    sys.stdout.isatty()
    and not os.environ.get("NO_COLOR")
    and not os.environ.get("MESHTERM_NO_COLOR")
    and os.environ.get("TERM", "") != "dumb"
)


def _c(text: str, code: str) -> str:
    return f"\033[{code}m{text}\033[0m" if _USE_COLOR else text


def bold(t: str) -> str:
    return _c(t, "1")


def dim(t: str) -> str:
    return _c(t, "2")


def green(t: str) -> str:
    return _c(t, "32")


def red(t: str) -> str:
    return _c(t, "1;31")


def yellow(t: str) -> str:
    return _c(t, "33")


def cyan(t: str) -> str:
    return _c(t, "36")


def cmd(t: str) -> str:
    """Highlight a command that the user must type or copy."""
    return bold(cyan(t))


# --------------------------------------------------------------------------- #
# Small text helpers (plain on purpose: a little colour, and no decoration)
# --------------------------------------------------------------------------- #


def tag(ok: bool, warn: bool = False) -> str:
    if warn:
        return yellow("[warn]")
    return green("[ ok ]") if ok else red("[FAIL]")


def ask(prompt: str, default: str = "") -> str:
    try:
        return input(prompt).strip()
    except (EOFError, KeyboardInterrupt):
        print()
        return default


def confirm(prompt: str, default_yes: bool = False) -> bool:
    suffix = " [Y/n] " if default_yes else " [y/N] "
    ans = ask(prompt + suffix).lower()
    if not ans:
        return default_yes
    return ans in ("y", "yes")


# --------------------------------------------------------------------------- #
# Runtime discovery
# --------------------------------------------------------------------------- #


def python_runtime(python: str) -> str | None:
    """The node runtime that `python` can import (a name from RUNTIMES), or None.

    The function starts one subprocess for each interpreter. The subprocess reports the
    first runtime that it finds, in the order of RUNTIMES. Thus an interpreter that has both
    runtimes answers `openhop_core`.
    """
    if not python or not os.path.exists(python):
        return None
    probe = (
        "import importlib.util as u, sys\n"
        f"for m in {RUNTIMES!r}:\n"
        "    if u.find_spec(m):\n"
        "        print(m); break\n"
    )
    try:
        r = subprocess.run(
            [python, "-c", probe],
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            text=True,
            timeout=20,
        )
    except Exception:
        return None
    found = r.stdout.strip()
    return found if r.returncode == 0 and found in RUNTIMES else None


def python_has_pymc(python: str) -> bool:
    """Return whether `python` can import a node runtime. This function keeps the older name."""
    return python_runtime(python) is not None


def _meshcore_console_python() -> str | None:
    """Resolve the interpreter behind the `meshcore-console` command, if the command exists."""
    exe = shutil.which("meshcore-console")
    if not exe:
        return None
    try:
        first = Path(exe).resolve().read_text(errors="replace").splitlines()[0]
    except Exception:
        return None
    if first.startswith("#!"):
        cand = first[2:].strip().split()[0]
        if cand.endswith("python") or "python" in os.path.basename(cand):
            return cand
    return None


def find_runtime_python() -> tuple[str, str] | None:
    """Find an interpreter that has a node runtime: `(python, runtime)`, or None.

    The candidates, in order, are: the explicit override, this interpreter, the known venv
    paths, and the interpreter behind an installed `meshcore-console`. The newest runtime
    wins across all of them. An interpreter with `openhop_core` is better than an earlier
    candidate that has only `pymc_core`. The override is the one exception.
    MESHTERM_PYMC_PYTHON names a python, and the function uses the runtime that it has.
    """
    seen: set[str] = set()
    override = os.environ.get("MESHTERM_PYMC_PYTHON", "")
    if override:
        rt = python_runtime(override)
        if rt:
            return override, rt
    candidates = [sys.executable, *VENV_CANDIDATES]
    mcc = _meshcore_console_python()
    if mcc:
        candidates.append(mcc)
    found: list[tuple[str, str]] = []
    for c in candidates:
        if c and c not in seen:
            seen.add(c)
            rt = python_runtime(c)
            if rt:
                found.append((c, rt))
    for wanted in RUNTIMES:
        for python, rt in found:
            if rt == wanted:
                return python, rt
    return None


def find_pymc_python() -> str | None:
    """The interpreter that `find_runtime_python` chooses, or None.

    This function keeps the older name.
    """
    hit = find_runtime_python()
    return hit[0] if hit else None


# --------------------------------------------------------------------------- #
# Preflight checks
# --------------------------------------------------------------------------- #


def _dev_ok(path: str) -> tuple[bool, str]:
    if not os.path.exists(path):
        return False, "missing"
    if not os.access(path, os.R_OK | os.W_OK):
        return False, "present but no read/write permission"
    return True, "present, read/write ok"


def _first_spidev() -> str | None:
    import glob

    devs = sorted(glob.glob("/dev/spidev*"))
    if "/dev/spidev1.0" in devs:
        return "/dev/spidev1.0"
    return devs[0] if devs else None


def port_listening(host: str, port: int) -> bool:
    """Return whether a process already accepts connections on host:port."""
    try:
        with socket.create_connection((host, port), timeout=1.5):
            return True
    except OSError:
        return False


def meshcore_console_running() -> list[str]:
    """The command lines of the meshcore-console (GUI or CLI) processes that run.

    The list can be incomplete.
    """
    try:
        out = subprocess.check_output(
            ["pgrep", "-af", "meshcore-console"], text=True, stderr=subprocess.DEVNULL
        )
    except Exception:
        return []
    lines = []
    for line in out.splitlines():
        # Our own bridge calls the python of the meshcore-uconsole venv, not the
        # meshcore-console binary. Thus it never matches, but the check is a guard.
        if "--run-bridge" in line or APP_NAME in line:
            continue
        lines.append(line.strip())
    return lines


def systemctl_user(*args: str) -> subprocess.CompletedProcess:
    return subprocess.run(
        ["systemctl", "--user", *args],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )


def service_installed() -> bool:
    return UNIT_PATH.exists()


def service_active() -> bool:
    return systemctl_user("is-active", UNIT_NAME).stdout.strip() == "active"


class Preflight:
    def __init__(self) -> None:
        hit = find_runtime_python()
        self.python, self.runtime = hit if hit else (None, None)
        self.spidev = _first_spidev()
        self.spidev_ok, self.spidev_detail = (
            _dev_ok(self.spidev) if self.spidev else (False, "no /dev/spidev* found")
        )
        self.gpio_ok, self.gpio_detail = _dev_ok("/dev/gpiochip0")
        self.svc_active = service_active()
        self.port_busy = port_listening(DEFAULT_HOST, DEFAULT_PORT)
        self.gui = meshcore_console_running()

    @property
    def can_run(self) -> bool:
        return bool(self.python) and self.spidev_ok and self.gpio_ok

    def print(self) -> None:
        print(bold("Prerequisite check:"))
        # 1. runtime
        if self.python:
            print(f"  {tag(True)} node runtime         {self.python} ({self.runtime})")
        else:
            print(f"  {tag(False)} node runtime         not found (openhop_core or pymc_core)")
            print("         fix: give the bridge its own runtime, on any Debian-family host:")
            print(f"              python3 -m venv {VENV_CANDIDATES[0][: -len('/bin/python')]}")
            print(f"              {VENV_CANDIDATES[0][: -len('python')]}pip install 'openhop-core[hardware]'")
            print("              (or set MESHTERM_PYMC_PYTHON to a python that has one)")
        # 2. spi
        print(f"  {tag(self.spidev_ok)} SPI device           {self.spidev or '-'} — {self.spidev_detail}")
        if not self.spidev_ok and self.spidev is None:
            print("         fix: enable SPI in /boot/firmware/config.txt")
            print("              (dtoverlay=spi1-1cs on the uConsole) and reboot")
        elif not self.spidev_ok:
            print("         fix: add your user to the 'spi' group, then log out/in:")
            print("              sudo usermod -aG spi $USER")
        # 3. gpio
        print(f"  {tag(self.gpio_ok)} GPIO chip            /dev/gpiochip0 — {self.gpio_detail}")
        if not self.gpio_ok and "permission" in self.gpio_detail:
            print("         fix: sudo usermod -aG gpio $USER  (then log out/in)")
        # 4. radio availability (information only)
        if self.svc_active:
            print(f"  {tag(True)} bridge service       running (radio in use by the boot service)")
        elif self.gui:
            print(f"  {tag(False, warn=True)} radio in use         meshcore-console is running:")
            for line in self.gui[:2]:
                print(f"              {line[:70]}")
            print("         note: only one program can use the radio at a time —")
            print("               close meshcore-console before running the bridge")
        elif self.port_busy:
            print(f"  {tag(False, warn=True)} port {DEFAULT_PORT} busy         something is already on {DEFAULT_HOST}:{DEFAULT_PORT}")
        else:
            print(f"  {tag(True)} radio is free        nothing else is using it")


# --------------------------------------------------------------------------- #
# Caveats banner
# --------------------------------------------------------------------------- #

CAVEATS = """\
How this works, briefly:
  * Your LoRa radio is wired to the host's SPI bus — there is no separate
    MCU and no firmware to flash. The mesh node runs as software here.
  * This bridge runs that node and exposes it on a local network port so
    MeshTerm can connect with:  meshterm --tcp {host}:{port}
  * Two processes cooperate: this bridge (owns the radio) and MeshTerm
    (the terminal you use). Keep the bridge running while you use MeshTerm.

Caveat — only ONE program may use the radio at a time:
  * Run EITHER the meshcore-console GUI, OR this bridge — never both.
  * If the bridge won't start, close meshcore-console first."""


def print_caveats() -> None:
    text = CAVEATS.format(host=DEFAULT_HOST, port=DEFAULT_PORT)
    for line in text.splitlines():
        if line.endswith(":") and not line.startswith(" "):
            print(bold(line))
        elif "meshterm --tcp" in line:
            print(line.replace(
                f"meshterm --tcp {DEFAULT_HOST}:{DEFAULT_PORT}",
                cmd(f"meshterm --tcp {DEFAULT_HOST}:{DEFAULT_PORT}"),
            ))
        else:
            print(line)


# --------------------------------------------------------------------------- #
# Actions
# --------------------------------------------------------------------------- #


def action_run_now(pre: Preflight) -> None:
    if pre.svc_active:
        print()
        print("The bridge is already running as a boot service, so the radio is")
        print("in use and MeshTerm can connect right now:")
        print("    " + cmd(f"meshterm --tcp {DEFAULT_HOST}:{DEFAULT_PORT}"))
        print("To run it in this terminal instead, stop the service first (menu")
        print("option 3, or: systemctl --user stop " + UNIT_NAME + ").")
        return
    if pre.gui:
        print()
        print("meshcore-console is running and holding the radio. Close it first,")
        print("then try again.")
        return
    if not pre.can_run:
        print("\nCan't run yet — resolve the [FAIL] items above first.")
        return

    print()
    print(dim(THIN))
    print("Starting the bridge. Leave this running, open another terminal, and:")
    print("    " + cmd(f"meshterm --tcp {DEFAULT_HOST}:{DEFAULT_PORT}"))
    print("Press Ctrl+C here to stop the bridge.")
    print(dim(THIN))
    offer_meshterm_profile()
    print()
    run_cmd = [pre.python, os.path.abspath(__file__), "--run-bridge",
               "--host", DEFAULT_HOST, "--port", str(DEFAULT_PORT), "--name", DEFAULT_NODE_NAME]
    try:
        subprocess.run(run_cmd)
    except KeyboardInterrupt:
        pass
    print("\nBridge stopped. Back to the menu.")


def _write_unit(python: str) -> None:
    UNIT_PATH.parent.mkdir(parents=True, exist_ok=True)
    exec_start = (
        f"{python} {INSTALL_PATH} --run-bridge "
        f"--host {DEFAULT_HOST} --port {DEFAULT_PORT} --name {DEFAULT_NODE_NAME}"
    )
    UNIT_PATH.write_text(
        "[Unit]\n"
        "Description=MeshTerm SPI-radio companion bridge (software node -> TCP)\n"
        "After=network.target\n\n"
        "[Service]\n"
        "Type=simple\n"
        f"ExecStart={exec_start}\n"
        "Restart=on-failure\n"
        "RestartSec=3\n\n"
        "[Install]\n"
        "WantedBy=default.target\n"
    )


def action_install_service(pre: Preflight) -> None:
    if not pre.python:
        print("\nCan't install — no node runtime was found (see above).")
        return
    if pre.gui:
        print("\nmeshcore-console is running. The service would fight it for the")
        print("radio. Close meshcore-console first, then install the service.")
        if not confirm("Install anyway?"):
            return

    print()
    print("This will install a background service that:")
    print(f"  * runs the bridge on {DEFAULT_HOST}:{DEFAULT_PORT}")
    print("  * starts automatically on boot and restarts if it crashes")
    print("  * runs as YOU, using your node identity (same as the GUI)")
    print("Reminder: while it's running, don't also open the meshcore-console GUI.")
    if not confirm("\nInstall and start it now?", default_yes=True):
        print("Cancelled.")
        return

    # Install a stable copy of this script. The unit points to this copy.
    INSTALL_PATH.parent.mkdir(parents=True, exist_ok=True)
    if os.path.abspath(__file__) != str(INSTALL_PATH):
        shutil.copyfile(os.path.abspath(__file__), INSTALL_PATH)
    INSTALL_PATH.chmod(0o755)

    _write_unit(pre.python)
    systemctl_user("daemon-reload")
    r = systemctl_user("enable", "--now", UNIT_NAME)
    if r.returncode != 0:
        print("\nFailed to start the service:")
        print("  " + (r.stderr.strip() or "unknown error"))
        return

    # Try to make the service start at boot, also before you log in.
    linger = subprocess.run(
        ["loginctl", "enable-linger", os.environ.get("USER", "")],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
    )

    print(green("\nInstalled and started."))
    if service_active():
        print(f"  {tag(True)} service is running")
    print(f"  unit:  {UNIT_PATH}")
    print(f"  logs:  journalctl --user -u {UNIT_NAME} -f")
    if linger.returncode != 0:
        print("\nOne extra step for start-on-boot without logging in first:")
        print(f"    sudo loginctl enable-linger {os.environ.get('USER','')}")
    offer_meshterm_profile()
    print("\nConnect from MeshTerm with:  " + cmd(f"meshterm --tcp {DEFAULT_HOST}:{DEFAULT_PORT}"))


def action_uninstall_service() -> None:
    if not service_installed():
        print("\nNo bridge service is installed. Nothing to remove.")
        print("(This never touches the hackergadgets meshcore-console package.)")
        return
    print("\nThis removes ONLY the bridge service and its helper script.")
    print("It does NOT remove meshcore-console or the radio stack.")
    if not confirm("Remove the bridge service now?", default_yes=True):
        print("Cancelled.")
        return
    systemctl_user("disable", "--now", UNIT_NAME)
    try:
        UNIT_PATH.unlink()
    except FileNotFoundError:
        pass
    systemctl_user("daemon-reload")
    # Remove the installed copy of the helper. Do not remove a repository checkout that
    # you ran the script from.
    if INSTALL_PATH.exists() and os.path.abspath(__file__) != str(INSTALL_PATH):
        try:
            INSTALL_PATH.unlink()
        except OSError:
            pass
    print(green("\nRemoved.") + " The radio is free again for the meshcore-console GUI.")
    if MESHTERM_CONFIG.exists() and "[profiles.bridge]" in MESHTERM_CONFIG.read_text(errors="replace"):
        print("Note: a [profiles.bridge] block remains in your MeshTerm config;")
        print(f"      edit {MESHTERM_CONFIG} to remove it if you like.")


def action_status(pre: Preflight) -> None:
    print()
    print(bold("Bridge service:"))
    if not service_installed():
        print("  not installed")
    else:
        print(f"  installed  {UNIT_PATH}")
        print(f"  active:    {'yes' if service_active() else 'no'}")
        print(f"  logs:      journalctl --user -u {UNIT_NAME} -f")
    listening = port_listening(DEFAULT_HOST, DEFAULT_PORT)
    print(f"\nEndpoint {DEFAULT_HOST}:{DEFAULT_PORT}: {'listening' if listening else 'not listening'}")
    if listening:
        print("  MeshTerm can connect now:  " + cmd(f"meshterm --tcp {DEFAULT_HOST}:{DEFAULT_PORT}"))


def offer_meshterm_profile() -> None:
    """Offer to add a MeshTerm TCP profile, so that a bare `meshterm` connects here."""
    if MESHTERM_CONFIG.exists() and "[profiles.bridge]" in MESHTERM_CONFIG.read_text(errors="replace"):
        return  # the profile is already there
    if not confirm(
        f"\nAdd a 'bridge' profile to {MESHTERM_CONFIG} so you can just run "
        "`meshterm`?",
        default_yes=False,
    ):
        return
    block = (
        "\n# Added by meshterm-spi-bridge — the local SPI-radio bridge.\n"
        "[profiles.bridge]\n"
        f'host = "{DEFAULT_HOST}"\n'
        f"tcp_port = {DEFAULT_PORT}\n"
        'description = "Local SPI radio via meshterm-spi-bridge"\n'
    )
    try:
        MESHTERM_CONFIG.parent.mkdir(parents=True, exist_ok=True)
        with MESHTERM_CONFIG.open("a", encoding="utf-8") as fh:
            fh.write(block)
        print("Added [profiles.bridge]. It'll show up in MeshTerm's device")
        print("picker as 'bridge' — just select it there, or run:  meshterm -p bridge")
    except OSError as exc:
        print(f"Couldn't write the profile ({exc}); add it by hand if you want it.")


# --------------------------------------------------------------------------- #
# Interactive menu
# --------------------------------------------------------------------------- #


def menu() -> int:
    while True:
        pre = Preflight()
        print()
        print(cyan(RULE))
        print(bold(cyan(" MeshTerm SPI-radio bridge — setup")))
        print(cyan(RULE))
        print_caveats()
        print()
        pre.print()
        print()
        print(bold("What would you like to do?"))
        print(f"  {bold('1')}) Run the bridge now   (stays in this terminal; Ctrl+C to stop)")
        print(f"  {bold('2')}) Install as a service (starts on boot, restarts on failure)")
        print(f"  {bold('3')}) Uninstall the service")
        print(f"  {bold('4')}) Show status")
        print(f"  {bold('5')}) Quit")
        choice = ask("\n> ")
        if choice == "1":
            action_run_now(pre)
        elif choice == "2":
            action_install_service(pre)
        elif choice == "3":
            action_uninstall_service()
        elif choice == "4":
            action_status(pre)
        elif choice in ("5", "q", "quit", "exit", ""):
            print("Bye.")
            return 0
        else:
            print("Please choose 1-5.")
        ask("\nPress Enter to continue…")


# --------------------------------------------------------------------------- #
# The bridge itself (runs under the pymc_core interpreter, through --run-bridge)
# --------------------------------------------------------------------------- #


def _import_runtime():
    """Import the node runtime that this interpreter has: `(module_name, module)`.

    The function uses the order of RUNTIMES. Thus `openhop_core` wins over a `pymc_core`
    that is installed next to it. It raises ImportError when neither is importable.
    """
    import importlib

    for name in RUNTIMES:
        try:
            return name, importlib.import_module(name)
        except ImportError:
            continue
    raise ImportError("no node runtime: " + " or ".join(RUNTIMES))


def _rt_import(rt: str, dotted: str, attr: str | None = None):
    """`from {rt}.{dotted} import {attr}` for the runtime that is loaded."""
    import importlib

    module = importlib.import_module(f"{rt}.{dotted}")
    return getattr(module, attr) if attr else module


def _patch_pymc_ack_compat(rt: str, log) -> None:
    """Accept modern MeshCore ACKs that have extra bytes after the CRC.

    Newer firmware sends discrete ACK payloads that are longer than 4 bytes (the 4-byte CRC
    first, then extra bytes). The AckHandler of pymc_core needs exactly 4 bytes and drops the
    rest. Thus delivery confirmations never reach MeshTerm. This function overrides the
    handler to use the first 4 bytes as the CRC. This is the same tolerant parse that
    pymc_core already uses for ACKs that are bundled in a path.

    openhop_core has this change upstream. Its `process_discrete_ack` uses the first four
    bytes of any payload of four bytes or more. Thus, under that runtime, the shim does
    nothing, because the library already does the same thing.
    """
    if rt != "pymc_core":
        log.info("compat: %s accepts >4-byte ACK payloads itself; ack shim not needed", rt)
        return
    try:
        AckHandler = _rt_import(rt, "node.handlers.ack", "AckHandler")
    except Exception as exc:  # the package layout changed, so do not change the package
        log.warning("ack compat shim not applied (%s)", exc)
        return

    async def process_discrete_ack(self, packet):
        payload = packet.payload
        self.log(f"Processing discrete ACK: payload_len={len(payload)}")
        self.log(f"ACK payload (hex): {payload.hex().upper()}")
        if len(payload) < 4:
            self.log(f"Invalid ACK length: {len(payload)} bytes (expected >= 4)")
            return None
        crc = int.from_bytes(payload[:4], "little")
        if len(payload) > 4:
            self.log(f"ACK carries {len(payload) - 4} extra byte(s): {payload[4:].hex().upper()}")
        self.log(f"Discrete ACK received: CRC={crc:08X}")
        return crc

    AckHandler.process_discrete_ack = process_discrete_ack
    log.info("compat: AckHandler patched to accept >4-byte ACK payloads")


def _wire_contact_persistence(companion, log) -> None:
    """Keep the contact list of the node when the bridge restarts.

    The ContactStore of pymc_core keeps data in memory only. Thus a restart forgot each
    contact, and DMs failed with ERR_CODE_NOT_FOUND until the node heard the peer again.
    This function loads the last snapshot at startup and saves a new snapshot after each
    change.
    """
    import json

    store = getattr(companion, "contacts", None)
    if store is None or not hasattr(store, "load_from_dicts"):
        log.warning("contact persistence not wired (no contact store)")
        return

    path = HOME / ".local" / "share" / APP_NAME / "contacts.json"

    def to_rec(c) -> dict:
        rec = {
            "public_key": c.public_key.hex(),
            "name": c.name,
            "adv_type": c.adv_type,
            "flags": c.flags,
            "out_path_len": c.out_path_len,
            "out_path": c.out_path.hex() if c.out_path else "",
            "last_advert_timestamp": c.last_advert_timestamp,
            "lastmod": c.lastmod,
            "gps_lat": c.gps_lat,
            "gps_lon": c.gps_lon,
            "sync_since": c.sync_since,
        }
        if c.last_advert_packet:
            rec["last_advert_packet"] = c.last_advert_packet.hex()
        return rec

    if path.exists():
        try:
            store.load_from_dicts(json.loads(path.read_text()))
            log.info("restored %d contact(s) from %s", store.get_count(), path)
        except Exception as exc:
            log.warning("could not restore contacts (%s)", exc)

    def save() -> None:
        try:
            path.parent.mkdir(parents=True, exist_ok=True)
            tmp = path.with_name(path.name + ".tmp")
            tmp.write_text(json.dumps([to_rec(c) for c in store.get_all()]))
            tmp.replace(path)
        except Exception as exc:
            log.warning("could not save contacts (%s)", exc)

    def wrap(name: str) -> None:
        orig = getattr(store, name, None)
        if orig is None:
            return

        def wrapped(*args, _orig=orig, **kwargs):
            result = _orig(*args, **kwargs)
            save()
            return result

        setattr(store, name, wrapped)

    for name in ("add", "add_or_overwrite", "update", "remove", "clear"):
        wrap(name)
    log.info("contact persistence at %s", path)


def _wire_channel_persistence(companion, log) -> None:
    """Keep the channel table of the node when the bridge restarts.

    The ChannelStore of the runtime keeps data in memory only, as it does for contacts. Each
    restart started with no channels, and the Channels page of MeshTerm was empty until the
    user added each channel again. This function loads the last snapshot at startup and
    saves a new snapshot after each set or clear. The frame server does a set or a clear
    through ``set_channel`` and ``remove_channel`` on the companion.
    """
    import json

    store = getattr(companion, "channels", None)
    if store is None or not hasattr(companion, "set_channel"):
        log.warning("channel persistence not wired (no channel store)")
        return

    path = HOME / ".local" / "share" / APP_NAME / "channels.json"

    def snapshot() -> list:
        out = []
        for idx in range(getattr(store, "max_channels", 40)):
            ch = store.get(idx)
            if ch is not None and ch.name:
                out.append({"idx": idx, "name": ch.name, "secret": bytes(ch.secret).hex()})
        return out

    def save() -> None:
        try:
            path.parent.mkdir(parents=True, exist_ok=True)
            tmp = path.with_name(path.name + ".tmp")
            tmp.write_text(json.dumps(snapshot()))
            tmp.replace(path)
        except Exception as exc:
            log.warning("could not save channels (%s)", exc)

    if path.exists():
        try:
            restored = 0
            for entry in json.loads(path.read_text()):
                if entry.get("name"):
                    companion.set_channel(
                        int(entry["idx"]), entry["name"], bytes.fromhex(entry["secret"])
                    )
                    restored += 1
            log.info("restored %d channel(s) from %s", restored, path)
        except Exception as exc:
            log.warning("could not restore channels (%s)", exc)

    for name in ("set_channel", "remove_channel"):
        orig = getattr(companion, name, None)
        if orig is None:
            continue

        def wrapped(*args, _orig=orig, **kwargs):
            result = _orig(*args, **kwargs)
            save()
            return result

        setattr(companion, name, wrapped)
    log.info("channel persistence at %s", path)


def _wire_trace_push(rt: str, companion, server, log) -> None:
    """Forward over-the-air trace replies to the client as PUSH_CODE_TRACE_DATA.

    pymc_core parses the TRACE packets that it receives, but it never gives them to the
    companion client. Its frame server registers each push callback except
    ``on_trace_received``, and nothing in the library fires the ``trace_received`` event.
    Thus the whole path is dead code. The only ``push_trace_data`` call is the self-hop
    shortcut in ``_cmd_send_trace_path``. As a result, ``run_trace`` of MeshTerm sends the
    trace and gets ``RESP_CODE_SENT``. Then it waits for ``TRACE_DATA`` and times out,
    while the node drops the reply with no message.

    This function bridges the gap at the one place that sees each received packet with its
    measured SNR. This is the packet-received callback of the dispatcher (``CompanionRadio``
    already installs ``_on_packet_received`` there, and it acts only on group data). The
    function wraps the callback. It runs the original callback. Then, for a TRACE payload,
    it cuts the on-air bytes again into the hash and SNR blocks that ``push_trace_data``
    needs, and it pushes the standard 0x89 frame. The ``meshcore`` client on the side of
    MeshTerm decodes exactly this frame.

    The SNR values are the difficult part. A TRACE packet keeps the route (the hop hashes)
    in its payload, and the payload does not change while the packet travels. The SNR of
    each hop is in the **path field of the packet header** instead. Each hop appends one
    signed byte (SNR × 4). The SNR block of ``push_trace_data`` and the decoder of MeshTerm
    both take the SNR values from this field. Thus the payload never grows. An earlier
    attempt waited for SNR values to appear in the payload, and it never fired.

    A received TRACE packet is not a reply only because it exists. The dispatcher is
    promiscuous. It runs the trace handler and this callback for each TRACE frame that it
    overhears and that is not a duplicate. This includes the traces of third parties. It
    also includes our own outbound trace that the first repeater echoes back (the same
    tag, and the header path is still filling). If the function pushes these packets, the
    wait of MeshTerm, which filters by tag, resolves at once against a trace that is half
    complete. Then each trace reads as a success. Thus the function uses a strict gate. It
    remembers the tag and the route of each trace that we originate (``send_trace_path_raw``
    is the one send point that the frame server calls). It pushes only a received TRACE that
    has a tag that we sent and a header that has an SNR for each hop of that route. This is
    the completed reply.

    Upstream, the clean repair is the same as for each other push. ``_on_packet_received``
    fires ``trace_received`` for a completed trace, and
    ``frame_server._setup_push_callbacks`` registers ``on_trace_received`` for
    ``push_trace_data``. Until a pymc_core release has this, this shim stays (it has the
    same purpose as :func:`_patch_pymc_ack_compat`).
    """
    import struct
    import time

    # openhop_core has this upstream. Its CompanionRadio fires ``trace_received`` for a
    # completed trace, and its frame server registers ``_on_trace_received`` for
    # ``push_trace_data``. A second bridge pushes each reply two
    # times. Thus, where the server has the handler, the shim does nothing.
    if hasattr(server, "_on_trace_received"):
        log.info("compat: %s pushes completed TRACE replies itself; trace shim not needed", rt)
        return

    try:
        PAYLOAD_TYPE_TRACE = _rt_import(rt, "protocol.constants", "PAYLOAD_TYPE_TRACE")
    except Exception as exc:  # the package layout changed, so do not change the package
        log.warning("trace push not wired (%s)", exc)
        return

    dispatcher = getattr(getattr(companion, "node", None), "dispatcher", None)
    if dispatcher is None or not hasattr(dispatcher, "set_packet_received_callback"):
        log.warning("trace push not wired (no dispatcher)")
        return

    orig_send = getattr(companion, "send_trace_path_raw", None)
    if orig_send is None:
        log.warning("trace push not wired (no send_trace_path_raw)")
        return

    # tag -> (flags, path_bytes, deadline). Only a trace that we originated, and that walked
    # the full route, becomes a push. The entries expire. Thus a trace that nobody answered
    # cannot match a tag that the node uses again later.
    SENT_TTL_S = 90.0
    sent: dict = {}

    def _prune(now: float) -> None:
        for tag in [t for t, (_f, _p, dl) in sent.items() if dl < now]:
            sent.pop(tag, None)

    async def send_trace_path_raw(tag, auth_code, flags, path_bytes, _orig=orig_send):
        now = time.monotonic()
        _prune(now)
        key = tag & 0xFFFFFFFF
        sent[key] = (flags, bytes(path_bytes), now + SENT_TTL_S)
        log.info(
            "trace sent tag=0x%08X flags=0x%02x path=%s",
            key, flags, bytes(path_bytes).hex().upper(),
        )
        return await _orig(tag, auth_code, flags, path_bytes)

    companion.send_trace_path_raw = send_trace_path_raw

    prev = getattr(companion, "_on_packet_received", None)

    def push_reply(pkt) -> None:
        payload = bytes(pkt.payload or b"")
        if len(payload) < 9:
            return
        tag, auth_code, flags = struct.unpack_from("<IIB", payload, 0)
        rec = sent.get(tag & 0xFFFFFFFF)
        if rec is None:
            return  # a trace that we did not send (a third party that we overheard): ignore it
        _sent_flags, _sent_path, _dl = rec
        hw = 1 << (flags & 0x03)
        route = payload[9:]                 # the hop hashes, which do not change in flight
        hops = len(route) // hw if hw else 0
        # The SNR of each hop is in the header path: one signed byte (SNR*4) for each hop.
        header_path = bytes(getattr(pkt, "path", b"") or b"")
        snr_count = getattr(pkt, "path_len", 0) & 0x3F
        header_snrs = header_path[:snr_count]
        log.info(
            "trace rx tag=0x%08X hops=%d hdr_snrs=%d payload=%s hdrpath=%s",
            tag, hops, len(header_snrs), payload.hex().upper(), header_path.hex().upper(),
        )
        if hops == 0 or len(header_snrs) < hops:
            return  # the trace is still in flight: not all hops have reported an SNR
        path_snrs = header_snrs[:hops]
        snr = pkt.get_snr() if hasattr(pkt, "get_snr") else getattr(pkt, "_snr", 0.0)
        final_snr_byte = max(-128, min(127, int(round((snr or 0.0) * 4)))) & 0xFF
        sent.pop(tag & 0xFFFFFFFF, None)  # one reply for each trace
        server.push_trace_data(
            hops * hw, flags, tag, auth_code, route, path_snrs, final_snr_byte
        )
        log.info("pushed TRACE_DATA tag=0x%08X hops=%d width=%d", tag, hops, hw)

    async def on_packet(pkt) -> None:
        if prev is not None:
            await prev(pkt)
        try:
            if pkt.get_payload_type() != PAYLOAD_TYPE_TRACE:
                return
        except Exception:
            return
        try:
            push_reply(pkt)
        except Exception as exc:
            log.warning("trace push failed: %s", exc)

    dispatcher.set_packet_received_callback(on_packet)
    log.info("compat: completed TRACE replies push PUSH_CODE_TRACE_DATA to the client")


def _wire_rx_log_push(companion, server, log) -> None:
    """Send the promiscuous RX packet log to the client (PUSH_CODE_LOG_RX_DATA).

    This is the same gap in the wiring as the trace push, in a place next to it. The live
    feed of MeshTerm comes from RX_LOG_DATA events. Each overheard frame has its SNR and
    RSSI. The frame server of pymc_core has ``push_rx_raw`` (which builds the 0x88 push),
    but its ``_setup_push_callbacks`` never registers ``on_rx_log_data``, so nothing calls
    it. Adverts still show, because they have their own decoded push (``advert_received``,
    which is wired). Everything else that we overhear travels only through this raw log, and
    it is dropped. This includes requests, channel text that we cannot decrypt, and acks.
    Thus the feed looks as if it has only adverts.

    This function subscribes to the raw-packet feed of the dispatcher directly, and it
    pushes each frame. The path through the dispatcher (not ``on_rx_log_data``) does not
    meet the ``clear_push_callbacks()`` that the frame server calls for each connection.
    That feed fires before the removal of duplicates, so the client sees each repeat. A
    live feed needs this.

    Upstream, the repair is again the same as for the wired pushes: register
    ``on_rx_log_data`` for ``push_rx_raw`` in ``frame_server._setup_push_callbacks``.
    """
    dispatcher = getattr(getattr(companion, "node", None), "dispatcher", None)
    if dispatcher is None or not hasattr(dispatcher, "add_raw_packet_subscriber"):
        log.warning("rx-log push not wired (no dispatcher)")
        return
    if not hasattr(server, "push_rx_raw"):
        log.warning("rx-log push not wired (no push_rx_raw)")
        return

    async def on_raw(pkt, data, _analysis=None) -> None:
        try:
            snr = getattr(pkt, "snr", getattr(pkt, "_snr", 0.0))
            rssi = getattr(pkt, "rssi", getattr(pkt, "_rssi", 0))
            server.push_rx_raw(snr or 0.0, rssi or 0, bytes(data))
        except Exception as exc:
            log.warning("rx-log push failed: %s", exc)

    dispatcher.add_raw_packet_subscriber(on_raw)
    log.info("compat: raw RX packets now push PUSH_CODE_LOG_RX_DATA to the client")


def run_bridge(host: str, port: int, node_name: str) -> int:
    import asyncio
    import logging

    logging.basicConfig(
        level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s"
    )
    log = logging.getLogger(APP_NAME)

    try:
        rt, runtime = _import_runtime()
    except ImportError:
        sys.stderr.write(
            "This must run under a Python that has openhop_core or pymc_core. Run the "
            "menu instead (it finds the right one), or set MESHTERM_PYMC_PYTHON.\n"
        )
        return 2
    try:
        import importlib.metadata as _md

        rt_version = _md.version(rt.replace("_", "-"))
    except Exception:
        rt_version = getattr(runtime, "__version__", "?")
    log.info("node runtime %s %s (%s)", rt, rt_version, sys.executable)

    _patch_pymc_ack_compat(rt, log)

    CompanionFrameServer = _rt_import(rt, "companion", "CompanionFrameServer")
    CompanionRadio = _rt_import(rt, "companion", "CompanionRadio")

    # Prefer the hackergadgets meshcore_console helpers. They have the correct board pins and
    # the same node identity that the GUI uses. Fall back to a direct setup for a generic
    # install (a Waveshare HAT with a runtime from pip), or for the bridge's own venv on
    # a host where the GUI package is older than the runtime.
    identity = None
    radio = None
    try:
        from meshcore_console.meshcore import runtime as _mc_runtime
        from meshcore_console.meshcore.config import load_hardware_config_from_env
        from meshcore_console.meshcore.paths import identity_key_path

        # The name of the helper comes from the runtime that the GUI was built against.
        # Each spelling returns the same five-tuple.
        _import_stack = getattr(_mc_runtime, "import_openhop_core", None) or getattr(
            _mc_runtime, "import_pymc_core"
        )
        create_radio = _mc_runtime.create_radio
        SX1262Radio, _ES, _ESub, _MN, LocalIdentity = _import_stack()
        hw = load_hardware_config_from_env()
        log.info("radio config %s", hw.to_log_string())
        radio = create_radio(SX1262Radio, hw, log.info)
        # This is the same preamble rule as the direct path below. Unless the environment
        # names a preamble, follow MeshCore (32 symbols up to SF8, 16 above). Thus the radio
        # hears the mesh. begin() reads the value, so the change is in time.
        if "MESHCORE_PREAMBLE_LENGTH" not in os.environ and hasattr(radio, "preamble_length"):
            radio.preamble_length = 32 if hw.spreading_factor <= 8 else 16
        key_path = identity_key_path()
        if key_path.exists():
            identity = LocalIdentity(key_path.read_bytes())
            log.info("using existing node identity at %s", key_path)
        radio_config = {
            "frequency": hw.frequency, "bandwidth": hw.bandwidth,
            "spreading_factor": hw.spreading_factor, "coding_rate": hw.coding_rate,
            "tx_power": hw.tx_power,
        }
    except ImportError:
        log.info("meshcore_console not present — using direct %s defaults", rt)
        import inspect

        SX1262Radio = _rt_import(rt, "hardware.sx1262_wrapper", "SX1262Radio")
        LocalIdentity = _rt_import(rt, "protocol.identity", "LocalIdentity")

        # The uConsole AIO preset. Each value is the same as the default of the
        # HardwareRadioConfig of meshcore_console. The defaults of the runtime itself are
        # the defaults of a Waveshare HAT (TX-enable on pin 6, no DIO2 RF switch, no DIO3
        # TCXO, 868 MHz), and on the AIO they make the radio deaf. The environment can
        # override each value, with the same MESHCORE_* names that the GUI reads.
        env = os.environ.get

        def _int(name: str, default: int) -> int:
            return int(env(name, default))

        def _bool(name: str, default: bool) -> bool:
            raw = env(name)
            return default if raw is None else raw.strip().lower() in ("1", "true", "yes", "on")

        def _pins(name: str) -> list[int] | None:
            raw = env(name, "").strip()
            return [int(p) for p in raw.split(",") if p.strip()] or None

        wanted = {
            "bus_id": _int("MESHCORE_BUS_ID", 1),
            "cs_id": _int("MESHCORE_CS_ID", 0),
            "cs_pin": _int("MESHCORE_CS_PIN", -1),
            "gpio_chip": _int("MESHCORE_GPIO_CHIP", 0),
            "use_gpiod_backend": _bool("MESHCORE_USE_GPIOD_BACKEND", False),
            "reset_pin": _int("MESHCORE_RESET_PIN", 25),
            "busy_pin": _int("MESHCORE_BUSY_PIN", 24),
            "irq_pin": _int("MESHCORE_IRQ_PIN", 26),
            "txen_pin": _int("MESHCORE_TXEN_PIN", -1),
            "rxen_pin": _int("MESHCORE_RXEN_PIN", -1),
            "en_pins": _pins("MESHCORE_EN_PINS"),
            "frequency": _int("MESHCORE_FREQUENCY", 910_525_000),
            "tx_power": _int("MESHCORE_TX_POWER", 22),
            "spreading_factor": _int("MESHCORE_SPREADING_FACTOR", 7),
            "bandwidth": _int("MESHCORE_BANDWIDTH", 62_500),
            "coding_rate": _int("MESHCORE_CODING_RATE", 5),
            # The rule of MeshCore (RadioLibWrapper::preambleLengthForSF): 32 symbols up
            # to SF8, 16 above. The receiver must use the same value as the sender. The
            # SX1262 waits for the sync word only for approximately the length of the
            # preamble that it expects. Thus the old fixed value of 12, against a mesh that
            # sent 32, decoded almost nothing.
            "preamble_length": _int(
                "MESHCORE_PREAMBLE_LENGTH",
                32 if _int("MESHCORE_SPREADING_FACTOR", 7) <= 8 else 16,
            ),
            "is_waveshare": _bool("MESHCORE_IS_WAVESHARE", False),
            "use_dio2_rf": _bool("MESHCORE_USE_DIO2_RF", True),
            "use_dio3_tcxo": _bool("MESHCORE_USE_DIO3_TCXO", True),
        }
        # Pass only the arguments that the constructor of this runtime knows. pymc_core 1.0.x
        # does not have some of them (en_pins, the gpiod backend) and refuses the call.
        accepted = inspect.signature(SX1262Radio.__init__).parameters
        kwargs = {k: v for k, v in wanted.items() if k in accepted and v is not None}
        log.info("radio config %s", " ".join(f"{k}={v}" for k, v in kwargs.items()))
        radio = SX1262Radio(**kwargs)
        radio_config = {
            "frequency": kwargs["frequency"], "bandwidth": kwargs["bandwidth"],
            "spreading_factor": kwargs["spreading_factor"],
            "coding_rate": kwargs["coding_rate"], "tx_power": kwargs["tx_power"],
        }
        # Use the identity of the GUI first, in the place where the meshcore-uconsole stack
        # keeps it. Thus a bridge that runs from its own venv is the same node as the GUI.
        # Then use the bridge's own identity.
        xdg = Path(env("XDG_DATA_HOME") or (HOME / ".local" / "share"))
        for key_path in (
            xdg / "meshcore-uconsole" / "identity.key",
            HOME / ".local" / "share" / APP_NAME / "identity.key",
        ):
            if key_path.exists():
                identity = LocalIdentity(key_path.read_bytes())
                log.info("using existing node identity at %s", key_path)
                break

    if identity is None:
        # There is no stored identity anywhere. Make one and save it. Thus the node keeps a
        # stable identity when it restarts.
        _LI = _rt_import(rt, "protocol.identity", "LocalIdentity")

        tmp = _LI()
        seed = tmp.get_signing_key_bytes()
        save_path = HOME / ".local" / "share" / APP_NAME / "identity.key"
        save_path.parent.mkdir(parents=True, exist_ok=True)
        save_path.write_bytes(seed)
        identity = _LI(seed)
        log.info("generated a new node identity at %s", save_path)

    async def _main() -> int:
        # radio.begin() with backoff. A session that just closed can still hold the GPIO
        # lines for a short time.
        import time

        for attempt in range(4):
            log.info("radio.begin() attempt %d/4", attempt + 1)
            try:
                if not radio.begin():
                    raise RuntimeError("radio.begin() returned False")
                break
            except SystemExit:
                if attempt == 3:
                    raise RuntimeError("GPIO pins still busy after retries")
                try:
                    radio.cleanup()
                except Exception:
                    pass
                time.sleep(2.0 * (attempt + 1))
        log.info("radio ready")

        companion = CompanionRadio(radio, identity, node_name=node_name, radio_config=radio_config)
        await companion.start()
        _wire_contact_persistence(companion, log)
        _wire_channel_persistence(companion, log)
        pub = companion.get_public_key()
        log.info("node identity %s… (hash 0x%02x, name %r)", pub.hex()[:16], pub[0], node_name)

        server = CompanionFrameServer(
            bridge=companion,
            companion_hash=f"{pub[0]:02x}",
            port=port,
            bind_address=host,
            local_hash=pub[0],
            device_model="pyMC-spi-bridge",
        )
        await server.start()
        _wire_trace_push(rt, companion, server, log)
        _wire_rx_log_push(companion, server, log)
        log.info("READY — connect with:  meshterm --tcp %s:%d", host, port)

        import signal

        stop = asyncio.Event()
        loop = asyncio.get_running_loop()
        for sig in (signal.SIGINT, signal.SIGTERM):
            try:
                loop.add_signal_handler(sig, stop.set)
            except NotImplementedError:
                pass
        try:
            await stop.wait()
        finally:
            log.info("shutting down…")
            try:
                await server.stop()
                await companion.stop()
            except Exception:
                pass
        return 0

    try:
        return asyncio.run(_main())
    except KeyboardInterrupt:
        return 0
    except Exception as exc:  # show a clean message, not a long traceback
        log.error("bridge failed: %s", exc)
        return 1


# --------------------------------------------------------------------------- #
# Entry point
# --------------------------------------------------------------------------- #


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Expose an SPI-attached MeshCore radio to MeshTerm over TCP."
    )
    ap.add_argument("--check", action="store_true", help="run prerequisite checks and exit")
    ap.add_argument("--run", action="store_true", help="run the bridge now (finds the runtime)")
    ap.add_argument("--run-bridge", action="store_true", help=argparse.SUPPRESS)
    ap.add_argument("--host", default=DEFAULT_HOST)
    ap.add_argument("--port", type=int, default=DEFAULT_PORT)
    ap.add_argument("--name", default=DEFAULT_NODE_NAME)
    args = ap.parse_args()

    # Internal: the script already ran again under the interpreter of the node runtime.
    if args.run_bridge:
        return run_bridge(args.host, args.port, args.name)

    if args.check:
        Preflight().print()
        return 0

    if args.run:
        pre = Preflight()
        pre.print()
        action_run_now(pre)
        return 0

    return menu()


if __name__ == "__main__":
    raise SystemExit(main())