meshcore_reachability.py

#!/usr/bin/env python3
import asyncio
import argparse
import threading
import time
from meshcore import MeshCore
from meshcore.events import EventType
from meshcoredecoder import MeshCoreDecoder
from meshcoredecoder.types.enums import PayloadType
from meshcoredecoder.utils.enum_names import (
    get_route_type_name,
    get_payload_type_name,
    get_device_role_name,
)
import sqlite3
import os
from datetime import datetime
import json
import random
from urllib.parse import urlencode
import base64
import io

import qrcode


# --- Database helper functions --------------------------------------------


def init_db(db_path: str):
    """Initialize the reachability database schema (nodes, paths, traces)."""
    init_db_needed = not os.path.exists(db_path)
    conn = sqlite3.connect(db_path, check_same_thread=False)
    if init_db_needed:
        c = conn.cursor()
        # stores nodes from adverts
        c.execute(
            """CREATE TABLE IF NOT EXISTS nodes (
            public_key CHAR(64) PRIMARY KEY,
            name VARCHAR(50),
            role VARCHAR(5),
            latitude REAL,
            longitude REAL,
            lastpath TEXT,
            lastmod TEXT
        )"""
        )
        # stores partial and full paths from adverts if a trace was executed
        c.execute(
            """CREATE TABLE IF NOT EXISTS paths (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            path TEXT,
            target_node TEXT,
            count INTEGER DEFAULT 1,
            lastmod TEXT
        )"""
        )
        # stores SNR results of trace executions; snr_values is NULL if trace failed
        c.execute(
            """CREATE TABLE IF NOT EXISTS traces (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            path_id INTEGER,
            timestamp TEXT,
            snr_values TEXT,
            FOREIGN KEY(path_id) REFERENCES paths(id)
        )"""
        )
        # convenience view for quickly joining paths and traces with node names
        c.execute(
            """CREATE VIEW IF NOT EXISTS pathtraces AS
                SELECT
                    p.id         AS path_id,
                    t.id         AS trace_id,
                    p.path,
                    p.count,
                    t.snr_values,
                    COALESCE(
                        (
                            SELECT n.name
                            FROM nodes AS n
                            WHERE n.public_key = p.target_node
                        ),
                        (
                            SELECT GROUP_CONCAT(n.name, ', ')
                            FROM nodes AS n
                            WHERE SUBSTR(n.public_key, 1, 2)
                                = SUBSTR(p.target_node, 1, 2)
                        )
                        ) AS node_name,
                    t.timestamp
                FROM paths AS p
                JOIN traces AS t
                    ON p.id = t.path_id;
            """
        )
        conn.commit()
    return conn


def formatPath(path_elems: list, default=""):
    """Return a comma-separated representation of a path element list."""
    if not path_elems:
        return default
    return ",".join(path_elems)


def write_node_to_db(
    conn: sqlite3.Connection,
    public_key,
    name,
    role,
    latitude,
    longitude,
    lastpath,
):
    """Insert or update a node entry in the ``nodes`` table."""
    c = conn.cursor()
    timestamp = datetime.now().isoformat(" ", "seconds")
    c.execute("SELECT public_key FROM nodes WHERE public_key = ?", (public_key,))
    result = c.fetchone()
    if result is None:
        c.execute(
            "INSERT INTO nodes (public_key, name, role, latitude, longitude, lastpath, lastmod) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (public_key, name, role, latitude, longitude, lastpath, timestamp),
        )
    else:
        update_fields = []
        update_values = []
        if name is not None:
            update_fields.append("name = ?")
            update_values.append(name)
        if role is not None:
            update_fields.append("role = ?")
            update_values.append(role)
        if latitude is not None:
            update_fields.append("latitude = ?")
            update_values.append(latitude)
        if longitude is not None:
            update_fields.append("longitude = ?")
            update_values.append(longitude)
        if lastpath is not None:
            update_fields.append("lastpath = ?")
            update_values.append(lastpath)
        update_fields.append("lastmod = ?")
        update_values.append(timestamp)
        update_values.append(public_key)
        sql = "UPDATE nodes SET " + ", ".join(update_fields) + " WHERE public_key = ?"
        c.execute(sql, tuple(update_values))
    conn.commit()


# --- Combined thread: collect adverts and process paths --------------------


