diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index b85007f0..3c8c73ff 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/roborock/devices/traits/b01/q10/__init__.py @@ -16,7 +16,7 @@ from .consumable import ConsumableTrait from .do_not_disturb import DoNotDisturbTrait from .dust_collection import DustCollectionTrait -from .map import MapContentTrait +from .map import MapContentTrait, MapDpsTrait from .network_info import NetworkInfoTrait from .remote import RemoteTrait from .status import StatusTrait @@ -77,7 +77,10 @@ class Q10PropertiesApi(Trait): """Trait exposing remaining life of consumables.""" map: MapContentTrait - """Trait for fetching the current parsed map (image + rooms).""" + """Composed map image plus caller-facing map and trace data.""" + + _map_dps: MapDpsTrait + """Private source of restricted zones and virtual walls received through DPS.""" clean_history: CleanHistoryTrait """Trait for fetching the device clean-record history (``dpCleanRecord``).""" @@ -96,7 +99,8 @@ def __init__(self, channel: B01Q10Channel) -> None: self.button_light = ButtonLightTrait(self.command) self.network_info = NetworkInfoTrait() self.consumable = ConsumableTrait() - self.map = MapContentTrait() + self._map_dps = MapDpsTrait() + self.map = MapContentTrait(self._map_dps) self.clean_history = CleanHistoryTrait(self.command) # Read-model traits updated from the device's DPS push stream. self._updatable_traits = [ @@ -108,6 +112,7 @@ def __init__(self, channel: B01Q10Channel) -> None: self.network_info, self.consumable, self.clean_history, + self._map_dps, ] self._subscribe_task: asyncio.Task[None] | None = None diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index c132def4..ee51352a 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -1,110 +1,151 @@ -"""Map content trait for B01 Q10 devices. - -Unlike the v1 / Q7 maps, the Q10 has no synchronous "get map" command, so this -trait is purely push-driven and mirrors the Q10 ``StatusTrait`` contract: - -- The device pushes its current map/path as protocol-301 ``MAP_RESPONSE`` - messages (a ``dpRequestDps`` nudges it to do so). The protocol layer decodes - those into :class:`Q10MapPacket` / :class:`Q10TracePacket` objects and the - ``Q10PropertiesApi`` subscribe loop routes them to - :meth:`MapContentTrait.update_from_map_packet` / - :meth:`MapContentTrait.update_from_trace_packet`. -- Those methods render/cache the content and notify update listeners (register - via :meth:`add_update_listener`). -- ``image_content``, ``map_data``, ``rooms``, ``path`` and ``robot_position`` - are readable and reflect the most recently pushed map. - -Unlike the Q7, the Q10 map payload is unencrypted, so no map key is required. +"""Push-driven map traits for B01 Q10 devices. + +Map-related state arrives on three independent streams: + +* 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. + +``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; +calibration, path placement and overlay placement remain inside the renderer. """ import logging from dataclasses import dataclass, field - -from vacuum_map_parser_base.map_data import MapData +from typing import Any from roborock.data import RoborockBase -from roborock.devices.traits.common import TraitUpdateListener +from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP +from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener +from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import ( - B01Q10MapParser, B01Q10MapParserConfig, Q10MapPacket, Q10Point, Q10Room, Q10TracePacket, ) +from roborock.map.b01_q10_overlays import parse_virtual_wall_blob, parse_zone_blob +from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map -_LOGGER = logging.getLogger(__name__) +from .common import UpdatableTrait -_TRUNCATE_LENGTH = 20 +_LOGGER = logging.getLogger(__name__) @dataclass -class MapContent(RoborockBase): - """Dataclass representing Q10 map content.""" +class MapDps(RoborockBase): + """Low-level map values delivered in the Q10 DPS stream.""" - image_content: bytes | None = None - """The rendered image of the map in PNG format.""" + 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}) - map_data: MapData | None = None - """Parsed map data (image metadata + room names).""" - rooms: list[Q10Room] = field(default_factory=list) - """Rooms (segments) reported by the device, with ids and names.""" +class MapDpsTrait(MapDps, UpdatableTrait): + """Private read model for map-related DPS values and decoded overlays.""" - path: list[Q10Point] = field(default_factory=list) - """Full path of the current cleaning session (oldest point first). + _CONVERTER = DpsDataConverter.from_dataclass(MapDps) - The robot accumulates this server-side and serves the whole trajectory so - far in one packet, so it is complete even if we connect mid-session. Only - populated while a cleaning session is active.""" + def __init__(self) -> None: + MapDps.__init__(self) + UpdatableTrait.__init__(self, command=None, logger=_LOGGER) + self._overlays = Q10MapOverlays() - robot_position: Q10Point | None = None - """Current robot position (the most recent path point), if known.""" + @property + def overlays(self) -> Q10MapOverlays: + """Overlays decoded once from the latest relevant DPS update.""" + return self._overlays - def __repr__(self) -> str: - img = self.image_content - if img and len(img) > _TRUNCATE_LENGTH: - img = img[: _TRUNCATE_LENGTH - 3] + b"..." - return f"MapContent(image_content={img!r}, rooms={self.rooms!r})" + def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: + """Decode overlay blobs when they arrive, then notify dependents.""" + 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)), + ) + self._notify_update() -class MapContentTrait(MapContent, TraitUpdateListener): - """Trait holding the most recently pushed parsed map content for Q10 devices. +class MapContentTrait(TraitUpdateListener): + """High-level composed Q10 map view. - The Q10 has no synchronous get-map request; the device pushes map and trace - packets, which the protocol layer decodes and the ``Q10PropertiesApi`` - subscribe loop feeds into :meth:`update_from_map_packet` / - :meth:`update_from_trace_packet`. Consumers read the cached fields and/or - register a callback with :meth:`add_update_listener` to be notified when new - map content arrives. + The latest map and trace packets are combined with the injected + :class:`MapDpsTrait` whenever any of those three sources changes. """ def __init__( self, + map_dps: MapDpsTrait, *, map_parser_config: B01Q10MapParserConfig | None = None, ) -> None: - super().__init__() TraitUpdateListener.__init__(self, logger=_LOGGER) - self._map_parser = B01Q10MapParser(map_parser_config) + self._config = map_parser_config or B01Q10MapParserConfig() + self._map_dps = map_dps + self._map_packet: Q10MapPacket | None = None + self._trace_packet: Q10TracePacket | 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 composed map PNG, if the latest map rendered successfully.""" + return self._image_content + + @property + def rooms(self) -> list[Q10Room]: + """Rooms reported by the device.""" + return self._map_packet.rooms if self._map_packet else [] + + @property + def path(self) -> list[Q10Point]: + """Full path for live status and callers drawing their own map overlay.""" + return self._trace_packet.points if self._trace_packet else [] + + @property + def robot_position(self) -> Q10Point | None: + """Current position for live status and caller-rendered map overlays.""" + return self._trace_packet.robot_position if self._trace_packet else None + + @property + def robot_heading(self) -> int | None: + """Current heading for orienting a robot marker on a caller-rendered map.""" + return self._trace_packet.heading if self._trace_packet else None def update_from_map_packet(self, packet: Q10MapPacket) -> None: - """Render a pushed full-map packet into the cached image/rooms. - - Rendering failures are logged and skipped (listeners are not notified) so - a single bad push cannot tear down the subscribe loop. - """ - parsed = self._map_parser.parse_packet(packet) - if parsed.image_content is None: - _LOGGER.debug("Failed to render Q10 map image") - return - self.image_content = parsed.image_content - self.map_data = parsed.map_data - self.rooms = packet.rooms + """Store a map-protocol update and render the latest sources.""" + self._map_packet = packet + self._render() self._notify_update() def update_from_trace_packet(self, packet: Q10TracePacket) -> None: - """Cache the path/robot position from a pushed trace packet.""" - self.path = packet.points - self.robot_position = packet.robot_position + """Store a trace-protocol update and render the latest sources.""" + self._trace_packet = 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: + return + self._render() + self._notify_update() + + def _render(self) -> None: + """Render the required map with the latest optional trace and overlays.""" + if self._map_packet is None: + return + try: + self._image_content = render_q10_map( + self._map_packet, + self._trace_packet, + self._map_dps.overlays, + config=self._config, + ) + except RoborockException as ex: + _LOGGER.debug("Failed to render Q10 map packet: %s", ex) + self._image_content = None diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 65216496..7bb32a23 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -1,34 +1,60 @@ """Tests for the Q10 B01 map content trait. The Q10 map API is push-driven: the device publishes ``MAP_RESPONSE`` messages -and the trait updates its cached state from them via ``update_from_map_response`` -(there is no synchronous get-map request). +which the protocol layer decodes into typed map/trace packets; the trait updates +its cached state from them via ``update_from_map_packet`` / +``update_from_trace_packet`` (there is no synchronous get-map request). These +tests cover that state management; the pixel/geometry work it drives is tested in +``tests/map/test_b01_q10_render.py``. """ import asyncio +import base64 from collections.abc import AsyncGenerator from pathlib import Path from typing import cast -from unittest.mock import Mock +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.devices.traits.b01.q10 import Q10PropertiesApi, create -from roborock.devices.traits.b01.q10.map import MapContentTrait -from roborock.map.b01_q10_map_parser import Q10Point, parse_map_packet, parse_trace_packet +from roborock.devices.traits.b01.q10.map import MapContentTrait, MapDpsTrait +from roborock.exceptions import RoborockException +from roborock.map.b01_q10_map_parser import ( + Q10Point, + Q10TracePacket, + parse_map_packet, + parse_trace_packet, +) +from roborock.map.b01_q10_render import Q10MapOverlays from roborock.protocols.b01_q10_protocol import Q10Message from .conftest import FakeB01Q10Channel FIXTURE = Path("tests/map/testdata/b01_q10_map.bin") -TRACE_FIXTURE = Path("tests/map/testdata/b01_q10_trace_multi.bin") +TRACE_SESSION_FIXTURE = Path("tests/map/testdata/b01_q10_trace_session.bin") + + +def _map_trait() -> MapContentTrait: + """Create a high-level trait with its required low-level dependency.""" + return MapContentTrait(MapDpsTrait()) + + +def _zone_blob() -> str: + """Return one base64-encoded restricted-zone DPS value.""" + vertices = [(0, 0), (40, 0), (40, 40), (0, 40)] + record = bytes([0, len(vertices)]) + b"".join( + int.to_bytes(value & 0xFFFF, 2, "big") for point in vertices for value in point + ) + return base64.b64encode(bytes([1, 1]) + record).decode() def test_update_from_map_packet_populates_image_and_rooms() -> None: - """A parsed 01 01 map packet populates the image, rooms and map data.""" + """A pushed 01 01 map packet populates the image and rooms.""" packet = parse_map_packet(FIXTURE.read_bytes()) - trait = MapContentTrait() + trait = _map_trait() updates: list[None] = [] trait.add_update_listener(lambda: updates.append(None)) @@ -37,22 +63,23 @@ def test_update_from_map_packet_populates_image_and_rooms() -> None: assert trait.image_content is not None assert trait.image_content[:8] == b"\x89PNG\r\n\x1a\n" assert {room.id: room.name for room in trait.rooms} == {2: "Living Room", 3: "Bedroom"} - assert trait.map_data is not None assert len(updates) == 1 def test_update_from_trace_packet_populates_path_and_position() -> None: - """A parsed 02 01 trace packet populates the path and robot position.""" - packet = parse_trace_packet(TRACE_FIXTURE.read_bytes()) - trait = MapContentTrait() + """A pushed 02 01 trace packet populates the path, position and heading.""" + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + trait = _map_trait() updates: list[None] = [] trait.add_update_listener(lambda: updates.append(None)) - trait.update_from_trace_packet(packet) + trait.update_from_trace_packet(trace) - assert [(p.x, p.y) for p in trait.path] == [(100, 200), (150, 250), (-50, 300)] + assert len(trait.path) == 14 + assert (trait.path[0].x, trait.path[0].y) == (41, 64) assert trait.robot_position is not None - assert (trait.robot_position.x, trait.robot_position.y) == (-50, 300) + assert (trait.robot_position.x, trait.robot_position.y) == (276, -1) + assert trait.robot_heading == -34 assert len(updates) == 1 @@ -60,12 +87,19 @@ def test_q10_position_is_available_as_top_level_cli_command() -> None: assert "q10-position" in cli.commands +def test_q10_map_dps_trait_is_private() -> None: + api = create(FakeB01Q10Channel()) + + assert not hasattr(api, "map_dps") + assert api._map_dps in api._updatable_traits + + # --- CLI push waiting -------------------------------------------------------- class _FakeQ10Properties: def __init__(self) -> None: - self.map = MapContentTrait() + self.map = _map_trait() self.refresh_count = 0 async def refresh(self) -> None: @@ -75,13 +109,13 @@ async def refresh(self) -> None: class _FakeQ10PropertiesWithTrace(_FakeQ10Properties): async def refresh(self) -> None: await super().refresh() - self.map.update_from_trace_packet(parse_trace_packet(TRACE_FIXTURE.read_bytes())) + self.map.update_from_trace_packet(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes())) async def test_await_q10_map_push_waits_for_fresh_update() -> None: """A cached trace alone is not treated as a successful new map push.""" properties = _FakeQ10Properties() - properties.map.path = [Q10Point(1, 2)] + properties.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1, 2)])) got_trace = await _await_q10_map_push( cast(Q10PropertiesApi, properties), lambda: bool(properties.map.path), timeout=0.01 @@ -99,12 +133,12 @@ async def test_await_q10_map_push_returns_true_after_update() -> None: ) assert got_trace is True - assert [(p.x, p.y) for p in properties.map.path] == [(100, 200), (150, 250), (-50, 300)] + assert len(properties.map.path) == 14 async def test_await_q10_map_push_can_fall_back_to_cached_map_on_timeout() -> None: properties = _FakeQ10Properties() - properties.map.image_content = b"cached-png" + properties.map.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) got_map = await _await_q10_map_push( cast(Q10PropertiesApi, properties), @@ -171,7 +205,117 @@ async def test_subscribe_loop_routes_trace_push( """A trace pushed onto the stream is routed to the map trait by the loop.""" assert not q10_api.map.path - message_queue.put_nowait(parse_trace_packet(TRACE_FIXTURE.read_bytes())) + message_queue.put_nowait(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes())) await _wait_for(lambda: bool(q10_api.map.path)) assert q10_api.map.robot_position is not None + + +# --- Source composition + rendering ------------------------------------------ + + +def test_trace_without_map_is_retained_without_rendering() -> None: + """A trace is retained even when no map is available to render yet.""" + trait = _map_trait() + trait.update_from_trace_packet(Q10TracePacket(points=[Q10Point(i, 0) for i in range(30)])) + assert len(trait.path) == 30 + assert trait.image_content is None + + +def test_render_failure_clears_stale_image() -> 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) + + assert trait.path == trace.points + assert trait.image_content is None + + +# --- Overlays ---------------------------------------------------------------- + + +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) + 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()}) + + 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 + + +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) + + with ( + patch("roborock.devices.traits.b01.q10.map.parse_zone_blob", return_value=[]) as parse_zones, + patch("roborock.devices.traits.b01.q10.map.parse_virtual_wall_blob", return_value=[]) as parse_walls, + ): + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()}) + trait.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) + trait.update_from_trace_packet(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes())) + + parse_zones.assert_called_once_with(_zone_blob()) + parse_walls.assert_called_once_with(None) + + +def test_load_overlays_partial_update_keeps_existing_zones() -> None: + """A status push without the zone DP (None) must not wipe loaded zones.""" + map_dps = MapDpsTrait() + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()}) + assert len(map_dps.overlays.zones) == 1 + # A later partial update carrying only the (empty) virtual-wall DP. + map_dps.update_from_dps({B01_Q10_DP.VIRTUAL_WALL_UP: "AA=="}) + assert len(map_dps.overlays.zones) == 1 # zones preserved + assert map_dps.overlays.virtual_walls == () + + +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) + notified = [] + trait.add_update_listener(lambda: notified.append(True)) + + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()}) + + assert len(map_dps.overlays.zones) == 1 + assert not notified + + +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) + 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