chore: refactor codebase before meshcore release (#682)

* chore: refactor codebase before meshcore release

* data: run black

* fix: resolve SonarCloud S1244/S5796 reliability issues in test files

Replace floating-point equality comparisons with pytest.approx() to
satisfy S1244, and replace the `is` identity operator with id()-based
comparison to satisfy S5796.

* fix: remove duplicate encrypted_flag assignment in store_packet_dict

The encrypted_flag was computed identically on lines 307 and 345 with no
mutation of `encrypted` between them. Remove the dead second assignment.

data/mesh_ingestor/channels.py

@@ -182,6 +182,9 @@ def capture_from_interface(iface: Any) -> None:
     channels_obj = getattr(local_node, "channels", None) if local_node else None
 
     channel_entries: list[tuple[int, str]] = []
+    # Use a set for O(1) duplicate-index checks; Meshtastic occasionally
+    # emits the same channel index twice when the channel list is partially
+    # initialised, so we keep only the first valid entry per index.
     seen_indices: set[int] = set()
     for candidate in _iter_channel_objects(channels_obj):
         result = _channel_tuple(candidate)

data/mesh_ingestor/daemon.py

@@ -26,6 +26,7 @@ from pubsub import pub
 
 from . import config, handlers, ingestors, interfaces
 from .provider import Provider
+from .utils import _retry_dict_snapshot
 
 _RECEIVE_TOPICS = (
     "meshtastic.receive",
@@ -82,10 +83,15 @@ def _subscribe_receive_topics() -> list[str]:
 
 
 def _node_items_snapshot(
-    nodes_obj, retries: int = 3
+    nodes_obj: object, retries: int = 3
 ) -> list[tuple[str, object]] | None:
     """Snapshot ``nodes_obj`` to avoid iteration errors during updates.
 
+    Uses :func:`~data.mesh_ingestor.utils._retry_dict_snapshot` to handle
+    both dict-like objects (``items()`` callable) and sequence-like objects
+    (``__iter__`` + ``__getitem__``) that Meshtastic may return depending on
+    firmware version.
+
     Parameters:
         nodes_obj: Meshtastic nodes mapping or iterable.
         retries: Number of attempts when encountering "dictionary changed"
@@ -101,25 +107,15 @@ def _node_items_snapshot(
 
     items_callable = getattr(nodes_obj, "items", None)
     if callable(items_callable):
-        for _ in range(max(1, retries)):
-            try:
-                return list(items_callable())
-            except RuntimeError as err:
-                if "dictionary changed size during iteration" not in str(err):
-                    raise
-                time.sleep(0)
-        return None
+        return _retry_dict_snapshot(lambda: list(items_callable()), retries)
 
     if hasattr(nodes_obj, "__iter__") and hasattr(nodes_obj, "__getitem__"):
-        for _ in range(max(1, retries)):
-            try:
-                keys = list(nodes_obj)
-                return [(key, nodes_obj[key]) for key in keys]
-            except RuntimeError as err:
-                if "dictionary changed size during iteration" not in str(err):
-                    raise
-                time.sleep(0)
-        return None
+
+        def _snapshot_via_keys() -> list[tuple[str, object]]:
+            keys = list(nodes_obj)
+            return [(key, nodes_obj[key]) for key in keys]
+
+        return _retry_dict_snapshot(_snapshot_via_keys, retries)
 
     return []
 
@@ -321,6 +317,9 @@ def _try_connect(state: _DaemonState) -> bool:
                 target=state.resolved_target,
             )
             state.announced_target = True
+        # Set an absolute monotonic deadline for this energy-saving session.
+        # When the deadline passes, _check_energy_saving() will close the
+        # interface and sleep until the next wake interval.
         if state.energy_saving_enabled and state.energy_online_secs > 0:
             state.energy_session_deadline = time.monotonic() + state.energy_online_secs
         else:
@@ -588,9 +587,16 @@ def main(*, provider: Provider | None = None) -> None:
     )
 
     def handle_sigterm(*_args) -> None:
+        """Set the stop flag so the daemon loop exits cleanly on SIGTERM."""
         state.stop.set()
 
     def handle_sigint(signum, frame) -> None:
+        """Handle SIGINT (Ctrl-C) with graceful-first, hard-exit-second behaviour.
+
+        The first SIGINT sets the stop flag and lets the loop finish its
+        current iteration.  A second SIGINT delegates to the default handler,
+        which raises :class:`KeyboardInterrupt` and terminates immediately.
+        """
         if state.stop.is_set():
             signal.default_int_handler(signum, frame)
             return

data/mesh_ingestor/decode_payload.py

@@ -59,6 +59,17 @@ def _decode_payload(portnum: int, payload_b64: str) -> dict[str, Any]:
 
 
 def main() -> int:
+    """Read a JSON request from stdin and write a decoded protobuf response to stdout.
+
+    Reads a single JSON object containing ``portnum`` (int) and
+    ``payload_b64`` (base-64 encoded bytes) from standard input, decodes the
+    protobuf payload via :func:`_decode_payload`, and writes the result as
+    JSON to standard output.
+
+    Returns:
+        ``0`` on success, ``1`` when the input is malformed or required fields
+        are absent.
+    """
     raw = sys.stdin.read()
     try:
         request = json.loads(raw)

data/mesh_ingestor/events.py

@@ -28,12 +28,21 @@ from typing import NotRequired, TypedDict
 
 
 class _MessageEventRequired(TypedDict):
+    """Required fields shared by all :class:`MessageEvent` payloads."""
+
     id: int
     rx_time: int
     rx_iso: str
 
 
 class MessageEvent(_MessageEventRequired, total=False):
+    """Payload for the ``/api/messages`` ingest route.
+
+    Maps to the ``MessageEvent`` contract described in ``CONTRACTS.md``.
+    Required fields are inherited from :class:`_MessageEventRequired`;
+    all other fields are optional.
+    """
+
     from_id: object
     to_id: object
     channel: int
@@ -52,12 +61,21 @@ class MessageEvent(_MessageEventRequired, total=False):
 
 
 class _PositionEventRequired(TypedDict):
+    """Required fields shared by all :class:`PositionEvent` payloads."""
+
     id: int
     rx_time: int
     rx_iso: str
 
 
 class PositionEvent(_PositionEventRequired, total=False):
+    """Payload for the ``/api/positions`` ingest route.
+
+    Maps to the ``PositionEvent`` contract described in ``CONTRACTS.md``.
+    Coordinates may be supplied as floating-point degrees or derived from
+    Meshtastic's integer-scaled ``latitudeI``/``longitudeI`` fields.
+    """
+
     node_id: str
     node_num: int | None
     num: int | None
@@ -85,12 +103,21 @@ class PositionEvent(_PositionEventRequired, total=False):
 
 
 class _TelemetryEventRequired(TypedDict):
+    """Required fields shared by all :class:`TelemetryEvent` payloads."""
+
     id: int
     rx_time: int
     rx_iso: str
 
 
 class TelemetryEvent(_TelemetryEventRequired, total=False):
+    """Payload for the ``/api/telemetry`` ingest route.
+
+    Maps to the ``TelemetryEvent`` contract described in ``CONTRACTS.md``.
+    Metric keys beyond the required ones are open-ended; the web layer accepts
+    any additional device, environment, power, or air-quality fields.
+    """
+
     node_id: str | None
     node_num: int | None
     from_id: object
@@ -112,23 +139,39 @@ class TelemetryEvent(_TelemetryEventRequired, total=False):
 
 
 class _NeighborEntryRequired(TypedDict):
+    """Required fields for a single entry within a :class:`NeighborsSnapshot`."""
+
     rx_time: int
     rx_iso: str
 
 
 class NeighborEntry(_NeighborEntryRequired, total=False):
+    """A single observed neighbour node within a :class:`NeighborsSnapshot`.
+
+    Each entry describes one node heard by the reporting device, including
+    optional signal-quality metrics.
+    """
+
     neighbor_id: str
     neighbor_num: int | None
     snr: float | None
 
 
 class _NeighborsSnapshotRequired(TypedDict):
+    """Required fields shared by all :class:`NeighborsSnapshot` payloads."""
+
     node_id: str
     rx_time: int
     rx_iso: str
 
 
 class NeighborsSnapshot(_NeighborsSnapshotRequired, total=False):
+    """Payload for the ``/api/neighbors`` ingest route.
+
+    Maps to the ``NeighborsSnapshot`` contract described in ``CONTRACTS.md``.
+    Encapsulates the full list of neighbours heard by a single reporting node.
+    """
+
     node_num: int | None
     neighbors: list[NeighborEntry]
     node_broadcast_interval_secs: int | None
@@ -139,12 +182,21 @@ class NeighborsSnapshot(_NeighborsSnapshotRequired, total=False):
 
 
 class _TraceEventRequired(TypedDict):
+    """Required fields shared by all :class:`TraceEvent` payloads."""
+
     hops: list[int]
     rx_time: int
     rx_iso: str
 
 
 class TraceEvent(_TraceEventRequired, total=False):
+    """Payload for the ``/api/traceroutes`` ingest route.
+
+    Maps to the ``TraceEvent`` contract described in ``CONTRACTS.md``.
+    The ``hops`` list contains node numbers in transmission order from
+    source to destination.
+    """
+
     id: int | None
     request_id: int | None
     src: int | None
@@ -158,6 +210,13 @@ class TraceEvent(_TraceEventRequired, total=False):
 
 
 class IngestorHeartbeat(TypedDict):
+    """Payload for the ``/api/ingestors`` heartbeat route.
+
+    Maps to the ``IngestorHeartbeat`` contract described in ``CONTRACTS.md``.
+    Sent periodically to signal that the ingestor process is alive and
+    associated with a particular radio node.
+    """
+
     node_id: str
     start_time: int
     last_seen_time: int

data/mesh_ingestor/handlers/__init__.py

@@ -0,0 +1,100 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Packet handlers that serialise mesh data and push it to the HTTP queue.
+
+This package is organised into focused submodules:
+
+- :mod:`._state` — shared mutable state (host node ID, packet timestamps)
+- :mod:`.radio` — radio metadata enrichment helpers
+- :mod:`.ignored` — debug-mode logging of dropped packets
+- :mod:`.position` — GPS position and traceroute handlers
+- :mod:`.telemetry` — device/environment telemetry and router heartbeat handlers
+- :mod:`.nodeinfo` — node information update handler
+- :mod:`.neighborinfo` — neighbour topology snapshot handler
+- :mod:`.generic` — packet dispatcher, node upsert, and the main receive callback
+
+All public names from the original flat ``handlers`` module are re-exported
+here so existing callers (e.g. ``daemon.py``, ``providers/``) require no
+changes.
+"""
+
+from __future__ import annotations
+
+from .. import queue as _queue
+from ._state import (
+    host_node_id,
+    last_packet_monotonic,
+    register_host_node_id,
+)
+from .generic import (
+    _is_encrypted_flag,
+    _portnum_candidates,
+    on_receive,
+    store_packet_dict,
+    upsert_node,
+)
+from .ignored import (
+    _IGNORED_PACKET_LOCK,
+    _IGNORED_PACKET_LOG_PATH,
+    _record_ignored_packet,
+)
+from .neighborinfo import store_neighborinfo_packet
+from .nodeinfo import store_nodeinfo_packet
+from .position import (
+    _normalize_trace_hops,
+    base64_payload,
+    store_position_packet,
+    store_traceroute_packet,
+)
+from .radio import (
+    _apply_radio_metadata,
+    _apply_radio_metadata_to_nodes,
+    _radio_metadata_fields,
+)
+from .telemetry import (
+    _VALID_TELEMETRY_TYPES,
+    store_router_heartbeat_packet,
+    store_telemetry_packet,
+)
+
+# Re-export the queue alias for any callers that reference handlers._queue_post_json
+_queue_post_json = _queue._queue_post_json
+
+__all__ = [
+    "_IGNORED_PACKET_LOCK",
+    "_IGNORED_PACKET_LOG_PATH",
+    "_VALID_TELEMETRY_TYPES",
+    "_apply_radio_metadata",
+    "_apply_radio_metadata_to_nodes",
+    "_is_encrypted_flag",
+    "_normalize_trace_hops",
+    "_portnum_candidates",
+    "_queue_post_json",
+    "_radio_metadata_fields",
+    "_record_ignored_packet",
+    "base64_payload",
+    "host_node_id",
+    "last_packet_monotonic",
+    "on_receive",
+    "register_host_node_id",
+    "store_neighborinfo_packet",
+    "store_nodeinfo_packet",
+    "store_packet_dict",
+    "store_position_packet",
+    "store_router_heartbeat_packet",
+    "store_telemetry_packet",
+    "store_traceroute_packet",
+    "upsert_node",
+]

data/mesh_ingestor/handlers/_state.py

@@ -0,0 +1,157 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Shared mutable state and state accessors for the handlers subpackage.
+
+All mutable globals that span multiple handler modules live here so that each
+handler submodule can import this module and get a consistent view of state
+without risking stale references from bare ``from ... import`` bindings.
+"""
+
+from __future__ import annotations
+
+import math
+import time
+
+from .. import config
+from ..serialization import _canonical_node_id
+
+# ---------------------------------------------------------------------------
+# Host device identity
+# ---------------------------------------------------------------------------
+
+_host_node_id: str | None = None
+"""Canonical ``!xxxxxxxx`` identifier for the connected host device."""
+
+_host_telemetry_last_rx: int | None = None
+"""Receive timestamp of the last accepted host telemetry packet."""
+
+_HOST_TELEMETRY_INTERVAL_SECS: int = 60 * 60
+"""Minimum interval (seconds) between accepted host telemetry packets.
+
+Meshtastic devices report their own telemetry at regular intervals. Accepting
+every packet would overwrite the host's profile too aggressively; this window
+throttles updates to at most once per hour.
+"""
+
+# ---------------------------------------------------------------------------
+# Packet receipt tracking
+# ---------------------------------------------------------------------------
+
+_last_packet_monotonic: float | None = None
+"""Monotonic timestamp of the most recently processed packet."""
+
+
+# ---------------------------------------------------------------------------
+# Public accessors
+# ---------------------------------------------------------------------------
+
+
+def register_host_node_id(node_id: str | None) -> None:
+    """Record the canonical identifier for the connected host device.
+
+    Resetting the host node also clears the telemetry suppression window so
+    the first telemetry packet from the new host is always accepted.
+
+    Parameters:
+        node_id: Identifier reported by the connected device. ``None`` clears
+            the current host assignment.
+    """
+
+    global _host_node_id, _host_telemetry_last_rx
+    canonical = _canonical_node_id(node_id)
+    _host_node_id = canonical
+    _host_telemetry_last_rx = None
+    if canonical:
+        config._debug_log(
+            "Registered host device node id",
+            context="handlers.host_device",
+            host_node_id=canonical,
+        )
+
+
+def host_node_id() -> str | None:
+    """Return the canonical identifier for the connected host device.
+
+    Returns:
+        The canonical ``!xxxxxxxx`` node identifier, or ``None`` when no host
+        has been registered yet.
+    """
+
+    return _host_node_id
+
+
+def _mark_host_telemetry_seen(rx_time: int) -> None:
+    """Update the last receive timestamp for the host telemetry window.
+
+    Parameters:
+        rx_time: Unix timestamp of the accepted host telemetry packet.
+    """
+
+    global _host_telemetry_last_rx
+    _host_telemetry_last_rx = rx_time
+
+
+def _host_telemetry_suppressed(rx_time: int) -> tuple[bool, int]:
+    """Return suppression state and minutes remaining for host telemetry.
+
+    Host telemetry is suppressed when it arrives within
+    :data:`_HOST_TELEMETRY_INTERVAL_SECS` of the previous accepted packet.
+    This avoids flooding the API with high-frequency device metrics from the
+    locally connected node.
+
+    Parameters:
+        rx_time: Unix timestamp of the candidate telemetry packet.
+
+    Returns:
+        A ``(suppressed, minutes_remaining)`` tuple.  ``suppressed`` is
+        ``True`` when the packet should be dropped; ``minutes_remaining``
+        is the whole number of minutes until the next packet will be accepted.
+    """
+
+    if _host_telemetry_last_rx is None:
+        return False, 0
+    remaining_secs = (_host_telemetry_last_rx + _HOST_TELEMETRY_INTERVAL_SECS) - rx_time
+    if remaining_secs <= 0:
+        return False, 0
+    return True, int(math.ceil(remaining_secs / 60.0))
+
+
+def last_packet_monotonic() -> float | None:
+    """Return the monotonic timestamp of the most recently processed packet.
+
+    Returns:
+        A :func:`time.monotonic` value, or ``None`` before any packet has been
+        received.
+    """
+
+    return _last_packet_monotonic
+
+
+def _mark_packet_seen() -> None:
+    """Record that a packet has been processed by updating the monotonic clock."""
+
+    global _last_packet_monotonic
+    _last_packet_monotonic = time.monotonic()
+
+
+__all__ = [
+    "_HOST_TELEMETRY_INTERVAL_SECS",
+    "_host_telemetry_suppressed",
+    "_mark_host_telemetry_seen",
+    "_mark_packet_seen",
+    "host_node_id",
+    "last_packet_monotonic",
+    "register_host_node_id",
+]

data/mesh_ingestor/handlers/generic.py

@@ -0,0 +1,478 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Generic packet dispatcher, node upsert, and the main receive callback."""
+
+from __future__ import annotations
+
+import base64
+import contextlib
+import importlib
+import json
+import sys
+import time
+from collections.abc import Mapping
+
+from .. import channels, config, queue
+from ..serialization import (
+    _canonical_node_id,
+    _coerce_int,
+    _first,
+    _iso,
+    _pkt_to_dict,
+    upsert_payload,
+)
+from . import _state, ignored as _ignored_mod
+from .neighborinfo import store_neighborinfo_packet
+from .nodeinfo import store_nodeinfo_packet
+from .position import store_position_packet
+from .radio import _apply_radio_metadata, _apply_radio_metadata_to_nodes
+from .telemetry import store_router_heartbeat_packet, store_telemetry_packet
+from .position import store_traceroute_packet
+
+
+def _portnum_candidates(name: str) -> set[int]:
+    """Return Meshtastic port number candidates for ``name``.
+
+    Meshtastic ships two protobuf module layouts (legacy and modern).  Both are
+    probed so that port-number comparisons work regardless of which firmware
+    version is installed.
+
+    Parameters:
+        name: Port name to look up in Meshtastic ``PortNum`` enums.
+
+    Returns:
+        Set of integer port numbers resolved from all available Meshtastic
+        modules.
+    """
+
+    candidates: set[int] = set()
+    for module_name in (
+        "meshtastic.portnums_pb2",
+        "meshtastic.protobuf.portnums_pb2",
+    ):
+        module = sys.modules.get(module_name)
+        if module is None:
+            with contextlib.suppress(ModuleNotFoundError):
+                module = importlib.import_module(module_name)
+        if module is None:
+            continue
+        portnum_enum = getattr(module, "PortNum", None)
+        value_lookup = getattr(portnum_enum, "Value", None) if portnum_enum else None
+        if callable(value_lookup):
+            with contextlib.suppress(Exception):
+                candidate = _coerce_int(value_lookup(name))
+                if candidate is not None:
+                    candidates.add(candidate)
+        constant_value = getattr(module, name, None)
+        candidate = _coerce_int(constant_value)
+        if candidate is not None:
+            candidates.add(candidate)
+    return candidates
+
+
+def _is_encrypted_flag(value: object) -> bool:
+    """Return ``True`` when ``value`` represents an encrypted payload.
+
+    Meshtastic may express the encrypted flag as a boolean, an integer, or a
+    string depending on how the packet was decoded.  All representations are
+    normalised to a Python bool.
+
+    Parameters:
+        value: Raw encrypted field from a Meshtastic packet.
+
+    Returns:
+        ``True`` when the payload is considered encrypted, ``False`` otherwise.
+    """
+
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)):
+        return value != 0
+    if isinstance(value, str):
+        normalized = value.strip().lower()
+        if normalized in {"", "0", "false", "no"}:
+            return False
+        return True
+    return bool(value)
+
+
+def upsert_node(node_id: object, node: object) -> None:
+    """Schedule an upsert for a single node.
+
+    Serialises ``node`` via :func:`upsert_payload`, enriches the result with
+    radio metadata and the current host node identifier, then enqueues a POST
+    to ``/api/nodes``.
+
+    Parameters:
+        node_id: Canonical identifier for the node in the ``!xxxxxxxx`` format.
+        node: Node object or mapping to serialise for the API payload.
+
+    Returns:
+        ``None``. The payload is forwarded to the shared HTTP queue.
+    """
+
+    payload = _apply_radio_metadata_to_nodes(upsert_payload(node_id, node))
+    payload["ingestor"] = _state.host_node_id()
+    queue._queue_post_json("/api/nodes", payload, priority=queue._NODE_POST_PRIORITY)
+
+    if config.DEBUG:
+        from ..serialization import _get
+
+        user = _get(payload[node_id], "user") or {}
+        short = _get(user, "shortName")
+        long = _get(user, "longName")
+        config._debug_log(
+            "Queued node upsert payload",
+            context="handlers.upsert_node",
+            node_id=node_id,
+            short_name=short,
+            long_name=long,
+        )
+
+
+def store_packet_dict(packet: Mapping) -> None:
+    """Route a decoded packet to the appropriate storage handler.
+
+    Inspects ``portnum`` (string and integer forms) and the presence of
+    well-known decoded sub-sections to determine packet type, then delegates
+    to the corresponding ``store_*`` handler.
+
+    Parameters:
+        packet: Packet dictionary emitted by the mesh interface.
+
+    Returns:
+        ``None``. Side-effects depend on the specific handler invoked.
+    """
+
+    decoded = packet.get("decoded") or {}
+
+    portnum_raw = _first(decoded, "portnum", default=None)
+    portnum = str(portnum_raw).upper() if portnum_raw is not None else None
+    portnum_int = _coerce_int(portnum_raw)
+
+    telemetry_section = (
+        decoded.get("telemetry") if isinstance(decoded, Mapping) else None
+    )
+    if (
+        portnum == "TELEMETRY_APP"
+        or portnum_int == 65
+        or isinstance(telemetry_section, Mapping)
+    ):
+        store_telemetry_packet(packet, decoded)
+        return
+
+    traceroute_section = (
+        decoded.get("traceroute") if isinstance(decoded, Mapping) else None
+    )
+    traceroute_port_ints = _portnum_candidates("TRACEROUTE_APP")
+
+    if (
+        portnum == "TRACEROUTE_APP"
+        or (portnum_int is not None and portnum_int in traceroute_port_ints)
+        or isinstance(traceroute_section, Mapping)
+    ):
+        store_traceroute_packet(packet, decoded)
+        return
+
+    if portnum in {"5", "NODEINFO_APP"}:
+        store_nodeinfo_packet(packet, decoded)
+        return
+
+    if portnum in {"4", "POSITION_APP"}:
+        store_position_packet(packet, decoded)
+        return
+
+    neighborinfo_section = (
+        decoded.get("neighborinfo") if isinstance(decoded, Mapping) else None
+    )
+    if portnum == "NEIGHBORINFO_APP" or isinstance(neighborinfo_section, Mapping):
+        store_neighborinfo_packet(packet, decoded)
+        return
+
+    store_forward_port_candidates = _portnum_candidates("STORE_FORWARD_APP")
+    store_forward_section = (
+        decoded.get("storeforward") if isinstance(decoded, Mapping) else None
+    )
+    if portnum == "STORE_FORWARD_APP" or (
+        portnum_int is not None and portnum_int in store_forward_port_candidates
+    ):
+        if not isinstance(store_forward_section, Mapping):
+            _ignored_mod._record_ignored_packet(
+                packet, reason="unsupported-store-forward"
+            )
+            return
+        rr = str(store_forward_section.get("rr") or "").upper()
+        if rr == "ROUTER_HEARTBEAT":
+            store_router_heartbeat_packet(packet)
+            return
+        _ignored_mod._record_ignored_packet(
+            packet, reason="unsupported-store-forward-rr"
+        )
+        return
+
+    text = _first(decoded, "payload.text", "text", "data.text", default=None)
+    encrypted = _first(decoded, "payload.encrypted", "encrypted", default=None)
+    if encrypted is None:
+        encrypted = _first(packet, "encrypted", default=None)
+    reply_id_raw = _first(
+        decoded,
+        "payload.replyId",
+        "payload.reply_id",
+        "data.replyId",
+        "data.reply_id",
+        "replyId",
+        "reply_id",
+        default=None,
+    )
+    reply_id = _coerce_int(reply_id_raw)
+    emoji_raw = _first(
+        decoded,
+        "payload.emoji",
+        "data.emoji",
+        "emoji",
+        default=None,
+    )
+    emoji = None
+    if emoji_raw is not None:
+        try:
+            emoji_text = str(emoji_raw)
+        except Exception:
+            emoji_text = None
+        else:
+            emoji_text = emoji_text.strip()
+            if emoji_text:
+                emoji = emoji_text
+
+    routing_section = decoded.get("routing") if isinstance(decoded, Mapping) else None
+    routing_port_candidates = _portnum_candidates("ROUTING_APP")
+    if text is None and (
+        portnum == "ROUTING_APP"
+        or (portnum_int is not None and portnum_int in routing_port_candidates)
+        or isinstance(routing_section, Mapping)
+    ):
+        routing_payload = _first(decoded, "payload", "data", default=None)
+        if routing_payload is not None:
+            if isinstance(routing_payload, bytes):
+                text = base64.b64encode(routing_payload).decode("ascii")
+            elif isinstance(routing_payload, str):
+                text = routing_payload
+            else:
+                try:
+                    text = json.dumps(routing_payload, ensure_ascii=True)
+                except TypeError:
+                    text = str(routing_payload)
+            if isinstance(text, str):
+                text = text.strip() or None
+
+    allowed_port_values = {"1", "TEXT_MESSAGE_APP", "REACTION_APP", "ROUTING_APP"}
+    allowed_port_ints = {1}
+
+    reaction_port_candidates = _portnum_candidates("REACTION_APP")
+    for candidate in reaction_port_candidates:
+        allowed_port_ints.add(candidate)
+        allowed_port_values.add(str(candidate))
+
+    for candidate in routing_port_candidates:
+        allowed_port_ints.add(candidate)
+        allowed_port_values.add(str(candidate))
+
+    if isinstance(routing_section, Mapping) and portnum_int is not None:
+        allowed_port_ints.add(portnum_int)
+        allowed_port_values.add(str(portnum_int))
+
+    is_reaction_packet = portnum == "REACTION_APP" or (
+        reply_id is not None and emoji is not None
+    )
+    if is_reaction_packet and portnum_int is not None:
+        allowed_port_ints.add(portnum_int)
+        allowed_port_values.add(str(portnum_int))
+
+    if portnum and portnum not in allowed_port_values:
+        if portnum_int not in allowed_port_ints:
+            _ignored_mod._record_ignored_packet(packet, reason="unsupported-port")
+            return
+
+    encrypted_flag = _is_encrypted_flag(encrypted)
+    if not any([text, encrypted_flag, emoji is not None, reply_id is not None]):
+        _ignored_mod._record_ignored_packet(packet, reason="no-message-payload")
+        return
+
+    channel = _first(decoded, "channel", default=None)
+    if channel is None:
+        channel = _first(packet, "channel", default=0)
+    try:
+        channel = int(channel)
+    except Exception:
+        channel = 0
+
+    channel_name_value = channels.channel_name(channel)
+
+    pkt_id = _first(packet, "id", "packet_id", "packetId", default=None)
+    if pkt_id is None:
+        _ignored_mod._record_ignored_packet(packet, reason="missing-packet-id")
+        return
+    rx_time = int(_first(packet, "rxTime", "rx_time", default=time.time()))
+    from_id = _first(packet, "fromId", "from_id", "from", default=None)
+    to_id = _first(packet, "toId", "to_id", "to", default=None)
+
+    if (from_id is None or str(from_id) == "") and config.DEBUG:
+        try:
+            raw = json.dumps(packet, default=str)
+        except Exception:
+            raw = str(packet)
+        config._debug_log(
+            "Packet missing from_id",
+            context="handlers.store_packet_dict",
+            packet=raw,
+        )
+
+    snr = _first(packet, "snr", "rx_snr", "rxSnr", default=None)
+    rssi = _first(packet, "rssi", "rx_rssi", "rxRssi", default=None)
+    hop = _first(packet, "hopLimit", "hop_limit", default=None)
+
+    to_id_normalized = str(to_id).strip() if to_id is not None else ""
+
+    if (
+        not is_reaction_packet
+        and channel == 0
+        and not encrypted_flag
+        and to_id_normalized
+        and to_id_normalized.lower() != "^all"
+    ):
+        if config.DEBUG:
+            config._debug_log(
+                "Skipped direct message on primary channel",
+                context="handlers.store_packet_dict",
+                from_id=_canonical_node_id(from_id) or from_id,
+                to_id=_canonical_node_id(to_id) or to_id,
+                channel=channel,
+            )
+        _ignored_mod._record_ignored_packet(packet, reason="skipped-direct-message")
+        return
+
+    if not channels.is_allowed_channel(channel_name_value):
+        _ignored_mod._record_ignored_packet(packet, reason="disallowed-channel")
+        if config.DEBUG:
+            config._debug_log(
+                "Ignored packet on disallowed channel",
+                context="handlers.store_packet_dict",
+                channel=channel,
+                channel_name=channel_name_value,
+                allowed_channels=channels.allowed_channel_names(),
+            )
+        return
+
+    if channels.is_hidden_channel(channel_name_value):
+        _ignored_mod._record_ignored_packet(packet, reason="hidden-channel")
+        if config.DEBUG:
+            config._debug_log(
+                "Ignored packet on hidden channel",
+                context="handlers.store_packet_dict",
+                channel=channel,
+                channel_name=channel_name_value,
+            )
+        return
+
+    message_payload = {
+        "id": int(pkt_id),
+        "rx_time": rx_time,
+        "rx_iso": _iso(rx_time),
+        "from_id": from_id,
+        "to_id": to_id,
+        "channel": channel,
+        "portnum": str(portnum) if portnum is not None else None,
+        "text": text,
+        "encrypted": encrypted,
+        "snr": float(snr) if snr is not None else None,
+        "rssi": int(rssi) if rssi is not None else None,
+        "hop_limit": int(hop) if hop is not None else None,
+        "reply_id": reply_id,
+        "emoji": emoji,
+        "ingestor": _state.host_node_id(),
+    }
+
+    if not encrypted_flag and channel_name_value:
+        message_payload["channel_name"] = channel_name_value
+    queue._queue_post_json(
+        "/api/messages",
+        _apply_radio_metadata(message_payload),
+        priority=queue._MESSAGE_POST_PRIORITY,
+    )
+
+    if config.DEBUG:
+        from_label = _canonical_node_id(from_id) or from_id
+        to_label = _canonical_node_id(to_id) or to_id
+        payload_desc = "Encrypted" if text is None and encrypted else text
+        log_kwargs = {
+            "context": "handlers.store_packet_dict",
+            "from_id": from_label,
+            "to_id": to_label,
+            "channel": channel,
+            "channel_display": channel_name_value or channel,
+            "payload": payload_desc,
+        }
+        if channel_name_value:
+            log_kwargs["channel_name"] = channel_name_value
+        config._debug_log("Queued message payload", **log_kwargs)
+
+
+def on_receive(packet: object, interface: object) -> None:
+    """Callback registered with Meshtastic to capture incoming packets.
+
+    Subscribed to all ``meshtastic.receive.*`` pubsub topics.  The packet is
+    deduplicated via a ``_potatomesh_seen`` flag before being normalised and
+    dispatched to :func:`store_packet_dict`.
+
+    Parameters:
+        packet: Packet payload supplied by the Meshtastic pubsub topic.
+        interface: Interface instance that produced the packet. Only used for
+            compatibility with Meshtastic's callback signature.
+
+    Returns:
+        ``None``. Packets are serialised and enqueued asynchronously.
+    """
+
+    if isinstance(packet, dict):
+        if packet.get("_potatomesh_seen"):
+            return
+        packet["_potatomesh_seen"] = True
+
+    _state._mark_packet_seen()
+
+    packet_dict = None
+    try:
+        packet_dict = _pkt_to_dict(packet)
+        store_packet_dict(packet_dict)
+    except Exception as exc:
+        info = (
+            list(packet_dict.keys()) if isinstance(packet_dict, dict) else type(packet)
+        )
+        config._debug_log(
+            "Failed to store packet",
+            context="handlers.on_receive",
+            severity="warn",
+            error_class=exc.__class__.__name__,
+            error_message=str(exc),
+            packet_info=info,
+        )
+
+
+__all__ = [
+    "_is_encrypted_flag",
+    "_portnum_candidates",
+    "on_receive",
+    "store_packet_dict",
+    "upsert_node",
+]

data/mesh_ingestor/handlers/ignored.py

@@ -0,0 +1,103 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Debug-mode logging of ignored Meshtastic packets.
+
+When :data:`config.DEBUG` is set the ingestor appends a JSON record for each
+packet that is filtered out (unsupported port, missing fields, disallowed
+channel, etc.) to a plain-text log file.  This aids offline debugging without
+adding overhead in production.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import threading
+from collections.abc import Mapping
+from datetime import datetime, timezone
+from pathlib import Path
+
+from .. import config
+
+_IGNORED_PACKET_LOG_PATH = (
+    Path(__file__).resolve().parents[3] / "ignored-meshtastic.txt"
+)
+"""Filesystem path that stores ignored Meshtastic packets when debug mode is active."""
+
+_IGNORED_PACKET_LOCK = threading.Lock()
+"""Lock serialising concurrent appends to :data:`_IGNORED_PACKET_LOG_PATH`."""
+
+
+def _ignored_packet_default(value: object) -> object:
+    """Return a JSON-serialisable representation for an ignored packet value.
+
+    Called as the ``default`` argument to :func:`json.dumps` when serialising
+    ignored packet entries.  Handles container types and raw bytes so the log
+    file contains readable text rather than ``repr()`` fragments.
+
+    Parameters:
+        value: Arbitrary value encountered during packet serialisation.
+
+    Returns:
+        A JSON-compatible object derived from ``value``.
+    """
+
+    if isinstance(value, (list, tuple, set)):
+        return list(value)
+    if isinstance(value, bytes):
+        return base64.b64encode(value).decode("ascii")
+    if isinstance(value, Mapping):
+        return {
+            str(key): _ignored_packet_default(sub_value)
+            for key, sub_value in value.items()
+        }
+    return str(value)
+
+
+def _record_ignored_packet(packet: Mapping | object, *, reason: str) -> None:
+    """Persist packet details to :data:`_IGNORED_PACKET_LOG_PATH` during debugging.
+
+    Does nothing when :data:`config.DEBUG` is ``False``.  Each call appends a
+    single newline-delimited JSON record with a timestamp, drop reason, and a
+    sanitised copy of the packet.
+
+    Parameters:
+        packet: Packet object or mapping to record.
+        reason: Short machine-readable label describing why the packet was
+            ignored (e.g. ``"unsupported-port"``, ``"missing-packet-id"``).
+    """
+
+    if not config.DEBUG:
+        return
+
+    timestamp = datetime.now(timezone.utc).isoformat()
+    entry = {
+        "timestamp": timestamp,
+        "reason": reason,
+        "packet": _ignored_packet_default(packet),
+    }
+    payload = json.dumps(entry, ensure_ascii=False, sort_keys=True)
+    with _IGNORED_PACKET_LOCK:
+        _IGNORED_PACKET_LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
+        with _IGNORED_PACKET_LOG_PATH.open("a", encoding="utf-8") as handle:
+            handle.write(f"{payload}\n")
+
+
+__all__ = [
+    "_IGNORED_PACKET_LOCK",
+    "_IGNORED_PACKET_LOG_PATH",
+    "_ignored_packet_default",
+    "_record_ignored_packet",
+]

