Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 18 additions & 3 deletions roborock/data/v1/v1_clean_modes.py
Original file line number Diff line number Diff line change
Expand Up @@ -199,14 +199,20 @@ def get_water_modes(features: DeviceFeatures) -> list[WaterModes]:

def get_water_slide_mode_name(code: int) -> str | None:
"""Get the app's range name for a water slide code, or None if out of range."""
mode = _resolve_water_slide_mode(code)
return mode.value if mode is not None else None


def _resolve_water_slide_mode(code: int) -> WaterModes | None:
"""Resolve a raw slider code to the water mode for its app range."""
if code == 200:
return WaterModes.OFF.value
return WaterModes.OFF
position = code - 220
if position < 1:
return None
for last_position, mode in _WATER_SLIDE_RANGES:
if position <= last_position:
return mode.value
return mode
return None


Expand Down Expand Up @@ -300,7 +306,14 @@ def get_cleaning_mode_parameters(cleaning_mode: CleaningMode, features: DeviceFe
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}
water_code = water_box_mode.code
if features.is_water_slide_mode_supported:
# Aliased water enums retain legacy codes; the device mapping preserves
# slide-specific wire values without changing the user-facing names.
water_code = next(
code for code, name in get_water_mode_mapping(features).items() if name == water_box_mode.value
)
params: dict[str, int] = {"fan_power": fan_power.code, "water_box_mode": water_code}
if features.is_clean_route_setting_supported:
params["mop_mode"] = mop_mode.code
return [params]
Expand Down Expand Up @@ -337,6 +350,8 @@ def get_current_cleaning_mode(
return None
clean_mode_enum = _resolve_clean_mode(clean_mode, features)
water_mode_enum = _resolve_mode_code(water_mode, WaterModes)
if features.is_water_slide_mode_supported and isinstance(water_mode, int):
water_mode_enum = _resolve_water_slide_mode(water_mode) or water_mode_enum
mop_mode_enum = _resolve_mode_code(mop_mode, CleanRoutes)
if clean_mode_enum is None or water_mode_enum is None:
return None
Expand Down
126 changes: 122 additions & 4 deletions tests/devices/traits/v1/test_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -404,7 +404,7 @@ def test_current_cleaning_mode_with_main_brush_lift() -> None:
is_water_slide_mode_supported=True,
)
status_trait.fan_power = VacuumModes.OFF.code
status_trait.water_box_mode = WaterModes.PURE_WATER_FLOW_MIDDLE.code
status_trait.water_box_mode = 235
status_trait.mop_mode = CleanRoutes.STANDARD.code

assert status_trait.current_cleaning_mode == CleaningMode.MOP
Expand Down Expand Up @@ -521,7 +521,7 @@ def test_get_cleaning_mode_parameters_qrevo_edge_2() -> None:
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,
"water_box_mode": 235,
"mop_mode": CleanRoutes.STANDARD.code,
}
]
Expand Down Expand Up @@ -570,19 +570,137 @@ def test_get_cleaning_mode_parameters_water_slide_device() -> None:
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,
"water_box_mode": 235,
"mop_mode": CleanRoutes.STANDARD.code,
}
]
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,
"water_box_mode": 235,
"mop_mode": CleanRoutes.STANDARD.code,
}
]


@pytest.mark.parametrize("water_box_mode", range(221, 251))
@pytest.mark.parametrize(
("fan_power", "expected_mode"),
[(102, CleaningMode.VAC_AND_MOP), (104, CleaningMode.VAC_AND_MOP), (105, CleaningMode.MOP)],
)
def test_current_cleaning_mode_water_slide_codes(
water_box_mode: int, fan_power: int, expected_mode: CleaningMode
) -> None:
"""Classify every raw slider position without changing the reported water code."""
status_trait = _create_cleaning_mode_status_trait(is_water_slide_mode_supported=True)
status_trait.fan_power = fan_power
status_trait.water_box_mode = water_box_mode
status_trait.mop_mode = 303

assert status_trait.current_cleaning_mode == expected_mode
assert status_trait.current_cleaning_mode_name == expected_mode.value
assert status_trait.water_box_mode == water_box_mode


