Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Automatically determine the advertising interval for bluetooth devices (#79669)

pull/80358/head
J. Nick Koston 2022-10-14 08:39:18 -10:00 committed by GitHub
parent a68bd8df6f
commit 0c76e3a97e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 625 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/bluetooth/__init__.py
View File

@ -39,6 +39,7 @@ from .const import (
DATA_MANAGER,
DEFAULT_ADDRESS,
DOMAIN,
FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
SOURCE_LOCAL,
AdapterDetails,
)
@ -81,6 +82,7 @@ __all__ = [
"BluetoothCallback",
"HaBluetoothConnector",
"SOURCE_LOCAL",
"FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS",
]
_LOGGER = logging.getLogger(__name__)

68
homeassistant/components/bluetooth/advertisement_tracker.py Normal file
View File

@ -0,0 +1,68 @@
"""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
class AdvertisementTracker:
"""Tracker to determine the interval that a device is advertising."""
def __init__(self) -> None:
"""Initialize the tracker."""
self.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,
"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_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)

16
homeassistant/components/bluetooth/const.py
View File

@ -31,11 +31,17 @@ UNAVAILABLE_TRACK_SECONDS: Final = 60 * 5
START_TIMEOUT = 15
MAX_DBUS_SETUP_SECONDS = 5
# Anything after 30s is considered stale, we have buffer
# for start timeouts and execution time
STALE_ADVERTISEMENT_SECONDS: Final = 30 + START_TIMEOUT + MAX_DBUS_SETUP_SECONDS
# The maximum time between advertisements for a device to be considered
# stale when the advertisement tracker cannot determine the interval.
#
# We have to set this quite high as we don't know
# when devices fall out of the ESPHome device (and other non-local scanners)'s
# stack like we do with BlueZ so its safer to assume its available
# since if it does go out of range and it is in range
# of another device the timeout is much shorter and it will
# switch over to using that adapter anyways.
#
FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS: Final = 60 * 15
# We must recover before we hit the 180s mark

173
homeassistant/components/bluetooth/manager.py
View File

@ -6,6 +6,7 @@ from collections.abc import Callable, Iterable
from datetime import datetime, timedelta
import itertools
import logging
import time
from typing import TYPE_CHECKING, Any, Final
from bleak.backends.scanner import AdvertisementDataCallback
@ -20,11 +21,12 @@ from homeassistant.core import (
from homeassistant.helpers import discovery_flow
from homeassistant.helpers.event import async_track_time_interval
from .advertisement_tracker import AdvertisementTracker
from .const import (
ADAPTER_ADDRESS,
ADAPTER_PASSIVE_SCAN,
FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
NO_RSSI_VALUE,
STALE_ADVERTISEMENT_SECONDS,
UNAVAILABLE_TRACK_SECONDS,
AdapterDetails,
)
@ -66,49 +68,11 @@ APPLE_START_BYTES_WANTED: Final = {
RSSI_SWITCH_THRESHOLD = 6
MONOTONIC_TIME: Final = time.monotonic
_LOGGER = logging.getLogger(__name__)
def _prefer_previous_adv(
old: BluetoothServiceInfoBleak, new: BluetoothServiceInfoBleak
) -> bool:
"""Prefer previous advertisement if it is better."""
if new.time - old.time > STALE_ADVERTISEMENT_SECONDS:
# If the old advertisement is stale, any new advertisement is preferred
if new.source != old.source:
_LOGGER.debug(
"%s (%s): Switching from %s[%s] to %s[%s] (time elapsed:%s > stale seconds:%s)",
new.advertisement.local_name,
new.device.address,
old.source,
old.connectable,
new.source,
new.connectable,
new.time - old.time,
STALE_ADVERTISEMENT_SECONDS,
)
return False
if new.device.rssi - RSSI_SWITCH_THRESHOLD > (old.device.rssi or NO_RSSI_VALUE):
# If new advertisement is RSSI_SWITCH_THRESHOLD more, the new one is preferred
if new.source != old.source:
_LOGGER.debug(
"%s (%s): Switching from %s[%s] to %s[%s] (new rssi:%s - threshold:%s > old rssi:%s)",
new.advertisement.local_name,
new.device.address,
old.source,
old.connectable,
new.source,
new.connectable,
new.device.rssi,
RSSI_SWITCH_THRESHOLD,
old.device.rssi,
)
return False
# If the source is the different, the old one is preferred because its
# not stale and its RSSI_SWITCH_THRESHOLD less than the new one
return old.source != new.source
def _dispatch_bleak_callback(
callback: AdvertisementDataCallback | None,
filters: dict[str, set[str]],
@ -142,13 +106,17 @@ class BluetoothManager:
"""Init bluetooth manager."""
self.hass = hass
self._integration_matcher = integration_matcher
self._cancel_unavailable_tracking: list[CALLBACK_TYPE] = []
self._cancel_unavailable_tracking: CALLBACK_TYPE | None = None
self._advertisement_tracker = AdvertisementTracker()
self._unavailable_callbacks: dict[
str, list[Callable[[BluetoothServiceInfoBleak], None]]
] = {}
self._connectable_unavailable_callbacks: dict[
str, list[Callable[[BluetoothServiceInfoBleak], None]]
] = {}
self._callback_index = BluetoothCallbackMatcherIndex()
self._bleak_callbacks: list[
tuple[AdvertisementDataCallback, dict[str, set[str]]]
@ -190,6 +158,7 @@ class BluetoothManager:
"history": [
service_info.as_dict() for service_info in self._history.values()
],
"advertisement_tracker": self._advertisement_tracker.async_diagnostics(),
}
def _find_adapter_by_address(self, address: str) -> str | None:
@ -229,9 +198,8 @@ class BluetoothManager:
"""Stop the Bluetooth integration at shutdown."""
_LOGGER.debug("Stopping bluetooth manager")
if self._cancel_unavailable_tracking:
for cancel in self._cancel_unavailable_tracking:
cancel()
self._cancel_unavailable_tracking.clear()
self._cancel_unavailable_tracking()
self._cancel_unavailable_tracking = None
uninstall_multiple_bleak_catcher()
async def async_get_devices_by_address(
@ -274,18 +242,24 @@ class BluetoothManager:
@hass_callback
def async_setup_unavailable_tracking(self) -> None:
"""Set up the unavailable tracking."""
self._async_setup_unavailable_tracking(True)
self._async_setup_unavailable_tracking(False)
self._cancel_unavailable_tracking = async_track_time_interval(
self.hass,
self._async_check_unavailable,
timedelta(seconds=UNAVAILABLE_TRACK_SECONDS),
)
@hass_callback
def _async_setup_unavailable_tracking(self, connectable: bool) -> None:
"""Set up the unavailable tracking."""
unavailable_callbacks = self._get_unavailable_callbacks_by_type(connectable)
history = self._get_history_by_type(connectable)
def _async_check_unavailable(self, now: datetime) -> None:
"""Watch for unavailable devices and cleanup state history."""
monotonic_now = MONOTONIC_TIME()
connectable_history = self._connectable_history
all_history = self._history
removed_addresses: set[str] = set()
@hass_callback
def _async_check_unavailable(now: datetime) -> None:
"""Watch for unavailable devices."""
for connectable in (True, False):
unavailable_callbacks = self._get_unavailable_callbacks_by_type(connectable)
intervals = self._advertisement_tracker.intervals
history = connectable_history if connectable else all_history
history_set = set(history)
active_addresses = {
device.address
@ -293,35 +267,79 @@ class BluetoothManager:
}
disappeared = history_set.difference(active_addresses)
for address in disappeared:
#
# For non-connectable devices we also check the device has exceeded
# the advertising interval before we mark it as unavailable
# since it may have gone to sleep and since we do not need an active connection
# to it we can only determine its availability by the lack of advertisements
#
if not connectable and (advertising_interval := intervals.get(address)):
time_since_seen = monotonic_now - history[address].time
if time_since_seen <= advertising_interval:
continue
service_info = history.pop(address)
removed_addresses.add(address)
if not (callbacks := unavailable_callbacks.get(address)):
continue
for callback in callbacks:
try:
callback(service_info)
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Error in unavailable callback")
self._cancel_unavailable_tracking.append(
async_track_time_interval(
self.hass,
_async_check_unavailable,
timedelta(seconds=UNAVAILABLE_TRACK_SECONDS),
# If we removed the device from both the connectable history
# and all history then we can remove it from the advertisement tracker
for address in removed_addresses:
if address not in connectable_history and address not in all_history:
self._advertisement_tracker.async_remove_address(address)
def _prefer_previous_adv_from_different_source(
self, old: BluetoothServiceInfoBleak, new: BluetoothServiceInfoBleak
) -> bool:
"""Prefer previous advertisement from a different source if it is better."""
if new.time - old.time > (
stale_seconds := self._advertisement_tracker.intervals.get(
new.address, FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
)
)
):
# If the old advertisement is stale, any new advertisement is preferred
_LOGGER.debug(
"%s (%s): Switching from %s[%s] to %s[%s] (time elapsed:%s > stale seconds:%s)",
new.advertisement.local_name,
new.device.address,
old.source,
old.connectable,
new.source,
new.connectable,
new.time - old.time,
stale_seconds,
)
return False
if new.device.rssi - RSSI_SWITCH_THRESHOLD > (old.device.rssi or NO_RSSI_VALUE):
# If new advertisement is RSSI_SWITCH_THRESHOLD more, the new one is preferred
_LOGGER.debug(
"%s (%s): Switching from %s[%s] to %s[%s] (new rssi:%s - threshold:%s > old rssi:%s)",
new.advertisement.local_name,
new.device.address,
old.source,
old.connectable,
new.source,
new.connectable,
new.device.rssi,
RSSI_SWITCH_THRESHOLD,
old.device.rssi,
)
return False
return True
@hass_callback
def scanner_adv_received(self, service_info: BluetoothServiceInfoBleak) -> None:
"""Handle a new advertisement from any scanner.
Callbacks from all the scanners arrive here.
In the future we will only process callbacks if
- The device is not in the history
- The RSSI is above a certain threshold better than
than the source from the history or the timestamp
in the history is older than 180s
"""
# Pre-filter noisy apple devices as they can account for 20-35% of the
@ -340,8 +358,14 @@ class BluetoothManager:
connectable = service_info.connectable
address = device.address
all_history = self._connectable_history if connectable else self._history
old_service_info = all_history.get(address)
if old_service_info and _prefer_previous_adv(old_service_info, service_info):
source = service_info.source
if (
(old_service_info := all_history.get(address))
and source != old_service_info.source
and self._prefer_previous_adv_from_different_source(
old_service_info, service_info
)
):
return
self._history[address] = service_info
@ -350,6 +374,15 @@ class BluetoothManager:
self._connectable_history[address] = service_info
# Bleak callbacks must get a connectable device
# Track advertisement intervals to determine when we need to
# switch adapters or mark a device as unavailable
tracker = self._advertisement_tracker
if (last_source := tracker.sources.get(address)) and last_source != source:
# Source changed, remove the old address from the tracker
tracker.async_remove_address(address)
if address not in tracker.intervals:
tracker.async_collect(service_info)
# If the advertisement data is the same as the last time we saw it, we
# don't need to do anything else.
if old_service_info and not (
@ -360,7 +393,6 @@ class BluetoothManager:
):
return
source = service_info.source
if connectable:
# Bleak callbacks must get a connectable device
for callback_filters in self._bleak_callbacks:
@ -515,6 +547,7 @@ class BluetoothManager:
scanners = self._get_scanners_by_type(connectable)
def _unregister_scanner() -> None:
self._advertisement_tracker.async_remove_source(scanner.source)
scanners.remove(scanner)
scanners.append(scanner)

7
homeassistant/components/bluetooth/models.py
View File

@ -20,7 +20,7 @@ from bleak.backends.scanner import (
)
from bleak_retry_connector import freshen_ble_device
from homeassistant.core import CALLBACK_TYPE, callback as hass_callback
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback
from homeassistant.helpers.frame import report
from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo
@ -105,6 +105,11 @@ class _HaWrappedBleakBackend:
class BaseHaScanner:
"""Base class for Ha Scanners."""
def __init__(self, hass: HomeAssistant, source: str) -> None:
"""Initialize the scanner."""
self.hass = hass
self.source = source
@property
@abstractmethod
def discovered_devices(self) -> list[BLEDevice]:

6
homeassistant/components/bluetooth/scanner.py
View File

@ -50,8 +50,6 @@ PASSIVE_SCANNER_ARGS = BlueZScannerArgs(
_LOGGER = logging.getLogger(__name__)
MONOTONIC_TIME = time.monotonic
# If the adapter is in a stuck state the following errors are raised:
NEED_RESET_ERRORS = [
"org.bluez.Error.Failed",
@ -130,7 +128,8 @@ class HaScanner(BaseHaScanner):
address: str,
) -> None:
"""Init bluetooth discovery."""
self.hass = hass
source = address if address != DEFAULT_ADDRESS else adapter or SOURCE_LOCAL
super().__init__(hass, source)
self.mode = mode
self.adapter = adapter
self._start_stop_lock = asyncio.Lock()
@ -139,7 +138,6 @@ class HaScanner(BaseHaScanner):
self._start_time = 0.0
self._callbacks: list[Callable[[BluetoothServiceInfoBleak], None]] = []
self.name = adapter_human_name(adapter, address)
self.source = address if address != DEFAULT_ADDRESS else adapter or SOURCE_LOCAL
@property
def discovered_devices(self) -> list[BLEDevice]:

25
homeassistant/components/esphome/bluetooth/scanner.py
View File

@ -11,19 +11,15 @@ from aioesphomeapi import BluetoothLEAdvertisement
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BaseHaScanner, HaBluetoothConnector
from homeassistant.components.bluetooth.models import BluetoothServiceInfoBleak
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
# We have to set this quite high as we don't know
# when devices fall out of the esphome device's stack
# like we do with BlueZ so its safer to assume its available
# since if it does go out of range and it is in range
# of another device the timeout is much shorter and it will
# switch over to using that adapter anyways.
ADV_STALE_TIME = 60 * 15 # seconds
TWO_CHAR = re.compile("..")
@ -39,11 +35,10 @@ class ESPHomeScanner(BaseHaScanner):
connectable: bool,
) -> None:
"""Initialize the scanner."""
self._hass = hass
super().__init__(hass, scanner_id)
self._new_info_callback = new_info_callback
self._discovered_devices: dict[str, BLEDevice] = {}
self._discovered_device_timestamps: dict[str, float] = {}
self._source = scanner_id
self._connector = connector
self._connectable = connectable
self._details: dict[str, str | HaBluetoothConnector] = {"source": scanner_id}
@ -54,7 +49,7 @@ class ESPHomeScanner(BaseHaScanner):
def async_setup(self) -> CALLBACK_TYPE:
"""Set up the scanner."""
return async_track_time_interval(
self._hass, self._async_expire_devices, timedelta(seconds=30)
self.hass, self._async_expire_devices, timedelta(seconds=30)
)
def _async_expire_devices(self, _datetime: datetime.datetime) -> None:
@ -63,7 +58,7 @@ class ESPHomeScanner(BaseHaScanner):
expired = [
address
for address, timestamp in self._discovered_device_timestamps.items()
if now - timestamp > ADV_STALE_TIME
if now - timestamp > FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
]
for address in expired:
del self._discovered_devices[address]
@ -113,7 +108,7 @@ class ESPHomeScanner(BaseHaScanner):
manufacturer_data=advertisement_data.manufacturer_data,
service_data=advertisement_data.service_data,
service_uuids=advertisement_data.service_uuids,
source=self._source,
source=self.source,
device=device,
advertisement=advertisement_data,
connectable=self._connectable,

405
tests/components/bluetooth/test_advertisement_tracker.py Normal file
View File

@ -0,0 +1,405 @@
"""Tests for the Bluetooth integration advertisement tracking."""
from datetime import timedelta
import time
from unittest.mock import patch
from bleak.backends.scanner import AdvertisementData, BLEDevice
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 (
SOURCE_LOCAL,
UNAVAILABLE_TRACK_SECONDS,
)
from homeassistant.components.bluetooth.models import BaseHaScanner
from homeassistant.core import callback
from homeassistant.util import dt as dt_util
from . import inject_advertisement_with_time_and_source
from tests.common import async_fire_time_changed
ONE_HOUR_SECONDS = 3600
async def test_advertisment_interval_shorter_than_adapter_stack_timeout(
hass, caplog, enable_bluetooth, macos_adapter
):
"""Test we can determine the advertisement interval."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
@callback
def _switchbot_device_unavailable_callback(_address: str) -> None:
"""Switchbot device unavailable callback."""
nonlocal switchbot_device_went_unavailable
switchbot_device_went_unavailable = True
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass,
switchbot_device,
switchbot_adv,
start_monotonic_time + (i * 2),
SOURCE_LOCAL,
)
switchbot_device_unavailable_cancel = async_track_unavailable(
hass, _switchbot_device_unavailable_callback, switchbot_device.address
)
monotonic_now = start_monotonic_time + ((ADVERTISING_TIMES_NEEDED - 1) * 2)
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
):
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
)
await hass.async_block_till_done()
assert switchbot_device_went_unavailable is True
switchbot_device_unavailable_cancel()
async def test_advertisment_interval_longer_than_adapter_stack_timeout_connectable(
hass, caplog, enable_bluetooth, macos_adapter
):
"""Test device with a long advertisement interval."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
@callback
def _switchbot_device_unavailable_callback(_address: str) -> None:
"""Switchbot device unavailable callback."""
nonlocal switchbot_device_went_unavailable
switchbot_device_went_unavailable = True
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass,
switchbot_device,
switchbot_adv,
start_monotonic_time + (i * ONE_HOUR_SECONDS),
SOURCE_LOCAL,
)
switchbot_device_unavailable_cancel = async_track_unavailable(
hass, _switchbot_device_unavailable_callback, switchbot_device.address
)
monotonic_now = start_monotonic_time + (
(ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS
)
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
):
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
)
await hass.async_block_till_done()
assert switchbot_device_went_unavailable is True
switchbot_device_unavailable_cancel()
async def test_advertisment_interval_longer_than_adapter_stack_timeout_adapter_change_connectable(
hass, caplog, enable_bluetooth, macos_adapter
):
"""Test device with a long advertisement interval with an adapter change."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
@callback
def _switchbot_device_unavailable_callback(_address: str) -> None:
"""Switchbot device unavailable callback."""
nonlocal switchbot_device_went_unavailable
switchbot_device_went_unavailable = True
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass,
switchbot_device,
switchbot_adv,
start_monotonic_time + (i * 2),
"original",
)
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass,
switchbot_device,
switchbot_adv,
start_monotonic_time + (i * ONE_HOUR_SECONDS),
"new",
)
switchbot_device_unavailable_cancel = async_track_unavailable(
hass, _switchbot_device_unavailable_callback, switchbot_device.address
)
monotonic_now = start_monotonic_time + (
(ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS
)
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
):
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
)
await hass.async_block_till_done()
assert switchbot_device_went_unavailable is True
switchbot_device_unavailable_cancel()
async def test_advertisment_interval_longer_than_adapter_stack_timeout_not_connectable(
hass, caplog, enable_bluetooth, macos_adapter
):
"""Test device with a long advertisement interval that is not connectable not reaching the advertising interval."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
@callback
def _switchbot_device_unavailable_callback(_address: str) -> None:
"""Switchbot device unavailable callback."""
nonlocal switchbot_device_went_unavailable
switchbot_device_went_unavailable = True
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass,
switchbot_device,
switchbot_adv,
start_monotonic_time + (i * ONE_HOUR_SECONDS),
SOURCE_LOCAL,
)
switchbot_device_unavailable_cancel = async_track_unavailable(
hass,
_switchbot_device_unavailable_callback,
switchbot_device.address,
connectable=False,
)
monotonic_now = start_monotonic_time + (
(ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS
)
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
):
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
)
await hass.async_block_till_done()
assert switchbot_device_went_unavailable is False
switchbot_device_unavailable_cancel()
async def test_advertisment_interval_shorter_than_adapter_stack_timeout_adapter_change_not_connectable(
hass, caplog, enable_bluetooth, macos_adapter
):
"""Test device with a short advertisement interval with an adapter change that is not connectable."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
@callback
def _switchbot_device_unavailable_callback(_address: str) -> None:
"""Switchbot device unavailable callback."""
nonlocal switchbot_device_went_unavailable
switchbot_device_went_unavailable = True
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass,
switchbot_device,
switchbot_adv,
start_monotonic_time + (i * ONE_HOUR_SECONDS),
"original",
)
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass, switchbot_device, switchbot_adv, start_monotonic_time + (i * 2), "new"
)
switchbot_device_unavailable_cancel = async_track_unavailable(
hass,
_switchbot_device_unavailable_callback,
switchbot_device.address,
connectable=False,
)
monotonic_now = start_monotonic_time + (
(ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS
)
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
):
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
)
await hass.async_block_till_done()
assert switchbot_device_went_unavailable is True
switchbot_device_unavailable_cancel()
async def test_advertisment_interval_longer_than_adapter_stack_timeout_adapter_change_not_connectable(
hass, caplog, enable_bluetooth, macos_adapter
):
"""Test device with a long advertisement interval with an adapter change that is not connectable."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
class FakeScanner(BaseHaScanner):
"""Fake scanner."""
@property
def discovered_devices(self) -> list[BLEDevice]:
return []
scanner = FakeScanner(hass, "new")
cancel_scanner = async_register_scanner(hass, scanner, False)
@callback
def _switchbot_device_unavailable_callback(_address: str) -> None:
"""Switchbot device unavailable callback."""
nonlocal switchbot_device_went_unavailable
switchbot_device_went_unavailable = True
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass,
switchbot_device,
switchbot_adv,
start_monotonic_time + (i * 2),
"original",
)
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass,
switchbot_device,
switchbot_adv,
start_monotonic_time + (i * ONE_HOUR_SECONDS),
"new",
)
switchbot_device_unavailable_cancel = async_track_unavailable(
hass,
_switchbot_device_unavailable_callback,
switchbot_device.address,
connectable=False,
)
monotonic_now = start_monotonic_time + (
(ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS
)
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
):
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
)
await hass.async_block_till_done()
assert switchbot_device_went_unavailable is False
cancel_scanner()
# Now that the scanner is gone we should go back to the stack default timeout
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
):
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
)
await hass.async_block_till_done()
assert switchbot_device_went_unavailable is True
switchbot_device_unavailable_cancel()
async def test_advertisment_interval_longer_increasing_than_adapter_stack_timeout_adapter_change_not_connectable(
hass, caplog, enable_bluetooth, macos_adapter
):
"""Test device with a increasing advertisement interval with an adapter change that is not connectable."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
@callback
def _switchbot_device_unavailable_callback(_address: str) -> None:
"""Switchbot device unavailable callback."""
nonlocal switchbot_device_went_unavailable
switchbot_device_went_unavailable = True
for i in range(ADVERTISING_TIMES_NEEDED, 2 * ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass,
switchbot_device,
switchbot_adv,
start_monotonic_time + (i**2),
"new",
)
switchbot_device_unavailable_cancel = async_track_unavailable(
hass,
_switchbot_device_unavailable_callback,
switchbot_device.address,
connectable=False,
)
monotonic_now = start_monotonic_time + UNAVAILABLE_TRACK_SECONDS + 1
with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
):
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
)
await hass.async_block_till_done()
assert switchbot_device_went_unavailable is False
switchbot_device_unavailable_cancel()

