diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index b85007f0..b5d14525 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 @@ -32,6 +32,7 @@ "DoNotDisturbTrait", "DustCollectionTrait", "MapContentTrait", + "MapDpsTrait", "NetworkInfoTrait", "SoundVolumeTrait", "StatusTrait", @@ -79,6 +80,9 @@ class Q10PropertiesApi(Trait): map: MapContentTrait """Trait for fetching the current parsed map (image + rooms).""" + map_dps: MapDpsTrait + """Low-level DPS values used to compose map overlays.""" + clean_history: CleanHistoryTrait """Trait for fetching the device clean-record history (``dpCleanRecord``).""" @@ -96,7 +100,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 +113,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..1d3c3ba4 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -1,110 +1,144 @@ -"""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 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 Q10Zone, 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): + """Converter-backed read model for map-related DPS values.""" - 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) - robot_position: Q10Point | None = None - """Current robot position (the most recent path point), if known.""" + @property + def zones(self) -> list[Q10Zone]: + """Restricted zones decoded from the latest DPS value.""" + return parse_zone_blob(self.restricted_zone_up) - 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})" + @property + def virtual_walls(self) -> list[Q10Zone]: + """Virtual walls decoded from the latest DPS value.""" + return parse_virtual_wall_blob(self.virtual_wall_up) -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 | None = None, *, 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 or MapDpsTrait() + 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 a map has been pushed.""" + 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 from the latest trace packet.""" + return self._trace_packet.points if self._trace_packet else [] + + @property + def robot_position(self) -> Q10Point | None: + """Current robot position from the latest trace packet.""" + return self._trace_packet.robot_position if self._trace_packet else None + + @property + def robot_heading(self) -> int | None: + """Current robot heading from the latest trace packet.""" + 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.""" + self._render() self._notify_update() + + def _render(self) -> None: + """Render the latest map, trace and DPS sources, if a map is available.""" + if self._map_packet is None: + return + try: + self._image_content = render_q10_map( + self._map_packet, + self._trace_packet, + Q10MapOverlays( + zones=tuple(self._map_dps.zones), + virtual_walls=tuple(self._map_dps.virtual_walls), + ), + config=self._config, + ) + except RoborockException as ex: + _LOGGER.debug("Failed to render Q10 map packet: %s", ex) diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 65216496..50622600 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -1,12 +1,17 @@ """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 dataclasses import replace from pathlib import Path from typing import cast from unittest.mock import Mock @@ -14,19 +19,51 @@ 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.map.b01_grid_layers import GridCalibration +from roborock.map.b01_q10_map_parser import ( + Q10HeaderCalibration, + Q10MapPacket, + Q10Point, + Q10TracePacket, + parse_map_packet, + parse_trace_packet, +) 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") + +# A header calibration whose pixel origin (0, 5) is usable (not a keepalive +# frame), so a short path can calibrate the fixture map. +_USABLE_HEADER = Q10HeaderCalibration(origin_x=0, origin_y=50, resolution=5, charger_x=0, charger_y=0, charger_phi=0) + + +def _trait_with_map() -> MapContentTrait: + """A trait with the fixture map already pushed into it.""" + trait = MapContentTrait() + trait.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) + return trait + + +def _floor_world_points(packet: Q10MapPacket, cal: GridCalibration, count: int) -> list[Q10Point]: + """``count`` world points lying on the map's floor under ``cal``.""" + layers = packet.layers + floor = [ + (px, py) + for py in range(layers.height) + for px in range(layers.width) + if layers.cell_class(layers.grid[py * layers.width + px]) == "floor" + ] + return [Q10Point(*(int(v) for v in cal.pixel_to_world(px, py))) for px, py in floor[:count]] 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, rooms and map data.""" packet = parse_map_packet(FIXTURE.read_bytes()) trait = MapContentTrait() updates: list[None] = [] @@ -37,22 +74,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()) + """A pushed 02 01 trace packet populates the path, position and heading.""" + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) trait = MapContentTrait() 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 @@ -75,13 +113,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 +137,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 +209,119 @@ 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 = MapContentTrait() + 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_trace_update_projects_short_path_using_header() -> None: + """A map header and short trace are sufficient to render a path.""" + trait = MapContentTrait() + packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER) + trait.update_from_map_packet(packet) + base = trait.image_content + assert base is not None + true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1) + trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) + assert len(trait.path) < 20 # far too short for the full origin+resolution fit + + assert trait.image_content is not None + assert trait.image_content != base + + +def test_short_trace_without_header_cannot_be_projected() -> None: + """Without a header origin a short trace cannot be placed on the map.""" + packet = parse_map_packet(FIXTURE.read_bytes()) + trait = MapContentTrait() + trait.update_from_map_packet(packet) # the fixture header is a keepalive frame + base = trait.image_content + true = GridCalibration(resolution=10.0, origin_x=0.0, origin_y=5.0, y_sign=1) + trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) + assert trait.image_content == base + + +# --- Overlays ---------------------------------------------------------------- + + +def test_load_overlays_places_zones_after_calibration() -> None: + """Decoded no-go / no-mop zones are drawn once the sources calibrate.""" + map_dps = MapDpsTrait() + trait = MapContentTrait(map_dps) + packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER) + trait.update_from_map_packet(packet) + true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1) + trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) + before = trait.image_content + assert before is not None + + def rect(zone_type: int, corners: list[tuple[int, int]]) -> bytes: + out = bytes([zone_type, len(corners)]) + for x, y in corners: + out += int.to_bytes(x & 0xFFFF, 2, "big") + int.to_bytes(y & 0xFFFF, 2, "big") + return out.ljust(18, b"\x00") + + blob = bytes([1, 1]) + rect(0, [(0, 0), (40, 0), (40, 40), (0, 40)]) + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()}) + + assert len(map_dps.zones) == 1 + assert trait.image_content != before + + +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() + blob = ( + bytes([1, 1]) + + bytes([0, 4]) + + b"".join(int.to_bytes(v & 0xFFFF, 2, "big") for xy in [(0, 0), (4, 0), (4, 4), (0, 4)] for v in xy) + ) + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()}) + assert len(map_dps.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.zones) == 1 # zones preserved + assert map_dps.virtual_walls == [] + + +def test_map_dps_trait_updates_high_level_map_content() -> None: + """The low-level DPS trait notifies the dependent high-level map trait.""" + map_dps = MapDpsTrait() + trait = MapContentTrait(map_dps) + blob = ( + bytes([1, 1]) + + bytes([0, 4]) + + b"".join(int.to_bytes(v & 0xFFFF, 2, "big") for xy in [(0, 0), (4, 0), (4, 4), (0, 4)] for v in xy) + ) + notified = [] + trait.add_update_listener(lambda: notified.append(True)) + + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()}) + + assert len(map_dps.zones) == 1 + assert notified # listeners learn the overlays changed + + +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.zones == [] + assert map_dps.virtual_walls == [] + assert not notified