feat(dyad): add F25 Ultra setting and self-clean commands

roborock/devices/traits/a01/__init__.py

@@ -96,6 +96,13 @@ def _decode_dyad_schedule(value: str) -> list[DyadScheduleTask]:
     return [DyadScheduleTask.from_dict(task) for task in json.loads(value)]
 
 
+# The F25 Ultra app pairs each suction level (eco, standard, turbo) with a brush speed.
+_F25_SUCTION_BRUSH_SPEED = {
+    DyadSuction.l2: DyadBrushSpeed.l1,
+    DyadSuction.l3: DyadBrushSpeed.l2,
+    DyadSuction.l4: DyadBrushSpeed.l2,
+}
+
 DYAD_PROTOCOL_ENTRIES: dict[RoborockDyadDataProtocol, Callable] = {
     RoborockDyadDataProtocol.STATUS: lambda val: RoborockDyadStateCode(val).name,
     RoborockDyadDataProtocol.SELF_CLEAN_MODE: lambda val: DyadSelfCleanMode(val).name,
@@ -338,6 +345,98 @@ async def stop_drying(self) -> None:
         """Stop a running drying cycle."""
         await send_decoded_command(self._channel, {RoborockDyadDataProtocol.START: 0})
 
+    async def set_drying_mode(self, mode: DyadDryingMode) -> None:
+        """Set the drying mode on the F25 Ultra.
+
+        Do not use this on older Dyad models, where the same data point is the
+        silent drying duration in minutes. The device reports the selected mode
+        on `DRY_CUR_MODE`.
+        """
+        if mode == DyadDryingMode.unknown:
+            raise ValueError("Cannot set drying mode to unknown")
+        await send_decoded_command(self._channel, {RoborockDyadDataProtocol.SILENT_DRY_DURATION: int(mode)})
+
+    async def set_self_clean_intensity(self, intensity: DyadSelfCleanIntensity) -> None:
+        """Set the self-clean intensity on the F25 Ultra.
+
+        The F25 Ultra stores this setting across three data points, and smart differs
+        from standard only in `AUTO_SMART_SELF_CLEAN_SET`. Do not use this on other
+        models. `SELF_CLEAN_CUR_MODE` reports the current or last run, not this setting.
+        """
+        if intensity == DyadSelfCleanIntensity.unknown:
+            raise ValueError("Cannot set self-clean intensity to unknown")
+        level = DyadSelfCleanLevel.deep if intensity == DyadSelfCleanIntensity.deep else DyadSelfCleanLevel.normal
+        await send_decoded_command(
+            self._channel,
+            {
+                RoborockDyadDataProtocol.SELF_CLEAN_LEVEL: int(level),
+                RoborockDyadDataProtocol.AUTO_SELF_CLEAN_SET_MODE: int(level),
+                RoborockDyadDataProtocol.AUTO_SMART_SELF_CLEAN_SET: int(intensity == DyadSelfCleanIntensity.smart),
+            },
+        )
+
+    async def set_suction(self, suction: DyadSuction) -> None:
+        """Set the auto mode suction on the F25 Ultra.
+
+        Supports `l2` (eco), `l3` (standard) and `l4` (turbo). The device does not
+        change the brush speed by itself, so it is sent along with the suction.
+        """
+        await self._set_f25_suction(
+            RoborockDyadDataProtocol.SUCTION,
+            RoborockDyadDataProtocol.BRUSH_SPEED,
+            suction,
+        )
+
+    async def set_sponge_suction(self, suction: DyadSuction) -> None:
+        """Set the sponge mode suction on the F25 Ultra.
+
+        Supports `l2` (eco), `l3` (standard) and `l4` (turbo). The device does not
+        change the brush speed by itself, so it is sent along with the suction.
+        """
+        await self._set_f25_suction(
+            RoborockDyadDataProtocol.WATER_MODE_SUCTION,
+            RoborockDyadDataProtocol.WATER_MODE_BRUSH_SPEED,
+            suction,
+        )
+
+    async def _set_f25_suction(
+        self,
+        suction_protocol: RoborockDyadDataProtocol,
+        brush_protocol: RoborockDyadDataProtocol,
+        suction: DyadSuction,
+    ) -> None:
+        """Send a suction level with the brush speed the F25 Ultra app pairs with it."""
+        if (brush_speed := _F25_SUCTION_BRUSH_SPEED.get(suction)) is None:
+            raise ValueError(f"Unsupported suction level: {suction.name}")
+        await send_decoded_command(
+            self._channel,
+            {suction_protocol: int(suction), brush_protocol: int(brush_speed)},
+        )
+
+    async def start_self_clean(self) -> None:
+        """Start self-clean with the settings currently set on the F25 Ultra.
+
+        The app includes the current `SELF_CLEAN_LEVEL` in the start command, so it
+        is read from the device first. After changing the intensity, wait until the
+        device reports the new value before calling this.
+        """
+        response = await send_decoded_command(
+            self._channel,
+            {RoborockDyadDataProtocol.ID_QUERY: [RoborockDyadDataProtocol.SELF_CLEAN_LEVEL]},
+            value_encoder=json.dumps,
+        )
+        await send_decoded_command(
+            self._channel,
+            {
+                RoborockDyadDataProtocol.START: 1,
+                RoborockDyadDataProtocol.SELF_CLEAN_LEVEL: response[RoborockDyadDataProtocol.SELF_CLEAN_LEVEL],
+            },
+        )
+
+    async def stop_self_clean(self) -> None:
+        """Stop a running self-clean cycle."""
+        await send_decoded_command(self._channel, {RoborockDyadDataProtocol.START: 0})
+
     async def reset_filter(self) -> None:
         """Reset the filter usage counter."""
         await send_decoded_command(self._channel, {RoborockDyadDataProtocol.MESH_RESET: 1})

tests/devices/traits/a01/test_init.py

@@ -8,6 +8,7 @@
 from freezegun import freeze_time
 
 from roborock.data import DyadScheduleTask
+from roborock.data.dyad.dyad_code_mappings import DyadDryingMode, DyadSelfCleanIntensity, DyadSuction
 from roborock.devices.traits.a01 import DyadApi, ZeoApi, create
 from roborock.devices.traits.a01.device_feature import build_force_load_dp_list
 from roborock.roborock_message import RoborockDyadDataProtocol, RoborockMessageProtocol, RoborockZeoProtocol
@@ -520,6 +521,7 @@ async def test_dyad_schedule_task_invalid(dyad_api: DyadApi, fake_channel: FakeC
     [
         ("start_drying", {"200": 1, "204": 1}),
         ("stop_drying", {"200": 0}),
+        ("stop_self_clean", {"200": 0}),
         ("reset_filter", {"218": 1}),
         ("reset_brush", {"219": 1}),
     ],
@@ -538,6 +540,118 @@ async def test_dyad_action_commands(
     assert payload_data["dps"] == expected_dps
 
 
+@pytest.mark.parametrize(
+    ("mode", "expected_dps"),
+    [
+        (DyadDryingMode.fast, {"225": 5}),
+        (DyadDryingMode.silent, {"225": 22}),
+    ],
+)
+async def test_dyad_set_drying_mode(
+    mode: DyadDryingMode,
+    expected_dps: dict[str, int],
+    dyad_api: DyadApi,
+    fake_channel: FakeChannel,
+):
+    """The drying mode is written to the data point used by the app."""
+    await dyad_api.set_drying_mode(mode)
+
+    assert len(fake_channel.published_messages) == 1
+    payload_data = json.loads(unpad(fake_channel.published_messages[0].payload, AES.block_size))
+    assert payload_data["dps"] == expected_dps
+
+
+async def test_dyad_set_drying_mode_unknown(dyad_api: DyadApi, fake_channel: FakeChannel):
+    """The unknown placeholder is rejected and nothing is sent to the device."""
+    with pytest.raises(ValueError):
+        await dyad_api.set_drying_mode(DyadDryingMode.unknown)
+
+    assert fake_channel.published_messages == []
+
+
+@pytest.mark.parametrize(
+    ("intensity", "expected_dps"),
+    [
+        (DyadSelfCleanIntensity.smart, {"203": 1, "223": 1, "239": 1}),
+        (DyadSelfCleanIntensity.standard, {"203": 1, "223": 1, "239": 0}),
+        (DyadSelfCleanIntensity.deep, {"203": 2, "223": 2, "239": 0}),
+    ],
+)
+async def test_dyad_set_self_clean_intensity(
+    intensity: DyadSelfCleanIntensity,
+    expected_dps: dict[str, int],
+    dyad_api: DyadApi,
+    fake_channel: FakeChannel,
+):
+    """The intensity is written to all the data points the device reports it on."""
+    await dyad_api.set_self_clean_intensity(intensity)
+
+    assert len(fake_channel.published_messages) == 1
+    payload_data = json.loads(unpad(fake_channel.published_messages[0].payload, AES.block_size))
+    assert payload_data["dps"] == expected_dps
+
+
+async def test_dyad_set_self_clean_intensity_unknown(dyad_api: DyadApi, fake_channel: FakeChannel):
+    """The unknown placeholder is rejected and nothing is sent to the device."""
+    with pytest.raises(ValueError):
+        await dyad_api.set_self_clean_intensity(DyadSelfCleanIntensity.unknown)
+
+    assert fake_channel.published_messages == []
+
+
+@pytest.mark.parametrize(
+    ("method", "suction", "expected_dps"),
+    [
+        ("set_suction", DyadSuction.l2, {"206": 2, "208": 1}),
+        ("set_suction", DyadSuction.l3, {"206": 3, "208": 2}),
+        ("set_suction", DyadSuction.l4, {"206": 4, "208": 2}),
+        ("set_sponge_suction", DyadSuction.l2, {"244": 2, "245": 1}),
+        ("set_sponge_suction", DyadSuction.l3, {"244": 3, "245": 2}),
+        ("set_sponge_suction", DyadSuction.l4, {"244": 4, "245": 2}),
+    ],
+)
+async def test_dyad_set_suction(
+    method: str,
+    suction: DyadSuction,
+    expected_dps: dict[str, int],
+    dyad_api: DyadApi,
+    fake_channel: FakeChannel,
+):
+    """Suction is sent together with the brush speed the app pairs with it."""
+    await getattr(dyad_api, method)(suction)
+
+    assert len(fake_channel.published_messages) == 1
+    payload_data = json.loads(unpad(fake_channel.published_messages[0].payload, AES.block_size))
+    assert payload_data["dps"] == expected_dps
+
+
+@pytest.mark.parametrize("method", ["set_suction", "set_sponge_suction"])
+@pytest.mark.parametrize("suction", [DyadSuction.l1, DyadSuction.l5, DyadSuction.l6])
+async def test_dyad_set_suction_unsupported(
+    method: str,
+    suction: DyadSuction,
+    dyad_api: DyadApi,
+    fake_channel: FakeChannel,
+):
+    """Levels not seen on the F25 Ultra are rejected and nothing is sent."""
+    with pytest.raises(ValueError):
+        await getattr(dyad_api, method)(suction)
+
+    assert fake_channel.published_messages == []
+
+
+@pytest.mark.parametrize("level", [1, 2])
+async def test_dyad_start_self_clean(level: int, dyad_api: DyadApi, fake_channel: FakeChannel):
+    """Self-clean starts with the level currently set on the device."""
+    fake_channel.response_queue.append(build_a01_message({203: level}))
+
+    await dyad_api.start_self_clean()
+
+    assert len(fake_channel.published_messages) == 2
+    payload_data = json.loads(unpad(fake_channel.published_messages[1].payload, AES.block_size))
+    assert payload_data["dps"] == {"200": 1, "203": level}
+
+
 async def test_zeo_api_set_value(zeo_api: ZeoApi, fake_channel: FakeChannel):
     """Test ZeoApi set_value sends correct command."""
     await zeo_api.set_value(RoborockZeoProtocol.MODE, "standard")