feat: live Q7 (B01) map updates from unsolicited map pushes

Q7 devices stream full SCMap frames (protocol 301) on their own during
cleaning — no polling or request is needed. Verified against a live
Q7 Series (roborock.vacuum.sc05): a short clean produced a pushed frame
roughly every 10 seconds.

- B01Q7Channel.subscribe_map_pushes() decodes unsolicited MAP_RESPONSE
  frames with the device map key.
- MapContentTrait.update_from_push() re-parses pushed frames and notifies
  update listeners; malformed frames are dropped without clearing the
  cached map.
- Q7PropertiesApi.start()/close() subscribe for the device lifetime, wired
  up in RoborockDevice.connect()/close() like V1 and Q10.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: andig

roborock/devices/device.py

@@ -202,6 +202,8 @@ async def connect(self) -> None:
                 await self.v1_properties.start()
             elif self.b01_q10_properties is not None:
                 await self.b01_q10_properties.start()
+            elif self.b01_q7_properties is not None:
+                await self.b01_q7_properties.start()
         except RoborockException:
             # Expected: start() can fail transiently. Unsubscribe before propagating
             # so the retry by connect_loop() gets a clean channel.
@@ -230,6 +232,8 @@ async def close(self) -> None:
             self.v1_properties.close()
         if self.b01_q10_properties is not None:
             await self.b01_q10_properties.close()
+        if self.b01_q7_properties is not None:
+            await self.b01_q7_properties.close()
         if self._unsub:
             self._unsub()
             self._unsub = None

roborock/devices/rpc/b01_q7_channel.py

