refactor: route Q10 overlay DPs through the map trait's update_from_dps

The orchestrator special-cased the no-go / virtual-wall data points,
reaching into the map trait's load_overlays after the generic DPS fan-out.
Per review feedback on #848, the map trait now implements update_from_dps
and joins the _updatable_traits fan-out like every other read-model trait,
picking out the vector-overlay DPs it owns and ignoring the rest. The
orchestrator no longer needs to know which DPs the map cares about.

A DpsUpdatable Protocol in the Q10 common module types the (now
heterogeneous) fan-out list, since the map trait extends TraitUpdateListener
directly rather than UpdatableTrait.

Author: tubededentifrice

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

@@ -13,6 +13,7 @@
 from .button_light import ButtonLightTrait
 from .child_lock import ChildLockTrait
 from .command import CommandTrait
+from .common import DpsUpdatable
 from .consumable import ConsumableTrait
 from .do_not_disturb import DoNotDisturbTrait
 from .dust_collection import DustCollectionTrait
@@ -94,14 +95,18 @@ def __init__(self, channel: MqttChannel) -> None:
         self.consumable = ConsumableTrait()
         self.map = MapContentTrait()
         # Read-model traits updated from the device's DPS push stream.
-        self._updatable_traits = [
+        self._updatable_traits: list[DpsUpdatable] = [
             self.status,
             self.volume,
             self.child_lock,
             self.do_not_disturb,
             self.dust_collection,
             self.network_info,
             self.consumable,
+            # The map trait owns the vector-overlay data points (no-go zones /
+            # virtual walls), which arrive as status DPs rather than in the map
+            # packet, so it updates from the DPS stream like any other read-model.
+            self.map,
         ]
         self._subscribe_task: asyncio.Task[None] | None = None
 
@@ -145,18 +150,11 @@ def _handle_message(self, message: Q10Message) -> None:
         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.
+            # only updates the fields that it is responsible for (the map trait
+            # picks out the vector-overlay data points it owns).
             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:
     """Create traits for B01 devices."""

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

@@ -11,7 +11,7 @@
 """
 
 import logging
-from typing import Any, ClassVar, cast
+from typing import Any, ClassVar, Protocol, cast
 
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.data.containers import RoborockBase
@@ -20,6 +20,18 @@
 from .command import CommandTrait
 
 
+class DpsUpdatable(Protocol):
+    """A trait that updates itself from the Q10 DPS push stream.
+
+    Implemented by :class:`UpdatableTrait` (read-model traits) and by the map
+    trait, which owns the vector-overlay data points. The ``Q10PropertiesApi``
+    subscribe loop fans each push out to every such trait; each picks out the
+    data points it is responsible for and ignores the rest.
+    """
+
+    def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: ...
+
+
 class UpdatableTrait(TraitUpdateListener):
     """Base for Q10 traits backed by a read-model updated from the DPS stream.
 

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

@@ -24,11 +24,13 @@
 import logging
 import math
 from dataclasses import dataclass, field
+from typing import Any
 
 from PIL import Image, ImageDraw
 from vacuum_map_parser_base.map_data import Area, MapData, Path, Point, Wall
 
 from roborock.data import RoborockBase
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.devices.traits.common import TraitUpdateListener
 from roborock.exceptions import RoborockException
 from roborock.map.b01_grid_layers import (
@@ -256,6 +258,25 @@ def _calibration_from_fit(self, points: list[tuple[float, float]]) -> GridCalibr
             return None
         return solve_calibration(self.layers, points, resolutions=_Q10_RESOLUTIONS)
 
+    def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
+        """Decode any vector-overlay data points present in a DPS push.
+
+        The Q10 pushes no-go / no-mop zones (``dpRestrictedZoneUp``) and virtual
+        walls (``dpVirtualWallUp``) as status data points rather than inside the
+        map packet, so the map trait joins the ``Q10PropertiesApi`` DPS fan-out
+        like the other read-model traits instead of being special-cased by the
+        orchestrator. Data points absent from this push leave the existing
+        overlays untouched (a partial status push must not wipe them); a push
+        carrying neither is a no-op.
+        """
+        if B01_Q10_DP.RESTRICTED_ZONE_UP not in decoded_dps and B01_Q10_DP.VIRTUAL_WALL_UP not in decoded_dps:
+            return
+        self.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),
+        )
+        self._notify_update()
+
     def load_overlays(
         self,
         *,

tests/devices/traits/b01/q10/test_map.py

@@ -15,6 +15,7 @@
 import pytest
 from PIL import Image
 
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.devices.traits.b01.q10 import Q10PropertiesApi, create
 from roborock.devices.traits.b01.q10.map import _Q10_RESOLUTIONS, MapContentTrait
 from roborock.exceptions import RoborockException
@@ -335,3 +336,33 @@ def test_load_overlays_partial_update_keeps_existing_zones() -> None:
     trait.load_overlays(restricted_zone_up=None, virtual_wall_up=b"\x00")
     assert len(trait.zones) == 1  # zones preserved
     assert trait.virtual_walls == []
+
+
+def test_update_from_dps_decodes_overlay_data_points() -> None:
+    """The map trait picks the overlay DPs out of a DPS push and decodes them."""
+    trait = MapContentTrait()
+    blob = (
+        bytes([1, 1])
+        + bytes([0, 4])
+        + b"".join(int.to_bytes(v & 0xFFFF, 2, "big") for xy in [(0, 0), (4, 0), (4, 4), (0, 4)] for v in xy)
+    )
+    notified = []
+    trait.add_update_listener(lambda: notified.append(True))
+
+    trait.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: blob})
+
+    assert len(trait.zones) == 1
+    assert notified  # listeners learn the overlays changed
+
+
+def test_update_from_dps_without_overlay_data_points_is_noop() -> None:
+    """A DPS push carrying neither overlay DP leaves the trait untouched."""
+    trait = MapContentTrait()
+    notified = []
+    trait.add_update_listener(lambda: notified.append(True))
+
+    trait.update_from_dps({B01_Q10_DP.BATTERY: 50})
+
+    assert trait.zones == []
+    assert trait.virtual_walls == []
+    assert not notified