data/mesh_ingestor/handlers/neighborinfo.py

@@ -0,0 +1,150 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Handler for neighbour-information packets."""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Mapping
+
+from .. import config, queue
+from ..serialization import (
+    _canonical_node_id,
+    _coerce_float,
+    _coerce_int,
+    _first,
+    _iso,
+    _node_num_from_id,
+)
+from . import _state
+from .radio import _apply_radio_metadata
+
+
+def store_neighborinfo_packet(packet: Mapping, decoded: Mapping) -> None:
+    """Persist neighbour information gathered from a packet.
+
+    Meshtastic nodes periodically broadcast the set of nodes they can hear
+    directly along with the observed signal quality.  This handler serialises
+    that snapshot so the web dashboard can render a live RF topology graph.
+
+    Parameters:
+        packet: Raw Meshtastic packet metadata.
+        decoded: Decoded view containing the ``neighborinfo`` section.
+
+    Returns:
+        ``None``. The neighbour snapshot is queued for HTTP submission.
+    """
+
+    neighbor_section = (
+        decoded.get("neighborinfo") if isinstance(decoded, Mapping) else None
+    )
+    if not isinstance(neighbor_section, Mapping):
+        return
+
+    node_ref = _first(
+        neighbor_section,
+        "nodeId",
+        "node_id",
+        default=_first(packet, "fromId", "from_id", "from", default=None),
+    )
+    node_id = _canonical_node_id(node_ref)
+    if node_id is None:
+        return
+
+    node_num = _coerce_int(_first(neighbor_section, "nodeId", "node_id", default=None))
+    if node_num is None:
+        node_num = _node_num_from_id(node_id)
+
+    node_broadcast_interval = _coerce_int(
+        _first(
+            neighbor_section,
+            "nodeBroadcastIntervalSecs",
+            "node_broadcast_interval_secs",
+            default=None,
+        )
+    )
+
+    last_sent_by_ref = _first(
+        neighbor_section,
+        "lastSentById",
+        "last_sent_by_id",
+        default=None,
+    )
+    last_sent_by_id = _canonical_node_id(last_sent_by_ref)
+
+    rx_time = _coerce_int(_first(packet, "rxTime", "rx_time", default=time.time()))
+    if rx_time is None:
+        rx_time = int(time.time())
+
+    neighbors_payload = neighbor_section.get("neighbors")
+    neighbors_iterable = (
+        neighbors_payload if isinstance(neighbors_payload, list) else []
+    )
+
+    neighbor_entries: list[dict] = []
+    for entry in neighbors_iterable:
+        if not isinstance(entry, Mapping):
+            continue
+        neighbor_ref = _first(entry, "nodeId", "node_id", default=None)
+        neighbor_id = _canonical_node_id(neighbor_ref)
+        if neighbor_id is None:
+            continue
+        neighbor_num = _coerce_int(_first(entry, "nodeId", "node_id", default=None))
+        if neighbor_num is None:
+            neighbor_num = _node_num_from_id(neighbor_id)
+        snr = _coerce_float(_first(entry, "snr", default=None))
+        entry_rx_time = _coerce_int(_first(entry, "rxTime", "rx_time", default=None))
+        if entry_rx_time is None:
+            entry_rx_time = rx_time
+        neighbor_entries.append(
+            {
+                "neighbor_id": neighbor_id,
+                "neighbor_num": neighbor_num,
+                "snr": snr,
+                "rx_time": entry_rx_time,
+                "rx_iso": _iso(entry_rx_time),
+            }
+        )
+
+    payload = {
+        "node_id": node_id,
+        "node_num": node_num,
+        "neighbors": neighbor_entries,
+        "rx_time": rx_time,
+        "rx_iso": _iso(rx_time),
+        "ingestor": _state.host_node_id(),
+    }
+
+    if node_broadcast_interval is not None:
+        payload["node_broadcast_interval_secs"] = node_broadcast_interval
+    if last_sent_by_id is not None:
+        payload["last_sent_by_id"] = last_sent_by_id
+
+    queue._queue_post_json(
+        "/api/neighbors",
+        _apply_radio_metadata(payload),
+        priority=queue._NEIGHBOR_POST_PRIORITY,
+    )
+
+    if config.DEBUG:
+        config._debug_log(
+            "Queued neighborinfo payload",
+            context="handlers.store_neighborinfo",
+            node_id=node_id,
+            neighbors=len(neighbor_entries),
+        )
+
+
+__all__ = ["store_neighborinfo_packet"]

data/mesh_ingestor/handlers/nodeinfo.py

@@ -0,0 +1,219 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Handler for node-information packets."""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Mapping
+
+from .. import config, queue
+from ..serialization import (
+    _canonical_node_id,
+    _coerce_int,
+    _decode_nodeinfo_payload,
+    _extract_payload_bytes,
+    _first,
+    _merge_mappings,
+    _node_num_from_id,
+    _node_to_dict,
+    _nodeinfo_metrics_dict,
+    _nodeinfo_position_dict,
+    _nodeinfo_user_dict,
+)
+from . import _state
+from .radio import _apply_radio_metadata_to_nodes
+
+
+def store_nodeinfo_packet(packet: Mapping, decoded: Mapping) -> None:
+    """Persist node information updates.
+
+    Node info packets carry user profile data (short name, long name, hardware
+    model, public key) together with optional position and device-metrics
+    snapshots.  When a protobuf payload is present it is decoded first; any
+    fields missing from the protobuf are filled in from the ``decoded`` dict
+    so both firmware variants are handled.
+
+    Parameters:
+        packet: Raw packet metadata describing the update.
+        decoded: Decoded payload that may include ``user`` and ``position``
+            sections.
+
+    Returns:
+        ``None``. The node payload is merged into the API queue.
+    """
+
+    payload_bytes = _extract_payload_bytes(decoded)
+    node_info = _decode_nodeinfo_payload(payload_bytes)
+    decoded_user = decoded.get("user")
+    user_dict = _nodeinfo_user_dict(node_info, decoded_user)
+
+    node_info_fields = set()
+    if node_info:
+        node_info_fields = {field_desc.name for field_desc, _ in node_info.ListFields()}
+
+    node_id = None
+    if isinstance(user_dict, Mapping):
+        node_id = _canonical_node_id(user_dict.get("id"))
+
+    if node_id is None:
+        node_id = _canonical_node_id(
+            _first(packet, "fromId", "from_id", "from", default=None)
+        )
+
+    if node_id is None:
+        return
+
+    node_payload: dict = {}
+    if user_dict:
+        node_payload["user"] = user_dict
+
+    # Resolve node_num from protobuf first, then decoded dict, then from the
+    # canonical ID as a last resort.
+    node_num = None
+    if node_info and "num" in node_info_fields:
+        try:
+            node_num = int(node_info.num)
+        except (TypeError, ValueError):
+            node_num = None
+    if node_num is None:
+        decoded_num = decoded.get("num")
+        if decoded_num is not None:
+            try:
+                node_num = int(decoded_num)
+            except (TypeError, ValueError):
+                try:
+                    node_num = int(str(decoded_num).strip(), 0)
+                except Exception:
+                    node_num = None
+    if node_num is None:
+        node_num = _node_num_from_id(node_id)
+    if node_num is not None:
+        node_payload["num"] = node_num
+
+    rx_time = int(_first(packet, "rxTime", "rx_time", default=time.time()))
+    last_heard = None
+    if node_info and "last_heard" in node_info_fields:
+        try:
+            last_heard = int(node_info.last_heard)
+        except (TypeError, ValueError):
+            last_heard = None
+    if last_heard is None:
+        decoded_last_heard = decoded.get("lastHeard")
+        if decoded_last_heard is not None:
+            try:
+                last_heard = int(decoded_last_heard)
+            except (TypeError, ValueError):
+                last_heard = None
+    if last_heard is None or last_heard < rx_time:
+        last_heard = rx_time
+    node_payload["lastHeard"] = last_heard
+
+    snr = None
+    if node_info and "snr" in node_info_fields:
+        try:
+            snr = float(node_info.snr)
+        except (TypeError, ValueError):
+            snr = None
+    if snr is None:
+        snr = _first(packet, "snr", "rx_snr", "rxSnr", default=None)
+        if snr is not None:
+            try:
+                snr = float(snr)
+            except (TypeError, ValueError):
+                snr = None
+    if snr is not None:
+        node_payload["snr"] = snr
+
+    hops = None
+    if node_info and "hops_away" in node_info_fields:
+        try:
+            hops = int(node_info.hops_away)
+        except (TypeError, ValueError):
+            hops = None
+    if hops is None:
+        hops = decoded.get("hopsAway")
+        if hops is not None:
+            try:
+                hops = int(hops)
+            except (TypeError, ValueError):
+                hops = None
+    if hops is not None:
+        node_payload["hopsAway"] = hops
+
+    if node_info and "channel" in node_info_fields:
+        try:
+            node_payload["channel"] = int(node_info.channel)
+        except (TypeError, ValueError):
+            pass
+
+    if node_info and "via_mqtt" in node_info_fields:
+        node_payload["viaMqtt"] = bool(node_info.via_mqtt)
+
+    if node_info and "is_favorite" in node_info_fields:
+        node_payload["isFavorite"] = bool(node_info.is_favorite)
+    elif "isFavorite" in decoded:
+        node_payload["isFavorite"] = bool(decoded.get("isFavorite"))
+
+    if node_info and "is_ignored" in node_info_fields:
+        node_payload["isIgnored"] = bool(node_info.is_ignored)
+    if node_info and "is_key_manually_verified" in node_info_fields:
+        node_payload["isKeyManuallyVerified"] = bool(node_info.is_key_manually_verified)
+
+    metrics = _nodeinfo_metrics_dict(node_info)
+    decoded_metrics = decoded.get("deviceMetrics")
+    if isinstance(decoded_metrics, Mapping):
+        metrics = _merge_mappings(metrics, _node_to_dict(decoded_metrics))
+    if metrics:
+        node_payload["deviceMetrics"] = metrics
+
+    position = _nodeinfo_position_dict(node_info)
+    decoded_position = decoded.get("position")
+    if isinstance(decoded_position, Mapping):
+        position = _merge_mappings(position, _node_to_dict(decoded_position))
+    if position:
+        node_payload["position"] = position
+
+    hop_limit = _first(packet, "hopLimit", "hop_limit", default=None)
+    if hop_limit is not None and "hopLimit" not in node_payload:
+        try:
+            node_payload["hopLimit"] = int(hop_limit)
+        except (TypeError, ValueError):
+            pass
+
+    nodes_payload = _apply_radio_metadata_to_nodes({node_id: node_payload})
+    nodes_payload["ingestor"] = _state.host_node_id()
+    queue._queue_post_json(
+        "/api/nodes",
+        nodes_payload,
+        priority=queue._NODE_POST_PRIORITY,
+    )
+
+    if config.DEBUG:
+        short = None
+        long_name = None
+        if isinstance(user_dict, Mapping):
+            short = user_dict.get("shortName")
+            long_name = user_dict.get("longName")
+        config._debug_log(
+            "Queued nodeinfo payload",
+            context="handlers.store_nodeinfo",
+            node_id=node_id,
+            short_name=short,
+            long_name=long_name,
+        )
+
+
+__all__ = ["store_nodeinfo_packet"]

