perf(map): eliminate redundant PNG serialization in Q10 rendering pipeline

roborock/map/b01_q10_map_parser.py

@@ -801,8 +801,8 @@ def parse_packet(self, packet: Q10MapPacket) -> ParsedMapData:
         """
         return self.parsed_from_packet(packet)
 
-    def parsed_from_packet(self, packet: Q10MapPacket) -> ParsedMapData:
-        """Render a (possibly erase-modified) packet into a PNG + ``MapData``."""
+    def map_data_from_packet(self, packet: Q10MapPacket) -> MapData:
+        """Build MapData with a rendered base image, without PNG serialization."""
         image = self._render(packet)
 
         map_data = MapData()
@@ -830,8 +830,14 @@ def parsed_from_packet(self, packet: Q10MapPacket) -> ParsedMapData:
         if packet.carpet_mask is not None:
             map_data.carpet_map = {i for i, value in enumerate(packet.carpet_mask) if value}
 
+        return map_data
+
+    def parsed_from_packet(self, packet: Q10MapPacket) -> ParsedMapData:
+        """Render a (possibly erase-modified) packet into a PNG + ``MapData``."""
+        map_data = self.map_data_from_packet(packet)
         image_bytes = io.BytesIO()
-        image.save(image_bytes, format=_MAP_FILE_FORMAT)
+        if map_data.image is not None:
+            map_data.image.data.save(image_bytes, format=_MAP_FILE_FORMAT)
         return ParsedMapData(image_content=image_bytes.getvalue(), map_data=map_data)
 
     def _render(self, packet: Q10MapPacket) -> Image.Image:

roborock/map/b01_q10_render.py

@@ -109,10 +109,9 @@ def render_q10_map(
             # areas disappear (as the app shows).
             render_packet = erased_packet(packet, cells)
 
-    parsed = parser.parsed_from_packet(render_packet)
-    if parsed.image_content is None or parsed.map_data is None:
+    map_data = parser.map_data_from_packet(render_packet)
+    if map_data.image is None:
         raise RoborockException("Failed to render Q10 map image")
-    map_data = parsed.map_data
 
     has_drawables = False
     if trace_calibration is not None and trace is not None:
@@ -128,7 +127,9 @@ def render_q10_map(
     if has_drawables:
         return _draw_map_content(map_data, config=config)
 
-    return parsed.image_content
+    buffer = io.BytesIO()
+    map_data.image.data.save(buffer, format="PNG")
+    return buffer.getvalue()
 
 
 def solve_q10_calibration(

tests/map/test_benchmarks.py

@@ -37,7 +37,9 @@
     Q10MapPacketKind,
     Q10Room,
     parse_map_packet,
+    parse_trace_packet,
 )
+from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map
 from roborock.map.map_parser import MapParser, MapParserConfig, ParsedMapData
 
 if TYPE_CHECKING:
@@ -63,6 +65,7 @@ def _runner(func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
 _V1_S5_MAP = _TESTDATA_DIR / "v1" / "s5_fw2008_with_segments.bin"
 _V1_S6_MAP = _TESTDATA_DIR / "v1" / "s6_fw2652_with_active_segment_and_no_mop_zone.bin"
 _Q10_MAP = _TESTDATA_DIR / "b01_q10_map.bin"
+_Q10_TRACE = _TESTDATA_DIR / "b01_q10_trace_session.bin"
 
 
 def _build_synthetic_q10_grid(width: int = 200, height: int = 200) -> Q10MapPacket:
@@ -98,14 +101,15 @@ def _build_synthetic_q10_grid(width: int = 200, height: int = 200) -> Q10MapPack
 
 
 def _validate_image(
-    result: ParsedMapData | None,
+    result: ParsedMapData | bytes | None,
     expected_size: tuple[int, int] | None = None,
 ) -> None:
     """Validate rendered PNG image integrity and pixel bounds outside the timed benchmark."""
     assert result is not None
-    assert result.image_content is not None
-    assert len(result.image_content) > 0
-    image = Image.open(io.BytesIO(result.image_content))
+    image_bytes = result if isinstance(result, bytes) else result.image_content
+    assert image_bytes is not None
+    assert len(image_bytes) > 0
+    image = Image.open(io.BytesIO(image_bytes))
     assert image.format == "PNG"
     assert image.mode == "RGBA"
     assert image.getbbox() is not None  # Image has non-empty drawn content
@@ -203,6 +207,17 @@ def test_benchmark_q10_map_full_scale_render_scale_4(benchmark: BenchmarkFixture
     _validate_image(result, (800, 800))
 
 
+def test_benchmark_q10_map_render_with_trace(benchmark: BenchmarkFixture) -> None:
+    """Benchmark Q10 render_q10_map end-to-end composite pipeline with trace overlay."""
+    map_bytes = _Q10_MAP.read_bytes()
+    trace_bytes = _Q10_TRACE.read_bytes()
+    packet = parse_map_packet(map_bytes)
+    trace = parse_trace_packet(trace_bytes)
+    config = B01Q10MapParserConfig(map_scale=4)
+    result = benchmark(render_q10_map, packet, trace=trace, overlays=Q10MapOverlays(), config=config)
+    _validate_image(result, (32, 24))
+
+
 # ---------------------------------------------------------------------------
 # Standalone CLI Benchmark & Profiler
 # ---------------------------------------------------------------------------
@@ -225,6 +240,11 @@ def _run_benchmarks(iterations: int, warmup: int, profile: bool) -> None:
     p_q10_scale1 = B01Q10MapParser(B01Q10MapParserConfig(map_scale=1))
     p_q10_scale4 = B01Q10MapParser(B01Q10MapParserConfig(map_scale=4))
 
+    q10_trace_data = _Q10_TRACE.read_bytes()
+    q10_map_packet = parse_map_packet(q10_data)
+    q10_trace_packet = parse_trace_packet(q10_trace_data)
+    q10_render_config = B01Q10MapParserConfig(map_scale=4)
+
     benchmarks: list[tuple[str, Callable[[], Any]]] = [
         ("V1 S5 Map (Scale 4, Default)", lambda: p_v1_s5.parse(v1_s5_data)),
         ("V1 S6 Map (Scale 4, Default)", lambda: p_v1_s6_scale4.parse(v1_s6_data)),
@@ -236,6 +256,15 @@ def _run_benchmarks(iterations: int, warmup: int, profile: bool) -> None:
         ("Q10 Small Map (Scale 1)", lambda: p_q10_small.parse(q10_data)),
         ("Q10 200x200 Map (Scale 1)", lambda: p_q10_scale1.parse_packet(q10_packet_200)),
         ("Q10 200x200 Map (Scale 4)", lambda: p_q10_scale4.parse_packet(q10_packet_200)),
+        (
+            "Q10 Render with Trace",
+            lambda: render_q10_map(
+                q10_map_packet,
+                trace=q10_trace_packet,
+                overlays=Q10MapOverlays(),
+                config=q10_render_config,
+            ),
+        ),
     ]
 
     print(f"\nRunning {len(benchmarks)} benchmarks ({warmup} warmup, {iterations} timed iterations)...\n")
@@ -255,7 +284,7 @@ def _run_benchmarks(iterations: int, warmup: int, profile: bool) -> None:
         min_ms = timings[0]
         mean_ms = sum(timings) / len(timings)
         med_ms = timings[len(timings) // 2]
-        p95_ms = timings[int(len(timings) * 0.95)]
+        p95_ms = timings[max(0, int((len(timings) - 1) * 0.95))]
         ops_per_sec = 1000.0 / mean_ms if mean_ms > 0 else float("inf")
 
         results.append((name, min_ms, med_ms, mean_ms, p95_ms, ops_per_sec))