From 71e298ce8a2fe029308b75f867a4f0065af5677f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bojan=20Poto=C4=8Dnik?= Date: Thu, 24 Nov 2022 18:48:29 +0100 Subject: [PATCH] BleakScanner: Add ScanMode enum, support active scan mode fall-back Use ScanMode enum instead of a string literal for the `scanning_mode` parameter of the `BleakScanner`. Support fall-back to active scanning mode on platforms where passive scanning mode is not possible if `TRY_PASSIVE` mode is used. Add noqa comment to avoid "N806 variable ... in function should be lowercase". --- CHANGELOG.rst | 2 ++ bleak/__init__.py | 21 ++++++++--- bleak/backends/bluezdbus/scanner.py | 47 ++++++++++++++++--------- bleak/backends/corebluetooth/scanner.py | 17 ++++++--- bleak/backends/p4android/scanner.py | 16 ++++----- bleak/backends/scanner.py | 38 ++++++++++++++++++++ bleak/backends/winrt/scanner.py | 18 +++++----- examples/passive_scan.py | 21 ++--------- 8 files changed, 118 insertions(+), 62 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 193aaed8..b12c1a58 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -15,6 +15,7 @@ Added * Added optional hack to use Bluetooth address instead of UUID on macOS. * Added ``BleakScanner.find_device_by_name()`` class method. * Added ``BleakNoPassiveScanError`` exception. +* Added ``ScanMode`` enum for ``BleakScanner`` scanning type. Changed ------- @@ -24,6 +25,7 @@ Changed * ``BaseBleakClient.services`` is now ``None`` instead of empty service collection until services are discovered. * Include thread name in ``BLEAK_LOGGING`` output. Merged #1144. +* ``scanning_mode`` parameter of ``BleakScanner`` is now ``ScanMode`` enum instead of string literal. Fixed ----- diff --git a/bleak/__init__.py b/bleak/__init__.py index 2ced4b02..d0828796 100644 --- a/bleak/__init__.py +++ b/bleak/__init__.py @@ -14,6 +14,7 @@ import os import sys import uuid +import warnings from typing import ( TYPE_CHECKING, Awaitable, @@ -45,6 +46,7 @@ AdvertisementData, AdvertisementDataCallback, AdvertisementDataFilter, + ScanMode, BaseBleakScanner, get_platform_scanner_backend_type, ) @@ -88,9 +90,10 @@ class BleakScanner: containing this advertising data will be received. Required on macOS >= 12.0, < 12.3 (unless you create an app with ``py2app``). scanning_mode: - Set to ``"passive"`` to avoid the ``"active"`` scanning mode. + Set to ``ScanMode.PASSIVE`` or ``ScanMode.TRY_PASSIVE`` to avoid + the default active scanning mode. Passive scanning is not supported on macOS! Will raise - :class:`BleakError` if set to ``"passive"`` on macOS. + :class:`BleakError` if set to ``ScanMode.PASSIVE`` on macOS. bluez: Dictionary of arguments specific to the BlueZ backend. cb: @@ -114,14 +117,22 @@ def __init__( self, detection_callback: Optional[AdvertisementDataCallback] = None, service_uuids: Optional[List[str]] = None, - scanning_mode: Literal["active", "passive"] = "active", + scanning_mode: ScanMode = ScanMode.ACTIVE, *, bluez: BlueZScannerArgs = {}, cb: CBScannerArgs = {}, backend: Optional[Type[BaseBleakScanner]] = None, **kwargs, ): - PlatformBleakScanner = ( + if isinstance(scanning_mode, str): + scanning_mode = ScanMode(scanning_mode.lower()) + warnings.warn( + f"The scanning_mode is now {scanning_mode} instead of string literal", + DeprecationWarning, + stacklevel=2, + ) + + PlatformBleakScanner = ( # noqa: N806 get_platform_scanner_backend_type() if backend is None else backend ) @@ -418,7 +429,7 @@ def __init__( backend: Optional[Type[BaseBleakClient]] = None, **kwargs, ): - PlatformBleakClient = ( + PlatformBleakClient = ( # noqa: N806 get_platform_client_backend_type() if backend is None else backend ) diff --git a/bleak/backends/bluezdbus/scanner.py b/bleak/backends/bluezdbus/scanner.py index d9cce225..fbf90b83 100644 --- a/bleak/backends/bluezdbus/scanner.py +++ b/bleak/backends/bluezdbus/scanner.py @@ -6,12 +6,17 @@ from dbus_fast import Variant if sys.version_info[:2] < (3, 8): - from typing_extensions import Literal, TypedDict + from typing_extensions import TypedDict else: - from typing import Literal, TypedDict - -from ...exc import BleakError -from ..scanner import AdvertisementData, AdvertisementDataCallback, BaseBleakScanner + from typing import TypedDict + +from ...exc import BleakError, BleakNoPassiveScanError +from ..scanner import ( + AdvertisementData, + AdvertisementDataCallback, + BaseBleakScanner, + ScanMode, +) from .advertisement_monitor import OrPatternLike from .defs import Device1 from .manager import get_global_bluez_manager @@ -121,7 +126,7 @@ def __init__( self, detection_callback: Optional[AdvertisementDataCallback], service_uuids: Optional[List[str]], - scanning_mode: Literal["active", "passive"], + scanning_mode: ScanMode, *, bluez: BlueZScannerArgs, **kwargs, @@ -162,12 +167,12 @@ def __init__( self._or_patterns = bluez.get("or_patterns") - if self._scanning_mode == "passive" and service_uuids: + if self._scanning_mode.passive and service_uuids: logger.warning( "service uuid filtering is not implemented for passive scanning, use bluez or_patterns as a workaround" ) - if self._scanning_mode == "passive" and not self._or_patterns: + if self._scanning_mode.passive and not self._or_patterns: raise BleakError("passive scanning mode requires bluez or_patterns") async def start(self): @@ -180,14 +185,24 @@ async def start(self): self.seen_devices = {} - if self._scanning_mode == "passive": - self._stop = await manager.passive_scan( - adapter_path, - self._or_patterns, - self._handle_advertising_data, - self._handle_device_removed, - ) - else: + if self._scanning_mode.passive: + try: + self._stop = await manager.passive_scan( + adapter_path, + self._or_patterns, + self._handle_advertising_data, + self._handle_device_removed, + ) + except BleakNoPassiveScanError as e: + if self._scanning_mode == ScanMode.TRY_PASSIVE: + logger.warning( + f"Passive scan not possible, using active scan ({e})" + ) + self._scanning_mode = ScanMode.ACTIVE + else: + raise + + if self._scanning_mode == ScanMode.ACTIVE: self._stop = await manager.active_scan( adapter_path, self._filters, diff --git a/bleak/backends/corebluetooth/scanner.py b/bleak/backends/corebluetooth/scanner.py index 026187d3..ea136103 100644 --- a/bleak/backends/corebluetooth/scanner.py +++ b/bleak/backends/corebluetooth/scanner.py @@ -3,16 +3,21 @@ from typing import Any, Dict, List, Optional if sys.version_info[:2] < (3, 8): - from typing_extensions import Literal, TypedDict + from typing_extensions import TypedDict else: - from typing import Literal, TypedDict + from typing import TypedDict import objc from CoreBluetooth import CBPeripheral from Foundation import NSBundle from ...exc import BleakNoPassiveScanError -from ..scanner import AdvertisementData, AdvertisementDataCallback, BaseBleakScanner +from ..scanner import ( + AdvertisementData, + AdvertisementDataCallback, + BaseBleakScanner, + ScanMode, +) from .CentralManagerDelegate import CentralManagerDelegate from .utils import cb_uuid_to_str @@ -65,7 +70,7 @@ def __init__( self, detection_callback: Optional[AdvertisementDataCallback], service_uuids: Optional[List[str]], - scanning_mode: Literal["active", "passive"], + scanning_mode: ScanMode, *, cb: CBScannerArgs, **kwargs @@ -76,8 +81,10 @@ def __init__( self._use_bdaddr = cb.get("use_bdaddr", False) - if scanning_mode == "passive": + if scanning_mode == ScanMode.PASSIVE: raise BleakNoPassiveScanError("macOS does not support passive scanning") + elif scanning_mode == ScanMode.TRY_PASSIVE: + logger.warning("macOS does not support passive scanning, using active scan") self._manager = CentralManagerDelegate.alloc().init() self._timeout: float = kwargs.get("timeout", 5.0) diff --git a/bleak/backends/p4android/scanner.py b/bleak/backends/p4android/scanner.py index 609e5846..3fe38352 100644 --- a/bleak/backends/p4android/scanner.py +++ b/bleak/backends/p4android/scanner.py @@ -11,17 +11,17 @@ else: from asyncio import timeout as async_timeout -if sys.version_info[:2] < (3, 8): - from typing_extensions import Literal -else: - from typing import Literal - from android.broadcast import BroadcastReceiver from android.permissions import Permission, request_permissions from jnius import cast, java_method from ...exc import BleakError -from ..scanner import AdvertisementData, AdvertisementDataCallback, BaseBleakScanner +from ..scanner import ( + AdvertisementData, + AdvertisementDataCallback, + BaseBleakScanner, + ScanMode, +) from . import defs, utils logger = logging.getLogger(__name__) @@ -49,12 +49,12 @@ def __init__( self, detection_callback: Optional[AdvertisementDataCallback], service_uuids: Optional[List[str]], - scanning_mode: Literal["active", "passive"], + scanning_mode: ScanMode, **kwargs, ): super(BleakScannerP4Android, self).__init__(detection_callback, service_uuids) - if scanning_mode == "passive": + if scanning_mode.passive: self.__scan_mode = defs.ScanSettings.SCAN_MODE_OPPORTUNISTIC else: self.__scan_mode = defs.ScanSettings.SCAN_MODE_LOW_LATENCY diff --git a/bleak/backends/scanner.py b/bleak/backends/scanner.py index 359c6ce5..3b3dca58 100644 --- a/bleak/backends/scanner.py +++ b/bleak/backends/scanner.py @@ -1,5 +1,6 @@ import abc import asyncio +import enum import inspect import os import platform @@ -85,6 +86,43 @@ def __repr__(self) -> str: return f"AdvertisementData({', '.join(kwargs)})" +class ScanMode(enum.Enum): + """Bluetooth scanning mode""" + + ACTIVE = "active" + """ + Perform active scan (default for most platforms) + + This type of scan is typically used when the potential Central device would like more info + than can be provided in an ADV_IND packet, before making a decision to connect to it. After + advertisement is received, the Scanner will request more information by issuing a SCAN_REQ + packet in the advertising interval. The Advertiser responds with more information (like + friendly name and supported profiles) in a SCAN_RSP packet. + """ + + PASSIVE = "passive" + """ + Perform passive scan (not supported on all platforms) + + In passive scanning mode, the Scanner just listens for advertising packets. When such packet + is detected, the module reports the discovered device. The Advertiser is never aware that + packets were received by the Scanner. + """ + + TRY_PASSIVE = "try_passive" + """ + Try to use passive scanning mode, if not possible fall-back to active scanning mode + + As passive scanning mode is not possible on all platforms, this option can be used to prefer + passive scanning, or automatically switch to active scanning if passive is not possible. + """ + + @property + def passive(self) -> bool: + """Helper method to check whether passive mode should [try to] be used [first]""" + return self in (self.PASSIVE, self.TRY_PASSIVE) + + AdvertisementDataCallback = Callable[ [BLEDevice, AdvertisementData], Optional[Awaitable[None]], diff --git a/bleak/backends/winrt/scanner.py b/bleak/backends/winrt/scanner.py index 1da9553c..5629bb08 100644 --- a/bleak/backends/winrt/scanner.py +++ b/bleak/backends/winrt/scanner.py @@ -1,6 +1,5 @@ import asyncio import logging -import sys from typing import Dict, List, NamedTuple, Optional from uuid import UUID @@ -11,12 +10,12 @@ BluetoothLEAdvertisementType, ) -if sys.version_info[:2] < (3, 8): - from typing_extensions import Literal -else: - from typing import Literal - -from ..scanner import AdvertisementDataCallback, BaseBleakScanner, AdvertisementData +from ..scanner import ( + AdvertisementDataCallback, + BaseBleakScanner, + AdvertisementData, + ScanMode, +) from ...assigned_numbers import AdvertisementDataType @@ -73,7 +72,7 @@ def __init__( self, detection_callback: Optional[AdvertisementDataCallback], service_uuids: Optional[List[str]], - scanning_mode: Literal["active", "passive"], + scanning_mode: ScanMode, **kwargs, ): super(BleakScannerWinRT, self).__init__(detection_callback, service_uuids) @@ -82,8 +81,7 @@ def __init__( self._advertisement_pairs: Dict[int, _RawAdvData] = {} self._stopped_event = None - # case insensitivity is for backwards compatibility on Windows only - if scanning_mode.lower() == "passive": + if scanning_mode.passive: self._scanning_mode = BluetoothLEScanningMode.PASSIVE else: self._scanning_mode = BluetoothLEScanningMode.ACTIVE diff --git a/examples/passive_scan.py b/examples/passive_scan.py index 3d6af621..3e68b84e 100644 --- a/examples/passive_scan.py +++ b/examples/passive_scan.py @@ -9,12 +9,11 @@ """ import argparse import asyncio -import contextlib import logging from typing import Optional, List, Dict, Any import bleak -from bleak import BLEDevice, AdvertisementData +from bleak import BleakScanner, BLEDevice, AdvertisementData, ScanMode logger = logging.getLogger(__name__) @@ -70,30 +69,16 @@ def get_core_bluetooth_scanning_params() -> Dict[str, Any]: }.get(bleak.get_platform_scanner_backend_type().__name__, lambda: {})() -@contextlib.asynccontextmanager -async def scanner(**kwargs): - try: - async with bleak.BleakScanner(**kwargs) as bleak_scanner: - yield bleak_scanner - - except bleak.exc.BleakNoPassiveScanError as e: - logger.error(f"Passive scanning not possible, using active scanning ({e})") - - del kwargs["scanning_mode"] - async with bleak.BleakScanner(**kwargs) as bleak_scanner: - yield bleak_scanner - - async def main(args: argparse.Namespace): def scan_callback(device: BLEDevice, adv_data: AdvertisementData): logger.info("%s: %r", device.address, adv_data) - async with scanner( + async with BleakScanner( detection_callback=scan_callback, **_get_os_specific_scanning_params( uuids=args.services, macos_use_bdaddr=args.macos_use_bdaddr ), - scanning_mode="passive", + scanning_mode=ScanMode.TRY_PASSIVE, ): await asyncio.sleep(60)