10
tests/components/bluetooth/test_diagnostics.py
View File

@ -96,6 +96,11 @@ async def test_diagnostics(
}
},
"manager": {
"advertisement_tracker": {
"intervals": {},
"sources": {},
"timings": {},
},
"adapters": {
"hci0": {
"address": "00:00:00:00:00:01",
@ -198,6 +203,11 @@ async def test_diagnostics_macos(
}
},
"manager": {
"advertisement_tracker": {
"intervals": {},
"sources": {"44:44:33:11:23:45": "local"},
"timings": {"44:44:33:11:23:45": [ANY]},
},
"adapters": {
"Core Bluetooth": {
"address": "00:00:00:00:00:00",

4
tests/components/bluetooth/test_init.py
View File

@ -2595,7 +2595,7 @@ async def test_getting_the_scanner_returns_the_wrapped_instance(hass, enable_blu
async def test_scanner_count_connectable(hass, enable_bluetooth):
"""Test getting the connectable scanner count."""
scanner = models.BaseHaScanner()
scanner = models.BaseHaScanner(hass, "any")
cancel = bluetooth.async_register_scanner(hass, scanner, False)
assert bluetooth.async_scanner_count(hass, connectable=True) == 1
cancel()
@ -2603,7 +2603,7 @@ async def test_scanner_count_connectable(hass, enable_bluetooth):
async def test_scanner_count(hass, enable_bluetooth):
"""Test getting the connectable and non-connectable scanner count."""
scanner = models.BaseHaScanner()
scanner = models.BaseHaScanner(hass, "any")
cancel = bluetooth.async_register_scanner(hass, scanner, False)
assert bluetooth.async_scanner_count(hass, connectable=False) == 2
cancel()

6
tests/components/bluetooth/test_manager.py
View File

@ -6,7 +6,9 @@ from bleak.backends.scanner import AdvertisementData, BLEDevice
from bluetooth_adapters import AdvertisementHistory
from homeassistant.components import bluetooth
from homeassistant.components.bluetooth.manager import STALE_ADVERTISEMENT_SECONDS
from homeassistant.components.bluetooth.manager import (
FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
)
from homeassistant.setup import async_setup_component
from . import (
@ -227,7 +229,7 @@ async def test_switching_adapters_based_on_stale(hass, enable_bluetooth):
hass,
switchbot_device_poor_signal_hci1,
switchbot_adv_poor_signal_hci1,
start_time_monotonic + STALE_ADVERTISEMENT_SECONDS + 1,
start_time_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1,
"hci1",
)

4
tests/components/bluetooth/test_models.py
View File

@ -204,7 +204,7 @@ async def test_ble_device_with_proxy_client_out_of_connections_uses_best_availab
return switchbot_proxy_device_has_connection_slot
return None
scanner = FakeScanner()
scanner = FakeScanner(hass, "esp32")
cancel = manager.async_register_scanner(scanner, True)
assert manager.async_discovered_devices(True) == [
switchbot_proxy_device_no_connection_slot
@ -290,7 +290,7 @@ async def test_ble_device_with_proxy_client_out_of_connections_uses_best_availab
return switchbot_proxy_device_has_connection_slot
return None
scanner = FakeScanner()
scanner = FakeScanner(hass, "esp32")
cancel = manager.async_register_scanner(scanner, True)
assert manager.async_discovered_devices(True) == [
switchbot_proxy_device_no_connection_slot