data/mesh_ingestor/handlers/position.py

@@ -0,0 +1,413 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Handlers for position and traceroute packets."""
+
+from __future__ import annotations
+
+import base64
+import time
+from collections.abc import Mapping
+
+from .. import config, queue
+from ..serialization import (
+    _canonical_node_id,
+    _coerce_float,
+    _coerce_int,
+    _extract_payload_bytes,
+    _first,
+    _iso,
+    _node_num_from_id,
+    _node_to_dict,
+    _pkt_to_dict,
+)
+from . import _state
+from .ignored import _record_ignored_packet
+from .radio import _apply_radio_metadata
+
+
+def base64_payload(payload_bytes: bytes | None) -> str | None:
+    """Encode raw payload bytes as a Base64 string for JSON transport.
+
+    Parameters:
+        payload_bytes: Optional raw bytes to encode. When ``None`` or empty,
+            ``None`` is returned so callers can omit the field.
+
+    Returns:
+        The Base64-encoded ASCII string, or ``None`` when ``payload_bytes`` is
+        falsy.
+    """
+
+    if not payload_bytes:
+        return None
+    return base64.b64encode(payload_bytes).decode("ascii")
+
+
+def _normalize_trace_hops(hops_value: object) -> list[int]:
+    """Coerce hop entries to integer node numbers, preserving order.
+
+    Each hop can arrive as a plain integer, a canonical node-ID string
+    (``!xxxxxxxx``), or a mapping with a ``nodeId`` / ``node_id`` field.
+    All forms are normalised to the raw 32-bit node number used by the API.
+
+    Parameters:
+        hops_value: A single hop or list of hops in any supported form.
+
+    Returns:
+        List of integer node numbers with ``None``-coerced entries dropped.
+    """
+
+    if hops_value is None:
+        return []
+    hop_entries = hops_value if isinstance(hops_value, list) else [hops_value]
+    normalized: list[int] = []
+    for hop in hop_entries:
+        hop_value = hop
+        if isinstance(hop, Mapping):
+            hop_value = _first(hop, "node_id", "nodeId", "id", "num", default=None)
+
+        canonical = _canonical_node_id(hop_value)
+        hop_id = _node_num_from_id(canonical or hop_value)
+        if hop_id is None:
+            hop_id = _coerce_int(hop_value)
+        if hop_id is not None:
+            normalized.append(hop_id)
+    return normalized
+
+
+def store_position_packet(packet: Mapping, decoded: Mapping) -> None:
+    """Persist a decoded GPS position packet to the API.
+
+    Extracts coordinates from both the integer-scaled (``latitudeI`` /
+    ``longitudeI``) and floating-point (``latitude`` / ``longitude``) forms
+    that Meshtastic may produce depending on firmware version.
+
+    Parameters:
+        packet: Raw packet metadata emitted by the Meshtastic interface.
+        decoded: Decoded payload extracted from ``packet['decoded']``.
+
+    Returns:
+        ``None``. The formatted position payload is added to the HTTP queue.
+    """
+
+    node_ref = _first(packet, "fromId", "from_id", "from", default=None)
+    if node_ref is None:
+        node_ref = _first(decoded, "num", default=None)
+    node_id = _canonical_node_id(node_ref)
+    if node_id is None:
+        return
+
+    node_num = _coerce_int(_first(decoded, "num", default=None))
+    if node_num is None:
+        node_num = _node_num_from_id(node_id)
+
+    pkt_id = _coerce_int(_first(packet, "id", "packet_id", "packetId", default=None))
+    if pkt_id is None:
+        return
+
+    rx_time = _coerce_int(_first(packet, "rxTime", "rx_time", default=time.time()))
+    if rx_time is None:
+        rx_time = int(time.time())
+
+    to_id = _first(packet, "toId", "to_id", "to", default=None)
+    to_id = to_id if to_id not in {"", None} else None
+
+    position_section = decoded.get("position") if isinstance(decoded, Mapping) else None
+    if not isinstance(position_section, Mapping):
+        position_section = {}
+
+    # Meshtastic firmware may emit coordinates in one of two forms:
+    #   - Floating-point degrees: ``latitude`` / ``longitude``
+    #   - Integer-scaled (1e-7 degrees): ``latitudeI`` / ``longitudeI``
+    # Try the float form first and fall back to the integer form when absent.
+    latitude = _coerce_float(
+        _first(position_section, "latitude", "raw.latitude", default=None)
+    )
+    if latitude is None:
+        lat_i = _coerce_int(
+            _first(
+                position_section,
+                "latitudeI",
+                "latitude_i",
+                "raw.latitude_i",
+                default=None,
+            )
+        )
+        if lat_i is not None:
+            latitude = lat_i / 1e7
+
+    longitude = _coerce_float(
+        _first(position_section, "longitude", "raw.longitude", default=None)
+    )
+    if longitude is None:
+        lon_i = _coerce_int(
+            _first(
+                position_section,
+                "longitudeI",
+                "longitude_i",
+                "raw.longitude_i",
+                default=None,
+            )
+        )
+        if lon_i is not None:
+            longitude = lon_i / 1e7
+
+    altitude = _coerce_float(
+        _first(position_section, "altitude", "raw.altitude", default=None)
+    )
+    position_time = _coerce_int(
+        _first(position_section, "time", "raw.time", default=None)
+    )
+    location_source = _first(
+        position_section,
+        "locationSource",
+        "location_source",
+        "raw.location_source",
+        default=None,
+    )
+    location_source = (
+        str(location_source).strip() if location_source not in {None, ""} else None
+    )
+
+    precision_bits = _coerce_int(
+        _first(
+            position_section,
+            "precisionBits",
+            "precision_bits",
+            "raw.precision_bits",
+            default=None,
+        )
+    )
+    sats_in_view = _coerce_int(
+        _first(
+            position_section,
+            "satsInView",
+            "sats_in_view",
+            "raw.sats_in_view",
+            default=None,
+        )
+    )
+    pdop = _coerce_float(
+        _first(position_section, "PDOP", "pdop", "raw.PDOP", "raw.pdop", default=None)
+    )
+    ground_speed = _coerce_float(
+        _first(
+            position_section,
+            "groundSpeed",
+            "ground_speed",
+            "raw.ground_speed",
+            default=None,
+        )
+    )
+    ground_track = _coerce_float(
+        _first(
+            position_section,
+            "groundTrack",
+            "ground_track",
+            "raw.ground_track",
+            default=None,
+        )
+    )
+
+    snr = _coerce_float(_first(packet, "snr", "rx_snr", "rxSnr", default=None))
+    rssi = _coerce_int(_first(packet, "rssi", "rx_rssi", "rxRssi", default=None))
+    hop_limit = _coerce_int(_first(packet, "hopLimit", "hop_limit", default=None))
+    bitfield = _coerce_int(_first(decoded, "bitfield", default=None))
+
+    payload_bytes = _extract_payload_bytes(decoded)
+    payload_b64 = base64_payload(payload_bytes)
+
+    raw_section = decoded.get("raw") if isinstance(decoded, Mapping) else None
+    raw_payload = _node_to_dict(raw_section) if raw_section else None
+    if raw_payload is None and position_section:
+        raw_position = (
+            position_section.get("raw")
+            if isinstance(position_section, Mapping)
+            else None
+        )
+        if raw_position:
+            raw_payload = _node_to_dict(raw_position)
+
+    position_payload = {
+        "id": pkt_id,
+        "node_id": node_id or node_ref,
+        "node_num": node_num,
+        "num": node_num,
+        "from_id": node_id,
+        "to_id": to_id,
+        "rx_time": rx_time,
+        "rx_iso": _iso(rx_time),
+        "latitude": latitude,
+        "longitude": longitude,
+        "altitude": altitude,
+        "position_time": position_time,
+        "location_source": location_source,
+        "precision_bits": precision_bits,
+        "sats_in_view": sats_in_view,
+        "pdop": pdop,
+        "ground_speed": ground_speed,
+        "ground_track": ground_track,
+        "snr": snr,
+        "rssi": rssi,
+        "hop_limit": hop_limit,
+        "bitfield": bitfield,
+        "payload_b64": payload_b64,
+        "ingestor": _state.host_node_id(),
+    }
+    if raw_payload:
+        position_payload["raw"] = raw_payload
+
+    queue._queue_post_json(
+        "/api/positions",
+        _apply_radio_metadata(position_payload),
+        priority=queue._POSITION_POST_PRIORITY,
+    )
+
+    if config.DEBUG:
+        config._debug_log(
+            "Queued position payload",
+            context="handlers.store_position",
+            node_id=node_id,
+            latitude=latitude,
+            longitude=longitude,
+            position_time=position_time,
+        )
+
+
+def store_traceroute_packet(packet: Mapping, decoded: Mapping) -> None:
+    """Persist traceroute details and the observed hop path to the API.
+
+    Hop lists can arrive under several key names (``hops``, ``path``,
+    ``route``) and may appear at multiple nesting levels.  All candidates are
+    deduplicated and merged into a single ordered list.
+
+    Parameters:
+        packet: Raw packet metadata from the Meshtastic interface.
+        decoded: Decoded payload containing the traceroute section.
+
+    Returns:
+        ``None``. The traceroute payload is queued for HTTP submission, or
+        silently dropped when identifiers are entirely absent.
+    """
+
+    traceroute_section = (
+        decoded.get("traceroute") if isinstance(decoded, Mapping) else None
+    )
+    request_id = _coerce_int(
+        _first(
+            traceroute_section,
+            "requestId",
+            "request_id",
+            default=_first(decoded, "req", "requestId", "request_id", default=None),
+        )
+    )
+    pkt_id = _coerce_int(_first(packet, "id", "packet_id", "packetId", default=None))
+    if pkt_id is None:
+        pkt_id = request_id
+
+    rx_time = _coerce_int(_first(packet, "rxTime", "rx_time", default=time.time()))
+    if rx_time is None:
+        rx_time = int(time.time())
+
+    src = _coerce_int(
+        _first(
+            decoded,
+            "src",
+            "source",
+            default=_first(packet, "fromId", "from_id", "from", default=None),
+        )
+    )
+    dest = _coerce_int(
+        _first(
+            decoded,
+            "dest",
+            "destination",
+            default=_first(packet, "toId", "to_id", "to", default=None),
+        )
+    )
+
+    metrics = traceroute_section if isinstance(traceroute_section, Mapping) else {}
+    rssi = _coerce_int(
+        _first(metrics, "rssi", default=_first(packet, "rssi", "rx_rssi", "rxRssi"))
+    )
+    snr = _coerce_float(
+        _first(metrics, "snr", default=_first(packet, "snr", "rx_snr", "rxSnr"))
+    )
+    elapsed_ms = _coerce_int(
+        _first(metrics, "elapsed_ms", "latency_ms", "latencyMs", default=None)
+    )
+
+    # Hops can appear under multiple keys at different nesting levels; collect
+    # all candidates and deduplicate while preserving first-seen order.
+    hop_candidates = (
+        _first(metrics, "hops", default=None),
+        _first(metrics, "path", default=None),
+        _first(metrics, "route", default=None),
+        _first(decoded, "hops", default=None),
+        _first(decoded, "path", default=None),
+        (
+            _first(traceroute_section, "route", default=None)
+            if isinstance(traceroute_section, Mapping)
+            else None
+        ),
+    )
+    hops: list[int] = []
+    seen_hops: set[int] = set()
+    for candidate in hop_candidates:
+        for hop in _normalize_trace_hops(candidate):
+            if hop in seen_hops:
+                continue
+            seen_hops.add(hop)
+            hops.append(hop)
+
+    if pkt_id is None and request_id is None and not hops:
+        _record_ignored_packet(packet, reason="traceroute-missing-identifiers")
+        return
+
+    payload = {
+        "id": pkt_id,
+        "request_id": request_id,
+        "src": src,
+        "dest": dest,
+        "rx_time": rx_time,
+        "rx_iso": _iso(rx_time),
+        "hops": hops,
+        "rssi": rssi,
+        "snr": snr,
+        "elapsed_ms": elapsed_ms,
+        "ingestor": _state.host_node_id(),
+    }
+
+    queue._queue_post_json(
+        "/api/traces",
+        _apply_radio_metadata(payload),
+        priority=queue._TRACE_POST_PRIORITY,
+    )
+
+    if config.DEBUG:
+        config._debug_log(
+            "Queued traceroute payload",
+            context="handlers.store_traceroute_packet",
+            request_id=request_id,
+            src=src,
+            dest=dest,
+            hop_count=len(hops),
+        )
+
+
+__all__ = [
+    "base64_payload",
+    "store_position_packet",
+    "store_traceroute_packet",
+]

data/mesh_ingestor/handlers/radio.py

@@ -0,0 +1,94 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Radio metadata helpers for enriching API payloads.
+
+LoRa radio parameters (frequency and modem preset) are captured once at
+connection time by :mod:`data.mesh_ingestor.interfaces` and stored on the
+:mod:`data.mesh_ingestor.config` module.  The helpers here read those cached
+values and attach them to outgoing payloads so the web dashboard can display
+radio configuration alongside mesh data.
+"""
+
+from __future__ import annotations
+
+from .. import config
+
+
+def _radio_metadata_fields() -> dict[str, object]:
+    """Return the shared radio metadata fields for payload enrichment.
+
+    Reads ``LORA_FREQ`` and ``MODEM_PRESET`` from :mod:`config` and returns
+    only the keys that have been populated (i.e. skips ``None`` values).
+
+    Returns:
+        A dictionary containing zero, one, or both of ``lora_freq`` and
+        ``modem_preset`` depending on what is available.
+    """
+
+    metadata: dict[str, object] = {}
+    freq = getattr(config, "LORA_FREQ", None)
+    if freq is not None:
+        metadata["lora_freq"] = freq
+    preset = getattr(config, "MODEM_PRESET", None)
+    if preset is not None:
+        metadata["modem_preset"] = preset
+    return metadata
+
+
+def _apply_radio_metadata(payload: dict) -> dict:
+    """Augment a flat payload dict with radio metadata when available.
+
+    Parameters:
+        payload: Mutable dictionary that will receive radio metadata keys.
+
+    Returns:
+        The same ``payload`` dict with radio metadata keys merged in-place.
+    """
+
+    metadata = _radio_metadata_fields()
+    if metadata:
+        payload.update(metadata)
+    return payload
+
+
+def _apply_radio_metadata_to_nodes(payload: dict) -> dict:
+    """Attach radio metadata to each node entry stored in ``payload``.
+
+    Node upsert payloads are keyed by node ID; each value is a dict of node
+    attributes.  This function enriches every node-value dict with radio
+    metadata so the dashboard can show the radio configuration that was active
+    when the node was last heard.
+
+    Parameters:
+        payload: Mapping of ``node_id → node_dict`` to enrich in-place.
+
+    Returns:
+        The same ``payload`` dict after in-place mutation of its node entries.
+    """
+
+    metadata = _radio_metadata_fields()
+    if not metadata:
+        return payload
+    for value in payload.values():
+        if isinstance(value, dict):
+            value.update(metadata)
+    return payload
+
+
+__all__ = [
+    "_apply_radio_metadata",
+    "_apply_radio_metadata_to_nodes",
+    "_radio_metadata_fields",
+]

