Add roller shade and hubmini matter support (#315)

* Add roller shade and hubmini matter support

* add unit test

* chore(pre-commit.ci): auto fixes

* add unit tests for passive data, stop and set position case

* chore(pre-commit.ci): auto fixes

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: zerzhang

switchbot/__init__.py

@@ -29,6 +29,7 @@
 from .devices.lock import SwitchbotLock
 from .devices.plug import SwitchbotPlugMini
 from .devices.relay_switch import SwitchbotRelaySwitch
+from .devices.roller_shade import SwitchbotRollerShade
 from .discovery import GetSwitchbotDevices
 from .models import SwitchBotAdvertisement
 
@@ -58,6 +59,7 @@
     "SwitchbotPlugMini",
     "SwitchbotPlugMini",
     "SwitchbotRelaySwitch",
+    "SwitchbotRollerShade",
     "SwitchbotSupportedType",
     "SwitchbotSupportedType",
     "close_stale_connections",

switchbot/adv_parser.py

@@ -17,6 +17,7 @@
 from .adv_parsers.contact import process_wocontact
 from .adv_parsers.curtain import process_wocurtain
 from .adv_parsers.hub2 import process_wohub2
+from .adv_parsers.hubmini_matter import process_hubmini_matter
 from .adv_parsers.humidifier import process_evaporative_humidifier, process_wohumidifier
 from .adv_parsers.keypad import process_wokeypad
 from .adv_parsers.leak import process_leak
@@ -30,6 +31,7 @@
     process_worelay_switch_1pm,
 )
 from .adv_parsers.remote import process_woremote
+from .adv_parsers.roller_shade import process_worollershade
 from .const import SwitchbotModel
 from .models import SwitchBotAdvertisement
 
@@ -216,6 +218,18 @@ class SwitchbotSupportedType(TypedDict):
         "func": process_woremote,
         "manufacturer_id": 89,
     },
