From ab541e9f06cd2cdf472d0c6954b4484841e95906 Mon Sep 17 00:00:00 2001 From: testdev Date: Mon, 22 Jun 2026 18:37:12 +0800 Subject: [PATCH 1/2] feat: add roller mode for Relay Switch 2PM - Add open/close/set_position commands for 2PM roller/cover mode - Add _send_position() helper to deduplicate position command logic - Fix type annotation in adv_parsers/relay_switch.py - Clean up dead code (COMMAND_OPEN/COMMAND_CLOSE) - Add comprehensive tests for roller mode and short-response guards --- switchbot/adv_parsers/relay_switch.py | 12 +- switchbot/devices/base_cover.py | 16 +- switchbot/devices/blind_tilt.py | 2 +- switchbot/devices/curtain.py | 3 +- switchbot/devices/relay_switch.py | 116 ++++++++-- switchbot/devices/roller_shade.py | 2 +- tests/test_adv_parser.py | 12 + tests/test_base_cover.py | 28 ++- tests/test_relay_switch.py | 316 ++++++++++++++++++++++++-- 9 files changed, 467 insertions(+), 40 deletions(-) diff --git a/switchbot/adv_parsers/relay_switch.py b/switchbot/adv_parsers/relay_switch.py index bbac74ef..5e07759f 100644 --- a/switchbot/adv_parsers/relay_switch.py +++ b/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], } diff --git a/switchbot/devices/base_cover.py b/switchbot/devices/base_cover.py index a721c8a4..23e1ea35 100644 --- a/switchbot/devices/base_cover.py +++ b/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] = {} diff --git a/switchbot/devices/blind_tilt.py b/switchbot/devices/blind_tilt.py index 3fbfea86..d708e592 100644 --- a/switchbot/devices/blind_tilt.py +++ b/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] diff --git a/switchbot/devices/curtain.py b/switchbot/devices/curtain.py index 877aa994..a8c6d7e4 100644 --- a/switchbot/devices/curtain.py +++ b/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] diff --git a/switchbot/devices/relay_switch.py b/switchbot/devices/relay_switch.py index cf3fda28..8d89e35d 100644 --- a/switchbot/devices/relay_switch.py +++ b/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] >> 1, } 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 diff --git a/switchbot/devices/roller_shade.py b/switchbot/devices/roller_shade.py index 0cbdc7ea..8b06e292 100644 --- a/switchbot/devices/roller_shade.py +++ b/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] diff --git a/tests/test_adv_parser.py b/tests/test_adv_parser.py index e940c8dc..d96497ef 100644 --- a/tests/test_adv_parser.py +++ b/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, }, diff --git a/tests/test_base_cover.py b/tests/test_base_cover.py index 071ac79f..3d490c1a 100644 --- a/tests/test_base_cover.py +++ b/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 diff --git a/tests/test_relay_switch.py b/tests/test_relay_switch.py index 252a34e3..ba795e7e 100644 --- a/tests/test_relay_switch.py +++ b/tests/test_relay_switch.py @@ -4,6 +4,7 @@ from bleak.backends.device import BLEDevice from switchbot import SwitchBotAdvertisement, SwitchbotModel +from switchbot.adv_parsers.relay_switch import process_relay_switch_2pm from switchbot.devices import relay_switch from switchbot.devices.device import _merge_data as merge_data @@ -122,7 +123,7 @@ def make_advertisement_data( {1: {"isOn": True}, 2: {"isOn": True}}, ], ) -async def test_turn_on_2PM(common_parametrize_2pm, init_data): +async def test_turn_on_2PM(common_parametrize_2pm, init_data) -> None: """Test turn on command.""" device = create_device_for_command_testing( common_parametrize_2pm["rawAdvData"], common_parametrize_2pm["model"], init_data @@ -147,7 +148,7 @@ async def test_turn_on_2PM(common_parametrize_2pm, init_data): {1: {"isOn": False}, 2: {"isOn": False}}, ], ) -async def test_turn_off_2PM(common_parametrize_2pm, init_data): +async def test_turn_off_2PM(common_parametrize_2pm, init_data) -> None: """Test turn off command.""" device = create_device_for_command_testing( common_parametrize_2pm["rawAdvData"], common_parametrize_2pm["model"], init_data @@ -166,7 +167,7 @@ async def test_turn_off_2PM(common_parametrize_2pm, init_data): @pytest.mark.asyncio -async def test_turn_toggle_2PM(common_parametrize_2pm): +async def test_turn_toggle_2PM(common_parametrize_2pm) -> None: """Test toggle command.""" device = create_device_for_command_testing( common_parametrize_2pm["rawAdvData"], common_parametrize_2pm["model"] @@ -185,7 +186,7 @@ async def test_turn_toggle_2PM(common_parametrize_2pm): @pytest.mark.asyncio -async def test_get_switch_mode_2PM(common_parametrize_2pm): +async def test_get_switch_mode_2PM(common_parametrize_2pm) -> None: """Test get switch mode.""" device = create_device_for_command_testing( common_parametrize_2pm["rawAdvData"], common_parametrize_2pm["model"] @@ -216,7 +217,7 @@ async def test_get_switch_mode_2PM(common_parametrize_2pm): ), ], ) -async def test_get_basic_info_2PM(common_parametrize_2pm, info_data, result): +async def test_get_basic_info_2PM(common_parametrize_2pm, info_data, result) -> None: """Test get_basic_info for 2PM devices.""" device = create_device_for_command_testing( common_parametrize_2pm["rawAdvData"], common_parametrize_2pm["model"] @@ -279,7 +280,7 @@ async def mock_get_basic_info(arg): }, ], ) -async def test_basic_info_exceptions_2PM(common_parametrize_2pm, info_data): +async def test_basic_info_exceptions_2PM(common_parametrize_2pm, info_data) -> None: """Test get_basic_info exceptions.""" device = create_device_for_command_testing( common_parametrize_2pm["rawAdvData"], common_parametrize_2pm["model"] @@ -412,7 +413,7 @@ async def mock_get_basic_info(arg): @pytest.mark.asyncio -async def test_get_parsed_data_2PM(common_parametrize_2pm): +async def test_get_parsed_data_2PM(common_parametrize_2pm) -> None: """Test get_parsed_data for 2PM devices.""" device = create_device_for_command_testing( common_parametrize_2pm["rawAdvData"], common_parametrize_2pm["model"] @@ -430,7 +431,7 @@ async def test_get_parsed_data_2PM(common_parametrize_2pm): ("rawAdvData", "model"), common_params, ) -async def test_turn_on(rawAdvData, model): +async def test_turn_on(rawAdvData, model) -> None: """Test turn on command.""" device = create_device_for_command_testing(rawAdvData, model) await device.turn_on() @@ -443,7 +444,7 @@ async def test_turn_on(rawAdvData, model): ("rawAdvData", "model"), common_params, ) -async def test_turn_off(rawAdvData, model): +async def test_turn_off(rawAdvData, model) -> None: """Test turn off command.""" device = create_device_for_command_testing(rawAdvData, model, {"isOn": False}) await device.turn_off() @@ -456,7 +457,7 @@ async def test_turn_off(rawAdvData, model): ("rawAdvData", "model"), common_params, ) -async def test_toggle(rawAdvData, model): +async def test_toggle(rawAdvData, model) -> None: """Test toggle command.""" device = create_device_for_command_testing(rawAdvData, model) await device.async_toggle() @@ -478,7 +479,7 @@ async def test_toggle(rawAdvData, model): ) ], ) -async def test_get_basic_info_garage_door_opener(rawAdvData, model, info_data): +async def test_get_basic_info_garage_door_opener(rawAdvData, model, info_data) -> None: """Test get_basic_info for garage door opener.""" device = create_device_for_command_testing(rawAdvData, model) device.get_current_time_and_start_time = MagicMock( @@ -507,7 +508,7 @@ async def mock_get_basic_info(arg): (relay_switch.SwitchbotRelaySwitch2PM, SwitchbotModel.RELAY_SWITCH_2PM), ], ) -def test_default_model_classvar(dev_cls, expected_model): +def test_default_model_classvar(dev_cls, expected_model) -> None: ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any") device = dev_cls(ble_device, "ff", "ffffffffffffffffffffffffffffffff") assert device._model == expected_model @@ -534,14 +535,14 @@ def test_default_model_classvar(dev_cls, expected_model): ), ], ) -def test_merge_data(old_data, new_data, expected_result): +def test_merge_data(old_data, new_data, expected_result) -> None: """Test merging of data dictionaries.""" result = merge_data(old_data, new_data) assert result == expected_result @pytest.mark.asyncio -async def test_garage_door_opener_open(): +async def test_garage_door_opener_open() -> None: """Test open the garage door.""" device = create_device_for_command_testing( b">\x00\x00\x00", SwitchbotModel.GARAGE_DOOR_OPENER @@ -552,7 +553,7 @@ async def test_garage_door_opener_open(): @pytest.mark.asyncio -async def test_garage_door_opener_close(): +async def test_garage_door_opener_close() -> None: """Test close the garage door.""" device = create_device_for_command_testing( b">\x00\x00\x00", SwitchbotModel.GARAGE_DOOR_OPENER @@ -570,7 +571,7 @@ async def test_garage_door_opener_close(): ], ) @pytest.mark.asyncio -async def test_garage_door_opener_door_open(door_open): +async def test_garage_door_opener_door_open(door_open) -> None: """Test get garage door state.""" device = create_device_for_command_testing( b">\x00\x00\x00", SwitchbotModel.GARAGE_DOOR_OPENER, {"door_open": door_open} @@ -579,10 +580,291 @@ async def test_garage_door_opener_door_open(door_open): @pytest.mark.asyncio -async def test_press(): +async def test_press() -> None: """Test the press command for garage door opener.""" device = create_device_for_command_testing( b">\x00\x00\x00", SwitchbotModel.GARAGE_DOOR_OPENER ) await device.press() device._send_command.assert_awaited_once_with(device._press_command) + + +def create_2pm_device_with_position(position: int = 50, calibration: bool = True): + """Create a 2PM device with position/calibration data for cover testing.""" + return create_device_for_command_testing( + b"\x00\x00\x00\x00\x00\x00", + SwitchbotModel.RELAY_SWITCH_2PM, + { + 1: { + "switchMode": True, + "sequence_number": 99, + "isOn": True, + "position": position, + "calibration": calibration, + "mode": 0, + }, + 2: { + "switchMode": True, + "sequence_number": 99, + "isOn": False, + "position": position, + "calibration": calibration, + "mode": 0, + }, + }, + ) + + +@pytest.mark.asyncio +async def test_2pm_open() -> None: + """Test open command for 2PM roller mode.""" + device = create_2pm_device_with_position() + await device.open() + device._send_command.assert_called_with( + relay_switch.COMMAND_POSITION.format(f"{0:02X}") + ) + assert device.is_opening() is True + assert device.is_closing() is False + + +@pytest.mark.asyncio +async def test_2pm_close() -> None: + """Test close command for 2PM roller mode.""" + device = create_2pm_device_with_position() + await device.close() + device._send_command.assert_called_with( + relay_switch.COMMAND_POSITION.format(f"{100:02X}") + ) + assert device.is_opening() is False + assert device.is_closing() is True + + +@pytest.mark.asyncio +async def test_2pm_stop() -> None: + """Test stop command for 2PM roller mode.""" + device = create_2pm_device_with_position() + await device.stop() + device._send_command.assert_called_with(relay_switch.COMMAND_STOP) + assert device.is_opening() is False + assert device.is_closing() is False + + +@pytest.mark.asyncio +async def test_2pm_set_position_closing() -> None: + """Test set_position to a higher device position (closing in HA terms).""" + device = create_2pm_device_with_position(position=30) + await device.set_position(80) + device._send_command.assert_called_with( + relay_switch.COMMAND_POSITION.format(f"{80:02X}") + ) + assert device.is_opening() is False + assert device.is_closing() is True + + +@pytest.mark.asyncio +async def test_2pm_set_position_opening() -> None: + """Test set_position to a lower device position (opening in HA terms).""" + device = create_2pm_device_with_position(position=80) + await device.set_position(20) + device._send_command.assert_called_with( + relay_switch.COMMAND_POSITION.format(f"{20:02X}") + ) + assert device.is_opening() is True + assert device.is_closing() is False + + +@pytest.mark.asyncio +async def test_2pm_set_position_passthrough() -> None: + """Test set_position sends position directly without transformation.""" + ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any") + device = relay_switch.SwitchbotRelaySwitch2PM( + ble_device, "ff", "ffffffffffffffffffffffffffffffff" + ) + device.update_from_advertisement( + make_advertisement_data( + ble_device, + b"\x00\x00\x00\x00\x00\x00", + SwitchbotModel.RELAY_SWITCH_2PM, + { + 1: { + "switchMode": True, + "sequence_number": 99, + "isOn": True, + "position": 30, + "calibration": True, + "mode": 0, + }, + 2: { + "switchMode": True, + "sequence_number": 99, + "isOn": False, + "position": 30, + "calibration": True, + "mode": 0, + }, + }, + ) + ) + device._send_command = AsyncMock() + device._check_command_result = MagicMock() + device.update = AsyncMock() + + await device.set_position(40) + # position sent directly as-is + device._send_command.assert_called_with( + relay_switch.COMMAND_POSITION.format(f"{40:02X}") + ) + + +def test_2pm_position_property() -> None: + """Test position property returns value from channel 1.""" + device = create_2pm_device_with_position(position=42) + assert device.position == 42 + + +def test_2pm_mode_property() -> None: + """Test mode property returns value from channel 1.""" + device = create_2pm_device_with_position() + assert device.mode == 0 + + +def test_2pm_update_motion_direction_no_previous() -> None: + """Test _update_motion_direction with no previous position does nothing.""" + device = create_2pm_device_with_position() + device._update_motion_direction(True, None, 80) + assert device.is_opening() is False + assert device.is_closing() is False + + +def test_2pm_update_motion_direction_stop() -> None: + """Test _update_motion_direction with in_motion=False clears both flags.""" + device = create_2pm_device_with_position() + device._is_opening = True + device._is_closing = True + device._update_motion_direction(False, 50, 80) + assert device.is_opening() is False + assert device.is_closing() is False + + +def test_2pm_update_motion_direction_opening() -> None: + """Test _update_motion_direction detects opening.""" + device = create_2pm_device_with_position() + device._update_motion_direction(True, 30, 70) + assert device.is_opening() is True + assert device.is_closing() is False + + +def test_2pm_update_motion_direction_closing() -> None: + """Test _update_motion_direction detects closing.""" + device = create_2pm_device_with_position() + device._update_motion_direction(True, 70, 30) + assert device.is_opening() is False + assert device.is_closing() is True + + +@pytest.mark.asyncio +async def test_2pm_open_does_not_set_motion_flag_on_failure() -> None: + """If the open command result fails, _is_opening must remain False.""" + device = create_2pm_device_with_position() + device._check_command_result = MagicMock(return_value=False) + result = await device.open() + assert result is False + assert device.is_opening() is False + assert device.is_closing() is False + + +@pytest.mark.asyncio +async def test_2pm_close_does_not_set_motion_flag_on_failure() -> None: + """If the close command result fails, _is_closing must remain False.""" + device = create_2pm_device_with_position() + device._check_command_result = MagicMock(return_value=False) + result = await device.close() + assert result is False + assert device.is_opening() is False + assert device.is_closing() is False + + +@pytest.mark.asyncio +async def test_2pm_stop_does_not_clear_motion_flags_on_failure() -> None: + """If the stop command result fails, the prior motion flags persist.""" + device = create_2pm_device_with_position() + device._is_opening = True + device._is_closing = False + device._check_command_result = MagicMock(return_value=False) + result = await device.stop() + assert result is False + assert device.is_opening() is True + assert device.is_closing() is False + + +@pytest.mark.asyncio +async def test_2pm_set_position_does_not_update_direction_on_failure() -> None: + """If the set_position command result fails, motion flags must not change.""" + device = create_2pm_device_with_position(position=30) + device._check_command_result = MagicMock(return_value=False) + result = await device.set_position(80) + assert result is False + assert device.is_opening() is False + assert device.is_closing() is False + + +def test_parse_common_data_includes_sequence_number() -> None: + """ + `_parse_common_data` must expose `sequence_number` from raw_data[1]. + + Regression test: the key was inadvertently dropped during the 2PM roller + work, breaking any 1PM `get_basic_info` consumer reading it. + """ + ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any") + device = relay_switch.SwitchbotRelaySwitch( + ble_device, "ff", "ffffffffffffffffffffffffffffffff" + ) + raw_data = bytes( + [ + 0x01, + 0x2A, + 0x80, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x10, + ] + ) + parsed = device._parse_common_data(raw_data) + assert parsed["sequence_number"] == 0x2A + + +def test_2pm_adv_parses_distinct_per_channel_modes() -> None: + """ + Channel 1 mode is the lower nibble of mfr_data[9]; channel 2 the upper. + + Regression for the precedence bug `mfr_data[9] & 0b11110000 >> 4` which + Python parses as `mfr_data[9] & (0b11110000 >> 4)` = `mfr_data[9] & 0x0F`, + silently returning channel 1's mode for channel 2. + """ + # mfr_data[9] = 0x53 → lower nibble 3 (channel 1), upper nibble 5 (channel 2) + mfr_data = bytes(6) + bytes([0x8A, 0xC1, 0x00, 0x53, 0x00, 0x00, 0x00, 0x00, 0x00]) + parsed = process_relay_switch_2pm(None, mfr_data) + assert parsed[1]["mode"] == 0x3 + assert parsed[2]["mode"] == 0x5 + + +@pytest.mark.parametrize("mfr_data", [None, b"", b"\x00" * 4, b"\x00" * 14]) +def test_2pm_adv_short_mfr_data_returns_empty(mfr_data: bytes | None) -> None: + """ + Truncated advertisements must not raise. + + The parser reads up to mfr_data[14] (roller position); anything shorter + than 15 bytes must degrade to an empty dict instead of raising IndexError. + """ + assert process_relay_switch_2pm(None, mfr_data) == {} From 330ddd3d7d414fd386685cf8c77842988590689d Mon Sep 17 00:00:00 2001 From: testdev Date: Wed, 24 Jun 2026 17:04:05 +0800 Subject: [PATCH 2/2] fix: align position encoding between advertisement and get_basic_info Co-Authored-By: Claude Opus 4.8 --- switchbot/devices/relay_switch.py | 2 +- tests/test_relay_switch.py | 38 +++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/switchbot/devices/relay_switch.py b/switchbot/devices/relay_switch.py index 8d89e35d..436de137 100644 --- a/switchbot/devices/relay_switch.py +++ b/switchbot/devices/relay_switch.py @@ -81,7 +81,7 @@ def _parse_common_data(self, raw_data: bytes) -> dict[str, Any]: "channel2_isOn": bool(raw_data[2] & SWITCH2_ON_MASK), "calibration": bool(raw_data[3] & 0b01000000), "mode": raw_data[4] & 0b00001111, - "position": raw_data[9] >> 1, + "position": raw_data[9], } def _parse_user_data(self, raw_data: bytes) -> dict[str, Any]: diff --git a/tests/test_relay_switch.py b/tests/test_relay_switch.py index ba795e7e..c1bf4719 100644 --- a/tests/test_relay_switch.py +++ b/tests/test_relay_switch.py @@ -844,6 +844,44 @@ def test_parse_common_data_includes_sequence_number() -> None: assert parsed["sequence_number"] == 0x2A +def test_parse_common_data_position_is_unshifted_percentage() -> None: + """ + `_parse_common_data` must expose `position` as the raw 0-100 percentage. + + Regression test: the value was shifted right by 1 (`raw_data[9] >> 1`), + yielding 0-50 from `get_basic_info` while the advertisement parser reports + the raw 0-100 value, so the same logical position reached Home Assistant + with two different encodings. + """ + ble_device = generate_ble_device("aa:bb:cc:dd:ee:ff", "any") + device = relay_switch.SwitchbotRelaySwitch( + ble_device, "ff", "ffffffffffffffffffffffffffffffff" + ) + raw_data = bytes( + [ + 0x01, + 0x2A, + 0x80, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x64, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x10, + ] + ) + parsed = device._parse_common_data(raw_data) + assert parsed["position"] == 100 + + def test_2pm_adv_parses_distinct_per_channel_modes() -> None: """ Channel 1 mode is the lower nibble of mfr_data[9]; channel 2 the upper.