feat: add Q10 customized cleaning controls

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
@@ -356,6 +357,35 @@ class RemoteCommand(IntEnum):
     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
+
+
 class Q10RestrictionType(RoborockEnum):
     """Restriction types supported by the Q10 map editor."""
 

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,13 @@ 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."""
+        if self.clean_count is None:
+            return None
+        return Q10CleanCount.from_code_optional(self.clean_count) or Q10CleanCount.UNKNOWN
+
 
 @dataclass
 class SoundVolume(RoborockBase):
@@ -342,3 +352,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

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,6 +113,7 @@ def __init__(self, channel: B01Q10Channel, *, map_parser_config: B01Q10MapParser
         self._map_dps = MapDpsTrait()
         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.clean_history = CleanHistoryTrait(self.command)
         # Read-model traits updated from the device's DPS push stream.
@@ -122,6 +128,7 @@ def __init__(self, channel: B01Q10Channel, *, map_parser_config: B01Q10MapParser
             self.clean_history,
             self._map_dps,
             self.maps,
+            self.room_cleaning,
         ]
         self._subscribe_task: asyncio.Task[None] | None = None
 

roborock/devices/traits/b01/q10/room_cleaning.py

@@ -0,0 +1,69 @@
+"""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
+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: list[Q10RoomCleanSettings] | None = None
+
+
+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."""
+        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:
+        """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()
+
+    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 = list(update.settings)
+        else:
+            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 = list(by_room.values())
+        if settings != self.settings:
+            self.settings = settings
+            self._notify_update()

roborock/devices/traits/b01/q10/vacuum.py

@@ -2,18 +2,21 @@
 
 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 roborock.protocols.b01_q10_protocol import CleanParams, encode_clean_params, encode_room_clean_settings
 
 from .command import CommandTrait
 from .goto import GotoAction, GotoActionCommand, GotoSnapshot
@@ -167,6 +170,18 @@ 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."""
+        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(
         self,
         first_corner: Q10RoborockPoint,
@@ -278,14 +293,36 @@ 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,
         )
 
     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})