def advert_and_path_thread(
    port: str,
    host: str,
    ble: str,
    db_path: str,
    stop_event: threading.Event,
    home_latitude: float | None = None,
    home_longitude: float | None = None,
    checkradius_km: float | None = None,
):
    """Collect MeshCore adverts and test reachability of the source via send_msg and ACK."""

    async def _run():
        mc: MeshCore | None = None
        subscription = None
        # track the last full hour when we sent our own advert
        last_advert_hour = datetime.now().hour

        def _distance_km(lat1, lon1, lat2, lon2):
            """Return great-circle distance between two coordinates in kilometers."""
            if None in (lat1, lon1, lat2, lon2):
                return None
            from math import radians, sin, cos, sqrt, atan2

            R = 6371.0
            dlat = radians(lat2 - lat1)
            dlon = radians(lon2 - lon1)
            a = sin(dlat / 2) ** 2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon / 2) ** 2
            c = 2 * atan2(sqrt(a), sqrt(1 - a))
            return R * c

        async def handle_rf_packet(event):
            """Handle RX_LOG_DATA events and trigger reachability checks for adverts."""
            nonlocal mc, subscription, last_advert_hour

            packet = event.payload
            if isinstance(packet, dict) and "payload" in packet:
                packet = MeshCoreDecoder.decode(packet["payload"])
                # debug helpers for manual inspection:
                # print(f"  Route Type: {get_route_type_name(packet.route_type)}")
                # print(f"  Payload Type: {get_payload_type_name(packet.payload_type)}")
                # print(f"  Message Hash: {packet.message_hash}")
                # print(f"  Message Path: {packet.path}")

                if packet.payload_type == PayloadType.Advert and packet.payload.get("decoded"):

                    # Temporarily unsubscribe from further events while processing this advert
                    mc.unsubscribe(subscription)

                    advert = packet.payload["decoded"]
                    name = advert.app_data.get("name")
                    roleval = advert.app_data.get("device_role")
                    role = get_device_role_name(roleval) if roleval is not None else None
                    latitude = None
                    longitude = None
                    print("")
                    print(
                        f"Received Advert from {role} {name}, Pubkey-Prefix: {advert.public_key[:8]} via Path {formatPath(packet.path, '<direct>')}"
                    )
                    if advert.app_data.get("location"):
                        location = advert.app_data["location"]
                        latitude = location.get("latitude")
                        longitude = location.get("longitude")
                    write_node_to_db(
                        conn,
                        advert.public_key,
                        name,
                        role,
                        latitude,
                        longitude,
                        formatPath(packet.path),
                    )

                    # Radius limitation for additional processing
                    inside_radius = True
                    dist = None
                    if (
                        checkradius_km is not None
                        and home_latitude is not None
                        and home_longitude is not None
                    ):
                        if latitude is not None and longitude is not None:
                            dist = _distance_km(
                                home_latitude,
                                home_longitude,
                                latitude,
                                longitude,
                            )
                            inside_radius = dist is None or dist <= checkradius_km
                        else:
                            # no location info -> treat as inside to not lose nodes
                            inside_radius = True

                    if dist is not None:
                        print(
                            f"Distance from home: {dist:.2f} km (radius limit: {checkradius_km} km) -> {'inside' if inside_radius else 'outside'}"
                        )

                    # Send out our own advert once per full hour so peering chat nodes can respond
                    current_hour = datetime.now().hour
                    if last_advert_hour is None or current_hour != last_advert_hour:
                        last_advert_hour = current_hour
                        await mc.commands.send_advert(True)
                        print("Sent hourly flood advert so peers can respond upon requests")
                        await asyncio.sleep(3)

                    # Process the foreign advert only if the node is inside the radius (or has no geo-location)
                    if inside_radius:
                        await process_advert(role, advert.public_key, packet.path)
                    else:
                        print("Skipping advert processing for node outside radius constraint")

                    # Re-subscribe to RX log events
                    subscription = mc.subscribe(EventType.RX_LOG_DATA, handle_rf_packet)
                else:
                    # A non-advert packet; just print a dot so we see ongoing traffic
                    print(".", end="")

        async def process_advert(role, public_key, path):
            """Analyse the path of an advert, run traces, and test reverse messaging for chat nodes."""
            nonlocal mc

            try:
                # Normalise and reverse the given path as seen from our local node
                path_elems = []
                if path:
                    path_elems = list(reversed(path))

                # For non-chat nodes add the advertising node as final hop
                if role != "Chat Node":
                    path_elems += [public_key]

                # Check all path prefixes and, if needed, run traces for them
                prefix = []
                prefix_short = []
                for elem in path_elems:
                    prefix.append(elem)
                    prefix_short.append(elem[:2])
                    path_id, now_ts = _ensure_path_record(conn, prefix)

                    needsNewTrace, lastTraceFailed = _needs_new_trace(conn, path_id, now_ts)
                    if not needsNewTrace and lastTraceFailed:
                        # The prefix path was unreachable previously, skip processing of longer prefixes
                        return

                    if needsNewTrace:
                        # Prefix "92 c0 a1" becomes a roundtrip-like path: 92,c0,a1,c0,92
                        if len(prefix_short) == 1:
                            full_path = list(prefix_short)
                        else:
                            full_path = list(prefix_short) + list(reversed(prefix_short[:-1]))

                        full_path_flat = formatPath(full_path)
                        snr_values = await _execute_trace_for_path_async(mc, full_path_flat)
                        if not snr_values:
                            print("Second trace attempt:")
                            snr_values = await _execute_trace_for_path_async(mc, full_path_flat)
                        _insert_trace_result(conn, path_id, snr_values)
                        if not snr_values:
                            # This prefix path could not be reached, do not try longer prefixes
                            return

                # For non-chat nodes we are done after tracing all path prefixes
                if role != "Chat Node":
                    return

                # For chat nodes try to send a message and wait for an ACK once the path is traceable
                assumed_out_path = []
                full_path = [public_key]
                if path:
                    assumed_out_path = list(reversed(path))
                    full_path = assumed_out_path + [public_key]

                # Check if we already tested this full path
                path_id, now_ts = _ensure_path_record(conn, full_path)
                needsNewTrace, lastTraceFailed = _needs_new_trace(conn, path_id, now_ts)

                if not needsNewTrace:
                    return

                # Try to contact Chat Node
                contact = mc.get_contact_by_key_prefix(public_key)
                if contact is None:
                    # Refresh contact list and try again
                    await mc.commands.get_contacts()
                    contact = mc.get_contact_by_key_prefix(public_key)
                    if contact is None:
                        print(
                            "Your contact list seems to be filled totally; you need to remove unrequired records to process new Chat Node adverts"
                        )
                        return

                saved_out_path = contact["out_path"]

                # Temporarily change out path to assumed out path from advert
                if not assumed_out_path:
                    op_res = await mc.commands.reset_path(contact)
                    if op_res.type == EventType.ERROR:
                        return
                else:
                    op_res = await mc.commands.change_contact_path(
                        contact, path="".join(assumed_out_path)
                    )
                    if op_res.type == EventType.ERROR:
                        return

                # Send a message and wait for the confirmation
                send_event = await mc.commands.send_msg(
                    dst=contact,
                    msg="Received your advert, testing reverse connection",
                )
                if send_event.type == EventType.ERROR:
                    print(f"Error {send_event}")
                    await mc.commands.change_contact_path(contact, path=saved_out_path)
                    return

                # Wait for ACK
                exp_ack = send_event.payload["expected_ack"].hex()
                timeout = 0
                min_timeout = 8
                timeout = (
                    send_event.payload["suggested_timeout"] / 1000 * 1.2
                    if timeout == 0
                    else timeout
                )
                timeout = timeout if timeout > min_timeout else min_timeout
                send_resp_event = await mc.wait_for_event(
                    EventType.ACK,
                    attribute_filters={"code": exp_ack},
                    timeout=timeout,
                )
                if send_resp_event:
                    print(f"Reverse message to {contact['adv_name']} was confirmed (ACK)")
                    _insert_trace_result(conn, path_id, "ACK")
                else:
                    print(f"Reverse message to {contact['adv_name']} failed")
                    # Reset path to previous setting
                    if not saved_out_path:
                        await mc.commands.reset_path(contact)
                    else:
                        await mc.commands.change_contact_path(
                            contact, path=saved_out_path
                        )

            except Exception as e:
                print(f"[process_advert] Error: {e}")

            finally:
                # Give the event loop some breathing room between adverts
                await asyncio.sleep(0.1)

        try:
            # Open the database for this thread (schema already initialized in main)
            conn = sqlite3.connect(db_path, check_same_thread=False)

            if port:
                print(f"[collector] Connecting to {port}...")
                mc = await MeshCore.create_serial(port, 115200)
            elif host:
                print(f"[collector] Connecting to {host}...")
                mc = await MeshCore.create_tcp(host, 5000)
            elif ble:
                print(f"[collector] Connecting to {ble}...")
                mc = await MeshCore.create_ble(ble)


            # Forget all repeaters that were not updated in the last 2 days to free space
            days_ago_ts = int(time.time() - 2 * 24 * 60 * 60)

            # Load contacts and cleanup stale repeaters
            print(f"[collector] Fetch contacts and cleanup repeaters")
            contacts = await mc.commands.get_contacts()
            if contacts and contacts.payload:
                for contact in contacts.payload:
                    if (
                        contacts.payload[contact]["type"] == 2
                        and contacts.payload[contact]["lastmod"] < days_ago_ts
                    ):
                        await mc.commands.remove_contact(contact)

            print("[collector] Waiting for log data")

            # Subscribe to RX log data events
            subscription = mc.subscribe(EventType.RX_LOG_DATA, handle_rf_packet)

            # Main loop: keep the collector running until the stop event is set
            while not stop_event.is_set():
                await asyncio.sleep(2)

        except Exception as e:
            print(f"[collector] Error: {e}")

        finally:
            if mc is not None:
                await mc.disconnect()
            conn.close()

    asyncio.run(_run())


