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/4] 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/4] 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/4] 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/4] 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)