refactor: simplify Q10 map trait composition

Author: tubededentifrice

roborock/cli.py

@@ -721,39 +721,6 @@ async def q10_position(ctx, device_id: str, include_path: bool):
     click.echo(dump_json(summary))
 
 
-@session.command()
-@click.option("--device_id", required=True)
-@click.option("--output-file", required=True, help="Path to save the map image with the path drawn.")
-@click.pass_context
-@async_command
-async def q10_map_with_path(ctx, device_id: str, output_file: str):
-    """Render the Q10 map with the current cleaning path + robot position drawn.
-
-    Needs the robot to be actively cleaning so a live trace is available.
-    Fetches the map and path and writes the annotated PNG.
-    """
-    context: RoborockContext = ctx.obj
-    device_manager = await context.get_device_manager()
-    device = await device_manager.get_device(device_id)
-    if device.b01_q10_properties is None:
-        click.echo("Feature not supported by device")
-        return
-    properties = device.b01_q10_properties
-    map_trait = properties.map
-    await _await_q10_map_push(properties, lambda: map_trait.image_content is not None)
-    got_path = await _await_q10_map_push(properties, lambda: bool(map_trait.path))
-    if not got_path:
-        click.echo("No live path available (the robot only reports its path while cleaning).")
-        return
-    image = map_trait.image_content
-    if image is None:
-        click.echo("No map image content available.")
-        return
-    with open(output_file, "wb") as f:
-        f.write(image)
-    click.echo(f"Saved map with {len(map_trait.path)}-point path to {output_file}")
-
-
 @session.command()
 @click.option("--device_id", required=True)
 @click.pass_context

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

@@ -9,12 +9,12 @@
 ``MapDpsTrait`` owns the low-level DPS read model. ``MapContentTrait`` depends
 on it and combines that state with the latest map/trace packets through the pure
 functions in :mod:`roborock.map.b01_q10_render`. The high-level trait keeps only
-one grouped source snapshot and one replace-whole rendered image; calibration,
-path placement and overlay placement are not independently mutable trait state.
+the latest value from each source and one replace-whole rendered image;
+calibration, path placement and overlay placement remain inside the renderer.
 """
 
 import logging
-from dataclasses import dataclass, field, replace
+from dataclasses import dataclass, field
 
 from roborock.data import RoborockBase
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
@@ -52,21 +52,22 @@ def __init__(self) -> None:
         MapDps.__init__(self)
         UpdatableTrait.__init__(self, command=None, logger=_LOGGER)
 
+    @property
+    def zones(self) -> list[Q10Zone]:
+        """Restricted zones decoded from the latest DPS value."""
+        return parse_zone_blob(self.restricted_zone_up)
 
-@dataclass(frozen=True)
-class Q10MapSource:
-    """The latest map-protocol inputs, replaced atomically on every push."""
-
-    map_packet: Q10MapPacket | None = None
-    trace_packet: Q10TracePacket | None = None
+    @property
+    def virtual_walls(self) -> list[Q10Zone]:
+        """Virtual walls decoded from the latest DPS value."""
+        return parse_virtual_wall_blob(self.virtual_wall_up)
 
 
 class MapContentTrait(TraitUpdateListener):
     """High-level composed Q10 map view.
 