# --- Helper functions for storing paths and traces ------------------------


def _ensure_path_record(conn: sqlite3.Connection, path_elements):
    """Find or insert a path in ``paths`` and bump its usage counter.

    ``path_elements`` is a list of node key prefixes, e.g. ``['92']`` or ``['92', 'c0']``.
    Returns ``(path_id, now_ts)``.
    """
    c = conn.cursor()
    now_ts = datetime.now().isoformat(" ", "seconds")
    path_str = formatPath(path_elements)
    c.execute("SELECT id, count FROM paths WHERE path = ?", (path_str,))
    row = c.fetchone()
    if row:
        pid, cnt = row
        cnt_new = cnt + 1
        c.execute(
            "UPDATE paths SET count = ?, lastmod = ? WHERE id = ?",
            (cnt_new, now_ts, pid),
        )
    else:
        c.execute(
            "INSERT INTO paths (path, target_node, count, lastmod) VALUES (?, ?, ?, ?)",
            (path_str, path_elements[-1], 1, now_ts),
        )
        pid = c.lastrowid
    conn.commit()
    return pid, now_ts


def _needs_new_trace(conn: sqlite3.Connection, path_id: int, now_ts: str) -> tuple[bool, bool]:
    """Return whether a new trace for ``path_id`` is required and whether the last one failed.

    First return value (``needs_new_trace``):
    * ``True`` if no trace entry exists yet
    * ``True`` if the last trace is older than 72 hours

    Second return value (``last_trace_failed``):
    * ``True`` if the last trace failed (``snr_values`` is ``NULL``)
    """
    c = conn.cursor()
    c.execute(
        "SELECT timestamp, snr_values FROM traces WHERE path_id = ? ORDER BY id DESC LIMIT 1",
        (path_id,),
    )
    row = c.fetchone()
    if not row:
        return True, False
    last_ts = datetime.fromisoformat(row[0])
    now_dt = datetime.fromisoformat(now_ts)
    delta = now_dt - last_ts
    # do not re-trace paths younger than 72 hours
    return delta.total_seconds() > 3600 * 72, row[1] is None