@pytest.mark.parametrize(
("water_box_mode", "expected_mode"),
[
(200, CleaningMode.VACUUM),
(204, CleaningMode.CUSTOM),
(209, CleaningMode.SMART_MODE),
(220, None),
(251, None),
(999, None),
(None, None),
],
)
def test_current_cleaning_mode_water_slide_legacy_codes(
water_box_mode: int | None, expected_mode: CleaningMode | None
) -> None:
"""Keep legacy and unknown status decoding on water-slide devices."""
status_trait = _create_cleaning_mode_status_trait(is_water_slide_mode_supported=True)
status_trait.fan_power = 102
status_trait.water_box_mode = water_box_mode
status_trait.mop_mode = 300

assert status_trait.current_cleaning_mode == expected_mode
assert status_trait.current_cleaning_mode_name == (expected_mode.value if expected_mode is not None else None)
assert status_trait.water_box_mode == water_box_mode


@pytest.mark.parametrize("water_box_mode", [222, 223, 248])
def test_current_cleaning_mode_without_water_slide_support(water_box_mode: int) -> None:
"""Do not interpret slider positions without the water-slide capability."""
status_trait = _create_cleaning_mode_status_trait(is_water_slide_mode_supported=False)
status_trait.fan_power = 102
status_trait.water_box_mode = water_box_mode
status_trait.mop_mode = 300

assert status_trait.current_cleaning_mode is None
assert status_trait.current_cleaning_mode_name is None
assert status_trait.water_box_mode == water_box_mode


@pytest.mark.parametrize("is_water_slide_mode_supported", [False, True])
@pytest.mark.parametrize(
("water_box_mode", "expected_mode"),
[
(200, CleaningMode.VACUUM),
(201, CleaningMode.VAC_AND_MOP),
(202, CleaningMode.VAC_AND_MOP),
(203, CleaningMode.VAC_AND_MOP),
(None, None),
(999, None),
],
)
def test_current_cleaning_mode_water_code_fallback(
is_water_slide_mode_supported: bool, water_box_mode: int | None, expected_mode: CleaningMode | None
) -> None:
"""Preserve legacy water status codes and unknown-code fallback for both device types."""
status_trait = _create_cleaning_mode_status_trait(is_water_slide_mode_supported=is_water_slide_mode_supported)
status_trait.fan_power = 102
status_trait.water_box_mode = water_box_mode
status_trait.mop_mode = 300

assert status_trait.current_cleaning_mode == expected_mode


@pytest.mark.parametrize("water_mode", [WaterModes.MEDIUM, WaterModes.PURE_WATER_FLOW_MIDDLE])
def test_current_cleaning_mode_water_slide_accepts_enums(water_mode: WaterModes) -> None:
"""Enum inputs remain valid even when they alias a legacy water mode."""
status_trait = _create_cleaning_mode_status_trait(is_water_slide_mode_supported=True)
assert (
get_current_cleaning_mode(102, water_mode, 303, status_trait._device_features_trait) == CleaningMode.VAC_AND_MOP
)


@pytest.mark.parametrize("is_clean_route_setting_supported", [False, True])
@pytest.mark.parametrize(
("cleaning_mode", "fan_power", "water_box_mode"),
[("vacuum", 102, 200), ("vac_and_mop", 102, 235), ("mop", 105, 235)],
)
async def test_set_cleaning_mode_water_slide_wire_payload(
mock_rpc_channel: AsyncMock,
is_clean_route_setting_supported: bool,
cleaning_mode: str,
fan_power: int,
water_box_mode: int,
) -> None:
"""Assert the public trait emits valid slide RPC integers, with optional mop route."""
status_trait = _create_cleaning_mode_status_trait(
is_water_slide_mode_supported=True,
is_clean_route_setting_supported=is_clean_route_setting_supported,
)
status_trait._rpc_channel = mock_rpc_channel # type: ignore[assignment]

await status_trait.set_cleaning_mode(cleaning_mode)

expected = {"fan_power": fan_power, "water_box_mode": water_box_mode}
if is_clean_route_setting_supported:
expected["mop_mode"] = 300
mock_rpc_channel.send_command.assert_called_once_with(RoborockCommand.SET_CLEAN_MOTOR_MODE, params=[expected])


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(
Expand Down
Loading