refactor: cache one composed Q10 map image

Author: tubededentifrice

roborock/cli.py

@@ -745,10 +745,9 @@ async def q10_map_with_path(ctx, device_id: str, output_file: str):
     if not got_path:
         click.echo("No live path available (the robot only reports its path while cleaning).")
         return
-    try:
-        image = map_trait.render_path_on_map()
-    except RoborockException as err:
-        click.echo(f"Could not render path on map: {err}")
+    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)

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

@@ -9,7 +9,7 @@
 ``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 result; calibration,
+one grouped source snapshot and one replace-whole rendered image; calibration,
 path placement and overlay placement are not independently mutable trait state.
 """
 
@@ -28,7 +28,7 @@
     Q10TracePacket,
 )
 from roborock.map.b01_q10_overlays import Q10Zone, parse_virtual_wall_blob, parse_zone_blob
-from roborock.map.b01_q10_render import Q10MapOverlays, Q10MapRender, draw_path_on_map, render_q10_map
+from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map
 
 from .common import UpdatableTrait
 
@@ -66,7 +66,7 @@ class MapContentTrait(TraitUpdateListener):
 
     Map and trace packet updates replace :attr:`_source`; DPS updates are owned
     by the injected :class:`MapDpsTrait`. Rendering always produces one new
-    :class:`Q10MapRender`, keeping the externally visible fields consistent.
+    image, keeping the externally visible fields consistent.
     """
 
     def __init__(
@@ -79,13 +79,13 @@ def __init__(
         self._config = map_parser_config or B01Q10MapParserConfig()
         self._map_dps = map_dps or MapDpsTrait()
         self._source = Q10MapSource()
-        self._render: Q10MapRender | None = None
+        self._image_content: bytes | None = None
         self._map_dps.add_update_listener(self._map_dps_updated)
 
     @property
     def image_content(self) -> bytes | None:
-        """The rendered base map PNG, if a map has been pushed."""
-        return self._render.image_content if self._render else None
+        """The composed map PNG, if a map has been pushed."""
+        return self._image_content
 
     @property
     def rooms(self) -> list[Q10Room]:
@@ -128,7 +128,7 @@ def update_from_map_packet(self, packet: Q10MapPacket) -> None:
         if render is None:
             return
         self._source = source
-        self._render = render
+        self._image_content = render
         self._notify_update()
 
     def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
@@ -138,29 +138,13 @@ def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
             self._rebuild()
         self._notify_update()
 
-    def render_path_on_map(
-        self,
-        *,
-        line_color: tuple[int, int, int, int] = (235, 64, 52, 255),
-        position_color: tuple[int, int, int, int] = (255, 211, 0, 255),
-    ) -> bytes:
-        """Return a PNG with the path, robot, zones and walls drawn."""
-        if self._render is None:
-            raise RoborockException("No map available; no map has been pushed yet")
-        return draw_path_on_map(
-            self._render,
-            config=self._config,
-            line_color=line_color,
-            position_color=position_color,
-        )
-
     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()
         self._notify_update()
 
-    def _compose(self, source: Q10MapSource) -> Q10MapRender | None:
+    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
@@ -179,4 +163,4 @@ def _rebuild(self) -> None:
         """Replace the derived render from the current source snapshot."""
         render = self._compose(self._source)
         if render is not None:
-            self._render = render
+            self._image_content = render

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

@@ -22,7 +22,6 @@
 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 MapContentTrait, MapDpsTrait
-from roborock.exceptions import RoborockException
 from roborock.map.b01_grid_layers import GridCalibration
 from roborock.map.b01_q10_map_parser import (
     Q10HeaderCalibration,
@@ -232,48 +231,25 @@ def test_trace_update_projects_short_path_using_header() -> None:
     trait = MapContentTrait()
     packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER)
     trait.update_from_map_packet(packet)
+    base = trait.image_content
+    assert base is not None
     true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1)
     trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6)))
     assert len(trait.path) < 20  # far too short for the full origin+resolution fit
 
-    assert trait.render_path_on_map()[:8] == b"\x89PNG\r\n\x1a\n"
+    assert trait.image_content is not None
+    assert trait.image_content != base
 
 
 def test_short_trace_without_header_cannot_be_projected() -> None:
     """Without a header origin a short trace cannot be placed on the map."""
     packet = parse_map_packet(FIXTURE.read_bytes())
     trait = MapContentTrait()
     trait.update_from_map_packet(packet)  # the fixture header is a keepalive frame
+    base = trait.image_content
     true = GridCalibration(resolution=10.0, origin_x=0.0, origin_y=5.0, y_sign=1)
     trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6)))
-    with pytest.raises(RoborockException, match="No calibration available"):
-        trait.render_path_on_map()
-
-
-def test_render_path_on_map_requires_map() -> None:
-    trait = MapContentTrait()
-    with pytest.raises(RoborockException, match="No map available"):
-        trait.render_path_on_map()
-
-
-def test_render_path_on_map_uses_derived_content() -> None:
-    """The renderer draws the already-derived map content as a PNG."""
-    trait = MapContentTrait()
-    packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER)
-    trait.update_from_map_packet(packet)
-    true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1)
-    trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6)))
-
-    png = trait.render_path_on_map()
-
-    assert png[:8] == b"\x89PNG\r\n\x1a\n"
-
-
-def test_render_path_on_map_without_path_cannot_calibrate() -> None:
-    """A map but no cleaning path -> no calibration -> a clear error."""
-    trait = _trait_with_map()
-    with pytest.raises(RoborockException, match="No calibration available"):
-        trait.render_path_on_map()
+    assert trait.image_content == base
 
 
 # --- Overlays ----------------------------------------------------------------
@@ -287,7 +263,8 @@ def test_load_overlays_places_zones_after_calibration() -> None:
     trait.update_from_map_packet(packet)
     true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1)
     trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6)))
-    before = trait.render_path_on_map()
+    before = trait.image_content
+    assert before is not None
 
     def rect(zone_type: int, corners: list[tuple[int, int]]) -> bytes:
         out = bytes([zone_type, len(corners)])
@@ -299,7 +276,7 @@ def rect(zone_type: int, corners: list[tuple[int, int]]) -> bytes:
     map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()})
 
     assert len(trait.zones) == 1
-    assert trait.render_path_on_map() != before
+    assert trait.image_content != before
 
 
 def test_load_overlays_partial_update_keeps_existing_zones() -> None: