Fix device discovery when Bluetooth adapter is in passive scanning mode (#397)

bdraco · web-flow · commit 60c27250253f · 2025-09-10T12:37:05.000-05:00
diff --git a/switchbot/__init__.py b/switchbot/__init__.py
@@ -36,6 +36,7 @@
     SwitchbotDevice,
     SwitchbotEncryptedDevice,
     SwitchbotOperationError,
+    fetch_cloud_devices,
 )
 from .devices.evaporative_humidifier import SwitchbotEvaporativeHumidifier
 from .devices.fan import SwitchbotFan
@@ -104,6 +105,7 @@
     "SwitchbotVacuum",
     "close_stale_connections",
     "close_stale_connections_by_address",
+    "fetch_cloud_devices",
     "get_device",
     "parse_advertisement_data",
 ]
diff --git a/switchbot/adv_parser.py b/switchbot/adv_parser.py
@@ -45,6 +45,7 @@
 from .adv_parsers.vacuum import process_vacuum, process_vacuum_k
 from .const import SwitchbotModel
 from .models import SwitchBotAdvertisement
+from .utils import format_mac_upper
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -54,6 +55,8 @@
 )
 MFR_DATA_ORDER = (2409, 741, 89)
 
+_MODEL_TO_MAC_CACHE: dict[str, SwitchbotModel] = {}
+
 
 class SwitchbotSupportedType(TypedDict):
     """Supported type of Switchbot."""
@@ -383,6 +386,10 @@ def parse_advertisement_data(
     model: SwitchbotModel | None = None,
 ) -> SwitchBotAdvertisement | None:
     """Parse advertisement data."""
+    upper_mac = format_mac_upper(device.address)
+    if model is None and upper_mac in _MODEL_TO_MAC_CACHE:
+        model = _MODEL_TO_MAC_CACHE[upper_mac]
+
     service_data = advertisement_data.service_data
 
     _service_data = None
@@ -470,3 +477,8 @@ def _parse_data(
             )
 
     return data
