From 35e735443c550bf487be380e79f2c50c74020812 Mon Sep 17 00:00:00 2001 From: Luke Date: Fri, 24 Apr 2026 23:34:44 -0400 Subject: [PATCH 1/5] feat: add better cleaning mode support --- roborock/data/v1/v1_clean_modes.py | 101 +++++++- roborock/devices/traits/v1/status.py | 41 +++- .../devices/__snapshots__/test_v1_device.ambr | 2 +- tests/devices/traits/v1/test_status.py | 229 +++++++++++++++++- 4 files changed, 366 insertions(+), 7 deletions(-) diff --git a/roborock/data/v1/v1_clean_modes.py b/roborock/data/v1/v1_clean_modes.py index 2b709d0d..af0f791a 100644 --- a/roborock/data/v1/v1_clean_modes.py +++ b/roborock/data/v1/v1_clean_modes.py @@ -1,7 +1,9 @@ from __future__ import annotations import typing +from enum import StrEnum +from ...exceptions import RoborockUnsupportedFeature from ..code_mappings import RoborockModeEnum if typing.TYPE_CHECKING: @@ -68,6 +70,14 @@ class WashTowelModes(RoborockModeEnum): SUPER_DEEP = ("super_deep", 8) +class CleaningModes(StrEnum): + VACUUM = "vacuum" + VAC_AND_MOP = "vac_and_mop" + MOP = "mop" + CUSTOM = "custom" + SMART_MODE = "smart_mode" + + WATER_SLIDE_MODE_MAPPING: dict[int, WaterModes] = { 200: WaterModes.OFF, 221: WaterModes.PURE_WATER_FLOW_START, @@ -174,7 +184,90 @@ def get_water_mode_mapping(features: DeviceFeatures) -> dict[int, str]: return {mode.code: mode.value for mode in get_water_modes(features)} -def is_mode_customized(clean_mode: VacuumModes, water_mode: WaterModes, mop_mode: CleanRoutes) -> bool: +def get_cleaning_mode_options(features: DeviceFeatures) -> list[CleaningModes]: + """Get the supported high-level cleaning modes for the device.""" + if not features.is_support_water_mode: + return [] + + options = [CleaningModes.VACUUM, CleaningModes.VAC_AND_MOP] + if features.is_pure_clean_mop_supported: + options.append(CleaningModes.MOP) + if features.is_customized_clean_supported: + options.append(CleaningModes.CUSTOM) + if features.is_smart_clean_mode_set_supported: + options.append(CleaningModes.SMART_MODE) + return options + + +def get_mop_only_vacuum_mode(features: DeviceFeatures) -> VacuumModes: + if not features.is_pure_clean_mop_supported: + raise RoborockUnsupportedFeature("Mop-only cleaning is not supported") + if features.is_support_main_brush_up_down_supported: + return VacuumModes.OFF_RAISE_MAIN_BRUSH + return VacuumModes.OFF + + +_CLEAN_MOTOR_MODE_PARAMS: dict[CleaningModes, tuple[int, int, int]] = { + CleaningModes.VACUUM: (VacuumModes.BALANCED.code, WaterModes.OFF.code, CleanRoutes.STANDARD.code), + CleaningModes.VAC_AND_MOP: (VacuumModes.BALANCED.code, WaterModes.STANDARD.code, CleanRoutes.STANDARD.code), + CleaningModes.CUSTOM: (VacuumModes.CUSTOMIZED.code, WaterModes.CUSTOMIZED.code, CleanRoutes.CUSTOMIZED.code), + CleaningModes.SMART_MODE: (VacuumModes.SMART_MODE.code, WaterModes.SMART_MODE.code, CleanRoutes.SMART_MODE.code), +} + + +def get_cleaning_mode_parameters(cleaning_mode: str | CleaningModes, features: DeviceFeatures) -> list[dict[str, int]]: + """Get the RPC payload for switching the high-level cleaning mode.""" + try: + mode = CleaningModes(cleaning_mode) + except ValueError as err: + raise RoborockUnsupportedFeature(f"Cleaning mode {cleaning_mode!r} is not supported") from err + if mode not in get_cleaning_mode_options(features): + raise RoborockUnsupportedFeature(f"Cleaning mode {mode.value!r} is not supported") + + if mode == CleaningModes.MOP: + fan_power = get_mop_only_vacuum_mode(features).code + water_box_mode = WaterModes.STANDARD.code + mop_mode = CleanRoutes.STANDARD.code + else: + fan_power, water_box_mode, mop_mode = _CLEAN_MOTOR_MODE_PARAMS[mode] + + params: dict[str, int] = {"fan_power": fan_power, "water_box_mode": water_box_mode} + if features.is_clean_route_setting_supported: + params["mop_mode"] = mop_mode + return [params] + + +def get_current_cleaning_mode( + clean_mode: int | None, + water_mode: int | None, + mop_mode: int | None, + features: DeviceFeatures, +) -> CleaningModes | None: + """Classify the current high-level cleaning mode from individual mode codes.""" + if not features.is_support_water_mode: + return None + if clean_mode is None or water_mode is None: + return None + + if is_smart_mode_set(water_mode, clean_mode, mop_mode): + return CleaningModes.SMART_MODE + if is_mode_customized(clean_mode, water_mode, mop_mode): + return CleaningModes.CUSTOM + try: + if clean_mode == get_mop_only_vacuum_mode(features).code: + return CleaningModes.MOP + except RoborockUnsupportedFeature: + pass + if water_mode == WaterModes.OFF.code: + return CleaningModes.VACUUM + return CleaningModes.VAC_AND_MOP + + +def is_mode_customized( + clean_mode: int | VacuumModes | None, + water_mode: int | WaterModes | None, + mop_mode: int | CleanRoutes | None, +) -> bool: """Check if any of the cleaning modes are set to a custom value.""" return ( clean_mode == VacuumModes.CUSTOMIZED @@ -183,7 +276,11 @@ def is_mode_customized(clean_mode: VacuumModes, water_mode: WaterModes, mop_mode ) -def is_smart_mode_set(water_mode: WaterModes, clean_mode: VacuumModes, mop_mode: CleanRoutes) -> bool: +def is_smart_mode_set( + water_mode: int | WaterModes | None, + clean_mode: int | VacuumModes | None, + mop_mode: int | CleanRoutes | None, +) -> bool: """Check if the smart mode is set for the given water mode and clean mode""" return ( water_mode == WaterModes.SMART_MODE diff --git a/roborock/devices/traits/v1/status.py b/roborock/devices/traits/v1/status.py index 82371c15..bc581c6d 100644 --- a/roborock/devices/traits/v1/status.py +++ b/roborock/devices/traits/v1/status.py @@ -1,12 +1,16 @@ from functools import cached_property from roborock import ( + CleaningModes, CleanRoutes, StatusV2, VacuumModes, WaterModes, get_clean_modes, get_clean_routes, + get_cleaning_mode_options, + get_cleaning_mode_parameters, + get_current_cleaning_mode, get_water_mode_mapping, get_water_modes, ) @@ -34,9 +38,10 @@ class StatusTrait(StatusV2, common.V1TraitMixin): - Water Mode - Mop Route - You should call the _options() version of the attribute to know which are supported for your device - (i.e. fan_speed_options()) - Then you can call the _mapping to convert an int value to the actual Enum. (i.e. fan_speed_mapping()) + You should call the _options() version of the attribute to know which are + supported for your device (i.e. fan_speed_options()) + Then you can call the _mapping to convert an int value to the actual Enum. + (i.e. fan_speed_mapping()) You can call the _name property to get the str value of the enum. (i.e. fan_speed_name) """ @@ -74,6 +79,10 @@ def mop_route_options(self) -> list[CleanRoutes]: def mop_route_mapping(self) -> dict[int, str]: return {route.code: route.value for route in self.mop_route_options} + @cached_property + def cleaning_mode_options(self) -> list[CleaningModes]: + return get_cleaning_mode_options(self._device_features_trait) + @property def fan_speed_name(self) -> str | None: if self.fan_power is None: @@ -91,3 +100,29 @@ def mop_route_name(self) -> str | None: if self.mop_mode is None: return None return self.mop_route_mapping.get(self.mop_mode) + + @property + def current_cleaning_mode(self) -> CleaningModes | None: + return get_current_cleaning_mode( + clean_mode=self.fan_power, + water_mode=self.water_box_mode, + mop_mode=self.mop_mode, + features=self._device_features_trait, + ) + + @property + def cleaning_mode_name(self) -> str | None: + if (cleaning_mode := self.current_cleaning_mode) is None: + return None + return cleaning_mode.value + + def get_cleaning_mode_parameters(self, cleaning_mode: str | CleaningModes) -> list[dict[str, int]]: + """Get the RPC payload for the selected high-level cleaning mode.""" + return get_cleaning_mode_parameters(cleaning_mode, self._device_features_trait) + + async def set_cleaning_mode(self, cleaning_mode: str | CleaningModes) -> None: + """Set the high-level cleaning mode.""" + await self.rpc_channel.send_command( + RoborockCommand.SET_CLEAN_MOTOR_MODE, + params=self.get_cleaning_mode_parameters(cleaning_mode), + ) diff --git a/tests/devices/__snapshots__/test_v1_device.ambr b/tests/devices/__snapshots__/test_v1_device.ambr index 8da5d44c..84c285a5 100644 --- a/tests/devices/__snapshots__/test_v1_device.ambr +++ b/tests/devices/__snapshots__/test_v1_device.ambr @@ -870,7 +870,7 @@ }) # --- # name: test_device_trait_command_parsing[status] - StatusTrait(adbumper_status=None, auto_dust_collection=None, avoid_count=None, back_type=None, battery=100, camera_status=None, charge_status=None, clean_area=91287500, clean_fluid_status=None, clean_percent=None, clean_time=5405, clear_water_box_status=None, collision_avoid_status=None, command=, common_status=None, converter=DefaultConverter, corner_clean_mode=None, current_map=0, debug_mode=None, dirty_water_box_status=None, distance_off=0, dnd_enabled=1, dock_cool_fan_status=None, dock_error_status=None, dock_type=None, dry_status=None, dss=None, dust_bag_status=None, dust_collection_status=None, error_code=, error_code_name='none', fan_power=106, fan_speed_mapping={101: 'quiet', 102: 'balanced', 103: 'turbo', 104: 'max', 108: 'max_plus', 105: 'off', 106: 'custom'}, fan_speed_name='custom', fan_speed_options=[, , , , , , ], hatch_door_status=None, home_sec_enable_password=None, home_sec_status=None, in_cleaning=, in_fresh_state=1, in_returning=0, in_warmup=None, is_exploring=None, is_locating=0, kct=None, lab_status=1, last_clean_t=None, lock_status=0, map_present=1, map_status=3, mop_forbidden_enable=0, mop_mode=None, mop_route_mapping={300: 'standard', 301: 'deep', 302: 'custom'}, mop_route_name=None, mop_route_options=[, , ], msg_seq=515, msg_ver=2, rdt=None, repeat=None, replenish_mode=None, rss=None, square_meter_clean_area=91.3, state=, state_name='charging', subdivision_sets=None, switch_map_mode=None, unsave_map_flag=0, unsave_map_reason=4, wash_phase=None, wash_ready=None, wash_status=None, water_box_carriage_status=0, water_box_filter_status=None, water_box_mode=204, water_box_status=0, water_mode_mapping={200: 'off', 201: 'mild', 202: 'standard', 203: 'intense', 207: 'custom_water_flow', 204: 'custom'}, water_mode_name='custom', water_mode_options=[, , , , , ], water_shortage_status=None) + StatusTrait(adbumper_status=None, auto_dust_collection=None, avoid_count=None, back_type=None, battery=100, camera_status=None, charge_status=None, clean_area=91287500, clean_fluid_status=None, clean_percent=None, clean_time=5405, cleaning_mode_name='custom', cleaning_mode_options=[, , , ], clear_water_box_status=None, collision_avoid_status=None, command=, common_status=None, converter=DefaultConverter, corner_clean_mode=None, current_cleaning_mode=, current_map=0, debug_mode=None, dirty_water_box_status=None, distance_off=0, dnd_enabled=1, dock_cool_fan_status=None, dock_error_status=None, dock_type=None, dry_status=None, dss=None, dust_bag_status=None, dust_collection_status=None, error_code=, error_code_name='none', fan_power=106, fan_speed_mapping={101: 'quiet', 102: 'balanced', 103: 'turbo', 104: 'max', 108: 'max_plus', 105: 'off', 106: 'custom'}, fan_speed_name='custom', fan_speed_options=[, , , , , , ], hatch_door_status=None, home_sec_enable_password=None, home_sec_status=None, in_cleaning=, in_fresh_state=1, in_returning=0, in_warmup=None, is_exploring=None, is_locating=0, kct=None, lab_status=1, last_clean_t=None, lock_status=0, map_present=1, map_status=3, mop_forbidden_enable=0, mop_mode=None, mop_route_mapping={300: 'standard', 301: 'deep', 302: 'custom'}, mop_route_name=None, mop_route_options=[, , ], msg_seq=515, msg_ver=2, rdt=None, repeat=None, replenish_mode=None, rss=None, square_meter_clean_area=91.3, state=, state_name='charging', subdivision_sets=None, switch_map_mode=None, unsave_map_flag=0, unsave_map_reason=4, wash_phase=None, wash_ready=None, wash_status=None, water_box_carriage_status=0, water_box_filter_status=None, water_box_mode=204, water_box_status=0, water_mode_mapping={200: 'off', 201: 'mild', 202: 'standard', 203: 'intense', 207: 'custom_water_flow', 204: 'custom'}, water_mode_name='custom', water_mode_options=[, , , , , ], water_shortage_status=None) # --- # name: test_device_trait_command_parsing[status].1 dict({ diff --git a/tests/devices/traits/v1/test_status.py b/tests/devices/traits/v1/test_status.py index a308dbca..52c3f64c 100644 --- a/tests/devices/traits/v1/test_status.py +++ b/tests/devices/traits/v1/test_status.py @@ -5,6 +5,7 @@ import pytest +from roborock import CleaningModes, CleanRoutes, VacuumModes, WaterModes from roborock.data import SHORT_MODEL_TO_ENUM from roborock.data.v1 import ( RoborockStateCode, @@ -13,7 +14,7 @@ from roborock.devices.device import RoborockDevice from roborock.devices.traits.v1.device_features import DeviceFeaturesTrait from roborock.devices.traits.v1.status import StatusTrait -from roborock.exceptions import RoborockException +from roborock.exceptions import RoborockException, RoborockUnsupportedFeature from roborock.roborock_typing import RoborockCommand from tests import mock_data from tests.mock_data import STATUS @@ -26,6 +27,23 @@ def status_trait(device: RoborockDevice) -> StatusTrait: return device.v1_properties.status +def _create_cleaning_mode_status_trait(**feature_overrides: bool) -> StatusTrait: + """Create a status trait with mop-capable V1 features for cleaning mode tests.""" + short_model = mock_data.A27_PRODUCT_DATA["model"].split(".")[-1] + features = DeviceFeatures.from_feature_flags( + new_feature_info=0, + new_feature_info_str="", + feature_info=[], + product_nickname=SHORT_MODEL_TO_ENUM[short_model], + ) + features.is_support_water_mode = True + features.is_pure_clean_mop_supported = True + features.is_customized_clean_supported = True + for feature_name, value in feature_overrides.items(): + setattr(features, feature_name, value) + return StatusTrait(cast(DeviceFeaturesTrait, features), region="us") + + async def test_refresh_status(status_trait: StatusTrait, mock_rpc_channel: AsyncMock) -> None: """Test successfully refreshing status.""" mock_rpc_channel.send_command.return_value = [STATUS] @@ -87,6 +105,215 @@ def test_options(status_trait: StatusTrait) -> None: assert len(status_trait.mop_route_options) > 0 +def test_cleaning_mode_options() -> None: + """Test the high-level cleaning mode options for the device.""" + status_trait = _create_cleaning_mode_status_trait() + assert status_trait.cleaning_mode_options == [ + CleaningModes.VACUUM, + CleaningModes.VAC_AND_MOP, + CleaningModes.MOP, + CleaningModes.CUSTOM, + ] + + +@pytest.mark.parametrize( + ("fan_power", "water_box_mode", "mop_mode", "expected_mode"), + [ + ( + VacuumModes.BALANCED.code, + WaterModes.STANDARD.code, + CleanRoutes.STANDARD.code, + CleaningModes.VAC_AND_MOP, + ), + ( + VacuumModes.BALANCED.code, + WaterModes.OFF.code, + CleanRoutes.STANDARD.code, + CleaningModes.VACUUM, + ), + ( + VacuumModes.OFF.code, + WaterModes.STANDARD.code, + CleanRoutes.STANDARD.code, + CleaningModes.MOP, + ), + ( + VacuumModes.CUSTOMIZED.code, + WaterModes.STANDARD.code, + CleanRoutes.STANDARD.code, + CleaningModes.CUSTOM, + ), + ( + VacuumModes.BALANCED.code, + WaterModes.SMART_MODE.code, + CleanRoutes.STANDARD.code, + CleaningModes.SMART_MODE, + ), + ], +) +def test_current_cleaning_mode( + fan_power: int, + water_box_mode: int, + mop_mode: int, + expected_mode: CleaningModes, +) -> None: + """Test the current high-level cleaning mode classification.""" + status_trait = _create_cleaning_mode_status_trait(is_smart_clean_mode_set_supported=True) + status_trait.fan_power = fan_power + status_trait.water_box_mode = water_box_mode + status_trait.mop_mode = mop_mode + + assert status_trait.current_cleaning_mode == expected_mode + assert status_trait.cleaning_mode_name == expected_mode.value + + +def test_current_cleaning_mode_with_brush_up_mop() -> None: + """Test brush-up mop-only classification on supported devices.""" + status_trait = _create_cleaning_mode_status_trait(is_support_main_brush_up_down_supported=True) + status_trait.fan_power = VacuumModes.OFF_RAISE_MAIN_BRUSH.code + status_trait.water_box_mode = WaterModes.STANDARD.code + status_trait.mop_mode = CleanRoutes.STANDARD.code + + assert status_trait.current_cleaning_mode == CleaningModes.MOP + + +def test_current_cleaning_mode_none() -> None: + """Test that incomplete status values do not classify a cleaning mode.""" + status_trait = _create_cleaning_mode_status_trait() + status_trait.fan_power = None + assert status_trait.current_cleaning_mode is None + assert status_trait.cleaning_mode_name is None + + +def test_current_cleaning_mode_without_mop_route_status() -> None: + """Test older V1 devices can classify cleaning mode without mop route status.""" + status_trait = _create_cleaning_mode_status_trait( + is_clean_route_setting_supported=False, + is_customized_clean_supported=False, + ) + status_trait.fan_power = VacuumModes.BALANCED.code + status_trait.water_box_mode = WaterModes.OFF.code + status_trait.mop_mode = None + + assert status_trait.current_cleaning_mode == CleaningModes.VACUUM + + +def test_get_cleaning_mode_parameters() -> None: + """Test payload generation for supported high-level cleaning modes.""" + status_trait = _create_cleaning_mode_status_trait() + assert status_trait.get_cleaning_mode_parameters(CleaningModes.VACUUM) == [ + { + "fan_power": VacuumModes.BALANCED.code, + "water_box_mode": WaterModes.OFF.code, + "mop_mode": CleanRoutes.STANDARD.code, + } + ] + assert status_trait.get_cleaning_mode_parameters("custom") == [ + { + "fan_power": VacuumModes.CUSTOMIZED.code, + "water_box_mode": WaterModes.CUSTOMIZED.code, + "mop_mode": CleanRoutes.CUSTOMIZED.code, + } + ] + + +def test_get_cleaning_mode_parameters_unsupported() -> None: + """Test unsupported cleaning modes raise a clear error.""" + status_trait = _create_cleaning_mode_status_trait() + with pytest.raises(RoborockUnsupportedFeature, match="not supported"): + status_trait.get_cleaning_mode_parameters(CleaningModes.SMART_MODE) + + +def test_get_cleaning_mode_parameters_invalid_name() -> None: + """Test invalid cleaning mode names raise a Roborock exception.""" + status_trait = _create_cleaning_mode_status_trait() + with pytest.raises(RoborockUnsupportedFeature, match="not supported"): + status_trait.get_cleaning_mode_parameters("invalid_mode") + + +async def test_set_cleaning_mode( + mock_rpc_channel: AsyncMock, +) -> None: + """Test setting the high-level cleaning mode.""" + status_trait = _create_cleaning_mode_status_trait() + status_trait._rpc_channel = mock_rpc_channel # type: ignore[assignment] + await status_trait.set_cleaning_mode(CleaningModes.CUSTOM) + + mock_rpc_channel.send_command.assert_called_once_with( + RoborockCommand.SET_CLEAN_MOTOR_MODE, + params=[ + { + "fan_power": VacuumModes.CUSTOMIZED.code, + "water_box_mode": WaterModes.CUSTOMIZED.code, + "mop_mode": CleanRoutes.CUSTOMIZED.code, + } + ], + ) + + +def test_cleaning_mode_options_with_smart_mode() -> None: + """Test SmartPlan support is reflected in the available options.""" + status_trait = _create_cleaning_mode_status_trait(is_smart_clean_mode_set_supported=True) + + assert status_trait.cleaning_mode_options == [ + CleaningModes.VACUUM, + CleaningModes.VAC_AND_MOP, + CleaningModes.MOP, + CleaningModes.CUSTOM, + CleaningModes.SMART_MODE, + ] + + +def test_get_cleaning_mode_parameters_with_brush_up_mop() -> None: + """Test mop-only uses the brush-up mode when supported.""" + status_trait = _create_cleaning_mode_status_trait(is_support_main_brush_up_down_supported=True) + + assert status_trait.get_cleaning_mode_parameters(CleaningModes.MOP) == [ + { + "fan_power": VacuumModes.OFF_RAISE_MAIN_BRUSH.code, + "water_box_mode": WaterModes.STANDARD.code, + "mop_mode": CleanRoutes.STANDARD.code, + } + ] + + +def test_get_cleaning_mode_parameters_without_clean_route_setting() -> None: + """Test older V1 devices use the 2-field clean motor payload.""" + status_trait = _create_cleaning_mode_status_trait( + is_clean_route_setting_supported=False, + is_customized_clean_supported=False, + ) + + assert status_trait.get_cleaning_mode_parameters(CleaningModes.VACUUM) == [ + { + "fan_power": VacuumModes.BALANCED.code, + "water_box_mode": WaterModes.OFF.code, + } + ] + assert status_trait.get_cleaning_mode_parameters(CleaningModes.VAC_AND_MOP) == [ + { + "fan_power": VacuumModes.BALANCED.code, + "water_box_mode": WaterModes.STANDARD.code, + } + ] + assert status_trait.get_cleaning_mode_parameters(CleaningModes.MOP) == [ + { + "fan_power": VacuumModes.OFF.code, + "water_box_mode": WaterModes.STANDARD.code, + } + ] + + +def test_current_cleaning_mode_gentle_not_mop_without_pure_mop() -> None: + """Test code 105 is not treated as mop-only on devices without pure mop.""" + status_trait = _create_cleaning_mode_status_trait(is_pure_clean_mop_supported=False) + status_trait.fan_power = VacuumModes.GENTLE.code + status_trait.water_box_mode = WaterModes.STANDARD.code + status_trait.mop_mode = CleanRoutes.STANDARD.code + + assert status_trait.current_cleaning_mode == CleaningModes.VAC_AND_MOP + + def test_water_slide_mode_mapping() -> None: """Test feature-aware water mode mapping for water slide mode devices.""" short_model = mock_data.A114_PRODUCT_DATA["model"].split(".")[-1] From c29c38f875ccfce0b53627182798cff3b4f968b4 Mon Sep 17 00:00:00 2001 From: Luke Date: Sat, 25 Apr 2026 09:24:59 -0400 Subject: [PATCH 2/5] chore: address comments from copilot --- roborock/data/v1/v1_clean_modes.py | 11 ++++++----- tests/devices/traits/v1/test_status.py | 1 + 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/roborock/data/v1/v1_clean_modes.py b/roborock/data/v1/v1_clean_modes.py index af0f791a..ab52882b 100644 --- a/roborock/data/v1/v1_clean_modes.py +++ b/roborock/data/v1/v1_clean_modes.py @@ -253,11 +253,12 @@ def get_current_cleaning_mode( return CleaningModes.SMART_MODE if is_mode_customized(clean_mode, water_mode, mop_mode): return CleaningModes.CUSTOM - try: - if clean_mode == get_mop_only_vacuum_mode(features).code: - return CleaningModes.MOP - except RoborockUnsupportedFeature: - pass + if water_mode != WaterModes.OFF.code: + try: + if clean_mode == get_mop_only_vacuum_mode(features).code: + return CleaningModes.MOP + except RoborockUnsupportedFeature: + pass if water_mode == WaterModes.OFF.code: return CleaningModes.VACUUM return CleaningModes.VAC_AND_MOP diff --git a/tests/devices/traits/v1/test_status.py b/tests/devices/traits/v1/test_status.py index 52c3f64c..ee079995 100644 --- a/tests/devices/traits/v1/test_status.py +++ b/tests/devices/traits/v1/test_status.py @@ -39,6 +39,7 @@ def _create_cleaning_mode_status_trait(**feature_overrides: bool) -> StatusTrait features.is_support_water_mode = True features.is_pure_clean_mop_supported = True features.is_customized_clean_supported = True + features.is_clean_route_setting_supported = True for feature_name, value in feature_overrides.items(): setattr(features, feature_name, value) return StatusTrait(cast(DeviceFeaturesTrait, features), region="us") From 8bb33cc9a45e395515db8dd391b1730a254c233d Mon Sep 17 00:00:00 2001 From: Luke Date: Sat, 25 Apr 2026 09:41:09 -0400 Subject: [PATCH 3/5] chore: address comments from copilot --- roborock/data/v1/v1_clean_modes.py | 40 +++++++++++++++++--------- roborock/devices/traits/v1/status.py | 10 +++---- tests/devices/traits/v1/test_status.py | 29 +++++++++++++++++++ 3 files changed, 61 insertions(+), 18 deletions(-) diff --git a/roborock/data/v1/v1_clean_modes.py b/roborock/data/v1/v1_clean_modes.py index ab52882b..e84e5224 100644 --- a/roborock/data/v1/v1_clean_modes.py +++ b/roborock/data/v1/v1_clean_modes.py @@ -207,12 +207,32 @@ def get_mop_only_vacuum_mode(features: DeviceFeatures) -> VacuumModes: return VacuumModes.OFF -_CLEAN_MOTOR_MODE_PARAMS: dict[CleaningModes, tuple[int, int, int]] = { - CleaningModes.VACUUM: (VacuumModes.BALANCED.code, WaterModes.OFF.code, CleanRoutes.STANDARD.code), - CleaningModes.VAC_AND_MOP: (VacuumModes.BALANCED.code, WaterModes.STANDARD.code, CleanRoutes.STANDARD.code), - CleaningModes.CUSTOM: (VacuumModes.CUSTOMIZED.code, WaterModes.CUSTOMIZED.code, CleanRoutes.CUSTOMIZED.code), - CleaningModes.SMART_MODE: (VacuumModes.SMART_MODE.code, WaterModes.SMART_MODE.code, CleanRoutes.SMART_MODE.code), -} +def _get_default_mopping_water_code(features: DeviceFeatures) -> int: + """Pick a sensible default water code when mopping for the device.""" + # Water-slide devices use a disjoint set of water codes; pick a mid-flow + # slide code instead of the standard 202, which they don't accept. + if features.is_water_slide_mode_supported: + return WaterModes.PURE_WATER_FLOW_MIDDLE.code + return WaterModes.STANDARD.code + + +def _get_clean_motor_mode_params(mode: CleaningModes, features: DeviceFeatures) -> tuple[int, int, int]: + """Return (fan_power, water_box_mode, mop_mode) codes for the high-level mode.""" + if mode == CleaningModes.VACUUM: + return (VacuumModes.BALANCED.code, WaterModes.OFF.code, CleanRoutes.STANDARD.code) + if mode == CleaningModes.VAC_AND_MOP: + return (VacuumModes.BALANCED.code, _get_default_mopping_water_code(features), CleanRoutes.STANDARD.code) + if mode == CleaningModes.MOP: + return ( + get_mop_only_vacuum_mode(features).code, + _get_default_mopping_water_code(features), + CleanRoutes.STANDARD.code, + ) + if mode == CleaningModes.CUSTOM: + return (VacuumModes.CUSTOMIZED.code, WaterModes.CUSTOMIZED.code, CleanRoutes.CUSTOMIZED.code) + if mode == CleaningModes.SMART_MODE: + return (VacuumModes.SMART_MODE.code, WaterModes.SMART_MODE.code, CleanRoutes.SMART_MODE.code) + raise RoborockUnsupportedFeature(f"Cleaning mode {mode.value!r} is not supported") def get_cleaning_mode_parameters(cleaning_mode: str | CleaningModes, features: DeviceFeatures) -> list[dict[str, int]]: @@ -224,13 +244,7 @@ def get_cleaning_mode_parameters(cleaning_mode: str | CleaningModes, features: D if mode not in get_cleaning_mode_options(features): raise RoborockUnsupportedFeature(f"Cleaning mode {mode.value!r} is not supported") - if mode == CleaningModes.MOP: - fan_power = get_mop_only_vacuum_mode(features).code - water_box_mode = WaterModes.STANDARD.code - mop_mode = CleanRoutes.STANDARD.code - else: - fan_power, water_box_mode, mop_mode = _CLEAN_MOTOR_MODE_PARAMS[mode] - + fan_power, water_box_mode, mop_mode = _get_clean_motor_mode_params(mode, features) params: dict[str, int] = {"fan_power": fan_power, "water_box_mode": water_box_mode} if features.is_clean_route_setting_supported: params["mop_mode"] = mop_mode diff --git a/roborock/devices/traits/v1/status.py b/roborock/devices/traits/v1/status.py index bc581c6d..b31b32f2 100644 --- a/roborock/devices/traits/v1/status.py +++ b/roborock/devices/traits/v1/status.py @@ -38,11 +38,11 @@ class StatusTrait(StatusV2, common.V1TraitMixin): - Water Mode - Mop Route - You should call the _options() version of the attribute to know which are - supported for your device (i.e. fan_speed_options()) - Then you can call the _mapping to convert an int value to the actual Enum. - (i.e. fan_speed_mapping()) - You can call the _name property to get the str value of the enum. (i.e. fan_speed_name) + You should use the _options version of the attribute to know which are + supported for your device (i.e. fan_speed_options) + Then you can use the _mapping to convert an int value to the actual Enum. + (i.e. fan_speed_mapping) + You can use the _name property to get the str value of the enum. (i.e. fan_speed_name) """ diff --git a/tests/devices/traits/v1/test_status.py b/tests/devices/traits/v1/test_status.py index ee079995..816dbb18 100644 --- a/tests/devices/traits/v1/test_status.py +++ b/tests/devices/traits/v1/test_status.py @@ -41,6 +41,8 @@ def _create_cleaning_mode_status_trait(**feature_overrides: bool) -> StatusTrait features.is_customized_clean_supported = True features.is_clean_route_setting_supported = True for feature_name, value in feature_overrides.items(): + if not hasattr(features, feature_name): + raise AttributeError(f"Unknown DeviceFeatures override: {feature_name}") setattr(features, feature_name, value) return StatusTrait(cast(DeviceFeaturesTrait, features), region="us") @@ -305,6 +307,33 @@ def test_get_cleaning_mode_parameters_without_clean_route_setting() -> None: ] +def test_get_cleaning_mode_parameters_water_slide_device() -> None: + """Water-slide devices should use a slide-compatible water code, not 202.""" + status_trait = _create_cleaning_mode_status_trait(is_water_slide_mode_supported=True) + + assert status_trait.get_cleaning_mode_parameters(CleaningModes.VACUUM) == [ + { + "fan_power": VacuumModes.BALANCED.code, + "water_box_mode": WaterModes.OFF.code, + "mop_mode": CleanRoutes.STANDARD.code, + } + ] + assert status_trait.get_cleaning_mode_parameters(CleaningModes.VAC_AND_MOP) == [ + { + "fan_power": VacuumModes.BALANCED.code, + "water_box_mode": WaterModes.PURE_WATER_FLOW_MIDDLE.code, + "mop_mode": CleanRoutes.STANDARD.code, + } + ] + assert status_trait.get_cleaning_mode_parameters(CleaningModes.MOP) == [ + { + "fan_power": VacuumModes.OFF.code, + "water_box_mode": WaterModes.PURE_WATER_FLOW_MIDDLE.code, + "mop_mode": CleanRoutes.STANDARD.code, + } + ] + + def test_current_cleaning_mode_gentle_not_mop_without_pure_mop() -> None: """Test code 105 is not treated as mop-only on devices without pure mop.""" status_trait = _create_cleaning_mode_status_trait(is_pure_clean_mop_supported=False) From 68797de525b7b7056b4e0924178570ed2696d7d3 Mon Sep 17 00:00:00 2001 From: Luke Date: Sat, 25 Apr 2026 10:45:24 -0400 Subject: [PATCH 4/5] chore: address comments from copilot --- tests/devices/traits/v1/test_status.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/devices/traits/v1/test_status.py b/tests/devices/traits/v1/test_status.py index 816dbb18..dec04d52 100644 --- a/tests/devices/traits/v1/test_status.py +++ b/tests/devices/traits/v1/test_status.py @@ -228,7 +228,7 @@ def test_get_cleaning_mode_parameters_unsupported() -> None: def test_get_cleaning_mode_parameters_invalid_name() -> None: - """Test invalid cleaning mode names raise a Roborock exception.""" + """Test invalid cleaning mode names raise RoborockUnsupportedFeature.""" status_trait = _create_cleaning_mode_status_trait() with pytest.raises(RoborockUnsupportedFeature, match="not supported"): status_trait.get_cleaning_mode_parameters("invalid_mode") From ed4e3c83d6582645cb86b2682f7898b068f14f9e Mon Sep 17 00:00:00 2001 From: Luke Date: Sun, 26 Apr 2026 20:21:57 -0400 Subject: [PATCH 5/5] chore: address comments --- roborock/data/v1/v1_clean_modes.py | 151 ++++++++++++------ roborock/devices/traits/v1/status.py | 19 +-- .../devices/__snapshots__/test_v1_device.ambr | 2 +- tests/devices/traits/v1/test_status.py | 105 ++++++++---- 4 files changed, 181 insertions(+), 96 deletions(-) diff --git a/roborock/data/v1/v1_clean_modes.py b/roborock/data/v1/v1_clean_modes.py index e84e5224..cf1337cf 100644 --- a/roborock/data/v1/v1_clean_modes.py +++ b/roborock/data/v1/v1_clean_modes.py @@ -2,6 +2,7 @@ import typing from enum import StrEnum +from typing import TypeVar from ...exceptions import RoborockUnsupportedFeature from ..code_mappings import RoborockModeEnum @@ -70,7 +71,16 @@ class WashTowelModes(RoborockModeEnum): SUPER_DEEP = ("super_deep", 8) -class CleaningModes(StrEnum): +class CleaningMode(StrEnum): + """High-level cleaning intent derived from the lower-level motor settings. + + Prefer this abstraction when you want to present or switch between the + user-facing cleaning behaviors exposed by the app. The lower-level + `VacuumModes`, `WaterModes`, and `CleanRoutes` enums are still useful for + advanced tuning, but most integrations should treat them as implementation + details of a single high-level cleaning mode. + """ + VACUUM = "vacuum" VAC_AND_MOP = "vac_and_mop" MOP = "mop" @@ -88,6 +98,8 @@ class CleaningModes(StrEnum): 250: WaterModes.PURE_WATER_FLOW_END, } +ModeEnumT = TypeVar("ModeEnumT", bound=RoborockModeEnum) + def get_wash_towel_modes(features: DeviceFeatures) -> list[WashTowelModes]: """Get the valid wash towel modes for the device""" @@ -184,18 +196,25 @@ def get_water_mode_mapping(features: DeviceFeatures) -> dict[int, str]: return {mode.code: mode.value for mode in get_water_modes(features)} -def get_cleaning_mode_options(features: DeviceFeatures) -> list[CleaningModes]: - """Get the supported high-level cleaning modes for the device.""" +def get_cleaning_mode_options(features: DeviceFeatures) -> list[CleaningMode]: + """Return the supported high-level cleaning modes for the device. + + These options are the preferred user-facing choices because they bundle the + correct fan, water, and mop-route settings together for the device. Callers + should generally present these instead of mixing lower-level mode enums + unless they explicitly need fine-grained control. + """ if not features.is_support_water_mode: return [] - options = [CleaningModes.VACUUM, CleaningModes.VAC_AND_MOP] + supported_water_modes = get_water_modes(features) + options = [CleaningMode.VACUUM, CleaningMode.VAC_AND_MOP] if features.is_pure_clean_mop_supported: - options.append(CleaningModes.MOP) - if features.is_customized_clean_supported: - options.append(CleaningModes.CUSTOM) - if features.is_smart_clean_mode_set_supported: - options.append(CleaningModes.SMART_MODE) + options.append(CleaningMode.MOP) + if features.is_customized_clean_supported and WaterModes.CUSTOMIZED in supported_water_modes: + options.append(CleaningMode.CUSTOM) + if features.is_smart_clean_mode_set_supported and WaterModes.SMART_MODE in supported_water_modes: + options.append(CleaningMode.SMART_MODE) return options @@ -207,75 +226,107 @@ def get_mop_only_vacuum_mode(features: DeviceFeatures) -> VacuumModes: return VacuumModes.OFF -def _get_default_mopping_water_code(features: DeviceFeatures) -> int: - """Pick a sensible default water code when mopping for the device.""" +def _get_default_mopping_water_mode(features: DeviceFeatures) -> WaterModes: + """Pick a sensible default water mode when mopping for the device.""" # Water-slide devices use a disjoint set of water codes; pick a mid-flow # slide code instead of the standard 202, which they don't accept. if features.is_water_slide_mode_supported: - return WaterModes.PURE_WATER_FLOW_MIDDLE.code - return WaterModes.STANDARD.code + return WaterModes.PURE_WATER_FLOW_MIDDLE + return WaterModes.STANDARD -def _get_clean_motor_mode_params(mode: CleaningModes, features: DeviceFeatures) -> tuple[int, int, int]: - """Return (fan_power, water_box_mode, mop_mode) codes for the high-level mode.""" - if mode == CleaningModes.VACUUM: - return (VacuumModes.BALANCED.code, WaterModes.OFF.code, CleanRoutes.STANDARD.code) - if mode == CleaningModes.VAC_AND_MOP: - return (VacuumModes.BALANCED.code, _get_default_mopping_water_code(features), CleanRoutes.STANDARD.code) - if mode == CleaningModes.MOP: +def _get_clean_motor_mode_params( + mode: CleaningMode, + features: DeviceFeatures, +) -> tuple[VacuumModes, WaterModes, CleanRoutes]: + """Return (fan_power, water_box_mode, mop_mode) enums for the high-level mode.""" + if mode == CleaningMode.VACUUM: + return (VacuumModes.BALANCED, WaterModes.OFF, CleanRoutes.STANDARD) + if mode == CleaningMode.VAC_AND_MOP: + return (VacuumModes.BALANCED, _get_default_mopping_water_mode(features), CleanRoutes.STANDARD) + if mode == CleaningMode.MOP: return ( - get_mop_only_vacuum_mode(features).code, - _get_default_mopping_water_code(features), - CleanRoutes.STANDARD.code, + get_mop_only_vacuum_mode(features), + _get_default_mopping_water_mode(features), + CleanRoutes.STANDARD, ) - if mode == CleaningModes.CUSTOM: - return (VacuumModes.CUSTOMIZED.code, WaterModes.CUSTOMIZED.code, CleanRoutes.CUSTOMIZED.code) - if mode == CleaningModes.SMART_MODE: - return (VacuumModes.SMART_MODE.code, WaterModes.SMART_MODE.code, CleanRoutes.SMART_MODE.code) + if mode == CleaningMode.CUSTOM: + return (VacuumModes.CUSTOMIZED, WaterModes.CUSTOMIZED, CleanRoutes.CUSTOMIZED) + if mode == CleaningMode.SMART_MODE: + return (VacuumModes.SMART_MODE, WaterModes.SMART_MODE, CleanRoutes.SMART_MODE) raise RoborockUnsupportedFeature(f"Cleaning mode {mode.value!r} is not supported") -def get_cleaning_mode_parameters(cleaning_mode: str | CleaningModes, features: DeviceFeatures) -> list[dict[str, int]]: - """Get the RPC payload for switching the high-level cleaning mode.""" +def resolve_cleaning_mode(cleaning_mode: str | CleaningMode) -> CleaningMode: + """Resolve a string or enum into a CleaningMode value.""" + if isinstance(cleaning_mode, CleaningMode): + return cleaning_mode try: - mode = CleaningModes(cleaning_mode) + return CleaningMode(cleaning_mode) except ValueError as err: raise RoborockUnsupportedFeature(f"Cleaning mode {cleaning_mode!r} is not supported") from err - if mode not in get_cleaning_mode_options(features): - raise RoborockUnsupportedFeature(f"Cleaning mode {mode.value!r} is not supported") - fan_power, water_box_mode, mop_mode = _get_clean_motor_mode_params(mode, features) - params: dict[str, int] = {"fan_power": fan_power, "water_box_mode": water_box_mode} + +def get_cleaning_mode_parameters(cleaning_mode: CleaningMode, features: DeviceFeatures) -> list[dict[str, int]]: + """Get the RPC payload for switching the high-level cleaning mode.""" + if cleaning_mode not in get_cleaning_mode_options(features): + raise RoborockUnsupportedFeature(f"Cleaning mode {cleaning_mode.value!r} is not supported") + + fan_power, water_box_mode, mop_mode = _get_clean_motor_mode_params(cleaning_mode, features) + params: dict[str, int] = {"fan_power": fan_power.code, "water_box_mode": water_box_mode.code} if features.is_clean_route_setting_supported: - params["mop_mode"] = mop_mode + params["mop_mode"] = mop_mode.code return [params] +def _resolve_mode_code(value: int | ModeEnumT | None, mode_cls: type[ModeEnumT]) -> ModeEnumT | None: + """Resolve a raw code or enum into a RoborockModeEnum.""" + if value is None: + return None + if isinstance(value, mode_cls): + return value + return mode_cls.from_code_optional(int(value)) + + +def _resolve_clean_mode(value: int | VacuumModes | None, features: DeviceFeatures) -> VacuumModes | None: + """Resolve a vacuum mode code, accounting for feature-specific code aliases.""" + if value is None or isinstance(value, VacuumModes): + return value + if value == VacuumModes.OFF.code: + if features.is_pure_clean_mop_supported: + return get_mop_only_vacuum_mode(features) + return VacuumModes.GENTLE + return VacuumModes.from_code_optional(value) + + def get_current_cleaning_mode( - clean_mode: int | None, - water_mode: int | None, - mop_mode: int | None, + clean_mode: int | VacuumModes | None, + water_mode: int | WaterModes | None, + mop_mode: int | CleanRoutes | None, features: DeviceFeatures, -) -> CleaningModes | None: +) -> CleaningMode | None: """Classify the current high-level cleaning mode from individual mode codes.""" if not features.is_support_water_mode: return None - if clean_mode is None or water_mode is None: + clean_mode_enum = _resolve_clean_mode(clean_mode, features) + water_mode_enum = _resolve_mode_code(water_mode, WaterModes) + mop_mode_enum = _resolve_mode_code(mop_mode, CleanRoutes) + if clean_mode_enum is None or water_mode_enum is None: return None - if is_smart_mode_set(water_mode, clean_mode, mop_mode): - return CleaningModes.SMART_MODE - if is_mode_customized(clean_mode, water_mode, mop_mode): - return CleaningModes.CUSTOM - if water_mode != WaterModes.OFF.code: + if is_smart_mode_set(water_mode_enum, clean_mode_enum, mop_mode_enum): + return CleaningMode.SMART_MODE + if is_mode_customized(clean_mode_enum, water_mode_enum, mop_mode_enum): + return CleaningMode.CUSTOM + if water_mode_enum != WaterModes.OFF: try: - if clean_mode == get_mop_only_vacuum_mode(features).code: - return CleaningModes.MOP + if clean_mode_enum == get_mop_only_vacuum_mode(features): + return CleaningMode.MOP except RoborockUnsupportedFeature: pass - if water_mode == WaterModes.OFF.code: - return CleaningModes.VACUUM - return CleaningModes.VAC_AND_MOP + if water_mode_enum == WaterModes.OFF: + return CleaningMode.VACUUM + return CleaningMode.VAC_AND_MOP def is_mode_customized( diff --git a/roborock/devices/traits/v1/status.py b/roborock/devices/traits/v1/status.py index b31b32f2..f3f9368a 100644 --- a/roborock/devices/traits/v1/status.py +++ b/roborock/devices/traits/v1/status.py @@ -1,7 +1,7 @@ from functools import cached_property from roborock import ( - CleaningModes, + CleaningMode, CleanRoutes, StatusV2, VacuumModes, @@ -13,6 +13,7 @@ get_current_cleaning_mode, get_water_mode_mapping, get_water_modes, + resolve_cleaning_mode, ) from roborock.roborock_typing import RoborockCommand @@ -80,7 +81,7 @@ def mop_route_mapping(self) -> dict[int, str]: return {route.code: route.value for route in self.mop_route_options} @cached_property - def cleaning_mode_options(self) -> list[CleaningModes]: + def cleaning_mode_options(self) -> list[CleaningMode]: return get_cleaning_mode_options(self._device_features_trait) @property @@ -102,7 +103,7 @@ def mop_route_name(self) -> str | None: return self.mop_route_mapping.get(self.mop_mode) @property - def current_cleaning_mode(self) -> CleaningModes | None: + def current_cleaning_mode(self) -> CleaningMode | None: return get_current_cleaning_mode( clean_mode=self.fan_power, water_mode=self.water_box_mode, @@ -111,18 +112,14 @@ def current_cleaning_mode(self) -> CleaningModes | None: ) @property - def cleaning_mode_name(self) -> str | None: + def current_cleaning_mode_name(self) -> str | None: if (cleaning_mode := self.current_cleaning_mode) is None: return None return cleaning_mode.value - def get_cleaning_mode_parameters(self, cleaning_mode: str | CleaningModes) -> list[dict[str, int]]: - """Get the RPC payload for the selected high-level cleaning mode.""" - return get_cleaning_mode_parameters(cleaning_mode, self._device_features_trait) - - async def set_cleaning_mode(self, cleaning_mode: str | CleaningModes) -> None: - """Set the high-level cleaning mode.""" + async def set_cleaning_mode(self, cleaning_mode: str | CleaningMode) -> None: + """Set the preferred high-level cleaning mode for the device.""" await self.rpc_channel.send_command( RoborockCommand.SET_CLEAN_MOTOR_MODE, - params=self.get_cleaning_mode_parameters(cleaning_mode), + params=get_cleaning_mode_parameters(resolve_cleaning_mode(cleaning_mode), self._device_features_trait), ) diff --git a/tests/devices/__snapshots__/test_v1_device.ambr b/tests/devices/__snapshots__/test_v1_device.ambr index 84c285a5..8ac141bd 100644 --- a/tests/devices/__snapshots__/test_v1_device.ambr +++ b/tests/devices/__snapshots__/test_v1_device.ambr @@ -870,7 +870,7 @@ }) # --- # name: test_device_trait_command_parsing[status] - StatusTrait(adbumper_status=None, auto_dust_collection=None, avoid_count=None, back_type=None, battery=100, camera_status=None, charge_status=None, clean_area=91287500, clean_fluid_status=None, clean_percent=None, clean_time=5405, cleaning_mode_name='custom', cleaning_mode_options=[, , , ], clear_water_box_status=None, collision_avoid_status=None, command=, common_status=None, converter=DefaultConverter, corner_clean_mode=None, current_cleaning_mode=, current_map=0, debug_mode=None, dirty_water_box_status=None, distance_off=0, dnd_enabled=1, dock_cool_fan_status=None, dock_error_status=None, dock_type=None, dry_status=None, dss=None, dust_bag_status=None, dust_collection_status=None, error_code=, error_code_name='none', fan_power=106, fan_speed_mapping={101: 'quiet', 102: 'balanced', 103: 'turbo', 104: 'max', 108: 'max_plus', 105: 'off', 106: 'custom'}, fan_speed_name='custom', fan_speed_options=[, , , , , , ], hatch_door_status=None, home_sec_enable_password=None, home_sec_status=None, in_cleaning=, in_fresh_state=1, in_returning=0, in_warmup=None, is_exploring=None, is_locating=0, kct=None, lab_status=1, last_clean_t=None, lock_status=0, map_present=1, map_status=3, mop_forbidden_enable=0, mop_mode=None, mop_route_mapping={300: 'standard', 301: 'deep', 302: 'custom'}, mop_route_name=None, mop_route_options=[, , ], msg_seq=515, msg_ver=2, rdt=None, repeat=None, replenish_mode=None, rss=None, square_meter_clean_area=91.3, state=, state_name='charging', subdivision_sets=None, switch_map_mode=None, unsave_map_flag=0, unsave_map_reason=4, wash_phase=None, wash_ready=None, wash_status=None, water_box_carriage_status=0, water_box_filter_status=None, water_box_mode=204, water_box_status=0, water_mode_mapping={200: 'off', 201: 'mild', 202: 'standard', 203: 'intense', 207: 'custom_water_flow', 204: 'custom'}, water_mode_name='custom', water_mode_options=[, , , , , ], water_shortage_status=None) + StatusTrait(adbumper_status=None, auto_dust_collection=None, avoid_count=None, back_type=None, battery=100, camera_status=None, charge_status=None, clean_area=91287500, clean_fluid_status=None, clean_percent=None, clean_time=5405, cleaning_mode_options=[, , , ], clear_water_box_status=None, collision_avoid_status=None, command=, common_status=None, converter=DefaultConverter, corner_clean_mode=None, current_cleaning_mode=, current_cleaning_mode_name='custom', current_map=0, debug_mode=None, dirty_water_box_status=None, distance_off=0, dnd_enabled=1, dock_cool_fan_status=None, dock_error_status=None, dock_type=None, dry_status=None, dss=None, dust_bag_status=None, dust_collection_status=None, error_code=, error_code_name='none', fan_power=106, fan_speed_mapping={101: 'quiet', 102: 'balanced', 103: 'turbo', 104: 'max', 108: 'max_plus', 105: 'off', 106: 'custom'}, fan_speed_name='custom', fan_speed_options=[, , , , , , ], hatch_door_status=None, home_sec_enable_password=None, home_sec_status=None, in_cleaning=, in_fresh_state=1, in_returning=0, in_warmup=None, is_exploring=None, is_locating=0, kct=None, lab_status=1, last_clean_t=None, lock_status=0, map_present=1, map_status=3, mop_forbidden_enable=0, mop_mode=None, mop_route_mapping={300: 'standard', 301: 'deep', 302: 'custom'}, mop_route_name=None, mop_route_options=[, , ], msg_seq=515, msg_ver=2, rdt=None, repeat=None, replenish_mode=None, rss=None, square_meter_clean_area=91.3, state=, state_name='charging', subdivision_sets=None, switch_map_mode=None, unsave_map_flag=0, unsave_map_reason=4, wash_phase=None, wash_ready=None, wash_status=None, water_box_carriage_status=0, water_box_filter_status=None, water_box_mode=204, water_box_status=0, water_mode_mapping={200: 'off', 201: 'mild', 202: 'standard', 203: 'intense', 207: 'custom_water_flow', 204: 'custom'}, water_mode_name='custom', water_mode_options=[, , , , , ], water_shortage_status=None) # --- # name: test_device_trait_command_parsing[status].1 dict({ diff --git a/tests/devices/traits/v1/test_status.py b/tests/devices/traits/v1/test_status.py index dec04d52..254e882d 100644 --- a/tests/devices/traits/v1/test_status.py +++ b/tests/devices/traits/v1/test_status.py @@ -5,7 +5,15 @@ import pytest -from roborock import CleaningModes, CleanRoutes, VacuumModes, WaterModes +from roborock import ( + CleaningMode, + CleanRoutes, + VacuumModes, + WaterModes, + get_cleaning_mode_parameters, + get_current_cleaning_mode, + resolve_cleaning_mode, +) from roborock.data import SHORT_MODEL_TO_ENUM from roborock.data.v1 import ( RoborockStateCode, @@ -112,10 +120,10 @@ def test_cleaning_mode_options() -> None: """Test the high-level cleaning mode options for the device.""" status_trait = _create_cleaning_mode_status_trait() assert status_trait.cleaning_mode_options == [ - CleaningModes.VACUUM, - CleaningModes.VAC_AND_MOP, - CleaningModes.MOP, - CleaningModes.CUSTOM, + CleaningMode.VACUUM, + CleaningMode.VAC_AND_MOP, + CleaningMode.MOP, + CleaningMode.CUSTOM, ] @@ -126,31 +134,31 @@ def test_cleaning_mode_options() -> None: VacuumModes.BALANCED.code, WaterModes.STANDARD.code, CleanRoutes.STANDARD.code, - CleaningModes.VAC_AND_MOP, + CleaningMode.VAC_AND_MOP, ), ( VacuumModes.BALANCED.code, WaterModes.OFF.code, CleanRoutes.STANDARD.code, - CleaningModes.VACUUM, + CleaningMode.VACUUM, ), ( VacuumModes.OFF.code, WaterModes.STANDARD.code, CleanRoutes.STANDARD.code, - CleaningModes.MOP, + CleaningMode.MOP, ), ( VacuumModes.CUSTOMIZED.code, WaterModes.STANDARD.code, CleanRoutes.STANDARD.code, - CleaningModes.CUSTOM, + CleaningMode.CUSTOM, ), ( VacuumModes.BALANCED.code, WaterModes.SMART_MODE.code, CleanRoutes.STANDARD.code, - CleaningModes.SMART_MODE, + CleaningMode.SMART_MODE, ), ], ) @@ -158,7 +166,7 @@ def test_current_cleaning_mode( fan_power: int, water_box_mode: int, mop_mode: int, - expected_mode: CleaningModes, + expected_mode: CleaningMode, ) -> None: """Test the current high-level cleaning mode classification.""" status_trait = _create_cleaning_mode_status_trait(is_smart_clean_mode_set_supported=True) @@ -167,7 +175,7 @@ def test_current_cleaning_mode( status_trait.mop_mode = mop_mode assert status_trait.current_cleaning_mode == expected_mode - assert status_trait.cleaning_mode_name == expected_mode.value + assert status_trait.current_cleaning_mode_name == expected_mode.value def test_current_cleaning_mode_with_brush_up_mop() -> None: @@ -177,7 +185,22 @@ def test_current_cleaning_mode_with_brush_up_mop() -> None: status_trait.water_box_mode = WaterModes.STANDARD.code status_trait.mop_mode = CleanRoutes.STANDARD.code - assert status_trait.current_cleaning_mode == CleaningModes.MOP + assert status_trait.current_cleaning_mode == CleaningMode.MOP + + +def test_current_cleaning_mode_accepts_enums() -> None: + """Test direct enum inputs are resolved before classification.""" + status_trait = _create_cleaning_mode_status_trait(is_smart_clean_mode_set_supported=True) + + assert ( + get_current_cleaning_mode( + clean_mode=VacuumModes.BALANCED, + water_mode=WaterModes.SMART_MODE, + mop_mode=CleanRoutes.STANDARD, + features=status_trait._device_features_trait, + ) + == CleaningMode.SMART_MODE + ) def test_current_cleaning_mode_none() -> None: @@ -185,7 +208,7 @@ def test_current_cleaning_mode_none() -> None: status_trait = _create_cleaning_mode_status_trait() status_trait.fan_power = None assert status_trait.current_cleaning_mode is None - assert status_trait.cleaning_mode_name is None + assert status_trait.current_cleaning_mode_name is None def test_current_cleaning_mode_without_mop_route_status() -> None: @@ -198,20 +221,20 @@ def test_current_cleaning_mode_without_mop_route_status() -> None: status_trait.water_box_mode = WaterModes.OFF.code status_trait.mop_mode = None - assert status_trait.current_cleaning_mode == CleaningModes.VACUUM + assert status_trait.current_cleaning_mode == CleaningMode.VACUUM def test_get_cleaning_mode_parameters() -> None: """Test payload generation for supported high-level cleaning modes.""" status_trait = _create_cleaning_mode_status_trait() - assert status_trait.get_cleaning_mode_parameters(CleaningModes.VACUUM) == [ + assert get_cleaning_mode_parameters(CleaningMode.VACUUM, status_trait._device_features_trait) == [ { "fan_power": VacuumModes.BALANCED.code, "water_box_mode": WaterModes.OFF.code, "mop_mode": CleanRoutes.STANDARD.code, } ] - assert status_trait.get_cleaning_mode_parameters("custom") == [ + assert get_cleaning_mode_parameters(resolve_cleaning_mode("custom"), status_trait._device_features_trait) == [ { "fan_power": VacuumModes.CUSTOMIZED.code, "water_box_mode": WaterModes.CUSTOMIZED.code, @@ -224,14 +247,13 @@ def test_get_cleaning_mode_parameters_unsupported() -> None: """Test unsupported cleaning modes raise a clear error.""" status_trait = _create_cleaning_mode_status_trait() with pytest.raises(RoborockUnsupportedFeature, match="not supported"): - status_trait.get_cleaning_mode_parameters(CleaningModes.SMART_MODE) + get_cleaning_mode_parameters(CleaningMode.SMART_MODE, status_trait._device_features_trait) def test_get_cleaning_mode_parameters_invalid_name() -> None: """Test invalid cleaning mode names raise RoborockUnsupportedFeature.""" - status_trait = _create_cleaning_mode_status_trait() with pytest.raises(RoborockUnsupportedFeature, match="not supported"): - status_trait.get_cleaning_mode_parameters("invalid_mode") + resolve_cleaning_mode("invalid_mode") async def test_set_cleaning_mode( @@ -240,7 +262,7 @@ async def test_set_cleaning_mode( """Test setting the high-level cleaning mode.""" status_trait = _create_cleaning_mode_status_trait() status_trait._rpc_channel = mock_rpc_channel # type: ignore[assignment] - await status_trait.set_cleaning_mode(CleaningModes.CUSTOM) + await status_trait.set_cleaning_mode(CleaningMode.CUSTOM) mock_rpc_channel.send_command.assert_called_once_with( RoborockCommand.SET_CLEAN_MOTOR_MODE, @@ -259,11 +281,11 @@ def test_cleaning_mode_options_with_smart_mode() -> None: status_trait = _create_cleaning_mode_status_trait(is_smart_clean_mode_set_supported=True) assert status_trait.cleaning_mode_options == [ - CleaningModes.VACUUM, - CleaningModes.VAC_AND_MOP, - CleaningModes.MOP, - CleaningModes.CUSTOM, - CleaningModes.SMART_MODE, + CleaningMode.VACUUM, + CleaningMode.VAC_AND_MOP, + CleaningMode.MOP, + CleaningMode.CUSTOM, + CleaningMode.SMART_MODE, ] @@ -271,7 +293,7 @@ def test_get_cleaning_mode_parameters_with_brush_up_mop() -> None: """Test mop-only uses the brush-up mode when supported.""" status_trait = _create_cleaning_mode_status_trait(is_support_main_brush_up_down_supported=True) - assert status_trait.get_cleaning_mode_parameters(CleaningModes.MOP) == [ + assert get_cleaning_mode_parameters(CleaningMode.MOP, status_trait._device_features_trait) == [ { "fan_power": VacuumModes.OFF_RAISE_MAIN_BRUSH.code, "water_box_mode": WaterModes.STANDARD.code, @@ -287,19 +309,19 @@ def test_get_cleaning_mode_parameters_without_clean_route_setting() -> None: is_customized_clean_supported=False, ) - assert status_trait.get_cleaning_mode_parameters(CleaningModes.VACUUM) == [ + assert get_cleaning_mode_parameters(CleaningMode.VACUUM, status_trait._device_features_trait) == [ { "fan_power": VacuumModes.BALANCED.code, "water_box_mode": WaterModes.OFF.code, } ] - assert status_trait.get_cleaning_mode_parameters(CleaningModes.VAC_AND_MOP) == [ + assert get_cleaning_mode_parameters(CleaningMode.VAC_AND_MOP, status_trait._device_features_trait) == [ { "fan_power": VacuumModes.BALANCED.code, "water_box_mode": WaterModes.STANDARD.code, } ] - assert status_trait.get_cleaning_mode_parameters(CleaningModes.MOP) == [ + assert get_cleaning_mode_parameters(CleaningMode.MOP, status_trait._device_features_trait) == [ { "fan_power": VacuumModes.OFF.code, "water_box_mode": WaterModes.STANDARD.code, @@ -311,21 +333,21 @@ def test_get_cleaning_mode_parameters_water_slide_device() -> None: """Water-slide devices should use a slide-compatible water code, not 202.""" status_trait = _create_cleaning_mode_status_trait(is_water_slide_mode_supported=True) - assert status_trait.get_cleaning_mode_parameters(CleaningModes.VACUUM) == [ + assert get_cleaning_mode_parameters(CleaningMode.VACUUM, status_trait._device_features_trait) == [ { "fan_power": VacuumModes.BALANCED.code, "water_box_mode": WaterModes.OFF.code, "mop_mode": CleanRoutes.STANDARD.code, } ] - assert status_trait.get_cleaning_mode_parameters(CleaningModes.VAC_AND_MOP) == [ + assert get_cleaning_mode_parameters(CleaningMode.VAC_AND_MOP, status_trait._device_features_trait) == [ { "fan_power": VacuumModes.BALANCED.code, "water_box_mode": WaterModes.PURE_WATER_FLOW_MIDDLE.code, "mop_mode": CleanRoutes.STANDARD.code, } ] - assert status_trait.get_cleaning_mode_parameters(CleaningModes.MOP) == [ + assert get_cleaning_mode_parameters(CleaningMode.MOP, status_trait._device_features_trait) == [ { "fan_power": VacuumModes.OFF.code, "water_box_mode": WaterModes.PURE_WATER_FLOW_MIDDLE.code, @@ -334,6 +356,21 @@ def test_get_cleaning_mode_parameters_water_slide_device() -> None: ] +def test_cleaning_mode_options_water_slide_device() -> None: + """Water-slide devices should not expose unsupported custom or smart water modes.""" + status_trait = _create_cleaning_mode_status_trait( + is_water_slide_mode_supported=True, + is_customized_clean_supported=True, + is_smart_clean_mode_set_supported=True, + ) + + assert status_trait.cleaning_mode_options == [ + CleaningMode.VACUUM, + CleaningMode.VAC_AND_MOP, + CleaningMode.MOP, + ] + + def test_current_cleaning_mode_gentle_not_mop_without_pure_mop() -> None: """Test code 105 is not treated as mop-only on devices without pure mop.""" status_trait = _create_cleaning_mode_status_trait(is_pure_clean_mop_supported=False) @@ -341,7 +378,7 @@ def test_current_cleaning_mode_gentle_not_mop_without_pure_mop() -> None: status_trait.water_box_mode = WaterModes.STANDARD.code status_trait.mop_mode = CleanRoutes.STANDARD.code - assert status_trait.current_cleaning_mode == CleaningModes.VAC_AND_MOP + assert status_trait.current_cleaning_mode == CleaningMode.VAC_AND_MOP def test_water_slide_mode_mapping() -> None: