fix: show docked Q10 robot on map

Author: tubededentifrice

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

@@ -4,12 +4,12 @@
 
 * map packets are decoded from map-protocol responses;
 * trace packets are decoded from trace-protocol responses;
-* restricted zones and virtual walls arrive as ordinary DPS values.
+* restricted zones, virtual walls and dock state arrive as ordinary DPS values.
 
-``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
-the latest value from each source and one replace-whole rendered image;
+``MapDpsTrait`` owns the low-level map-specific DPS read model.
+``MapContentTrait`` 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 the latest value from each source and one replace-whole image;
 calibration, path placement and overlay placement remain inside the renderer.
 """
 
@@ -18,7 +18,7 @@
 from typing import Any
 
 from roborock.data import RoborockBase
-from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, YXDeviceState
 from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
 from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import (
@@ -34,12 +34,14 @@
 from .common import UpdatableTrait
 
 _LOGGER = logging.getLogger(__name__)
+_DOCKED_STATES = {YXDeviceState.CHARGING, YXDeviceState.EMPTYING_THE_BIN}
 
 
 @dataclass
 class MapDps(RoborockBase):
     """Low-level map values delivered in the Q10 DPS stream."""
 
+    status: YXDeviceState | None = field(default=None, metadata={"dps": B01_Q10_DP.STATUS})
     restricted_zone_up: str | None = field(default=None, metadata={"dps": B01_Q10_DP.RESTRICTED_ZONE_UP})
     virtual_wall_up: str | None = field(default=None, metadata={"dps": B01_Q10_DP.VIRTUAL_WALL_UP})
 
@@ -59,22 +61,36 @@ def overlays(self) -> Q10MapOverlays:
         """Overlays decoded once from the latest relevant DPS update."""
         return self._overlays
 
+    @property
+    def robot_at_dock(self) -> bool:
+        """Whether status places the idle robot at the saved dock."""
+        return self.status in _DOCKED_STATES
+
     def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
-        """Decode overlay blobs when they arrive, then notify dependents."""
+        """Update one coherent snapshot of the DPS inputs used by the map."""
+        previous_robot_at_dock = self.robot_at_dock
+        previous_restricted_zones = self.restricted_zone_up
+        previous_virtual_walls = self.virtual_wall_up
         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)),
+        overlays_changed = (
+            self.restricted_zone_up != previous_restricted_zones or self.virtual_wall_up != previous_virtual_walls
         )
+        if overlays_changed:
+            self._overlays = Q10MapOverlays(
+                zones=tuple(parse_zone_blob(self.restricted_zone_up)),
+                virtual_walls=tuple(parse_virtual_wall_blob(self.virtual_wall_up)),
+            )
+        if not overlays_changed and self.robot_at_dock == previous_robot_at_dock:
+            return
         self._notify_update()
 
 
 class MapContentTrait(TraitUpdateListener):
     """High-level composed Q10 map view.
 
-    The latest map and trace packets are combined with the injected
-    :class:`MapDpsTrait` whenever any of those three sources changes.
+    The latest map and trace packets are combined with the injected map DPS
+    whenever any source changes.
     """
 
     def __init__(
@@ -86,6 +102,7 @@ def __init__(
         TraitUpdateListener.__init__(self, logger=_LOGGER)
         self._config = map_parser_config or B01Q10MapParserConfig()
         self._map_dps = map_dps
+        self._robot_at_dock = self._map_dps.robot_at_dock
         self._map_packet: Q10MapPacket | None = None
         self._trace_packet: Q10TracePacket | None = None
         self._image_content: bytes | None = None
@@ -124,13 +141,19 @@ def update_from_map_packet(self, packet: Q10MapPacket) -> None:
 
     def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
         """Store a trace-protocol update and render the latest sources."""
-        self._trace_packet = packet
+        # A late packet from the completed clean cannot move a robot that the
+        # map's DPS snapshot already confirmed is docked.
+        self._trace_packet = None if self._robot_at_dock else packet
         self._render()
         self._notify_update()
 
     def _map_dps_updated(self) -> None:
-        """Render after the low-level DPS source changes."""
-        if self._map_packet is None:
+        """Apply the latest map-specific DPS snapshot and recompose."""
+        self._robot_at_dock = self._map_dps.robot_at_dock
+        trace_cleared = self._robot_at_dock and self._trace_packet is not None
+        if self._robot_at_dock:
+            self._trace_packet = None
+        if self._map_packet is None and not trace_cleared:
             return
         self._render()
         self._notify_update()
@@ -145,6 +168,7 @@ def _render(self) -> None:
                 self._trace_packet,
                 self._map_dps.overlays,
                 config=self._config,
+                robot_at_dock=self._robot_at_dock,
             )
         except RoborockException as ex:
             _LOGGER.debug("Failed to render Q10 map packet: %s", ex)

roborock/map/b01_q10_render.py

@@ -14,10 +14,12 @@
 """
 
 import io
+import math
 from collections.abc import Sequence
 from dataclasses import dataclass
 
 from vacuum_map_parser_base.config.drawable import Drawable