+
+
+def populate_model_to_mac_cache(mac: str, model: SwitchbotModel) -> None:
+    """Populate the model to MAC address cache."""
+    _MODEL_TO_MAC_CACHE[mac] = model
diff --git a/switchbot/devices/device.py b/switchbot/devices/device.py
@@ -24,6 +24,7 @@
 )
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 
+from ..adv_parser import populate_model_to_mac_cache
 from ..api_config import SWITCHBOT_APP_API_BASE_URL, SWITCHBOT_APP_CLIENT_ID
 from ..const import (
     DEFAULT_RETRY_COUNT,
@@ -37,9 +38,43 @@
 from ..discovery import GetSwitchbotDevices
 from ..helpers import create_background_task
 from ..models import SwitchBotAdvertisement
+from ..utils import format_mac_upper
 
 _LOGGER = logging.getLogger(__name__)
 
+
+def _extract_region(userinfo: dict[str, Any]) -> str:
+    """Extract region from user info, defaulting to 'us'."""
+    if "botRegion" in userinfo and userinfo["botRegion"] != "":
+        return userinfo["botRegion"]
+    return "us"
+
+
+# Mapping from API model names to SwitchbotModel enum values
+API_MODEL_TO_ENUM: dict[str, SwitchbotModel] = {
+    "WoHand": SwitchbotModel.BOT,
+    "WoCurtain": SwitchbotModel.CURTAIN,
+    "WoHumi": SwitchbotModel.HUMIDIFIER,
+    "WoPlug": SwitchbotModel.PLUG_MINI,
+    "WoPlugUS": SwitchbotModel.PLUG_MINI,
+    "WoContact": SwitchbotModel.CONTACT_SENSOR,
+    "WoStrip": SwitchbotModel.LIGHT_STRIP,
+    "WoSensorTH": SwitchbotModel.METER,
+    "WoMeter": SwitchbotModel.METER,
+    "WoMeterPlus": SwitchbotModel.METER_PRO,
+    "WoPresence": SwitchbotModel.MOTION_SENSOR,
+    "WoBulb": SwitchbotModel.COLOR_BULB,
+    "WoCeiling": SwitchbotModel.CEILING_LIGHT,
+    "WoLock": SwitchbotModel.LOCK,
+    "WoBlindTilt": SwitchbotModel.BLIND_TILT,
+    "WoIOSensor": SwitchbotModel.IO_METER,  # Outdoor Meter
+    "WoButton": SwitchbotModel.REMOTE,  # Remote button
+    "WoLinkMini": SwitchbotModel.HUBMINI_MATTER,  # Hub Mini
+    "W1083002": SwitchbotModel.RELAY_SWITCH_1,  # Relay Switch 1
+    "W1079000": SwitchbotModel.METER_PRO,  # Meter Pro (another variant)
+    "W1102001": SwitchbotModel.STRIP_LIGHT_3,  # RGBWW Strip Light 3
+}
+
 REQ_HEADER = "570f"
 
 
@@ -164,6 +199,113 @@ def __init__(
         self._last_full_update: float = -PASSIVE_POLL_INTERVAL
         self._timed_disconnect_task: asyncio.Task[None] | None = None
 
+    @classmethod
+    async def _async_get_user_info(
+        cls,
+        session: aiohttp.ClientSession,
+        auth_headers: dict[str, str],
+    ) -> dict[str, Any]:
+        try:
+            return await cls.api_request(
+                session, "account", "account/api/v1/user/userinfo", {}, auth_headers
+            )
+        except Exception as err:
+            raise SwitchbotAccountConnectionError(
+                f"Failed to retrieve SwitchBot Account user details: {err}"
+            ) from err
+
+    @classmethod
+    async def _get_auth_result(
+        cls,
+        session: aiohttp.ClientSession,
+        username: str,
+        password: str,
+    ) -> dict[str, Any]:
+        """Authenticate with SwitchBot API."""
+        try:
+            return await cls.api_request(
+                session,
+                "account",
+                "account/api/v1/user/login",
+                {
+                    "clientId": SWITCHBOT_APP_CLIENT_ID,
+                    "username": username,
+                    "password": password,
+                    "grantType": "password",
+                    "verifyCode": "",
+                },
+            )
+        except Exception as err:
+            raise SwitchbotAuthenticationError(f"Authentication failed: {err}") from err
+
+    @classmethod
+    async def get_devices(
+        cls,
+        session: aiohttp.ClientSession,
+        username: str,
+        password: str,
+    ) -> dict[str, SwitchbotModel]:
+        """Get devices from SwitchBot API and return formatted MAC to model mapping."""
+        try:
+            auth_result = await cls._get_auth_result(session, username, password)
+            auth_headers = {"authorization": auth_result["access_token"]}
+        except Exception as err:
+            raise SwitchbotAuthenticationError(f"Authentication failed: {err}") from err
+
+        userinfo = await cls._async_get_user_info(session, auth_headers)
+        region = _extract_region(userinfo)
+
+        try:
+            device_info = await cls.api_request(
+                session,
+                f"wonderlabs.{region}",
+                "wonder/device/v3/getdevice",
+                {
+                    "required_type": "All",
+                },
+                auth_headers,
+            )
+        except Exception as err:
+            raise SwitchbotAccountConnectionError(
+                f"Failed to retrieve devices from SwitchBot Account: {err}"
+            ) from err
+
+        items: list[dict[str, Any]] = device_info["Items"]
+        mac_to_model: dict[str, SwitchbotModel] = {}
+
+        for item in items:
+            if "device_mac" not in item:
+                continue
+
+            if (
+                "device_detail" not in item
+                or "device_type" not in item["device_detail"]
+            ):
+                continue
+
+            mac = item["device_mac"]
+            model_name = item["device_detail"]["device_type"]
+
+            # Format MAC to uppercase with colons
+            formatted_mac = format_mac_upper(mac)
+
+            # Map API model name to SwitchbotModel enum if possible
+            if model_name in API_MODEL_TO_ENUM:
+                model = API_MODEL_TO_ENUM[model_name]
+                mac_to_model[formatted_mac] = model
+                # Populate the cache
+                populate_model_to_mac_cache(formatted_mac, model)
+            else:
+                # Log the full item payload for unknown models
+                _LOGGER.debug(
+                    "Unknown model %s for device %s, full item: %s",
+                    model_name,
+                    formatted_mac,
+                    item,
+                )
+
+        return mac_to_model
+
     @classmethod
     async def api_request(
         cls,
@@ -809,34 +951,13 @@ async def async_retrieve_encryption_key(
         device_mac = device_mac.replace(":", "").replace("-", "").upper()
 
         try:
-            auth_result = await cls.api_request(
-                session,
-                "account",
-                "account/api/v1/user/login",
-                {
-                    "clientId": SWITCHBOT_APP_CLIENT_ID,
-                    "username": username,
-                    "password": password,
-                    "grantType": "password",
-                    "verifyCode": "",
-                },
-            )
+            auth_result = await cls._get_auth_result(session, username, password)
             auth_headers = {"authorization": auth_result["access_token"]}
         except Exception as err:
             raise SwitchbotAuthenticationError(f"Authentication failed: {err}") from err
 
-        try:
-            userinfo = await cls.api_request(
-                session, "account", "account/api/v1/user/userinfo", {}, auth_headers
-            )
-            if "botRegion" in userinfo and userinfo["botRegion"] != "":
-                region = userinfo["botRegion"]
-            else:
-                region = "us"
-        except Exception as err:
-            raise SwitchbotAccountConnectionError(
-                f"Failed to retrieve SwitchBot Account user details: {err}"
-            ) from err
+        userinfo = await cls._async_get_user_info(session, auth_headers)
+        region = _extract_region(userinfo)
 
         try:
             device_info = await cls.api_request(
@@ -1023,3 +1144,13 @@ def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> No
         )
         if current_state != new_state:
             create_background_task(self.update())
+
+
+async def fetch_cloud_devices(
+    session: aiohttp.ClientSession,
+    username: str,
+    password: str,
+) -> dict[str, SwitchbotModel]:
+    """Fetch devices from SwitchBot API and return MAC to model mapping."""
+    # Get devices from the API (which also populates the cache)
+    return await SwitchbotBaseDevice.get_devices(session, username, password)
diff --git a/switchbot/utils.py b/switchbot/utils.py
@@ -0,0 +1,24 @@
+"""Utility functions for switchbot."""
+
+from functools import lru_cache
+
+
+@lru_cache(maxsize=512)
+def format_mac_upper(mac: str) -> str:
+    """Format the mac address string to uppercase with colons."""
+    to_test = mac
+
+    if len(to_test) == 17 and to_test.count(":") == 5:
+        return to_test.upper()
+
+    if len(to_test) == 17 and to_test.count("-") == 5:
+        to_test = to_test.replace("-", "")
+    elif len(to_test) == 14 and to_test.count(".") == 2:
+        to_test = to_test.replace(".", "")
+
+    if len(to_test) == 12:
+        # no : included
+        return ":".join(to_test.upper()[i : i + 2] for i in range(0, 12, 2))
+
+    # Not sure how formatted, return original
+    return mac.upper()
diff --git a/tests/test_adv_parser.py b/tests/test_adv_parser.py
@@ -8,7 +8,11 @@
 from bleak.backends.scanner import AdvertisementData
 
 from switchbot import HumidifierMode, SwitchbotModel
-from switchbot.adv_parser import parse_advertisement_data
+from switchbot.adv_parser import (
+    _MODEL_TO_MAC_CACHE,
+    parse_advertisement_data,
+    populate_model_to_mac_cache,
+)
 from switchbot.const.lock import LockStatus
 from switchbot.models import SwitchBotAdvertisement
 
@@ -3790,3 +3794,73 @@ def test_adv_with_empty_data(test_case: AdvTestCase) -> None:
         rssi=-97,
         active=True,
     )
+
+
+def test_parse_advertisement_with_mac_cache() -> None:
+    """Test that populating the MAC cache helps identify unknown passive devices."""
+    # Clear the cache to ensure clean test
+    _MODEL_TO_MAC_CACHE.clear()
+
+    # Create a passive lock device with manufacturer data only (no service data)
+    # This would normally not be identifiable without the cache
+    mac_address = "C1:64:B8:7D:06:05"
+    ble_device = generate_ble_device(mac_address, "WoLock")
+
+    # Lock passive advertisement with manufacturer data
+    # This is real data from a WoLock device in passive mode
+    adv_data = generate_advertisement_data(
+        manufacturer_data={2409: b"\xc1d\xb8}\x06\x05\x00\x00\x00\x00\x00"},
+        service_data={},
+        rssi=-70,
+    )
+
+    # First attempt: Without cache, parser cannot identify the device model
+    result_without_cache = parse_advertisement_data(ble_device, adv_data)
+    assert result_without_cache is None, "Should not decode without model hint"
+
+    # Now populate the cache with the device's MAC and model
+    populate_model_to_mac_cache(mac_address, SwitchbotModel.LOCK)
+
+    # Second attempt: With cache, parser can now identify and decode the device
+    result_with_cache = parse_advertisement_data(ble_device, adv_data)
+    assert result_with_cache is not None, "Should decode with MAC cache"
+    assert result_with_cache.data["modelName"] == SwitchbotModel.LOCK
+    assert result_with_cache.data["modelFriendlyName"] == "Lock"
+    assert result_with_cache.active is False  # Passive advertisement
+
+    # Clean up
+    _MODEL_TO_MAC_CACHE.clear()
+
+
+def test_parse_advertisement_with_mac_cache_curtain() -> None:
+    """Test MAC cache with a passive curtain device."""
+    # Clear the cache
+    _MODEL_TO_MAC_CACHE.clear()
+
+    # Create a passive curtain device
+    mac_address = "CC:F4:C4:F9:AC:6C"
+    ble_device = generate_ble_device(mac_address, None)
+
+    # Curtain passive advertisement with only manufacturer data
+    adv_data = generate_advertisement_data(
+        manufacturer_data={2409: b"\xccOLG\x00c\x00\x00\x11\x00\x00"},
+        service_data={},
+        rssi=-85,
+    )
+
+    # Without cache, cannot identify
+    result_without_cache = parse_advertisement_data(ble_device, adv_data)
+    assert result_without_cache is None
+
+    # Populate cache
+    populate_model_to_mac_cache(mac_address, SwitchbotModel.CURTAIN)
+
+    # With cache, can identify and parse
+    result_with_cache = parse_advertisement_data(ble_device, adv_data)
+    assert result_with_cache is not None
+    assert result_with_cache.data["modelName"] == SwitchbotModel.CURTAIN
+    assert result_with_cache.data["modelFriendlyName"] == "Curtain"
+    assert result_with_cache.active is False
+
+    # Clean up
+    _MODEL_TO_MAC_CACHE.clear()
diff --git a/tests/test_device.py b/tests/test_device.py
diff --git a/tests/test_utils.py b/tests/test_utils.py
