feat: add roller mode for switch2pm

switchbot/adv_parsers/relay_switch.py

@@ -45,21 +45,29 @@ def process_garage_door_opener(
 
 def process_relay_switch_2pm(
     data: bytes | None, mfr_data: bytes | None
-) -> dict[int, dict[str, Any]]:
+) -> dict[int | str, dict[str, Any] | int]:
     """Process Relay Switch 2PM services data."""
-    if mfr_data is None:
+    # Highest index read below is mfr_data[14] (roller position), so guard
+    # against truncated advertisements that would otherwise raise IndexError.
+    if mfr_data is None or len(mfr_data) < 15:
         return {}
 
     return {
         1: {
             **process_relay_switch_common_data(data, mfr_data),
             "power": parse_power_data(mfr_data, 10),
+            "mode": mfr_data[9] & 0b00001111,
+            "position": mfr_data[14],
+            "calibration": bool(mfr_data[8] & 0b01000000),
         },
         2: {
             "switchMode": True,  # for compatibility, useless
             "sequence_number": mfr_data[6],
             "isOn": bool(mfr_data[7] & 0b01000000),
             "power": parse_power_data(mfr_data, 12),
+            "mode": (mfr_data[9] & 0b11110000) >> 4,
+            "position": mfr_data[14],
+            "calibration": bool(mfr_data[8] & 0b01000000),
         },
         "sequence_number": mfr_data[6],
     }

switchbot/devices/base_cover.py

@@ -33,8 +33,20 @@
 class SwitchbotBaseCover(SwitchbotDevice):
     """Representation of a Switchbot Cover devices for both curtains and tilt blinds."""
 
-    def __init__(self, reverse: bool, *args: Any, **kwargs: Any) -> None:
-        """Switchbot Cover device constructor."""
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """
+        Switchbot Cover device constructor.
+
+        ``reverse`` may be passed either as the first positional argument
+        (legacy form, kept for backwards compatibility) or as a keyword
+        argument (preferred — required for cooperative multiple inheritance
+        where ``reverse`` must travel through ``**kwargs`` via the MRO).
+        """
+        if args and isinstance(args[0], bool):
+            reverse: bool = args[0]
+            args = args[1:]
+        else:
+            reverse = kwargs.pop("reverse", False)
         super().__init__(*args, **kwargs)
         self._reverse = reverse
         self._settings: dict[str, Any] = {}

switchbot/devices/blind_tilt.py

@@ -45,7 +45,7 @@ class SwitchbotBlindTilt(SwitchbotBaseCover, SwitchbotSequenceDevice):
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         """Switchbot Blind Tilt/woBlindTilt constructor."""
         self._reverse: bool = kwargs.pop("reverse_mode", False)
-        super().__init__(self._reverse, *args, **kwargs)
+        super().__init__(*args, reverse=self._reverse, **kwargs)
 
     def _set_parsed_data(
         self, advertisement: SwitchBotAdvertisement, data: dict[str, Any]

switchbot/devices/curtain.py

@@ -47,10 +47,9 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         # 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)
+        super().__init__(*args, reverse=self._reverse, **kwargs)
         self._settings: dict[str, Any] = {}
         self.ext_info_sum: dict[str, Any] = {}
-        self.ext_info_adv: dict[str, Any] = {}
 
     def _set_parsed_data(
         self, advertisement: SwitchBotAdvertisement, data: dict[str, Any]

switchbot/devices/relay_switch.py

@@ -2,6 +2,8 @@
 import time
 from typing import Any
 
+from switchbot.devices.base_cover import SwitchbotBaseCover
+
 from ..const import SwitchbotModel
 from ..helpers import parse_power_data, parse_uint24_be
 from ..models import SwitchBotAdvertisement
@@ -53,6 +55,10 @@
     }
 }
 
+# roller mode command
+COMMAND_POSITION = f"{COMMAND_CONTROL}0D04{{}}01"
+COMMAND_STOP = f"{COMMAND_CONTROL}0D00"
+
 
 class SwitchbotRelaySwitch(SwitchbotSequenceDevice, SwitchbotEncryptedDevice):
     """Representation of a Switchbot relay switch 1pm."""
@@ -73,6 +79,9 @@ def _parse_common_data(self, raw_data: bytes) -> dict[str, Any]:
             "isOn": bool(raw_data[2] & SWITCH1_ON_MASK),
             "firmware": raw_data[16] / 10.0,
             "channel2_isOn": bool(raw_data[2] & SWITCH2_ON_MASK),
