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

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()
  elif self.zeo is not None:
  await self.zeo.start()
  except RoborockException:
@@ -232,6 +234,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.zeo is not None:
  self.zeo.close()
  if self._unsub:

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,19 @@ 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."""
+ 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)
+
+ 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,20 +9,24 @@
 """
 
 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.map.b01_map_parser import B01MapParser, B01MapParserConfig, parse_map_type
 from roborock.roborock_typing import RoborockB01Q7Methods
 
 from .map import MapTrait
 
+_LOGGER = logging.getLogger(__name__)
 _TRUNCATE_LENGTH = 20
+_LIVE_MAP_TYPE = 0
 
 
 @dataclass
@@ -49,7 +53,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 +63,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 +87,34 @@ 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. Frames for other maps
+ than the live one are ignored to not overwrite the current map.
+ """
+ try:
+ map_type = parse_map_type(raw_payload)
+ except RoborockException as ex:
+ _LOGGER.debug("Failed to parse pushed B01 map frame: %s", ex)
+ return
+
+ if map_type != _LIVE_MAP_TYPE:
+ _LOGGER.debug("Ignoring pushed B01 map frame of type %s", map_type)
+ return
+
+ 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:

roborock/map/b01_map_parser.py

@@ -65,6 +65,15 @@ def parse(self, payload: bytes) -> ParsedMapData:
  )
 
 
+def parse_map_type(payload: bytes) -> int:
+ """Return the SCMap frame's map type without rendering the image.
+
+ Type 0 is the live map of the current cleaning run, other types are
+ historic or saved maps.
+ """
+ return _parse_scmap_payload(payload).mapType
+
+
 def _parse_scmap_payload(payload: bytes) -> RobotMap:
  """Parse inflated SCMap bytes into a generated protobuf message."""
  parsed = RobotMap()

tests/devices/rpc/test_b01_q7_channel.py

@@ -293,3 +293,60 @@ async def test_send_command_general_exception(
 
  with pytest.raises(RuntimeError, match="Generic publish crash"):
  await channel.send_command("prop.get", {"property": ["status"]})
+
+
+async def test_subscribe_map_pushes_delivers_decoded_frames(
+ device: HomeDataDevice,
+ product: HomeDataProduct,
+ fake_channel: FakeChannel,
+ message_builder: B01MessageBuilder,
+) -> None:
+ """Unsolicited MAP_RESPONSE frames are decoded and delivered to the callback."""
+ channel = create_b01_q7_channel(device, product, fake_channel) # type: ignore[arg-type]
+
+ received: list[bytes] = []
+ unsub = await channel.subscribe_map_pushes(received.append)
+
+ with patch(
+ "roborock.devices.rpc.b01_q7_channel.decode_map_payload",
+ return_value=b"inflated-payload",
+ ):
+ fake_channel.notify_subscribers(message_builder.build_map_response(b"raw-map-payload"))
+
+ assert received == [b"inflated-payload"]
+
+ # Non-map messages are filtered out.
+ fake_channel.notify_subscribers(message_builder.build({"status": 1}))
+ assert received == [b"inflated-payload"]
+
+ # After unsubscribing, further frames are not delivered.
+ unsub()
+ with patch(
+ "roborock.devices.rpc.b01_q7_channel.decode_map_payload",
+ return_value=b"inflated-payload",
+ ):
+ fake_channel.notify_subscribers(message_builder.build_map_response(b"raw-map-payload"))
+ assert received == [b"inflated-payload"]
+
+
+async def test_subscribe_map_pushes_skips_undecodable_frames(
+ device: HomeDataDevice,
+ product: HomeDataProduct,
+ fake_channel: FakeChannel,
+ message_builder: B01MessageBuilder,
+) -> None:
+ """Frames that fail map decoding are skipped without breaking the subscription."""
+ channel = create_b01_q7_channel(device, product, fake_channel) # type: ignore[arg-type]
+
+ received: list[bytes] = []
+ await channel.subscribe_map_pushes(received.append)
+
+ fake_channel.notify_subscribers(message_builder.build_map_response(b"!!! not base64 !!!"))
+ assert received == []
+
+ with patch(
+ "roborock.devices.rpc.b01_q7_channel.decode_map_payload",
+ return_value=b"inflated-payload",
+ ):
+ fake_channel.notify_subscribers(message_builder.build_map_response(b"raw-map-payload"))
+ assert received == [b"inflated-payload"]

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,8 @@ 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
+ self.map_push_subscribe_count = 0
 
  async def send_command(self, command: Any, params: Any = None) -> Any:
  if self.side_effect:
@@ -29,6 +32,15 @@ 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_subscribe_count += 1
+ 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: