From 9a6161b28275fc5cd53fdaa9bb3a99ddd52a50c5 Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 22:57:09 +0400 Subject: [PATCH 1/3] feat: compose Q10 map content from grouped traits --- roborock/devices/traits/b01/q10/__init__.py | 10 +- roborock/devices/traits/b01/q10/map.py | 170 ++++++++++-------- tests/devices/traits/b01/q10/test_map.py | 184 ++++++++++++++++++-- 3 files changed, 277 insertions(+), 87 deletions(-) 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 From b85893dd9d472d8484a7abfef68f69e0377325b5 Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 23:08:44 +0400 Subject: [PATCH 2/3] refactor: tighten Q10 map trait lifecycle --- roborock/devices/traits/b01/q10/__init__.py | 4 +- roborock/devices/traits/b01/q10/map.py | 9 +- tests/devices/traits/b01/q10/test_map.py | 136 ++++++++------------ 3 files changed, 61 insertions(+), 88 deletions(-) diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index b5d14525..19421ffc 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/roborock/devices/traits/b01/q10/__init__.py @@ -78,10 +78,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 - """Low-level DPS values used to compose map overlays.""" + """Restricted zones and virtual walls received through DPS.""" clean_history: CleanHistoryTrait """Trait for fetching the device clean-record history (``dpCleanRecord``).""" diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index 1d3c3ba4..ec1e24f5 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -72,13 +72,13 @@ class MapContentTrait(TraitUpdateListener): def __init__( self, - map_dps: MapDpsTrait | None = None, + map_dps: MapDpsTrait, *, map_parser_config: B01Q10MapParserConfig | None = None, ) -> None: TraitUpdateListener.__init__(self, logger=_LOGGER) self._config = map_parser_config or B01Q10MapParserConfig() - self._map_dps = map_dps or MapDpsTrait() + self._map_dps = map_dps self._map_packet: Q10MapPacket | None = None self._trace_packet: Q10TracePacket | None = None self._image_content: bytes | None = None @@ -86,7 +86,7 @@ def __init__( @property def image_content(self) -> bytes | None: - """The composed map PNG, if a map has been pushed.""" + """The composed map PNG, if the latest map rendered successfully.""" return self._image_content @property @@ -123,6 +123,8 @@ def update_from_trace_packet(self, packet: Q10TracePacket) -> None: 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() @@ -142,3 +144,4 @@ def _render(self) -> None: ) 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 50622600..ae2fcb38 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -11,10 +11,9 @@ 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 +from unittest.mock import Mock, patch import pytest @@ -22,10 +21,8 @@ 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, MapDpsTrait -from roborock.map.b01_grid_layers import GridCalibration +from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import ( - Q10HeaderCalibration, - Q10MapPacket, Q10Point, Q10TracePacket, parse_map_packet, @@ -38,34 +35,25 @@ FIXTURE = Path("tests/map/testdata/b01_q10_map.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 _map_trait() -> MapContentTrait: + """Create a high-level trait with its required low-level dependency.""" + return MapContentTrait(MapDpsTrait()) -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 _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 pushed 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)) @@ -80,7 +68,7 @@ def test_update_from_map_packet_populates_image_and_rooms() -> None: def test_update_from_trace_packet_populates_path_and_position() -> None: """A pushed 02 01 trace packet populates the path, position and heading.""" trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) - trait = MapContentTrait() + trait = _map_trait() updates: list[None] = [] trait.add_update_listener(lambda: updates.append(None)) @@ -103,7 +91,7 @@ def test_q10_position_is_available_as_top_level_cli_command() -> None: class _FakeQ10Properties: def __init__(self) -> None: - self.map = MapContentTrait() + self.map = _map_trait() self.refresh_count = 0 async def refresh(self) -> None: @@ -220,74 +208,61 @@ async def test_subscribe_loop_routes_trace_push( 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 = _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_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_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) -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 + assert trait.path == trace.points + assert trait.image_content is None # --- Overlays ---------------------------------------------------------------- -def test_load_overlays_places_zones_after_calibration() -> None: - """Decoded no-go / no-mop zones are drawn once the sources calibrate.""" +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 = 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") + packet = parse_map_packet(FIXTURE.read_bytes()) + notified: list[None] = [] + trait.add_update_listener(lambda: notified.append(None)) - 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()}) + 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.zones) == 1 - assert trait.image_content != before + 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 tuple(render.call_args.args[2].zones) == tuple(map_dps.zones) 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()}) + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()}) 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=="}) @@ -295,22 +270,17 @@ def test_load_overlays_partial_update_keeps_existing_zones() -> None: 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.""" +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) - 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()}) + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()}) assert len(map_dps.zones) == 1 - assert notified # listeners learn the overlays changed + assert not notified def test_map_dps_push_without_overlay_data_points_is_noop() -> None: From 033aabc9afe26dfd6a5147cad8f53fd13d82b23f Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Mon, 20 Jul 2026 18:26:56 +0400 Subject: [PATCH 3/3] refactor: simplify Q10 map trait state --- roborock/devices/traits/b01/q10/__init__.py | 11 +++--- roborock/devices/traits/b01/q10/map.py | 38 +++++++++++--------- tests/devices/traits/b01/q10/test_map.py | 40 ++++++++++++++++----- 3 files changed, 58 insertions(+), 31 deletions(-) diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index 19421ffc..3c8c73ff 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/roborock/devices/traits/b01/q10/__init__.py @@ -32,7 +32,6 @@ "DoNotDisturbTrait", "DustCollectionTrait", "MapContentTrait", - "MapDpsTrait", "NetworkInfoTrait", "SoundVolumeTrait", "StatusTrait", @@ -80,8 +79,8 @@ class Q10PropertiesApi(Trait): map: MapContentTrait """Composed map image plus caller-facing map and trace data.""" - map_dps: MapDpsTrait - """Restricted zones and virtual walls received through DPS.""" + _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``).""" @@ -100,8 +99,8 @@ def __init__(self, channel: B01Q10Channel) -> None: self.button_light = ButtonLightTrait(self.command) self.network_info = NetworkInfoTrait() self.consumable = ConsumableTrait() - self.map_dps = MapDpsTrait() - self.map = MapContentTrait(self.map_dps) + 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 = [ @@ -113,7 +112,7 @@ def __init__(self, channel: B01Q10Channel) -> None: self.network_info, self.consumable, self.clean_history, - self.map_dps, + 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 ec1e24f5..ee51352a 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -15,6 +15,7 @@ import logging from dataclasses import dataclass, field +from typing import Any from roborock.data import RoborockBase from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP @@ -27,7 +28,7 @@ Q10Room, Q10TracePacket, ) -from roborock.map.b01_q10_overlays import Q10Zone, parse_virtual_wall_blob, parse_zone_blob +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 from .common import UpdatableTrait @@ -44,23 +45,29 @@ class MapDps(RoborockBase): class MapDpsTrait(MapDps, UpdatableTrait): - """Converter-backed read model for map-related DPS values.""" + """Private read model for map-related DPS values and decoded overlays.""" _CONVERTER = DpsDataConverter.from_dataclass(MapDps) def __init__(self) -> None: MapDps.__init__(self) UpdatableTrait.__init__(self, command=None, logger=_LOGGER) + self._overlays = Q10MapOverlays() @property - def zones(self) -> list[Q10Zone]: - """Restricted zones decoded from the latest DPS value.""" - return parse_zone_blob(self.restricted_zone_up) + def overlays(self) -> Q10MapOverlays: + """Overlays decoded once from the latest relevant DPS update.""" + return self._overlays - @property - def virtual_walls(self) -> list[Q10Zone]: - """Virtual walls decoded from the latest DPS value.""" - return parse_virtual_wall_blob(self.virtual_wall_up) + 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(TraitUpdateListener): @@ -96,17 +103,17 @@ def rooms(self) -> list[Q10Room]: @property def path(self) -> list[Q10Point]: - """Full path from the latest trace packet.""" + """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 robot position from the latest trace packet.""" + """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 robot heading from the latest trace packet.""" + """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: @@ -129,17 +136,14 @@ def _map_dps_updated(self) -> None: self._notify_update() def _render(self) -> None: - """Render the latest map, trace and DPS sources, if a map is available.""" + """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, - Q10MapOverlays( - zones=tuple(self._map_dps.zones), - virtual_walls=tuple(self._map_dps.virtual_walls), - ), + self._map_dps.overlays, config=self._config, ) except RoborockException as ex: diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index ae2fcb38..7bb32a23 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -28,6 +28,7 @@ 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 @@ -86,6 +87,13 @@ 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 -------------------------------------------------------- @@ -250,24 +258,41 @@ def test_map_dps_update_renders_decoded_overlays() -> None: notified.clear() map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()}) - assert len(map_dps.zones) == 1 + 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 tuple(render.call_args.args[2].zones) == tuple(map_dps.zones) + 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.zones) == 1 + 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.zones) == 1 # zones preserved - assert map_dps.virtual_walls == [] + 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: @@ -279,7 +304,7 @@ def test_map_dps_update_without_map_does_not_notify_map_content() -> None: map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()}) - assert len(map_dps.zones) == 1 + assert len(map_dps.overlays.zones) == 1 assert not notified @@ -292,6 +317,5 @@ def test_map_dps_push_without_overlay_data_points_is_noop() -> None: map_dps.update_from_dps({B01_Q10_DP.BATTERY: 50}) - assert map_dps.zones == [] - assert map_dps.virtual_walls == [] + assert map_dps.overlays == Q10MapOverlays() assert not notified