Add quietdrift (quiet mode) support for roller shades (#456)

Co-authored-by: J. Nick Koston <nick@home-assistant.io>

switchbot/devices/roller_shade.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import warnings
 from typing import Any
 
 from ..models import SwitchBotAdvertisement
@@ -14,11 +15,11 @@ _LOGGER = logging.getLogger(__name__)
 
 OPEN_KEYS = [
     f"{REQ_HEADER}{ROLLERSHADE_COMMAND}01{CONTROL_SOURCE}0100",
-    f"{REQ_HEADER}{ROLLERSHADE_COMMAND}05{CONTROL_SOURCE}0000",
+    f"{REQ_HEADER}{ROLLERSHADE_COMMAND}05{CONTROL_SOURCE}",  # +mode + "00"
 ]
 CLOSE_KEYS = [
     f"{REQ_HEADER}{ROLLERSHADE_COMMAND}01{CONTROL_SOURCE}0164",
-    f"{REQ_HEADER}{ROLLERSHADE_COMMAND}05{CONTROL_SOURCE}0064",
+    f"{REQ_HEADER}{ROLLERSHADE_COMMAND}05{CONTROL_SOURCE}",  # +mode + "64"
 ]
 POSITION_KEYS = [
     f"{REQ_HEADER}{ROLLERSHADE_COMMAND}01{CONTROL_SOURCE}01",
@@ -48,37 +49,74 @@ class SwitchbotRollerShade(SwitchbotBaseCover, SwitchbotSequenceDevice):
         self._update_motion_direction(in_motion, previous_position, new_position)
         super()._set_parsed_data(advertisement, data)
 
+    @staticmethod
+    def _validate_mode(mode: int) -> None:
+        """Validate the motor mode (0 = performance, 1 = quiet)."""
+        if mode not in (0, 1):
+            raise ValueError(f"mode must be 0 (performance) or 1 (quiet), got {mode!r}")
+
     @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)
+        """Send open command. 0 - performance mode, 1 - quiet mode."""
+        self._validate_mode(mode)
+        if success := await self._send_multiple_commands(
+            [OPEN_KEYS[0], f"{OPEN_KEYS[1]}{mode:02X}00"]
+        ):
+            self._is_opening = True
+            self._is_closing = False
+        return success
 
     @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)
+    async def close(self, mode: int = 0, **kwargs: Any) -> bool:
+        """
+        Send close command. 0 - performance mode, 1 - quiet mode.
+
+        ``speed`` is accepted as a deprecated alias for ``mode``: prior to
+        the quietdrift work the ``speed`` parameter existed on this method
+        but was a no-op. Callers passing ``speed=`` get a
+        ``DeprecationWarning`` and the value is forwarded as ``mode``.
+        """
+        if "speed" in kwargs:
+            warnings.warn(
+                "`speed` kwarg on close() is deprecated; use `mode` instead",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+            mode = kwargs.pop("speed")
+        if kwargs:
+            raise TypeError(
+                f"close() got unexpected keyword arguments: {sorted(kwargs)}"
+            )
+        self._validate_mode(mode)
+        if success := await self._send_multiple_commands(
+            [CLOSE_KEYS[0], f"{CLOSE_KEYS[1]}{mode:02X}64"]
+        ):
+            self._is_closing = True
+            self._is_opening = False
+        return success
 
     @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)
