Move bluetooth remote scanner implementation into a base class

homeassistant/components/bluetooth/__init__.py

@@ -50,6 +50,7 @@
 from .manager import BluetoothManager
 from .match import BluetoothCallbackMatcher, IntegrationMatcher
 from .models import (
+    BaseHaRemoteScanner,
     BaseHaScanner,
     BluetoothCallback,
     BluetoothChange,
@@ -80,6 +81,7 @@
     "async_track_unavailable",
     "async_scanner_count",
     "BaseHaScanner",
+    "BaseHaRemoteScanner",
     "BluetoothServiceInfo",
     "BluetoothServiceInfoBleak",
     "BluetoothScanningMode",

homeassistant/components/bluetooth/models.py

@@ -6,6 +6,8 @@
 from collections.abc import Callable
 import contextlib
 from dataclasses import dataclass
+import datetime
+from datetime import timedelta
 from enum import Enum
 import logging
 from typing import TYPE_CHECKING, Any, Final
@@ -21,8 +23,21 @@
 from bleak_retry_connector import NO_RSSI_VALUE, freshen_ble_device
 
 from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback
+from homeassistant.helpers.event import async_track_time_interval
 from homeassistant.helpers.frame import report
 from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo
+from homeassistant.util.dt import monotonic_time_coarse
+
+from .const import FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
+
+# The maximum time between advertisements for a device to be considered
+# stale when the advertisement tracker can determine the interval for
+# connectable devices.
+#
+# BlueZ uses 180 seconds by default but we give it a bit more time
+# to account for the esp32's bluetooth stack being a bit slower
+# than BlueZ's.
+CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS: Final = 195
 
 if TYPE_CHECKING:
 
@@ -35,6 +50,8 @@
 
 MANAGER: BluetoothManager | None = None
 
+MONOTONIC_TIME: Final = monotonic_time_coarse
+
 
 @dataclass
 class BluetoothServiceInfoBleak(BluetoothServiceInfo):
@@ -134,6 +151,142 @@ async def async_diagnostics(self) -> dict[str, Any]:
         }
 
 
