feat: update scanner mode from callback (#137)

Author: bdraco

src/bleak_esphome/backend/scanner.py

@@ -2,13 +2,19 @@
 
 from __future__ import annotations
 
-from aioesphomeapi import BluetoothLEAdvertisement, BluetoothLERawAdvertisementsResponse
+from aioesphomeapi import (
+    BluetoothLEAdvertisement,
+    BluetoothLERawAdvertisementsResponse,
+    BluetoothScannerMode,
+    BluetoothScannerStateResponse,
+)
 from bluetooth_data_tools import (
     int_to_bluetooth_address,
 )
 from bluetooth_data_tools import (
     monotonic_time_coarse as MONOTONIC_TIME,
 )
+from habluetooth import BluetoothScanningMode
 from habluetooth.base_scanner import BaseHaRemoteScanner
 
 
@@ -17,6 +23,18 @@ class ESPHomeScanner(BaseHaRemoteScanner):
 
     __slots__ = ()
 
+    def async_update_scanner_state(self, state: BluetoothScannerStateResponse) -> None:
+        """Update the scanner state."""
+        if state.mode == BluetoothScannerMode.ACTIVE:
+            self.current_mode = BluetoothScanningMode.ACTIVE  # type: ignore[misc]
+            self.requested_mode = BluetoothScanningMode.ACTIVE  # type: ignore[misc]
+        elif state.mode == BluetoothScannerMode.PASSIVE:
+            self.current_mode = BluetoothScanningMode.PASSIVE
+            self.requested_mode = BluetoothScanningMode.PASSIVE
+        else:
+            self.current_mode = None
+            self.requested_mode = None
+
     def async_on_advertisement(self, adv: BluetoothLEAdvertisement) -> None:
         """Call the registered callback."""
         # The mac address is a uint64, but we need a string

src/bleak_esphome/connect.py

@@ -95,6 +95,12 @@ def connect_scanner(
             bluetooth_device.async_update_ble_connection_limits
         )
 
+    if feature_flags & BluetoothProxyFeature.FEATURE_STATE_AND_MODE:
+        _LOGGER.debug(
+            "%s [%s]: Bluetooth scanner state and mode support available", name, source
+        )
+        cli.subscribe_bluetooth_scanner_state(scanner.async_update_scanner_state)
+
     if feature_flags & BluetoothProxyFeature.RAW_ADVERTISEMENTS:
         cli.subscribe_bluetooth_le_raw_advertisements(
             scanner.async_on_raw_advertisements

tests/backend/test_scanner.py

@@ -1,9 +1,18 @@
+import pytest
 from aioesphomeapi import (
+    APIClient,
     BluetoothLERawAdvertisement,
     BluetoothLERawAdvertisementsResponse,
+    BluetoothScannerMode,
+    BluetoothScannerStateResponse,
 )
 from bluetooth_data_tools import int_to_bluetooth_address
-from habluetooth import BaseHaRemoteScanner, HaBluetoothConnector, get_manager
+from habluetooth import (
+    BaseHaRemoteScanner,
+    BluetoothScanningMode,
+    HaBluetoothConnector,
+    get_manager,
+)
 
 from bleak_esphome.backend.client import ESPHomeClientData
 from bleak_esphome.backend.scanner import ESPHomeScanner
@@ -12,15 +21,18 @@
 ESP_NAME = "proxy"
 
 
-def test_scanner() -> None:
+@pytest.fixture
+def scanner() -> ESPHomeScanner:
+    """Fixture to create an ESPHomeScanner instance."""
     connector = HaBluetoothConnector(ESPHomeClientData, ESP_MAC_ADDRESS, lambda: True)
-    scanner = ESPHomeScanner(ESP_MAC_ADDRESS, ESP_NAME, connector, True)
+    return ESPHomeScanner(ESP_MAC_ADDRESS, ESP_NAME, connector, True)
+
+
+def test_scanner(scanner: ESPHomeScanner) -> None:
     assert isinstance(scanner, BaseHaRemoteScanner)
 
 
-def test_scanner_async_on_advertisement() -> None:
-    connector = HaBluetoothConnector(ESPHomeClientData, ESP_MAC_ADDRESS, lambda: True)
-    scanner = ESPHomeScanner(ESP_MAC_ADDRESS, ESP_NAME, connector, True)
+def test_scanner_async_on_advertisement(scanner: ESPHomeScanner) -> None:
     adv = BluetoothLERawAdvertisementsResponse(
         advertisements=[
             BluetoothLERawAdvertisement(
@@ -45,3 +57,26 @@ def test_scanner_async_on_advertisement() -> None:
     assert manager.async_last_service_info(
         int_to_bluetooth_address(246965243285491), True
     )
+
+
+def test_scanner_async_update_scanner_state(
+    scanner: ESPHomeScanner, mock_client: APIClient
+) -> None:
+    mock_client.subscribe_bluetooth_scanner_state(scanner.async_update_scanner_state)
+    scanner.async_update_scanner_state(
+        BluetoothScannerStateResponse(
+            mode=BluetoothScannerMode.ACTIVE,
+        )
+    )
+    assert scanner.current_mode == BluetoothScanningMode.ACTIVE
+    assert scanner.requested_mode == BluetoothScanningMode.ACTIVE
+    scanner.async_update_scanner_state(
+        BluetoothScannerStateResponse(
+            mode=BluetoothScannerMode.PASSIVE,
+        )
+    )
+    assert scanner.current_mode == BluetoothScanningMode.PASSIVE
+    assert scanner.requested_mode == BluetoothScanningMode.PASSIVE
+    scanner.async_update_scanner_state(BluetoothScannerStateResponse(mode=None))
+    assert scanner.current_mode is None
+    assert scanner.requested_mode is None

tests/conftest.py

@@ -4,6 +4,7 @@
 from aioesphomeapi import (
     APIClient,
     APIVersion,
+    BluetoothProxyFeature,
     DeviceInfo,
     ReconnectLogic,
 )
@@ -32,6 +33,14 @@ def mock_device_info() -> DeviceInfo:
         legacy_bluetooth_proxy_version=0,
         # ESPHome mac addresses are UPPER case
         mac_address="11:22:33:44:55:AA",
+        bluetooth_mac_address="11:22:33:44:55:AC",
+        bluetooth_proxy_feature_flags=BluetoothProxyFeature.PASSIVE_SCAN
+        | BluetoothProxyFeature.ACTIVE_CONNECTIONS
+        | BluetoothProxyFeature.REMOTE_CACHING
+        | BluetoothProxyFeature.PAIRING
+        | BluetoothProxyFeature.CACHE_CLEARING
+        | BluetoothProxyFeature.RAW_ADVERTISEMENTS
+        | BluetoothProxyFeature.FEATURE_STATE_AND_MODE,
         esphome_version="1.0.0",
     )
 

tests/test_connect.py

@@ -0,0 +1,16 @@
+import pytest
+from aioesphomeapi import APIClient
+
+from bleak_esphome.connect import connect_scanner
+
+
+@pytest.mark.asyncio
+async def test_connect(mock_client: APIClient) -> None:
+    """Test the connect_scanner function."""
+    device_info = await mock_client.device_info()
+    scanner = connect_scanner(mock_client, device_info, available=True)
+    assert scanner is not None
+    assert (
+        scanner.device_info.bluetooth_mac_address == device_info.bluetooth_mac_address
+    )
+    assert scanner.client is mock_client