async def _execute_trace_for_path_async(mc: MeshCore, full_path):
    """Execute a trace for the comma-separated path and return SNR values or ``None``.

    Returns a comma-separated string of per-hop SNR values in case of success,
    otherwise ``None``.
    """
    await asyncio.sleep(1)

    try:
        tag = random.randint(1, 0xFFFFFFFF)
        result = await mc.commands.send_trace(path=full_path, tag=tag)

        if result.type == EventType.ERROR:
            print(
                f"Failed to send trace packet with path={full_path}: {result.payload.get('reason', 'unknown error')}"
            )
            return None
        if result.type != EventType.MSG_SENT:
            print(f"Failed to send trace packet with path={full_path}")
            return None

        print(
            f"Sent trace packet with path={full_path} and tag={tag}, waiting for response ..."
        )

        event = await mc.wait_for_event(
            EventType.TRACE_DATA,
            attribute_filters={"tag": tag},
            timeout=15,
        )
        if not event:
            print(
                f"No trace response received with path={full_path} and tag={tag} within timeout"
            )
            return None

        trace = event.payload
        print(f"Trace data received for path={full_path} and tag={tag}:")
        # Debug fields if needed:
        # print(f"  Tag: {trace['tag']}")
        # print(f"  Flags: {trace.get('flags', 0)}")
        # print(f"  Path Length: {trace.get('path_len', 0)}")

        if not trace.get("path"):
            return None

        snr_list = []
        for node in trace["path"]:
            if "snr" not in node:
                return None
            print(f"  {node}")
            snr_list.append(str(node["snr"]))

        return ",".join(snr_list)

    except Exception as e:
        print(f"Trace exception: {e}")
        return None


def _insert_trace_result(
    conn: sqlite3.Connection, path_id: int, snr_values: str | None
):
    """Insert a new entry into the ``traces`` table for a path trace run."""
    c = conn.cursor()
    now_ts = datetime.now().isoformat(" ", "seconds")
    c.execute(
        "INSERT INTO traces (path_id, timestamp, snr_values) VALUES (?, ?, ?)",
        (path_id, now_ts, snr_values),
    )
    conn.commit()


# --- Thread 3: Visualisation (Dash + Leaflet) -----------------------------


