refactor: reconcile Q10 map dispatch onto #847's typed-message model

PR #847 (the PR this branch is stacked on) landed a typed-message decode/
dispatch model on main: stream_decoded_messages() decodes each MAP_RESPONSE
push into a typed Q10Message (Q10DpsUpdate | Q10MapPacket | Q10TracePacket)
via decode_message(), and Q10PropertiesApi._handle_message routes by
isinstance. This branch predated that and carried the older push-driven model
(channel yielding DPS-only dicts, _handle_message taking the raw RoborockMessage
and calling map.update_from_map_response()), so the merges reconciled the
parsing details but left the dispatch architecture reverted -- decode_message
ended up orphaned (defined + unit-tested but with no production caller), which
is the revert @allenporter flagged.

Re-reconcile onto main's typed model instead of replacing it:

- b01_q10_channel.py: restore stream_decoded_messages() + decode_message()
  (now byte-identical to main); drop the DPS-only stream_decoded_responses().
- traits/b01/q10/__init__.py: restore main's typed isinstance dispatch. The
  branch's one net-new behavior -- feeding the no-go / virtual-wall overlay DPs
  to the map trait -- now rides on the Q10DpsUpdate branch.
- traits/b01/q10/map.py: replace update_from_map_response(message) with main's
  typed update_from_map_packet(packet) / update_from_trace_packet(packet). The
  trait caches the parsed Q10MapPacket (self._packet) so erase zones / overlays
  re-render without re-parsing wire bytes; parse_map_content() re-renders the
  cached packet. Drop raw_api_response (no raw bytes reach the trait in the
  typed model, matching main). Calibration / layers / erase / overlay features
  are unchanged, just layered on the typed entry points.
- tests: drive the typed methods; the trait-level "ignores non-map" case is
  now covered by the decode_message protocol tests and the dispatch integration
  tests.

decode_message is back on the production path. Full suite green (611 passed);
mypy + ruff clean.

roborock/devices/rpc/b01_q10_channel.py

@@ -2,41 +2,42 @@
 
 import logging
 from collections.abc import AsyncGenerator
-from typing import Any
 
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.devices.transport.mqtt_channel import MqttChannel
 from roborock.exceptions import RoborockException
 from roborock.protocols.b01_q10_protocol import (
     ParamsType,
-    decode_rpc_response,
+    Q10Message,
+    decode_message,
     encode_mqtt_payload,
 )
 
 _LOGGER = logging.getLogger(__name__)
 
 
