Add support for polling when passive scanning is used (#161)

Author: bdraco

switchbot/adv_parser.py

@@ -177,7 +177,9 @@ def parse_advertisement_data(
     if not data:
         return None
 
-    return SwitchBotAdvertisement(device.address, data, device, advertisement_data.rssi)
+    return SwitchBotAdvertisement(
+        device.address, data, device, advertisement_data.rssi, bool(_service_data)
+    )
 
 
 @lru_cache(maxsize=128)

switchbot/adv_parsers/curtain.py

@@ -25,4 +25,4 @@ def process_wocurtain(
         "position": (100 - _position) if reverse else _position,
         "lightLevel": _light_level,
         "deviceChain": _device_chain,
-    }
+    }

switchbot/adv_parsers/meter.py

@@ -25,7 +25,9 @@ def process_wosensorth(data: bytes | None, mfr_data: bytes | None) -> dict[str,
     _temp_f = (_temp_f * 10) / 10
 
     _wosensorth_data = {
+        # Data should be flat, but we keep the original structure for now
         "temp": {"c": _temp_c, "f": _temp_f},
+        "temperature": _temp_c,
         "fahrenheit": bool(temp_data[2] & 0b10000000),
         "humidity": temp_data[2] & 0b01111111,
         "battery": battery,

switchbot/devices/base_light.py

@@ -8,6 +8,7 @@
 
 _LOGGER = logging.getLogger(__name__)
 import asyncio
+import time
 
 from ..models import SwitchBotAdvertisement
 
@@ -81,6 +82,14 @@ async def set_color_temp(self, brightness: int, color_temp: int) -> bool:
     async def set_rgb(self, brightness: int, r: int, g: int, b: int) -> bool:
         """Set rgb."""
 
+    def poll_needed(self, last_poll_time: float | None) -> bool:
+        """Return if poll is needed."""
+        return False
+
+    async def update(self) -> None:
+        """Update device data."""
+        self._last_full_update = time.monotonic()
+
 
 class SwitchbotSequenceBaseLight(SwitchbotBaseLight):
     """Representation of a Switchbot light."""

switchbot/devices/bot.py

@@ -8,6 +8,7 @@
     DEVICE_SET_EXTENDED_KEY,
     DEVICE_SET_MODE_KEY,
     SwitchbotDeviceOverrideStateDuringConnection,
+    update_after_operation,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -30,10 +31,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
         self._inverse: bool = kwargs.pop("inverse_mode", False)
 
-    async def update(self, interface: int | None = None) -> None:
-        """Update mode, battery percent and state of device."""
-        await self.get_device_data(retry=self._retry_count, interface=interface)
-
+    @update_after_operation
     async def turn_on(self) -> bool:
         """Turn device on."""
         result = await self._send_command(ON_KEY)
@@ -48,6 +46,7 @@ async def turn_on(self) -> bool:
         self._fire_callbacks()
         return ret
 
+    @update_after_operation
     async def turn_off(self) -> bool:
         """Turn device off."""
         result = await self._send_command(OFF_KEY)
@@ -62,21 +61,25 @@ async def turn_off(self) -> bool:
         self._fire_callbacks()
         return ret
 
+    @update_after_operation
     async def hand_up(self) -> bool:
         """Raise device arm."""
         result = await self._send_command(UP_KEY)
         return self._check_command_result(result, 0, {1, 5})
 
+    @update_after_operation
     async def hand_down(self) -> bool:
         """Lower device arm."""
         result = await self._send_command(DOWN_KEY)
         return self._check_command_result(result, 0, {1, 5})
 
+    @update_after_operation
     async def press(self) -> bool:
         """Press command to device."""
         result = await self._send_command(PRESS_KEY)
         return self._check_command_result(result, 0, {1, 5})
 
+    @update_after_operation
     async def set_switch_mode(
         self, switch_mode: bool = False, strength: int = 100, inverse: bool = False
     ) -> bool:
@@ -86,6 +89,7 @@ async def set_switch_mode(
         result = await self._send_command(DEVICE_SET_MODE_KEY + strength_key + mode_key)
         return self._check_command_result(result, 0, {1})
 
+    @update_after_operation
     async def set_long_press(self, duration: int = 0) -> bool:
         """Set bot long press duration."""
         duration_key = f"{duration:0{2}x}"  # to hex with padding to double digit

switchbot/devices/bulb.py

@@ -34,6 +34,7 @@ async def update(self) -> None:
         """Update state of device."""
         result = await self._send_command(BULB_REQUEST)
         self._update_state(result)
+        await super().update()
 
     async def turn_on(self) -> bool:
         """Turn device on."""

switchbot/devices/ceiling_light.py

@@ -27,9 +27,6 @@ def color_modes(self) -> set[ColorMode]:
         """Return the supported color modes."""
         return {ColorMode.COLOR_TEMP}
 
-    async def update(self) -> None:
-        """Update state of device."""
-
     async def turn_on(self) -> bool:
         """Turn device on."""
         result = await self._send_command(CEILING_LIGHT_ON_KEY)

switchbot/devices/curtain.py

@@ -4,7 +4,7 @@
 import logging
 from typing import Any
 
-from .device import REQ_HEADER, SwitchbotDevice
+from .device import REQ_HEADER, SwitchbotDevice, update_after_operation
 
 # Curtain keys
 CURTAIN_COMMAND = "4501"
@@ -62,18 +62,22 @@ async def _send_multiple_commands(self, keys: list[str]) -> bool:
             final_result |= self._check_command_result(result, 0, {1})
         return final_result
 
+    @update_after_operation
     async def open(self) -> bool:
         """Send open command."""
         return await self._send_multiple_commands(OPEN_KEYS)
 
+    @update_after_operation
     async def close(self) -> bool:
         """Send close command."""
         return await self._send_multiple_commands(CLOSE_KEYS)
 
+    @update_after_operation
     async def stop(self) -> bool:
         """Send stop command to device."""
         return await self._send_multiple_commands(STOP_KEYS)
 
+    @update_after_operation
     async def set_position(self, position: int) -> bool:
         """Send position command (0-100) to device."""
         position = (100 - position) if self._reverse else position
@@ -82,10 +86,6 @@ async def set_position(self, position: int) -> bool:
             [key + hex_position for key in POSITION_KEYS]
         )
 
-    async def update(self, interface: int | None = None) -> None:
-        """Update position, battery percent and light level of device."""
-        await self.get_device_data(retry=self._retry_count, interface=interface)
-
     def get_position(self) -> Any:
         """Return cached position (0-100) of Curtain."""
         # To get actual position call update() first.
@@ -108,6 +108,7 @@ async def get_basic_info(self) -> dict[str, Any] | None:
             "light": bool(_data[4] & 0b00100000),
             "fault": bool(_data[4] & 0b00001000),
             "solarPanel": bool(_data[5] & 0b00001000),
+            "calibration": bool(_data[5] & 0b00000100),
             "calibrated": bool(_data[5] & 0b00000100),
             "inMotion": bool(_data[5] & 0b01000011),
             "position": (100 - _position) if self._reverse else _position,

switchbot/devices/device.py

@@ -4,8 +4,10 @@
 import asyncio
 import binascii
 import logging
+import time
+from dataclasses import replace
 from enum import Enum
-from typing import Any, Callable
+from typing import Any, Callable, TypeVar, cast
 from uuid import UUID
 
 import async_timeout
@@ -53,6 +55,13 @@ class ColorMode(Enum):
     EFFECT = 3
 
 
+# If the scanner is in passive mode, we
+# need to poll the device to get the
+# battery and a few rarely updating
+# values.
+PASSIVE_POLL_INTERVAL = 60 * 60 * 24
+
+
 class CharacteristicMissingError(Exception):
     """Raised when a characteristic is missing."""
 
@@ -76,6 +85,32 @@ def _sb_uuid(comms_type: str = "service") -> UUID | str:
 WRITE_CHAR_UUID = _sb_uuid(comms_type="tx")
 
 
+WrapFuncType = TypeVar("WrapFuncType", bound=Callable[..., Any])
+
+
+def update_after_operation(func: WrapFuncType) -> WrapFuncType:
+    """Define a wrapper to update after an operation."""
+
+    async def _async_update_after_operation_wrap(
+        self: SwitchbotBaseDevice, *args: Any, **kwargs: Any
+    ) -> None:
+        ret = await func(self, *args, **kwargs)
+        await self.update()
+        self._fire_callbacks()
+        return ret
+
+    return cast(WrapFuncType, _async_update_after_operation_wrap)
+
+
+def _merge_data(old_data: dict[str, Any], new_data: dict[str, Any]) -> dict[str, Any]:
+    """Merge data but only add None keys if they are missing."""
+    merged = old_data.copy()
+    for key, value in new_data.items():
+        if value is not None or key not in old_data:
+            merged[key] = value
+    return merged
+
+
 class SwitchbotBaseDevice:
     """Base Representation of a Switchbot Device."""
 
@@ -109,6 +144,7 @@ def __init__(
         self.loop = asyncio.get_event_loop()
         self._callbacks: list[Callable[[], None]] = []
         self._notify_future: asyncio.Future[bytearray] | None = None
+        self._last_full_update: float = -PASSIVE_POLL_INTERVAL
 
     def advertisement_changed(self, advertisement: SwitchBotAdvertisement) -> bool:
         """Check if the advertisement has changed."""
@@ -190,6 +226,18 @@ def name(self) -> str:
         """Return device name."""
         return f"{self._device.name} ({self._device.address})"
 
+    @property
+    def data(self) -> dict[str, Any]:
+        """Return device data."""
+        if self._sb_adv_data:
+            return self._sb_adv_data.data
+        return {}
+
+    @property
+    def parsed_data(self) -> dict[str, Any]:
+        """Return parsed device data."""
+        return self.data.get("data") or {}
+
     @property
     def rssi(self) -> int:
         """Return RSSI of device."""
@@ -392,6 +440,7 @@ def _override_state(self, state: dict[str, Any]) -> None:
         if self._override_adv_data is None:
             self._override_adv_data = {}
         self._override_adv_data.update(state)
+        self._update_parsed_data(state)
 
     def _get_adv_value(self, key: str) -> Any:
         """Return value from advertisement data."""
@@ -466,8 +515,20 @@ def _unsub() -> None:
 
         return _unsub
 
-    async def update(self) -> None:
-        """Update state of device."""
+    async def update(self, interface: int | None = None) -> None:
+        """Update position, battery percent and light level of device."""
+        if info := await self.get_basic_info():
+            self._last_full_update = time.monotonic()
+            self._update_parsed_data(info)
+
+    async def get_basic_info(self) -> dict[str, Any] | None:
+        """Get device basic settings."""
+        if not (_data := await self._get_basic_info()):
+            return None
+        return {
+            "battery": _data[1],
+            "firmware": _data[2] / 10.0,
+        }
 
     def _check_command_result(
         self, result: bytes | None, index: int, values: set[int]
@@ -480,21 +541,54 @@ def _check_command_result(
             )
         return result[index] in values
 
+    def _update_parsed_data(self, new_data: dict[str, Any]) -> None:
+        """Update data."""
+        if not self._sb_adv_data:
+            _LOGGER.exception("No advertisement data to update")
+            return
+        self._set_parsed_data(
+            self._sb_adv_data,
+            _merge_data(self._sb_adv_data.data.get("data") or {}, new_data),
+        )
+
+    def _set_parsed_data(
+        self, advertisement: SwitchBotAdvertisement, data: dict[str, Any]
+    ) -> None:
+        """Set data."""
+        self._sb_adv_data = replace(
+            advertisement, data=self._sb_adv_data.data | {"data": data}
+        )
+
     def _set_advertisement_data(self, advertisement: SwitchBotAdvertisement) -> None:
         """Set advertisement data."""
-        if (
-            advertisement.data.get("data")
-            or not self._sb_adv_data
-            or not self._sb_adv_data.data.get("data")
-        ):
+        new_data = advertisement.data.get("data") or {}
+        if advertisement.active:
+            # If we are getting active data, we can assume we are
+            # getting active scans and we do not need to poll
+            self._last_full_update = time.monotonic()
+        if not self._sb_adv_data:
             self._sb_adv_data = advertisement
+        elif new_data:
+            self._update_parsed_data(new_data)
         self._override_adv_data = None
 
     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 poll_needed(self, seconds_since_last_poll: float | None) -> bool:
+        """Return if device needs polling."""
+        if (
+            seconds_since_last_poll is not None
+            and seconds_since_last_poll < PASSIVE_POLL_INTERVAL
+        ):
+            return False
+        time_since_last_full_update = time.monotonic() - self._last_full_update
+        if time_since_last_full_update < PASSIVE_POLL_INTERVAL:
+            return False
+        return True
+
 
 class SwitchbotDevice(SwitchbotBaseDevice):
     """Base Representation of a Switchbot Device.

switchbot/devices/humidifier.py

@@ -1,6 +1,8 @@
 """Library to handle connection with Switchbot."""
 from __future__ import annotations
 
+import time
+
 from .device import REQ_HEADER, SwitchbotDevice
 
 HUMIDIFIER_COMMAND_HEADER = "4381"
@@ -28,7 +30,8 @@ class SwitchbotHumidifier(SwitchbotDevice):
 
     async def update(self, interface: int | None = None) -> None:
         """Update state of device."""
-        await self.get_device_data(retry=self._retry_count, interface=interface)
+        # No battery here
+        self._last_full_update = time.monotonic()
 
     def _generate_command(
         self, on: bool | None = None, level: int | None = None
@@ -96,3 +99,7 @@ def get_target_humidity(self) -> int | None:
         if self.is_auto():
             return None
         return MANUAL_BUTTON_PRESSES_TO_LEVEL.get(level, level)
+
+    def poll_needed(self, last_poll_time: float | None) -> bool:
+        """Return if device needs polling."""
+        return False