data/mesh_ingestor/handlers/telemetry.py

@@ -0,0 +1,563 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Handlers for telemetry and router-heartbeat packets."""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Mapping
+
+from .. import config, queue
+from ..serialization import (
+    _canonical_node_id,
+    _coerce_float,
+    _coerce_int,
+    _extract_payload_bytes,
+    _first,
+    _iso,
+    _node_num_from_id,
+)
+from . import _state
+from .position import base64_payload
+from .radio import _apply_radio_metadata, _apply_radio_metadata_to_nodes
+
+_VALID_TELEMETRY_TYPES: frozenset[str] = frozenset(
+    {"device", "environment", "power", "air_quality"}
+)
+"""Allowed discriminator values for the ``telemetry_type`` field.
+
+Meshtastic uses a protobuf ``oneof`` so only one metric sub-object can be
+populated per packet.  Values outside this set indicate a firmware version
+that added a new type not yet handled here; those are logged and dropped to
+avoid persisting unexpected data shapes.
+"""
+
+
+def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None:
+    """Persist telemetry metrics extracted from a packet.
+
+    Handles all four Meshtastic telemetry sub-types (device, environment,
+    power, air quality) by extracting common fields first and then
+    conditionally adding type-specific metric keys.
+
+    Host telemetry is rate-limited: if the locally connected node's own
+    telemetry arrives within the suppression window it is silently dropped to
+    avoid constant self-updates overwriting other node data.
+
+    Parameters:
+        packet: Packet metadata received from the radio interface.
+        decoded: Meshtastic-decoded view containing telemetry structures.
+
+    Returns:
+        ``None``. The telemetry payload is added to the HTTP queue.
+    """
+
+    telemetry_section = (
+        decoded.get("telemetry") if isinstance(decoded, Mapping) else None
+    )
+    if not isinstance(telemetry_section, Mapping):
+        return
+
+    pkt_id = _coerce_int(_first(packet, "id", "packet_id", "packetId", default=None))
+    if pkt_id is None:
+        return
+
+    raw_from = _first(packet, "fromId", "from_id", "from", default=None)
+    node_id = _canonical_node_id(raw_from)
+    node_num = _coerce_int(_first(decoded, "num", "node_num", default=None))
+    if node_num is None:
+        node_num = _node_num_from_id(node_id or raw_from)
+
+    to_id = _first(packet, "toId", "to_id", "to", default=None)
+
+    raw_rx_time = _first(packet, "rxTime", "rx_time", default=time.time())
+    try:
+        rx_time = int(raw_rx_time)
+    except (TypeError, ValueError):
+        rx_time = int(time.time())
+    rx_iso = _iso(rx_time)
+
+    host_id = _state.host_node_id()
+    # The locally connected node broadcasts its own telemetry frequently.
+    # Accepting every packet would overwrite the host's profile more often
+    # than necessary; the suppression window (default 1 h) rate-limits
+    # self-updates without blocking telemetry from other nodes.
+    if host_id is not None and node_id == host_id:
+        suppressed, minutes_remaining = _state._host_telemetry_suppressed(rx_time)
+        if suppressed:
+            config._debug_log(
+                "Suppressed host telemetry update",
+                context="handlers.store_telemetry",
+                host_node_id=host_id,
+                minutes_remaining=minutes_remaining,
+            )
+            return
+        _state._mark_host_telemetry_seen(rx_time)
+
+    telemetry_time = _coerce_int(_first(telemetry_section, "time", default=None))
+
+    _dm = telemetry_section.get("deviceMetrics") or telemetry_section.get(
+        "device_metrics"
+    )
+    _em = telemetry_section.get("environmentMetrics") or telemetry_section.get(
+        "environment_metrics"
+    )
+    _pm = telemetry_section.get("powerMetrics") or telemetry_section.get(
+        "power_metrics"
+    )
+    _aq = telemetry_section.get("airQualityMetrics") or telemetry_section.get(
+        "air_quality_metrics"
+    )
+    # Priority order matters: deviceMetrics is checked first because the device
+    # sub-object also carries a voltage field that overlaps with powerMetrics.
+    # Meshtastic uses a protobuf oneof so only one sub-object can be populated per
+    # packet; the elif chain handles any hypothetical overlap from future providers.
+    if isinstance(_dm, Mapping):
+        telemetry_type: str | None = "device"
+    elif isinstance(_em, Mapping):
+        telemetry_type = "environment"
+    elif isinstance(_pm, Mapping):
+        telemetry_type = "power"
+    elif isinstance(_aq, Mapping):
+        telemetry_type = "air_quality"
+    else:
+        telemetry_type = None
+
+    if telemetry_type is not None and telemetry_type not in _VALID_TELEMETRY_TYPES:
+        config._debug_log(
+            "Unexpected telemetry_type value; dropping field",
+            context="handlers.store_telemetry",
+            severity="warning",
+            always=True,
+            telemetry_type=telemetry_type,
+        )
+        telemetry_type = None
+
+    channel = _coerce_int(_first(decoded, "channel", default=None))
+    if channel is None:
+        channel = _coerce_int(_first(packet, "channel", default=None))
+    if channel is None:
+        channel = 0
+
+    portnum = _first(decoded, "portnum", default=None)
+    portnum = str(portnum) if portnum not in {None, ""} else None
+
+    bitfield = _coerce_int(_first(decoded, "bitfield", default=None))
+
+    snr = _coerce_float(_first(packet, "snr", "rx_snr", "rxSnr", default=None))
+    rssi = _coerce_int(_first(packet, "rssi", "rx_rssi", "rxRssi", default=None))
+    hop_limit = _coerce_int(_first(packet, "hopLimit", "hop_limit", default=None))
+
+    payload_bytes = _extract_payload_bytes(decoded)
+    payload_b64 = base64_payload(payload_bytes) or ""
+
+    battery_level = _coerce_float(
+        _first(
+            telemetry_section,
+            "batteryLevel",
+            "battery_level",
+            "deviceMetrics.batteryLevel",
+            "environmentMetrics.battery_level",
+            "deviceMetrics.battery_level",
+            default=None,
+        )
+    )
+    voltage = _coerce_float(
+        _first(
+            telemetry_section,
+            "voltage",
+            "environmentMetrics.voltage",
+            "deviceMetrics.voltage",
+            default=None,
+        )
+    )
+    channel_utilization = _coerce_float(
+        _first(
+            telemetry_section,
+            "channelUtilization",
+            "channel_utilization",
+            "deviceMetrics.channelUtilization",
+            "deviceMetrics.channel_utilization",
+            default=None,
+        )
+    )
+    air_util_tx = _coerce_float(
+        _first(
+            telemetry_section,
+            "airUtilTx",
+            "air_util_tx",
+            "deviceMetrics.airUtilTx",
+            "deviceMetrics.air_util_tx",
+            default=None,
+        )
+    )
+    uptime_seconds = _coerce_int(
+        _first(
+            telemetry_section,
+            "uptimeSeconds",
+            "uptime_seconds",
+            "deviceMetrics.uptimeSeconds",
+            "deviceMetrics.uptime_seconds",
+            default=None,
+        )
+    )
+
+    temperature = _coerce_float(
+        _first(
+            telemetry_section,
+            "temperature",
+            "environmentMetrics.temperature",
+            default=None,
+        )
+    )
+    relative_humidity = _coerce_float(
+        _first(
+            telemetry_section,
+            "relativeHumidity",
+            "relative_humidity",
+            "environmentMetrics.relativeHumidity",
+            "environmentMetrics.relative_humidity",
+            default=None,
+        )
+    )
+    barometric_pressure = _coerce_float(
+        _first(
+            telemetry_section,
+            "barometricPressure",
+            "barometric_pressure",
+            "environmentMetrics.barometricPressure",
+            "environmentMetrics.barometric_pressure",
+            default=None,
+        )
+    )
+
+    current = _coerce_float(
+        _first(
+            telemetry_section,
+            "current",
+            "deviceMetrics.current",
+            "deviceMetrics.current_ma",
+            "deviceMetrics.currentMa",
+            "environmentMetrics.current",
+            default=None,
+        )
+    )
+    gas_resistance = _coerce_float(
+        _first(
+            telemetry_section,
+            "gasResistance",
+            "gas_resistance",
+            "environmentMetrics.gasResistance",
+            "environmentMetrics.gas_resistance",
+            default=None,
+        )
+    )
+    iaq = _coerce_int(
+        _first(
+            telemetry_section,
+            "iaq",
+            "environmentMetrics.iaq",
+            "environmentMetrics.iaqIndex",
+            "environmentMetrics.iaq_index",
+            default=None,
+        )
+    )
+    distance = _coerce_float(
+        _first(
+            telemetry_section,
+            "distance",
+            "environmentMetrics.distance",
+            "environmentMetrics.range",
+            "environmentMetrics.rangeMeters",
+            default=None,
+        )
+    )
+    lux = _coerce_float(
+        _first(
+            telemetry_section,
+            "lux",
+            "environmentMetrics.lux",
+            "environmentMetrics.illuminance",
+            default=None,
+        )
+    )
+    white_lux = _coerce_float(
+        _first(
+            telemetry_section,
+            "whiteLux",
+            "white_lux",
+            "environmentMetrics.whiteLux",
+            "environmentMetrics.white_lux",
+            default=None,
+        )
+    )
+    ir_lux = _coerce_float(
+        _first(
+            telemetry_section,
+            "irLux",
+            "ir_lux",
+            "environmentMetrics.irLux",
+            "environmentMetrics.ir_lux",
+            default=None,
+        )
+    )
+    uv_lux = _coerce_float(
+        _first(
+            telemetry_section,
+            "uvLux",
+            "uv_lux",
+            "environmentMetrics.uvLux",
+            "environmentMetrics.uv_lux",
+            "environmentMetrics.uvIndex",
+            default=None,
+        )
+    )
+    wind_direction = _coerce_int(
+        _first(
+            telemetry_section,
+            "windDirection",
+            "wind_direction",
+            "environmentMetrics.windDirection",
+            "environmentMetrics.wind_direction",
+            default=None,
+        )
+    )
+    wind_speed = _coerce_float(
+        _first(
+            telemetry_section,
+            "windSpeed",
+            "wind_speed",
+            "environmentMetrics.windSpeed",
+            "environmentMetrics.wind_speed",
+            "environmentMetrics.windSpeedMps",
+            default=None,
+        )
+    )
+    wind_gust = _coerce_float(
+        _first(
+            telemetry_section,
+            "windGust",
+            "wind_gust",
+            "environmentMetrics.windGust",
+            "environmentMetrics.wind_gust",
+            default=None,
+        )
+    )
+    wind_lull = _coerce_float(
+        _first(
+            telemetry_section,
+            "windLull",
+            "wind_lull",
+            "environmentMetrics.windLull",
+            "environmentMetrics.wind_lull",
+            default=None,
+        )
+    )
+    weight = _coerce_float(
+        _first(
+            telemetry_section,
+            "weight",
+            "environmentMetrics.weight",
+            "environmentMetrics.mass",
+            default=None,
+        )
+    )
+    radiation = _coerce_float(
+        _first(
+            telemetry_section,
+            "radiation",
+            "environmentMetrics.radiation",
+            "environmentMetrics.radiationLevel",
+            default=None,
+        )
+    )
+    rainfall_1h = _coerce_float(
+        _first(
+            telemetry_section,
+            "rainfall1h",
+            "rainfall_1h",
+            "environmentMetrics.rainfall1h",
+            "environmentMetrics.rainfall_1h",
+            "environmentMetrics.rainfallOneHour",
+            default=None,
+        )
+    )
+    rainfall_24h = _coerce_float(
+        _first(
+            telemetry_section,
+            "rainfall24h",
+            "rainfall_24h",
+            "environmentMetrics.rainfall24h",
+            "environmentMetrics.rainfall_24h",
+            "environmentMetrics.rainfallTwentyFourHour",
+            default=None,
+        )
+    )
+    soil_moisture = _coerce_int(
+        _first(
+            telemetry_section,
+            "soilMoisture",
+            "soil_moisture",
+            "environmentMetrics.soilMoisture",
+            "environmentMetrics.soil_moisture",
+            default=None,
+        )
+    )
+    soil_temperature = _coerce_float(
+        _first(
+            telemetry_section,
+            "soilTemperature",
+            "soil_temperature",
+            "environmentMetrics.soilTemperature",
+            "environmentMetrics.soil_temperature",
+            default=None,
+        )
+    )
+
+    telemetry_payload = {
+        "id": pkt_id,
+        "node_id": node_id,
+        "node_num": node_num,
+        "from_id": node_id or raw_from,
+        "to_id": to_id,
+        "rx_time": rx_time,
+        "rx_iso": rx_iso,
+        "telemetry_time": telemetry_time,
+        "channel": channel,
+        "portnum": portnum,
+        "bitfield": bitfield,
+        "snr": snr,
+        "rssi": rssi,
+        "hop_limit": hop_limit,
+        "payload_b64": payload_b64,
+        "ingestor": _state.host_node_id(),
+    }
+
+    # Conditionally include metric keys so the API ignores absent fields rather
+    # than overwriting existing values with null.
+    if battery_level is not None:
+        telemetry_payload["battery_level"] = battery_level
+    if voltage is not None:
+        telemetry_payload["voltage"] = voltage
+    if channel_utilization is not None:
+        telemetry_payload["channel_utilization"] = channel_utilization
+    if air_util_tx is not None:
+        telemetry_payload["air_util_tx"] = air_util_tx
+    if uptime_seconds is not None:
+        telemetry_payload["uptime_seconds"] = uptime_seconds
+    if temperature is not None:
+        telemetry_payload["temperature"] = temperature
+    if relative_humidity is not None:
+        telemetry_payload["relative_humidity"] = relative_humidity
+    if barometric_pressure is not None:
+        telemetry_payload["barometric_pressure"] = barometric_pressure
+    if current is not None:
+        telemetry_payload["current"] = current
+    if gas_resistance is not None:
+        telemetry_payload["gas_resistance"] = gas_resistance
+    if iaq is not None:
+        telemetry_payload["iaq"] = iaq
+    if distance is not None:
+        telemetry_payload["distance"] = distance
+    if lux is not None:
+        telemetry_payload["lux"] = lux
+    if white_lux is not None:
+        telemetry_payload["white_lux"] = white_lux
+    if ir_lux is not None:
+        telemetry_payload["ir_lux"] = ir_lux
+    if uv_lux is not None:
+        telemetry_payload["uv_lux"] = uv_lux
+    if wind_direction is not None:
+        telemetry_payload["wind_direction"] = wind_direction
+    if wind_speed is not None:
+        telemetry_payload["wind_speed"] = wind_speed
+    if wind_gust is not None:
+        telemetry_payload["wind_gust"] = wind_gust
+    if wind_lull is not None:
+        telemetry_payload["wind_lull"] = wind_lull
+    if weight is not None:
+        telemetry_payload["weight"] = weight
+    if radiation is not None:
+        telemetry_payload["radiation"] = radiation
+    if rainfall_1h is not None:
+        telemetry_payload["rainfall_1h"] = rainfall_1h
+    if rainfall_24h is not None:
+        telemetry_payload["rainfall_24h"] = rainfall_24h
+    if soil_moisture is not None:
+        telemetry_payload["soil_moisture"] = soil_moisture
+    if soil_temperature is not None:
+        telemetry_payload["soil_temperature"] = soil_temperature
+    if telemetry_type is not None:
+        telemetry_payload["telemetry_type"] = telemetry_type
+
+    queue._queue_post_json(
+        "/api/telemetry",
+        _apply_radio_metadata(telemetry_payload),
+        priority=queue._TELEMETRY_POST_PRIORITY,
+    )
+
+    if config.DEBUG:
+        config._debug_log(
+            "Queued telemetry payload",
+            context="handlers.store_telemetry",
+            node_id=node_id,
+            battery_level=battery_level,
+            voltage=voltage,
+        )
+
+
+def store_router_heartbeat_packet(packet: Mapping) -> None:
+    """Persist a ``STORE_FORWARD_APP ROUTER_HEARTBEAT`` as a node presence update.
+
+    The heartbeat carries no message payload — the only actionable signal is
+    that the store-and-forward router is alive at the observed ``rx_time``.
+    All other fields are left untouched so the router's existing profile is
+    not overwritten.
+
+    Parameters:
+        packet: Raw packet metadata.
+
+    Returns:
+        ``None``. A minimal node upsert is enqueued at low priority.
+    """
+
+    node_id = _canonical_node_id(
+        _first(packet, "fromId", "from_id", "from", default=None)
+    )
+    if node_id is None:
+        return
+
+    rx_time = int(_first(packet, "rxTime", "rx_time", default=time.time()))
+
+    node_payload: dict = {"lastHeard": rx_time}
+    nodes_payload = _apply_radio_metadata_to_nodes({node_id: node_payload})
+    nodes_payload["ingestor"] = _state.host_node_id()
+    queue._queue_post_json(
+        "/api/nodes", nodes_payload, priority=queue._DEFAULT_POST_PRIORITY
+    )
+
+    if config.DEBUG:
+        config._debug_log(
+            "Queued router heartbeat node upsert",
+            context="handlers.store_router_heartbeat",
+            node_id=node_id,
+            rx_time=rx_time,
+        )
+
+
+__all__ = [
+    "store_router_heartbeat_packet",
+    "store_telemetry_packet",
+]

data/mesh_ingestor/interfaces.py

@@ -157,7 +157,21 @@ def _candidate_node_id(mapping: Mapping | None) -> str | None:
 
 
 def _extract_host_node_id(iface) -> str | None:
-    """Return the canonical node identifier for the connected host device."""
+    """Return the canonical node identifier for the connected host device.
+
+    Searches a sequence of well-known attribute names (``myInfo``,
+    ``my_node_info``, etc.) on ``iface`` for a mapping that contains a
+    recognisable node identifier, then falls back to the raw ``myNodeNum``
+    integer attribute.
+
+    Parameters:
+        iface: Live Meshtastic interface object, or any object that exposes
+            node-identity attributes in one of the expected forms.
+
+    Returns:
+        A canonical ``!xxxxxxxx`` node identifier, or ``None`` when no
+        identifiable host node information is available.
+    """
 
     if iface is None:
         return None
@@ -245,6 +259,9 @@ def _patch_meshtastic_nodeinfo_handler() -> None:
         with contextlib.suppress(Exception):
             mesh_interface_module = importlib.import_module("meshtastic.mesh_interface")
 
+    # Replace the module-level handler only once; the sentinel attribute prevents
+    # re-wrapping if _patch_meshtastic_nodeinfo_handler() is called again after
+    # the interface module is reloaded or re-imported.
     if not getattr(original, "_potato_mesh_safe_wrapper", False):
         module._onNodeInfoReceive = _build_safe_nodeinfo_callback(original)
 
@@ -303,6 +320,22 @@ def _patch_nodeinfo_handler_class(
         """Subclass that guards against missing node identifiers."""
 
         def onReceive(self, iface, packet):  # type: ignore[override]
+            """Normalise ``packet`` before dispatching to the parent handler.
+
+            Injects a canonical ``id`` field when one can be inferred from the
+            packet's other fields, then delegates to the original
+            ``NodeInfoHandler.onReceive``.  A ``KeyError`` on ``"id"`` is
+            suppressed because some firmware versions omit the field entirely.
+
+            Parameters:
+                iface: The Meshtastic interface that received the packet.
+                packet: Raw nodeinfo packet dict, possibly lacking an ``id``
+                    key.
+
+            Returns:
+                The return value of the parent handler, or ``None`` when a
+                missing ``"id"`` key would otherwise raise.
+            """
             normalised = _normalise_nodeinfo_packet(packet)
             if normalised is not None:
                 packet = normalised
@@ -638,6 +671,7 @@ class _DummySerialInterface:
         self.nodes: dict = {}
 
     def close(self) -> None:  # pragma: no cover - nothing to close
+        """No-op: the dummy interface holds no resources to release."""
         pass
 
 
@@ -688,6 +722,9 @@ def _parse_network_target(value: str) -> tuple[str, int] | None:
         if result:
             return result
 
+    # For bare "host:port" strings that urlparse may misparse, try a manual
+    # partition. The `startswith("[")` guard excludes IPv6 bracket notation
+    # (e.g. "[::1]:8080") because those already succeed via urlparse above.
     if value.count(":") == 1 and not value.startswith("["):
         host, _, port_text = value.partition(":")
         try:

data/mesh_ingestor/providers/meshtastic.py

@@ -16,11 +16,10 @@
 
 from __future__ import annotations
 
-import time
-
 from pubsub import pub
 
 from .. import config, daemon as _daemon, handlers, interfaces
+from ..utils import _retry_dict_snapshot
 
 
 class MeshtasticProvider:
@@ -73,19 +72,29 @@ class MeshtasticProvider:
         return interfaces._extract_host_node_id(iface)
 
     def node_snapshot_items(self, iface: object) -> list[tuple[str, object]]:
+        """Return a stable snapshot of all known nodes from ``iface``.
+
+        Uses :func:`~data.mesh_ingestor.utils._retry_dict_snapshot` to
+        tolerate concurrent modifications from the Meshtastic background
+        thread.
+
+        Parameters:
+            iface: Live Meshtastic interface whose ``nodes`` dict to snapshot.
+
+        Returns:
+            List of ``(node_id, node_dict)`` tuples, or an empty list when
+            the snapshot fails after retries.
+        """
+
         nodes = getattr(iface, "nodes", {}) or {}
-        for _ in range(3):
-            try:
-                return list(nodes.items())
-            except RuntimeError as err:
-                if "dictionary changed size during iteration" not in str(err):
-                    raise
-                time.sleep(0)
-        config._debug_log(
-            "Skipping node snapshot due to concurrent modification",
-            context="meshtastic.snapshot",
-        )
-        return []
+        result = _retry_dict_snapshot(lambda: list(nodes.items()))
+        if result is None:
+            config._debug_log(
+                "Skipping node snapshot due to concurrent modification",
+                context="meshtastic.snapshot",
+            )
+            return []
+        return result
 
 
 __all__ = ["MeshtasticProvider"]

data/mesh_ingestor/queue.py

@@ -172,6 +172,10 @@ def _enqueue_post_json(
 
     with state.lock:
         counter = next(state.counter)
+        # Heap tuple: (priority, counter, path, payload).  Lower priority
+        # values are dequeued first (min-heap semantics).  The monotonically
+        # increasing counter breaks ties so equal-priority items are processed
+        # in FIFO order without comparing the non-orderable payload dict.
         heapq.heappush(state.queue, (priority, counter, path, payload))
 
 

data/mesh_ingestor/serialization.py

@@ -128,6 +128,10 @@ def _load_cli_role_lookup() -> dict[int, str]:
                 mapping[key_int] = str(value)
         return mapping
 
+    # Iterate through candidate module paths in preference order.  The CLI
+    # package ships several role-enum locations across versions; we stop at
+    # the first module that yields a non-empty mapping so we do not silently
+    # merge partial enums from two different meshtastic-cli releases.
     for module_name in _CLI_ROLE_MODULE_NAMES:
         try:
             module = importlib.import_module(module_name)

data/mesh_ingestor/utils.py

@@ -0,0 +1,56 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Shared utility helpers for the mesh ingestor package."""
+
+from __future__ import annotations
+
+import time
+from typing import Callable, TypeVar
+
+_T = TypeVar("_T")
+
+
+def _retry_dict_snapshot(fn: Callable[[], _T], retries: int = 3) -> _T | None:
+    """Call ``fn()`` retrying on concurrent dictionary-modification errors.
+
+    Meshtastic's node dictionary is updated on a background thread. Iterating
+    it can raise a :class:`RuntimeError` with the message "dictionary changed
+    size during iteration".  This helper retries the call up to ``retries``
+    times, yielding the thread scheduler between attempts via :func:`time.sleep`.
+
+    Parameters:
+        fn: Zero-argument callable that performs the iteration.
+        retries: Maximum number of attempts before giving up.
+
+    Returns:
+        The return value of ``fn`` on success, or ``None`` when all retries are
+        exhausted.
+    """
+
+    for _ in range(max(1, retries)):
+        try:
+            return fn()
+        except RuntimeError as err:
+            # Only retry the specific concurrent-modification error; re-raise
+            # anything else so genuine bugs surface immediately.
+            if "dictionary changed size during iteration" not in str(err):
+                raise
+            # Yield to the thread scheduler to let the mutating thread complete
+            # before we attempt the snapshot again.
+            time.sleep(0)
+    return None
+
+
+__all__ = ["_retry_dict_snapshot"]