feat: expose Q7 robot position

roborock/data/b01_q7/b01_q7_code_mappings.py

@@ -77,6 +77,21 @@ class SCDeviceCleanParam(RoborockModeEnum):
     PAUSE = ("pause", 2)
 
 
+class Q7RemoteDirection(RoborockModeEnum):
+    """Direction codes for ``service.set_direction`` (remote / manual control).
+
+    Verified live on a Q7 (sc05): the robot must first be in manual mode
+    (``prop.set sweep_type=5`` -> ``work_mode`` 2). Each command moves the robot
+    for a short burst, so callers repeat it while a direction is held. Code 0
+    leaves manual mode.
+    """
+
+    EXIT = ("exit", 0)
+    FORWARD = ("forward", 1)
+    LEFT = ("left", 2)
+    RIGHT = ("right", 3)
+
+
 class WorkModeMapping(RoborockModeEnum):
     """Maps the detailed work modes of the robot."""
 

roborock/data/b01_q7/b01_q7_containers.py

@@ -2,6 +2,7 @@
 import json
 from dataclasses import dataclass, field
 from functools import cached_property
+from typing import Self
 
 from ...exceptions import RoborockException
 from ..containers import RoborockBase
@@ -77,12 +78,47 @@ class Recommend(RoborockBase):
     room_id: list[int] = field(default_factory=list)
 
 
+_ROBOROCK_COORDINATE_OFFSET_MM = 25_500
+_MM_PER_METER = 1000
+
+
+@dataclass
+class Q7RoborockPoint(RoborockBase):
+    """A point in the common Roborock millimetre coordinate space.
+
+    Q7 firmware uses world coordinates in meters (floats) whose origin is the
+    map origin next to the dock, with the y axis pointing up. Public Q7 APIs use
+    this common coordinate system instead, matching other Roborock devices and
+    placing the map origin at ``(25500, 25500)``.
+    """
+
+    x: int
+    y: int
+
+    @classmethod
+    def from_world(cls, x: float, y: float) -> Self:
+        """Convert Q7 world coordinates (meters) to common Roborock coordinates."""
+        return cls(
+            x=_ROBOROCK_COORDINATE_OFFSET_MM + round(x * _MM_PER_METER),
+            y=_ROBOROCK_COORDINATE_OFFSET_MM + round(y * _MM_PER_METER),
+        )
+
+    def to_world(self) -> tuple[float, float]:
+        """Convert common Roborock coordinates to Q7 world coordinates (meters)."""
+        return (
+            (self.x - _ROBOROCK_COORDINATE_OFFSET_MM) / _MM_PER_METER,
+            (self.y - _ROBOROCK_COORDINATE_OFFSET_MM) / _MM_PER_METER,
+        )
+
+
 @dataclass
 class Q7MapListEntry(RoborockBase):
     """Single map list entry returned by `service.get_map_list`."""
 
     id: int | None = None
     cur: bool | None = None
+    name: str | None = None
+    """User-visible map name as shown in the app (e.g. ``"Temporary map"`` for an unsaved map)."""
 
 
 @dataclass

roborock/devices/device_manager.py

@@ -282,6 +282,7 @@ def device_creator(home_data: HomeData, device: HomeDataDevice, product: HomeDat
                         device,
                         rpc_channel=b01_q7_channel,
                         map_rpc_channel=b01_q7_channel,
+                        push_channel=b01_q7_channel,
                     )
                 else:
                     raise UnsupportedDeviceError(f"Device {device.name} has unsupported B01 model: {product.model}")

roborock/devices/rpc/b01_q7_channel.py

@@ -16,8 +16,10 @@
     CommandType,
     MapKey,
     ParamsType,
+    Q7DevicePush,
     Q7RequestMessage,
     create_map_key,
+    decode_device_push,
     decode_map_payload,
     decode_rpc_response,
     encode_mqtt_payload,
@@ -58,6 +60,14 @@ async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Calla
         ...
 
 
+class Q7PushChannel(Protocol):
+    """Protocol for channels delivering unsolicited Q7 device pushes."""
+
+    async def subscribe_device_pushes(self, callback: Callable[[Q7DevicePush], None]) -> Callable[[], None]:
+        """Subscribe to ``*.post`` messages the device publishes on its own."""
+        ...
+
+
 def _matches_map_response(response_message: RoborockMessage, *, version: bytes | None) -> bytes | None:
     """Return raw map payload bytes for matching MAP_RESPONSE messages."""
     if (
@@ -166,7 +176,7 @@ def find_response(response_message: RoborockMessage) -> DecodedB01Response | Non
         raise
 
 
-class B01Q7Channel(Channel, Q7RpcChannel, Q7MapRpcChannel):
+class B01Q7Channel(Channel, Q7RpcChannel, Q7MapRpcChannel, Q7PushChannel):
     """Unified B01 Q7 channel wrapping MQTT transport."""
 
     def __init__(self, mqtt_channel: MqttChannel, map_key: MapKey) -> None:
@@ -231,6 +241,15 @@ def on_message(message: RoborockMessage) -> None:
 
         return await self._mqtt_channel.subscribe(on_message)
 
+    async def subscribe_device_pushes(self, callback: Callable[[Q7DevicePush], None]) -> Callable[[], None]:
+        """Subscribe to unsolicited ``*.post`` messages (property changes, live path, events)."""
+
+        def on_message(message: RoborockMessage) -> None:
+            if (push := decode_device_push(message)) is not None:
+                callback(push)
+
+        return await self._mqtt_channel.subscribe(on_message)
+
 
 def create_b01_q7_channel(
     device: HomeDataDevice,

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

@@ -7,7 +7,7 @@
 from typing import Any
 
 from roborock import B01Props
-from roborock.data import HomeDataDevice, HomeDataProduct, Q7MapList, Q7MapListEntry
+from roborock.data import HomeDataDevice, HomeDataProduct, Q7MapList, Q7MapListEntry, Q7RoborockPoint
 from roborock.data.b01_q7.b01_q7_code_mappings import (
     CleanPathPreferenceMapping,
     CleanRepeatMapping,
@@ -17,26 +17,32 @@
     SCWindMapping,
     WaterLevelMapping,
 )
-from roborock.devices.rpc.b01_q7_channel import Q7MapRpcChannel, Q7RpcChannel
+from roborock.devices.rpc.b01_q7_channel import Q7MapRpcChannel, Q7PushChannel, Q7RpcChannel
 from roborock.devices.traits import Trait
 from roborock.exceptions import RoborockException
-from roborock.protocols.b01_q7_protocol import CommandType, ParamsType
+from roborock.protocols.b01_q7_protocol import CommandType, ParamsType, Q7DevicePush
 from roborock.roborock_message import RoborockB01Props
 from roborock.roborock_typing import RoborockB01Q7Methods
 
 from .clean_summary import CleanSummaryTrait
 from .map import MapTrait
 from .map_content import MapContentTrait
+from .remote import RemoteTrait
 
 __all__ = [
     "Q7PropertiesApi",
     "CleanSummaryTrait",
     "MapTrait",
     "MapContentTrait",
+    "RemoteTrait",
     "Q7MapList",
     "Q7MapListEntry",
+    "Q7RoborockPoint",
 ]
 
+# ``service.set_preference_type`` value that selects zone cleaning (``custom_type`` 4).
+_ZONE_PREFERENCE_TYPE = 4
+
 
 class Q7PropertiesApi(Trait):
     """API for interacting with B01 Q7 devices."""
@@ -50,41 +56,56 @@ class Q7PropertiesApi(Trait):
     map_content: MapContentTrait
     """Trait for fetching parsed current map content."""
 
+    remote: RemoteTrait
+    """Trait for remote (manual) control of the robot."""
+
     def __init__(
         self,
         rpc_channel: Q7RpcChannel,
         map_rpc_channel: Q7MapRpcChannel,
+        push_channel: Q7PushChannel,
         device: HomeDataDevice,
         product: HomeDataProduct,
     ) -> None:
         """Initialize the Q7 API."""
         self._rpc_channel = rpc_channel
         self._map_rpc_channel = map_rpc_channel
+        self._push_channel = push_channel
         self._device = device
         self._product = product
 
         if not device.sn or not product.model:
             raise ValueError("B01 Q7 map content requires device serial number and product model metadata")
 
         self.clean_summary = CleanSummaryTrait(rpc_channel)
+        self.remote = RemoteTrait(rpc_channel)
         self.map = MapTrait(rpc_channel)
         self.map_content = MapContentTrait(
             self._map_rpc_channel,
             self.map,
         )
         self._unsub_map_pushes: Callable[[], None] | None = None
+        self._unsub_device_pushes: Callable[[], None] | None = None
 
     async def start(self) -> None:
-        """Start listening for unsolicited map pushes from the device."""
-        if self._unsub_map_pushes is not None:
-            return
-        self._unsub_map_pushes = await self._map_rpc_channel.subscribe_map_pushes(self.map_content.update_from_push)
+        """Start listening for unsolicited map frames and device pushes."""
+        if self._unsub_map_pushes is None:
+            self._unsub_map_pushes = await self._map_rpc_channel.subscribe_map_pushes(self.map_content.update_from_push)
+        if self._unsub_device_pushes is None:
+            self._unsub_device_pushes = await self._push_channel.subscribe_device_pushes(self._handle_device_push)
 
     async def close(self) -> None:
-        """Stop listening for unsolicited map pushes."""
+        """Stop listening for unsolicited map frames and device pushes."""
         if self._unsub_map_pushes is not None:
             self._unsub_map_pushes()
             self._unsub_map_pushes = None
+        if self._unsub_device_pushes is not None:
+            self._unsub_device_pushes()
+            self._unsub_device_pushes = None
+
+    def _handle_device_push(self, push: Q7DevicePush) -> None:
+        """Route an unsolicited device push to the state that depends on it."""
+        self.map_content.update_from_device_push(push)
 
     async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None:
         """Query the device for the values of the given Q7 properties."""
@@ -190,6 +211,41 @@ async def clean_segments(self, segment_ids: list[int]) -> None:
             },
         )
 