@@ -53,6 +53,10 @@ async def send_map_command(
         """Send a map command and get decoded bytes."""
         ...
 
+    async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Callable[[], None]:
+        """Subscribe to unsolicited map pushes, invoking callback with decoded SCMap bytes."""
+        ...
+
 
 def _matches_map_response(response_message: RoborockMessage, *, version: bytes | None) -> bytes | None:
     """Return raw map payload bytes for matching MAP_RESPONSE messages."""
@@ -208,6 +212,25 @@ async def send_map_command(
 
         return decode_map_payload(raw_payload, map_key=self._map_key)
 
+    async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Callable[[], None]:
+        """Subscribe to unsolicited ``MAP_RESPONSE`` pushes.
+
+        The device streams full SCMap frames on its own during cleaning; the
+        callback receives the decoded (inflated) SCMap bytes for each frame.
+        """
+
+        def on_message(message: RoborockMessage) -> None:
+            if (raw_payload := _matches_map_response(message, version=B01_VERSION)) is None:
+                return
+            try:
+                decoded = decode_map_payload(raw_payload, map_key=self._map_key)
+            except RoborockException as ex:
+                _LOGGER.debug("Failed to decode pushed B01 map payload: %s", ex)
+                return
+            callback(decoded)
+
+        return await self._mqtt_channel.subscribe(on_message)
+
 
 def create_b01_q7_channel(
     device: HomeDataDevice,

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

@@ -3,6 +3,7 @@
 Potentially other devices may fall into this category in the future.
 """
 
+from collections.abc import Callable
 from typing import Any
 
 from roborock import B01Props
@@ -71,6 +72,17 @@ def __init__(
             self._map_rpc_channel,
             self.map,
         )
+        self._unsub_map_pushes: Callable[[], None] | None = None
+
+    async def start(self) -> None:
+        """Start listening for unsolicited map pushes from the device."""
+        self._unsub_map_pushes = await self._map_rpc_channel.subscribe_map_pushes(self.map_content.update_from_push)
+
+    async def close(self) -> None:
+        """Stop listening for unsolicited map pushes."""
+        if self._unsub_map_pushes is not None:
+            self._unsub_map_pushes()
+            self._unsub_map_pushes = None
 
     async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None:
         """Query the device for the values of the given Q7 properties."""

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

@@ -9,19 +9,22 @@
 """
 
 import asyncio
+import logging
 from dataclasses import dataclass
 
 from vacuum_map_parser_base.map_data import MapData
 
 from roborock.data import 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
 from roborock.roborock_typing import RoborockB01Q7Methods
 
 from .map import MapTrait
 
+_LOGGER = logging.getLogger(__name__)
 _TRUNCATE_LENGTH = 20
 
 
@@ -49,7 +52,7 @@ def __repr__(self) -> str:
         return f"MapContent(image_content={img!r}, map_data={self.map_data!r})"
 
 
-class MapContentTrait(MapContent, Trait):
+class MapContentTrait(MapContent, Trait, TraitUpdateListener):
     """Trait for fetching parsed map content for Q7 devices."""
 
     def __init__(
@@ -59,7 +62,8 @@ def __init__(
         *,
         map_parser_config: B01MapParserConfig | None = None,
     ) -> None:
-        super().__init__()
+        MapContent.__init__(self)
+        TraitUpdateListener.__init__(self, logger=_LOGGER)
         self._map_rpc_channel = map_rpc_channel
         self._map_trait = map_trait
         self._map_parser = B01MapParser(map_parser_config)
@@ -82,6 +86,23 @@ async def refresh(self) -> None:
                 {"map_id": map_id},
             )
 
+        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.
+
+        Pushed frames carry the live robot pose and cleaning path, so the
+        rendered image stays current without polling.
+        """
+        try:
+            self._parse_and_store(raw_payload)
+        except RoborockException as ex:
+            _LOGGER.debug("Failed to parse pushed B01 map frame: %s", ex)
+            return
+        self._notify_update()
+
+    def _parse_and_store(self, raw_payload: bytes) -> None:
+        """Parse decoded SCMap bytes and update the cached fields."""
         try:
             parsed_data = self._map_parser.parse(raw_payload)
         except RoborockException:

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

@@ -1,3 +1,4 @@
+from collections.abc import Callable
 from typing import Any
 
 import pytest
@@ -14,6 +15,7 @@ def __init__(self) -> None:
         self.published_commands: list[tuple[Any, Any]] = []
         self.response_queue: list[Any] = []
         self.side_effect: Exception | None = None
+        self.map_push_callback: Callable[[bytes], None] | None = None
 
     async def send_command(self, command: Any, params: Any = None) -> Any:
         if self.side_effect:
@@ -29,6 +31,14 @@ async def send_map_command(self, command: Any, params: Any = None) -> bytes:
             return self.response_queue.pop(0)
         return b""
 
+    async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Callable[[], None]:
+        self.map_push_callback = callback
+
+        def unsub() -> None:
+            self.map_push_callback = None
+
+        return unsub
+
 
 @pytest.fixture(name="fake_channel")
 def fake_channel_fixture() -> FakeQ7Channel:

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

@@ -87,3 +87,63 @@ async def test_q7_map_content_refresh_errors_without_map_list(
 
     with pytest.raises(RoborockException, match="Unable to determine current map ID"):
         await q7_api.map_content.refresh()
+
+
+async def test_q7_map_content_updates_from_push(
+    q7_api: Q7PropertiesApi,
+    fake_channel: FakeQ7Channel,
+):
+    """Unsolicited map pushes update the cached map and notify listeners."""
+    await q7_api.start()
+    assert fake_channel.map_push_callback is not None
+
+    updates: list[bool] = []
+    q7_api.map_content.add_update_listener(lambda: updates.append(True))
+
+    dummy_map_data = MapData()
+    parsed_map_data = ParsedMapData(
+        image_content=b"pngbytes",
+        map_data=dummy_map_data,
+    )
+    with patch(
+        "roborock.devices.traits.b01.q7.map_content.B01MapParser.parse",
+        return_value=parsed_map_data,
+    ):
+        fake_channel.map_push_callback(b"pushed-payload")
+
+    assert q7_api.map_content.image_content == b"pngbytes"
+    assert q7_api.map_content.raw_api_response == b"pushed-payload"
+    assert updates == [True]
+
+    await q7_api.close()
+    assert fake_channel.map_push_callback is None
+
+
+async def test_q7_map_content_push_parse_failure_keeps_previous_map(
+    q7_api: Q7PropertiesApi,
+    fake_channel: FakeQ7Channel,
+):
+    """A malformed pushed frame is dropped without clearing cached content."""
+    await q7_api.start()
+    assert fake_channel.map_push_callback is not None
+
+    dummy_map_data = MapData()
+    parsed_map_data = ParsedMapData(
+        image_content=b"pngbytes",
+        map_data=dummy_map_data,
+    )
+    with patch(
+        "roborock.devices.traits.b01.q7.map_content.B01MapParser.parse",
+        return_value=parsed_map_data,
+    ):
+        fake_channel.map_push_callback(b"good-payload")
+
+    updates: list[bool] = []
+    q7_api.map_content.add_update_listener(lambda: updates.append(True))
+
+    fake_channel.map_push_callback(b"not a map")
+
+    assert q7_api.map_content.image_content == b"pngbytes"
+    assert q7_api.map_content.raw_api_response == b"good-payload"
+    assert updates == []
+    await q7_api.close()