🦎 q7: fetch map payload by map_id from map list

Author: arduano

roborock/devices/traits/b01/q7/map_content.py

@@ -3,12 +3,18 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from typing import Any
 
-from roborock.devices.rpc.b01_q7_channel import send_map_command
-from roborock.devices.transport.mqtt_channel import MqttChannel
+from roborock.devices.rpc.b01_q7_channel import send_decoded_command, send_map_command
 from roborock.devices.traits import Trait
 from roborock.devices.traits.v1.map_content import MapContent
-from roborock.map.b01_map_parser import decode_b01_map_payload, parse_scmap_payload, render_map_png
+from roborock.devices.transport.mqtt_channel import MqttChannel
+from roborock.exceptions import RoborockException
+from roborock.map.b01_map_parser import (
+    decode_b01_map_payload,
+    parse_scmap_payload,
+    render_map_png,
+)
 from roborock.protocols.b01_q7_protocol import Q7RequestMessage
 from roborock.roborock_typing import RoborockB01Q7Methods
 
@@ -18,6 +24,23 @@ class B01MapContent(MapContent):
     """B01 map content wrapper."""
 
 
+def _extract_current_map_id(map_list_response: dict[str, Any] | None) -> int | None:
+    if not isinstance(map_list_response, dict):
+        return None
+    map_list = map_list_response.get("map_list")
+    if not isinstance(map_list, list) or not map_list:
+        return None
+
+    for entry in map_list:
+        if isinstance(entry, dict) and entry.get("cur") and isinstance(entry.get("id"), int):
+            return entry["id"]
+
+    first = map_list[0]
+    if isinstance(first, dict) and isinstance(first.get("id"), int):
+        return first["id"]
+    return None
+
+
 class Q7MapContentTrait(B01MapContent, Trait):
     """Fetch and parse map content from B01/Q7 devices."""
 
@@ -29,9 +52,21 @@ def __init__(self, channel: MqttChannel, *, local_key: str, serial: str, model:
         self._model = model
 
     async def refresh(self) -> B01MapContent:
+        map_list_response = await send_decoded_command(
+            self._channel,
+            Q7RequestMessage(dps=10000, command=RoborockB01Q7Methods.GET_MAP_LIST, params={}),
+        )
+        map_id = _extract_current_map_id(map_list_response)
+        if map_id is None:
+            raise RoborockException(f"Unable to determine map_id from map list response: {map_list_response!r}")
+
         raw_payload = await send_map_command(
             self._channel,
-            Q7RequestMessage(dps=10000, command=RoborockB01Q7Methods.UPLOAD_BY_MAPTYPE, params={"maptype": 301}),
+            Q7RequestMessage(
+                dps=10000,
+                command=RoborockB01Q7Methods.UPLOAD_BY_MAPID,
+                params={"map_id": map_id},
+            ),
         )
         inflated = decode_b01_map_payload(
             raw_payload,

roborock/map/b01_map_parser.py

@@ -5,7 +5,6 @@
 import base64
 import hashlib
 import io
-import math
 import zlib
 from dataclasses import dataclass
 

tests/devices/traits/b01/q7/test_init.py

@@ -16,8 +16,9 @@
 from roborock.devices.rpc.b01_q7_channel import send_decoded_command
 from roborock.devices.traits.b01.q7 import Q7PropertiesApi
 from roborock.exceptions import RoborockException
+from roborock.map.b01_map_parser import B01MapData
 from roborock.protocols.b01_q7_protocol import B01_VERSION, Q7RequestMessage
-from roborock.roborock_message import RoborockB01Props, RoborockMessageProtocol
+from roborock.roborock_message import RoborockB01Props, RoborockMessage, RoborockMessageProtocol
 from tests.fixtures.channel_fixtures import FakeChannel
 
 from . import B01MessageBuilder
@@ -257,3 +258,70 @@ async def test_q7_api_find_me(q7_api: Q7PropertiesApi, fake_channel: FakeChannel
     payload_data = json.loads(unpad(message.payload, AES.block_size))
     assert payload_data["dps"]["10000"]["method"] == "service.find_device"
     assert payload_data["dps"]["10000"]["params"] == {}
+
+
+async def test_q7_api_clean_segments(
+    q7_api: Q7PropertiesApi, fake_channel: FakeChannel, message_builder: B01MessageBuilder
+):
+    """Test room/segment cleaning helper for Q7."""
+    fake_channel.response_queue.append(message_builder.build({"result": "ok"}))
+    await q7_api.clean_segments([10, 11])
+
+    assert len(fake_channel.published_messages) == 1
+    message = fake_channel.published_messages[0]
+    payload_data = json.loads(unpad(message.payload, AES.block_size))
+    assert payload_data["dps"]["10000"]["method"] == "service.set_room_clean"
+    assert payload_data["dps"]["10000"]["params"] == {
+        "clean_type": CleanTaskTypeMapping.ROOM.code,
+        "ctrl_value": SCDeviceCleanParam.START.code,
+        "room_ids": [10, 11],
+    }
+
+
+async def test_q7_map_content_refresh_from_map_response(
+    q7_api: Q7PropertiesApi,
+    fake_channel: FakeChannel,
+    message_builder: B01MessageBuilder,
+    monkeypatch: pytest.MonkeyPatch,
+):
+    """Test Q7 map content refresh wiring through map list + MAP_RESPONSE payload path."""
+
+    fake_channel.response_queue.append(message_builder.build({"map_list": [{"id": 1772093512, "cur": True}]}))
+    fake_channel.response_queue.append(
+        RoborockMessage(
+            protocol=RoborockMessageProtocol.MAP_RESPONSE,
+            payload=b"raw-map-payload",
+            version=b"B01",
+            seq=message_builder.seq + 1,
+        )
+    )
+
+    monkeypatch.setattr(
+        "roborock.devices.traits.b01.q7.map_content.decode_b01_map_payload",
+        lambda raw_payload, **kwargs: b"inflated-payload",
+    )
+    monkeypatch.setattr(
+        "roborock.devices.traits.b01.q7.map_content.parse_scmap_payload",
+        lambda payload: B01MapData(size_x=1, size_y=1, map_data=b"\x01"),
+    )
+    monkeypatch.setattr(
+        "roborock.devices.traits.b01.q7.map_content.render_map_png",
+        lambda parsed: b"\x89PNG-test",
+    )
+
+    result = await q7_api.map_content.refresh()
+
+    assert result.image_content == b"\x89PNG-test"
+    assert result.raw_api_response == b"raw-map-payload"
+
+    assert len(fake_channel.published_messages) == 2
+
+    first = fake_channel.published_messages[0]
+    first_payload = json.loads(unpad(first.payload, AES.block_size))
+    assert first_payload["dps"]["10000"]["method"] == "service.get_map_list"
+    assert first_payload["dps"]["10000"]["params"] == {}
+
+    second = fake_channel.published_messages[1]
+    second_payload = json.loads(unpad(second.payload, AES.block_size))
+    assert second_payload["dps"]["10000"]["method"] == "service.upload_by_mapid"
+    assert second_payload["dps"]["10000"]["params"] == {"map_id": 1772093512}