+from vacuum_map_parser_base.config.size import Size, Sizes
 from vacuum_map_parser_base.map_data import Area, MapData, Path, Point, Wall
 
 from roborock.exceptions import RoborockException
@@ -86,6 +88,7 @@ def render_q10_map(
     overlays: Q10MapOverlays,
     *,
     config: B01Q10MapParserConfig,
+    robot_at_dock: bool = False,
 ) -> bytes:
     """Compose the latest map, trace and DPS inputs into one PNG image.
 
@@ -117,6 +120,8 @@ def render_q10_map(
         _place_trace(map_data, trace_calibration, trace, charger_heading=charger_heading)
         has_drawables = True
     has_drawables = _place_charger_from_header(map_data, packet) or has_drawables
+    if robot_at_dock:
+        has_drawables = _place_docked_robot(map_data) or has_drawables
     if vector_calibration is not None:
         _place_overlays(map_data, vector_calibration, overlays)
         has_drawables = has_drawables or bool(map_data.no_go_areas or map_data.no_mopping_areas or map_data.walls)
@@ -262,6 +267,27 @@ def _place_charger_from_header(
     return True
 
 
+def _place_docked_robot(map_data: MapData) -> bool:
+    """Place a charging robot immediately in front of the saved dock.
+
+    A zero-point idle trace has no robot coordinates. The dock heading does,
+    however, identify its outward-facing side. Offset the robot by the shared
+    unscaled V1 charger radius so the two standard glyphs meet without one
+    covering the other, and preserve the saved dock heading.
+    """
+    charger = map_data.charger
+    if charger is None or charger.a is None:
+        return False
+    angle = math.radians(charger.a)
+    offset = Sizes.SIZES[Size.CHARGER_RADIUS]
+    map_data.vacuum_position = Point(
+        charger.x + offset * math.cos(angle),
+        charger.y - offset * math.sin(angle),
+        charger.a,
+    )
+    return True
+
+
 def _place_overlays(
     map_data: MapData,
     calibration: GridCalibration,

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

@@ -18,7 +18,7 @@
 import pytest
 
 from roborock.cli import _await_q10_map_push, cli
-from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, YXDeviceState
 from roborock.devices.traits.b01.q10 import Q10PropertiesApi, create
 from roborock.devices.traits.b01.q10.map import MapContentTrait, MapDpsTrait
 from roborock.exceptions import RoborockException
@@ -37,9 +37,9 @@
 TRACE_SESSION_FIXTURE = Path("tests/map/testdata/b01_q10_trace_session.bin")
 
 
-def _map_trait() -> MapContentTrait:
+def _map_trait(map_dps: MapDpsTrait | None = None) -> MapContentTrait:
     """Create a high-level trait with its required low-level dependency."""
-    return MapContentTrait(MapDpsTrait())
+    return MapContentTrait(map_dps or MapDpsTrait())
 
 
 def _zone_blob() -> str:
@@ -238,7 +238,7 @@ def test_render_failure_clears_stale_image() -> None:
 def test_map_dps_update_renders_decoded_overlays() -> None:
     """A DPS update recomposes an existing map with decoded overlays."""
     map_dps = MapDpsTrait()
-    trait = MapContentTrait(map_dps)
+    trait = _map_trait(map_dps)
     packet = parse_map_packet(FIXTURE.read_bytes())
     notified: list[None] = []
     trait.add_update_listener(lambda: notified.append(None))
@@ -263,7 +263,7 @@ def test_map_dps_update_renders_decoded_overlays() -> None:
 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)
+    trait = _map_trait(map_dps)
 
     with (
         patch("roborock.devices.traits.b01.q10.map.parse_zone_blob", return_value=[]) as parse_zones,
@@ -291,7 +291,7 @@ def test_load_overlays_partial_update_keeps_existing_zones() -> None:
 def test_map_dps_update_without_map_does_not_notify_map_content() -> None:
     """A DPS update cannot change high-level content before a map arrives."""
     map_dps = MapDpsTrait()
-    trait = MapContentTrait(map_dps)
+    trait = _map_trait(map_dps)
     notified = []
     trait.add_update_listener(lambda: notified.append(True))
 
@@ -304,11 +304,110 @@ def test_map_dps_update_without_map_does_not_notify_map_content() -> None:
 def test_map_dps_push_without_overlay_data_points_is_noop() -> None:
     """A DPS push carrying neither overlay DP leaves both traits untouched."""
     map_dps = MapDpsTrait()
-    trait = MapContentTrait(map_dps)
+    trait = _map_trait(map_dps)
     notified = []
     trait.add_update_listener(lambda: notified.append(True))
 
     map_dps.update_from_dps({B01_Q10_DP.BATTERY: 50})
 
     assert map_dps.overlays == Q10MapOverlays()
     assert not notified
+
+
+# --- Dock state --------------------------------------------------------------
+
+
+def test_charging_status_renders_robot_at_dock() -> None:
+    """Charging status adds the idle robot marker without inventing a path."""
+    map_dps = MapDpsTrait()
+    trait = _map_trait(map_dps)
+    packet = parse_map_packet(FIXTURE.read_bytes())
+    notified: list[None] = []
+    trait.add_update_listener(lambda: notified.append(None))
+
+    with patch(
+        "roborock.devices.traits.b01.q10.map.render_q10_map",
+        side_effect=[b"map with dock", b"map with docked robot"],
+    ) as render:
+        trait.update_from_map_packet(packet)
+        notified.clear()
+        map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code})
+        map_dps.update_from_dps({B01_Q10_DP.BATTERY: 50})
+
+    assert trait.image_content == b"map with docked robot"
+    assert trait.path == []
+    assert notified == [None]
+    assert render.call_count == 2
+    assert render.call_args.kwargs["robot_at_dock"] is True
+
+
+def test_entering_docked_state_clears_stale_live_trace() -> None:
+    """A completed cleaning path cannot remain the caller-facing live position."""
+    map_dps = MapDpsTrait()
+    trait = _map_trait(map_dps)
+    trait.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1, 2), Q10Point(3, 4)]))
+    notified: list[None] = []
+    trait.add_update_listener(lambda: notified.append(None))
+
+    map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code})
+
+    assert trait.path == []
+    assert trait.robot_position is None
+    assert trait.robot_heading is None
+    assert notified == [None]
+
+
+def test_late_trace_does_not_move_docked_robot() -> None:
+    """A delayed trace packet cannot revive a completed cleaning path."""
+    map_dps = MapDpsTrait()
+    map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code})
+    trait = _map_trait(map_dps)
+
+    trait.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1, 2)]))
+
+    assert trait.path == []
+    assert trait.robot_position is None
+
+
+def test_emptying_state_keeps_robot_at_dock() -> None:
+    """Dock emptying must not briefly remove the docked robot marker."""
+    map_dps = MapDpsTrait()
+    trait = _map_trait(map_dps)
+    packet = parse_map_packet(FIXTURE.read_bytes())
+
+    with patch(
+        "roborock.devices.traits.b01.q10.map.render_q10_map",
+        side_effect=[b"map with dock", b"map while emptying"],
+    ) as render:
+        trait.update_from_map_packet(packet)
+        map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.EMPTYING_THE_BIN.code})
+
+    assert trait.image_content == b"map while emptying"
+    assert render.call_args.kwargs["robot_at_dock"] is True
+
+
+def test_combined_status_and_overlay_update_renders_once() -> None:
+    """One map DPS update publishes the complete new rendering state."""
+    map_dps = MapDpsTrait()
+    trait = _map_trait(map_dps)
+    notified: list[None] = []
+    trait.add_update_listener(lambda: notified.append(None))
+
+    with patch(
+        "roborock.devices.traits.b01.q10.map.render_q10_map",
+        side_effect=[b"base map", b"combined map"],
+    ) as render:
+        trait.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes()))
+        notified.clear()
+        map_dps.update_from_dps(
+            {
+                B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code,
+                B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob(),
+            }
+        )
+
+    assert render.call_count == 2
+    assert len(render.call_args.args[2].zones) == 1
+    assert render.call_args.kwargs["robot_at_dock"] is True
+    assert notified == [None]
+    assert trait.image_content == b"combined map"

tests/map/test_b01_q10_render.py

@@ -10,6 +10,8 @@
 from pathlib import Path
 
 from PIL import Image
+from vacuum_map_parser_base.config.size import Size, Sizes
+from vacuum_map_parser_base.map_data import MapData, Point
 
 from roborock.map.b01_grid_layers import GridCalibration
 from roborock.map.b01_q10_map_parser import (
@@ -32,6 +34,7 @@
     Q10MapOverlays,
     _calibration_from_header_metadata,
     _erased_cells,
+    _place_docked_robot,
     _vector_calibration,
     render_q10_map,
     solve_q10_calibration,
@@ -159,6 +162,35 @@ def test_render_draws_dock_from_header_without_trace() -> None:
     assert rendered != base
 
 
+def test_place_docked_robot_uses_shared_v1_marker_geometry() -> None:
+    """The idle robot sits beside the dock, facing it, without a path."""
+    map_data = MapData()
+    map_data.charger = Point(20, 30, 90)
+
+    assert _place_docked_robot(map_data)
+
+    assert map_data.vacuum_position == Point(
+        20,
+        30 - Sizes.SIZES[Size.CHARGER_RADIUS],
+        90,
+    )
+    assert map_data.path is None
+
+
+def test_zero_degree_dock_places_robot_to_its_right() -> None:
+    """The Q10 dock heading is already its outward-facing direction."""
+    map_data = MapData()
+    map_data.charger = Point(3, 3, 0)
+
+    assert _place_docked_robot(map_data)
+
+    assert map_data.vacuum_position == Point(
+        3 + Sizes.SIZES[Size.CHARGER_RADIUS],
+        3,
+        0,
+    )
+
+
 def test_render_applies_erase_zones() -> None:
     """With a calibration, erase-zone cells are blanked from the image."""
     packet, trace = _calibrated_inputs()