def create_dash_app_from_db(
    db_path,
    maptiler_api_key: str | None = None,
    home_latitude: float | None = None,
    home_longitude: float | None = None,
    checkradius_km: float | None = None,
):
    """Create and configure the Dash app that visualises the reachability data."""
    mclink_qr_cache: dict[str, str] = {}
    statistics: dict[str, str] = {
        "chatnodes_rcvd_adverts": "",
        "chatnodes_reachable": "",
        "repeaters_rcvd_adverts": "",
        "repeaters_reachable": "",
        "roomservers_rcvd_adverts": "",
        "roomservers_reachable": "",
        "checked_paths": "",
    }
    import dash
    from dash import html, dcc, Output, Input, State
    import dash_leaflet as dl
    import sqlite3

    def load_node_meta_from_db():
        """Load node metadata and basic statistics from the SQLite database."""
        conn = sqlite3.connect(db_path, check_same_thread=False)
        c = conn.cursor()
        c.execute(
            """SELECT
                n.public_key,
                n.name,
                n.role,
                n.latitude,
                n.longitude,
                n.lastpath,
                n.lastmod,
                EXISTS(
                    SELECT 1
                    FROM paths AS p
                    JOIN traces AS t ON t.path_id = p.id
                    WHERE p.target_node = n.public_key
                    AND t.snr_values IS NOT NULL
                ) AS reachable
            FROM nodes AS n"""
        )
        rows = c.fetchall()

        # derive statistics from the DB content
        # Chat Nodes
        c.execute("SELECT COUNT(*) FROM nodes WHERE role = 'Chat Node'")
        statistics["chatnodes_rcvd_adverts"] = c.fetchone()[0]
        c.execute(
            """SELECT COUNT(*)
            FROM nodes AS n
            WHERE n.role = 'Chat Node'
            AND EXISTS (
                SELECT 1
                FROM paths AS p
                JOIN traces AS t ON t.path_id = p.id
                WHERE p.target_node = n.public_key
                AND t.snr_values IS NOT NULL
            )"""
        )
        statistics["chatnodes_reachable"] = c.fetchone()[0]

        # Repeaters
        c.execute("SELECT COUNT(*) FROM nodes WHERE role = 'Repeater'")
        statistics["repeaters_rcvd_adverts"] = c.fetchone()[0]
        c.execute(
            """SELECT COUNT(*)
            FROM nodes AS n
            WHERE n.role = 'Repeater'
            AND EXISTS (
                SELECT 1
                FROM paths AS p
                JOIN traces AS t ON t.path_id = p.id
                WHERE p.target_node = n.public_key
                AND t.snr_values IS NOT NULL
            )"""
        )
        statistics["repeaters_reachable"] = c.fetchone()[0]

        # Room Servers
        c.execute("SELECT COUNT(*) FROM nodes WHERE role = 'Room Server'")
        statistics["roomservers_rcvd_adverts"] = c.fetchone()[0]
        c.execute(
            """SELECT COUNT(*)
            FROM nodes AS n
            WHERE n.role = 'Room Server'
            AND EXISTS (
                SELECT 1
                FROM paths AS p
                JOIN traces AS t ON t.path_id = p.id
                WHERE p.target_node = n.public_key
                AND t.snr_values IS NOT NULL
            )"""
        )
        statistics["roomservers_reachable"] = c.fetchone()[0]

        # Total number of checked paths (including sub-paths) = number of entries in ``paths``
        c.execute("SELECT COUNT(*) FROM paths")
        statistics["checked_paths"] = c.fetchone()[0]

        conn.close()

        node_meta_local = {}
        for row in rows:
            node_meta_local[row[0]] = {
                "public_key": row[0],
                "name": row[1],
                "role": row[2],
                "latitude": row[3],
                "longitude": row[4],
                "lastpath": row[5],
                "lastmod": row[6],
                "reachable": row[7],
            }
        return node_meta_local

    node_meta = load_node_meta_from_db()

    def node_label(n):
        """Return a short label for a node based on its public key."""
        return n[:4] if len(n) == 64 else n

    def val_ok(val):
        """Return ``True`` if a stored coordinate value looks usable."""
        return val not in (None, 0, 0.0)

    coords = [
        (meta["longitude"], meta["latitude"])
        for meta in node_meta.values()
        if val_ok(meta["longitude"]) and val_ok(meta["latitude"])
    ]

    if coords:
        min_lon = min(c[0] for c in coords)
        max_lon = max(c[0] for c in coords)
        min_lat = min(c[1] for c in coords)
        max_lat = max(c[1] for c in coords)
        width = max_lon - min_lon or 1
        height = max_lat - min_lat or 1
    else:
        min_lon = max_lon = min_lat = max_lat = width = height = 1

    IMG_W, IMG_H = 1000, 800

    def map_coords_to_latlon(lon, lat):
        """Convert stored lon/lat coordinates to Leaflet's [lat, lon] representation."""
        return [lat, lon]

    # Determine Leaflet map center and zoom
    if coords:
        center_lat = (min_lat + max_lat) / 2
        center_lon = (min_lon + max_lon) / 2
        zoom = 8
    else:
        center_lat = home_latitude
        center_lon = home_longitude
        zoom = 8

    app = dash.Dash(__name__)
    app.layout = dcc.Loading(
        type="circle",
        children=html.Div(
            className="app-root",
            children=[
                html.H2(
                    "MeshCore Reachability",
                    className="app-title",
                ),
                dcc.Checklist(
                    id="reachable-filter",
                    # If enabled, only nodes with at least one successful path trace are shown
                    options=[
                        {
                            "label": "show reachable nodes only; the checks are limited to nodes inside the given radius",
                            "value": "reachable_only",
                        }
                    ],
                    value=[],
                    className="reachable-filter",
                ),
                html.Div(
                    className="map-container",
                    children=[
                        dl.Map(
                            center=[center_lat, center_lon],
                            zoom=zoom,
                            className="leaflet-map",
                            children=[
                                dl.TileLayer(
                                    url=(
                                        f"https://api.maptiler.com/maps/topo-v4/{{z}}/{{x}}/{{y}}.png?key={maptiler_api_key}"
                                        if maptiler_api_key
                                        else "https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png"
                                    ),
                                    attribution=(
                                        "<a href='https://www.maptiler.com/copyright/' target='_blank'>&copy; MapTiler</a> "
                                        "<a href='https://www.openstreetmap.org/copyright' target='_blank'>&copy; OpenStreetMap contributors</a>"
                                        if maptiler_api_key
                                        else "&copy; OpenStreetMap contributors"
                                    ),
                                    tileSize=512 if maptiler_api_key else 256,
                                    zoomOffset=-1 if maptiler_api_key else 0,
                                ),
                                dl.LayerGroup(id="node-layer"),
                                dl.LayerGroup(id="radius-layer"),
                                dl.Pane(
                                    id="home-marker-pane",
                                    style={"zIndex": 601},  # higher than overlays / default markers
                                    name="home-marker-pane"
                                ),
                                dl.Pane(
                                    id="other-markers-pane",
                                    style={"zIndex": 600},
                                    name="other-markers-pane"
                                ),
                                dl.Marker(
                                    id="home-location-marker",
                                    position=map_coords_to_latlon(
                                        home_longitude, home_latitude
                                    ),
                                    pane="home-marker-pane",
                                    icon={
                                        "iconUrl": dash.get_asset_url("homelocation.svg"),
                                        "iconSize": "28",
                                        "shadowUrl": dash.get_asset_url("iconbg.svg"),
                                        "shadowSize": "32",
                                    },
                                ),
                            ],
                        ),
                    ],
                ),
                html.Div(
                    id="stats-container",
                    className="stats-container",
                    children=[
                        html.Table(
                            className="stats-table",
                            children=[
                                html.Thead(
                                    children=html.Tr(
                                        children=[
                                            html.Th(""),
                                            html.Th("Adverts received"),
                                            html.Th(
                                                "Reachable Nodes inside check radius"
                                            ),
                                        ],
                                    ),
                                ),
                                html.Tbody(
                                    children=[
                                        html.Tr(
                                            children=[
                                                html.Td(
                                                    "Chat Nodes",
                                                    className="stat-label",
                                                ),
                                                html.Td(
                                                    id="stat-chatnodes-rcvd",
                                                    className="stat-value-center",
                                                ),
                                                html.Td(
                                                    id="stat-chatnodes-reach",
                                                    className="stat-value-right",
                                                ),
                                            ],
                                        ),
                                        html.Tr(
                                            children=[
                                                html.Td(
                                                    "Repeaters",
                                                    className="stat-label",
                                                ),
                                                html.Td(
                                                    id="stat-repeaters-rcvd",
                                                    className="stat-value-center",
                                                ),
                                                html.Td(
                                                    id="stat-repeaters-reach",
                                                    className="stat-value-right",
                                                ),
                                            ],
                                        ),
                                        html.Tr(
                                            children=[
                                                html.Td(
                                                    "Room Servers",
                                                    className="stat-label",
                                                ),
                                                html.Td(
                                                    id="stat-roomservers-rcvd",
                                                    className="stat-value-center",
                                                ),
                                                html.Td(
                                                    id="stat-roomservers-reach",
                                                    className="stat-value-right",
                                                ),
                                            ],
                                        ),
                                    ],
                                ),
                            ],
                        ),
                        html.P(
                            id="stat-checked-paths",
                            className="stat-checked-paths",
                        ),
                        html.H3(
                            "Nodes without geo location:",
                            className="nodes-without-geo-title",
                        ),
                        html.Table(
                            className="nodes-without-geo-table",
                            children=[
                                html.Thead(
                                    children=html.Tr(
                                        children=[
                                            html.Th("Name"),
                                            html.Th("Role"),
                                            html.Th("Public-Key"),
                                            html.Th("Contact-QR-Code"),
                                        ],
                                    ),
                                ),
                                html.Tbody(id="nodes-without-geo-body"),
                            ],
                        ),
                    ],
                ),
                html.Div(
                    id="node-details-overlay",
                    className="node-details-overlay",
                ),
                dcc.Store(id="node-meta-store", data=node_meta),
                dcc.Interval(
                    id="db-refresh-interval",
                    interval=2 * 60 * 1000,  # refresh every 2 minutes
                    n_intervals=0,
                ),
            ],
        ),
    )

    @app.callback(
        Output("node-meta-store", "data"),
        Input("db-refresh-interval", "n_intervals"),
        State("node-meta-store", "data"),
    )
    def refresh_node_meta(n_intervals, current_data):
        """Periodically reload node metadata from the database."""
        if n_intervals is None:
            return current_data
        # Could be further throttled via (n_intervals % N) if required
        return load_node_meta_from_db()

    @app.callback(
        Output("node-layer", "children"),
        Output("radius-layer", "children"),
        Output("stat-chatnodes-rcvd", "children"),
        Output("stat-chatnodes-reach", "children"),
        Output("stat-repeaters-rcvd", "children"),
        Output("stat-repeaters-reach", "children"),
        Output("stat-roomservers-rcvd", "children"),
        Output("stat-roomservers-reach", "children"),
        Output("stat-checked-paths", "children"),
        Output("nodes-without-geo-body", "children"),
        Input("reachable-filter", "value"),
        Input("node-meta-store", "data"),
    )
    def update_node_markers(filter_values, node_meta_store):
        """Update map markers, radius overlay and statistics whenever data changes."""
        show_reachable_only = "reachable_only" in (filter_values or [])

        def val_ok(val):
            return val not in (None, 0, 0.0)

        def get_role_id(meta):
            """Return the numeric MeshCore role ID for a node."""
            role_id = 1
            if meta.get("role") == "Repeater":
                role_id = 2
            elif meta.get("role") == "Room Server":
                role_id = 3
            return role_id

        def get_mclink_qr(meta):
            """Return (and cache) a data-url QR code for adding this node as contact."""
            role_id = get_role_id(meta)
            mclink = f"meshcore://contact/add?{urlencode({'name': meta.get('name')})}&public_key={meta['public_key']}&type={role_id}"

            if mclink in mclink_qr_cache:
                return mclink_qr_cache[mclink]

            qr = qrcode.QRCode(box_size=4, border=2)
            qr.add_data(mclink)
            qr.make(fit=True)
            img = qr.make_image(fill_color="black", back_color="white")
            buf = io.BytesIO()
            img.save(buf, format="PNG")
            img_b64 = base64.b64encode(buf.getvalue()).decode("ascii")
            mclink_qr_data_url = f"data:image/png;base64,{img_b64}"
            mclink_qr_cache[mclink] = mclink_qr_data_url
            return mclink_qr_data_url

        # Create markers for all nodes with location information
        markers = []

        # Radius circle (if provided)
        radius_markers = []
        if checkradius_km is not None and val_ok(home_latitude) and val_ok(home_longitude):
            # Leaflet expects radius in meters
            radius_m = checkradius_km * 1000.0
            radius_markers.append(
                dl.Circle(
                    center=[home_latitude, home_longitude],
                    radius=radius_m,
                    color="black",
                    fill=False,
                    opacity=0.8,
                    weight=2,
                )
            )
        for meta in node_meta_store.values():
            if show_reachable_only and meta.get("reachable", 0) == 0:
                continue
            if val_ok(meta.get("longitude")) and val_ok(meta.get("latitude")):

                mclink_qr_data_url = get_mclink_qr(meta)

                markers.append(
                    dl.Marker(
                        id=f"node-marker-{meta['public_key']}",
                        position=map_coords_to_latlon(
                            meta["longitude"], meta["latitude"]
                        ),
                        pane="other-markers-pane",
                        children=dl.Tooltip(
                            content=(
                                f"<span style='font-weight:bold;color:{'blue' if meta['reachable']==1 else 'red'}'>{meta['role']} {meta['name']}</span>"  # noqa: E501
                                f"<br/>Latest Path: {meta['lastpath'] or '[direct]'}"  # noqa: E501
                                f"<br/>Public-Key: {meta['public_key'][:10]}..."  # noqa: E501
                                f"<br/><img src='{mclink_qr_data_url}' alt='MC-Link QR Code'/>"  # noqa: E501
                            )
                        ),
                        icon={
                            "iconUrl": dash.get_asset_url(
                                f"{meta['role'].lower().replace(' ', '-')}{'_reachable' if meta['reachable']==1 else ''}.svg"
                            ),
                            "iconSize": "24",
                            "shadowUrl": dash.get_asset_url("iconbg.svg"),
                            "shadowSize": "28",
                        },
                    )
                )

        # Table "Nodes without geo location"
        nodes_no_geo = [
            meta
            for meta in node_meta_store.values()
            if not val_ok(meta.get("longitude"))
            or not val_ok(meta.get("latitude"))
        ]

        def sort_key(m):
            """Sort reachable nodes first and then by most recent advert time."""
            reachable_flag = 0 if m.get("reachable") == 1 else 1
            try:
                dt = datetime.fromisoformat(
                    m.get("lastmod") or "1970-01-01 00:00:00"
                )
                ts = dt.timestamp()
            except Exception:
                ts = 0
            return (reachable_flag, -ts)

        nodes_no_geo_sorted = sorted(nodes_no_geo, key=sort_key)

        rows_no_geo = []
        for meta in nodes_no_geo_sorted:
            mclink_qr_data_url = get_mclink_qr(meta)

            name_style = {
                "padding": "6px 4px",
                "borderBottom": "1px solid #272727",
                "color": "#e0e0e0",
            }
            if meta.get("reachable") == 1:
                name_style["color"] = "#81d4fa"

            cell_style = {
                "padding": "6px 4px",
                "borderBottom": "1px solid #272727",
                "color": "#e0e0e0",
            }

            rows_no_geo.append(
                html.Tr(
                    children=[
                        html.Td(meta.get("name") or "", style=name_style),
                        html.Td(
                            meta.get("role") or "",
                            style=cell_style,
                        ),
                        html.Td(
                            meta.get("public_key") or "",
                            style=cell_style,
                        ),
                        html.Td(
                            html.Div(
                                className="qr-button-with-tooltip",
                                children=[
                                    html.Img(
                                        src=dash.get_asset_url("qrcode.svg"),
                                        className="qr-icon",
                                    ),
                                    html.Div(
                                        className="qr-tooltip",
                                        children=[
                                            html.Img(
                                                src=mclink_qr_data_url,
                                                className="qr-tooltip-image",
                                            )
                                        ],
                                    ),
                                ],
                            ),
                            style=cell_style,
                        ),
                    ]
                )
            )

        return (
            markers,
            radius_markers,
            statistics["chatnodes_rcvd_adverts"],
            statistics["chatnodes_reachable"],
            statistics["repeaters_rcvd_adverts"],
            statistics["repeaters_reachable"],
            statistics["roomservers_rcvd_adverts"],
            statistics["roomservers_reachable"],
            f"Checked {statistics['checked_paths']} paths including sub-paths.",
            rows_no_geo,
        )

    return app


def dash_server_thread(
    db_path: str,
    stop_event: threading.Event,
    maptiler_api_key: str | None = None,
    home_latitude: float | None = None,
    home_longitude: float | None = None,
    checkradius_km: float | None = None,
):
    """Start the Dash application (blocking) in this thread."""
    app = create_dash_app_from_db(
        db_path,
        maptiler_api_key=maptiler_api_key,
        home_latitude=home_latitude,
        home_longitude=home_longitude,
        checkradius_km=checkradius_km,
    )
    # Dash itself has no built-in way to be shut down via an event.
    # We simply start the server and rely on process termination.
    print("[dash] Starting Dash server on http://0.0.0.0:5342 ...")
    app.run(host="0.0.0.0", port=5342, debug=False)


# --- main() ---------------------------------------------------------------


async def main(stop_event: threading.Event):
    """Parse CLI arguments, start collector thread and (optionally) the Dash UI."""

    parser = argparse.ArgumentParser(description="MeshCore Reachability Graph")

    parser.add_argument("-p", "--port", required=False, help="LoRa-Device serial port")
    parser.add_argument("-ip", "--host", required=False, help="LoRa-Device host ip")
    parser.add_argument("-ble", "--ble", required=False, help="LoRa-Device ble address")

    parser.add_argument(
        "--db", default="mcreach.sqlite", help="SQLite database file"
    )
    parser.add_argument(
        "-lat",
        "--latitude",
        dest="latitude",
        type=float,
        required=True,
        help="Latitude of home location, e.g. 47.73322",
    )
    parser.add_argument(
        "-lon",
        "--longitude",
        dest="longitude",
        type=float,
        required=True,
        help="Longitude of home location, e.g. 12.11043",
    )
    parser.add_argument(
        "-rad",
        "--checkradius_km",
        dest="checkradius_km",
        type=float,
        default=200.0,
        help=(
            "limit path checks on nodes within this radius from the home location (in km) to reduce traffic"
        ),
    )
    parser.add_argument(
        "--headless",
        action="store_true",
        required=False,
        help="Run packet collection without web-ui",
    )
    parser.add_argument(
        "--guionly",
        action="store_true",
        required=False,
        help="Run web-ui without packet collection",
    )
    parser.add_argument(
        "-ak",
        "--maptiler_api_key",
        dest="maptiler_api_key",
        help="Optional MapTiler API key for background map",
        required=False,
    )
    args = parser.parse_args()

    db_path = args.db
    maptiler_api_key = args.maptiler_api_key
    home_latitude = args.latitude
    home_longitude = args.longitude
    checkradius_km = args.checkradius_km

    # Ensure database and schema are initialized even in GUI-only mode
    conn = init_db(db_path)
    conn.close()

    # Thread that collects adverts, evaluates paths and runs traces sequentially
    t_collect_paths = threading.Thread(
        target=advert_and_path_thread,
        args=(
            args.port,
            args.host,
            args.ble,
            db_path,
            stop_event,
            home_latitude,
            home_longitude,
            checkradius_km,
        ),
        daemon=True,
    )

    if args.headless and args.guionly:
        print(
            "Error in arguments: disable either packet collection (--headless) or gui (--guionly), not both."
        )
        return

    if not args.guionly:
        t_collect_paths.start()
        print("[main] Collector thread started")

    if args.headless:
        # In pure headless mode just keep the main coroutine alive
        while True:
            await asyncio.sleep(2)
    else:
        dash_server_thread(
            db_path,
            stop_event,
            maptiler_api_key,
            home_latitude,
            home_longitude,
            checkradius_km,
        )
        print("[main] Launching Dash app")


if __name__ == "__main__":
    import asyncio as _asyncio

    stop_event = threading.Event()

    try:
        print("[main] Started app (Ctrl+C to stop)...")
        _asyncio.run(main(stop_event))
    except KeyboardInterrupt:
        print("[main] Stopping ...")
        stop_event.set()