From 67ce5a180b217e570801e1bb906b8abe01c17223 Mon Sep 17 00:00:00 2001 From: Ramon Angosto Artigues Date: Thu, 12 Nov 2026 16:43:32 +0100 Subject: [PATCH 1/5] feat: add Q7 zone cleaning Add `clean_zone` and `clean_zones` to the Q7 properties API. Zones are given as two opposite corners in the common Roborock millimetre coordinate space through a new `Q7RoborockPoint`, which converts to and from the Q7 world coordinates (meters, origin next to the dock). The device expects the zone preference type to be selected first (`service.set_preference_type`), then the zone corners as a flat list (`service.set_zone_points`), and finally `service.set_zone_clean` to start. Several zones are concatenated into a single task. --- roborock/data/b01_q7/b01_q7_containers.py | 34 ++++++++++++++ roborock/devices/traits/b01/q7/__init__.py | 33 ++++++++++++- tests/data/b01_q7/test_b01_q7_containers.py | 17 +++++++ tests/devices/traits/b01/q7/test_init.py | 51 ++++++++++++++++++++- 4 files changed, 133 insertions(+), 2 deletions(-) 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 == [] From 6bbd0cc5365c0e51d92ca41e4ed7188db00bfb0e Mon Sep 17 00:00:00 2001 From: Ramon Angosto Artigues Date: Thu, 12 Nov 2026 16:44:38 +0100 Subject: [PATCH 2/5] feat: add Q7 remote control trait Add a `RemoteTrait` to the Q7 properties API, mirroring the Q10 remote API. Manual mode is entered with `prop.set sweep_type=5` and the robot is driven with `service.set_direction` (1 forward, 2 left, 3 right, 0 exit). Each command moves the robot for a short burst, so callers repeat it while a direction is held. Leaving remote mode restores `sweep_type=0`. --- roborock/data/b01_q7/b01_q7_code_mappings.py | 15 ++++++ roborock/devices/traits/b01/q7/__init__.py | 6 +++ roborock/devices/traits/b01/q7/remote.py | 54 ++++++++++++++++++++ tests/devices/traits/b01/q7/test_remote.py | 43 ++++++++++++++++ 4 files changed, 118 insertions(+) create mode 100644 roborock/devices/traits/b01/q7/remote.py create mode 100644 tests/devices/traits/b01/q7/test_remote.py 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/devices/traits/b01/q7/__init__.py b/roborock/devices/traits/b01/q7/__init__.py index f2ae48d7f..792b1fb5b 100644 --- a/roborock/devices/traits/b01/q7/__init__.py +++ b/roborock/devices/traits/b01/q7/__init__.py @@ -27,12 +27,14 @@ 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", @@ -54,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, @@ -71,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, 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/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 From 63c929051d67efd83698a91558af23f3309998a8 Mon Sep 17 00:00:00 2001 From: Ramon Angosto Artigues Date: Thu, 12 Nov 2026 16:45:41 +0100 Subject: [PATCH 3/5] feat: add Q7 carpet settings and map names Add `set_carpet_turbo` and `set_carpet_avoid` to the Q7 properties API. Both are written individually with `prop.set` (`carpet_turbo` / `carpet_avoid`) and reported back nested under `privacy`. Keep the user-visible `name` of each entry returned by `service.get_map_list`, which was previously dropped. --- roborock/data/b01_q7/b01_q7_containers.py | 2 ++ roborock/devices/traits/b01/q7/__init__.py | 8 ++++++++ roborock/roborock_message.py | 3 +++ tests/data/b01_q7/test_b01_q7_containers.py | 16 +++++++++++++++ tests/devices/traits/b01/q7/test_init.py | 22 +++++++++++++++++++++ tests/devices/traits/b01/q7/test_map.py | 2 +- 6 files changed, 52 insertions(+), 1 deletion(-) diff --git a/roborock/data/b01_q7/b01_q7_containers.py b/roborock/data/b01_q7/b01_q7_containers.py index 6fa949175..b345ba486 100644 --- a/roborock/data/b01_q7/b01_q7_containers.py +++ b/roborock/data/b01_q7/b01_q7_containers.py @@ -117,6 +117,8 @@ class Q7MapListEntry(RoborockBase): 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 792b1fb5b..0038b86de 100644 --- a/roborock/devices/traits/b01/q7/__init__.py +++ b/roborock/devices/traits/b01/q7/__init__.py @@ -227,6 +227,14 @@ async def clean_zones(self, zones: list[tuple[Q7RoborockPoint, Q7RoborockPoint]] 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/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 8132d1df4..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,7 @@ CleanRecordList, CleanRepeatMapping, DustCollectionStateMapping, + Q7MapList, Q7RoborockPoint, SCWindMapping, StationStateMapping, @@ -294,3 +295,18 @@ def test_q7_roborock_point_conversion(world: tuple[float, float], common: tuple[ 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 0d9423e01..baab91fa8 100644 --- a/tests/devices/traits/b01/q7/test_init.py +++ b/tests/devices/traits/b01/q7/test_init.py @@ -409,3 +409,25 @@ async def test_q7_api_clean_zones_invalid( 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")] From 42e117a8e6c81cc16e602a06454a38cfb2b6bc40 Mon Sep 17 00:00:00 2001 From: Ramon Angosto Artigues Date: Thu, 12 Nov 2026 16:46:45 +0100 Subject: [PATCH 4/5] feat: expose Q7 robot position Add `MapContentTrait.robot_position` (common Roborock coordinates) for Q7 devices, kept up to date from two sources: - the current pose in live map frames (`currentPose`), ignoring the `(1100, 1100)` and `(0, 0)` placeholders the firmware reports when the pose is unknown or outside the map; - `cur_path` samples in unsolicited `prop.post` device pushes, which the device streams while a task is running. To receive those pushes, `B01Q7Channel` now implements a `Q7PushChannel` protocol that decodes `*.post` messages into `Q7DevicePush`, and `Q7PropertiesApi.start()` subscribes to them. Add `MapContentTrait.refresh_live()`, which requests the live map by type (`service.upload_by_maptype`). Unlike the saved map it carries the current pose and makes the device stream `cur_path` positions. --- roborock/devices/device_manager.py | 1 + roborock/devices/rpc/b01_q7_channel.py | 21 +++- roborock/devices/traits/b01/q7/__init__.py | 27 +++-- roborock/devices/traits/b01/q7/map_content.py | 69 ++++++++++- roborock/map/b01_map_parser.py | 16 +++ roborock/protocols/b01_q7_protocol.py | 41 +++++++ tests/devices/rpc/test_b01_q7_channel.py | 40 ++++++- tests/devices/traits/b01/q7/conftest.py | 20 +++- .../devices/traits/b01/q7/test_map_content.py | 107 +++++++++++++++++- 9 files changed, 325 insertions(+), 17 deletions(-) diff --git a/roborock/devices/device_manager.py b/roborock/devices/device_manager.py index f7d3a150a..4245e3335 100644 --- a/roborock/devices/device_manager.py +++ b/roborock/devices/device_manager.py @@ -282,6 +282,7 @@ def device_creator(home_data: HomeData, device: HomeDataDevice, product: HomeDat device, rpc_channel=b01_q7_channel, map_rpc_channel=b01_q7_channel, + push_channel=b01_q7_channel, ) else: raise UnsupportedDeviceError(f"Device {device.name} has unsupported B01 model: {product.model}") diff --git a/roborock/devices/rpc/b01_q7_channel.py b/roborock/devices/rpc/b01_q7_channel.py index a26611d22..1c3325b03 100644 --- a/roborock/devices/rpc/b01_q7_channel.py +++ b/roborock/devices/rpc/b01_q7_channel.py @@ -16,8 +16,10 @@ CommandType, MapKey, ParamsType, + Q7DevicePush, Q7RequestMessage, create_map_key, + decode_device_push, decode_map_payload, decode_rpc_response, encode_mqtt_payload, @@ -58,6 +60,14 @@ async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Calla ... +class Q7PushChannel(Protocol): + """Protocol for channels delivering unsolicited Q7 device pushes.""" + + async def subscribe_device_pushes(self, callback: Callable[[Q7DevicePush], None]) -> Callable[[], None]: + """Subscribe to ``*.post`` messages the device publishes on its own.""" + ... + + def _matches_map_response(response_message: RoborockMessage, *, version: bytes | None) -> bytes | None: """Return raw map payload bytes for matching MAP_RESPONSE messages.""" if ( @@ -166,7 +176,7 @@ def find_response(response_message: RoborockMessage) -> DecodedB01Response | Non raise -class B01Q7Channel(Channel, Q7RpcChannel, Q7MapRpcChannel): +class B01Q7Channel(Channel, Q7RpcChannel, Q7MapRpcChannel, Q7PushChannel): """Unified B01 Q7 channel wrapping MQTT transport.""" def __init__(self, mqtt_channel: MqttChannel, map_key: MapKey) -> None: @@ -231,6 +241,15 @@ def on_message(message: RoborockMessage) -> None: return await self._mqtt_channel.subscribe(on_message) + async def subscribe_device_pushes(self, callback: Callable[[Q7DevicePush], None]) -> Callable[[], None]: + """Subscribe to unsolicited ``*.post`` messages (property changes, live path, events).""" + + def on_message(message: RoborockMessage) -> None: + if (push := decode_device_push(message)) is not None: + callback(push) + + return await self._mqtt_channel.subscribe(on_message) + def create_b01_q7_channel( device: HomeDataDevice, diff --git a/roborock/devices/traits/b01/q7/__init__.py b/roborock/devices/traits/b01/q7/__init__.py index 0038b86de..ee424a5f5 100644 --- a/roborock/devices/traits/b01/q7/__init__.py +++ b/roborock/devices/traits/b01/q7/__init__.py @@ -17,10 +17,10 @@ SCWindMapping, WaterLevelMapping, ) -from roborock.devices.rpc.b01_q7_channel import Q7MapRpcChannel, Q7RpcChannel +from roborock.devices.rpc.b01_q7_channel import Q7MapRpcChannel, Q7PushChannel, Q7RpcChannel from roborock.devices.traits import Trait from roborock.exceptions import RoborockException -from roborock.protocols.b01_q7_protocol import CommandType, ParamsType +from roborock.protocols.b01_q7_protocol import CommandType, ParamsType, Q7DevicePush from roborock.roborock_message import RoborockB01Props from roborock.roborock_typing import RoborockB01Q7Methods @@ -63,12 +63,14 @@ def __init__( self, rpc_channel: Q7RpcChannel, map_rpc_channel: Q7MapRpcChannel, + push_channel: Q7PushChannel, device: HomeDataDevice, product: HomeDataProduct, ) -> None: """Initialize the Q7 API.""" self._rpc_channel = rpc_channel self._map_rpc_channel = map_rpc_channel + self._push_channel = push_channel self._device = device self._product = product @@ -83,18 +85,27 @@ def __init__( self.map, ) self._unsub_map_pushes: Callable[[], None] | None = None + self._unsub_device_pushes: Callable[[], None] | None = None async def start(self) -> None: - """Start listening for unsolicited map pushes from the device.""" - if self._unsub_map_pushes is not None: - return - self._unsub_map_pushes = await self._map_rpc_channel.subscribe_map_pushes(self.map_content.update_from_push) + """Start listening for unsolicited map frames and device pushes.""" + if self._unsub_map_pushes is None: + self._unsub_map_pushes = await self._map_rpc_channel.subscribe_map_pushes(self.map_content.update_from_push) + if self._unsub_device_pushes is None: + self._unsub_device_pushes = await self._push_channel.subscribe_device_pushes(self._handle_device_push) async def close(self) -> None: - """Stop listening for unsolicited map pushes.""" + """Stop listening for unsolicited map frames and device pushes.""" if self._unsub_map_pushes is not None: self._unsub_map_pushes() self._unsub_map_pushes = None + if self._unsub_device_pushes is not None: + self._unsub_device_pushes() + self._unsub_device_pushes = None + + def _handle_device_push(self, push: Q7DevicePush) -> None: + """Route an unsolicited device push to the state that depends on it.""" + self.map_content.update_from_device_push(push) async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None: """Query the device for the values of the given Q7 properties.""" @@ -281,6 +292,7 @@ def create( device: HomeDataDevice, rpc_channel: Q7RpcChannel, map_rpc_channel: Q7MapRpcChannel, + push_channel: Q7PushChannel, ) -> Q7PropertiesApi: """Create traits for B01 Q7 devices.""" if device.sn is None or product.model is None: @@ -290,6 +302,7 @@ def create( return Q7PropertiesApi( rpc_channel=rpc_channel, map_rpc_channel=map_rpc_channel, + push_channel=push_channel, device=device, product=product, ) diff --git a/roborock/devices/traits/b01/q7/map_content.py b/roborock/devices/traits/b01/q7/map_content.py index 56c180ce3..9b378146f 100644 --- a/roborock/devices/traits/b01/q7/map_content.py +++ b/roborock/devices/traits/b01/q7/map_content.py @@ -15,15 +15,17 @@ import asyncio import logging from dataclasses import dataclass +from typing import Any from vacuum_map_parser_base.map_data import MapData -from roborock.data import RoborockBase +from roborock.data import Q7RoborockPoint, RoborockBase from roborock.devices.rpc.b01_q7_channel import Q7MapRpcChannel from roborock.devices.traits import Trait from roborock.devices.traits.common import TraitUpdateListener from roborock.exceptions import RoborockException -from roborock.map.b01_map_parser import B01MapParser, B01MapParserConfig, parse_map_type +from roborock.map.b01_map_parser import B01MapParser, B01MapParserConfig, parse_map_type, parse_robot_pose +from roborock.protocols.b01_q7_protocol import Q7DevicePush from roborock.roborock_typing import RoborockB01Q7Methods from .map import MapTrait @@ -50,11 +52,34 @@ class MapContent(RoborockBase): library to re-parse the map data when needed. """ + robot_position: Q7RoborockPoint | None = None + """Latest known robot position in common Roborock coordinates. + + Updated from map frames and, while the robot moves, from the live + ``cur_path`` pushes the device publishes roughly once per second. + """ + def __repr__(self) -> str: img = self.image_content if img and len(img) > _TRUNCATE_LENGTH: img = img[: _TRUNCATE_LENGTH - 3] + b"..." - return f"MapContent(image_content={img!r}, map_data={self.map_data!r})" + return f"MapContent(image_content={img!r}, map_data={self.map_data!r}, robot_position={self.robot_position!r})" + + +def latest_position_from_cur_path(cur_path: list[Any]) -> Q7RoborockPoint | None: + """Return the newest pose of a ``cur_path`` push, or ``None`` if malformed. + + The device sends ``[start_index, x, y, phi, kind, (x, y, phi, kind)..., timestamp]`` + with world coordinates in meters; ``kind`` is 0 while travelling and 5 while + cleaning. + """ + samples = cur_path[1:-1] + if not samples or len(samples) % 4: + return None + x, y = samples[-4], samples[-3] + if not all(isinstance(value, int | float) and not isinstance(value, bool) for value in (x, y)): + return None + return Q7RoborockPoint.from_world(x, y) class MapContentTrait(MapContent, Trait, TraitUpdateListener): @@ -93,6 +118,24 @@ async def refresh(self) -> None: self._parse_and_store(raw_payload) + async def refresh_live(self) -> None: + """Fetch and parse the live map of the current task, including the robot pose. + + Unlike :meth:`refresh` (the saved map, which carries no live pose), this + requests the live map (``service.upload_by_maptype`` with map type 0), as the + official app does. Requesting it also makes the device stream ``cur_path`` + pushes while it moves (verified on a Q7 sc05: without a live-map request no + ``cur_path`` is published), keeping :attr:`robot_position` current about + once per second. + """ + async with self._map_command_lock: + raw_payload = await self._map_rpc_channel.send_map_command( + RoborockB01Q7Methods.UPLOAD_BY_MAPTYPE, + {"map_type": _LIVE_MAP_TYPE}, + ) + + self._parse_and_store(raw_payload) + def update_from_push(self, raw_payload: bytes) -> None: """Store an unsolicited SCMap frame pushed by the device during cleaning. @@ -132,3 +175,23 @@ def _parse_and_store(self, raw_payload: bytes) -> None: self.image_content = parsed_data.image_content self.map_data = parsed_data.map_data self.raw_api_response = raw_payload + # Saved maps and frames without a fix carry no usable pose: keep the last known one. + try: + pose = parse_robot_pose(raw_payload) + except RoborockException as ex: + _LOGGER.debug("Failed to read robot pose from B01 map frame: %s", ex) + return + if pose is not None: + self.robot_position = Q7RoborockPoint.from_world(*pose) + + def update_from_device_push(self, push: Q7DevicePush) -> None: + """Update the robot position from a live ``cur_path`` push and notify listeners.""" + if push.method != "prop.post" or not isinstance(cur_path := push.params.get("cur_path"), list): + return + if (position := latest_position_from_cur_path(cur_path)) is None: + _LOGGER.debug("Ignoring malformed Q7 cur_path push: %s", cur_path) + return + if position == self.robot_position: + return + self.robot_position = position + self._notify_update() diff --git a/roborock/map/b01_map_parser.py b/roborock/map/b01_map_parser.py index 7ca00ea1f..9b7ce0aec 100644 --- a/roborock/map/b01_map_parser.py +++ b/roborock/map/b01_map_parser.py @@ -113,6 +113,22 @@ def parse_map_type(payload: bytes) -> int: return _parse_scmap_payload(payload).mapType +def parse_robot_pose(payload: bytes) -> tuple[float, float] | None: + """Return the robot's world position (meters) from an SCMap frame, if valid. + + Frames without a live pose yield ``None``, as do the placeholders the firmware + reports while it has no fix: ``(1100.0, 1100.0)`` (outside the map bounds) and + an exact ``(0.0, 0.0)`` while a task is starting. + """ + parsed = _parse_scmap_payload(payload) + if not parsed.HasField("mapHead") or not parsed.HasField("currentPose"): + return None + pose = parsed.currentPose + if (pose.x == 0.0 and pose.y == 0.0) or not _WorldToPixel(parsed).in_bounds(pose.x, pose.y): + return None + return pose.x, pose.y + + def _parse_scmap_payload(payload: bytes) -> RobotMap: """Parse inflated SCMap bytes into a generated protobuf message.""" parsed = RobotMap() diff --git a/roborock/protocols/b01_q7_protocol.py b/roborock/protocols/b01_q7_protocol.py index 1eac3dcb7..9bb70ff14 100644 --- a/roborock/protocols/b01_q7_protocol.py +++ b/roborock/protocols/b01_q7_protocol.py @@ -89,6 +89,47 @@ def decode_rpc_response(message: RoborockMessage) -> dict[int, Any]: ) from err +@dataclass +class Q7DevicePush: + """An unsolicited message the device publishes on its own. + + Examples observed on a Q7 (sc05): ``prop.post`` with changed properties + (``{"status": 5, "fault": 2110}``, ``{"cur_path": [...]}``) and events such + as ``event.startClean.post``. + """ + + method: str + params: dict[str, Any] + + +def decode_device_push(message: RoborockMessage) -> Q7DevicePush | None: + """Return the device push carried by ``message``, or ``None`` if it is not one. + + Pushes are distinguished from command responses by their ``*.post`` method + and their ``params`` payload (responses carry ``data`` and a ``code``). + """ + if message.protocol != RoborockMessageProtocol.RPC_RESPONSE or message.version != B01_VERSION: + return None + try: + datapoints = decode_rpc_response(message) + except RoborockException: + return None + for value in datapoints.values(): + if not isinstance(value, str): + continue + try: + inner = json.loads(value) + except json.JSONDecodeError: + continue + if not isinstance(inner, dict): + continue + method = inner.get("method") + params = inner.get("params") + if isinstance(method, str) and method.endswith(".post") and isinstance(params, dict): + return Q7DevicePush(method=method, params=params) + return None + + @dataclass class MapKey: """Data class for holding a B01 map decryption key.""" diff --git a/tests/devices/rpc/test_b01_q7_channel.py b/tests/devices/rpc/test_b01_q7_channel.py index 087291d9c..c2cb0b766 100644 --- a/tests/devices/rpc/test_b01_q7_channel.py +++ b/tests/devices/rpc/test_b01_q7_channel.py @@ -16,7 +16,7 @@ send_decoded_command, ) from roborock.exceptions import RoborockException -from roborock.protocols.b01_q7_protocol import B01_VERSION, Q7RequestMessage +from roborock.protocols.b01_q7_protocol import B01_VERSION, Q7DevicePush, Q7RequestMessage from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol from tests.fixtures.channel_fixtures import FakeChannel @@ -52,6 +52,17 @@ def _build_dps(self, message: dict[str, Any] | str) -> RoborockMessage: seq=self.seq, ) + def build_push(self, method: str, params: dict[str, Any]) -> RoborockMessage: + """Build an unsolicited device push (``*.post``) as published on dps 10001.""" + self.seq += 1 + message = {"msgId": "323022", "method": method, "params": params} + return RoborockMessage( + protocol=RoborockMessageProtocol.RPC_RESPONSE, + payload=pad(json.dumps({"dps": {"10001": json.dumps(message)}}).encode(), AES.block_size), + version=b"B01", + seq=self.seq, + ) + def build_map_response(self, payload: bytes) -> RoborockMessage: """Build a dummy MAP_RESPONSE message.""" self.seq += 1 @@ -329,6 +340,33 @@ async def test_subscribe_map_pushes_delivers_decoded_frames( assert received == [b"inflated-payload"] +async def test_subscribe_device_pushes_delivers_post_messages( + device: HomeDataDevice, + product: HomeDataProduct, + fake_channel: FakeChannel, + message_builder: B01MessageBuilder, +) -> None: + """``*.post`` messages are delivered; command responses and map frames are not.""" + channel = create_b01_q7_channel(device, product, fake_channel) # type: ignore[arg-type] + + received: list[Q7DevicePush] = [] + unsub = await channel.subscribe_device_pushes(received.append) + + fake_channel.notify_subscribers(message_builder.build_push("prop.post", {"status": 5, "fault": 2110})) + fake_channel.notify_subscribers(message_builder.build_push("event.startClean.post", {})) + fake_channel.notify_subscribers(message_builder.build({"status": 1})) + fake_channel.notify_subscribers(message_builder.build_map_response(b"raw-map-payload")) + + assert received == [ + Q7DevicePush(method="prop.post", params={"status": 5, "fault": 2110}), + Q7DevicePush(method="event.startClean.post", params={}), + ] + + unsub() + fake_channel.notify_subscribers(message_builder.build_push("prop.post", {"quantity": 89})) + assert len(received) == 2 + + async def test_subscribe_map_pushes_skips_undecodable_frames( device: HomeDataDevice, product: HomeDataProduct, diff --git a/tests/devices/traits/b01/q7/conftest.py b/tests/devices/traits/b01/q7/conftest.py index 03f36db6f..10ab4fc71 100644 --- a/tests/devices/traits/b01/q7/conftest.py +++ b/tests/devices/traits/b01/q7/conftest.py @@ -4,12 +4,13 @@ import pytest from roborock.data import HomeDataDevice, HomeDataProduct, RoborockCategory -from roborock.devices.rpc.b01_q7_channel import Q7MapRpcChannel, Q7RpcChannel +from roborock.devices.rpc.b01_q7_channel import Q7MapRpcChannel, Q7PushChannel, Q7RpcChannel from roborock.devices.traits.b01.q7 import Q7PropertiesApi, create +from roborock.protocols.b01_q7_protocol import Q7DevicePush -class FakeQ7Channel(Q7RpcChannel, Q7MapRpcChannel): - """A plaintext mock for Q7 Rpc and Map Channel.""" +class FakeQ7Channel(Q7RpcChannel, Q7MapRpcChannel, Q7PushChannel): + """A plaintext mock for Q7 Rpc, Map and device-push channels.""" def __init__(self) -> None: self.published_commands: list[tuple[Any, Any]] = [] @@ -17,6 +18,8 @@ def __init__(self) -> None: self.side_effect: Exception | None = None self.map_push_callback: Callable[[bytes], None] | None = None self.map_push_subscribe_count = 0 + self.device_push_callback: Callable[[Q7DevicePush], None] | None = None + self.device_push_subscribe_count = 0 async def send_command(self, command: Any, params: Any = None) -> Any: if self.side_effect: @@ -41,6 +44,15 @@ def unsub() -> None: return unsub + async def subscribe_device_pushes(self, callback: Callable[[Q7DevicePush], None]) -> Callable[[], None]: + self.device_push_subscribe_count += 1 + self.device_push_callback = callback + + def unsub() -> None: + self.device_push_callback = None + + return unsub + @pytest.fixture(name="fake_channel") def fake_channel_fixture() -> FakeQ7Channel: @@ -74,4 +86,4 @@ def q7_api_fixture( device: HomeDataDevice, product: HomeDataProduct, ) -> Q7PropertiesApi: - return create(product, device, fake_channel, fake_channel) + return create(product, device, fake_channel, fake_channel, fake_channel) diff --git a/tests/devices/traits/b01/q7/test_map_content.py b/tests/devices/traits/b01/q7/test_map_content.py index e07e7720e..5529d5bdb 100644 --- a/tests/devices/traits/b01/q7/test_map_content.py +++ b/tests/devices/traits/b01/q7/test_map_content.py @@ -3,10 +3,11 @@ import pytest from vacuum_map_parser_base.map_data import MapData -from roborock.devices.traits.b01.q7 import Q7PropertiesApi +from roborock.devices.traits.b01.q7 import Q7PropertiesApi, Q7RoborockPoint from roborock.exceptions import RoborockException from roborock.map.b01_map_parser import ParsedMapData from roborock.map.proto.b01_scmap_pb2 import RobotMap # type: ignore[attr-defined] +from roborock.protocols.b01_q7_protocol import Q7DevicePush from roborock.roborock_typing import RoborockB01Q7Methods from .conftest import FakeQ7Channel @@ -205,5 +206,109 @@ async def test_q7_start_is_idempotent( await q7_api.start() assert fake_channel.map_push_subscribe_count == 1 + assert fake_channel.device_push_subscribe_count == 1 await q7_api.close() assert fake_channel.map_push_callback is None + assert fake_channel.device_push_callback is None + + +def cur_path_push(*samples: tuple[float, float, float, int]) -> Q7DevicePush: + """Build a ``prop.post`` push carrying a ``cur_path`` with the given poses.""" + flat = [value for sample in samples for value in sample] + return Q7DevicePush(method="prop.post", params={"cur_path": [35, *flat, 1794476592], "real_clean_time": 7}) + + +async def test_q7_robot_position_from_cur_path_push( + q7_api: Q7PropertiesApi, + fake_channel: FakeQ7Channel, +): + """Live cur_path pushes update the robot position (latest sample) and notify listeners.""" + await q7_api.start() + assert fake_channel.device_push_callback is not None + updates: list[Q7RoborockPoint | None] = [] + q7_api.map_content.add_update_listener(lambda: updates.append(q7_api.map_content.robot_position)) + + fake_channel.device_push_callback(cur_path_push((-0.2, -3.0, -2.7, 5), (-0.25, -3.1, -1.9, 5))) + # The same position again is not a change. + fake_channel.device_push_callback(cur_path_push((-0.25, -3.1, -1.9, 5))) + + assert updates == [Q7RoborockPoint(x=25250, y=22400)] + assert fake_channel.published_commands == [] + await q7_api.close() + + +@pytest.mark.parametrize( + "push", + [ + Q7DevicePush(method="prop.post", params={"real_clean_time": 7}), + Q7DevicePush(method="prop.post", params={"cur_path": [35, 1794476592]}), + Q7DevicePush(method="prop.post", params={"cur_path": [35, -0.2, -3.0, 1794476592]}), + Q7DevicePush(method="prop.post", params={"cur_path": [35, "x", "y", 0.0, 5, 1794476592]}), + Q7DevicePush(method="event.startClean.post", params={"cur_path": [35, 1.0, 1.0, 0.0, 5, 1]}), + ], +) +async def test_q7_robot_position_ignores_unrelated_or_malformed_pushes( + push: Q7DevicePush, + q7_api: Q7PropertiesApi, + fake_channel: FakeQ7Channel, +): + """Pushes without a well-formed cur_path leave the position untouched.""" + await q7_api.start() + assert fake_channel.device_push_callback is not None + updates: list[bool] = [] + q7_api.map_content.add_update_listener(lambda: updates.append(True)) + + fake_channel.device_push_callback(push) + + assert q7_api.map_content.robot_position is None + assert updates == [] + await q7_api.close() + + +def scmap_frame_with_pose(x: float, y: float) -> bytes: + """Serialize a live SCMap frame with a map head and a current pose.""" + frame = RobotMap(mapType=0) + frame.mapHead.minX, frame.mapHead.minY = -10.0, -13.0 + frame.mapHead.maxX, frame.mapHead.maxY = 6.0, 5.0 + frame.currentPose.x, frame.currentPose.y = x, y + return frame.SerializeToString() + + +async def test_q7_robot_position_from_map_frames_ignores_placeholder( + q7_api: Q7PropertiesApi, + fake_channel: FakeQ7Channel, +): + """A frame pose inside the map sets the position; the (1100, 1100) placeholder does not.""" + await q7_api.start() + assert fake_channel.map_push_callback is not None + parsed_map_data = ParsedMapData(image_content=b"pngbytes", map_data=MapData()) + + with patch("roborock.devices.traits.b01.q7.map_content.B01MapParser.parse", return_value=parsed_map_data): + fake_channel.map_push_callback(scmap_frame_with_pose(2.89, -2.78)) + assert q7_api.map_content.robot_position == Q7RoborockPoint(x=28390, y=22720) + + fake_channel.map_push_callback(scmap_frame_with_pose(1100.0, 1100.0)) + assert q7_api.map_content.robot_position == Q7RoborockPoint(x=28390, y=22720) + + fake_channel.map_push_callback(scmap_frame_with_pose(0.0, 0.0)) + assert q7_api.map_content.robot_position == Q7RoborockPoint(x=28390, y=22720) + + await q7_api.close() + + +async def test_q7_map_content_refresh_live_requests_live_map( + q7_api: Q7PropertiesApi, + fake_channel: FakeQ7Channel, +): + """refresh_live() requests the live map by type (no map list needed) and stores the pose.""" + payload = scmap_frame_with_pose(1.87, -3.45) + fake_channel.response_queue.append(payload) + parsed_map_data = ParsedMapData(image_content=b"pngbytes", map_data=MapData()) + + with patch("roborock.devices.traits.b01.q7.map_content.B01MapParser.parse", return_value=parsed_map_data): + await q7_api.map_content.refresh_live() + + assert fake_channel.published_commands == [(RoborockB01Q7Methods.UPLOAD_BY_MAPTYPE, {"map_type": 0})] + assert q7_api.map_content.image_content == b"pngbytes" + assert q7_api.map_content.raw_api_response == payload + assert q7_api.map_content.robot_position == Q7RoborockPoint(x=27370, y=22050) From 1f77ea9e3bf57c37011ba3d1e32a8b21588970aa Mon Sep 17 00:00:00 2001 From: Ramon Angosto Artigues Date: Thu, 12 Nov 2026 16:47:41 +0100 Subject: [PATCH 5/5] feat: add Q7 goto position Add `Q7PropertiesApi.goto_position`. The Q7 firmware has no goto command, so (like the Q10) it is emulated with a small zone-clean task around the target that is stopped once the live position is within 30 cm, leaving the robot there. A `GotoAction` state machine decides when to stop, based on the robot position and the `work_mode` reported in device pushes: - near the target while the zone task runs: stop the task; - the zone finished and the robot heads home (`PLAN_AREA_GO_HOME`) before a close position arrived: stop, since it already reached the target; - the zone task ended or was replaced any other way: complete silently. While a goto is active the live map is requested every 3 s, because `cur_path` pushes are not always streamed. A 300 s safety timeout stops a zone task that is still running. Any other cleaning command cancels the goto. --- roborock/devices/traits/b01/q7/__init__.py | 132 ++++++++++++++++ roborock/devices/traits/b01/q7/goto.py | 122 +++++++++++++++ tests/devices/traits/b01/q7/test_goto.py | 166 +++++++++++++++++++++ 3 files changed, 420 insertions(+) create mode 100644 roborock/devices/traits/b01/q7/goto.py create mode 100644 tests/devices/traits/b01/q7/test_goto.py diff --git a/roborock/devices/traits/b01/q7/__init__.py b/roborock/devices/traits/b01/q7/__init__.py index ee424a5f5..e4c2c3943 100644 --- a/roborock/devices/traits/b01/q7/__init__.py +++ b/roborock/devices/traits/b01/q7/__init__.py @@ -3,6 +3,8 @@ Potentially other devices may fall into this category in the future. """ +import asyncio +import logging from collections.abc import Callable from typing import Any @@ -16,6 +18,7 @@ SCDeviceCleanParam, SCWindMapping, WaterLevelMapping, + WorkModeMapping, ) from roborock.devices.rpc.b01_q7_channel import Q7MapRpcChannel, Q7PushChannel, Q7RpcChannel from roborock.devices.traits import Trait @@ -25,6 +28,7 @@ from roborock.roborock_typing import RoborockB01Q7Methods from .clean_summary import CleanSummaryTrait +from .goto import GotoAction, GotoActionCommand, GotoSnapshot from .map import MapTrait from .map_content import MapContentTrait from .remote import RemoteTrait @@ -40,9 +44,21 @@ "Q7RoborockPoint", ] +_LOGGER = logging.getLogger(__name__) + # ``service.set_preference_type`` value that selects zone cleaning (``custom_type`` 4). _ZONE_PREFERENCE_TYPE = 4 +# Emulated goto: a small zone around the target, stopped once the robot is close. +_GOTO_HALF_ZONE_MM = 250 +_GOTO_TOLERANCE_MM = 300 +# Leaving the dock alone takes ~30 s before the robot moves. +_GOTO_TIMEOUT = 300 +_GOTO_RETRY_INTERVAL = 1 +# While a goto is active the live map is requested periodically: it carries the +# current pose and keeps the device streaming ``cur_path`` positions. +_GOTO_LIVE_MAP_INTERVAL = 3 + class Q7PropertiesApi(Trait): """API for interacting with B01 Q7 devices.""" @@ -86,6 +102,12 @@ def __init__( ) self._unsub_map_pushes: Callable[[], None] | None = None self._unsub_device_pushes: Callable[[], None] | None = None + self._remove_map_listener: Callable[[], None] | None = None + self._work_mode: WorkModeMapping | None = None + self._goto_action: GotoAction | None = None + self._goto_action_remove_listener: Callable[[], None] | None = None + self._goto_track_task: asyncio.Task[None] | None = None + self._goto_command_task: asyncio.Task[None] | None = None async def start(self) -> None: """Start listening for unsolicited map frames and device pushes.""" @@ -93,19 +115,123 @@ async def start(self) -> None: self._unsub_map_pushes = await self._map_rpc_channel.subscribe_map_pushes(self.map_content.update_from_push) if self._unsub_device_pushes is None: self._unsub_device_pushes = await self._push_channel.subscribe_device_pushes(self._handle_device_push) + if self._remove_map_listener is None: + self._remove_map_listener = self.map_content.add_update_listener(self._goto_source_updated) async def close(self) -> None: """Stop listening for unsolicited map frames and device pushes.""" + self.cancel_goto() if self._unsub_map_pushes is not None: self._unsub_map_pushes() self._unsub_map_pushes = None if self._unsub_device_pushes is not None: self._unsub_device_pushes() self._unsub_device_pushes = None + if self._remove_map_listener is not None: + self._remove_map_listener() + self._remove_map_listener = None def _handle_device_push(self, push: Q7DevicePush) -> None: """Route an unsolicited device push to the state that depends on it.""" self.map_content.update_from_device_push(push) + if push.method == "prop.post" and isinstance(work_mode := push.params.get("work_mode"), int): + self._work_mode = WorkModeMapping.from_code_optional(work_mode) + self._goto_source_updated() + + def cancel_goto(self) -> None: + """Cancel an emulated goto, e.g. because another command replaced it.""" + if self._goto_action is not None: + self._goto_action.complete() + self._goto_action = None + if self._goto_action_remove_listener is not None: + self._goto_action_remove_listener() + self._goto_action_remove_listener = None + current_task = asyncio.current_task() + for task_name in ("_goto_track_task", "_goto_command_task"): + if (task := getattr(self, task_name)) is not None: + if task is not current_task: + task.cancel() + setattr(self, task_name, None) + + def _goto_snapshot(self) -> GotoSnapshot: + """Return the latest push-derived state used by an active goto action.""" + return GotoSnapshot(position=self.map_content.robot_position, work_mode=self._work_mode) + + def _goto_source_updated(self) -> None: + """Feed the latest position / work mode to the active goto action.""" + if self._goto_action is not None: + self._goto_action.update(self._goto_snapshot()) + + def _goto_action_updated(self, action: GotoAction, command: GotoActionCommand) -> None: + """Schedule a device command requested by the active goto action.""" + if action is not self._goto_action: + return + if command is GotoActionCommand.COMPLETE: + self.cancel_goto() + return + if self._goto_command_task is None: + self._goto_command_task = asyncio.create_task( + self._async_stop_goto_zone(action), + name="roborock_q7_goto_stop", + ) + + async def _async_stop_goto_zone(self, action: GotoAction) -> None: + """Stop the goto zone task, leaving the robot where it is.""" + current_task = asyncio.current_task() + try: + await self.send(RoborockB01Q7Methods.SET_ZONE_CLEAN, {"ctrl_value": SCDeviceCleanParam.STOP.code}) + except RoborockException as err: + _LOGGER.warning("Failed to stop Q7 goto zone task; retrying: %s", err) + await asyncio.sleep(_GOTO_RETRY_INTERVAL) + if action is self._goto_action: + self._goto_command_task = None + action.retry() + return + if action is self._goto_action: + action.complete() + self.cancel_goto() + if self._goto_command_task is current_task: + self._goto_command_task = None + + async def _async_track_goto(self, action: GotoAction) -> None: + """Keep live positions flowing while the goto is active, then apply the safety timeout.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + _GOTO_TIMEOUT + try: + while action is self._goto_action and loop.time() < deadline: + try: + await self.map_content.refresh_live() + except RoborockException as err: + _LOGGER.debug("Failed to refresh the Q7 live map during goto: %s", err) + await asyncio.sleep(_GOTO_LIVE_MAP_INTERVAL) + except asyncio.CancelledError: + return + if action is self._goto_action: + action.timeout(self._goto_snapshot()) + + async def goto_position(self, target: Q7RoborockPoint) -> None: + """Move to a coordinate (common Roborock coordinates), then stay there. + + The firmware has no goto command, so (like the Q10) this starts a small + zone-clean task around ``target`` and stops it once the live position is + within ``_GOTO_TOLERANCE_MM``. Requires :meth:`start` so that live + positions are received. Any other cleaning command cancels the goto. + """ + probe = GotoAction(target, tolerance=_GOTO_TOLERANCE_MM) + if probe.is_near_target(self.map_content.robot_position): + return + + await self.clean_zone( + Q7RoborockPoint(x=target.x - _GOTO_HALF_ZONE_MM, y=target.y - _GOTO_HALF_ZONE_MM), + Q7RoborockPoint(x=target.x + _GOTO_HALF_ZONE_MM, y=target.y + _GOTO_HALF_ZONE_MM), + ) + action = probe + self._goto_action = action + self._goto_action_remove_listener = action.add_update_listener( + lambda command: self._goto_action_updated(action, command) + ) + self._goto_track_task = asyncio.create_task(self._async_track_goto(action), name="roborock_q7_goto_track") + action.update(self._goto_snapshot()) async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None: """Query the device for the values of the given Q7 properties.""" @@ -199,6 +325,7 @@ async def start_clean(self) -> None: "room_ids": [], }, ) + self.cancel_goto() async def clean_segments(self, segment_ids: list[int]) -> None: """Start segment cleaning for the given ids (Q7 uses room ids).""" @@ -210,6 +337,7 @@ async def clean_segments(self, segment_ids: list[int]) -> None: "room_ids": segment_ids, }, ) + self.cancel_goto() async def clean_zone(self, first_corner: Q7RoborockPoint, second_corner: Q7RoborockPoint) -> None: """Clean one rectangular zone given two opposite corners in common Roborock coordinates.""" @@ -237,6 +365,7 @@ async def clean_zones(self, zones: list[tuple[Q7RoborockPoint, Q7RoborockPoint]] 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}) + self.cancel_goto() async def set_carpet_turbo(self, enabled: bool) -> None: """Enable or disable increased suction on carpets.""" @@ -256,6 +385,7 @@ async def pause_clean(self) -> None: "room_ids": [], }, ) + self.cancel_goto() async def stop_clean(self) -> None: """Stop cleaning.""" @@ -267,6 +397,7 @@ async def stop_clean(self) -> None: "room_ids": [], }, ) + self.cancel_goto() async def return_to_dock(self) -> None: """Return to dock.""" @@ -274,6 +405,7 @@ async def return_to_dock(self) -> None: command=RoborockB01Q7Methods.START_RECHARGE, params={}, ) + self.cancel_goto() async def find_me(self) -> None: """Locate the robot.""" diff --git a/roborock/devices/traits/b01/q7/goto.py b/roborock/devices/traits/b01/q7/goto.py new file mode 100644 index 000000000..94f818c0c --- /dev/null +++ b/roborock/devices/traits/b01/q7/goto.py @@ -0,0 +1,122 @@ +"""State management for an emulated Q7 goto action. + +The Q7 firmware has no "go to point" command, so a goto is emulated (like on the +Q10) with a small zone-clean task around the target that is stopped once the +robot gets close enough. +""" + +import logging +from collections.abc import Callable +from dataclasses import dataclass +from enum import StrEnum +from math import hypot + +from roborock.callbacks import CallbackList +from roborock.data.b01_q7.b01_q7_code_mappings import WorkModeMapping +from roborock.data.b01_q7.b01_q7_containers import Q7RoborockPoint + +_LOGGER = logging.getLogger(__name__) + +# ``work_mode`` reported while a zone-clean task is running (verified on a Q7 sc05). +_ZONE_WORK_MODE = WorkModeMapping.PLAN_AREA +# Reported once the zone is done and the robot heads home: it already reached the target. +_ZONE_DONE_WORK_MODE = WorkModeMapping.PLAN_AREA_GO_HOME + + +class GotoActionCommand(StrEnum): + """A command requested by a Q7 goto action.""" + + STOP = "stop" + COMPLETE = "complete" + + +@dataclass(frozen=True) +class GotoSnapshot: + """Device state needed to advance a goto action.""" + + position: Q7RoborockPoint | None + work_mode: WorkModeMapping | None + + +class GotoAction: + """Decide how one emulated goto should react to device updates. + + The action owns no tasks and sends no device commands. ``Q7PropertiesApi`` + feeds it push-derived snapshots and performs the commands it requests. + """ + + def __init__(self, target: Q7RoborockPoint, *, tolerance: int) -> None: + """Initialize a goto action waiting for its zone task to start.""" + self._target = target + self._tolerance = tolerance + self._task_seen = False + self._command_pending = False + self._finished = False + self._latest_snapshot: GotoSnapshot | None = None + self._callbacks: CallbackList[GotoActionCommand] = CallbackList(logger=_LOGGER) + + def add_update_listener(self, callback: Callable[[GotoActionCommand], None]) -> Callable[[], None]: + """Register a callback for the commands requested by the action.""" + return self._callbacks.add_callback(callback) + + def update(self, snapshot: GotoSnapshot) -> None: + """Process the latest push-derived device state.""" + self._latest_snapshot = snapshot + self._evaluate(snapshot) + + def retry(self) -> None: + """Re-evaluate the latest state after a requested command failed.""" + if self._finished or self._latest_snapshot is None: + return + self._command_pending = False + self._evaluate(self._latest_snapshot) + + def timeout(self, snapshot: GotoSnapshot) -> None: + """Stop the zone task if it is still running when the safety timeout expires.""" + if self._finished or self._command_pending: + return + self._latest_snapshot = snapshot + if self._task_seen and snapshot.work_mode is _ZONE_WORK_MODE: + self._emit(GotoActionCommand.STOP) + else: + self._emit(GotoActionCommand.COMPLETE) + + def complete(self) -> None: + """Mark the action complete after its requested command succeeds.""" + self._finished = True + self._command_pending = False + + def is_near_target(self, position: Q7RoborockPoint | None) -> bool: + """Return whether ``position`` is within the arrival tolerance of the target.""" + return ( + position is not None and hypot(position.x - self._target.x, position.y - self._target.y) <= self._tolerance + ) + + def _evaluate(self, snapshot: GotoSnapshot) -> None: + """Derive the next command from the latest device state.""" + if self._finished or self._command_pending: + return + + if snapshot.work_mode is _ZONE_WORK_MODE: + self._task_seen = True + elif self._task_seen and snapshot.work_mode is _ZONE_DONE_WORK_MODE: + # The small zone around the target was finished before a close-enough + # position arrived: stop the trip home so the robot stays at the target. + self._emit(GotoActionCommand.STOP) + return + elif self._task_seen: + # The zone task ended (finished, paused, docking) or was replaced. + _LOGGER.debug("Q7 goto zone task ended or was replaced (work_mode=%s)", snapshot.work_mode) + self._emit(GotoActionCommand.COMPLETE) + return + + if self._task_seen and self.is_near_target(snapshot.position): + self._emit(GotoActionCommand.STOP) + + def _emit(self, command: GotoActionCommand) -> None: + """Publish a requested command once until it is handled.""" + if command is GotoActionCommand.COMPLETE: + self._finished = True + else: + self._command_pending = True + self._callbacks(command) diff --git a/tests/devices/traits/b01/q7/test_goto.py b/tests/devices/traits/b01/q7/test_goto.py new file mode 100644 index 000000000..f34a1c562 --- /dev/null +++ b/tests/devices/traits/b01/q7/test_goto.py @@ -0,0 +1,166 @@ +"""Tests for the Q7 emulated goto (state machine and API integration).""" + +import asyncio + +import pytest + +from roborock.data.b01_q7 import Q7RoborockPoint, SCDeviceCleanParam, WorkModeMapping +from roborock.devices.traits.b01.q7 import Q7PropertiesApi +from roborock.devices.traits.b01.q7.goto import GotoAction, GotoActionCommand, GotoSnapshot +from roborock.protocols.b01_q7_protocol import Q7DevicePush +from roborock.roborock_typing import RoborockB01Q7Methods + +from .conftest import FakeQ7Channel + +TARGET = Q7RoborockPoint(x=26500, y=22200) +NEAR = Q7RoborockPoint(x=26590, y=22220) +FAR = Q7RoborockPoint(x=29060, y=22090) + + +def _snapshot( + position: Q7RoborockPoint | None = FAR, work_mode: WorkModeMapping | None = WorkModeMapping.PLAN_AREA +) -> GotoSnapshot: + return GotoSnapshot(position=position, work_mode=work_mode) + + +def _action() -> tuple[GotoAction, list[GotoActionCommand]]: + action = GotoAction(TARGET, tolerance=300) + commands: list[GotoActionCommand] = [] + action.add_update_listener(commands.append) + return action, commands + + +def test_goto_action_stops_near_target() -> None: + """Once the zone task runs, getting within tolerance requests a stop.""" + action, commands = _action() + + action.update(_snapshot()) + action.update(_snapshot(position=NEAR)) + + assert commands == [GotoActionCommand.STOP] + + +def test_goto_action_waits_for_zone_task() -> None: + """Being near the target before the zone task started does not stop anything.""" + action, commands = _action() + + action.update(_snapshot(position=NEAR, work_mode=WorkModeMapping.IDLE)) + + assert commands == [] + + +def test_goto_action_stops_trip_home_after_zone_done() -> None: + """If the zone finished before a close position arrived, the trip home is stopped.""" + action, commands = _action() + + action.update(_snapshot()) + action.update(_snapshot(work_mode=WorkModeMapping.PLAN_AREA_GO_HOME)) + + assert commands == [GotoActionCommand.STOP] + + +@pytest.mark.parametrize("work_mode", [WorkModeMapping.IDLE, WorkModeMapping.MOPPING, None]) +def test_goto_action_completes_when_task_replaced(work_mode: WorkModeMapping | None) -> None: + """Leaving the zone task any other way completes the action without commands.""" + action, commands = _action() + + action.update(_snapshot()) + action.update(_snapshot(work_mode=work_mode)) + + assert commands == [GotoActionCommand.COMPLETE] + + +@pytest.mark.parametrize( + ("work_mode", "expected"), + [(WorkModeMapping.PLAN_AREA, GotoActionCommand.STOP), (WorkModeMapping.IDLE, GotoActionCommand.COMPLETE)], +) +def test_goto_action_timeout(work_mode: WorkModeMapping, expected: GotoActionCommand) -> None: + """The safety timeout stops a still-running zone task, otherwise just completes.""" + action, commands = _action() + action.update(_snapshot()) + + action.timeout(_snapshot(work_mode=work_mode)) + + assert commands == [expected] + + +def test_goto_action_retry_after_failed_stop() -> None: + """A failed stop can be retried from the latest state; one command is pending at a time.""" + action, commands = _action() + action.update(_snapshot()) + action.update(_snapshot(position=NEAR)) + action.update(_snapshot(position=NEAR)) # pending: no duplicate + + action.retry() + + assert commands == [GotoActionCommand.STOP, GotoActionCommand.STOP] + + +def _push(**params: object) -> Q7DevicePush: + return Q7DevicePush(method="prop.post", params=dict(params)) + + +def _cur_path(position: Q7RoborockPoint) -> Q7DevicePush: + x, y = position.to_world() + return _push(cur_path=[1, x, y, 0.0, 0, 1794490763]) + + +async def _settle() -> None: + for _ in range(5): + await asyncio.sleep(0) + + +async def test_q7_goto_position_stops_zone_at_target(q7_api: Q7PropertiesApi, fake_channel: FakeQ7Channel): + """goto_position starts a small zone and stops it when the live position arrives.""" + await q7_api.start() + assert fake_channel.device_push_callback is not None + fake_channel.device_push_callback(_cur_path(FAR)) + + await q7_api.goto_position(TARGET) + await _settle() + + zone_points = [round(value, 3) for value in fake_channel.published_commands[1][1]["zone_points"]] + assert fake_channel.published_commands[:3] == [ + (RoborockB01Q7Methods.SET_PREFERENCE_TYPE, {"prefer_type": 4}), + (RoborockB01Q7Methods.SET_ZONE_POINTS, fake_channel.published_commands[1][1]), + (RoborockB01Q7Methods.SET_ZONE_CLEAN, {"ctrl_value": SCDeviceCleanParam.START.code}), + ] + assert zone_points == [0.75, -3.55, 1.25, -3.55, 1.25, -3.05, 0.75, -3.05] + # The live map is requested to track the position. + assert (RoborockB01Q7Methods.UPLOAD_BY_MAPTYPE, {"map_type": 0}) in fake_channel.published_commands + + fake_channel.device_push_callback(_push(work_mode=WorkModeMapping.PLAN_AREA.code, status=5)) + fake_channel.device_push_callback(_cur_path(NEAR)) + await _settle() + + stop = (RoborockB01Q7Methods.SET_ZONE_CLEAN, {"ctrl_value": SCDeviceCleanParam.STOP.code}) + assert fake_channel.published_commands.count(stop) == 1 + await q7_api.close() + + +async def test_q7_goto_position_already_there(q7_api: Q7PropertiesApi, fake_channel: FakeQ7Channel): + """No task is started when the robot is already within tolerance of the target.""" + await q7_api.start() + assert fake_channel.device_push_callback is not None + fake_channel.device_push_callback(_cur_path(NEAR)) + + await q7_api.goto_position(TARGET) + + assert fake_channel.published_commands == [] + await q7_api.close() + + +async def test_q7_goto_cancelled_by_other_command(q7_api: Q7PropertiesApi, fake_channel: FakeQ7Channel): + """Another cleaning command replaces the goto, which then never sends its stop.""" + await q7_api.start() + assert fake_channel.device_push_callback is not None + + await q7_api.goto_position(TARGET) + await q7_api.clean_segments([11]) + fake_channel.device_push_callback(_push(work_mode=WorkModeMapping.PLAN_AREA.code)) + fake_channel.device_push_callback(_cur_path(NEAR)) + await _settle() + + stop = (RoborockB01Q7Methods.SET_ZONE_CLEAN, {"ctrl_value": SCDeviceCleanParam.STOP.code}) + assert stop not in fake_channel.published_commands + await q7_api.close()