Add Q10 zone cleaning and position coordinates

Author: hmmbob

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

@@ -0,0 +1,18 @@
+"""Coordinate conversion helpers for Q10 B01 devices."""
+
+# Q10 trace coordinates are relative to the dock and use 2.5 mm units. The
+# public Roborock actions use millimetres with the dock at (25500, 25500).
+ROBOROCK_COORDINATE_OFFSET = 25500
+Q10_TRACE_UNIT_MM = 2.5
+# Zone and restriction vectors use 5 mm units in the same dock-relative frame.
+Q10_VECTOR_UNIT_MM = 5
+
+
+def trace_to_roborock_coordinate(value: int) -> int:
+    """Convert a Q10 trace value to the common Roborock coordinate space."""
+    return round(ROBOROCK_COORDINATE_OFFSET + value * Q10_TRACE_UNIT_MM)
+
+
+def roborock_to_vector_coordinate(value: int) -> int:
+    """Convert a common Roborock coordinate to the Q10 vector format."""
+    return round((value - ROBOROCK_COORDINATE_OFFSET) / Q10_VECTOR_UNIT_MM)

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

@@ -32,6 +32,7 @@
 from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map
 
 from .common import UpdatableTrait
+from .coordinates import trace_to_roborock_coordinate
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -111,6 +112,16 @@ def robot_position(self) -> Q10Point | None:
         """Current position for live status and caller-rendered map overlays."""
         return self._trace_packet.robot_position if self._trace_packet else None
 
+    @property
+    def roborock_position(self) -> Q10Point | None:
+        """Current position in the common Roborock millimetre coordinate space."""
+        if (position := self.robot_position) is None:
+            return None
+        return Q10Point(
+            x=trace_to_roborock_coordinate(position.x),
+            y=trace_to_roborock_coordinate(position.y),
+        )
+
     @property
     def robot_heading(self) -> int | None:
         """Current heading for orienting a robot marker on a caller-rendered map."""

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

@@ -1,5 +1,9 @@
 """Traits for Q10 B01 devices."""
 
+from base64 import b64encode
+from struct import error as StructError
+from struct import pack
+
 from roborock.data.b01_q10.b01_q10_code_mappings import (
     B01_Q10_DP,
     YXCleanType,
@@ -8,6 +12,41 @@
 )
 
 from .command import CommandTrait
+from .coordinates import roborock_to_vector_coordinate
+
+_ZONE_NAME_FIELD_LENGTH = 19
+
+
+def _encode_zone(x1: int, y1: int, x2: int, y2: int, clean_count: int) -> str:
+    """Encode one rectangular Q10 cleaning zone."""
+    if not 1 <= clean_count <= 3:
+        raise ValueError("clean_count must be between 1 and 3")
+
+    min_x, max_x = sorted((x1, x2))
+    min_y, max_y = sorted((y1, y2))
+    points = (
+        (min_x, min_y),
+        (max_x, min_y),
+        (max_x, max_y),
+        (min_x, max_y),
+    )
+    payload = bytearray((1, clean_count, 1, len(points)))
+    try:
+        for point_x, point_y in points:
+            payload.extend(
+                pack(
+                    ">hh",
+                    roborock_to_vector_coordinate(point_x),
+                    roborock_to_vector_coordinate(point_y),
+                )
+            )
+    except StructError as err:
+        raise ValueError("zone coordinates are outside the supported range") from err
+
+    # The app protocol reserves a fixed 19-byte UTF-8 name field per zone.
+    payload.append(0)
+    payload.extend(bytes(_ZONE_NAME_FIELD_LENGTH))
+    return b64encode(payload).decode()
 
 
 class VacuumTrait:
@@ -56,6 +95,25 @@ async def clean_segments(self, segment_ids: list[int]) -> None:
             params={"cmd": YXDeviceCleanTask.ELECTORAL.code, "clean_paramters": segment_ids},
         )
 
+    async def clean_zone(
+        self,
+        x1: int,
+        y1: int,
+        x2: int,
+        y2: int,
+        *,
+        clean_count: int = 1,
+    ) -> None:
+        """Clean one rectangular zone in the common Roborock coordinate space."""
+        await self._command.send(
+            command=B01_Q10_DP.START_CLEAN,
+            params={
+                "cmd": YXDeviceCleanTask.DIVIDE_AREAS.code,
+                # "clean_paramters" is the spelling required by the firmware.
+                "clean_paramters": _encode_zone(x1, y1, x2, y2, clean_count),
+            },
+        )
+
     async def spot_clean(self) -> None:
         """Start a spot / part clean around the robot's current position.
 

tests/devices/traits/b01/q10/test_map.py

@@ -79,6 +79,8 @@ def test_update_from_trace_packet_populates_path_and_position() -> None:
     assert (trait.path[0].x, trait.path[0].y) == (41, 64)
     assert trait.robot_position is not None
     assert (trait.robot_position.x, trait.robot_position.y) == (276, -1)
+    assert trait.roborock_position is not None
+    assert (trait.roborock_position.x, trait.roborock_position.y) == (26190, 25498)
     assert trait.robot_heading == -34
     assert len(updates) == 1
 

tests/devices/traits/b01/q10/test_vacuum.py

@@ -1,3 +1,4 @@
+from base64 import b64decode
 from collections.abc import Awaitable, Callable
 from typing import Any
 
@@ -49,3 +50,51 @@ async def test_vacuum_commands(
 
     assert command.code == dp_code
     assert params == expected_params
+
+
+async def test_clean_zone(
+    vacuum: VacuumTrait,
+    fake_channel: FakeB01Q10Channel,
+) -> None:
+    """Test the source-verified Q10 zone payload."""
+    await vacuum.clean_zone(25550, 25600, 25650, 25700, clean_count=2)
+
+    command, params = fake_channel.published_commands[0]
+    assert command.code == 201
+    assert params["cmd"] == 3
+    assert b64decode(params["clean_paramters"]) == bytes(
+        (
+            1,
+            2,
+            1,
+            4,
+            0,
+            10,
+            0,
+            20,
+            0,
+            30,
+            0,
+            20,
+            0,
+            30,
+            0,
+            40,
+            0,
+            10,
+            0,
+            40,
+            0,
+            *([0] * 19),
+        )
+    )
+
+
+@pytest.mark.parametrize("clean_count", [0, 4])
+async def test_clean_zone_rejects_invalid_clean_count(
+    vacuum: VacuumTrait,
+    clean_count: int,
+) -> None:
+    """Test that the device clean-count range is validated."""
+    with pytest.raises(ValueError, match="clean_count must be between 1 and 3"):
+        await vacuum.clean_zone(25550, 25600, 25650, 25700, clean_count=clean_count)