feat: add Q10 customized cleaning controls

roborock/cli.py

@@ -603,6 +603,7 @@ async def maps(ctx, device_id: str):
 async def _await_q10_map_push(
     properties: Q10PropertiesApi,
     predicate: Callable[[], bool],
+    revision: Callable[[], int],
     *,
     timeout: float = _Q10_MAP_PUSH_TIMEOUT,
     allow_cached_on_timeout: bool = False,
@@ -615,9 +616,10 @@ async def _await_q10_map_push(
     """
     loop = asyncio.get_running_loop()
     updated: asyncio.Future[None] = loop.create_future()
+    initial_revision = revision()
 
     def on_update() -> None:
-        if predicate() and not updated.done():
+        if revision() > initial_revision and predicate() and not updated.done():
             updated.set_result(None)
 
     unsub = properties.map.add_update_listener(on_update)
@@ -647,6 +649,7 @@ async def map_image(ctx, device_id: str, output_file: str):
         await _await_q10_map_push(
             properties,
             lambda: properties.map.image_content is not None,
+            lambda: properties.map.map_revision,
             allow_cached_on_timeout=True,
         )
         image_content = properties.map.image_content
@@ -695,8 +698,8 @@ async def map_data(ctx, device_id: str, include_path: bool):
 async def q10_position(ctx, device_id: str, include_path: bool):
     """Get the current Q10 robot position and live cleaning path.
 
-    The Q10 only streams its position/path while it is actively cleaning, so this
-    will report that no live trace is available for an idle/docked robot.
+    The Q10 normally streams position/path while it is actively cleaning, so an
+    idle device may report that no fresh live trace is available.
     """
     context: RoborockContext = ctx.obj
     device_manager = await context.get_device_manager()
@@ -708,9 +711,10 @@ async def q10_position(ctx, device_id: str, include_path: bool):
     got_trace = await _await_q10_map_push(
         properties,
         lambda: bool(properties.map.path),
+        lambda: properties.map.trace_revision,
     )
     if not got_trace:
-        click.echo("No live trace available (the robot only reports position while cleaning).")
+        click.echo("No fresh live trace available.")
         return
     map_trait = properties.map
     position = map_trait.robot_position
@@ -873,6 +877,7 @@ async def rooms(ctx, device_id: str):
         await _await_q10_map_push(
             properties,
             lambda: properties.map.image_content is not None,
+            lambda: properties.map.map_revision,
             allow_cached_on_timeout=True,
         )
         click.echo(dump_json({room.id: room.name for room in properties.map.rooms}))

roborock/data/b01_q10/b01_q10_code_mappings.py

@@ -148,6 +148,17 @@ class YXFanLevel(RoborockModeEnum):
     MAX_PLUS = "max_plus", 8  # super
 
 
+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 YXWaterLevel(RoborockModeEnum):
     UNKNOWN = "unknown", -1
     OFF = "off", 0  # close
@@ -162,6 +173,24 @@ class YXCleanLine(RoborockModeEnum):
     FINE = "fine", 2
 
 
+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 YXRoomMaterial(RoborockModeEnum):
     HORIZONTAL_FLOOR_BOARD = "horizontalfloorboard", 0
     VERTICAL_FLOOR_BOARD = "verticalfloorboard", 1

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,
@@ -28,6 +31,69 @@
     YXWaterLevel,
 )
 
+_ROBOROCK_COORDINATE_OFFSET_MM = 25_500
+_Q10_VECTOR_UNIT_MM = 5
+
+
+@dataclass(frozen=True)
+class Q10RoborockPoint:
+    """A point in the common Roborock millimetre coordinate space."""
+
+    x: int
+    y: int
+
+    @classmethod
+    def from_vector(cls, x: int, y: int) -> "Q10RoborockPoint":
+        """Convert Q10 vector coordinates to common Roborock coordinates."""
+        for value in (x, y):
+            if isinstance(value, bool) or not isinstance(value, int):
+                raise ValueError("vector coordinates must be integers")
+            if not -(2**15) <= value < 2**15:
+                raise ValueError("vector coordinates are outside the Q10 map range")
+        return cls(
+            x=_ROBOROCK_COORDINATE_OFFSET_MM + x * _Q10_VECTOR_UNIT_MM,
+            y=_ROBOROCK_COORDINATE_OFFSET_MM + y * _Q10_VECTOR_UNIT_MM,
+        )
+
+    def to_vector(self) -> tuple[int, int]:
+        """Convert common Roborock coordinates to the Q10 vector grid."""
+        coordinates: list[int] = []
+        for value in (self.x, self.y):
+            if isinstance(value, bool) or not isinstance(value, int):
+                raise ValueError("coordinates must be integers")
+            relative_mm = value - _ROBOROCK_COORDINATE_OFFSET_MM
+            if relative_mm % _Q10_VECTOR_UNIT_MM:
+                raise ValueError("coordinates must align to the Q10 5 mm grid")
+            coordinate = relative_mm // _Q10_VECTOR_UNIT_MM
+            if not -(2**15) <= coordinate < 2**15:
+                raise ValueError("coordinates are outside the Q10 map range")
+            coordinates.append(coordinate)
+        return coordinates[0], coordinates[1]
+
+
+@dataclass(frozen=True)
+class Q10RoomCleanSettings:
+    """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
+
+
+@dataclass(frozen=True)
+class Q10ReportedRoomCleanSettings:
+    """Cleaning settings reported by a Q10, preserving unknown wire values."""
+
+    room_id: int
+    fan_level: Q10RoomFanLevel | int
+    water_level: YXWaterLevel | int
+    clean_type: Q10RoomCleanType | int
+    clean_count: Q10CleanCount | int
+    clean_line: YXCleanLine | int
+
 
 @dataclass
 class dpCleanRecord(RoborockBase):
@@ -88,7 +154,9 @@ class Q10MapInfo(RoborockBase):
     """A saved map reported by ``dpMultiMap``.
 
     Q10 firmware represents the map identifier as a string on the wire. The
-    value is sent back unchanged in a subsequent ``{"op": "get"}`` request.
+    value is sent back unchanged in a subsequent ``{"op": "select"}`` detail
+    request. On Q10 firmware, ``select`` previews a saved map without applying
+    it as the active map.
     """
 
     id: str
@@ -230,6 +298,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 without changing the raw status field."""
+        return Q10CleanCount.from_code_optional(self.clean_count) if self.clean_count is not None else None
+
 
 @dataclass
 class SoundVolume(RoborockBase):

roborock/devices/device_manager.py

@@ -19,6 +19,7 @@
 from roborock.devices.device import DeviceReadyCallback, RoborockDevice
 from roborock.diagnostics import Diagnostics, redact_device_data
 from roborock.exceptions import RoborockException
+from roborock.map.b01_q10_map_parser import B01Q10MapParserConfig
 from roborock.map.map_parser import MapParserConfig
 from roborock.mqtt.roborock_session import create_lazy_mqtt_session
 from roborock.mqtt.session import MqttSession, SessionUnauthorizedHook
@@ -262,7 +263,18 @@ def device_creator(home_data: HomeData, device: HomeDataDevice, product: HomeDat
                 if "ss" in model_part:
                     b01_q10_channel = create_b01_q10_channel(mqtt_channel)
                     channel = b01_q10_channel
-                    trait = b01.q10.create(channel)
+                    trait = b01.q10.create(
+                        channel,
+                        model=product.model,
+                        map_parser_config=(
+                            B01Q10MapParserConfig(
+                                map_scale=map_parser_config.map_scale,
+                                drawables=map_parser_config.drawables,
+                            )
+                            if map_parser_config
+                            else None
+                        ),
+                    )
                 elif "sc" in model_part:
                     # Q7 devices start with 'sc' in their model naming.
                     b01_q7_channel = create_b01_q7_channel(device, product, mqtt_channel)

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

@@ -8,7 +8,12 @@
 from roborock.data.containers import RoborockBase
 from roborock.devices.rpc.b01_q10_channel import B01Q10Channel
 from roborock.devices.traits import Trait
-from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10TracePacket
+from roborock.map.b01_q10_map_parser import (
+    B01Q10MapParserConfig,
+    Q10MapPacket,
+    Q10MapPacketKind,
+    Q10TracePacket,
+)
 from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message
 
 from .button_light import ButtonLightTrait
@@ -22,6 +27,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 +43,7 @@
     "MapContentTrait",
     "MapsTrait",
     "NetworkInfoTrait",
+    "RoomCleaningTrait",
     "SoundVolumeTrait",
     "StatusTrait",
 ]
@@ -59,6 +66,9 @@ class Q10PropertiesApi(Trait):
     remote: RemoteTrait
     """Trait for sending remote control related commands to Q10 devices."""
 
+    room_cleaning: RoomCleaningTrait
+    """Trait for customized per-room cleaning settings and commands."""
+
     volume: SoundVolumeTrait
     """Trait for reading / setting the speaker volume."""
 
@@ -92,12 +102,20 @@ class Q10PropertiesApi(Trait):
     clean_history: CleanHistoryTrait
     """Trait for fetching the device clean-record history (``dpCleanRecord``)."""
 
-    def __init__(self, channel: B01Q10Channel) -> None:
+    def __init__(
+        self,
+        channel: B01Q10Channel,
+        *,
+        model: str | None = None,
+        map_parser_config: B01Q10MapParserConfig | None = None,
+    ) -> None:
         """Initialize the B01Props API."""
         self._channel = channel
         self.command = CommandTrait(channel)
-        self.vacuum = VacuumTrait(self.command)
+        advanced_cleaning_supported = model is None or model == "roborock.vacuum.ss07"
+        self.vacuum = VacuumTrait(self.command, advanced_cleaning_supported=advanced_cleaning_supported)
         self.remote = RemoteTrait(self.command)
+        self.room_cleaning = RoomCleaningTrait(self.command, supported=advanced_cleaning_supported)
         self.status = StatusTrait()
         self.volume = SoundVolumeTrait(self.command)
         self.child_lock = ChildLockTrait(self.command)
@@ -107,9 +125,20 @@ def __init__(self, channel: B01Q10Channel) -> None:
         self.network_info = NetworkInfoTrait()
         self.consumable = ConsumableTrait()
         self._map_dps = MapDpsTrait()
-        self.maps = MapsTrait(self.command)
-        self.map = MapContentTrait(self._map_dps, self.maps, self.command)
-        self.clean_history = CleanHistoryTrait(self.command)
+        self.maps = MapsTrait(
+            self.command,
+            map_parser_config=map_parser_config,
+            map_changed_callback=self.room_cleaning.invalidate,
+        )
+        self.map = MapContentTrait(
+            self._map_dps,
+            self.command,
+            map_parser_config=map_parser_config,
+        )
+        self.clean_history = CleanHistoryTrait(
+            self.command,
+            map_parser_config=map_parser_config,
+        )
         # Read-model traits updated from the device's DPS push stream.
         self._updatable_traits = [
             self.status,
@@ -122,6 +151,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
 
@@ -142,7 +172,8 @@ async def close(self) -> None:
     async def refresh(self) -> None:
         """Refresh all traits."""
         # Sending REQUEST_DPS causes the device to publish its ordinary status
-        # values. Map-list and map-content refreshes have separate schedules.
+        # values. Map, map-list, and customized-room settings have separate
+        # refresh methods so callers can schedule those larger responses.
         await self.command.send(B01_Q10_DP.REQUEST_DPS, params={})
 
     async def _subscribe_loop(self) -> None:
@@ -157,7 +188,12 @@ def _handle_message(self, message: Q10Message) -> None:
         Map-list DPS responses and other DPS updates feed the read-model traits.
         """
         if isinstance(message, Q10MapPacket):
-            self.map.update_from_map_packet(message)
+            if message.kind is Q10MapPacketKind.CURRENT:
+                self.map.update_from_map_packet(message)
+            elif message.kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL:
+                self.clean_history.update_from_map_packet(message)
+            elif message.kind is Q10MapPacketKind.SAVED_MAP_DETAIL:
+                self.maps.update_from_map_packet(message)
         elif isinstance(message, Q10TracePacket):
             self.map.update_from_trace_packet(message)
         elif isinstance(message, Q10DpsUpdate):
@@ -178,6 +214,11 @@ def as_dict(self) -> dict[str, Any]:
         return result
 
 
-def create(channel: B01Q10Channel) -> Q10PropertiesApi:
+def create(
+    channel: B01Q10Channel,
+    *,
+    model: str | None = None,
+    map_parser_config: B01Q10MapParserConfig | None = None,
+) -> Q10PropertiesApi:
     """Create traits for B01 devices."""
-    return Q10PropertiesApi(channel)
+    return Q10PropertiesApi(channel, model=model, map_parser_config=map_parser_config)