+    ",": {
+        "modelName": SwitchbotModel.ROLLER_SHADE,
+        "modelFriendlyName": "Roller Shade",
+        "func": process_worollershade,
+        "manufacturer_id": 2409,
+    },
+    "%": {
+        "modelName": SwitchbotModel.HUBMINI_MATTER,
+        "modelFriendlyName": "HubMini Matter",
+        "func": process_hubmini_matter,
+        "manufacturer_id": 2409,
+    },
 }
 
 _SWITCHBOT_MODEL_TO_CHAR = {

switchbot/adv_parsers/hubmini_matter.py

@@ -0,0 +1,37 @@
+"""Hubmini matter parser."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def process_hubmini_matter(
+    data: bytes | None, mfr_data: bytes | None
+) -> dict[str, Any]:
+    """Process Hubmini matter sensor manufacturer data."""
+    temp_data = None
+
+    if mfr_data:
+        temp_data = mfr_data[13:16]
+
+    if not temp_data:
+        return {}
+
+    _temp_sign = 1 if temp_data[1] & 0b10000000 else -1
+    _temp_c = _temp_sign * (
+        (temp_data[1] & 0b01111111) + ((temp_data[0] & 0b00001111) / 10)
+    )
+    _temp_f = (_temp_c * 9 / 5) + 32
+    _temp_f = (_temp_f * 10) / 10
+    humidity = temp_data[2] & 0b01111111
+
+    if _temp_c == 0 and humidity == 0:
+        return {}
+
+    paraser_data = {
+        "temp": {"c": _temp_c, "f": _temp_f},
+        "temperature": _temp_c,
+        "fahrenheit": bool(temp_data[2] & 0b10000000),
+        "humidity": humidity,
+    }
+    return paraser_data

switchbot/adv_parsers/roller_shade.py

@@ -0,0 +1,29 @@
+"""Library to handle connection with Switchbot."""
+
+from __future__ import annotations
+
+
+def process_worollershade(
+    data: bytes | None, mfr_data: bytes | None, reverse: bool = True
+) -> dict[str, bool | int]:
+    """Process woRollerShade services data."""
+    if mfr_data is None:
+        return {}
+
+    device_data = mfr_data[6:]
+
+    _position = max(min(device_data[2] & 0b01111111, 100), 0)
+    _calibrated = bool(device_data[2] & 0b10000000)
+    _in_motion = bool(device_data[1] & 0b00000110)
+    _light_level = (device_data[3] >> 4) & 0b00001111
+    _device_chain = device_data[3] & 0b00001111
+
+    return {
+        "calibration": _calibrated,
+        "battery": data[2] & 0b01111111 if data else None,
+        "inMotion": _in_motion,
+        "position": (100 - _position) if reverse else _position,
+        "lightLevel": _light_level,
+        "deviceChain": _device_chain,
+        "sequence_number": device_data[0],
+    }

switchbot/const/__init__.py

@@ -63,3 +63,5 @@ class SwitchbotModel(StrEnum):
     RELAY_SWITCH_1 = "Relay Switch 1"
     REMOTE = "WoRemote"
     EVAPORATIVE_HUMIDIFIER = "Evaporative Humidifier"
+    ROLLER_SHADE = "Roller Shade"
+    HUBMINI_MATTER = "HubMini Matter"

switchbot/devices/base_cover.py

@@ -10,6 +10,8 @@
 
 # Cover keys
 COVER_COMMAND = "4501"
+ROLLERSHADE_COMMAND = "4701"
+CONTROL_SOURCE = "00"
 
 # For second element of open and close arrs we should add two bytes i.e. ff00
 # First byte [ff] stands for speed (00 or ff - normal, 01 - slow) *

switchbot/devices/roller_shade.py

@@ -0,0 +1,129 @@
+"""Library to handle connection with Switchbot."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from ..models import SwitchBotAdvertisement
+from .base_cover import CONTROL_SOURCE, ROLLERSHADE_COMMAND, SwitchbotBaseCover
+from .device import REQ_HEADER, SwitchbotSequenceDevice, update_after_operation
+
+_LOGGER = logging.getLogger(__name__)
+
+
+OPEN_KEYS = [
+    f"{REQ_HEADER}{ROLLERSHADE_COMMAND}01{CONTROL_SOURCE}0100",
+    f"{REQ_HEADER}{ROLLERSHADE_COMMAND}05{CONTROL_SOURCE}0000",
+]
+CLOSE_KEYS = [
+    f"{REQ_HEADER}{ROLLERSHADE_COMMAND}01{CONTROL_SOURCE}0164",
+    f"{REQ_HEADER}{ROLLERSHADE_COMMAND}05{CONTROL_SOURCE}0064",
+]
+POSITION_KEYS = [
+    f"{REQ_HEADER}{ROLLERSHADE_COMMAND}01{CONTROL_SOURCE}01",
+    f"{REQ_HEADER}{ROLLERSHADE_COMMAND}05{CONTROL_SOURCE}",
+]  # +actual_position
+STOP_KEYS = [f"{REQ_HEADER}{ROLLERSHADE_COMMAND}00{CONTROL_SOURCE}01"]
+
+
+class SwitchbotRollerShade(SwitchbotBaseCover, SwitchbotSequenceDevice):
+    """Representation of a Switchbot Roller Shade."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Switchbot roller shade constructor."""
+        # The position of the roller shade is saved returned with 0 = open and 100 = closed.
+        # the definition of position is the same as in Home Assistant.
+
+        self._reverse: bool = kwargs.pop("reverse_mode", True)
+        super().__init__(self._reverse, *args, **kwargs)
+
+    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)
+
+    @update_after_operation
+    async def open(self, mode: int = 0) -> bool:
+        """Send open command. 0 - performance mode, 1 - unfelt mode."""
+        self._is_opening = True
+        self._is_closing = False
+        return await self._send_multiple_commands(OPEN_KEYS)
+
+    @update_after_operation
+    async def close(self, speed: int = 0) -> bool:
+        """Send close command. 0 - performance mode, 1 - unfelt mode."""
+        self._is_closing = True
+        self._is_opening = False
+        return await self._send_multiple_commands(CLOSE_KEYS)
+
+    @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, mode: int = 0) -> bool:
+        """Send position command (0-100) to device. 0 - performance mode, 1 - unfelt mode."""
+        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}",
+                f"{POSITION_KEYS[1]}{mode:02X}{position:02X}",
+            ]
+        )
+
+    def get_position(self) -> Any:
+        """Return cached position (0-100) of Curtain."""
+        # To get actual position call update() first.
+        return self._get_adv_value("position")
+
+    async def get_basic_info(self) -> dict[str, Any] | None:
+        """Get device basic settings."""
+        if not (_data := await self._get_basic_info()):
+            return None
+
+        _position = max(min(_data[5], 100), 0)
+        _direction_adjusted_position = (100 - _position) if self._reverse else _position
+        _previous_position = self._get_adv_value("position")
+        _in_motion = bool(_data[4] & 0b00000011)
+        self._update_motion_direction(
+            _in_motion, _previous_position, _direction_adjusted_position
+        )
+
+        return {
+            "battery": _data[1],
+            "firmware": _data[2] / 10.0,
+            "chainLength": _data[3],
+            "openDirection": (
+                "clockwise" if _data[4] & 0b10000000 == 128 else "anticlockwise"
+            ),
+            "fault": bool(_data[4] & 0b00010000),
+            "solarPanel": bool(_data[4] & 0b00001000),
+            "calibration": bool(_data[4] & 0b00000100),
+            "calibrated": bool(_data[4] & 0b00000100),
+            "inMotion": _in_motion,
+            "position": _direction_adjusted_position,
+            "timers": _data[6],
+        }
+
+    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

switchbot/discovery.py

@@ -111,7 +111,14 @@ async def get_tempsensors(self) -> dict[str, SwitchBotAdvertisement]:
         plus_meters = await self._get_devices_by_model("i")
         io_meters = await self._get_devices_by_model("w")
         hub2_meters = await self._get_devices_by_model("v")
-        return {**base_meters, **plus_meters, **io_meters, **hub2_meters}
+        hubmini_matter_meters = await self._get_devices_by_model("%")
+        return {
+            **base_meters,
+            **plus_meters,
+            **io_meters,
+            **hub2_meters,
+            **hubmini_matter_meters,
+        }
 
     async def get_contactsensors(self) -> dict[str, SwitchBotAdvertisement]:
         """Return all WoContact/Contact sensor devices with services data."""