jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Code styling tweaks to Bluetooth (#85448)

Browse Source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Franck Nijhof 2023-01-08 22:20:02 +01:00 committed by GitHub
parent 800b8abe39
commit 487782a6d1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 87 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
homeassistant/components/bluetooth/active_update_coordinator.py
View File

@ -1,4 +1,7 @@
"""A Bluetooth passive coordinator that receives data from advertisements but can also poll."""
"""A Bluetooth passive coordinator.
Receives data from advertisements but can also poll.
"""
from __future__ import annotations
from collections.abc import Callable, Coroutine
@ -33,16 +36,19 @@ class ActiveBluetoothDataUpdateCoordinator(
out if a poll is needed. This should return True if it is and False if it is
not needed.
def needs_poll_method(svc_info: BluetoothServiceInfoBleak, last_poll: float | None) -> bool:
def needs_poll_method(
svc_info: BluetoothServiceInfoBleak,
last_poll: float | None
) -> bool:
return True
If there has been no poll since HA started, `last_poll` will be None. Otherwise it is
the number of seconds since one was last attempted.
If there has been no poll since HA started, `last_poll` will be None.
Otherwise it is the number of seconds since one was last attempted.
If a poll is needed, the coordinator will call poll_method. This is a coroutine.
It should return the same type of data as your update_method. The expectation is that
data from advertisements and from polling are being parsed and fed into a shared
object that represents the current state of the device.
It should return the same type of data as your update_method. The expectation is
that data from advertisements and from polling are being parsed and fed into
a shared object that represents the current state of the device.
async def poll_method(svc_info: BluetoothServiceInfoBleak) -> YourDataType:
return YourDataType(....)

25
homeassistant/components/bluetooth/active_update_processor.py
View File

@ -1,4 +1,7 @@
"""A Bluetooth passive processor coordinator that collects data from advertisements but can also poll."""
"""A Bluetooth passive processor coordinator.
Collects data from advertisements but can also poll.
"""
from __future__ import annotations
from collections.abc import Callable, Coroutine
@ -23,23 +26,27 @@ _T = TypeVar("_T")
class ActiveBluetoothProcessorCoordinator(
Generic[_T], PassiveBluetoothProcessorCoordinator[_T]
):
"""
A processor coordinator that parses passive data from advertisements but can also poll.
"""A processor coordinator that parses passive data.
Parses passive data from advertisements but can also poll.
Every time an advertisement is received, needs_poll_method is called to work
out if a poll is needed. This should return True if it is and False if it is
not needed.
def needs_poll_method(svc_info: BluetoothServiceInfoBleak, last_poll: float | None) -> bool:
def needs_poll_method(
svc_info: BluetoothServiceInfoBleak,
last_poll: float | None
) -> bool:
return True
If there has been no poll since HA started, `last_poll` will be None. Otherwise it is
the number of seconds since one was last attempted.
If there has been no poll since HA started, `last_poll` will be None.
Otherwise it is the number of seconds since one was last attempted.
If a poll is needed, the coordinator will call poll_method. This is a coroutine.
It should return the same type of data as your update_method. The expectation is that
data from advertisements and from polling are being parsed and fed into a shared
object that represents the current state of the device.
It should return the same type of data as your update_method. The expectation is
that data from advertisements and from polling are being parsed and fed into a
shared object that represents the current state of the device.
async def poll_method(svc_info: BluetoothServiceInfoBleak) -> YourDataType:
return YourDataType(....)

9
homeassistant/components/bluetooth/base_scanner.py
View File

@ -107,7 +107,8 @@ class BaseHaScanner(ABC):
def _async_scanner_watchdog(self, now: datetime.datetime) -> None:
"""Check if the scanner is running.
Override this method if you need to do something else when the watchdog is triggered.
Override this method if you need to do something else when the watchdog
is triggered.
"""
if self._async_watchdog_triggered():
_LOGGER.info(
@ -144,6 +145,7 @@ class BaseHaScanner(ABC):
async def async_diagnostics(self) -> dict[str, Any]:
"""Return diagnostic information about the scanner."""
device_adv_datas = self.discovered_devices_and_advertisement_data.values()
return {
"name": self.name,
"start_time": self._start_time,
@ -160,7 +162,7 @@ class BaseHaScanner(ABC):
"advertisement_data": device_adv[1],
"details": device_adv[0].details,
}
for device_adv in self.discovered_devices_and_advertisement_data.values()
for device_adv in device_adv_datas
],
}
@ -258,9 +260,10 @@ class BaseHaRemoteScanner(BaseHaScanner):
@property
def discovered_devices(self) -> list[BLEDevice]:
"""Return a list of discovered devices."""
device_adv_datas = self._discovered_device_advertisement_datas.values()
return [
device_advertisement_data[0]
for device_advertisement_data in self._discovered_device_advertisement_datas.values()
for device_advertisement_data in device_adv_datas
]
@property

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

@ -225,15 +225,17 @@ class BluetoothManager:
results: list[tuple[BaseHaScanner, BLEDevice, AdvertisementData]] = []
for type_ in types_:
for scanner in self._get_scanners_by_type(type_):
if device_advertisement_data := scanner.discovered_devices_and_advertisement_data.get(
address
):
results.append((scanner, *device_advertisement_data))
devices_and_adv_data = scanner.discovered_devices_and_advertisement_data
if device_adv_data := devices_and_adv_data.get(address):
results.append((scanner, *device_adv_data))
return results
@hass_callback
def _async_all_discovered_addresses(self, connectable: bool) -> Iterable[str]:
"""Return all of discovered addresses from all the scanners including duplicates."""
"""Return all of discovered addresses.
Include addresses from all the scanners including duplicates.
"""
yield from itertools.chain.from_iterable(
scanner.discovered_devices_and_advertisement_data
for scanner in self._get_scanners_by_type(True)
@ -281,9 +283,9 @@ class BluetoothManager:
#
# 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
#
# 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 advertising_interval := intervals.get(address):
time_since_seen = monotonic_now - all_history[address].time
if time_since_seen <= advertising_interval:
@ -335,7 +337,8 @@ class BluetoothManager:
if (new.rssi or NO_RSSI_VALUE) - RSSI_SWITCH_THRESHOLD > (
old.rssi or NO_RSSI_VALUE
):
# If new advertisement is RSSI_SWITCH_THRESHOLD more, the new one is preferred
# If new advertisement is RSSI_SWITCH_THRESHOLD more,
# the new one is preferred.
if debug:
_LOGGER.debug(
(
@ -381,19 +384,21 @@ class BluetoothManager:
source = service_info.source
debug = _LOGGER.isEnabledFor(logging.DEBUG)
# This logic is complex due to the many combinations of scanners that are supported.
# This logic is complex due to the many combinations of scanners
# that are supported.
#
# We need to handle multiple connectable and non-connectable scanners
# and we need to handle the case where a device is connectable on one scanner
# but not on another.
#
# The device may also be connectable only by a scanner that has worse signal strength
# than a non-connectable scanner.
# The device may also be connectable only by a scanner that has worse
# signal strength than a non-connectable scanner.
#
# all_history - the history of all advertisements from all scanners with the best
# advertisement from each scanner
# connectable_history - the history of all connectable advertisements from all scanners
# with the best advertisement from each connectable scanner
# all_history - the history of all advertisements from all scanners with the
# best advertisement from each scanner
# connectable_history - the history of all connectable advertisements from all
# scanners with the best advertisement from each
# connectable scanner
#
if (
(old_service_info := all_history.get(address))

5
homeassistant/components/bluetooth/match.py
View File

@ -282,7 +282,10 @@ class BluetoothMatcherIndex(BluetoothMatcherIndexBase[BluetoothMatcher]):
class BluetoothCallbackMatcherIndex(
BluetoothMatcherIndexBase[BluetoothCallbackMatcherWithCallback]
):
"""Bluetooth matcher for the bluetooth integration that supports matching on addresses."""
"""Bluetooth matcher for the bluetooth integration.
Supports matching on addresses.
"""
def __init__(self) -> None:
"""Initialize the matcher index."""

11
homeassistant/components/bluetooth/usage.py
View File

@ -16,17 +16,22 @@ ORIGINAL_BLEAK_RETRY_CONNECTOR_CLIENT = (
def install_multiple_bleak_catcher() -> None:
"""Wrap the bleak classes to return the shared instance if multiple instances are detected."""
"""Wrap the bleak classes to return the shared instance.
In case multiple instances are detected.
"""
bleak.BleakScanner = HaBleakScannerWrapper # type: ignore[misc, assignment]
bleak.BleakClient = HaBleakClientWrapper # type: ignore[misc]
bleak_retry_connector.BleakClientWithServiceCache = HaBleakClientWithServiceCache # type: ignore[misc,assignment]
bleak_retry_connector.BleakClientWithServiceCache = HaBleakClientWithServiceCache # type: ignore[misc,assignment] # noqa: E501
def uninstall_multiple_bleak_catcher() -> None:
"""Unwrap the bleak classes."""
bleak.BleakScanner = ORIGINAL_BLEAK_SCANNER # type: ignore[misc]
bleak.BleakClient = ORIGINAL_BLEAK_CLIENT # type: ignore[misc]
bleak_retry_connector.BleakClientWithServiceCache = ORIGINAL_BLEAK_RETRY_CONNECTOR_CLIENT # type: ignore[misc]
bleak_retry_connector.BleakClientWithServiceCache = ( # type: ignore[misc]
ORIGINAL_BLEAK_RETRY_CONNECTOR_CLIENT
)
class HaBleakClientWithServiceCache(HaBleakClientWrapper):

5
homeassistant/components/bluetooth/util.py
View File

@ -15,7 +15,10 @@ from .storage import BluetoothStorage
def async_load_history_from_system(
adapters: BluetoothAdapters, storage: BluetoothStorage
) -> tuple[dict[str, BluetoothServiceInfoBleak], dict[str, BluetoothServiceInfoBleak]]:
"""Load the device and advertisement_data history if available on the current system."""
"""Load the device and advertisement_data history.
Only loads if available on the current system.
"""
now_monotonic = monotonic_time_coarse()
connectable_loaded_history: dict[str, BluetoothServiceInfoBleak] = {}
all_loaded_history: dict[str, BluetoothServiceInfoBleak] = {}

25
homeassistant/components/bluetooth/wrappers.py
View File

@ -119,10 +119,11 @@ class HaBleakScannerWrapper(BaseBleakScanner):
def register_detection_callback(
self, callback: AdvertisementDataCallback | None
) -> None:
"""Register a callback that is called when a device is discovered or has a property changed.
"""Register a detection callback.
This method takes the callback and registers it with the long running
scanner.
The callback is called when a device is discovered or has a property changed.
This method takes the callback and registers it with the long running sscanner.
"""
self._advertisement_data_callback = callback
self._setup_detection_callback()
@ -154,7 +155,9 @@ def _rssi_sorter_with_connection_failure_penalty(
connection_failure_count: dict[BaseHaScanner, int],
rssi_diff: int,
) -> float:
"""Get a sorted list of scanner, device, advertisement data adjusting for previous connection failures.
"""Get a sorted list of scanner, device, advertisement data.
Adjusting for previous connection failures.
When a connection fails, we want to try the next best adapter so we
apply a penalty to the RSSI value to make it less likely to be chosen
@ -227,7 +230,10 @@ class HaBleakClientWrapper(BleakClient):
"""Set the disconnect callback."""
self.__disconnected_callback = callback
if self._backend:
self._backend.set_disconnected_callback(callback, **kwargs) # type: ignore[arg-type]
self._backend.set_disconnected_callback(
callback, # type: ignore[arg-type]
**kwargs,
)
async def connect(self, **kwargs: Any) -> bool:
"""Connect to the specified GATT server."""
@ -294,15 +300,14 @@ class HaBleakClientWrapper(BleakClient):
that has a free connection slot.
"""
address = self.__address
scanner_device_advertisement_datas = manager.async_get_scanner_discovered_devices_and_advertisement_data_by_address(
scanner_device_advertisement_datas = manager.async_get_scanner_discovered_devices_and_advertisement_data_by_address( # noqa: E501
address, True
)
sorted_scanner_device_advertisement_datas = sorted(
scanner_device_advertisement_datas,
key=lambda scanner_device_advertisement_data: scanner_device_advertisement_data[
2
].rssi
or NO_RSSI_VALUE,
key=lambda scanner_device_advertisement_data: (
scanner_device_advertisement_data[2].rssi or NO_RSSI_VALUE
),
reverse=True,
)