+    async def clean_zone(self, first_corner: Q7RoborockPoint, second_corner: Q7RoborockPoint) -> None:
+        """Clean one rectangular zone given two opposite corners in common Roborock coordinates."""
+        await self.clean_zones([(first_corner, second_corner)])
+
+    async def clean_zones(self, zones: list[tuple[Q7RoborockPoint, Q7RoborockPoint]]) -> None:
+        """Clean one or more rectangular zones in a single task.
+
+        Each zone is given by two opposite corners in common Roborock coordinates.
+        The device takes the zones as a flat list of rectangle corners in world
+        meters (``[x1, y1, x2, y1, x2, y2, x1, y2, ...]``) and needs the zone
+        preference type selected first; ``set_zone_clean`` then starts the task.
+        """
+        if not zones:
+            raise ValueError("at least one zone is required")
+        zone_points: list[float] = []
+        for first_corner, second_corner in zones:
+            x1, y1 = first_corner.to_world()
+            x2, y2 = second_corner.to_world()
+            if x1 == x2 or y1 == y2:
+                raise ValueError(f"zone corners must span an area: {first_corner}, {second_corner}")
+            min_x, max_x = sorted((x1, x2))
+            min_y, max_y = sorted((y1, y2))
+            zone_points += [min_x, min_y, max_x, min_y, max_x, max_y, min_x, max_y]
+        await self.send(RoborockB01Q7Methods.SET_PREFERENCE_TYPE, {"prefer_type": _ZONE_PREFERENCE_TYPE})
+        await self.send(RoborockB01Q7Methods.SET_ZONE_POINTS, {"zone_points": zone_points})
+        await self.send(RoborockB01Q7Methods.SET_ZONE_CLEAN, {"ctrl_value": SCDeviceCleanParam.START.code})
+
+    async def set_carpet_turbo(self, enabled: bool) -> None:
+        """Enable or disable increased suction on carpets."""
+        await self.set_prop(RoborockB01Props.CARPET_TURBO, int(enabled))
+
+    async def set_carpet_avoid(self, enabled: bool) -> None:
+        """Enable or disable carpet avoidance."""
+        await self.set_prop(RoborockB01Props.CARPET_AVOID, int(enabled))
+
     async def pause_clean(self) -> None:
         """Pause cleaning."""
         await self.send(
@@ -236,6 +292,7 @@ def create(
     device: HomeDataDevice,
     rpc_channel: Q7RpcChannel,
     map_rpc_channel: Q7MapRpcChannel,
+    push_channel: Q7PushChannel,
 ) -> Q7PropertiesApi:
     """Create traits for B01 Q7 devices."""
     if device.sn is None or product.model is None:
@@ -245,6 +302,7 @@ def create(
     return Q7PropertiesApi(
         rpc_channel=rpc_channel,
         map_rpc_channel=map_rpc_channel,
+        push_channel=push_channel,
         device=device,
         product=product,
     )