feat(v1): add set_clean_then_mop and document how it relates to CleaningMode

Author: klowdo

roborock/devices/traits/v1/status.py

@@ -18,6 +18,7 @@
     resolve_cleaning_mode,
 )
 from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
+from roborock.exceptions import RoborockUnsupportedFeature
 from roborock.roborock_message import RoborockDataProtocol
 from roborock.roborock_typing import RoborockCommand
 
@@ -48,6 +49,7 @@ class StatusTrait(StatusV2, common.V1TraitMixin, TraitUpdateListener):
     - Fan Speed
     - Water Mode
     - Mop Route
+    - Clean then mop (``clean_then_mop``, independent of the cleaning mode)
 
     You should use the _options version of the attribute to know which are
     supported for your device (i.e. fan_speed_options)
@@ -115,10 +117,18 @@ def mop_route_name(self) -> str | None:
 
     @property
     def clean_then_mop(self) -> bool | None:
-        """Whether the current run vacuums each room fully before mopping it.
+        """Whether each room is vacuumed fully before it is mopped.
 
-        Reported by the device as ``seq_type``. It describes the run in
-        progress; it is not a persisted setting and cannot be set directly.
+        Reported by the device as ``seq_type`` and persisted on the device: it
+        tracks the "vacuum then mop" toggle in the vendor app, also while docked.
+
+        This is a separate axis from ``current_cleaning_mode``. The cleaning
+        mode is derived from ``fan_power``, ``water_box_mode`` and ``mop_mode``
+        and describes *what* the robot does in a room (vacuum, mop, both). This
+        flag describes the *order* when it does both. Changing one leaves the
+        other untouched; set it with ``set_clean_then_mop``.
+
+        ``None`` means the device does not support the feature.
         """
         if self.seq_type is None:
             return None
@@ -163,6 +173,27 @@ async def resolve_error(self, error_code: int | None = None) -> None:
         await self.rpc_channel.send_command(RoborockCommand.RESOLVE_ERROR, params={"error_code": error_code})
         await self.refresh()
 
+    async def set_clean_then_mop(self, enabled: bool) -> None:
+        """Set whether each room is vacuumed fully before it is mopped.
+
+        Independent of ``set_cleaning_mode``. The firmware requires the current
+        motor-mode values in this request, so they are echoed back unchanged.
+        """
+        if not self._device_features_trait.is_clean_then_mop_mode_supported:
+            raise RoborockUnsupportedFeature("Clean then mop is not supported")
+        seq_type = _SEQ_TYPE_CLEAN_THEN_MOP if enabled else 0
+        params: dict[str, int | None] = {
+            "type": seq_type,
+            "fan_power": self.fan_power,
+            "water_box_mode": self.water_box_mode,
+        }
+        if self.mop_mode is not None:
+            params["mop_mode"] = self.mop_mode
+        if enabled and self._device_features_trait.is_ctm_with_repeat_supported:
+            params["repeat"] = 1
+        await self.rpc_channel.send_command(RoborockCommand.APP_SET_CLEAN_SEQUENCE_TYPE, params=params)
+        self.seq_type = seq_type
+
     def update_from_dps(self, decoded_dps: dict[RoborockDataProtocol, Any]) -> None:
         """Update the trait from data protocol push message data.
 

roborock/roborock_typing.py

@@ -34,6 +34,7 @@ class RoborockCommand(str, Enum):
     APP_SET_CROSS_CARPET_CLEANING_STATUS = "app_set_cross_carpet_cleaning_status"
     APP_SET_DOOR_SILL_BLOCKS = "app_set_door_sill_blocks"
     APP_SET_DIRTY_REPLENISH_CLEAN_STATUS = "app_set_dirty_replenish_clean_status"
+    APP_SET_CLEAN_SEQUENCE_TYPE = "app_set_clean_sequence_type"
     APP_SET_DRYER_SETTING = "app_set_dryer_setting"
     APP_SET_DRYER_STATUS = "app_set_dryer_status"
     APP_SET_DYNAMIC_CONFIG = "app_set_dynamic_config"

tests/devices/traits/v1/test_status.py

@@ -702,3 +702,47 @@ def test_status_parses_seq_type() -> None:
     status = StatusV2.from_dict({**STATUS, "seq_type": 1})
 
     assert status.seq_type == 1
+
+
+@pytest.mark.parametrize(
+    ("enabled", "ctm_with_repeat", "expected_params"),
+    [
+        (True, True, {"type": 1, "fan_power": 102, "water_box_mode": 202, "mop_mode": 300, "repeat": 1}),
+        (True, False, {"type": 1, "fan_power": 102, "water_box_mode": 202, "mop_mode": 300}),
+        (False, True, {"type": 0, "fan_power": 102, "water_box_mode": 202, "mop_mode": 300}),
+    ],
+)
+async def test_set_clean_then_mop(
+    mock_rpc_channel: AsyncMock,
+    enabled: bool,
+    ctm_with_repeat: bool,
+    expected_params: dict[str, int],
+) -> None:
+    """The setter echoes the current motor modes and updates seq_type optimistically."""
+    status_trait = _create_cleaning_mode_status_trait(
+        is_clean_then_mop_mode_supported=True,
+        is_ctm_with_repeat_supported=ctm_with_repeat,
+    )
+    status_trait._rpc_channel = mock_rpc_channel  # type: ignore[assignment]
+    status_trait.fan_power = 102
+    status_trait.water_box_mode = 202
+    status_trait.mop_mode = 300
+    status_trait.seq_type = 0 if enabled else 1
+
+    await status_trait.set_clean_then_mop(enabled)
+
+    mock_rpc_channel.send_command.assert_called_once_with(
+        RoborockCommand.APP_SET_CLEAN_SEQUENCE_TYPE, params=expected_params
+    )
+    assert status_trait.clean_then_mop is enabled
+
+
+async def test_set_clean_then_mop_unsupported(mock_rpc_channel: AsyncMock) -> None:
+    """Devices without the feature reject the setter before sending anything."""
+    status_trait = _create_cleaning_mode_status_trait(is_clean_then_mop_mode_supported=False)
+    status_trait._rpc_channel = mock_rpc_channel  # type: ignore[assignment]
+
+    with pytest.raises(RoborockUnsupportedFeature):
+        await status_trait.set_clean_then_mop(True)
+
+    mock_rpc_channel.send_command.assert_not_called()