+        if success := await self._send_multiple_commands(STOP_KEYS):
+            self._is_opening = self._is_closing = False
+        return success
 
     @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."""
+        """Send position command (0-100) to device. 0 - performance mode, 1 - quiet mode."""
+        self._validate_mode(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(
+        if success := await self._send_multiple_commands(
             [
                 f"{POSITION_KEYS[0]}{position:02X}",
                 f"{POSITION_KEYS[1]}{mode:02X}{position:02X}",
             ]
-        )
+        ):
+            self._update_motion_direction(
+                True, self._get_adv_value("position"), position
+            )
+        return success
 
     def get_position(self) -> Any:
         """Return cached position (0-100) of Curtain."""

tests/test_roller_shade.py

@@ -58,7 +58,18 @@ async def test_open():
     assert roller_shade_device.is_opening() is True
     assert roller_shade_device.is_closing() is False
     roller_shade_device._send_multiple_commands.assert_awaited_once_with(
-        roller_shade.OPEN_KEYS
+        [roller_shade.OPEN_KEYS[0], f"{roller_shade.OPEN_KEYS[1]}0000"]
+    )
+
+
+@pytest.mark.asyncio
+async def test_open_quietdrift():
+    roller_shade_device = create_device_for_command_testing()
+    await roller_shade_device.open(mode=1)
+    assert roller_shade_device.is_opening() is True
+    assert roller_shade_device.is_closing() is False
+    roller_shade_device._send_multiple_commands.assert_awaited_once_with(
+        [roller_shade.OPEN_KEYS[0], f"{roller_shade.OPEN_KEYS[1]}0100"]
     )
 
 
@@ -69,7 +80,18 @@ async def test_close():
     assert roller_shade_device.is_opening() is False
     assert roller_shade_device.is_closing() is True
     roller_shade_device._send_multiple_commands.assert_awaited_once_with(
-        roller_shade.CLOSE_KEYS
+        [roller_shade.CLOSE_KEYS[0], f"{roller_shade.CLOSE_KEYS[1]}0064"]
+    )
+
+
+@pytest.mark.asyncio
+async def test_close_quietdrift():
+    roller_shade_device = create_device_for_command_testing()
+    await roller_shade_device.close(mode=1)
+    assert roller_shade_device.is_opening() is False
+    assert roller_shade_device.is_closing() is True
+    roller_shade_device._send_multiple_commands.assert_awaited_once_with(
+        [roller_shade.CLOSE_KEYS[0], f"{roller_shade.CLOSE_KEYS[1]}0164"]
     )
 
 
@@ -204,6 +226,144 @@ async def test_set_position_closing():
     curtain_device._send_multiple_commands.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_set_position_default_mode_performance():
+    """`mode=0` (default) must send the same wire bytes as before quiet mode."""
+    curtain_device = create_device_for_command_testing()
+    await curtain_device.set_position(50)
+    curtain_device._send_multiple_commands.assert_awaited_once_with(
+        [
+            f"{roller_shade.POSITION_KEYS[0]}32",
+            f"{roller_shade.POSITION_KEYS[1]}0032",
+        ]
+    )
+
+
+@pytest.mark.asyncio
+async def test_set_position_quietdrift():
+    """`mode=1` flips the mode byte while leaving the position byte alone."""
+    curtain_device = create_device_for_command_testing()
+    await curtain_device.set_position(50, mode=1)
+    curtain_device._send_multiple_commands.assert_awaited_once_with(
+        [
+            f"{roller_shade.POSITION_KEYS[0]}32",
+            f"{roller_shade.POSITION_KEYS[1]}0132",
+        ]
+    )
+
+
+@pytest.mark.asyncio
+async def test_set_position_quietdrift_reversed():
+    """Quiet mode and reverse mode are independent — both apply correctly."""
+    curtain_device = create_device_for_command_testing(reverse_mode=True)
+    # position=30 with reverse → device position 70 (0x46), mode byte 01
+    await curtain_device.set_position(30, mode=1)
+    curtain_device._send_multiple_commands.assert_awaited_once_with(
+        [
+            f"{roller_shade.POSITION_KEYS[0]}46",
+            f"{roller_shade.POSITION_KEYS[1]}0146",
+        ]
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("invalid_mode", [-1, 2, 255])
+async def test_open_rejects_invalid_mode(invalid_mode):
+    roller_shade_device = create_device_for_command_testing()
+    with pytest.raises(ValueError, match="mode must be 0"):
+        await roller_shade_device.open(mode=invalid_mode)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("invalid_mode", [-1, 2, 255])
+async def test_close_rejects_invalid_mode(invalid_mode):
+    roller_shade_device = create_device_for_command_testing()
+    with pytest.raises(ValueError, match="mode must be 0"):
+        await roller_shade_device.close(mode=invalid_mode)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("invalid_mode", [-1, 2, 255])
+async def test_set_position_rejects_invalid_mode(invalid_mode):
+    roller_shade_device = create_device_for_command_testing()
+    with pytest.raises(ValueError, match="mode must be 0"):
+        await roller_shade_device.set_position(50, mode=invalid_mode)
+
+
+@pytest.mark.asyncio
+async def test_open_does_not_set_motion_flag_on_failure():
+    """If the open command fails, _is_opening must remain False."""
+    roller_shade_device = create_device_for_command_testing()
+    roller_shade_device._send_multiple_commands = AsyncMock(return_value=False)
+    result = await roller_shade_device.open()
+    assert result is False
+    assert roller_shade_device.is_opening() is False
+    assert roller_shade_device.is_closing() is False
+
+
+@pytest.mark.asyncio
+async def test_close_speed_kwarg_is_deprecated_alias_for_mode():
+    """`close(speed=1)` continues to work but emits DeprecationWarning."""
+    roller_shade_device = create_device_for_command_testing()
+    with pytest.warns(DeprecationWarning, match="speed.*deprecated"):
+        await roller_shade_device.close(speed=1)
+    roller_shade_device._send_multiple_commands.assert_awaited_once_with(
+        [roller_shade.CLOSE_KEYS[0], f"{roller_shade.CLOSE_KEYS[1]}0164"]
+    )
+
+
+@pytest.mark.asyncio
+async def test_close_speed_kwarg_validates_mode():
+    """A bad value via `speed=` is still rejected by `_validate_mode`."""
+    roller_shade_device = create_device_for_command_testing()
+    with (
+        pytest.warns(DeprecationWarning, match="speed.*deprecated"),
+        pytest.raises(ValueError, match="mode must be 0"),
+    ):
+        await roller_shade_device.close(speed=2)
+
+
+@pytest.mark.asyncio
+async def test_close_rejects_other_unexpected_kwargs():
+    """Unknown kwargs (other than `speed`) should still raise TypeError."""
+    roller_shade_device = create_device_for_command_testing()
+    with pytest.raises(TypeError, match="unexpected keyword arguments"):
+        await roller_shade_device.close(turbo=True)
+
+
+@pytest.mark.asyncio
+async def test_close_does_not_set_motion_flag_on_failure():
+    """If the close command fails, _is_closing must remain False."""
+    roller_shade_device = create_device_for_command_testing()
+    roller_shade_device._send_multiple_commands = AsyncMock(return_value=False)
+    result = await roller_shade_device.close()
+    assert result is False
+    assert roller_shade_device.is_opening() is False
+    assert roller_shade_device.is_closing() is False
+
+
+@pytest.mark.asyncio
+async def test_stop_does_not_clear_motion_flags_on_failure():
+    """If the stop command fails, prior motion flags persist."""
+    roller_shade_device = create_device_for_command_testing()
+    roller_shade_device._is_opening = True
+    roller_shade_device._send_multiple_commands = AsyncMock(return_value=False)
+    result = await roller_shade_device.stop()
+    assert result is False
+    assert roller_shade_device.is_opening() is True
+
+
+@pytest.mark.asyncio
+async def test_set_position_does_not_update_direction_on_failure():
+    """If set_position fails, the motion direction must not be touched."""
+    roller_shade_device = create_device_for_command_testing(position=50)
+    roller_shade_device._send_multiple_commands = AsyncMock(return_value=False)
+    result = await roller_shade_device.set_position(80)
+    assert result is False
+    assert roller_shade_device.is_opening() is False
+    assert roller_shade_device.is_closing() is False
+
+
 def test_get_position():
     curtain_device = create_device_for_command_testing()
     assert curtain_device.get_position() == 50