Add bluetooth API to allow rediscovery of address (#76005)

* Add API to allow rediscovery of domains

* Switch to clearing per device

* Drop unneded change
This commit is contained in:
Joakim Plate 2022-08-05 14:49:34 +02:00 committed by GitHub
parent 741efb89d5
commit cdde4f9925
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 69 additions and 3 deletions

View File

@ -214,6 +214,13 @@ def async_track_unavailable(
return manager.async_track_unavailable(callback, address)
@hass_callback
def async_rediscover_address(hass: HomeAssistant, address: str) -> None:
"""Trigger discovery of devices which have already been seen."""
manager: BluetoothManager = hass.data[DOMAIN]
manager.async_rediscover_address(address)
async def _async_has_bluetooth_adapter() -> bool:
"""Return if the device has a bluetooth adapter."""
return bool(await async_get_bluetooth_adapters())
@ -545,3 +552,8 @@ class BluetoothManager:
# change the bluetooth dongle.
_LOGGER.error("Error stopping scanner: %s", ex)
uninstall_multiple_bleak_catcher()
@hass_callback
def async_rediscover_address(self, address: str) -> None:
"""Trigger discovery of devices which have already been seen."""
self._integration_matcher.async_clear_address(address)

View File

@ -10,7 +10,7 @@ from lru import LRU # pylint: disable=no-name-in-module
from homeassistant.loader import BluetoothMatcher, BluetoothMatcherOptional
if TYPE_CHECKING:
from collections.abc import Mapping
from collections.abc import MutableMapping
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
@ -70,7 +70,7 @@ class IntegrationMatcher:
self._integration_matchers = integration_matchers
# Some devices use a random address so we need to use
# an LRU to avoid memory issues.
self._matched: Mapping[str, IntegrationMatchHistory] = LRU(
self._matched: MutableMapping[str, IntegrationMatchHistory] = LRU(
MAX_REMEMBER_ADDRESSES
)
@ -78,6 +78,10 @@ class IntegrationMatcher:
"""Clear the history."""
self._matched = {}
def async_clear_address(self, address: str) -> None:
"""Clear the history matches for a set of domains."""
self._matched.pop(address, None)
def match_domains(self, device: BLEDevice, adv_data: AdvertisementData) -> set[str]:
"""Return the domains that are matched."""
matched_domains: set[str] = set()
@ -98,7 +102,7 @@ class IntegrationMatcher:
previous_match.service_data |= bool(adv_data.service_data)
previous_match.service_uuids |= bool(adv_data.service_uuids)
else:
self._matched[device.address] = IntegrationMatchHistory( # type: ignore[index]
self._matched[device.address] = IntegrationMatchHistory(
manufacturer_data=bool(adv_data.manufacturer_data),
service_data=bool(adv_data.service_data),
service_uuids=bool(adv_data.service_uuids),

View File

@ -14,11 +14,13 @@ from homeassistant.components.bluetooth import (
BluetoothScanningMode,
BluetoothServiceInfoBleak,
async_address_present,
async_rediscover_address,
async_register_callback,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
@ -113,6 +115,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
device = Device(service_info.device)
device_info = DeviceInfo(
connections={(dr.CONNECTION_BLUETOOTH, service_info.address)},
identifiers={(DOMAIN, service_info.address)},
manufacturer="Fjäråskupan",
name="Fjäråskupan",
@ -175,4 +178,11 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
for device_entry in dr.async_entries_for_config_entry(
dr.async_get(hass), entry.entry_id
):
for conn in device_entry.connections:
if conn[0] == dr.CONNECTION_BLUETOOTH:
async_rediscover_address(hass, conn[1])
return unload_ok

View File

@ -16,6 +16,7 @@ from homeassistant.components.bluetooth import (
BluetoothScanningMode,
BluetoothServiceInfo,
async_process_advertisements,
async_rediscover_address,
async_track_unavailable,
models,
)
@ -560,6 +561,45 @@ async def test_discovery_match_first_by_service_uuid_and_then_manufacturer_id(
assert len(mock_config_flow.mock_calls) == 0
async def test_rediscovery(hass, mock_bleak_scanner_start, enable_bluetooth):
"""Test bluetooth discovery can be re-enabled for a given domain."""
mock_bt = [
{"domain": "switchbot", "service_uuid": "cba20d00-224d-11e6-9fb8-0002a5d5c51b"}
]
with patch(
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
), patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
assert await async_setup_component(
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
)
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
assert len(mock_bleak_scanner_start.mock_calls) == 1
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
_get_underlying_scanner()._callback(switchbot_device, switchbot_adv)
await hass.async_block_till_done()
_get_underlying_scanner()._callback(switchbot_device, switchbot_adv)
await hass.async_block_till_done()
assert len(mock_config_flow.mock_calls) == 1
assert mock_config_flow.mock_calls[0][1][0] == "switchbot"
async_rediscover_address(hass, "44:44:33:11:23:45")
_get_underlying_scanner()._callback(switchbot_device, switchbot_adv)
await hass.async_block_till_done()
assert len(mock_config_flow.mock_calls) == 2
assert mock_config_flow.mock_calls[1][1][0] == "switchbot"
async def test_async_discovered_device_api(hass, mock_bleak_scanner_start):
"""Test the async_discovered_device API."""
mock_bt = []