diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index ee51352a..e206623d 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/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) diff --git a/roborock/devices/traits/common.py b/roborock/devices/traits/common.py index f1b3be76..35a6220e 100644 --- a/roborock/devices/traits/common.py +++ b/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 diff --git a/roborock/map/b01_q10_render.py b/roborock/map/b01_q10_render.py index d4e37f64..97ab6797 100644 --- a/roborock/map/b01_q10_render.py +++ b/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, diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 61120a7e..724d5006 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -10,7 +10,7 @@ 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 @@ -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: @@ -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,7 +306,7 @@ 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)) @@ -312,3 +314,99 @@ def test_map_dps_push_without_overlay_data_points_is_noop() -> None: 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" diff --git a/tests/devices/traits/test_common.py b/tests/devices/traits/test_common.py new file mode 100644 index 00000000..2ae0ec35 --- /dev/null +++ b/tests/devices/traits/test_common.py @@ -0,0 +1,30 @@ +"""Tests for common trait utilities.""" + +from dataclasses import dataclass, field +from enum import IntEnum + +from roborock.data import RoborockBase +from roborock.devices.traits.common import DpsDataConverter + + +class FakeDps(IntEnum): + """Data points for the test model.""" + + VALUE = 1 + + +@dataclass +class FakeData(RoborockBase): + """Small data model for converter tests.""" + + value: int | None = field(default=None, metadata={"dps": FakeDps.VALUE}) + + +def test_update_from_dps_reports_only_value_changes() -> None: + """The converter reports a change only when the target value changes.""" + converter = DpsDataConverter.from_dataclass(FakeData) + target = FakeData() + + assert converter.update_from_dps(target, {FakeDps.VALUE: 1}) + assert not converter.update_from_dps(target, {FakeDps.VALUE: 1}) + assert converter.update_from_dps(target, {FakeDps.VALUE: 2}) diff --git a/tests/map/test_b01_q10_render.py b/tests/map/test_b01_q10_render.py index 6620ece2..153f25b3 100644 --- a/tests/map/test_b01_q10_render.py +++ b/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()