From 487782a6d12019b6ffca6a51080fc5b8779f4715 Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Sun, 8 Jan 2023 22:20:02 +0100 Subject: [PATCH] Code styling tweaks to Bluetooth (#85448) Co-authored-by: J. Nick Koston --- .../bluetooth/active_update_coordinator.py | 20 ++++++---- .../bluetooth/active_update_processor.py | 25 ++++++++----- .../components/bluetooth/base_scanner.py | 9 +++-- homeassistant/components/bluetooth/manager.py | 37 +++++++++++-------- homeassistant/components/bluetooth/match.py | 5 ++- homeassistant/components/bluetooth/usage.py | 11 ++++-- homeassistant/components/bluetooth/util.py | 5 ++- .../components/bluetooth/wrappers.py | 25 ++++++++----- 8 files changed, 87 insertions(+), 50 deletions(-) diff --git a/homeassistant/components/bluetooth/active_update_coordinator.py b/homeassistant/components/bluetooth/active_update_coordinator.py index 5371d9f99fa..09567aada05 100644 --- a/homeassistant/components/bluetooth/active_update_coordinator.py +++ b/homeassistant/components/bluetooth/active_update_coordinator.py @@ -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(....) diff --git a/homeassistant/components/bluetooth/active_update_processor.py b/homeassistant/components/bluetooth/active_update_processor.py index e175fc665f4..b91ac2cbf4d 100644 --- a/homeassistant/components/bluetooth/active_update_processor.py +++ b/homeassistant/components/bluetooth/active_update_processor.py @@ -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(....) diff --git a/homeassistant/components/bluetooth/base_scanner.py b/homeassistant/components/bluetooth/base_scanner.py index b4c88260591..8868b0a0883 100644 --- a/homeassistant/components/bluetooth/base_scanner.py +++ b/homeassistant/components/bluetooth/base_scanner.py @@ -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 diff --git a/homeassistant/components/bluetooth/manager.py b/homeassistant/components/bluetooth/manager.py index 748b685d866..c863299d206 100644 --- a/homeassistant/components/bluetooth/manager.py +++ b/homeassistant/components/bluetooth/manager.py @@ -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)) diff --git a/homeassistant/components/bluetooth/match.py b/homeassistant/components/bluetooth/match.py index 1a59ee6fe4c..a7308bfd7ff 100644 --- a/homeassistant/components/bluetooth/match.py +++ b/homeassistant/components/bluetooth/match.py @@ -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.""" diff --git a/homeassistant/components/bluetooth/usage.py b/homeassistant/components/bluetooth/usage.py index 0b1e615ddda..b751559e7a4 100644 --- a/homeassistant/components/bluetooth/usage.py +++ b/homeassistant/components/bluetooth/usage.py @@ -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): diff --git a/homeassistant/components/bluetooth/util.py b/homeassistant/components/bluetooth/util.py index 5419fa79e1c..e78eb51a38c 100644 --- a/homeassistant/components/bluetooth/util.py +++ b/homeassistant/components/bluetooth/util.py @@ -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] = {} diff --git a/homeassistant/components/bluetooth/wrappers.py b/homeassistant/components/bluetooth/wrappers.py index a2c417ca382..4a1be63903f 100644 --- a/homeassistant/components/bluetooth/wrappers.py +++ b/homeassistant/components/bluetooth/wrappers.py @@ -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, )