Add support for is_opening and is_closing for curtains (#226)

Co-authored-by: J. Nick Koston <[email protected]>

Author: dcmeglio

switchbot/devices/bulb.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
 
 from .base_light import SwitchbotSequenceBaseLight
 from .device import REQ_HEADER, ColorMode

switchbot/devices/ceiling_light.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
 
 from .base_light import SwitchbotBaseLight
 from .device import REQ_HEADER, ColorMode

switchbot/devices/curtain.py

@@ -4,6 +4,8 @@
 import logging
 from typing import Any
 
+from switchbot.models import SwitchBotAdvertisement
+
 from .device import REQ_HEADER, SwitchbotDevice, update_after_operation
 
 # Curtain keys
@@ -54,6 +56,18 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         self._settings: dict[str, Any] = {}
         self.ext_info_sum: dict[str, Any] = {}
         self.ext_info_adv: dict[str, Any] = {}
+        self._is_opening: bool = False
+        self._is_closing: bool = False
+
+    def _set_parsed_data(
+        self, advertisement: SwitchBotAdvertisement, data: dict[str, Any]
+    ) -> None:
+        """Set data."""
+        in_motion = data["inMotion"]
+        previous_position = self._get_adv_value("position")
+        new_position = data["position"]
+        self._update_motion_direction(in_motion, previous_position, new_position)
+        super()._set_parsed_data(advertisement, data)
 
     async def _send_multiple_commands(self, keys: list[str]) -> bool:
         """Send multiple commands to device.
@@ -70,26 +84,32 @@ async def _send_multiple_commands(self, keys: list[str]) -> bool:
     @update_after_operation
     async def open(self, speed: int = 255) -> bool:
         """Send open command. Speed 255 - normal, 1 - slow"""
+        self._is_opening = True
+        self._is_closing = False
         return await self._send_multiple_commands(
             [OPEN_KEYS[0], f"{OPEN_KEYS[1]}{speed:02X}00"]
         )
 
     @update_after_operation
     async def close(self, speed: int = 255) -> bool:
         """Send close command. Speed 255 - normal, 1 - slow"""
+        self._is_closing = True
+        self._is_opening = False
         return await self._send_multiple_commands(
             [CLOSE_KEYS[0], f"{CLOSE_KEYS[1]}{speed:02X}64"]
         )
 
     @update_after_operation
     async def stop(self) -> bool:
         """Send stop command to device."""
+        self._is_opening = self._is_closing = False
         return await self._send_multiple_commands(STOP_KEYS)
 
     @update_after_operation
     async def set_position(self, position: int, speed: int = 255) -> bool:
         """Send position command (0-100) to device. Speed 255 - normal, 1 - slow"""
         position = (100 - position) if self._reverse else position
+        self._update_motion_direction(True, self._get_adv_value("position"), position)
         return await self._send_multiple_commands(
             [
                 f"{POSITION_KEYS[0]}{position:02X}",
@@ -108,6 +128,13 @@ async def get_basic_info(self) -> dict[str, Any] | None:
             return None
 
         _position = max(min(_data[6], 100), 0)
+        _direction_adjusted_position = (100 - _position) if self._reverse else _position
+        _previous_position = self._get_adv_value("position")
+        _in_motion = bool(_data[5] & 0b01000011)
+        self._update_motion_direction(
+            _in_motion, _previous_position, _direction_adjusted_position
+        )
+
         return {
             "battery": _data[1],
             "firmware": _data[2] / 10.0,
@@ -121,11 +148,25 @@ async def get_basic_info(self) -> dict[str, Any] | None:
             "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,
+            "inMotion": _in_motion,
+            "position": _direction_adjusted_position,
             "timers": _data[7],
         }
 
+    def _update_motion_direction(
+        self, in_motion: bool, previous_position: int | None, new_position: int
+    ) -> None:
+        """Update opening/closing status based on movement."""
+        if previous_position is None:
+            return
+        if in_motion is False:
+            self._is_closing = self._is_opening = False
+            return
+
+        if new_position != previous_position:
+            self._is_opening = new_position > previous_position
+            self._is_closing = new_position < previous_position
+
     async def get_extended_info_summary(self) -> dict[str, Any] | None:
         """Get basic info for all devices in chain."""
         _data = await self._send_command(key=CURTAIN_EXT_SUM_KEY)
@@ -210,3 +251,11 @@ def is_calibrated(self) -> Any:
         """Return True curtain is calibrated."""
         # To get actual light level call update() first.
         return self._get_adv_value("calibration")
+
+    def is_opening(self) -> bool:
+        """Return True if the curtain is opening."""
+        return self._is_opening
+
+    def is_closing(self) -> bool:
+        """Return True if the curtain is closing."""
+        return self._is_closing

switchbot/devices/light_strip.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
 
 from .base_light import SwitchbotSequenceBaseLight
 from .device import REQ_HEADER, ColorMode

tests/test_adv_parser.py

@@ -358,7 +358,9 @@ def test_parse_advertisement_data_curtain3():
     """Test parse_advertisement_data for curtain 3."""
     ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
     adv_data = generate_advertisement_data(
-        manufacturer_data={2409: b"\xaa\xbb\xcc\xdd\xee\xff\xf7\x07\x00\x11\x04\x00\x49"},
+        manufacturer_data={
+            2409: b"\xaa\xbb\xcc\xdd\xee\xff\xf7\x07\x00\x11\x04\x00\x49"
+        },
         service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"{\xc0\x49\x00\x11\x04"},
         rssi=-80,
     )
@@ -391,7 +393,9 @@ def test_parse_advertisement_data_curtain3_passive():
     """Test parse_advertisement_data for curtain passive."""
     ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
     adv_data = generate_advertisement_data(
-        manufacturer_data={2409: b"\xaa\xbb\xcc\xdd\xee\xff\xf7\x07\x00\x11\x04\x00\x49"},
+        manufacturer_data={
+            2409: b"\xaa\xbb\xcc\xdd\xee\xff\xf7\x07\x00\x11\x04\x00\x49"
+        },
         service_data={},
         rssi=-80,
     )
@@ -1364,6 +1368,7 @@ def test_parsing_lock_passive():
         active=False,
     )
 
+
 def test_parsing_lock_active_old_firmware():
     """Test parsing lock with active data. Old firmware."""
     ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")