From 0c49cd3ef350c4c84c02528fbf5056c98089fc04 Mon Sep 17 00:00:00 2001 From: infy1995 Date: Sun, 8 Nov 2026 19:04:29 +0100 Subject: [PATCH 1/2] feat(dyad): add F25 Ultra setting and self-clean commands Add typed DyadApi commands for F25 Ultra settings that span several data points or need a value from the device: - set_drying_mode writes DP 225, while the mode is read from DP 272 - set_self_clean_intensity writes DPs 203, 223 and 239 together - set_suction and set_sponge_suction send the brush speed the app pairs with each level, because the device does not change it by itself - start_self_clean sends the current DP 203 with the start command, as the app does - stop_self_clean stops a running cycle --- roborock/devices/traits/a01/__init__.py | 98 ++++++++++++++++++++ tests/devices/traits/a01/test_init.py | 114 ++++++++++++++++++++++++ 2 files changed, 212 insertions(+) diff --git a/roborock/devices/traits/a01/__init__.py b/roborock/devices/traits/a01/__init__.py index 7682bc7f..65a74b43 100644 --- a/roborock/devices/traits/a01/__init__.py +++ b/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,97 @@ 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. + """ + 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}) diff --git a/tests/devices/traits/a01/test_init.py b/tests/devices/traits/a01/test_init.py index aea824e6..a1bfa441 100644 --- a/tests/devices/traits/a01/test_init.py +++ b/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") From 208bd52c8e4e9fcd0df2047c4017c93df5baf160 Mon Sep 17 00:00:00 2001 From: infy1995 Date: Sun, 8 Nov 2026 19:35:41 +0100 Subject: [PATCH 2/2] docs(dyad): note intensity change ordering for start_self_clean --- roborock/devices/traits/a01/__init__.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/roborock/devices/traits/a01/__init__.py b/roborock/devices/traits/a01/__init__.py index 65a74b43..8d19d4f2 100644 --- a/roborock/devices/traits/a01/__init__.py +++ b/roborock/devices/traits/a01/__init__.py @@ -417,7 +417,8 @@ 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. + 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,