-    Map and trace packet updates replace :attr:`_source`; DPS updates are owned
-    by the injected :class:`MapDpsTrait`. Rendering always produces one new
-    image, keeping the externally visible fields consistent.
+    The latest map and trace packets are combined with the injected
+    :class:`MapDpsTrait` whenever any of those three sources changes.
     """
 
     def __init__(
@@ -78,7 +79,8 @@ def __init__(
         TraitUpdateListener.__init__(self, logger=_LOGGER)
         self._config = map_parser_config or B01Q10MapParserConfig()
         self._map_dps = map_dps or MapDpsTrait()
-        self._source = Q10MapSource()
+        self._map_packet: Q10MapPacket | None = None
+        self._trace_packet: Q10TracePacket | None = None
         self._image_content: bytes | None = None
         self._map_dps.add_update_listener(self._map_dps_updated)
 
@@ -90,77 +92,53 @@ def image_content(self) -> bytes | None:
     @property
     def rooms(self) -> list[Q10Room]:
         """Rooms reported by the device."""
-        packet = self._source.map_packet
-        return packet.rooms if packet else []
+        return self._map_packet.rooms if self._map_packet else []
 
     @property
     def path(self) -> list[Q10Point]:
         """Full path from the latest trace packet."""
-        trace = self._source.trace_packet
-        return trace.points if trace else []
+        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."""
-        trace = self._source.trace_packet
-        return trace.robot_position if trace else None
+        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."""
-        trace = self._source.trace_packet
-        return trace.heading if trace else None
-
-    @property
-    def zones(self) -> list[Q10Zone]:
-        """Restricted zones decoded from the low-level DPS trait."""
-        return parse_zone_blob(self._map_dps.restricted_zone_up)
-
-    @property
-    def virtual_walls(self) -> list[Q10Zone]:
-        """Virtual walls decoded from the low-level DPS trait."""
-        return parse_virtual_wall_blob(self._map_dps.virtual_wall_up)
+        return self._trace_packet.heading if self._trace_packet else None
 
     def update_from_map_packet(self, packet: Q10MapPacket) -> None:
-        """Replace the current map packet and compose a consistent result."""
-        source = replace(self._source, map_packet=packet)
-        render = self._compose(source)
-        if render is None:
-            return
-        self._source = source
-        self._image_content = render
+        """Store a map-protocol update and render the latest sources."""
+        self._map_packet = packet
+        self._render()
         self._notify_update()
 
     def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
-        """Replace the complete current-session trace packet."""
-        self._source = replace(self._source, trace_packet=packet)
-        if self._source.map_packet is not None:
-            self._rebuild()
+        """Store a trace-protocol update and render the latest sources."""
+        self._trace_packet = packet
+        self._render()
         self._notify_update()
 
     def _map_dps_updated(self) -> None:
-        """Recompose placed overlays after the low-level DPS state changes."""
-        if self._source.map_packet is not None:
-            self._rebuild()
+        """Render after the low-level DPS source changes."""
+        self._render()
         self._notify_update()
 
-    def _compose(self, source: Q10MapSource) -> bytes | None:
-        """Compose a source snapshot, preserving the previous result on error."""
-        if source.map_packet is None:
-            return None
+    def _render(self) -> None:
+        """Render the latest map, trace and DPS sources, if a map is available."""
+        if self._map_packet is None:
+            return
         try:
-            return render_q10_map(
-                source.map_packet,
-                source.trace_packet,
-                Q10MapOverlays(zones=tuple(self.zones), virtual_walls=tuple(self.virtual_walls)),
+            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),
+                ),
                 config=self._config,
             )
         except RoborockException as ex:
             _LOGGER.debug("Failed to render Q10 map packet: %s", ex)
-            return None
-
-    def _rebuild(self) -> None:
-        """Replace the derived render from the current source snapshot."""
-        render = self._compose(self._source)
-        if render is not None:
-            self._image_content = render

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

@@ -256,7 +256,7 @@ def test_short_trace_without_header_cannot_be_projected() -> None:
 
 
 def test_load_overlays_places_zones_after_calibration() -> None:
-    """Decoded no-go / no-mop zones are placed on MapData once calibrated."""
+    """Decoded no-go / no-mop zones are drawn once the sources calibrate."""
     map_dps = MapDpsTrait()
     trait = MapContentTrait(map_dps)
     packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER)
@@ -275,25 +275,24 @@ def rect(zone_type: int, corners: list[tuple[int, int]]) -> bytes:
     blob = bytes([1, 1]) + rect(0, [(0, 0), (40, 0), (40, 40), (0, 40)])
     map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()})
 
-    assert len(trait.zones) == 1
+    assert len(map_dps.zones) == 1
     assert trait.image_content != before
 
 
 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()
-    trait = MapContentTrait(map_dps)
     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)
     )
     map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()})
-    assert len(trait.zones) == 1
+    assert len(map_dps.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(trait.zones) == 1  # zones preserved
-    assert trait.virtual_walls == []
+    assert len(map_dps.zones) == 1  # zones preserved
+    assert map_dps.virtual_walls == []
 
 
 def test_map_dps_trait_updates_high_level_map_content() -> None:
@@ -310,7 +309,7 @@ def test_map_dps_trait_updates_high_level_map_content() -> None:
 
     map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()})
 
-    assert len(trait.zones) == 1
+    assert len(map_dps.zones) == 1
     assert notified  # listeners learn the overlays changed
 
 
@@ -323,6 +322,6 @@ def test_map_dps_push_without_overlay_data_points_is_noop() -> None:
 
     map_dps.update_from_dps({B01_Q10_DP.BATTERY: 50})
 
-    assert trait.zones == []
-    assert trait.virtual_walls == []
+    assert map_dps.zones == []
+    assert map_dps.virtual_walls == []
     assert not notified