feat: add Q7 remote control trait

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,6 +78,39 @@ 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`."""

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,
@@ -27,16 +27,22 @@
 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,6 +56,9 @@ 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,
@@ -67,6 +76,7 @@ def __init__(
             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,
@@ -190,6 +200,33 @@ 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 pause_clean(self) -> None:
         """Pause cleaning."""
         await self.send(

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

@@ -0,0 +1,54 @@
+"""Remote (manual) control trait for B01 Q7 devices."""
+
+from roborock.data.b01_q7.b01_q7_code_mappings import Q7RemoteDirection
+from roborock.devices.rpc.b01_q7_channel import Q7RpcChannel
+from roborock.roborock_message import RoborockB01Props
+from roborock.roborock_typing import RoborockB01Q7Methods
+
+# ``sweep_type`` values observed while the official app drives the remote control.
+_MANUAL_SWEEP_TYPE = 5
+_NORMAL_SWEEP_TYPE = 0
+
+
+class RemoteTrait:
+    """Trait for driving a Q7 manually, mirroring the Q10 ``RemoteTrait`` API.
+
+    Manual mode is entered with ``prop.set sweep_type=5`` (the device then reports
+    ``work_mode`` manual). Each movement command only moves the robot for a short
+    burst, so callers repeat it (the official app sends several per second) while
+    a direction is held.
+    """
+
+    def __init__(self, channel: Q7RpcChannel) -> None:
+        """Initialize the RemoteTrait."""
+        self._channel = channel
+
+    async def _send_direction(self, direction: Q7RemoteDirection) -> None:
+        await self._channel.send_command(RoborockB01Q7Methods.SET_DIRECTION, {"direction": direction.code})
+
+    async def stop(self) -> None:
+        """Stop moving, entering manual (remote control) mode if needed."""
+        await self._channel.send_command(
+            RoborockB01Q7Methods.SET_PROP,
+            {RoborockB01Props.SWEEP_TYPE: _MANUAL_SWEEP_TYPE},
+        )
+
+    async def forward(self) -> None:
+        """Move forward."""
+        await self._send_direction(Q7RemoteDirection.FORWARD)
+
+    async def left(self) -> None:
+        """Turn left."""
+        await self._send_direction(Q7RemoteDirection.LEFT)
+
+    async def right(self) -> None:
+        """Turn right."""
+        await self._send_direction(Q7RemoteDirection.RIGHT)
+
+    async def exit_remote(self) -> None:
+        """Exit remote control and return to the normal idle state."""
+        await self._send_direction(Q7RemoteDirection.EXIT)
+        await self._channel.send_command(
+            RoborockB01Q7Methods.SET_PROP,
+            {RoborockB01Props.SWEEP_TYPE: _NORMAL_SWEEP_TYPE},
+        )

tests/data/b01_q7/test_b01_q7_containers.py

@@ -13,6 +13,7 @@
     CleanRecordList,
     CleanRepeatMapping,
     DustCollectionStateMapping,
+    Q7RoborockPoint,
     SCWindMapping,
     StationStateMapping,
     WorkStatusMapping,
@@ -277,3 +278,19 @@ def test_b01_q7_clean_record_list_parses_detail_fields():
     assert detail.record_faultcode == 0
     assert detail.record_dust_num == 0
     assert detail.clean_current_map == 0
+
+
+@pytest.mark.parametrize(
+    ("world", "common"),
+    [
+        ((0.0, 0.0), (25500, 25500)),
+        ((-0.258, 0.025), (25242, 25525)),
+        ((3.5, -3.2), (29000, 22300)),
+    ],
+)
+def test_q7_roborock_point_conversion(world: tuple[float, float], common: tuple[int, int]) -> None:
+    """Test Q7 world meters convert to the common Roborock millimetre space and back."""
+    point = Q7RoborockPoint.from_world(*world)
+
+    assert (point.x, point.y) == common
+    assert point.to_world() == pytest.approx(world, abs=0.001)

tests/devices/traits/b01/q7/test_init.py

@@ -12,7 +12,7 @@
     WaterLevelMapping,
     WorkStatusMapping,
 )
-from roborock.devices.traits.b01.q7 import Q7PropertiesApi
+from roborock.devices.traits.b01.q7 import Q7PropertiesApi, Q7RoborockPoint
 from roborock.roborock_message import RoborockB01Props
 from roborock.roborock_typing import RoborockB01Q7Methods
 
@@ -360,3 +360,52 @@ async def test_q7_api_clean_segments(q7_api: Q7PropertiesApi, fake_channel: Fake
         "ctrl_value": SCDeviceCleanParam.START.code,
         "room_ids": [10, 11],
     }
+
+
+async def test_q7_api_clean_zone(q7_api: Q7PropertiesApi, fake_channel: FakeQ7Channel):
+    """Test zone cleaning converts common coordinates to Q7 world meters."""
+    await q7_api.clean_zone(Q7RoborockPoint(x=27000, y=21900), Q7RoborockPoint(x=26000, y=22700))
+
+    assert fake_channel.published_commands == [
+        (RoborockB01Q7Methods.SET_PREFERENCE_TYPE, {"prefer_type": 4}),
+        (RoborockB01Q7Methods.SET_ZONE_POINTS, {"zone_points": [0.5, -3.6, 1.5, -3.6, 1.5, -2.8, 0.5, -2.8]}),
+        (RoborockB01Q7Methods.SET_ZONE_CLEAN, {"ctrl_value": SCDeviceCleanParam.START.code}),
+    ]
+
+
+async def test_q7_api_clean_zones_multiple(q7_api: Q7PropertiesApi, fake_channel: FakeQ7Channel):
+    """Test several zones are concatenated into a single task."""
+    await q7_api.clean_zones(
+        [
+            (Q7RoborockPoint(x=26000, y=21900), Q7RoborockPoint(x=27000, y=22700)),
+            (Q7RoborockPoint(x=28000, y=21900), Q7RoborockPoint(x=28800, y=22700)),
+        ]
+    )
+
+    _, params = fake_channel.published_commands[1]
+    assert params == {
+        "zone_points": [
+            *[0.5, -3.6, 1.5, -3.6, 1.5, -2.8, 0.5, -2.8],
+            *[2.5, -3.6, 3.3, -3.6, 3.3, -2.8, 2.5, -2.8],
+        ]
+    }
+
+
+@pytest.mark.parametrize(
+    "zones",
+    [
+        [],
+        [(Q7RoborockPoint(x=26000, y=21900), Q7RoborockPoint(x=26000, y=22700))],
+        [(Q7RoborockPoint(x=26000, y=21900), Q7RoborockPoint(x=27000, y=21900))],
+    ],
+)
+async def test_q7_api_clean_zones_invalid(
+    zones: list[tuple[Q7RoborockPoint, Q7RoborockPoint]],
+    q7_api: Q7PropertiesApi,
+    fake_channel: FakeQ7Channel,
+):
+    """Test empty or degenerate zones are rejected before anything is sent."""
+    with pytest.raises(ValueError):
+        await q7_api.clean_zones(zones)
+
+    assert fake_channel.published_commands == []

tests/devices/traits/b01/q7/test_remote.py

@@ -0,0 +1,43 @@
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+import pytest
+
+from roborock.devices.traits.b01.q7 import Q7PropertiesApi, RemoteTrait
+from roborock.roborock_message import RoborockB01Props
+from roborock.roborock_typing import RoborockB01Q7Methods
+
+from .conftest import FakeQ7Channel
+
+
+@pytest.fixture(name="remote")
+def remote_fixture(q7_api: Q7PropertiesApi) -> RemoteTrait:
+    return q7_api.remote
+
+
+@pytest.mark.parametrize(
+    ("command_fn", "expected_commands"),
+    [
+        (lambda x: x.stop(), [(RoborockB01Q7Methods.SET_PROP, {RoborockB01Props.SWEEP_TYPE: 5})]),
+        (lambda x: x.forward(), [(RoborockB01Q7Methods.SET_DIRECTION, {"direction": 1})]),
+        (lambda x: x.left(), [(RoborockB01Q7Methods.SET_DIRECTION, {"direction": 2})]),
+        (lambda x: x.right(), [(RoborockB01Q7Methods.SET_DIRECTION, {"direction": 3})]),
+        (
+            lambda x: x.exit_remote(),
+            [
+                (RoborockB01Q7Methods.SET_DIRECTION, {"direction": 0}),
+                (RoborockB01Q7Methods.SET_PROP, {RoborockB01Props.SWEEP_TYPE: 0}),
+            ],
+        ),
+    ],
+)
+async def test_remote_commands(
+    remote: RemoteTrait,
+    fake_channel: FakeQ7Channel,
+    command_fn: Callable[[RemoteTrait], Awaitable[None]],
+    expected_commands: list[tuple[Any, Any]],
+) -> None:
+    """Test remote control commands are sent with the verified Q7 payloads."""
+    await command_fn(remote)
+
+    assert fake_channel.published_commands == expected_commands