feat: add Q7 goto 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,