diff --git a/roborock/data/b01_q7/b01_q7_containers.py b/roborock/data/b01_q7/b01_q7_containers.py index 144dd84b8..6fa949175 100644 --- a/roborock/data/b01_q7/b01_q7_containers.py +++ b/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`.""" diff --git a/roborock/devices/traits/b01/q7/__init__.py b/roborock/devices/traits/b01/q7/__init__.py index 295d09e67..f2ae48d7f 100644 --- a/roborock/devices/traits/b01/q7/__init__.py +++ b/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( diff --git a/tests/data/b01_q7/test_b01_q7_containers.py b/tests/data/b01_q7/test_b01_q7_containers.py index 44515139c..8132d1df4 100644 --- a/tests/data/b01_q7/test_b01_q7_containers.py +++ b/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) diff --git a/tests/devices/traits/b01/q7/test_init.py b/tests/devices/traits/b01/q7/test_init.py index 13c02312f..0d9423e01 100644 --- a/tests/devices/traits/b01/q7/test_init.py +++ b/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 == []