diff --git a/roborock/data/b01_q7/b01_q7_code_mappings.py b/roborock/data/b01_q7/b01_q7_code_mappings.py index df3f8cc0b..e32a213d9 100644 --- a/roborock/data/b01_q7/b01_q7_code_mappings.py +++ b/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.""" diff --git a/roborock/data/b01_q7/b01_q7_containers.py b/roborock/data/b01_q7/b01_q7_containers.py index 144dd84b8..b345ba486 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,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 diff --git a/roborock/devices/traits/b01/q7/__init__.py b/roborock/devices/traits/b01/q7/__init__.py index 295d09e67..0038b86de 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, @@ -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,41 @@ 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 set_carpet_turbo(self, enabled: bool) -> None: + """Enable or disable increased suction on carpets.""" + await self.set_prop(RoborockB01Props.CARPET_TURBO, int(enabled)) + + async def set_carpet_avoid(self, enabled: bool) -> None: + """Enable or disable carpet avoidance.""" + await self.set_prop(RoborockB01Props.CARPET_AVOID, int(enabled)) + async def pause_clean(self) -> None: """Pause cleaning.""" await self.send( diff --git a/roborock/devices/traits/b01/q7/remote.py b/roborock/devices/traits/b01/q7/remote.py new file mode 100644 index 000000000..bf3d1b942 --- /dev/null +++ b/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}, + ) diff --git a/roborock/roborock_message.py b/roborock/roborock_message.py index e3620d069..625174958 100644 --- a/roborock/roborock_message.py +++ b/roborock/roborock_message.py @@ -329,6 +329,9 @@ class RoborockB01Props(StrEnum): SERIAL_NUMBER = "serial_number" RECOMMEND = "recommend" ADD_SWEEP_STATUS = "add_sweep_status" + # Settable individually via prop.set (reported nested under ``privacy`` by prop.get). + CARPET_TURBO = "carpet_turbo" + CARPET_AVOID = "carpet_avoid" ROBOROCK_DATA_STATUS_PROTOCOL = [ diff --git a/tests/data/b01_q7/test_b01_q7_containers.py b/tests/data/b01_q7/test_b01_q7_containers.py index 44515139c..a61aad88e 100644 --- a/tests/data/b01_q7/test_b01_q7_containers.py +++ b/tests/data/b01_q7/test_b01_q7_containers.py @@ -13,6 +13,8 @@ CleanRecordList, CleanRepeatMapping, DustCollectionStateMapping, + Q7MapList, + Q7RoborockPoint, SCWindMapping, StationStateMapping, WorkStatusMapping, @@ -277,3 +279,34 @@ 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) + + +def test_q7_map_list_names() -> None: + """Test map list entries keep the user-visible map name.""" + map_list = Q7MapList.from_dict( + { + "map_list": [ + {"name": "Home", "id": 1791875181, "cur": True}, + {"name": "Temporary map", "id": 0, "cur": False}, + ] + } + ) + + assert [entry.name for entry in map_list.map_list] == ["Home", "Temporary map"] + assert map_list.current_map_id == 1791875181 diff --git a/tests/devices/traits/b01/q7/test_init.py b/tests/devices/traits/b01/q7/test_init.py index 13c02312f..baab91fa8 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,74 @@ 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 == [] + + +@pytest.mark.parametrize( + ("method", "prop"), + [ + ("set_carpet_turbo", RoborockB01Props.CARPET_TURBO), + ("set_carpet_avoid", RoborockB01Props.CARPET_AVOID), + ], +) +@pytest.mark.parametrize(("enabled", "expected_code"), [(True, 1), (False, 0)]) +async def test_q7_api_set_carpet_settings( + method: str, + prop: RoborockB01Props, + enabled: bool, + expected_code: int, + q7_api: Q7PropertiesApi, + fake_channel: FakeQ7Channel, +): + """Test carpet settings are written as individual properties.""" + await getattr(q7_api, method)(enabled) + + assert fake_channel.published_commands == [(RoborockB01Q7Methods.SET_PROP, {prop: expected_code})] diff --git a/tests/devices/traits/b01/q7/test_map.py b/tests/devices/traits/b01/q7/test_map.py index e20115012..96b0f1bc6 100644 --- a/tests/devices/traits/b01/q7/test_map.py +++ b/tests/devices/traits/b01/q7/test_map.py @@ -14,4 +14,4 @@ async def test_q7_map_refresh(q7_api: Q7PropertiesApi, fake_channel: FakeQ7Chann command, params = fake_channel.published_commands[0] assert command == RoborockB01Q7Methods.GET_MAP_LIST assert params == {} - assert q7_api.map.map_list == [Q7MapListEntry(id=111)] + assert q7_api.map.map_list == [Q7MapListEntry(id=111, name="Map 1")] diff --git a/tests/devices/traits/b01/q7/test_remote.py b/tests/devices/traits/b01/q7/test_remote.py new file mode 100644 index 000000000..f2b47ee01 --- /dev/null +++ b/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