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".

Author: bojanpotocnik

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
 -----

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
         )
 

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,

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)

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

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]],

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

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)