diff --git a/custom_components/meshcore/binary_sensor.py b/custom_components/meshcore/binary_sensor.py index c84ecd24..892b9a70 100644 --- a/custom_components/meshcore/binary_sensor.py +++ b/custom_components/meshcore/binary_sensor.py @@ -728,7 +728,11 @@ def extra_state_attributes(self) -> Dict[str, Any]: # 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 diff --git a/custom_components/meshcore/coordinator.py b/custom_components/meshcore/coordinator.py index 02ae53f2..d2254b3a 100644 --- a/custom_components/meshcore/coordinator.py +++ b/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() diff --git a/docs/docs/contacts.md b/docs/docs/contacts.md index a213c471..0b1c1c53 100644 --- a/docs/docs/contacts.md +++ b/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 diff --git a/tests_integration/test_advert_path.py b/tests_integration/test_advert_path.py new file mode 100644 index 00000000..d03cd8f6 --- /dev/null +++ b/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