From 05948195e20132538564179703dfaf168e1519ff Mon Sep 17 00:00:00 2001 From: Harry Coureau Date: Sat, 7 Nov 2026 19:02:25 +0000 Subject: [PATCH 1/2] feat: add Q10 customized cleaning controls --- .../data/b01_q10/b01_q10_code_mappings.py | 30 +++ roborock/data/b01_q10/b01_q10_containers.py | 20 ++ roborock/devices/traits/b01/q10/__init__.py | 9 +- .../devices/traits/b01/q10/room_cleaning.py | 81 +++++++ roborock/devices/traits/b01/q10/vacuum.py | 37 +++- roborock/protocols/b01_q10_protocol.py | 128 ++++++++++- .../b01/q10/__snapshots__/test_status.ambr | 15 ++ .../traits/b01/q10/test_room_cleaning.py | 138 ++++++++++++ tests/devices/traits/b01/q10/test_vacuum.py | 32 +++ tests/protocols/test_b01_q10_protocol.py | 203 +++++++++++++++++- .../customer_clean_full_sanitized.b64 | 1 + 11 files changed, 687 insertions(+), 7 deletions(-) create mode 100644 roborock/devices/traits/b01/q10/room_cleaning.py create mode 100644 tests/devices/traits/b01/q10/test_room_cleaning.py create mode 100644 tests/protocols/testdata/b01_q10_protocol/customer_clean_full_sanitized.b64 diff --git a/roborock/data/b01_q10/b01_q10_code_mappings.py b/roborock/data/b01_q10/b01_q10_code_mappings.py index a35f81f24..06f8f8d88 100644 --- a/roborock/data/b01_q10/b01_q10_code_mappings.py +++ b/roborock/data/b01_q10/b01_q10_code_mappings.py @@ -152,6 +152,7 @@ class YXWaterLevel(RoborockModeEnum): class YXCleanLine(RoborockModeEnum): + UNKNOWN = "unknown", -1 FAST = "fast", 0 DAILY = "daily", 1 FINE = "fine", 2 @@ -354,3 +355,32 @@ class RemoteCommand(IntEnum): RIGHT = 3 STOP = 4 EXIT = 5 + + +class Q10RoomFanLevel(RoborockModeEnum): + """Suction levels used inside a Q10 customized-room payload.""" + + UNKNOWN = "unknown", -1 + QUIET = "quiet", 1 + BALANCED = "balanced", 2 + TURBO = "turbo", 3 + MAX = "max", 4 + MAX_PLUS = "max_plus", 5 + + +class Q10CleanCount(RoborockModeEnum): + """Number of passes for a Q10 cleaning task.""" + + UNKNOWN = "unknown", -1 + ONCE = "once", 1 + TWICE = "twice", 2 + THREE_TIMES = "three_times", 3 + + +class Q10RoomCleanType(RoborockModeEnum): + """Work performed in one Q10 customized-room setting.""" + + UNKNOWN = "unknown", -1 + VAC_AND_MOP = "vac_and_mop", 1 + VACUUM = "vacuum", 2 + MOP = "mop", 3 diff --git a/roborock/data/b01_q10/b01_q10_containers.py b/roborock/data/b01_q10/b01_q10_containers.py index 8550bd7c6..89220dbc2 100644 --- a/roborock/data/b01_q10/b01_q10_containers.py +++ b/roborock/data/b01_q10/b01_q10_containers.py @@ -12,6 +12,9 @@ from ..containers import RoborockBase from .b01_q10_code_mappings import ( B01_Q10_DP, + Q10CleanCount, + Q10RoomCleanType, + Q10RoomFanLevel, YXAreaUnit, YXBackType, YXCarpetCleanType, @@ -286,6 +289,11 @@ def fault_name(self) -> str | None: """Returns the name of the current fault.""" return self.fault.value if self.fault is not None else None + @property + def clean_count_mode(self) -> Q10CleanCount | None: + """Return the typed cleaning pass count from the reported status.""" + return Q10CleanCount.from_code_optional(self.clean_count) if self.clean_count is not None else None + @dataclass class SoundVolume(RoborockBase): @@ -342,3 +350,15 @@ class Q10NetworkInfo(RoborockBase): """ net_info: dpNetInfo | None = field(default=None, metadata={"dps": B01_Q10_DP.NET_INFO}) + + +@dataclass +class Q10RoomCleanSettings(RoborockBase): + """Writable cleaning settings for one Q10 room.""" + + room_id: int + fan_level: Q10RoomFanLevel + water_level: YXWaterLevel + clean_type: Q10RoomCleanType + clean_count: Q10CleanCount + clean_line: YXCleanLine diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index 5416b7d77..5e683d04d 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/roborock/devices/traits/b01/q10/__init__.py @@ -22,6 +22,7 @@ from .maps import MapsTrait from .network_info import NetworkInfoTrait from .remote import RemoteTrait +from .room_cleaning import RoomCleaningTrait from .status import StatusTrait from .vacuum import VacuumTrait from .volume import SoundVolumeTrait @@ -37,6 +38,7 @@ "MapContentTrait", "MapsTrait", "NetworkInfoTrait", + "RoomCleaningTrait", "SoundVolumeTrait", "StatusTrait", ] @@ -83,6 +85,9 @@ class Q10PropertiesApi(Trait): map: MapContentTrait """Composed map image plus caller-facing map and trace data.""" + room_cleaning: RoomCleaningTrait + """Customized cleaning settings and room-cleaning controls.""" + maps: MapsTrait """Saved-map list metadata.""" @@ -108,7 +113,8 @@ def __init__(self, channel: B01Q10Channel) -> None: self._map_dps = MapDpsTrait() self.maps = MapsTrait(self.command) self.map = MapContentTrait(self._map_dps, self.maps, self.command) - self.vacuum = VacuumTrait(self.command, self.status, self.map) + self.room_cleaning = RoomCleaningTrait(self.command) + self.vacuum = VacuumTrait(self.command, self.status, self.map, self.room_cleaning) self.clean_history = CleanHistoryTrait(self.command) # Read-model traits updated from the device's DPS push stream. self._updatable_traits = [ @@ -122,6 +128,7 @@ def __init__(self, channel: B01Q10Channel) -> None: self.clean_history, self._map_dps, self.maps, + self.room_cleaning, ] self._subscribe_task: asyncio.Task[None] | None = None diff --git a/roborock/devices/traits/b01/q10/room_cleaning.py b/roborock/devices/traits/b01/q10/room_cleaning.py new file mode 100644 index 000000000..6d191b5a1 --- /dev/null +++ b/roborock/devices/traits/b01/q10/room_cleaning.py @@ -0,0 +1,81 @@ +"""Customized room-cleaning controls for Q10 devices.""" + +import logging +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Any + +from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, YXCleanType, YXDeviceCleanTask +from roborock.data.b01_q10.b01_q10_containers import Q10RoomCleanSettings +from roborock.data.containers import RoborockBase +from roborock.exceptions import RoborockException +from roborock.protocols.b01_q10_protocol import decode_room_clean_settings, encode_room_clean_settings + +from .command import CommandTrait +from .common import UpdatableTrait + +_LOGGER = logging.getLogger(__name__) + + +@dataclass +class RoomCleaning(RoborockBase): + """Current customized room-cleaning state.""" + + settings: tuple[Q10RoomCleanSettings, ...] = () + settings_available: bool = False + + +class RoomCleaningTrait(RoomCleaning, UpdatableTrait): + """Read and control Q10 per-room cleaning settings.""" + + def __init__(self, command: CommandTrait) -> None: + RoomCleaning.__init__(self) + UpdatableTrait.__init__(self, command, _LOGGER) + self._command: CommandTrait = command + + def settings_for_room(self, room_id: int) -> Q10RoomCleanSettings | None: + """Return the latest reported settings for one room.""" + return next((room for room in self.settings if room.room_id == room_id), None) + + async def refresh(self) -> None: + """Request the complete customized-room settings.""" + await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CUSTOMER_CLEAN_REQUEST.code): 0}) + + async def set_settings(self, settings: Sequence[Q10RoomCleanSettings]) -> None: + """Publish customized settings, then request a complete settings push.""" + payload = encode_room_clean_settings(settings) + await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CUSTOMER_CLEAN.code): payload}) + await self.refresh() + + async def clean(self, settings: Sequence[Q10RoomCleanSettings]) -> None: + """Publish settings, select customized mode, and start the selected rooms.""" + selected = tuple(settings) + payload = encode_room_clean_settings(selected) + await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CUSTOMER_CLEAN.code): payload}) + await self._command.send(B01_Q10_DP.CLEAN_MODE, YXCleanType.CUSTOMIZED.code) + await self._command.send( + B01_Q10_DP.START_CLEAN, + {"cmd": YXDeviceCleanTask.ELECTORAL.code, "clean_paramters": [room.room_id for room in selected]}, + ) + + def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: + """Replace full settings or upsert the rooms in a compact push.""" + payload = decoded_dps.get(B01_Q10_DP.CUSTOMER_CLEAN) + if not isinstance(payload, str): + return + try: + update = decode_room_clean_settings(payload) + except RoborockException: + _LOGGER.debug("Ignoring malformed Q10 customized-room settings", exc_info=True) + return + if update.complete: + settings = update.settings + else: + by_room = {room.room_id: room for room in self.settings} + by_room.update((room.room_id, room) for room in update.settings) + settings = tuple(by_room.values()) + changed = settings != self.settings or not self.settings_available + self.settings = settings + self.settings_available = True + if changed: + self._notify_update() diff --git a/roborock/devices/traits/b01/q10/vacuum.py b/roborock/devices/traits/b01/q10/vacuum.py index 4b846576b..c1788f936 100644 --- a/roborock/devices/traits/b01/q10/vacuum.py +++ b/roborock/devices/traits/b01/q10/vacuum.py @@ -2,22 +2,26 @@ import asyncio import logging -from collections.abc import Callable +from collections.abc import Callable, Sequence from math import hypot from roborock.data.b01_q10.b01_q10_code_mappings import ( B01_Q10_DP, + Q10CleanCount, + YXCleanLine, YXCleanType, YXDeviceCleanTask, YXFanLevel, + YXWaterLevel, ) -from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint, Q10RoomCleanSettings from roborock.exceptions import RoborockException from roborock.protocols.b01_q10_protocol import CleanParams, encode_clean_params from .command import CommandTrait from .goto import GotoAction, GotoActionCommand, GotoSnapshot from .map import MapContentTrait +from .room_cleaning import RoomCleaningTrait from .status import StatusTrait _GOTO_HALF_ZONE_SIZE = 200 @@ -40,11 +44,13 @@ def __init__( command: CommandTrait, status: StatusTrait, map_content: MapContentTrait, + room_cleaning: RoomCleaningTrait, ) -> None: """Initialize the VacuumTrait.""" self._command = command self._status = status self._map = map_content + self._room_cleaning = room_cleaning self._goto_action: GotoAction | None = None self._goto_action_remove_listener: Callable[[], None] | None = None self._goto_timeout_task: asyncio.Task[None] | None = None @@ -167,6 +173,11 @@ async def clean_segments(self, segment_ids: list[int]) -> None: ) self.cancel_goto() + async def clean_rooms(self, settings: Sequence[Q10RoomCleanSettings]) -> None: + """Start a customized clean for the supplied rooms.""" + await self._room_cleaning.clean(settings) + self.cancel_goto() + async def clean_zone( self, first_corner: Q10RoborockPoint, @@ -278,6 +289,8 @@ async def empty_dustbin(self) -> None: async def set_clean_mode(self, mode: YXCleanType) -> None: """Set the cleaning mode (vacuum, mop, or both).""" + if mode is YXCleanType.UNKNOWN: + raise ValueError("mode must be a supported YXCleanType") await self._command.send( command=B01_Q10_DP.CLEAN_MODE, params=mode.code, @@ -285,7 +298,27 @@ async def set_clean_mode(self, mode: YXCleanType) -> None: async def set_fan_level(self, level: YXFanLevel) -> None: """Set the fan suction level.""" + if level is YXFanLevel.UNKNOWN: + raise ValueError("level must be a supported YXFanLevel") await self._command.send( command=B01_Q10_DP.FAN_LEVEL, params=level.code, ) + + async def set_water_level(self, level: YXWaterLevel) -> None: + """Set the mop water level.""" + if level is YXWaterLevel.UNKNOWN: + raise ValueError("level must be a supported YXWaterLevel") + await self._command.send(B01_Q10_DP.WATER_LEVEL, level.code) + + async def set_clean_count(self, count: Q10CleanCount) -> None: + """Set the number of passes for ordinary cleaning.""" + if count is Q10CleanCount.UNKNOWN: + raise ValueError("count must be a supported Q10CleanCount") + await self._command.send(B01_Q10_DP.CLEAN_COUNT, count.code) + + async def set_clean_line(self, line: YXCleanLine) -> None: + """Set the cleaning route preference.""" + if line is YXCleanLine.UNKNOWN: + raise ValueError("line must be a supported YXCleanLine") + await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CLEAN_LINE.code): line.code}) diff --git a/roborock/protocols/b01_q10_protocol.py b/roborock/protocols/b01_q10_protocol.py index a3a18fa1b..b46efe456 100644 --- a/roborock/protocols/b01_q10_protocol.py +++ b/roborock/protocols/b01_q10_protocol.py @@ -1,14 +1,24 @@ """Roborock B01 Protocol encoding and decoding.""" +import base64 +import binascii import json import logging from base64 import b64encode +from collections.abc import Sequence from dataclasses import dataclass from struct import pack from typing import Any -from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP -from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint +from roborock.data.b01_q10.b01_q10_code_mappings import ( + B01_Q10_DP, + Q10CleanCount, + Q10RoomCleanType, + Q10RoomFanLevel, + YXCleanLine, + YXWaterLevel, +) +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint, Q10RoomCleanSettings from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import ( Q10CleanRecordDetail, @@ -75,6 +85,120 @@ def encode_clean_params(params: CleanParams) -> str: return b64encode(payload).decode() +_Q10_CUSTOM_ROOM_COMPACT_SIZE = 6 +_Q10_CUSTOM_ROOM_PROPERTIES_SIZE = 26 +_Q10_CUSTOM_ROOM_NAME_SIZE = 20 + + +@dataclass(frozen=True) +class Q10RoomCleanUpdate: + """Decoded Q10 customized-room settings push.""" + + settings: tuple[Q10RoomCleanSettings, ...] + complete: bool + + +def encode_room_clean_settings(settings: Sequence[Q10RoomCleanSettings]) -> str: + """Encode the compact six-byte-per-room ``dpCustomerClean`` write.""" + selected = tuple(settings) + if not selected or len(selected) > 255: + raise ValueError("settings must contain between 1 and 255 rooms") + + payload = bytearray((len(selected),)) + room_ids: set[int] = set() + fields = ( + ("fan_level", Q10RoomFanLevel), + ("water_level", YXWaterLevel), + ("clean_type", Q10RoomCleanType), + ("clean_count", Q10CleanCount), + ("clean_line", YXCleanLine), + ) + for room in selected: + if not 0 <= room.room_id <= 255: + raise ValueError("room_id must be an integer between 0 and 255") + if room.room_id in room_ids: + raise ValueError(f"duplicate room_id: {room.room_id}") + room_ids.add(room.room_id) + values = [room.room_id] + for field_name, mode_type in fields: + value = getattr(room, field_name) + if value.code < 0: + raise ValueError(f"{field_name} must be a supported {mode_type.__name__}") + values.append(value.code) + payload.extend(values) + return base64.b64encode(payload).decode() + + +def decode_room_clean_settings(payload: str) -> Q10RoomCleanUpdate: + """Decode a compact echo or complete ``dpCustomerClean`` response. + + Complete records contain 26 property bytes, a 20-byte name block, one + vertex-count byte, then four bytes per vertex. Relevant property offsets + are room id 0, count 5, type 7, fan 8, water 9, and route 11. + """ + try: + raw = base64.b64decode(payload, validate=True) + except (ValueError, binascii.Error) as ex: + raise RoborockException("Invalid Q10 customized-room payload: malformed base64") from ex + if not raw: + raise RoborockException("Invalid Q10 customized-room payload: empty data") + + count = raw[0] + if count == 0 and len(raw) == 1: + return Q10RoomCleanUpdate((), complete=True) + if len(raw) == 1 + count * _Q10_CUSTOM_ROOM_COMPACT_SIZE: + offset = 1 + rooms = [] + room_ids: set[int] = set() + for _ in range(count): + room_id = raw[offset] + if room_id in room_ids: + raise RoborockException("Invalid Q10 customized-room payload: duplicate room id") + room_ids.add(room_id) + rooms.append( + Q10RoomCleanSettings( + room_id=room_id, + fan_level=(Q10RoomFanLevel.from_code_optional(raw[offset + 1]) or Q10RoomFanLevel.UNKNOWN), + water_level=(YXWaterLevel.from_code_optional(raw[offset + 2]) or YXWaterLevel.UNKNOWN), + clean_type=(Q10RoomCleanType.from_code_optional(raw[offset + 3]) or Q10RoomCleanType.UNKNOWN), + clean_count=(Q10CleanCount.from_code_optional(raw[offset + 4]) or Q10CleanCount.UNKNOWN), + clean_line=(YXCleanLine.from_code_optional(raw[offset + 5]) or YXCleanLine.UNKNOWN), + ) + ) + offset += _Q10_CUSTOM_ROOM_COMPACT_SIZE + return Q10RoomCleanUpdate(tuple(rooms), complete=False) + + offset = 1 + rooms = [] + room_ids = set() + for _ in range(count): + minimum_end = offset + _Q10_CUSTOM_ROOM_PROPERTIES_SIZE + _Q10_CUSTOM_ROOM_NAME_SIZE + 1 + if minimum_end > len(raw): + raise RoborockException("Invalid Q10 customized-room payload: truncated room record") + room_id = int.from_bytes(raw[offset : offset + 2], "big") + if room_id in room_ids: + raise RoborockException("Invalid Q10 customized-room payload: duplicate room id") + room_ids.add(room_id) + clean_count = int.from_bytes(raw[offset + 5 : offset + 7], "big") + rooms.append( + Q10RoomCleanSettings( + room_id=room_id, + fan_level=(Q10RoomFanLevel.from_code_optional(raw[offset + 8]) or Q10RoomFanLevel.UNKNOWN), + water_level=(YXWaterLevel.from_code_optional(raw[offset + 9]) or YXWaterLevel.UNKNOWN), + clean_type=(Q10RoomCleanType.from_code_optional(raw[offset + 7]) or Q10RoomCleanType.UNKNOWN), + clean_count=(Q10CleanCount.from_code_optional(clean_count) or Q10CleanCount.UNKNOWN), + clean_line=(YXCleanLine.from_code_optional(raw[offset + 11]) or YXCleanLine.UNKNOWN), + ) + ) + vertex_count = raw[offset + _Q10_CUSTOM_ROOM_PROPERTIES_SIZE + _Q10_CUSTOM_ROOM_NAME_SIZE] + offset = minimum_end + vertex_count * 4 + if offset > len(raw): + raise RoborockException("Invalid Q10 customized-room payload: truncated vertices") + if offset != len(raw): + raise RoborockException("Invalid Q10 customized-room payload: trailing data") + return Q10RoomCleanUpdate(tuple(rooms), complete=True) + + def encode_mqtt_payload(command: B01_Q10_DP, params: ParamsType) -> RoborockMessage: """Encode payload for B01 Q10 commands over MQTT. diff --git a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr index daac7f55b..2444ce418 100644 --- a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr +++ b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr @@ -41,6 +41,11 @@ 'wifiName': 'wifi-network-name', }), }), + 'room_cleaning': dict({ + 'settings': tuple( + ), + 'settingsAvailable': False, + }), 'status': dict({ 'addCleanState': False, 'areaUnit': 'square_meter', @@ -421,6 +426,11 @@ 'wifiName': '**REDACTED**', }), }), + 'room_cleaning': dict({ + 'settings': tuple( + ), + 'settingsAvailable': False, + }), 'status': dict({ 'addCleanState': False, 'areaUnit': 'square_meter', @@ -491,6 +501,11 @@ }), 'network_info': dict({ }), + 'room_cleaning': dict({ + 'settings': tuple( + ), + 'settingsAvailable': False, + }), 'status': dict({ }), 'volume': dict({ diff --git a/tests/devices/traits/b01/q10/test_room_cleaning.py b/tests/devices/traits/b01/q10/test_room_cleaning.py new file mode 100644 index 000000000..fe3f4f2b7 --- /dev/null +++ b/tests/devices/traits/b01/q10/test_room_cleaning.py @@ -0,0 +1,138 @@ +"""Tests for customized Q10 room-cleaning commands and pushed settings.""" + +import base64 + +import pytest + +from roborock.data.b01_q10.b01_q10_code_mappings import ( + B01_Q10_DP, + Q10CleanCount, + Q10RoomCleanType, + Q10RoomFanLevel, + YXCleanLine, + YXWaterLevel, +) +from roborock.data.b01_q10.b01_q10_containers import Q10RoomCleanSettings +from roborock.devices.traits.b01.q10 import Q10PropertiesApi +from roborock.exceptions import RoborockException +from roborock.protocols.b01_q10_protocol import encode_room_clean_settings + +from .conftest import FakeB01Q10Channel + + +def _settings(room_id: int = 3, *, fan: Q10RoomFanLevel = Q10RoomFanLevel.TURBO) -> Q10RoomCleanSettings: + return Q10RoomCleanSettings( + room_id=room_id, + fan_level=fan, + water_level=YXWaterLevel.HIGH, + clean_type=Q10RoomCleanType.VAC_AND_MOP, + clean_count=Q10CleanCount.TWICE, + clean_line=YXCleanLine.FINE, + ) + + +def _complete_payload(settings: Q10RoomCleanSettings) -> str: + properties = bytearray(26) + properties[0:2] = settings.room_id.to_bytes(2, "big") + properties[3:5] = b"\xff\xff" + properties[5:7] = settings.clean_count.code.to_bytes(2, "big") + properties[7] = settings.clean_type.code + properties[8] = settings.fan_level.code + properties[9] = settings.water_level.code + properties[10] = 0xFF + properties[11] = settings.clean_line.code + name = b"Example" + name_block = bytes((len(name),)) + name + bytes(19 - len(name)) + return base64.b64encode(b"\x01" + properties + name_block + b"\x00").decode() + + +async def test_set_settings_publishes_then_requests_full_settings( + q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel +) -> None: + settings = (_settings(),) + await q10_api.room_cleaning.set_settings(settings) + assert fake_channel.published_commands == [ + (B01_Q10_DP.COMMON, {"62": encode_room_clean_settings(settings)}), + (B01_Q10_DP.COMMON, {"63": 0}), + ] + assert q10_api.room_cleaning.settings == () + + +async def test_clean_rooms_sends_settings_mode_and_start_in_order( + q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel +) -> None: + settings = (_settings(3), _settings(9)) + await q10_api.vacuum.clean_rooms(settings) + assert fake_channel.published_commands == [ + (B01_Q10_DP.COMMON, {"62": encode_room_clean_settings(settings)}), + (B01_Q10_DP.CLEAN_MODE, 4), + (B01_Q10_DP.START_CLEAN, {"cmd": 2, "clean_paramters": [3, 9]}), + ] + + +@pytest.mark.parametrize("fail_on", [1, 2, 3]) +async def test_publish_failure_stops_remaining_clean_commands( + q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel, fail_on: int, monkeypatch: pytest.MonkeyPatch +) -> None: + original = fake_channel.send_command + calls = 0 + + async def send(command, params=None): + nonlocal calls + calls += 1 + if calls == fail_on: + raise RoborockException("publish failed") + await original(command, params) + + monkeypatch.setattr(fake_channel, "send_command", send) + with pytest.raises(RoborockException, match="publish failed"): + await q10_api.room_cleaning.clean((_settings(),)) + assert calls == fail_on + assert len(fake_channel.published_commands) == fail_on - 1 + + +async def test_set_settings_failure_does_not_request_refresh( + q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel, monkeypatch: pytest.MonkeyPatch +) -> None: + async def send(command, params=None): + raise RoborockException("publish failed") + + monkeypatch.setattr(fake_channel, "send_command", send) + with pytest.raises(RoborockException, match="publish failed"): + await q10_api.room_cleaning.set_settings((_settings(),)) + assert fake_channel.published_commands == [] + + +def test_full_push_replaces_and_compact_push_upserts(q10_api: Q10PropertiesApi) -> None: + trait = q10_api.room_cleaning + notified = [] + trait.add_update_listener(lambda: notified.append(True)) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: _complete_payload(_settings(3))}) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: encode_room_clean_settings((_settings(9),))}) + assert trait.settings == (_settings(3), _settings(9)) + changed = _settings(3, fan=Q10RoomFanLevel.MAX) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: encode_room_clean_settings((changed,))}) + assert trait.settings == (changed, _settings(9)) + assert trait.settings_for_room(3) == changed + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: _complete_payload(_settings(1))}) + assert trait.settings == (_settings(1),) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "invalid"}) + assert trait.settings == (_settings(1),) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AA=="}) + assert trait.settings == () + assert trait.settings_available + assert len(notified) == 5 + + +@pytest.mark.parametrize("field", ["fan_level", "water_level", "clean_type", "clean_count", "clean_line"]) +def test_unknown_setting_cannot_be_written(field: str) -> None: + settings = _settings() + setattr(settings, field, type(getattr(settings, field)).UNKNOWN) + with pytest.raises(ValueError, match=field): + encode_room_clean_settings((settings,)) + + +@pytest.mark.parametrize("rooms", [(), (_settings(-1),), (_settings(256),), (_settings(), _settings())]) +def test_semantic_room_guards(rooms: tuple[Q10RoomCleanSettings, ...]) -> None: + with pytest.raises(ValueError): + encode_room_clean_settings(rooms) diff --git a/tests/devices/traits/b01/q10/test_vacuum.py b/tests/devices/traits/b01/q10/test_vacuum.py index 6b6c477ef..f8517de36 100644 --- a/tests/devices/traits/b01/q10/test_vacuum.py +++ b/tests/devices/traits/b01/q10/test_vacuum.py @@ -8,10 +8,13 @@ from roborock.data.b01_q10.b01_q10_code_mappings import ( B01_Q10_DP, + Q10CleanCount, + YXCleanLine, YXCleanType, YXDeviceCleanTask, YXDeviceState, YXFanLevel, + YXWaterLevel, ) from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint from roborock.devices.traits.b01.q10 import Q10PropertiesApi @@ -236,3 +239,32 @@ async def test_goto_position_at_current_position_pauses_owned_zone( B01_Q10_DP.START_CLEAN, B01_Q10_DP.PAUSE, ] + + +@pytest.mark.parametrize( + ("method", "value", "command", "payload"), + [ + ("set_water_level", YXWaterLevel.HIGH, B01_Q10_DP.WATER_LEVEL, 3), + ("set_clean_count", Q10CleanCount.THREE_TIMES, B01_Q10_DP.CLEAN_COUNT, 3), + ("set_clean_line", YXCleanLine.FINE, B01_Q10_DP.COMMON, {"78": 2}), + ], +) +async def test_advanced_cleaning_controls(q10_api, fake_channel, method, value, command, payload) -> None: + await getattr(q10_api.vacuum, method)(value) + assert fake_channel.published_commands == [(command, payload)] + + +@pytest.mark.parametrize( + ("method", "value"), + [ + ("set_clean_mode", YXCleanType.UNKNOWN), + ("set_fan_level", YXFanLevel.UNKNOWN), + ("set_water_level", YXWaterLevel.UNKNOWN), + ("set_clean_count", Q10CleanCount.UNKNOWN), + ("set_clean_line", YXCleanLine.UNKNOWN), + ], +) +async def test_unknown_cleaning_modes_are_not_published(q10_api, fake_channel, method, value) -> None: + with pytest.raises(ValueError, match="supported"): + await getattr(q10_api.vacuum, method)(value) + assert fake_channel.published_commands == [] diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py index 16fe9a108..09052eeae 100644 --- a/tests/protocols/test_b01_q10_protocol.py +++ b/tests/protocols/test_b01_q10_protocol.py @@ -1,5 +1,6 @@ """Tests for the B01 protocol message encoding and decoding.""" +import base64 import json import logging import pathlib @@ -11,8 +12,15 @@ from freezegun import freeze_time from syrupy import SnapshotAssertion -from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, YXWaterLevel -from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint +from roborock.data.b01_q10.b01_q10_code_mappings import ( + B01_Q10_DP, + Q10CleanCount, + Q10RoomCleanType, + Q10RoomFanLevel, + YXCleanLine, + YXWaterLevel, +) +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint, Q10RoomCleanSettings from roborock.data.code_mappings import completed_warnings from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import Q10CleanRecordDetail, Q10MapPacket, Q10MapPacketKind, Q10TracePacket @@ -20,9 +28,11 @@ CleanParams, Q10DpsUpdate, decode_message, + decode_room_clean_settings, decode_rpc_response, encode_clean_params, encode_mqtt_payload, + encode_room_clean_settings, ) from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol @@ -289,3 +299,192 @@ def test_encode_mqtt_payload(command: B01_Q10_DP, params: dict[str, Any], snapsh assert snapshot == message.payload json.loads(message.payload.decode()) + + +def _room_settings( + room_id: int = 3, + *, + fan: Q10RoomFanLevel = Q10RoomFanLevel.TURBO, + water: YXWaterLevel = YXWaterLevel.HIGH, + clean_type: Q10RoomCleanType = Q10RoomCleanType.VAC_AND_MOP, + count: Q10CleanCount = Q10CleanCount.TWICE, + line: YXCleanLine = YXCleanLine.FINE, +) -> Q10RoomCleanSettings: + return Q10RoomCleanSettings(room_id, fan, water, clean_type, count, line) + + +def _full_room_record(settings: Q10RoomCleanSettings, name: str = "Example") -> bytes: + properties = bytearray(26) + properties[0:2] = settings.room_id.to_bytes(2, "big") + properties[2] = 1 + properties[3:5] = b"\xff\xff" + properties[5:7] = settings.clean_count.code.to_bytes(2, "big") + properties[7] = settings.clean_type.code + properties[8] = settings.fan_level.code + properties[9] = settings.water_level.code + properties[10] = 0xFF + properties[11] = settings.clean_line.code + encoded_name = name.encode() + name_block = bytes((len(encoded_name),)) + encoded_name + bytes(19 - len(encoded_name)) + return bytes(properties) + name_block + b"\x00" + + +def test_encode_room_clean_settings_uses_compact_format() -> None: + settings = [_room_settings(), _room_settings(9, fan=Q10RoomFanLevel.MAX_PLUS, water=YXWaterLevel.LOW)] + + encoded = encode_room_clean_settings(settings) + + assert base64.b64decode(encoded) == bytes((2, 3, 3, 3, 1, 2, 2, 9, 5, 1, 1, 2, 2)) + assert settings == [ + _room_settings(), + _room_settings(9, fan=Q10RoomFanLevel.MAX_PLUS, water=YXWaterLevel.LOW), + ] + + +def test_decode_room_clean_settings_distinguishes_echo_and_complete_response() -> None: + first = _room_settings() + second = _room_settings(9, fan=Q10RoomFanLevel.MAX_PLUS, water=YXWaterLevel.LOW) + echo = decode_room_clean_settings(encode_room_clean_settings((first, second))) + complete = decode_room_clean_settings( + base64.b64encode(b"\x02" + _full_room_record(first) + _full_room_record(second)).decode() + ) + + assert echo.settings == ( + Q10RoomCleanSettings(**first.__dict__), + Q10RoomCleanSettings(**second.__dict__), + ) + assert echo.complete is False + assert complete.settings == ( + Q10RoomCleanSettings(**first.__dict__), + Q10RoomCleanSettings(**second.__dict__), + ) + assert complete.complete is True + + +def test_decode_room_clean_settings_preserves_unknown_values() -> None: + record = bytearray(_full_room_record(_room_settings())) + record[8] = 42 + record[11] = 99 + + decoded = decode_room_clean_settings(base64.b64encode(b"\x01" + record).decode()) + + assert decoded.settings[0].fan_level is Q10RoomFanLevel.UNKNOWN + assert decoded.settings[0].clean_line is YXCleanLine.UNKNOWN + + +def test_decode_room_clean_settings_empty_complete_response() -> None: + decoded = decode_room_clean_settings("AA==") + + assert decoded.settings == () + assert decoded.complete is True + + +def test_encode_room_clean_settings_rejects_more_than_wire_count() -> None: + with pytest.raises(ValueError, match="between 1 and 255"): + encode_room_clean_settings([_room_settings(room_id) for room_id in range(256)]) + + +def test_encode_room_clean_settings_matches_physical_echoes() -> None: + candidate = Q10RoomCleanSettings( + 1, + Q10RoomFanLevel.TURBO, + YXWaterLevel.MEDIUM, + Q10RoomCleanType.VAC_AND_MOP, + Q10CleanCount.ONCE, + YXCleanLine.DAILY, + ) + restored = Q10RoomCleanSettings( + 1, + Q10RoomFanLevel.QUIET, + YXWaterLevel.HIGH, + Q10RoomCleanType.VAC_AND_MOP, + Q10CleanCount.TWICE, + YXCleanLine.FINE, + ) + + assert encode_room_clean_settings((candidate,)) == "AQEDAgEBAQ==" + assert encode_room_clean_settings((restored,)) == "AQEBAwECAg==" + assert decode_room_clean_settings("AQEDAgEBAQ==").settings == (Q10RoomCleanSettings(**candidate.__dict__),) + assert decode_room_clean_settings("AQEBAwECAg==").settings == (Q10RoomCleanSettings(**restored.__dict__),) + + +def test_encode_room_clean_settings_supports_three_pass_protocol_value() -> None: + settings = _room_settings(count=Q10CleanCount.THREE_TIMES) + + assert base64.b64decode(encode_room_clean_settings((settings,)))[5] == 3 + + +@pytest.mark.parametrize( + "payload", + [ + "not-base64", + "", + base64.b64encode(b"\x01\x00").decode(), + base64.b64encode(b"\x01" + _full_room_record(_room_settings())[:-1]).decode(), + base64.b64encode(b"\x01" + _full_room_record(_room_settings()) + b"extra").decode(), + base64.b64encode(b"\x02" + bytes((3, 3, 3, 1, 2, 2)) * 2).decode(), + base64.b64encode(b"\x02" + _full_room_record(_room_settings()) * 2).decode(), + ], +) +def test_decode_room_clean_settings_rejects_malformed_payload(payload: str) -> None: + with pytest.raises(RoborockException, match="customized-room"): + decode_room_clean_settings(payload) + + +def test_decode_sanitized_physical_six_room_capture() -> None: + payload = (TESTDATA_PATH / "customer_clean_full_sanitized.b64").read_text().strip() + + decoded = decode_room_clean_settings(payload) + + assert decoded.complete is True + assert len(base64.b64decode(payload)) == 283 + assert list(decoded.settings) == [ + Q10RoomCleanSettings( + 1, + Q10RoomFanLevel.QUIET, + YXWaterLevel.HIGH, + Q10RoomCleanType.VAC_AND_MOP, + Q10CleanCount.TWICE, + YXCleanLine.FINE, + ), + Q10RoomCleanSettings( + 2, + Q10RoomFanLevel.MAX_PLUS, + YXWaterLevel.HIGH, + Q10RoomCleanType.VAC_AND_MOP, + Q10CleanCount.TWICE, + YXCleanLine.FINE, + ), + Q10RoomCleanSettings( + 3, + Q10RoomFanLevel.MAX_PLUS, + YXWaterLevel.MEDIUM, + Q10RoomCleanType.VACUUM, + Q10CleanCount.TWICE, + YXCleanLine.FINE, + ), + Q10RoomCleanSettings( + 4, + Q10RoomFanLevel.MAX_PLUS, + YXWaterLevel.MEDIUM, + Q10RoomCleanType.VACUUM, + Q10CleanCount.TWICE, + YXCleanLine.FINE, + ), + Q10RoomCleanSettings( + 5, + Q10RoomFanLevel.QUIET, + YXWaterLevel.HIGH, + Q10RoomCleanType.VAC_AND_MOP, + Q10CleanCount.TWICE, + YXCleanLine.FINE, + ), + Q10RoomCleanSettings( + 6, + Q10RoomFanLevel.MAX_PLUS, + YXWaterLevel.HIGH, + Q10RoomCleanType.VAC_AND_MOP, + Q10CleanCount.TWICE, + YXCleanLine.FINE, + ), + ] diff --git a/tests/protocols/testdata/b01_q10_protocol/customer_clean_full_sanitized.b64 b/tests/protocols/testdata/b01_q10_protocol/customer_clean_full_sanitized.b64 new file mode 100644 index 000000000..dbc00f660 --- /dev/null +++ b/tests/protocols/testdata/b01_q10_protocol/customer_clean_full_sanitized.b64 @@ -0,0 +1 @@ +BgABAP//AAIBAQP/AgAAAAAAAAAAAAAAAAAABlJvb20gMQAAAAAAAAAAAAAAAAAAAAIA//8AAgEFA/8CAAAAAAAAAAAAAAAAAAAGUm9vbSAyAAAAAAAAAAAAAAAAAAAAAwL//wACAgUC/wIAAAAAAAAAAAAAAAAAAAZSb29tIDMAAAAAAAAAAAAAAAAAAAAEAf//AAICBQL/AgAAAAAAAAAAAAAAAAAABlJvb20gNAAAAAAAAAAAAAAAAAAAAAUJ//8AAgEBA/8CAAAAAAAAAAAAAAAAAAAGUm9vbSA1AAAAAAAAAAAAAAAAAAAABgD//wACAQUD/wIAAAAAAAAAAAAAAAAAAAZSb29tIDYAAAAAAAAAAAAAAAAAAA== From 575f824e9cee52a258fdba371632056df1017173 Mon Sep 17 00:00:00 2001 From: Harry Coureau Date: Sun, 8 Nov 2026 18:59:53 +0000 Subject: [PATCH 2/2] fix: simplify Q10 room cleaning state and commands --- roborock/data/b01_q10/b01_q10_containers.py | 4 +- roborock/devices/traits/b01/q10/__init__.py | 2 +- .../devices/traits/b01/q10/room_cleaning.py | 30 ++---- roborock/devices/traits/b01/q10/vacuum.py | 14 ++- roborock/protocols/b01_q10_protocol.py | 17 ++- .../b01/q10/__snapshots__/test_status.ambr | 9 -- .../traits/b01/q10/test_room_cleaning.py | 101 +++++++++--------- tests/devices/traits/b01/q10/test_status.py | 17 +++ tests/devices/traits/b01/q10/test_vacuum.py | 79 +++++++++++++- tests/protocols/test_b01_q10_protocol.py | 32 ++++-- 10 files changed, 198 insertions(+), 107 deletions(-) diff --git a/roborock/data/b01_q10/b01_q10_containers.py b/roborock/data/b01_q10/b01_q10_containers.py index 89220dbc2..5e4f6d73e 100644 --- a/roborock/data/b01_q10/b01_q10_containers.py +++ b/roborock/data/b01_q10/b01_q10_containers.py @@ -292,7 +292,9 @@ def fault_name(self) -> str | None: @property def clean_count_mode(self) -> Q10CleanCount | None: """Return the typed cleaning pass count from the reported status.""" - return Q10CleanCount.from_code_optional(self.clean_count) if self.clean_count is not None else None + if self.clean_count is None: + return None + return Q10CleanCount.from_code_optional(self.clean_count) or Q10CleanCount.UNKNOWN @dataclass diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index cbe804b6c..6e9a3adfb 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/roborock/devices/traits/b01/q10/__init__.py @@ -114,7 +114,7 @@ def __init__(self, channel: B01Q10Channel, *, map_parser_config: B01Q10MapParser self.maps = MapsTrait(self.command) self.map = MapContentTrait(self._map_dps, self.maps, self.command, map_parser_config=map_parser_config) self.room_cleaning = RoomCleaningTrait(self.command) - self.vacuum = VacuumTrait(self.command, self.status, self.map, self.room_cleaning) + self.vacuum = VacuumTrait(self.command, self.status, self.map) self.clean_history = CleanHistoryTrait(self.command) # Read-model traits updated from the device's DPS push stream. self._updatable_traits = [ diff --git a/roborock/devices/traits/b01/q10/room_cleaning.py b/roborock/devices/traits/b01/q10/room_cleaning.py index 6d191b5a1..61c567520 100644 --- a/roborock/devices/traits/b01/q10/room_cleaning.py +++ b/roborock/devices/traits/b01/q10/room_cleaning.py @@ -5,7 +5,7 @@ from dataclasses import dataclass from typing import Any -from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, YXCleanType, YXDeviceCleanTask +from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP from roborock.data.b01_q10.b01_q10_containers import Q10RoomCleanSettings from roborock.data.containers import RoborockBase from roborock.exceptions import RoborockException @@ -21,8 +21,7 @@ class RoomCleaning(RoborockBase): """Current customized room-cleaning state.""" - settings: tuple[Q10RoomCleanSettings, ...] = () - settings_available: bool = False + settings: list[Q10RoomCleanSettings] | None = None class RoomCleaningTrait(RoomCleaning, UpdatableTrait): @@ -35,6 +34,8 @@ def __init__(self, command: CommandTrait) -> None: def settings_for_room(self, room_id: int) -> Q10RoomCleanSettings | None: """Return the latest reported settings for one room.""" + if self.settings is None: + return None return next((room for room in self.settings if room.room_id == room_id), None) async def refresh(self) -> None: @@ -47,17 +48,6 @@ async def set_settings(self, settings: Sequence[Q10RoomCleanSettings]) -> None: await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CUSTOMER_CLEAN.code): payload}) await self.refresh() - async def clean(self, settings: Sequence[Q10RoomCleanSettings]) -> None: - """Publish settings, select customized mode, and start the selected rooms.""" - selected = tuple(settings) - payload = encode_room_clean_settings(selected) - await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CUSTOMER_CLEAN.code): payload}) - await self._command.send(B01_Q10_DP.CLEAN_MODE, YXCleanType.CUSTOMIZED.code) - await self._command.send( - B01_Q10_DP.START_CLEAN, - {"cmd": YXDeviceCleanTask.ELECTORAL.code, "clean_paramters": [room.room_id for room in selected]}, - ) - def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: """Replace full settings or upsert the rooms in a compact push.""" payload = decoded_dps.get(B01_Q10_DP.CUSTOMER_CLEAN) @@ -69,13 +59,11 @@ def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: _LOGGER.debug("Ignoring malformed Q10 customized-room settings", exc_info=True) return if update.complete: - settings = update.settings + settings = list(update.settings) else: - by_room = {room.room_id: room for room in self.settings} + by_room = {room.room_id: room for room in (self.settings or [])} by_room.update((room.room_id, room) for room in update.settings) - settings = tuple(by_room.values()) - changed = settings != self.settings or not self.settings_available - self.settings = settings - self.settings_available = True - if changed: + settings = list(by_room.values()) + if settings != self.settings: + self.settings = settings self._notify_update() diff --git a/roborock/devices/traits/b01/q10/vacuum.py b/roborock/devices/traits/b01/q10/vacuum.py index c1788f936..5c655e089 100644 --- a/roborock/devices/traits/b01/q10/vacuum.py +++ b/roborock/devices/traits/b01/q10/vacuum.py @@ -16,12 +16,11 @@ ) from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint, Q10RoomCleanSettings from roborock.exceptions import RoborockException -from roborock.protocols.b01_q10_protocol import CleanParams, encode_clean_params +from roborock.protocols.b01_q10_protocol import CleanParams, encode_clean_params, encode_room_clean_settings from .command import CommandTrait from .goto import GotoAction, GotoActionCommand, GotoSnapshot from .map import MapContentTrait -from .room_cleaning import RoomCleaningTrait from .status import StatusTrait _GOTO_HALF_ZONE_SIZE = 200 @@ -44,13 +43,11 @@ def __init__( command: CommandTrait, status: StatusTrait, map_content: MapContentTrait, - room_cleaning: RoomCleaningTrait, ) -> None: """Initialize the VacuumTrait.""" self._command = command self._status = status self._map = map_content - self._room_cleaning = room_cleaning self._goto_action: GotoAction | None = None self._goto_action_remove_listener: Callable[[], None] | None = None self._goto_timeout_task: asyncio.Task[None] | None = None @@ -175,7 +172,14 @@ async def clean_segments(self, segment_ids: list[int]) -> None: async def clean_rooms(self, settings: Sequence[Q10RoomCleanSettings]) -> None: """Start a customized clean for the supplied rooms.""" - await self._room_cleaning.clean(settings) + selected = tuple(settings) + payload = encode_room_clean_settings(selected) + await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CUSTOMER_CLEAN.code): payload}) + await self._command.send(B01_Q10_DP.CLEAN_MODE, YXCleanType.CUSTOMIZED.code) + await self._command.send( + B01_Q10_DP.START_CLEAN, + {"cmd": YXDeviceCleanTask.ELECTORAL.code, "clean_paramters": [room.room_id for room in selected]}, + ) self.cancel_goto() async def clean_zone( diff --git a/roborock/protocols/b01_q10_protocol.py b/roborock/protocols/b01_q10_protocol.py index f30640db9..6d7339b75 100644 --- a/roborock/protocols/b01_q10_protocol.py +++ b/roborock/protocols/b01_q10_protocol.py @@ -1,13 +1,10 @@ """Roborock B01 Protocol encoding and decoding.""" -import base64 import binascii - - import json import logging import struct -from base64 import b64encode +from base64 import b64decode, b64encode from collections.abc import Sequence from dataclasses import dataclass from struct import pack @@ -131,7 +128,7 @@ def encode_room_clean_settings(settings: Sequence[Q10RoomCleanSettings]) -> str: raise ValueError(f"{field_name} must be a supported {mode_type.__name__}") values.append(value.code) payload.extend(values) - return base64.b64encode(payload).decode() + return b64encode(payload).decode() def decode_room_clean_settings(payload: str) -> Q10RoomCleanUpdate: @@ -142,7 +139,7 @@ def decode_room_clean_settings(payload: str) -> Q10RoomCleanUpdate: are room id 0, count 5, type 7, fan 8, water 9, and route 11. """ try: - raw = base64.b64decode(payload, validate=True) + raw = b64decode(payload, validate=True) except (ValueError, binascii.Error) as ex: raise RoborockException("Invalid Q10 customized-room payload: malformed base64") from ex if not raw: @@ -253,7 +250,7 @@ def encode_restricted_zones(zones: Sequence[Q10RestrictedZone]) -> str: if len(zones) > 255: raise ValueError("zones must be a sequence of at most 255 restrictions") if not zones: - return base64.b64encode(b"\x01\x00").decode() + return b64encode(b"\x01\x00").decode() payload = bytearray((1, len(zones))) for zone in zones: @@ -276,7 +273,7 @@ def encode_restricted_zones(zones: Sequence[Q10RestrictedZone]) -> str: ) record.extend(bytes(_Q10_ZONE_RECORD_SIZE - len(record))) payload.extend(record) - return base64.b64encode(payload).decode() + return b64encode(payload).decode() def encode_virtual_walls(walls: Sequence[Q10VirtualWall]) -> str: @@ -297,7 +294,7 @@ def encode_virtual_walls(walls: Sequence[Q10VirtualWall]) -> str: y, ) ) - return base64.b64encode(payload).decode() + return b64encode(payload).decode() def encode_room_name(room_id: int, name: str) -> str: @@ -313,7 +310,7 @@ def encode_room_name(room_id: int, name: str) -> str: payload = bytearray((1, room_id, 0, len(encoded_name))) payload.extend(encoded_name) payload.extend(bytes(_Q10_ROOM_NAME_FIELD_SIZE - len(encoded_name))) - return base64.b64encode(payload).decode() + return b64encode(payload).decode() def encode_mqtt_payload(command: B01_Q10_DP, params: ParamsType) -> RoborockMessage: diff --git a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr index 2444ce418..8f317521e 100644 --- a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr +++ b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr @@ -42,9 +42,6 @@ }), }), 'room_cleaning': dict({ - 'settings': tuple( - ), - 'settingsAvailable': False, }), 'status': dict({ 'addCleanState': False, @@ -427,9 +424,6 @@ }), }), 'room_cleaning': dict({ - 'settings': tuple( - ), - 'settingsAvailable': False, }), 'status': dict({ 'addCleanState': False, @@ -502,9 +496,6 @@ 'network_info': dict({ }), 'room_cleaning': dict({ - 'settings': tuple( - ), - 'settingsAvailable': False, }), 'status': dict({ }), diff --git a/tests/devices/traits/b01/q10/test_room_cleaning.py b/tests/devices/traits/b01/q10/test_room_cleaning.py index fe3f4f2b7..4ca5f3039 100644 --- a/tests/devices/traits/b01/q10/test_room_cleaning.py +++ b/tests/devices/traits/b01/q10/test_room_cleaning.py @@ -55,40 +55,7 @@ async def test_set_settings_publishes_then_requests_full_settings( (B01_Q10_DP.COMMON, {"62": encode_room_clean_settings(settings)}), (B01_Q10_DP.COMMON, {"63": 0}), ] - assert q10_api.room_cleaning.settings == () - - -async def test_clean_rooms_sends_settings_mode_and_start_in_order( - q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel -) -> None: - settings = (_settings(3), _settings(9)) - await q10_api.vacuum.clean_rooms(settings) - assert fake_channel.published_commands == [ - (B01_Q10_DP.COMMON, {"62": encode_room_clean_settings(settings)}), - (B01_Q10_DP.CLEAN_MODE, 4), - (B01_Q10_DP.START_CLEAN, {"cmd": 2, "clean_paramters": [3, 9]}), - ] - - -@pytest.mark.parametrize("fail_on", [1, 2, 3]) -async def test_publish_failure_stops_remaining_clean_commands( - q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel, fail_on: int, monkeypatch: pytest.MonkeyPatch -) -> None: - original = fake_channel.send_command - calls = 0 - - async def send(command, params=None): - nonlocal calls - calls += 1 - if calls == fail_on: - raise RoborockException("publish failed") - await original(command, params) - - monkeypatch.setattr(fake_channel, "send_command", send) - with pytest.raises(RoborockException, match="publish failed"): - await q10_api.room_cleaning.clean((_settings(),)) - assert calls == fail_on - assert len(fake_channel.published_commands) == fail_on - 1 + assert q10_api.room_cleaning.settings is None async def test_set_settings_failure_does_not_request_refresh( @@ -109,30 +76,64 @@ def test_full_push_replaces_and_compact_push_upserts(q10_api: Q10PropertiesApi) trait.add_update_listener(lambda: notified.append(True)) trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: _complete_payload(_settings(3))}) trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: encode_room_clean_settings((_settings(9),))}) - assert trait.settings == (_settings(3), _settings(9)) + assert trait.settings == [_settings(3), _settings(9)] changed = _settings(3, fan=Q10RoomFanLevel.MAX) trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: encode_room_clean_settings((changed,))}) - assert trait.settings == (changed, _settings(9)) + assert trait.settings == [changed, _settings(9)] assert trait.settings_for_room(3) == changed trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: _complete_payload(_settings(1))}) - assert trait.settings == (_settings(1),) + assert trait.settings == [_settings(1)] trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "invalid"}) - assert trait.settings == (_settings(1),) + assert trait.settings == [_settings(1)] trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AA=="}) - assert trait.settings == () - assert trait.settings_available + assert trait.settings == [] assert len(notified) == 5 -@pytest.mark.parametrize("field", ["fan_level", "water_level", "clean_type", "clean_count", "clean_line"]) -def test_unknown_setting_cannot_be_written(field: str) -> None: - settings = _settings() - setattr(settings, field, type(getattr(settings, field)).UNKNOWN) - with pytest.raises(ValueError, match=field): - encode_room_clean_settings((settings,)) +async def test_refresh_requests_complete_settings(q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel) -> None: + await q10_api.room_cleaning.refresh() + assert fake_channel.published_commands == [(B01_Q10_DP.COMMON, {"63": 0})] + assert q10_api.room_cleaning.settings is None -@pytest.mark.parametrize("rooms", [(), (_settings(-1),), (_settings(256),), (_settings(), _settings())]) -def test_semantic_room_guards(rooms: tuple[Q10RoomCleanSettings, ...]) -> None: - with pytest.raises(ValueError): - encode_room_clean_settings(rooms) +def test_settings_for_room_unpopulated(q10_api: Q10PropertiesApi) -> None: + assert q10_api.room_cleaning.settings_for_room(3) is None + assert q10_api.room_cleaning.as_dict() == {} + assert q10_api.as_dict()["room_cleaning"] == {} + + +def test_settings_for_room_missing(q10_api: Q10PropertiesApi) -> None: + q10_api.room_cleaning.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: _complete_payload(_settings(3))}) + assert q10_api.room_cleaning.settings_for_room(9) is None + + +def test_repeated_pushes_do_not_notify(q10_api: Q10PropertiesApi) -> None: + trait = q10_api.room_cleaning + notified: list[bool] = [] + trait.add_update_listener(lambda: notified.append(True)) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "invalid"}) + assert trait.settings is None + assert not notified + compact = encode_room_clean_settings((_settings(3),)) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: compact}) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: compact}) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: _complete_payload(_settings(3))}) + assert trait.settings == [_settings(3)] + assert len(notified) == 1 + assert isinstance(trait.as_dict()["settings"], list) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AA=="}) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AA=="}) + assert trait.settings == [] + assert trait.as_dict() == {"settings": []} + assert trait.settings_for_room(3) is None + assert len(notified) == 2 + + +def test_first_empty_push_is_reported(q10_api: Q10PropertiesApi) -> None: + trait = q10_api.room_cleaning + notified: list[bool] = [] + trait.add_update_listener(lambda: notified.append(True)) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AA=="}) + assert trait.settings == [] + assert trait.as_dict() == {"settings": []} + assert notified == [True] diff --git a/tests/devices/traits/b01/q10/test_status.py b/tests/devices/traits/b01/q10/test_status.py index 240c87f5a..339a6ccc6 100644 --- a/tests/devices/traits/b01/q10/test_status.py +++ b/tests/devices/traits/b01/q10/test_status.py @@ -11,6 +11,7 @@ from roborock.data import HomeDataDevice, HomeDataProduct from roborock.data.b01_q10.b01_q10_code_mappings import ( B01_Q10_DP, + Q10CleanCount, YXAreaUnit, YXCarpetCleanType, YXCleanLine, @@ -282,3 +283,19 @@ async def test_q10_properties_as_dict_and_diagnostic_data( # Verify diagnostic_data extracts the properties correctly diagnostics = device.diagnostic_data() assert diagnostics == snapshot + + +@pytest.mark.parametrize( + ("code", "expected"), + [ + (None, None), + (1, Q10CleanCount.ONCE), + (2, Q10CleanCount.TWICE), + (3, Q10CleanCount.THREE_TIMES), + (99, Q10CleanCount.UNKNOWN), + ], +) +def test_clean_count_mode(q10_api: Q10PropertiesApi, code: int | None, expected: Q10CleanCount | None) -> None: + if code is not None: + q10_api.status.update_from_dps({B01_Q10_DP.CLEAN_COUNT: code}) + assert q10_api.status.clean_count_mode is expected diff --git a/tests/devices/traits/b01/q10/test_vacuum.py b/tests/devices/traits/b01/q10/test_vacuum.py index f8517de36..ec85d0a12 100644 --- a/tests/devices/traits/b01/q10/test_vacuum.py +++ b/tests/devices/traits/b01/q10/test_vacuum.py @@ -9,6 +9,8 @@ from roborock.data.b01_q10.b01_q10_code_mappings import ( B01_Q10_DP, Q10CleanCount, + Q10RoomCleanType, + Q10RoomFanLevel, YXCleanLine, YXCleanType, YXDeviceCleanTask, @@ -16,12 +18,13 @@ YXFanLevel, YXWaterLevel, ) -from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint, Q10RoomCleanSettings from roborock.devices.traits.b01.q10 import Q10PropertiesApi from roborock.devices.traits.b01.q10 import vacuum as vacuum_module from roborock.devices.traits.b01.q10.vacuum import VacuumTrait from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import Q10Point, Q10TracePacket +from roborock.protocols.b01_q10_protocol import encode_room_clean_settings from .conftest import FakeB01Q10Channel @@ -268,3 +271,77 @@ async def test_unknown_cleaning_modes_are_not_published(q10_api, fake_channel, m with pytest.raises(ValueError, match="supported"): await getattr(q10_api.vacuum, method)(value) assert fake_channel.published_commands == [] + + +def _room_settings(room_id: int = 3) -> Q10RoomCleanSettings: + return Q10RoomCleanSettings( + room_id, + Q10RoomFanLevel.TURBO, + YXWaterLevel.HIGH, + Q10RoomCleanType.VAC_AND_MOP, + Q10CleanCount.TWICE, + YXCleanLine.FINE, + ) + + +async def test_clean_rooms_sends_settings_mode_and_start_in_order( + q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel +) -> None: + settings = (_room_settings(3), _room_settings(9)) + await q10_api.vacuum.clean_rooms(settings) + assert fake_channel.published_commands == [ + (B01_Q10_DP.COMMON, {"62": encode_room_clean_settings(settings)}), + (B01_Q10_DP.CLEAN_MODE, 4), + (B01_Q10_DP.START_CLEAN, {"cmd": 2, "clean_paramters": [3, 9]}), + ] + + +@pytest.mark.parametrize("fail_on", [1, 2, 3]) +async def test_publish_failure_stops_remaining_clean_commands( + q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel, fail_on: int, monkeypatch: pytest.MonkeyPatch +) -> None: + original = fake_channel.send_command + calls = 0 + + async def send(command, params=None): + nonlocal calls + calls += 1 + if calls == fail_on: + raise RoborockException("publish failed") + await original(command, params) + + monkeypatch.setattr(fake_channel, "send_command", send) + with pytest.raises(RoborockException, match="publish failed"): + await q10_api.vacuum.clean_rooms((_room_settings(),)) + assert calls == fail_on + assert len(fake_channel.published_commands) == fail_on - 1 + + +async def test_clean_rooms_cancels_active_goto( + q10_api: Q10PropertiesApi, + fake_channel: FakeB01Q10Channel, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An old goto cannot pause or stop a successful customized room clean.""" + monkeypatch.setattr(vacuum_module, "_GOTO_TIMEOUT", 0.01) + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(0, 0)], sequence=1)) + q10_api.status.update_from_dps( + { + B01_Q10_DP.CLEAN_TASK_TYPE: YXDeviceCleanTask.DIVIDE_AREAS.code, + B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code, + } + ) + try: + await q10_api.vacuum.goto_position(Q10RoborockPoint(29900, 28650)) + await q10_api.vacuum.clean_rooms((_room_settings(),)) + # Keep the old zone status while its trace reaches the goto target. + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1760, 1260)], sequence=2)) + await asyncio.sleep(0.03) + assert [command for command, _ in fake_channel.published_commands] == [ + B01_Q10_DP.START_CLEAN, + B01_Q10_DP.COMMON, + B01_Q10_DP.CLEAN_MODE, + B01_Q10_DP.START_CLEAN, + ] + finally: + await q10_api.close() diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py index 05b37b91f..c4031482b 100644 --- a/tests/protocols/test_b01_q10_protocol.py +++ b/tests/protocols/test_b01_q10_protocol.py @@ -34,10 +34,8 @@ decode_rpc_response, encode_clean_params, encode_mqtt_payload, - encode_room_clean_settings, - - encode_restricted_zones, + encode_room_clean_settings, encode_room_name, encode_virtual_walls, ) @@ -357,13 +355,13 @@ def test_decode_room_clean_settings_distinguishes_echo_and_complete_response() - ) assert echo.settings == ( - Q10RoomCleanSettings(**first.__dict__), - Q10RoomCleanSettings(**second.__dict__), + first, + second, ) assert echo.complete is False assert complete.settings == ( - Q10RoomCleanSettings(**first.__dict__), - Q10RoomCleanSettings(**second.__dict__), + first, + second, ) assert complete.complete is True @@ -411,8 +409,8 @@ def test_encode_room_clean_settings_matches_physical_echoes() -> None: assert encode_room_clean_settings((candidate,)) == "AQEDAgEBAQ==" assert encode_room_clean_settings((restored,)) == "AQEBAwECAg==" - assert decode_room_clean_settings("AQEDAgEBAQ==").settings == (Q10RoomCleanSettings(**candidate.__dict__),) - assert decode_room_clean_settings("AQEBAwECAg==").settings == (Q10RoomCleanSettings(**restored.__dict__),) + assert decode_room_clean_settings("AQEDAgEBAQ==").settings == (candidate,) + assert decode_room_clean_settings("AQEBAwECAg==").settings == (restored,) def test_encode_room_clean_settings_supports_three_pass_protocol_value() -> None: @@ -677,3 +675,19 @@ def test_encode_room_name_rejects_invalid_room_id(room_id: int) -> None: def test_encode_room_name_rejects_invalid_name(name: str) -> None: with pytest.raises(ValueError, match="name"): encode_room_name(7, name) + + +@pytest.mark.parametrize("field", ["fan_level", "water_level", "clean_type", "clean_count", "clean_line"]) +def test_unknown_setting_cannot_be_written(field: str) -> None: + settings = _room_settings() + setattr(settings, field, type(getattr(settings, field)).UNKNOWN) + with pytest.raises(ValueError, match=field): + encode_room_clean_settings((settings,)) + + +@pytest.mark.parametrize( + "rooms", [(), (_room_settings(-1),), (_room_settings(256),), (_room_settings(), _room_settings())] +) +def test_semantic_room_guards(rooms: tuple[Q10RoomCleanSettings, ...]) -> None: + with pytest.raises(ValueError): + encode_room_clean_settings(rooms)