-async def stream_decoded_responses(
+async def stream_decoded_messages(
     mqtt_channel: MqttChannel,
-) -> AsyncGenerator[dict[B01_Q10_DP, Any], None]:
-    """Stream decoded DPS messages received via MQTT.
+) -> AsyncGenerator[Q10Message, None]:
+    """Stream decoded Q10 messages received via MQTT.
 
-    Messages that are not decodable DPS responses (e.g. protocol-301
-    ``MAP_RESPONSE`` map pushes) are skipped; callers that need the raw
-    messages should subscribe to :meth:`MqttChannel.subscribe_stream` directly.
+    Each pushed ``RoborockMessage`` is decoded into a typed :data:`Q10Message`
+    (a DPS status update, a map packet, or a trace packet). Messages that fail
+    to decode or carry an unrecognized payload are skipped.
     """
 
-    async for response_message in mqtt_channel.subscribe_stream():
+    async for message in mqtt_channel.subscribe_stream():
         try:
-            decoded_dps = decode_rpc_response(response_message)
+            decoded = decode_message(message)
         except RoborockException as ex:
             _LOGGER.debug(
-                "Failed to decode B01 Q10 RPC response: %s: %s",
-                response_message,
+                "Failed to decode B01 Q10 message: %s: %s",
+                message,
                 ex,
             )
             continue
-        yield decoded_dps
+        if decoded is not None:
+            yield decoded
 
 
 async def send_command(

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

@@ -4,11 +4,11 @@
 import logging
 
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
+from roborock.devices.rpc.b01_q10_channel import stream_decoded_messages
 from roborock.devices.traits import Trait
 from roborock.devices.transport.mqtt_channel import MqttChannel
-from roborock.exceptions import RoborockException
-from roborock.protocols.b01_q10_protocol import decode_rpc_response
-from roborock.roborock_message import RoborockMessage
+from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10TracePacket
+from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message
 
 from .button_light import ButtonLightTrait
 from .child_lock import ChildLockTrait
@@ -126,41 +126,36 @@ async def refresh(self) -> None:
         await self.command.send(B01_Q10_DP.REQUEST_DPS, params={})
 
     async def _subscribe_loop(self) -> None:
-        """Persistent loop dispatching pushed messages to the read-model traits."""
-        async for message in self._channel.subscribe_stream():
+        """Persistent loop dispatching decoded messages to the read-model traits."""
+        async for message in stream_decoded_messages(self._channel):
             self._handle_message(message)
 
-    def _handle_message(self, message: RoborockMessage) -> None:
-        """Route a single pushed message to the trait responsible for it.
+    def _handle_message(self, message: Q10Message) -> None:
+        """Route a single decoded message to the trait responsible for it.
 
-        Map/trace pushes arrive as protocol-301 ``MAP_RESPONSE`` messages (not
-        DPS), so they are handled separately from the status DPS stream. The Q10
-        is entirely push-driven: there is no synchronous get-map request, the
-        device just publishes its current map (a ``dpRequestDps`` nudges it to).
+        Map and trace packets arrive as protocol-301 ``MAP_RESPONSE`` pushes (the
+        Q10 is entirely push-driven: there is no synchronous get-map request, a
+        ``dpRequestDps`` just nudges the device to publish its current map). DPS
+        updates feed the read-model traits. More traits can be dispatched here below.
         """
-        if self.map.update_from_map_response(message):
-            return
-
-        try:
-            decoded_dps = decode_rpc_response(message)
-        except RoborockException as ex:
-            _LOGGER.debug("Failed to decode Q10 RPC response: %s: %s", message, ex)
-            return
-
-        _LOGGER.debug("Received Q10 status update: %s", decoded_dps)
-        # Notify all read-model traits about the new message; each trait
-        # only updates the fields that it is responsible for.
-        # More traits can be added here below.
-        for trait in self._updatable_traits:
-            trait.update_from_dps(decoded_dps)
-
-        # Feed the map's vector-overlay data points (no-go zones / virtual
-        # walls) to the map trait so they are decoded as they arrive.
-        if B01_Q10_DP.RESTRICTED_ZONE_UP in decoded_dps or B01_Q10_DP.VIRTUAL_WALL_UP in decoded_dps:
-            self.map.load_overlays(
-                restricted_zone_up=decoded_dps.get(B01_Q10_DP.RESTRICTED_ZONE_UP),
-                virtual_wall_up=decoded_dps.get(B01_Q10_DP.VIRTUAL_WALL_UP),
-            )
+        if isinstance(message, Q10MapPacket):
+            self.map.update_from_map_packet(message)
+        elif isinstance(message, Q10TracePacket):
+            self.map.update_from_trace_packet(message)
+        elif isinstance(message, Q10DpsUpdate):
+            _LOGGER.debug("Received Q10 status update: %s", message.dps)
+            # Notify all read-model traits about the new message; each trait
+            # only updates the fields that it is responsible for.
+            for trait in self._updatable_traits:
+                trait.update_from_dps(message.dps)
+
+            # Feed the map's vector-overlay data points (no-go zones / virtual
+            # walls) to the map trait so they are decoded as they arrive.
+            if B01_Q10_DP.RESTRICTED_ZONE_UP in message.dps or B01_Q10_DP.VIRTUAL_WALL_UP in message.dps:
+                self.map.load_overlays(
+                    restricted_zone_up=message.dps.get(B01_Q10_DP.RESTRICTED_ZONE_UP),
+                    virtual_wall_up=message.dps.get(B01_Q10_DP.VIRTUAL_WALL_UP),
+                )
 
 
 def create(channel: MqttChannel) -> Q10PropertiesApi:

roborock/devices/traits/b01/q10/map.py

@@ -4,13 +4,18 @@
 trait is purely push-driven and mirrors the Q10 ``StatusTrait`` contract:
 
 - The device pushes its current map/path as protocol-301 ``MAP_RESPONSE``
-  messages (a ``dpRequestDps`` nudges it to do so). The ``Q10PropertiesApi``
-  subscribe loop routes those messages to :meth:`MapContentTrait.update_from_map_response`.
-- ``update_from_map_response`` parses the payload, updates the cached fields and
-  notifies update listeners (register via :meth:`add_update_listener`).
-- ``parse_map_content()`` reparses the cached raw bytes without I/O.
-- ``image_content``, ``map_data``, ``rooms``, ``path``, ``robot_position`` and
-  ``raw_api_response`` are readable and reflect the most recently pushed map.
+  messages (a ``dpRequestDps`` nudges it to do so). The protocol layer decodes
+  each push into a typed :class:`~roborock.map.b01_q10_map_parser.Q10MapPacket`
+  or :class:`~roborock.map.b01_q10_map_parser.Q10TracePacket`, and the
+  ``Q10PropertiesApi`` subscribe loop routes those to
+  :meth:`MapContentTrait.update_from_map_packet` /
+  :meth:`MapContentTrait.update_from_trace_packet`.
+- Those methods render/cache the packet, update the cached fields and notify
+  update listeners (register via :meth:`add_update_listener`).
+- ``parse_map_content()`` re-renders the cached map packet without I/O (e.g.
+  after the calibration changed).
+- ``image_content``, ``map_data``, ``rooms``, ``path`` and ``robot_position``
+  are readable and reflect the most recently pushed map.
 
 Unlike the Q7, the Q10 map payload is unencrypted, so no map key is required.
 """
@@ -37,12 +42,12 @@
     B01Q10MapParserConfig,
     Q10EraseZone,
     Q10HeaderCalibration,
+    Q10MapPacket,
     Q10Point,
     Q10Room,
+    Q10TracePacket,
     decompose_layers,
     erased_packet,
-    parse_map_packet,
-    parse_trace_packet,
 )
 from roborock.map.b01_q10_overlays import (
     ZONE_TYPE_NO_GO,
@@ -51,17 +56,11 @@
     parse_virtual_wall_blob,
     parse_zone_blob,
 )
-from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol
 
 _LOGGER = logging.getLogger(__name__)
 
 _TRUNCATE_LENGTH = 20
 
-# MAP_RESPONSE (protocol 301) payloads start with a 2-byte marker identifying the
-# packet kind: a full map (``01 01``) or a live trace/path (``02 01``).
-_MAP_PACKET_MARKER = b"\x01\x01"
-_TRACE_PACKET_MARKER = b"\x02\x01"
-
 # Path-units-per-pixel candidates for calibration. A dense ss07 path lands a
 # best fit of 20.0 around the header origin -- ground-truthed June 2026 on the
 # R1: a corridor drive registered at 20 (matching the format author's
@@ -129,9 +128,6 @@ class MapContent(RoborockBase):
     map packet tail. Once a calibration is available the cells inside them are
     blanked from the rendered map and every layer (see :meth:`MapContentTrait`)."""
 
-    raw_api_response: bytes | None = None
-    """Raw bytes of the map payload from the device (opaque blob for re-parsing)."""
-
     header_calibration: Q10HeaderCalibration | None = None
     """Calibration read straight from the map packet's grid-frame header (ss07).
     Supplies the world<->pixel origin without a fit, so :meth:`solve_calibration`
@@ -149,8 +145,9 @@ class MapContentTrait(MapContent, TraitUpdateListener):
     """Trait holding the most recently pushed parsed map content for Q10 devices.
 
     The Q10 has no synchronous get-map request; the device pushes map and trace
-    packets, which the ``Q10PropertiesApi`` subscribe loop feeds into
-    :meth:`update_from_map_response`. Consumers read the cached fields and/or
+    packets, which the protocol layer decodes into typed packets and the
+    ``Q10PropertiesApi`` subscribe loop feeds into :meth:`update_from_map_packet`
+    / :meth:`update_from_trace_packet`. Consumers read the cached fields and/or
     register a callback with :meth:`add_update_listener` to be notified when new
     map content arrives.
     """
@@ -163,48 +160,44 @@ def __init__(
         super().__init__()
         TraitUpdateListener.__init__(self, logger=_LOGGER)
         self._map_parser = B01Q10MapParser(map_parser_config)
+        # The most recently pushed (parsed) map packet, cached so the map can be
+        # re-rendered (e.g. to apply erase zones / overlays once a calibration is
+        # known) without re-parsing wire bytes.
+        self._packet: Q10MapPacket | None = None
 
-    def update_from_map_response(self, message: RoborockMessage) -> bool:
-        """Update cached map/trace state from a pushed ``MAP_RESPONSE`` message.
+    def update_from_map_packet(self, packet: Q10MapPacket) -> None:
+        """Render a pushed full-map packet into the cached image/rooms/layers.
 
-        Returns ``True`` if the message was a recognized Q10 map (``01 01``) or
-        trace (``02 01``) packet (so the caller can stop processing it), and
-        ``False`` otherwise. Update listeners are notified only when a packet is
-        parsed successfully.
+        Rendering failures are logged and skipped (listeners are not notified) so
+        a single bad push cannot tear down the subscribe loop.
         """
-        if message.protocol != RoborockMessageProtocol.MAP_RESPONSE or not message.payload:
-            return False
-        marker = message.payload[:2]
-        if marker == _MAP_PACKET_MARKER:
-            self.raw_api_response = message.payload
-            try:
-                self.parse_map_content()
-            except RoborockException as ex:
-                _LOGGER.debug("Failed to parse Q10 map packet: %s", ex)
-                return True
-            self._notify_update()
-            return True
-        if marker == _TRACE_PACKET_MARKER:
-            try:
-                trace = parse_trace_packet(message.payload)
-            except RoborockException as ex:
-                _LOGGER.debug("Failed to parse Q10 trace packet: %s", ex)
-                return True
-            self.path = trace.points
-            self.robot_position = trace.robot_position
-            self.robot_heading = trace.heading
-            self._notify_update()
-            return True
-        return False
+        try:
+            self._render_packet(packet)
+        except RoborockException as ex:
+            _LOGGER.debug("Failed to render Q10 map packet: %s", ex)
+            return
+        self._notify_update()
+
+    def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
+        """Cache the path / robot position / heading from a pushed trace packet."""
+        self.path = packet.points
+        self.robot_position = packet.robot_position
+        self.robot_heading = packet.heading
+        self._notify_update()
 
     def parse_map_content(self) -> None:
-        """Reparse the cached raw map payload without performing any I/O."""
-        if self.raw_api_response is None:
+        """Re-render the cached map packet without performing any I/O.
+
+        Used to refresh the rendered image / ``map_data`` after the calibration
+        changed (so the path, overlays and erase zones are reapplied)."""
+        if self._packet is None:
             raise RoborockException("No map payload available; no map has been pushed yet")
+        self._render_packet(self._packet)
 
+    def _render_packet(self, packet: Q10MapPacket) -> None:
+        """Render a parsed map packet into the cached image / map_data / layers."""
         try:
-            parsed = self._map_parser.parse(self.raw_api_response)
-            packet = parse_map_packet(self.raw_api_response)
+            parsed = self._map_parser.parsed_from_packet(packet)
         except RoborockException:
             raise
         except Exception as ex:
@@ -213,6 +206,7 @@ def parse_map_content(self) -> None:
         if parsed.image_content is None:
             raise RoborockException("Failed to render Q10 map image")
 
+        self._packet = packet
         self.image_content = parsed.image_content
         self.map_data = parsed.map_data
         self.rooms = packet.rooms
@@ -342,12 +336,12 @@ def _apply_erase(self, calibration: GridCalibration) -> None:
         them in pixel space, blank those cells to background, and re-render so the
         phantom areas disappear -- matching what the app shows.
         """
-        if self.layers is None or self.raw_api_response is None:
+        if self.layers is None or self._packet is None:
             return
         cells = self._erased_cells(calibration)
         if not cells:
             return
-        packet = erased_packet(parse_map_packet(self.raw_api_response), cells)
+        packet = erased_packet(self._packet, cells)
         parsed = self._map_parser.parsed_from_packet(packet)
         self.image_content = parsed.image_content
         self.map_data = parsed.map_data