fix: show docked Q10 robot on map (#904)

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,8 +61,13 @@ 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."""
         if not self._CONVERTER.update_from_dps(self, decoded_dps):
             return
         self._overlays = Q10MapOverlays(
@@ -73,8 +80,8 @@ def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
 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__(
@@ -129,7 +136,7 @@ def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
         self._notify_update()
 
     def _map_dps_updated(self) -> None:
-        """Render after the low-level DPS source changes."""
+        """Render after the low-level map DPS source changes."""
         if self._map_packet is None:
             return
         self._render()
@@ -142,9 +149,10 @@ def _render(self) -> None:
         try:
             self._image_content = render_q10_map(
                 self._map_packet,
-                self._trace_packet,
+                self._trace_packet if not self._map_dps.robot_at_dock else None,
                 self._map_dps.overlays,
                 config=self._config,
+                robot_at_dock=self._map_dps.robot_at_dock,
             )
         except RoborockException as ex:
             _LOGGER.debug("Failed to render Q10 map packet: %s", ex)

roborock/devices/traits/common.py

@@ -105,10 +105,14 @@ def update_from_dps(self, target: RoborockBase, decoded_dps: dict[TDps, Any]) ->
             decoded_dps: The decoded DPS data to convert.
 
         Returns:
-            True if any values were updated, False otherwise.
+            True if any values changed, False otherwise.
         """
         conversions = RoborockBase.convert_dict(self._dps_type_map, decoded_dps)
+        changed = False
         for dps_id, value in conversions.items():
             field_name = self._dps_field_map[dps_id]
+            if getattr(target, field_name) == value:
+                continue
             setattr(target, field_name, value)
-        return bool(conversions)
+            changed = True
+        return changed

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

@@ -10,15 +10,15 @@
 
 import asyncio
 import base64
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Generator
 from pathlib import Path
 from typing import cast
 from unittest.mock import Mock, patch
 
 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:
@@ -51,6 +51,12 @@ def _zone_blob() -> str:
     return base64.b64encode(bytes([1, 1]) + record).decode()
 
 
+@pytest.fixture(name="render_map")
+def render_map_fixture() -> Generator[Mock, None, None]:
+    with patch("roborock.devices.traits.b01.q10.map.render_q10_map") as render:
+        yield render
+
+
 def test_update_from_map_packet_populates_image_and_rooms() -> None:
     """A pushed 01 01 map packet populates the image and rooms."""
     packet = parse_map_packet(FIXTURE.read_bytes())
@@ -215,18 +221,16 @@ def test_trace_without_map_is_retained_without_rendering() -> None:
     assert trait.image_content is None
 
 
-def test_render_failure_clears_stale_image() -> None:
+def test_render_failure_clears_stale_image(render_map: Mock) -> None:
     """A failed composition cannot leave an image from older source data."""
     packet = parse_map_packet(FIXTURE.read_bytes())
     trace = Q10TracePacket(points=[Q10Point(1, 2)])
     trait = _map_trait()
 
-    with patch(
-        "roborock.devices.traits.b01.q10.map.render_q10_map",
-        side_effect=[b"initial image", RoborockException("invalid map")],
-    ):
-        trait.update_from_map_packet(packet)
-        trait.update_from_trace_packet(trace)
+    render_map.side_effect = [b"initial image", RoborockException("invalid map")]
+
+    trait.update_from_map_packet(packet)
+    trait.update_from_trace_packet(trace)
 
     assert trait.path == trace.points
     assert trait.image_content is None
@@ -235,35 +239,33 @@ def test_render_failure_clears_stale_image() -> None:
 # --- Overlays ----------------------------------------------------------------
 
 
-def test_map_dps_update_renders_decoded_overlays() -> None:
+def test_map_dps_update_renders_decoded_overlays(render_map: Mock) -> 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))
 
-    with patch(
-        "roborock.devices.traits.b01.q10.map.render_q10_map",
-        side_effect=[b"base image", b"image with overlays"],
-    ) as render:
-        trait.update_from_map_packet(packet)
-        notified.clear()
-        map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()})
+    render_map.side_effect = [b"base image", b"image with overlays"]
+
+    trait.update_from_map_packet(packet)
+    notified.clear()
+    map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()})
 
     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 render.call_args.args[2] is map_dps.overlays
+    assert render_map.call_count == 2
+    assert render_map.call_args.args[0] is packet
+    assert render_map.call_args.args[1] is None
+    assert render_map.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)
+    trait = _map_trait(map_dps)
 
     with (
         patch("roborock.devices.traits.b01.q10.map.parse_zone_blob", return_value=[]) as parse_zones,
@@ -291,7 +293,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 +306,107 @@ 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
+
+
+async def test_charging_status_renders_robot_at_dock(render_map: Mock) -> 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())
+    updated = asyncio.Event()
+    trait.add_update_listener(updated.set)
+    render_map.side_effect = [b"map with dock", b"map with docked robot"]
+
+    trait.update_from_map_packet(packet)
+    updated.clear()
+    map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code})
+    map_dps.update_from_dps({B01_Q10_DP.BATTERY: 50})
+
+    await asyncio.wait_for(updated.wait(), timeout=1)
+
+    assert trait.image_content == b"map with docked robot"
+    assert trait.path == []
+    assert render_map.call_count == 2
+    assert render_map.call_args.kwargs["robot_at_dock"] is True
+
+
+def test_docked_state_hides_trace_only_from_rendering(render_map: Mock) -> None:
+    """A docked render omits the valid trace without deleting source data."""
+    map_dps = MapDpsTrait()
+    trait = _map_trait(map_dps)
+    packet = parse_map_packet(FIXTURE.read_bytes())
+    trace = Q10TracePacket(points=[Q10Point(1, 2), Q10Point(3, 4)])
+    render_map.return_value = b"map"
+
+    trait.update_from_map_packet(packet)
+    trait.update_from_trace_packet(trace)
+    assert render_map.call_args.args[1] is trace
+
+    map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code})
+
+    assert trait.path == trace.points
+    assert render_map.call_args.args[1] is None
+    assert render_map.call_args.kwargs["robot_at_dock"] is True
+
+
+def test_late_trace_is_retained_but_hidden_while_docked(render_map: Mock) -> None:
+    """A late trace stays available but is not part of a docked render."""
+    map_dps = MapDpsTrait()
+    map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code})
+    trait = _map_trait(map_dps)
+    trace = Q10TracePacket(points=[Q10Point(1, 2)])
+    render_map.return_value = b"map"
+
+    trait.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes()))
+    trait.update_from_trace_packet(trace)
+
+    assert trait.path == trace.points
+    assert render_map.call_args.args[1] is None
+
+
+def test_emptying_state_keeps_robot_at_dock(render_map: Mock) -> 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())
+
+    render_map.side_effect = [b"map with dock", b"map while emptying"]
+
+    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_map.call_args.kwargs["robot_at_dock"] is True
+
+
+async def test_combined_status_and_overlay_update_renders_once(render_map: Mock) -> None:
+    """One map DPS update publishes the complete new rendering state."""
+    map_dps = MapDpsTrait()
+    trait = _map_trait(map_dps)
+    updated = asyncio.Event()
+    trait.add_update_listener(updated.set)
+    render_map.side_effect = [b"base map", b"combined map"]
+
+    trait.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes()))
+    updated.clear()
+    map_dps.update_from_dps(
+        {
+            B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code,
+            B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob(),
+        }
+    )
+
+    await asyncio.wait_for(updated.wait(), timeout=1)
+
+    assert render_map.call_count == 2
+    assert len(render_map.call_args.args[2].zones) == 1
+    assert render_map.call_args.kwargs["robot_at_dock"] is True
+    assert trait.image_content == b"combined map"