rate limit host device telemetry (#467)

* rate limit host device telemetry

* Spec: add more unit tests

data/mesh_ingestor/daemon.py

@@ -284,6 +284,9 @@ def main(existing_interface=None) -> None:
                         active_candidate = resolved_target
                     interfaces._ensure_radio_metadata(iface)
                     interfaces._ensure_channel_metadata(iface)
+                    handlers.register_host_node_id(
+                        interfaces._extract_host_node_id(iface)
+                    )
                     retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS)
                     initial_snapshot_sent = False
                     if not announced_target and resolved_target:

data/mesh_ingestor/handlers.py

@@ -20,6 +20,7 @@ import base64
 import contextlib
 import importlib
 import json
+import math
 import sys
 import threading
 import time
@@ -35,6 +36,15 @@ _IGNORED_PACKET_LOG_PATH = Path(__file__).resolve().parents[2] / "ignored.txt"
 _IGNORED_PACKET_LOCK = threading.Lock()
 """Lock guarding writes to :data:`_IGNORED_PACKET_LOG_PATH`."""
 
+_HOST_TELEMETRY_INTERVAL_SECS = 60 * 60
+"""Minimum interval between accepted host telemetry packets."""
+
+_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."""
+
 
 def _ignored_packet_default(value: object) -> object:
     """Return a JSON-serialisable representation for ignored packet data."""
@@ -90,6 +100,50 @@ from .serialization import (
 )
 
 
+def register_host_node_id(node_id: str | None) -> None:
+    """Record the canonical identifier for the connected host device.
+
+    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."""
+
+    return _host_node_id
+
+
+def _mark_host_telemetry_seen(rx_time: int) -> None:
+    """Update the last receive time for the host telemetry window."""
+
+    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."""
+
+    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 _radio_metadata_fields() -> dict[str, object]:
     """Return the shared radio metadata fields for payload enrichment."""
 
@@ -534,6 +588,19 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None:
         rx_time = int(time.time())
     rx_iso = _iso(rx_time)
 
+    host_id = host_node_id()
+    if host_id is not None and node_id == host_id:
+        suppressed, minutes_remaining = _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
+        _mark_host_telemetry_seen(rx_time)
+
     telemetry_time = _coerce_int(_first(telemetry_section, "time", default=None))
 
     channel = _coerce_int(_first(decoded, "channel", default=None))