+class BaseHaRemoteScanner(BaseHaScanner):
+    """Base class for a Home Assistant remote BLE scanner."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        scanner_id: str,
+        new_info_callback: Callable[[BluetoothServiceInfoBleak], None],
+        connector: HaBluetoothConnector,
+        connectable: bool,
+    ) -> None:
+        """Initialize the scanner."""
+        super().__init__(hass, scanner_id)
+        self._new_info_callback = new_info_callback
+        self._discovered_device_advertisement_datas: dict[
+            str, tuple[BLEDevice, AdvertisementData]
+        ] = {}
+        self._discovered_device_timestamps: dict[str, float] = {}
+        self._connector = connector
+        self._connectable = connectable
+        self._details: dict[str, str | HaBluetoothConnector] = {"source": scanner_id}
+        self._expire_seconds = FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
+        if connectable:
+            self._details["connector"] = connector
+            self._expire_seconds = (
+                CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
+            )
+
+    @hass_callback
+    def async_setup(self) -> CALLBACK_TYPE:
+        """Set up the scanner."""
+        return async_track_time_interval(
+            self.hass, self._async_expire_devices, timedelta(seconds=30)
+        )
+
+    def _async_expire_devices(self, _datetime: datetime.datetime) -> None:
+        """Expire old devices."""
+        now = MONOTONIC_TIME()
+        expired = [
+            address
+            for address, timestamp in self._discovered_device_timestamps.items()
+            if now - timestamp > self._expire_seconds
+        ]
+        for address in expired:
+            del self._discovered_device_advertisement_datas[address]
+            del self._discovered_device_timestamps[address]
+
+    @property
+    def discovered_devices(self) -> list[BLEDevice]:
+        """Return a list of discovered devices."""
+        return [
+            device_advertisement_data[0]
+            for device_advertisement_data in self._discovered_device_advertisement_datas.values()
+        ]
+
+    @property
+    def discovered_devices_and_advertisement_data(
+        self,
+    ) -> dict[str, tuple[BLEDevice, AdvertisementData]]:
+        """Return a list of discovered devices and advertisement data."""
+        return self._discovered_device_advertisement_datas
+
+    @hass_callback
+    def _async_on_advertisement(
+        self,
+        address: str,
+        rssi: int,
+        local_name: str | None,
+        service_uuids: list[str],
+        service_data: dict[str, bytes],
+        manufacturer_data: dict[int, bytes],
+        tx_power: int | None,
+    ) -> None:
+        """Call the registered callback."""
+        now = MONOTONIC_TIME()
+        if prev_discovery := self._discovered_device_advertisement_datas.get(address):
+            # Merge the new data with the old data
+            # to function the same as BlueZ which
+            # merges the dicts on PropertiesChanged
+            prev_device = prev_discovery[0]
+            prev_advertisement = prev_discovery[1]
+            if (
+                local_name
+                and prev_device.name
+                and len(prev_device.name) > len(local_name)
+            ):
+                local_name = prev_device.name
+            if prev_advertisement.service_uuids:
+                service_uuids = list(
+                    set(service_uuids + prev_advertisement.service_uuids)
+                )
+            if prev_advertisement.service_data:
+                service_data = {**prev_advertisement.service_data, **service_data}
+            if prev_advertisement.manufacturer_data:
+                manufacturer_data = {
+                    **prev_advertisement.manufacturer_data,
+                    **manufacturer_data,
+                }
+
+        advertisement_data = AdvertisementData(
+            local_name=None if local_name == "" else local_name,
+            manufacturer_data=manufacturer_data,
+            service_data=service_data,
+            service_uuids=service_uuids,
+            rssi=rssi,
+            tx_power=NO_RSSI_VALUE if tx_power is None else tx_power,
+            platform_data=(),
+        )
+        device = BLEDevice(  # type: ignore[no-untyped-call]
+            address=address,
+            name=local_name,
+            details=self._details,
+            rssi=rssi,  # deprecated, will be removed in newer bleak
+        )
+        self._discovered_device_advertisement_datas[address] = (
+            device,
+            advertisement_data,
+        )
+        self._discovered_device_timestamps[address] = now
+        self._new_info_callback(
+            BluetoothServiceInfoBleak(
+                name=advertisement_data.local_name or device.name or device.address,
+                address=device.address,
+                rssi=rssi,
+                manufacturer_data=advertisement_data.manufacturer_data,
+                service_data=advertisement_data.service_data,
+                service_uuids=advertisement_data.service_uuids,
+                source=self.source,
+                device=device,
+                advertisement=advertisement_data,
+                connectable=self._connectable,
+                time=now,
+            )
+        )
+
+
 class HaBleakScannerWrapper(BaseBleakScanner):
     """A wrapper that uses the single instance."""
 

homeassistant/components/esphome/bluetooth/scanner.py

@@ -1,147 +1,29 @@
 """Bluetooth scanner for esphome."""
 from __future__ import annotations
 
-from collections.abc import Callable
-import datetime
-from datetime import timedelta
 import re
-import time
-from typing import Final
 
 from aioesphomeapi import BluetoothLEAdvertisement
-from bleak.backends.device import BLEDevice
-from bleak.backends.scanner import AdvertisementData
 
-from homeassistant.components.bluetooth import (
-    FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
-    BaseHaScanner,
-    BluetoothServiceInfoBleak,
-    HaBluetoothConnector,
-)
-from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
-from homeassistant.helpers.event import async_track_time_interval
-from homeassistant.util.dt import monotonic_time_coarse
+from homeassistant.components.bluetooth import BaseHaRemoteScanner
+from homeassistant.core import callback
 
 TWO_CHAR = re.compile("..")
 
-# The maximum time between advertisements for a device to be considered
-# stale when the advertisement tracker can determine the interval for
-# connectable devices.
-#
-# BlueZ uses 180 seconds by default but we give it a bit more time
-# to account for the esp32's bluetooth stack being a bit slower
-# than BlueZ's.
-CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS: Final = 195
 
-
-class ESPHomeScanner(BaseHaScanner):
+class ESPHomeScanner(BaseHaRemoteScanner):
     """Scanner for esphome."""
 
-    def __init__(
-        self,
-        hass: HomeAssistant,
-        scanner_id: str,
-        new_info_callback: Callable[[BluetoothServiceInfoBleak], None],
-        connector: HaBluetoothConnector,
-        connectable: bool,
-    ) -> None:
-        """Initialize the scanner."""
-        super().__init__(hass, scanner_id)
-        self._new_info_callback = new_info_callback
-        self._discovered_device_advertisement_datas: dict[
-            str, tuple[BLEDevice, AdvertisementData]
-        ] = {}
-        self._discovered_device_timestamps: dict[str, float] = {}
-        self._connector = connector
-        self._connectable = connectable
-        self._details: dict[str, str | HaBluetoothConnector] = {"source": scanner_id}
-        self._fallback_seconds = FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
-        if connectable:
-            self._details["connector"] = connector
-            self._fallback_seconds = (
-                CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
-            )
-
-    @callback
-    def async_setup(self) -> CALLBACK_TYPE:
-        """Set up the scanner."""
-        return async_track_time_interval(
-            self.hass, self._async_expire_devices, timedelta(seconds=30)
-        )
-
-    def _async_expire_devices(self, _datetime: datetime.datetime) -> None:
-        """Expire old devices."""
-        now = time.monotonic()
-        expired = [
-            address
-            for address, timestamp in self._discovered_device_timestamps.items()
-            if now - timestamp > self._fallback_seconds
-        ]
-        for address in expired:
-            del self._discovered_device_advertisement_datas[address]
-            del self._discovered_device_timestamps[address]
-
-    @property
-    def discovered_devices(self) -> list[BLEDevice]:
-        """Return a list of discovered devices."""
-        return [
-            device_advertisement_data[0]
-            for device_advertisement_data in self._discovered_device_advertisement_datas.values()
-        ]
-
-    @property
-    def discovered_devices_and_advertisement_data(
-        self,
-    ) -> dict[str, tuple[BLEDevice, AdvertisementData]]:
-        """Return a list of discovered devices and advertisement data."""
-        return self._discovered_device_advertisement_datas
-
     @callback
     def async_on_advertisement(self, adv: BluetoothLEAdvertisement) -> None:
         """Call the registered callback."""
-        now = monotonic_time_coarse()
         address = ":".join(TWO_CHAR.findall("%012X" % adv.address))  # must be upper
-        name = adv.name
-        if prev_discovery := self._discovered_device_advertisement_datas.get(address):
-            # If the last discovery had the full local name
-            # and this one doesn't, keep the old one as we
-            # always want the full local name over the short one
-            prev_device = prev_discovery[0]
-            if len(prev_device.name) > len(adv.name):
-                name = prev_device.name
-
-        advertisement_data = AdvertisementData(
-            local_name=None if name == "" else name,
-            manufacturer_data=adv.manufacturer_data,
-            service_data=adv.service_data,
-            service_uuids=adv.service_uuids,
-            rssi=adv.rssi,
-            tx_power=-127,
-            platform_data=(),
-        )
-        device = BLEDevice(  # type: ignore[no-untyped-call]
-            address=address,
-            name=name,
-            details=self._details,
-            rssi=adv.rssi,  # deprecated, will be removed in newer bleak
-        )
-        self._discovered_device_advertisement_datas[address] = (
-            device,
-            advertisement_data,
-        )
-        self._discovered_device_timestamps[address] = now
-        self._new_info_callback(
-            BluetoothServiceInfoBleak(
-                name=advertisement_data.local_name or device.name or device.address,
-                address=device.address,
-                rssi=adv.rssi,
-                manufacturer_data=advertisement_data.manufacturer_data,
-                service_data=advertisement_data.service_data,
-                service_uuids=advertisement_data.service_uuids,
-                source=self.source,
-                device=device,
-                advertisement=advertisement_data,
-                connectable=self._connectable,
-                time=now,
-            )
+        self._async_on_advertisement(
+            address,
+            adv.rssi,
+            adv.name,
+            adv.service_uuids,
+            adv.service_data,
+            adv.manufacturer_data,
+            None,
         )