feat: add Q7 zone cleaning

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,
@@ -35,8 +35,12 @@
     "MapContentTrait",
     "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."""
@@ -190,6 +194,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(

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 == []