Add --hci-id argument to python_ble_proxy

python_ble_proxy/matter_ble_proxy/bleak_backend.py

@@ -18,6 +18,7 @@
 if TYPE_CHECKING:
     from collections.abc import Callable
 
+    from bleak.args.bluez import BlueZScannerArgs
     from bleak.backends.device import BLEDevice
     from bleak.backends.scanner import AdvertisementData as BleakAdvertisementData
 
@@ -34,8 +35,9 @@ class BleakScanSource(BleScanSource):
     handshake) — typically tens to hundreds of milliseconds.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, hci_device: int | None = None) -> None:
         """Initialize."""
+        self._hci_device = hci_device
         self._scanner: BleakScanner | None = None
         self._callback: Callable[[AdvertisementData], None] | None = None
         # Cache the most recent BLEDevice per address so BleakDeviceResolver
@@ -53,7 +55,15 @@ async def start(self, callback: Callable[[AdvertisementData], None]) -> None:
         self._callback = callback
         if self._scanner is not None:
             return
-        self._scanner = BleakScanner(detection_callback=self._on_detection)
+
+        bluez_args: BlueZScannerArgs = {}
+        if self._hci_device is not None:
+            bluez_args["adapter"] = f"hci{self._hci_device}"
+
+        self._scanner = BleakScanner(
+            detection_callback=self._on_detection,
+            bluez=bluez_args,
+        )
         await self._scanner.start()
 
     async def stop(self) -> None:

python_ble_proxy/matter_ble_proxy/cli.py

@@ -37,6 +37,12 @@ def _parse_args(argv: list[str] | None) -> argparse.Namespace:
         default="ws://localhost:5580/ble",
         help="matter-server BLE proxy URL (default: %(default)s)",
     )
+    parser.add_argument(
+        "--hci-id",
+        default=None,
+        type=int,
+        help="Bluetooth adapter HCI ID (e.g., 0 for hci0)",
+    )
     parser.add_argument(
         "--log-level",
         default="INFO",
@@ -46,8 +52,8 @@ def _parse_args(argv: list[str] | None) -> argparse.Namespace:
     return parser.parse_args(argv)
 
 
-async def _run(server_url: str) -> int:
-    scan_source = BleakScanSource()
+async def _run(server_url: str, hci_id: int | None) -> int:
+    scan_source = BleakScanSource(hci_id)
     device_resolver = BleakDeviceResolver(scan_source)
     proxy = MatterBleProxy(server_url, scan_source, device_resolver)
 
@@ -96,7 +102,7 @@ def main(argv: list[str] | None = None) -> int:
         datefmt="%H:%M:%S",
     )
     try:
-        return asyncio.run(_run(args.server))
+        return asyncio.run(_run(args.server, args.hci_id))
     except KeyboardInterrupt:
         return 0