+            "calibration": bool(raw_data[3] & 0b01000000),
+            "mode": raw_data[4] & 0b00001111,
+            "position": raw_data[9],
         }
 
     def _parse_user_data(self, raw_data: bytes) -> dict[str, Any]:
@@ -162,7 +171,9 @@ async def get_basic_info(self) -> dict[str, Any] | None:
             return None
 
         _LOGGER.debug(
-            "on-off hex: %s, channel1_hex_data: %s", _data.hex(), _channel1_data.hex()
+            "get_basic_info raw: %s, channel1 raw: %s",
+            _data.hex(),
+            _channel1_data.hex(),
         )
 
         common_data = self._parse_common_data(_data)
@@ -210,7 +221,7 @@ class SwitchbotGarageDoorOpener(SwitchbotRelaySwitch):
     _press_command = f"{COMMAND_CONTROL}110329"  # for garage door opener toggle
 
 
-class SwitchbotRelaySwitch2PM(SwitchbotRelaySwitch):
+class SwitchbotRelaySwitch2PM(SwitchbotRelaySwitch, SwitchbotBaseCover):
     """Representation of a Switchbot relay switch 2pm."""
 
     _model = SwitchbotModel.RELAY_SWITCH_2PM
@@ -220,46 +231,111 @@ class SwitchbotRelaySwitch2PM(SwitchbotRelaySwitch):
     def channel(self) -> int:
         return self._channel
 
+    def get_position(self) -> Any:
+        """Return cached position (0-100) of Relay Switch 2PM."""
+        return self._get_adv_value("position", channel=1)
+
+    async def get_extended_info_summary(self) -> dict[str, Any] | None:
+        """Get extended info summary. Not supported for Relay Switch 2PM."""
+        return None
+
+    @property
+    def position(self) -> int | None:
+        """Return position."""
+        return self._get_adv_value("position", channel=1)
+
+    @property
+    def mode(self) -> int | None:
+        """Return mode."""
+        return self._get_adv_value("mode", channel=1)
+
+    async def _send_position(self, position: int) -> bool:
+        """Send a roller position command (0-100) and return success."""
+        result = await self._send_command(COMMAND_POSITION.format(f"{position:02X}"))
+        return self._check_command_result(result, 0, {1})
+
+    @update_after_operation
+    async def open(self) -> bool:
+        """Open the roller fully (device position 0, or 100 when reversed)."""
+        position = 100 if self._reverse else 0
+        if success := await self._send_position(position):
+            self._is_opening = True
+            self._is_closing = False
+        return success
+
+    @update_after_operation
+    async def close(self) -> bool:
+        """Close the roller fully (device position 100, or 0 when reversed)."""
+        position = 0 if self._reverse else 100
+        if success := await self._send_position(position):
+            self._is_closing = True
+            self._is_opening = False
+        return success
+
+    @update_after_operation
+    async def stop(self) -> bool:
+        """Send stop command to device."""
+        result = await self._send_command(COMMAND_STOP)
+        if success := self._check_command_result(result, 0, {1}):
+            self._is_opening = self._is_closing = False
+        return success
+
+    @update_after_operation
+    async def set_position(self, position: int) -> bool:
+        """Send position command (0-100) to device."""
+        if self._reverse:
+            position = 100 - position
+        position = max(0, min(100, position))
+        if success := await self._send_position(position):
+            prev = self._get_adv_value("position", channel=1)
+            self._update_motion_direction(
+                True,
+                (100 - prev) if prev is not None else None,
+                100 - position,
+            )
+        return success
+
     def get_parsed_data(self, channel: int | None = None) -> dict[str, Any]:
         """Return parsed device data, optionally for a specific channel."""
         data = self.data.get("data") or {}
         return data.get(channel, {})
 
     async def get_basic_info(self):
+        if not (channel1_data := await super().get_basic_info()):
+            return None
+
         current_time_hex, current_day_start_time_hex = (
             self.get_current_time_and_start_time()
         )
-        if not (common_data := await super().get_basic_info()):
-            return None
         if not (
-            _channel2_data := await self._get_basic_info(
+            _channel2_raw := await self._get_basic_info(
                 COMMAND_GET_CHANNEL2_INFO.format(
                     current_time_hex, current_day_start_time_hex
                 )
             )
         ):
             return None
-        if len(_channel2_data) < 15:
+        if len(_channel2_raw) < 15:
             _LOGGER.warning(
                 "%s: Short channel2 response (%d bytes): %s",
                 self.name,
-                len(_channel2_data),
-                _channel2_data.hex(),
+                len(_channel2_raw),
+                _channel2_raw.hex(),
             )
             return None
 
-        _LOGGER.debug("channel2_hex_data: %s", _channel2_data.hex())
+        _LOGGER.debug("channel2_raw: %s", _channel2_raw.hex())
 
-        channel2_data = self._parse_user_data(_channel2_data)
-        channel2_data["isOn"] = common_data["channel2_isOn"]
+        channel2_data = self._parse_user_data(_channel2_raw)
+        channel2_data["isOn"] = channel1_data["channel2_isOn"]
 
         if not channel2_data["isOn"]:
             self._reset_power_data(channel2_data)
 
         _LOGGER.debug(
-            "channel1_data: %s, channel2_data: %s", common_data, channel2_data
+            "channel1_data: %s, channel2_data: %s", channel1_data, channel2_data
         )
-        return {1: common_data, 2: channel2_data}
+        return {1: channel1_data, 2: channel2_data}
 
     @update_after_operation
     async def turn_on(self, channel: int) -> bool:
@@ -292,3 +368,17 @@ def is_on(self, channel: int) -> bool | None:
     def switch_mode(self, channel: int) -> bool | None:
         """Return true or false from cache."""
         return self._get_adv_value("switchMode", channel)
+
+    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/devices/roller_shade.py

@@ -37,7 +37,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         # 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)
+        super().__init__(*args, reverse=self._reverse, **kwargs)
 
     def _set_parsed_data(
         self, advertisement: SwitchBotAdvertisement, data: dict[str, Any]

tests/test_adv_parser.py

@@ -3474,12 +3474,18 @@ def test_humidifer_with_empty_data() -> None:
                     "sequence_number": 138,
                     "switchMode": True,
                     "power": 0.0,
+                    "mode": 0,
+                    "position": 0,
+                    "calibration": False,
                 },
                 2: {
                     "isOn": True,
                     "sequence_number": 138,
                     "switchMode": True,
                     "power": 70.0,
+                    "mode": 0,
+                    "position": 0,
+                    "calibration": False,
                 },
                 "sequence_number": 138,
             },
@@ -3940,12 +3946,18 @@ def test_adv_active(test_case: AdvTestCase) -> None:
                     "sequence_number": 138,
                     "switchMode": True,
                     "power": 0.0,
+                    "mode": 0,
+                    "position": 0,
+                    "calibration": False,
                 },
                 2: {
                     "isOn": True,
                     "sequence_number": 138,
                     "switchMode": True,
                     "power": 70.0,
+                    "mode": 0,
+                    "position": 0,
+                    "calibration": False,
                 },
                 "sequence_number": 138,
             },

tests/test_base_cover.py

@@ -11,7 +11,7 @@
 
 def create_device_for_command_testing(position=50, calibration=True):
     ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
-    base_cover_device = base_cover.SwitchbotBaseCover(False, ble_device)
+    base_cover_device = base_cover.SwitchbotBaseCover(ble_device, reverse=False)
     base_cover_device.update_from_advertisement(
         make_advertisement_data(ble_device, True, position, calibration)
     )
@@ -50,7 +50,7 @@ def make_advertisement_data(
 @pytest.mark.asyncio
 async def test_send_multiple_commands():
     ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
-    base_cover_device = base_cover.SwitchbotBaseCover(False, ble_device)
+    base_cover_device = base_cover.SwitchbotBaseCover(ble_device, reverse=False)
     base_cover_device.update_from_advertisement(
         make_advertisement_data(ble_device, True, 50, True)
     )
@@ -149,3 +149,27 @@ async def test_get_extended_info_adv_returns_device1_charge_states(data_value, r
     )
     ext_result = await base_cover_device.get_extended_info_adv()
     assert ext_result["device1"]["stateOfCharge"] == result
+
+
+def test_reverse_accepts_legacy_positional_arg():
+    """Legacy callers passed ``reverse`` as the first positional arg."""
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
+    device = base_cover.SwitchbotBaseCover(True, ble_device)
+    assert device.is_reversed() is True
+    assert device._device is ble_device
+
+
+def test_reverse_accepts_kwarg():
+    """Modern callers pass ``reverse`` as a kwarg (required for cooperative MRO)."""
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
+    device = base_cover.SwitchbotBaseCover(ble_device, reverse=True)
+    assert device.is_reversed() is True
+    assert device._device is ble_device
+
+
+def test_reverse_defaults_to_false():
+    """When ``reverse`` is omitted entirely it defaults to False."""
+    ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any")
+    device = base_cover.SwitchbotBaseCover(ble_device)
+    assert device.is_reversed() is False
+    assert device._device is ble_device