refactor: simplify Q10 map trait state

Author: tubededentifrice

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

@@ -32,7 +32,6 @@
     "DoNotDisturbTrait",
     "DustCollectionTrait",
     "MapContentTrait",
-    "MapDpsTrait",
     "NetworkInfoTrait",
     "SoundVolumeTrait",
     "StatusTrait",
@@ -80,8 +79,8 @@ class Q10PropertiesApi(Trait):
     map: MapContentTrait
     """Composed map image plus caller-facing map and trace data."""
 
-    map_dps: MapDpsTrait
-    """Restricted zones and virtual walls received through DPS."""
+    _map_dps: MapDpsTrait
+    """Private source of restricted zones and virtual walls received through DPS."""
 
     clean_history: CleanHistoryTrait
     """Trait for fetching the device clean-record history (``dpCleanRecord``)."""
@@ -100,8 +99,8 @@ def __init__(self, channel: B01Q10Channel) -> None:
         self.button_light = ButtonLightTrait(self.command)
         self.network_info = NetworkInfoTrait()
         self.consumable = ConsumableTrait()
-        self.map_dps = MapDpsTrait()
-        self.map = MapContentTrait(self.map_dps)
+        self._map_dps = MapDpsTrait()
+        self.map = MapContentTrait(self._map_dps)
         self.clean_history = CleanHistoryTrait(self.command)
         # Read-model traits updated from the device's DPS push stream.
         self._updatable_traits = [
@@ -113,7 +112,7 @@ def __init__(self, channel: B01Q10Channel) -> None:
             self.network_info,
             self.consumable,
             self.clean_history,
-            self.map_dps,
+            self._map_dps,
         ]
         self._subscribe_task: asyncio.Task[None] | None = None
 

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

@@ -15,6 +15,7 @@
 
 import logging
 from dataclasses import dataclass, field
+from typing import Any
 
 from roborock.data import RoborockBase
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
@@ -27,7 +28,7 @@
     Q10Room,
     Q10TracePacket,
 )
-from roborock.map.b01_q10_overlays import Q10Zone, parse_virtual_wall_blob, parse_zone_blob
+from roborock.map.b01_q10_overlays import parse_virtual_wall_blob, parse_zone_blob
 from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map
 
 from .common import UpdatableTrait
@@ -44,23 +45,29 @@ class MapDps(RoborockBase):
 
 
 class MapDpsTrait(MapDps, UpdatableTrait):
-    """Converter-backed read model for map-related DPS values."""
+    """Private read model for map-related DPS values and decoded overlays."""
 
     _CONVERTER = DpsDataConverter.from_dataclass(MapDps)
 
     def __init__(self) -> None:
         MapDps.__init__(self)
         UpdatableTrait.__init__(self, command=None, logger=_LOGGER)
+        self._overlays = Q10MapOverlays()
 
     @property
-    def zones(self) -> list[Q10Zone]:
-        """Restricted zones decoded from the latest DPS value."""
-        return parse_zone_blob(self.restricted_zone_up)
+    def overlays(self) -> Q10MapOverlays:
+        """Overlays decoded once from the latest relevant DPS update."""
+        return self._overlays
 
-    @property
-    def virtual_walls(self) -> list[Q10Zone]:
-        """Virtual walls decoded from the latest DPS value."""
-        return parse_virtual_wall_blob(self.virtual_wall_up)
+    def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
+        """Decode overlay blobs when they arrive, then notify dependents."""
+        if not self._CONVERTER.update_from_dps(self, decoded_dps):
+            return
+        self._overlays = Q10MapOverlays(
+            zones=tuple(parse_zone_blob(self.restricted_zone_up)),
+            virtual_walls=tuple(parse_virtual_wall_blob(self.virtual_wall_up)),
+        )
+        self._notify_update()
 
 
 class MapContentTrait(TraitUpdateListener):
@@ -96,17 +103,17 @@ def rooms(self) -> list[Q10Room]:
 
     @property
     def path(self) -> list[Q10Point]:
-        """Full path from the latest trace packet."""
+        """Full path for live status and callers drawing their own map overlay."""
         return self._trace_packet.points if self._trace_packet else []
 
     @property
     def robot_position(self) -> Q10Point | None:
-        """Current robot position from the latest trace packet."""
+        """Current position for live status and caller-rendered map overlays."""
         return self._trace_packet.robot_position if self._trace_packet else None
 
     @property
     def robot_heading(self) -> int | None:
-        """Current robot heading from the latest trace packet."""
+        """Current heading for orienting a robot marker on a caller-rendered map."""
         return self._trace_packet.heading if self._trace_packet else None
 
     def update_from_map_packet(self, packet: Q10MapPacket) -> None:
@@ -129,17 +136,14 @@ def _map_dps_updated(self) -> None:
         self._notify_update()
 
     def _render(self) -> None:
-        """Render the latest map, trace and DPS sources, if a map is available."""
+        """Render the required map with the latest optional trace and overlays."""
         if self._map_packet is None:
             return
         try:
             self._image_content = render_q10_map(
                 self._map_packet,
                 self._trace_packet,
-                Q10MapOverlays(
-                    zones=tuple(self._map_dps.zones),
-                    virtual_walls=tuple(self._map_dps.virtual_walls),
-                ),
+                self._map_dps.overlays,
                 config=self._config,
             )
         except RoborockException as ex:

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

@@ -28,6 +28,7 @@
     parse_map_packet,
     parse_trace_packet,
 )
+from roborock.map.b01_q10_render import Q10MapOverlays
 from roborock.protocols.b01_q10_protocol import Q10Message
 
 from .conftest import FakeB01Q10Channel
@@ -86,6 +87,13 @@ def test_q10_position_is_available_as_top_level_cli_command() -> None:
     assert "q10-position" in cli.commands
 
 
+def test_q10_map_dps_trait_is_private() -> None:
+    api = create(FakeB01Q10Channel())
+
+    assert not hasattr(api, "map_dps")
+    assert api._map_dps in api._updatable_traits
+
+
 # --- CLI push waiting --------------------------------------------------------
 
 
@@ -250,24 +258,41 @@ def test_map_dps_update_renders_decoded_overlays() -> None:
         notified.clear()
         map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()})
 
-    assert len(map_dps.zones) == 1
+    assert len(map_dps.overlays.zones) == 1
     assert trait.image_content == b"image with overlays"
     assert notified == [None]
     assert render.call_count == 2
     assert render.call_args.args[0] is packet
     assert render.call_args.args[1] is None
-    assert tuple(render.call_args.args[2].zones) == tuple(map_dps.zones)
+    assert render.call_args.args[2] is map_dps.overlays
+
+
+def test_map_dps_blobs_are_decoded_only_when_dps_arrives() -> None:
+    """Map and trace renders reuse the overlays decoded by the DPS trait."""
+    map_dps = MapDpsTrait()
+    trait = MapContentTrait(map_dps)
+
+    with (
+        patch("roborock.devices.traits.b01.q10.map.parse_zone_blob", return_value=[]) as parse_zones,
+        patch("roborock.devices.traits.b01.q10.map.parse_virtual_wall_blob", return_value=[]) as parse_walls,
+    ):
+        map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()})
+        trait.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes()))
+        trait.update_from_trace_packet(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()))
+
+    parse_zones.assert_called_once_with(_zone_blob())
+    parse_walls.assert_called_once_with(None)
 
 
 def test_load_overlays_partial_update_keeps_existing_zones() -> None:
     """A status push without the zone DP (None) must not wipe loaded zones."""
     map_dps = MapDpsTrait()
     map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()})
-    assert len(map_dps.zones) == 1
+    assert len(map_dps.overlays.zones) == 1
     # A later partial update carrying only the (empty) virtual-wall DP.
     map_dps.update_from_dps({B01_Q10_DP.VIRTUAL_WALL_UP: "AA=="})
-    assert len(map_dps.zones) == 1  # zones preserved
-    assert map_dps.virtual_walls == []
+    assert len(map_dps.overlays.zones) == 1  # zones preserved
+    assert map_dps.overlays.virtual_walls == ()
 
 
 def test_map_dps_update_without_map_does_not_notify_map_content() -> None:
@@ -279,7 +304,7 @@ def test_map_dps_update_without_map_does_not_notify_map_content() -> None:
 
     map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()})
 
-    assert len(map_dps.zones) == 1
+    assert len(map_dps.overlays.zones) == 1
     assert not notified
 
 
@@ -292,6 +317,5 @@ def test_map_dps_push_without_overlay_data_points_is_noop() -> None:
 
     map_dps.update_from_dps({B01_Q10_DP.BATTERY: 50})
 
-    assert map_dps.zones == []
-    assert map_dps.virtual_walls == []
+    assert map_dps.overlays == Q10MapOverlays()
     assert not notified