Add passive_scan.py example

Author: bojanpotocnik

examples/passive_scan.py

@@ -0,0 +1,125 @@
+"""
+Scanner using passive scanning mode
+--------------
+
+Example similar to detection_callback.py, but using passive scanning
+
+Updated on 2022-11-24 by bojanpotocnik <[email protected]>
+
+"""
+import argparse
+import asyncio
+import contextlib
+import logging
+from typing import Optional, List, Dict, Any
+
+import bleak
+from bleak import BLEDevice, AdvertisementData
+
+logger = logging.getLogger(__name__)
+
+
+def _get_os_specific_scanning_params(
+    uuids: Optional[List[str]],
+    rssi: Optional[int] = None,
+    macos_use_bdaddr: bool = False,
+) -> Dict[str, Any]:
+    def get_bluez_dbus_scanning_params() -> Dict[str, Any]:
+        from bleak.assigned_numbers import AdvertisementDataType
+        from bleak.backends.bluezdbus.advertisement_monitor import OrPattern
+        from bleak.backends.bluezdbus.scanner import (
+            BlueZScannerArgs,
+            BlueZDiscoveryFilters,
+        )
+
+        filters = BlueZDiscoveryFilters(
+            # UUIDs= Added below, because it cannot be None
+            # RSSI= Added below, because it cannot be None
+            Transport="le",
+            DuplicateData=True,
+        )
+        if uuids:
+            filters["UUIDs"] = uuids
+        if rssi:
+            filters["RSSI"] = rssi
+
+        # or_patterns ar required for BlueZ passive scanning
+        or_patterns = [
+            # General Discoverable (peripherals)
+            OrPattern(0, AdvertisementDataType.FLAGS, b"\x02"),
+            # BR/EDR Not Supported (BLE peripherals)
+            OrPattern(0, AdvertisementDataType.FLAGS, b"\x04"),
+            # General Discoverable, BR/EDR Not Supported (BLE peripherals)
+            OrPattern(0, AdvertisementDataType.FLAGS, b"\x06"),
+            # General Discoverable, LE and BR/EDR Capable (Controller), LE and BR/EDR Capable (Host) (computers, phones)
+            OrPattern(0, AdvertisementDataType.FLAGS, b"\x1A"),
+        ]
+
+        return {"bluez": BlueZScannerArgs(filters=filters, or_patterns=or_patterns)}
+
+    def get_core_bluetooth_scanning_params() -> Dict[str, Any]:
+        from bleak.backends.corebluetooth.scanner import CBScannerArgs
+
+        return {"cb": CBScannerArgs(use_bdaddr=macos_use_bdaddr)}
+
+    return {
+        "BleakScannerBlueZDBus": get_bluez_dbus_scanning_params,
+        "BleakScannerCoreBluetooth": get_core_bluetooth_scanning_params,
+        # "BleakScannerP4Android": get_p4android_scanning_params,
+        # "BleakScannerWinRT": get_winrt_scanning_params,
+    }.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(
+        detection_callback=scan_callback,
+        **_get_os_specific_scanning_params(
+            uuids=args.services, macos_use_bdaddr=args.macos_use_bdaddr
+        ),
+        scanning_mode="passive",
+    ):
+        await asyncio.sleep(60)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)-15s %(name)-8s %(levelname)s: %(message)s",
+    )
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--macos-use-bdaddr",
+        action="store_true",
+        help="when true use Bluetooth address instead of UUID on macOS",
+    )
+    parser.add_argument(
+        "--services",
+        metavar="<uuid>",
+        nargs="*",
+        help="UUIDs of one or more services to filter for",
+    )
+
+    arguments = parser.parse_args()
+
+    try:
+        asyncio.run(main(arguments))
+    except KeyboardInterrupt:
+        pass