Move Bluetooth advertisement tracker to habluetooth library (#105083)

homeassistant/components/bluetooth/advertisement_tracker.py

@@ -1,82 +0,0 @@
-"""The bluetooth integration advertisement tracker."""
-from __future__ import annotations
-
-from typing import Any
-
-from homeassistant.core import callback
-
-from .models import BluetoothServiceInfoBleak
-
-ADVERTISING_TIMES_NEEDED = 16
-
-# Each scanner may buffer incoming packets so
-# we need to give a bit of leeway before we
-# mark a device unavailable
-TRACKER_BUFFERING_WOBBLE_SECONDS = 5
-
-
-class AdvertisementTracker:
-    """Tracker to determine the interval that a device is advertising."""
-
-    __slots__ = ("intervals", "fallback_intervals", "sources", "_timings")
-
-    def __init__(self) -> None:
-        """Initialize the tracker."""
-        self.intervals: dict[str, float] = {}
-        self.fallback_intervals: dict[str, float] = {}
-        self.sources: dict[str, str] = {}
-        self._timings: dict[str, list[float]] = {}
-
-    @callback
-    def async_diagnostics(self) -> dict[str, dict[str, Any]]:
-        """Return diagnostics."""
-        return {
-            "intervals": self.intervals,
-            "fallback_intervals": self.fallback_intervals,
-            "sources": self.sources,
-            "timings": self._timings,
-        }
-
-    @callback
-    def async_collect(self, service_info: BluetoothServiceInfoBleak) -> None:
-        """Collect timings for the tracker.
-
-        For performance reasons, it is the responsibility of the
-        caller to check if the device already has an interval set or
-        the source has changed before calling this function.
-        """
-        address = service_info.address
-        self.sources[address] = service_info.source
-        timings = self._timings.setdefault(address, [])
-        timings.append(service_info.time)
-        if len(timings) != ADVERTISING_TIMES_NEEDED:
-            return
-
-        max_time_between_advertisements = timings[1] - timings[0]
-        for i in range(2, len(timings)):
-            time_between_advertisements = timings[i] - timings[i - 1]
-            if time_between_advertisements > max_time_between_advertisements:
-                max_time_between_advertisements = time_between_advertisements
-
-        # We now know the maximum time between advertisements
-        self.intervals[address] = max_time_between_advertisements
-        del self._timings[address]
-
-    @callback
-    def async_remove_address(self, address: str) -> None:
-        """Remove the tracker."""
-        self.intervals.pop(address, None)
-        self.sources.pop(address, None)
-        self._timings.pop(address, None)
-
-    @callback
-    def async_remove_fallback_interval(self, address: str) -> None:
-        """Remove fallback interval."""
-        self.fallback_intervals.pop(address, None)
-
-    @callback
-    def async_remove_source(self, source: str) -> None:
-        """Remove the tracker."""
-        for address, tracked_source in list(self.sources.items()):
-            if tracked_source == source:
-                self.async_remove_address(address)

homeassistant/components/bluetooth/manager.py

@@ -17,6 +17,7 @@ from bluetooth_adapters import (
     BluetoothAdapters,
 )
 from bluetooth_data_tools import monotonic_time_coarse
+from habluetooth import TRACKER_BUFFERING_WOBBLE_SECONDS, AdvertisementTracker
 
 from homeassistant import config_entries
 from homeassistant.const import EVENT_LOGGING_CHANGED
@@ -29,10 +30,6 @@ from homeassistant.core import (
 from homeassistant.helpers import discovery_flow
 from homeassistant.helpers.event import async_track_time_interval
 
-from .advertisement_tracker import (
-    TRACKER_BUFFERING_WOBBLE_SECONDS,
-    AdvertisementTracker,
-)
 from .base_scanner import BaseHaScanner, BluetoothScannerDevice
 from .const import (
     FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,

tests/components/bluetooth/test_advertisement_tracker.py

@@ -3,6 +3,7 @@ from datetime import timedelta
 import time
 from unittest.mock import patch
 
+from habluetooth.advertisement_tracker import ADVERTISING_TIMES_NEEDED
 import pytest
 
 from homeassistant.components.bluetooth import (
@@ -10,9 +11,6 @@ from homeassistant.components.bluetooth import (
     async_register_scanner,
     async_track_unavailable,
 )
-from homeassistant.components.bluetooth.advertisement_tracker import (
-    ADVERTISING_TIMES_NEEDED,
-)
 from homeassistant.components.bluetooth.const import (
     FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
     SOURCE_LOCAL,

tests/components/bluetooth/test_base_scanner.py

@@ -8,6 +8,7 @@ from unittest.mock import patch
 
 from bleak.backends.device import BLEDevice
 from bleak.backends.scanner import AdvertisementData
+from habluetooth.advertisement_tracker import TRACKER_BUFFERING_WOBBLE_SECONDS
 import pytest
 
 from homeassistant.components import bluetooth
@@ -17,9 +18,6 @@ from homeassistant.components.bluetooth import (
     HomeAssistantRemoteScanner,
     storage,
 )
-from homeassistant.components.bluetooth.advertisement_tracker import (
-    TRACKER_BUFFERING_WOBBLE_SECONDS,
-)
 from homeassistant.components.bluetooth.const import (
     CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
     FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,

tests/components/private_ble_device/test_device_tracker.py

@@ -3,9 +3,8 @@
 
 import time
 
-from homeassistant.components.bluetooth.advertisement_tracker import (
-    ADVERTISING_TIMES_NEEDED,
-)
+from habluetooth.advertisement_tracker import ADVERTISING_TIMES_NEEDED
+
 from homeassistant.components.bluetooth.api import (
     async_get_fallback_availability_interval,
 )

tests/components/private_ble_device/test_sensor.py

@@ -1,10 +1,9 @@
 """Tests for sensors."""
 
 
+from habluetooth.advertisement_tracker import ADVERTISING_TIMES_NEEDED
+
 from homeassistant.components.bluetooth import async_set_fallback_availability_interval
-from homeassistant.components.bluetooth.advertisement_tracker import (
-    ADVERTISING_TIMES_NEEDED,
-)
 from homeassistant.core import HomeAssistant
 
 from . import (