@@ -1492,8 +1559,10 @@ def on_receive(packet, interface) -> None:
 
 __all__ = [
     "_queue_post_json",
+    "host_node_id",
     "last_packet_monotonic",
     "on_receive",
+    "register_host_node_id",
     "store_neighborinfo_packet",
     "store_nodeinfo_packet",
     "store_packet_dict",

data/mesh_ingestor/interfaces.py

@@ -113,6 +113,47 @@ def _candidate_node_id(mapping: Mapping | None) -> str | None:
     return None
 
 
+def _extract_host_node_id(iface) -> str | None:
+    """Return the canonical node identifier for the connected host device."""
+
+    if iface is None:
+        return None
+
+    def _as_mapping(candidate) -> Mapping | None:
+        mapping = _ensure_mapping(candidate)
+        if mapping is not None:
+            return mapping
+        if callable(candidate):
+            with contextlib.suppress(Exception):
+                return _ensure_mapping(candidate())
+        return None
+
+    candidates: list[Mapping] = []
+    for attr in ("myInfo", "my_node_info", "myNodeInfo", "my_node", "localNode"):
+        mapping = _as_mapping(getattr(iface, attr, None))
+        if mapping is None:
+            continue
+        candidates.append(mapping)
+        nested_info = _ensure_mapping(mapping.get("info"))
+        if nested_info:
+            candidates.append(nested_info)
+
+    for mapping in candidates:
+        node_id = _candidate_node_id(mapping)
+        if node_id:
+            return node_id
+        for key in ("myNodeNum", "my_node_num", "myNodeId", "my_node_id"):
+            node_id = serialization._canonical_node_id(mapping.get(key))
+            if node_id:
+                return node_id
+
+    node_id = serialization._canonical_node_id(getattr(iface, "myNodeNum", None))
+    if node_id:
+        return node_id
+
+    return None
+
+
 def _normalise_nodeinfo_packet(packet) -> dict | None:
     """Return a dictionary view of ``packet`` with a guaranteed ``id`` when known."""
 
@@ -771,6 +812,7 @@ __all__ = [
     "NoAvailableMeshInterface",
     "_ensure_channel_metadata",
     "_ensure_radio_metadata",
+    "_extract_host_node_id",
     "_DummySerialInterface",
     "_DEFAULT_TCP_PORT",
     "_DEFAULT_TCP_TARGET",

tests/test_mesh.py

@@ -228,6 +228,60 @@ def test_snapshot_interval_defaults_to_60_seconds(mesh_module):
     assert mesh.SNAPSHOT_SECS == 60
 
 
+def test_extract_host_node_id_prefers_my_info_fields(mesh_module):
+    mesh = mesh_module
+
+    class DummyInterface:
+        def __init__(self):
+            self.myInfo = {"my_node_num": 0x9E95CF60}
+
+    iface = DummyInterface()
+
+    assert mesh._extract_host_node_id(iface) == "!9e95cf60"
+
+
+def test_extract_host_node_id_from_nested_info(mesh_module):
+    mesh = mesh_module
+
+    class DummyInterface:
+        def __init__(self):
+            self.myInfo = {"info": {"id": "!cafebabe"}}
+
+    iface = DummyInterface()
+
+    assert mesh._extract_host_node_id(iface) == "!cafebabe"
+
+
+def test_extract_host_node_id_from_callable(mesh_module):
+    mesh = mesh_module
+
+    class CallableNoDict:
+        __slots__ = ()
+
+        def __call__(self):
+            return {"id": "!f00ba4"}
+
+    class DummyInterface:
+        def __init__(self):
+            self.localNode = CallableNoDict()
+
+    iface = DummyInterface()
+
+    assert mesh._extract_host_node_id(iface) == "!00f00ba4"
+
+
+def test_extract_host_node_id_from_my_node_num_attribute(mesh_module):
+    mesh = mesh_module
+
+    class DummyInterface:
+        def __init__(self):
+            self.myNodeNum = 0xDEADBEEF
+
+    iface = DummyInterface()
+
+    assert mesh._extract_host_node_id(iface) == "!deadbeef"
+
+
 @pytest.mark.parametrize("value", ["mock", "Mock", " disabled "])
 def test_create_serial_interface_allows_mock(mesh_module, value):
     mesh = mesh_module
@@ -1978,6 +2032,57 @@ def test_store_packet_dict_handles_environment_telemetry(mesh_module, monkeypatc
     assert payload["modem_preset"] == "MediumFast"
 
 
+def test_store_packet_dict_throttles_host_telemetry(mesh_module, monkeypatch):
+    mesh = mesh_module
+    captured = []
+    logs = []
+    monkeypatch.setattr(
+        mesh,
+        "_queue_post_json",
+        lambda path, payload, *, priority: captured.append((path, payload, priority)),
+    )
+    monkeypatch.setattr(
+        mesh.config,
+        "_debug_log",
+        lambda message, **metadata: logs.append((message, metadata)),
+    )
+
+    mesh.register_host_node_id("!9e95cf60")
+
+    base_packet = {
+        "id": 1_234,
+        "fromId": "!9e95cf60",
+        "decoded": {
+            "portnum": "TELEMETRY_APP",
+            "telemetry": {
+                "time": 1_000,
+                "deviceMetrics": {
+                    "batteryLevel": 50,
+                },
+            },
+        },
+    }
+
+    mesh.store_packet_dict({**base_packet, "rxTime": 1_000})
+    mesh.store_packet_dict({**base_packet, "id": 1_235, "rxTime": 1_300})
+    mesh.store_packet_dict({**base_packet, "id": 1_236, "rxTime": 4_700})
+
+    assert len(captured) == 2
+    first_path, first_payload, _ = captured[0]
+    second_path, second_payload, _ = captured[1]
+    assert first_path == "/api/telemetry"
+    assert second_path == "/api/telemetry"
+    assert first_payload["id"] == 1_234
+    assert second_payload["id"] == 1_236
+
+    suppression_logs = [
+        entry for entry in logs if entry[0] == "Suppressed host telemetry update"
+    ]
+    assert suppression_logs
+    assert suppression_logs[0][1]["host_node_id"] == "!9e95cf60"
+    assert suppression_logs[0][1]["minutes_remaining"] == 55
+
+
 def test_store_packet_dict_handles_traceroute_packet(mesh_module, monkeypatch):
     mesh = mesh_module
     captured = []