Workaround stale advertisement data while connected to bots (#102)

Author: bdraco

switchbot/devices/bot.py

@@ -1,19 +1,28 @@
 """Library to handle connection with Switchbot."""
 from __future__ import annotations
 
+import logging
 from typing import Any
 
-from .device import DEVICE_SET_EXTENDED_KEY, DEVICE_SET_MODE_KEY, SwitchbotDevice
+from .device import (
+    DEVICE_SET_EXTENDED_KEY,
+    DEVICE_SET_MODE_KEY,
+    SwitchbotDeviceOverrideStateDuringConnection,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+BOT_COMMAND_HEADER = "5701"
 
 # Bot keys
-PRESS_KEY = "570100"
-ON_KEY = "570101"
-OFF_KEY = "570102"
-DOWN_KEY = "570103"
-UP_KEY = "570104"
+PRESS_KEY = f"{BOT_COMMAND_HEADER}00"
+ON_KEY = f"{BOT_COMMAND_HEADER}01"
+OFF_KEY = f"{BOT_COMMAND_HEADER}02"
+DOWN_KEY = f"{BOT_COMMAND_HEADER}03"
+UP_KEY = f"{BOT_COMMAND_HEADER}04"
 
 
-class Switchbot(SwitchbotDevice):
+class Switchbot(SwitchbotDeviceOverrideStateDuringConnection):
     """Representation of a Switchbot."""
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
@@ -28,12 +37,24 @@ async def update(self, interface: int | None = None) -> None:
     async def turn_on(self) -> bool:
         """Turn device on."""
         result = await self._send_command(ON_KEY)
-        return self._check_command_result(result, 0, {1, 5})
+        ret = self._check_command_result(result, 0, {1, 5})
+        self._override_adv_data = {"isOn": True}
+        _LOGGER.debug(
+            "%s: Turn on result: %s -> %s", self.name, result, self._override_adv_data
+        )
+        self._fire_callbacks()
+        return ret
 
     async def turn_off(self) -> bool:
         """Turn device off."""
         result = await self._send_command(OFF_KEY)
-        return self._check_command_result(result, 0, {1, 5})
+        ret = self._check_command_result(result, 0, {1, 5})
+        self._override_adv_data = {"isOn": False}
+        _LOGGER.debug(
+            "%s: Turn off result: %s -> %s", self.name, result, self._override_adv_data
+        )
+        self._fire_callbacks()
+        return ret
 
     async def hand_up(self) -> bool:
         """Raise device arm."""
@@ -79,12 +100,12 @@ async def get_basic_info(self) -> dict[str, Any] | None:
             "holdSeconds": _data[10],
         }
 
-    def switch_mode(self) -> Any:
+    def switch_mode(self) -> bool | None:
         """Return true or false from cache."""
         # To get actual position call update() first.
         return self._get_adv_value("switchMode")
 
-    def is_on(self) -> Any:
+    def is_on(self) -> bool | None:
         """Return switch state from cache."""
         # To get actual position call update() first.
         value = self._get_adv_value("isOn")

switchbot/devices/device.py

@@ -76,7 +76,7 @@ def _sb_uuid(comms_type: str = "service") -> UUID | str:
 WRITE_CHAR_UUID = _sb_uuid(comms_type="tx")
 
 
-class SwitchbotDevice:
+class SwitchbotBaseDevice:
     """Base Representation of a Switchbot Device."""
 
     def __init__(
@@ -349,6 +349,12 @@ def get_address(self) -> str:
     def _get_adv_value(self, key: str) -> Any:
         """Return value from advertisement data."""
         if self._override_adv_data and key in self._override_adv_data:
+            _LOGGER.debug(
+                "%s: Using override value for %s: %s",
+                self.name,
+                key,
+                self._override_adv_data[key],
+            )
             return self._override_adv_data[key]
         if not self._sb_adv_data:
             return None
@@ -362,9 +368,6 @@ def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> No
         """Update device data from advertisement."""
         # Only accept advertisements if the data is not missing
         # if we already have an advertisement with data
-        if advertisement.data.get("data") or not self._sb_adv_data.data.get("data"):
-            self._sb_adv_data = advertisement
-        self._override_adv_data = None
         if self._device and ble_device_has_changed(self._device, advertisement.device):
             self._cached_services = None
         self._device = advertisement.device
@@ -432,9 +435,51 @@ def _check_command_result(
             )
         return result[index] in values
 
+    def _set_advertisement_data(self, advertisement: SwitchBotAdvertisement) -> None:
+        """Set advertisement data."""
+        if advertisement.data.get("data") or not self._sb_adv_data.data.get("data"):
+            self._sb_adv_data = advertisement
+        self._override_adv_data = None
+
+
+class SwitchbotDevice(SwitchbotBaseDevice):
+    """Base Representation of a Switchbot Device.
+
+    This base class consumes the advertisement data during connection. If the device
+    sends stale advertisement data while connected, use
+    SwitchbotDeviceOverrideStateDuringConnection instead.
+    """
+
+    def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None:
+        """Update device data from advertisement."""
+        super().update_from_advertisement(advertisement)
+        self._set_advertisement_data(advertisement)
+
+
+class SwitchbotDeviceOverrideStateDuringConnection(SwitchbotBaseDevice):
+    """Base Representation of a Switchbot Device.
+
+    This base class ignores the advertisement data during connection and uses the
+    data from the device instead.
+    """
+
+    def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None:
+        super().update_from_advertisement(advertisement)
+        if self._client and self._client.is_connected:
+            # We do not consume the advertisement data if we are connected
+            # to the device. This is because the advertisement data is not
+            # updated when the device is connected for some devices.
+            _LOGGER.debug("%s: Ignore advertisement data during connection", self.name)
+            return
+        self._set_advertisement_data(advertisement)
+
 
 class SwitchbotSequenceDevice(SwitchbotDevice):
-    """A Switchbot sequence device."""
+    """A Switchbot sequence device.
+
+    This class must not use SwitchbotDeviceOverrideStateDuringConnection because
+    it needs to know when the sequence_number has changed.
+    """
 
     def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None:
         """Update device data from advertisement."""