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

Break out bluetooth apis into api.py (#82416) 

* Break out bluetooth apis into api.py

Like #82291 this is not a functional change.

* cleanups
pull/82427/head
J. Nick Koston 2022-11-20 11:33:03 -06:00 committed by GitHub
parent 6ec8c63b5c
commit 3f5649092e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 200 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,14 +1,11 @@
"""The bluetooth integration."""
from __future__ import annotations
from asyncio import Future
from collections.abc import Callable, Iterable
import datetime
import logging
import platform
from typing import TYPE_CHECKING, cast
from typing import TYPE_CHECKING
import async_timeout
from awesomeversion import AwesomeVersion
from bluetooth_adapters import (
ADAPTER_ADDRESS,
@ -29,7 +26,7 @@ from homeassistant.config_entries import (
ConfigEntry,
)
from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback
from homeassistant.core import HomeAssistant, callback as hass_callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr, discovery_flow
from homeassistant.helpers.debounce import Debouncer
@ -42,6 +39,21 @@ from homeassistant.helpers.issue_registry import (
from homeassistant.loader import async_get_bluetooth
from . import models
from .api import (
_get_manager,
async_address_present,
async_ble_device_from_address,
async_discovered_service_info,
async_get_advertisement_callback,
async_get_scanner,
async_last_service_info,
async_process_advertisements,
async_rediscover_address,
async_register_callback,
async_register_scanner,
async_scanner_count,
async_track_unavailable,
)
from .base_scanner import BaseHaRemoteScanner, BaseHaScanner
from .const import (
BLUETOOTH_DISCOVERY_COOLDOWN_SECONDS,
@ -56,21 +68,15 @@ from .const import (
)
from .manager import BluetoothManager
from .match import BluetoothCallbackMatcher, IntegrationMatcher
from .models import (
BluetoothCallback,
BluetoothChange,
BluetoothScanningMode,
ProcessAdvertisementCallback,
)
from .models import BluetoothCallback, BluetoothChange, BluetoothScanningMode
from .scanner import HaScanner, ScannerStartError
from .wrappers import HaBleakScannerWrapper, HaBluetoothConnector
from .wrappers import HaBluetoothConnector
if TYPE_CHECKING:
from bleak.backends.device import BLEDevice
from homeassistant.helpers.typing import ConfigType
__all__ = [
"async_address_present",
"async_ble_device_from_address",
"async_discovered_service_info",
"async_get_scanner",
@ -83,6 +89,8 @@ __all__ = [
"async_scanner_count",
"BaseHaScanner",
"BaseHaRemoteScanner",
"BluetoothCallbackMatcher",
"BluetoothChange",
"BluetoothServiceInfo",
"BluetoothServiceInfoBleak",
"BluetoothScanningMode",
@ -97,151 +105,7 @@ _LOGGER = logging.getLogger(__name__)
RECOMMENDED_MIN_HAOS_VERSION = AwesomeVersion("9.0.dev0")
def _get_manager(hass: HomeAssistant) -> BluetoothManager:
"""Get the bluetooth manager."""
return cast(BluetoothManager, hass.data[DATA_MANAGER])
@hass_callback
def async_get_scanner(hass: HomeAssistant) -> HaBleakScannerWrapper:
"""Return a HaBleakScannerWrapper.
This is a wrapper around our BleakScanner singleton that allows
multiple integrations to share the same BleakScanner.
"""
return HaBleakScannerWrapper()
@hass_callback
def async_scanner_count(hass: HomeAssistant, connectable: bool = True) -> int:
"""Return the number of scanners currently in use."""
return _get_manager(hass).async_scanner_count(connectable)
@hass_callback
def async_discovered_service_info(
hass: HomeAssistant, connectable: bool = True
) -> Iterable[BluetoothServiceInfoBleak]:
"""Return the discovered devices list."""
if DATA_MANAGER not in hass.data:
return []
return _get_manager(hass).async_discovered_service_info(connectable)
@hass_callback
def async_last_service_info(
hass: HomeAssistant, address: str, connectable: bool = True
) -> BluetoothServiceInfoBleak | None:
"""Return the last service info for an address."""
if DATA_MANAGER not in hass.data:
return None
return _get_manager(hass).async_last_service_info(address, connectable)
@hass_callback
def async_ble_device_from_address(
hass: HomeAssistant, address: str, connectable: bool = True
) -> BLEDevice | None:
"""Return BLEDevice for an address if its present."""
if DATA_MANAGER not in hass.data:
return None
return _get_manager(hass).async_ble_device_from_address(address, connectable)
@hass_callback
def async_address_present(
hass: HomeAssistant, address: str, connectable: bool = True
) -> bool:
"""Check if an address is present in the bluetooth device list."""
if DATA_MANAGER not in hass.data:
return False
return _get_manager(hass).async_address_present(address, connectable)
@hass_callback
def async_register_callback(
hass: HomeAssistant,
callback: BluetoothCallback,
match_dict: BluetoothCallbackMatcher | None,
mode: BluetoothScanningMode,
) -> Callable[[], None]:
"""Register to receive a callback on bluetooth change.
mode is currently not used as we only support active scanning.
Passive scanning will be available in the future. The flag
is required to be present to avoid a future breaking change
when we support passive scanning.
Returns a callback that can be used to cancel the registration.
"""
return _get_manager(hass).async_register_callback(callback, match_dict)
async def async_process_advertisements(
hass: HomeAssistant,
callback: ProcessAdvertisementCallback,
match_dict: BluetoothCallbackMatcher,
mode: BluetoothScanningMode,
timeout: int,
) -> BluetoothServiceInfoBleak:
"""Process advertisements until callback returns true or timeout expires."""
done: Future[BluetoothServiceInfoBleak] = Future()
@hass_callback
def _async_discovered_device(
service_info: BluetoothServiceInfoBleak, change: BluetoothChange
) -> None:
if not done.done() and callback(service_info):
done.set_result(service_info)
unload = _get_manager(hass).async_register_callback(
_async_discovered_device, match_dict
)
try:
async with async_timeout.timeout(timeout):
return await done
finally:
unload()
@hass_callback
def async_track_unavailable(
hass: HomeAssistant,
callback: Callable[[BluetoothServiceInfoBleak], None],
address: str,
connectable: bool = True,
) -> Callable[[], None]:
"""Register to receive a callback when an address is unavailable.
Returns a callback that can be used to cancel the registration.
"""
return _get_manager(hass).async_track_unavailable(callback, address, connectable)
@hass_callback
def async_rediscover_address(hass: HomeAssistant, address: str) -> None:
"""Trigger discovery of devices which have already been seen."""
_get_manager(hass).async_rediscover_address(address)
@hass_callback
def async_register_scanner(
hass: HomeAssistant, scanner: BaseHaScanner, connectable: bool
) -> CALLBACK_TYPE:
"""Register a BleakScanner."""
return _get_manager(hass).async_register_scanner(scanner, connectable)
@hass_callback
def async_get_advertisement_callback(
hass: HomeAssistant,
) -> Callable[[BluetoothServiceInfoBleak], None]:
"""Get the advertisement callback."""
return _get_manager(hass).scanner_adv_received
async def async_get_adapter_from_address(
async def _async_get_adapter_from_address(
hass: HomeAssistant, address: str
) -> str | None:
"""Get an adapter by the address."""
@ -419,7 +283,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry for a bluetooth scanner."""
address = entry.unique_id
assert address is not None
adapter = await async_get_adapter_from_address(hass, address)
adapter = await _async_get_adapter_from_address(hass, address)
if adapter is None:
raise ConfigEntryNotReady(
f"Bluetooth adapter {adapter} with address {address} not found"

173
homeassistant/components/bluetooth/api.py Normal file
View File

@ -0,0 +1,173 @@
"""The bluetooth integration apis.
These APIs are the only documented way to interact with the bluetooth integration.
"""
from __future__ import annotations
from asyncio import Future
from collections.abc import Callable, Iterable
from typing import TYPE_CHECKING, cast
import async_timeout
from home_assistant_bluetooth import BluetoothServiceInfoBleak
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback
from .base_scanner import BaseHaScanner
from .const import DATA_MANAGER
from .manager import BluetoothManager
from .match import BluetoothCallbackMatcher
from .models import (
BluetoothCallback,
BluetoothChange,
BluetoothScanningMode,
ProcessAdvertisementCallback,
)
from .wrappers import HaBleakScannerWrapper
if TYPE_CHECKING:
from bleak.backends.device import BLEDevice
def _get_manager(hass: HomeAssistant) -> BluetoothManager:
"""Get the bluetooth manager."""
return cast(BluetoothManager, hass.data[DATA_MANAGER])
@hass_callback
def async_get_scanner(hass: HomeAssistant) -> HaBleakScannerWrapper:
"""Return a HaBleakScannerWrapper.
This is a wrapper around our BleakScanner singleton that allows
multiple integrations to share the same BleakScanner.
"""
return HaBleakScannerWrapper()
@hass_callback
def async_scanner_count(hass: HomeAssistant, connectable: bool = True) -> int:
"""Return the number of scanners currently in use."""
return _get_manager(hass).async_scanner_count(connectable)
@hass_callback
def async_discovered_service_info(
hass: HomeAssistant, connectable: bool = True
) -> Iterable[BluetoothServiceInfoBleak]:
"""Return the discovered devices list."""
if DATA_MANAGER not in hass.data:
return []
return _get_manager(hass).async_discovered_service_info(connectable)
@hass_callback
def async_last_service_info(
hass: HomeAssistant, address: str, connectable: bool = True
) -> BluetoothServiceInfoBleak | None:
"""Return the last service info for an address."""
if DATA_MANAGER not in hass.data:
return None
return _get_manager(hass).async_last_service_info(address, connectable)
@hass_callback
def async_ble_device_from_address(
hass: HomeAssistant, address: str, connectable: bool = True
) -> BLEDevice | None:
"""Return BLEDevice for an address if its present."""
if DATA_MANAGER not in hass.data:
return None
return _get_manager(hass).async_ble_device_from_address(address, connectable)
@hass_callback
def async_address_present(
hass: HomeAssistant, address: str, connectable: bool = True
) -> bool:
"""Check if an address is present in the bluetooth device list."""
if DATA_MANAGER not in hass.data:
return False
return _get_manager(hass).async_address_present(address, connectable)
@hass_callback
def async_register_callback(
hass: HomeAssistant,
callback: BluetoothCallback,
match_dict: BluetoothCallbackMatcher | None,
mode: BluetoothScanningMode,
) -> Callable[[], None]:
"""Register to receive a callback on bluetooth change.
mode is currently not used as we only support active scanning.
Passive scanning will be available in the future. The flag
is required to be present to avoid a future breaking change
when we support passive scanning.
Returns a callback that can be used to cancel the registration.
"""
return _get_manager(hass).async_register_callback(callback, match_dict)
async def async_process_advertisements(
hass: HomeAssistant,
callback: ProcessAdvertisementCallback,
match_dict: BluetoothCallbackMatcher,
mode: BluetoothScanningMode,
timeout: int,
) -> BluetoothServiceInfoBleak:
"""Process advertisements until callback returns true or timeout expires."""
done: Future[BluetoothServiceInfoBleak] = Future()
@hass_callback
def _async_discovered_device(
service_info: BluetoothServiceInfoBleak, change: BluetoothChange
) -> None:
if not done.done() and callback(service_info):
done.set_result(service_info)
unload = _get_manager(hass).async_register_callback(
_async_discovered_device, match_dict
)
try:
async with async_timeout.timeout(timeout):
return await done
finally:
unload()
@hass_callback
def async_track_unavailable(
hass: HomeAssistant,
callback: Callable[[BluetoothServiceInfoBleak], None],
address: str,
connectable: bool = True,
) -> Callable[[], None]:
"""Register to receive a callback when an address is unavailable.
Returns a callback that can be used to cancel the registration.
"""
return _get_manager(hass).async_track_unavailable(callback, address, connectable)
@hass_callback
def async_rediscover_address(hass: HomeAssistant, address: str) -> None:
"""Trigger discovery of devices which have already been seen."""
_get_manager(hass).async_rediscover_address(address)
@hass_callback
def async_register_scanner(
hass: HomeAssistant, scanner: BaseHaScanner, connectable: bool
) -> CALLBACK_TYPE:
"""Register a BleakScanner."""
return _get_manager(hass).async_register_scanner(scanner, connectable)
@hass_callback
def async_get_advertisement_callback(
hass: HomeAssistant,
) -> Callable[[BluetoothServiceInfoBleak], None]:
"""Get the advertisement callback."""
return _get_manager(hass).scanner_adv_received

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

@ -9,12 +9,11 @@ from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
import pytest
from homeassistant.components.bluetooth import (
BaseHaScanner,
from homeassistant.components.bluetooth import BaseHaScanner, HaBluetoothConnector
from homeassistant.components.bluetooth.wrappers import (
HaBleakClientWrapper,
HaBleakScannerWrapper,
HaBluetoothConnector,
)
from homeassistant.components.bluetooth.wrappers import HaBleakClientWrapper
from . import (
MockBleakClient,