Track advert paths and expose them on contact sensors

custom_components/meshcore/binary_sensor.py

@@ -1,11 +1,11 @@
 """Binary sensor platform for MeshCore integration."""
 from __future__ import annotations
 
 import logging
 import time
 from collections.abc import Callable
 from datetime import datetime
 from typing import Any, Dict
 
 from meshcore.events import EventType
 
@@ -466,7 +466,7 @@
         return True
 
     @property
     def extra_state_attributes(self) -> Dict[str, Any]:
         """Return message details as attributes."""
         attributes = {}
 
@@ -555,7 +555,7 @@
         return self._connected
 
     @property
     def extra_state_attributes(self) -> Dict[str, Any]:
         """Return broker context attributes."""
         return {
             "broker_number": self._broker_num,
@@ -641,7 +641,7 @@
     def device_info(self):
         return DeviceInfo(**self.coordinator.device_info)
 
     def _get_contact_data(self) -> Dict[str, Any]:
         """Get the data for this contact from the coordinator."""
         # Use O(1) lookup by prefix if we have it
         if self.pubkey_prefix:
@@ -658,7 +658,7 @@
 
         return {}
 
     def _update_from_contact_data(self, contact: Dict[str, Any]):
         """Update entity state based on contact data."""
         if not contact:
             return
@@ -713,7 +713,7 @@
         return "fresh" if self.is_on else "stale"
 
     @property
     def extra_state_attributes(self) -> Dict[str, Any]:
         """Return the contact data as attributes."""
         if not self._contact_data:
             return {"status": "unknown"}
@@ -728,7 +728,11 @@
         # Add all contact properties as attributes
         for key, value in self._contact_data.items():
             attributes[key] = value
-
+
+        # Route the last advert took (adv_path/adv_path_len/adv_path_time),
+        # tracked by the coordinator from ADVERTISEMENT pushes.
+        attributes.update(self.coordinator.get_advert_path_data(self.public_key))
+
         attributes["pubkey_short"] = self.public_key[:2] if self.public_key else ""
 
         # Get node type string
@@ -821,7 +825,7 @@
         return (time.time() - last_success) < staleness_window
 
     @property
     def extra_state_attributes(self) -> Dict[str, Any]:
         """Return timing context for the online status."""
         attrs: Dict[str, Any] = {}
         last_success = self.coordinator._last_successful_request.get(self.pubkey_prefix)

custom_components/meshcore/coordinator.py

@@ -145,6 +145,13 @@ def __init__(
         self._hardware_model = None
         self._max_channels = 4  # Default to 4 channels, updated from DEVICE_INFO
         self._channel_info = {}  # Dict keyed by channel_idx to store channel info
+
+        # Advert path tracking: route the last advert from each contact took,
+        # keyed by 12-char public_key prefix. Populated on ADVERTISEMENT pushes.
+        self._advert_paths: Dict[str, Dict[str, Any]] = {}
+        self._advert_path_pending: set[str] = set()
+        self._advert_path_failures = 0
+        self._advert_path_disabled = False
 
         # Create a central device_info dict that all entities can reference
         self.device_info = {
@@ -800,6 +807,70 @@ def handle_channel_info(event: Event):
             handle_channel_info,
         )
         self.logger.debug("Registered CHANNEL_INFO event listener")
+
+    def _setup_advert_path_listener(self) -> None:
+        """Set up ADVERTISEMENT listener to fetch each advert's traversed path."""
+        def handle_advertisement(event: Event):
+            try:
+                public_key = event.payload.get("public_key")
+                if public_key:
+                    self.hass.async_create_task(self._fetch_advert_path(public_key))
+            except Exception as ex:
+                self.logger.error(f"Error handling ADVERTISEMENT event: {ex}")
+
+        self.api.mesh_core.dispatcher.subscribe(
+            EventType.ADVERTISEMENT,
+            handle_advertisement,
+        )
+        self.logger.debug("Registered ADVERTISEMENT event listener")
+
+    async def _fetch_advert_path(self, public_key: str) -> None:
+        """Fetch the path the last advert from this contact took.
+
+        GET_ADVERT_PATH is a local companion query (no RF traffic), so it is
+        not rate-limited. Fetching is disabled after repeated failures so
+        firmware without the command isn't queried on every advert.
+        """
+        prefix = public_key[:12]
+        if self._advert_path_disabled or prefix in self._advert_path_pending:
+            return
+        self._advert_path_pending.add(prefix)
+        success = False
+        try:
+            result = await self.api.mesh_core.commands.get_advert_path(public_key)
+            if result and result.type == EventType.ADVERT_PATH:
+                success = True
+                payload = result.payload or {}
+                self._advert_paths[prefix] = {
+                    "adv_path": payload.get("path", ""),
+                    "adv_path_len": payload.get("path_len", -1),
+                    "adv_path_time": payload.get("timestamp"),
+                }
+                self.mark_contact_dirty(prefix)
+                self.async_update_listeners()
+        except Exception as ex:
+            self.logger.debug(f"Error fetching advert path for {prefix}: {ex}")
+        finally:
+            self._advert_path_pending.discard(prefix)
+        if success:
+            self._advert_path_failures = 0
+        else:
+            self._advert_path_failures += 1
+            if self._advert_path_failures >= 3 and not self._advert_path_disabled:
+                self._advert_path_disabled = True
+                self.logger.info(
+                    "Disabling advert path fetches after repeated failures "
+                    "(firmware may not support GET_ADVERT_PATH)"
+                )
+
+    def get_advert_path_data(self, pubkey_prefix: str) -> Dict[str, Any]:
+        """Advert path attributes for a contact (empty dict until an advert is heard).
+
+        Accepts either full public key or 12-char prefix.
+        """
+        if not pubkey_prefix:
+            return {}
+        return self._advert_paths.get(pubkey_prefix[:12], {})
 
     async def fetch_all_channel_info(self) -> None:
         """Fetch channel info for all channels on startup."""
@@ -1574,6 +1645,9 @@ async def _async_update_data(self) -> Dict[str, Any]:
                     # Set up CHANNEL_INFO event listener
                     self._setup_channel_info_listener()
 
+                    # Set up ADVERTISEMENT listener for advert path tracking
+                    self._setup_advert_path_listener()
+
                     # Fetch channel info for all channels
                     await self.fetch_all_channel_info()
 

docs/docs/contacts.md

@@ -373,6 +373,10 @@ Each contact sensor includes detailed attributes:
 - `last_advert` - Unix timestamp of last advertisement
 - `last_advert_formatted` - ISO formatted timestamp
 - Location data (if available): `latitude`, `longitude`
+- Advert path (once an advert is heard while the integration is running; requires firmware with `GET_ADVERT_PATH` support):
+  - `adv_path` - Route the last advert took, as concatenated 1-byte hop hashes (hex); empty for zero-hop (direct) reception
+  - `adv_path_len` - Number of hops the advert traversed
+  - `adv_path_time` - Timestamp of that advert
 
 ### Entity Icons
 

tests_integration/test_advert_path.py

@@ -0,0 +1,134 @@
+"""Integration-tier tests for advert path tracking.
+
+The coordinator fetches the path each advert took (GET_ADVERT_PATH) when the
+companion pushes an ADVERTISEMENT event, stores it keyed by pubkey prefix, and
+the contact binary sensor merges it into its attributes. These tests drive the
+real coordinator methods and the real sensor class; only the MeshCore API is
+mocked.
+"""
+import logging
+from datetime import timedelta
+from unittest.mock import AsyncMock, MagicMock
+
+from homeassistant.core import HomeAssistant
+from meshcore.events import Event, EventType
+from pytest_homeassistant_custom_component.common import MockConfigEntry
+
+from custom_components.meshcore.binary_sensor import (
+    MeshCoreContactDiagnosticBinarySensor,
+)
+from custom_components.meshcore.const import DOMAIN
+from custom_components.meshcore.coordinator import MeshCoreDataUpdateCoordinator
+
+FULL_PK = "a1" * 32
+PREFIX = FULL_PK[:12]
+ADVERT_PATH_PAYLOAD = {"timestamp": 1754600000, "path_hash_mode": 0, "path_len": 2, "path": "b2c3"}
+
+
+def _coordinator(hass: HomeAssistant) -> MeshCoreDataUpdateCoordinator:
+    config_entry = MockConfigEntry(domain=DOMAIN, data={})
+    config_entry.add_to_hass(hass)
+    coordinator = MeshCoreDataUpdateCoordinator(
+        hass,
+        logging.getLogger(__name__),
+        "test",
+        timedelta(seconds=60),
+        MagicMock(),
+        config_entry,
+    )
+    return coordinator
+
+
+def _mock_advert_path(coordinator, result: Event) -> AsyncMock:
+    mock = AsyncMock(return_value=result)
+    coordinator.api.mesh_core.commands.get_advert_path = mock
+    return mock
+
+
+async def test_fetch_stores_path_and_marks_dirty(hass: HomeAssistant):
+    coordinator = _coordinator(hass)
+    _mock_advert_path(coordinator, Event(EventType.ADVERT_PATH, ADVERT_PATH_PAYLOAD))
+
+    await coordinator._fetch_advert_path(FULL_PK)
+
+    expected = {"adv_path": "b2c3", "adv_path_len": 2, "adv_path_time": 1754600000}
+    assert coordinator.get_advert_path_data(FULL_PK) == expected
+    assert coordinator.get_advert_path_data(PREFIX) == expected  # prefix lookup too
+    assert coordinator.is_contact_dirty(FULL_PK)
+
+
+async def test_unknown_contact_returns_empty(hass: HomeAssistant):
+    coordinator = _coordinator(hass)
+    assert coordinator.get_advert_path_data("") == {}
+    assert coordinator.get_advert_path_data("ff" * 32) == {}
+
+
+async def test_disabled_after_three_consecutive_failures(hass: HomeAssistant):
+    coordinator = _coordinator(hass)
+    mock = _mock_advert_path(coordinator, Event(EventType.ERROR, {}))
+
+    for _ in range(4):
+        await coordinator._fetch_advert_path(FULL_PK)
+
+    # Fourth call short-circuits: the latch opened after the third failure.
+    assert mock.await_count == 3
+    assert coordinator.get_advert_path_data(FULL_PK) == {}
+
+
+async def test_success_resets_failure_count(hass: HomeAssistant):
+    coordinator = _coordinator(hass)
+    err = Event(EventType.ERROR, {})
+    ok = Event(EventType.ADVERT_PATH, ADVERT_PATH_PAYLOAD)
+    mock = _mock_advert_path(coordinator, None)
+    mock.side_effect = [err, err, ok, err, err]
+
+    for _ in range(5):
+        await coordinator._fetch_advert_path(FULL_PK)
+
+    # The success in the middle resets the consecutive-failure count, so the
+    # latch never opens and all five calls reach the device.
+    assert mock.await_count == 5
+
+
+async def test_advertisement_event_triggers_fetch(hass: HomeAssistant):
+    """End to end: the subscribed handler fetches and stores on ADVERTISEMENT."""
+    coordinator = _coordinator(hass)
+    _mock_advert_path(coordinator, Event(EventType.ADVERT_PATH, ADVERT_PATH_PAYLOAD))
+
+    coordinator._setup_advert_path_listener()
+    subscribe_mock = coordinator.api.mesh_core.dispatcher.subscribe
+    event_type, handler = subscribe_mock.call_args[0]
+    assert event_type == EventType.ADVERTISEMENT
+
+    handler(Event(EventType.ADVERTISEMENT, {"public_key": FULL_PK}))
+    await hass.async_block_till_done()
+
+    assert coordinator.get_advert_path_data(FULL_PK)["adv_path"] == "b2c3"
+
+
+async def test_contact_sensor_merges_advert_path_attributes(hass: HomeAssistant):
+    contact = {"adv_name": "peer", "public_key": FULL_PK, "last_advert": 1754600000}
+    coordinator = MagicMock()
+    coordinator.get_contact_by_prefix.return_value = contact
+    coordinator.get_advert_path_data.return_value = {
+        "adv_path": "b2c3",
+        "adv_path_len": 2,
+        "adv_path_time": 1754600000,
+    }
+    sensor = MeshCoreContactDiagnosticBinarySensor(coordinator, "peer", FULL_PK, "uid")
+
+    attributes = sensor.extra_state_attributes
+    assert attributes["adv_path"] == "b2c3"
+    assert attributes["adv_path_len"] == 2
+    assert attributes["adv_path_time"] == 1754600000
+    coordinator.get_advert_path_data.assert_called_with(FULL_PK)
+
+
+async def test_contact_sensor_without_advert_path_has_no_attributes(hass: HomeAssistant):
+    contact = {"adv_name": "peer", "public_key": FULL_PK, "last_advert": 1754600000}
+    coordinator = MagicMock()
+    coordinator.get_contact_by_prefix.return_value = contact
+    coordinator.get_advert_path_data.return_value = {}
+    sensor = MeshCoreContactDiagnosticBinarySensor(coordinator, "peer", FULL_PK, "uid")
+
+    assert "adv_path" not in sensor.extra_state_attributes