Skip to content

Commit

Permalink
scanner: Add ScanMode enum, support active scan mode fall-back
Browse files Browse the repository at this point in the history
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".
  • Loading branch information
bojanpotocnik committed Nov 24, 2022
1 parent 68ee241 commit a74f1d9
Show file tree
Hide file tree
Showing 8 changed files with 118 additions and 62 deletions.
2 changes: 2 additions & 0 deletions CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ Added
* Added ``BleakScanner.find_device_by_name()`` class method.
* Added check to verify that BlueZ Advertisement Monitor was actually registered. Solves #1136.
* Added ``BleakNoPassiveScanError`` exception.
* Added ``ScanMode`` enum for ``BleakScanner`` scanning type.

Changed
-------
Expand All @@ -25,6 +26,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
-----
Expand Down
21 changes: 16 additions & 5 deletions bleak/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import os
import sys
import uuid
import warnings
from typing import (
TYPE_CHECKING,
Awaitable,
Expand Down Expand Up @@ -45,6 +46,7 @@
AdvertisementData,
AdvertisementDataCallback,
AdvertisementDataFilter,
ScanMode,
BaseBleakScanner,
get_platform_scanner_backend_type,
)
Expand Down Expand Up @@ -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:
Expand All @@ -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
)

Expand Down Expand Up @@ -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
)

Expand Down
47 changes: 31 additions & 16 deletions bleak/backends/bluezdbus/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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):
Expand All @@ -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,
Expand Down
17 changes: 12 additions & 5 deletions bleak/backends/corebluetooth/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand Down
16 changes: 8 additions & 8 deletions bleak/backends/p4android/scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)
Expand Down Expand Up @@ -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
Expand Down
38 changes: 38 additions & 0 deletions bleak/backends/scanner.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import abc
import asyncio
import enum
import inspect
import os
import platform
Expand Down Expand Up @@ -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]],
Expand Down
18 changes: 8 additions & 10 deletions bleak/backends/winrt/scanner.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import asyncio
import logging
import sys
from typing import Dict, List, NamedTuple, Optional
from uuid import UUID

Expand All @@ -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


Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand Down
21 changes: 3 additions & 18 deletions examples/passive_scan.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)

Expand Down Expand Up @@ -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)

Expand Down

0 comments on commit a74f1d9

Please sign in to comment.