diff --git a/roborock/data/b01_q10/b01_q10_containers.py b/roborock/data/b01_q10/b01_q10_containers.py index 8550bd7c6..90b23fbdf 100644 --- a/roborock/data/b01_q10/b01_q10_containers.py +++ b/roborock/data/b01_q10/b01_q10_containers.py @@ -144,7 +144,9 @@ class Q10MapInfo(RoborockBase): """A saved map reported by ``dpMultiMap``. Q10 firmware represents the map identifier as a string on the wire. The - value is sent back unchanged in a subsequent ``{"op": "get"}`` request. + value is sent back unchanged in a subsequent ``{"op": "select"}`` detail + request. On Q10 firmware, ``select`` previews a saved map without applying + it as the active map. """ id: str diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index c36b25af1..194568ab9 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/roborock/devices/traits/b01/q10/__init__.py @@ -8,7 +8,13 @@ from roborock.data.containers import RoborockBase from roborock.devices.rpc.b01_q10_channel import B01Q10Channel from roborock.devices.traits import Trait -from roborock.map.b01_q10_map_parser import B01Q10MapParserConfig, Q10MapPacket, Q10TracePacket +from roborock.map.b01_q10_map_parser import ( + B01Q10MapParserConfig, + Q10CleanRecordDetail, + Q10MapPacket, + Q10MapPacketKind, + Q10TracePacket, +) from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message from .button_light import ButtonLightTrait @@ -92,7 +98,12 @@ class Q10PropertiesApi(Trait): clean_history: CleanHistoryTrait """Trait for fetching the device clean-record history (``dpCleanRecord``).""" - def __init__(self, channel: B01Q10Channel, *, map_parser_config: B01Q10MapParserConfig | None = None) -> None: + def __init__( + self, + channel: B01Q10Channel, + *, + map_parser_config: B01Q10MapParserConfig | None = None, + ) -> None: """Initialize the B01Props API.""" self._channel = channel self.command = CommandTrait(channel) @@ -107,9 +118,13 @@ def __init__(self, channel: B01Q10Channel, *, map_parser_config: B01Q10MapParser self.consumable = ConsumableTrait() self._map_dps = MapDpsTrait() self.maps = MapsTrait(self.command) - self.map = MapContentTrait(self._map_dps, self.maps, self.command, map_parser_config=map_parser_config) - self.vacuum = VacuumTrait(self.command, self.status, self.map) + self.map = MapContentTrait( + self._map_dps, + self.command, + map_parser_config=map_parser_config, + ) self.clean_history = CleanHistoryTrait(self.command) + self.vacuum = VacuumTrait(self.command, self.status, self.map) # Read-model traits updated from the device's DPS push stream. self._updatable_traits = [ self.status, @@ -158,7 +173,12 @@ def _handle_message(self, message: Q10Message) -> None: Map-list DPS responses and other DPS updates feed the read-model traits. """ if isinstance(message, Q10MapPacket): - self.map.update_from_map_packet(message) + if message.kind is Q10MapPacketKind.CURRENT: + self.map.update_from_map_packet(message) + elif message.kind is Q10MapPacketKind.SAVED_MAP_DETAIL: + self.maps.update_from_map_packet(message) + elif isinstance(message, Q10CleanRecordDetail): + self.clean_history.update_from_detail(message) elif isinstance(message, Q10TracePacket): self.map.update_from_trace_packet(message) elif isinstance(message, Q10DpsUpdate): @@ -179,6 +199,13 @@ def as_dict(self) -> dict[str, Any]: return result -def create(channel: B01Q10Channel, *, map_parser_config: B01Q10MapParserConfig | None = None) -> Q10PropertiesApi: +def create( + channel: B01Q10Channel, + *, + map_parser_config: B01Q10MapParserConfig | None = None, +) -> Q10PropertiesApi: """Create traits for B01 devices.""" - return Q10PropertiesApi(channel, map_parser_config=map_parser_config) + return Q10PropertiesApi( + channel, + map_parser_config=map_parser_config, + ) diff --git a/roborock/devices/traits/b01/q10/clean_history.py b/roborock/devices/traits/b01/q10/clean_history.py index fde2bb2fe..3b1549b6a 100644 --- a/roborock/devices/traits/b01/q10/clean_history.py +++ b/roborock/devices/traits/b01/q10/clean_history.py @@ -14,6 +14,7 @@ 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, YXCleaningResult, @@ -22,11 +23,20 @@ YXStartMethod, ) from roborock.data.b01_q10.b01_q10_containers import Q10CleanRecord +from roborock.exceptions import RoborockException +from roborock.map.b01_q10_map_parser import ( + Q10CleanRecordDetail, + Q10HistoricalTracePacket, + Q10MapPacket, + Q10MapPacketKind, + Q10Point, +) from .command import CommandTrait from .common import UpdatableTrait __all__ = [ + "CleanHistory", "CleanHistoryTrait", "CleanRecordConverter", "CleanRecordPush", @@ -107,26 +117,53 @@ def parse_record(raw: Any | None) -> Q10CleanRecord | None: return None -class CleanHistoryTrait(UpdatableTrait): - """Access to the Q10 clean-record history (``dpCleanRecord``, DP 52). - - A read-model trait updated from the DPS stream like the others, but it overrides - :meth:`update_from_dps` because the payload is a structured push (a record list, - or a single ``op:"notify"`` record) rather than a flat data-point-to-field map. - """ +@dataclass +class CleanHistory(RoborockBase): + """Received clean-history data, independent of request and transport state.""" - def __init__(self, command: CommandTrait) -> None: - """Initialize the clean history trait.""" - UpdatableTrait.__init__(self, command, _LOGGER) - self._converter = CleanRecordConverter() - self.records: list[Q10CleanRecord] = [] - """Decoded clean records, most recent first.""" + records: list[Q10CleanRecord] = field(default_factory=list) + """Decoded clean records, most recent first.""" + detail: Q10CleanRecordDetail | None = None + """Latest received archive. Its wire payload contains no record identifier.""" + detail_path: list[Q10Point] = field(default_factory=list) + """Historical path points from the latest received archive, included in diagnostics.""" @property def last_record(self) -> Q10CleanRecord | None: """The most recent clean record, or ``None`` if there are none.""" return self.records[0] if self.records else None + @property + def detail_packet(self) -> Q10MapPacket | None: + """The map from the most recently received clean-record detail.""" + return self.detail.map if self.detail else None + + @property + def detail_trace(self) -> Q10HistoricalTracePacket | None: + """Historical path embedded in the latest received clean-record detail.""" + return self.detail.trace if self.detail else None + + def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]: + """Serialize history metadata and path without binary map packets and grids.""" + return super().as_dict(exclude=(exclude or set()) | {"detail"}) + + +class CleanHistoryTrait(CleanHistory, UpdatableTrait): + """Browse past cleaning runs and retrieve their parsed maps and historical paths. + + Subscribe with ``add_update_listener`` to receive requested or unsolicited + archives. Archive pushes contain no record ID, so the latest detail cannot + be attributed to a particular selection. + """ + + _command: CommandTrait + + def __init__(self, command: CommandTrait) -> None: + CleanHistory.__init__(self) + UpdatableTrait.__init__(self, command, _LOGGER) + self._command = command + self._converter = CleanRecordConverter() + async def refresh(self) -> None: """Request the clean-record list from the device. @@ -134,13 +171,24 @@ async def refresh(self) -> None: asynchronously on the device stream and populate :attr:`records` once :meth:`update_from_dps` processes the ``dpCleanRecord`` push. """ - if self._command is None: - raise ValueError("Trait is read-only; no command channel was provided") await self._command.send( B01_Q10_DP.COMMON, params={str(B01_Q10_DP.CLEAN_RECORD.code): {"op": "list"}}, ) + async def refresh_detail(self, record: Q10CleanRecord) -> None: + """Request the saved map and path for a past cleaning run. + + Returns after sending; listeners receive the archive separately. The + push contains no record ID to correlate it with this selection. + """ + if not record.raw or not record.map_len: + raise RoborockException("The Q10 clean record has no saved map detail") + await self._command.send( + B01_Q10_DP.COMMON, + {str(B01_Q10_DP.CLEAN_RECORD.code): {"op": "select", "id": record.raw}}, + ) + def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: """Apply a ``dpCleanRecord`` push (a full list reply or a single notify).""" envelope = decoded_dps.get(B01_Q10_DP.CLEAN_RECORD) @@ -151,6 +199,14 @@ def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: return self._apply(push) + def update_from_detail(self, detail: Q10CleanRecordDetail) -> None: + """Store a pushed clean-record map and path and notify listeners.""" + if detail.map.kind is not Q10MapPacketKind.CLEAN_RECORD_DETAIL: + raise ValueError(f"Expected a Q10 clean-record detail packet, got {detail.map.kind.value}") + self.detail = detail + self.detail_path = list(detail.trace.points) if detail.trace else [] + self._notify_update() + def _apply(self, push: CleanRecordPush) -> None: """Merge or replace the records from ``push``, then sort newest-first and notify.""" if push.replace: diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index cf4dbde5d..0fd869f55 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -26,6 +26,7 @@ from roborock.map.b01_q10_map_parser import ( B01Q10MapParserConfig, Q10MapPacket, + Q10MapPacketKind, Q10Point, Q10Room, Q10TracePacket, @@ -36,7 +37,6 @@ from .command import CommandTrait from .common import UpdatableTrait -from .maps import MapsTrait _LOGGER = logging.getLogger(__name__) _DOCKED_STATES = {YXDeviceState.CHARGING, YXDeviceState.EMPTYING_THE_BIN} @@ -90,15 +90,13 @@ class MapContentTrait(TraitUpdateListener): """High-level composed Q10 map view. The latest map and trace packets are combined with the injected - :class:`MapDpsTrait` whenever a source changes. The - :class:`MapsTrait` supplies a stored ID only when this trait requests - content. + :class:`MapDpsTrait` whenever a source changes. Current-map acquisition is + independent of the saved-map list. """ def __init__( self, map_dps: MapDpsTrait, - maps: MapsTrait, command: CommandTrait, *, map_parser_config: B01Q10MapParserConfig | None = None, @@ -106,7 +104,6 @@ def __init__( TraitUpdateListener.__init__(self, logger=_LOGGER) self._config = map_parser_config or B01Q10MapParserConfig() self._map_dps = map_dps - self._maps = maps self._command = command self._map_packet: Q10MapPacket | None = None self._trace_packet: Q10TracePacket | None = None @@ -214,6 +211,8 @@ def robot_heading(self) -> int | None: def update_from_map_packet(self, packet: Q10MapPacket) -> None: """Store a map-protocol update and render the latest sources.""" + if packet.kind is not Q10MapPacketKind.CURRENT: + raise ValueError(f"Expected a current Q10 map packet, got {packet.kind.value}") self._map_packet = packet self._render() self._notify_update() diff --git a/roborock/devices/traits/b01/q10/maps.py b/roborock/devices/traits/b01/q10/maps.py index a158c67ef..47c0fa2a0 100644 --- a/roborock/devices/traits/b01/q10/maps.py +++ b/roborock/devices/traits/b01/q10/maps.py @@ -9,6 +9,7 @@ from roborock.data.b01_q10.b01_q10_containers import Q10MapInfo, dpMultiMap from roborock.devices.traits.common import DpsDataConverter from roborock.exceptions import RoborockException +from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10MapPacketKind from .command import CommandTrait from .common import UpdatableTrait @@ -22,6 +23,15 @@ class Maps(RoborockBase): multi_map: dpMultiMap | None = field(default=None, metadata={"dps": B01_Q10_DP.MULTI_MAP}) + detail_packet: Q10MapPacket | None = None + """Most recently received saved-map preview.""" + detail_map_id: str | None = None + """Map ID supplied by the preview packet itself.""" + + def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]: + """Serialize metadata without binary map packets and grids.""" + return super().as_dict(exclude=(exclude or set()) | {"detail_packet"}) + @property def current_map_id(self) -> str | None: """Return the first saved-map ID for a content request, if available.""" @@ -38,7 +48,11 @@ def map_list(self) -> list[Q10MapInfo]: class MapsTrait(Maps, UpdatableTrait): - """Request and store the Q10 saved-map list.""" + """Browse saved maps and retrieve their parsed previews independently of the live map. + + Preview pushes identify their map, but carry no request correlation ID. + Consumers receive them through ``add_update_listener``. + """ _CONVERTER = DpsDataConverter.from_dataclass(Maps) _command: CommandTrait @@ -70,6 +84,24 @@ async def set_current_map(self, map_id: str) -> None: }, ) + async def refresh_detail(self, map_id: str | None = None) -> None: + """Request a read-only preview for one saved map. + + Returns after publishing. Subscribe with ``add_update_listener`` to + receive previews separately from the live map. ``detail_map_id`` is + read from each pushed packet; it is not a request correlation ID. + """ + if map_id is None: + map_id = self.current_map_id + if map_id is None: + raise RoborockException("Cannot request Q10 saved-map detail before the map list is available") + if map_id not in {map_info.id for map_info in self.map_list}: + raise RoborockException(f"Unknown Q10 saved-map ID: {map_id}") + await self._command.send( + B01_Q10_DP.COMMON, + {str(B01_Q10_DP.MULTI_MAP.code): {"op": "select", "id": map_id}}, + ) + def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: """Store a successful saved-map list response.""" response = decoded_dps.get(B01_Q10_DP.MULTI_MAP) @@ -78,3 +110,11 @@ def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: if not isinstance(response, dict) or response.get("op") != "list" or response.get("result") != 1: return super().update_from_dps(decoded_dps) + + def update_from_map_packet(self, packet: Q10MapPacket) -> None: + """Store a pushed saved-map detail packet and notify listeners.""" + if packet.kind is not Q10MapPacketKind.SAVED_MAP_DETAIL: + raise ValueError(f"Expected a Q10 saved-map detail packet, got {packet.kind.value}") + self.detail_map_id = str(packet.map_id) + self.detail_packet = packet + self._notify_update() diff --git a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr index daac7f55b..636079707 100644 --- a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr +++ b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr @@ -4,6 +4,12 @@ 'child_lock': dict({ 'childLock': False, }), + 'clean_history': dict({ + 'detailPath': list([ + ]), + 'records': list([ + ]), + }), 'consumable': dict({ 'filterLife': 0, 'mainBrushLife': 0, @@ -384,6 +390,12 @@ 'child_lock': dict({ 'childLock': False, }), + 'clean_history': dict({ + 'detailPath': list([ + ]), + 'records': list([ + ]), + }), 'consumable': dict({ 'filterLife': 0, 'mainBrushLife': 0, @@ -473,6 +485,12 @@ dict({ 'child_lock': dict({ }), + 'clean_history': dict({ + 'detailPath': list([ + ]), + 'records': list([ + ]), + }), 'consumable': dict({ }), 'do_not_disturb': dict({ diff --git a/tests/devices/traits/b01/q10/test_clean_history.py b/tests/devices/traits/b01/q10/test_clean_history.py index 10fce2b64..b03eed6c7 100644 --- a/tests/devices/traits/b01/q10/test_clean_history.py +++ b/tests/devices/traits/b01/q10/test_clean_history.py @@ -1,3 +1,6 @@ +import json +from pathlib import Path + import pytest from roborock.data.b01_q10.b01_q10_code_mappings import ( @@ -9,7 +12,14 @@ ) from roborock.data.b01_q10.b01_q10_containers import Q10CleanRecord from roborock.devices.traits.b01.q10 import Q10PropertiesApi -from roborock.devices.traits.b01.q10.clean_history import CleanHistoryTrait, CleanRecordConverter +from roborock.devices.traits.b01.q10.clean_history import CleanHistory, CleanHistoryTrait, CleanRecordConverter +from roborock.exceptions import RoborockException +from roborock.map.b01_q10_map_parser import ( + Q10CleanRecordDetail, + Q10HistoricalTracePacket, + Q10Point, + parse_clean_record_detail, +) from .conftest import FakeB01Q10Channel @@ -175,3 +185,70 @@ async def test_refresh_sends_op_list(q10_api: Q10PropertiesApi, fake_channel: Fa B01_Q10_DP.COMMON, {"52": {"op": "list"}}, ) + + +async def test_refresh_detail_sends_op_select( + clean_history: CleanHistoryTrait, + fake_channel: FakeB01Q10Channel, +) -> None: + record = CleanRecordConverter.parse_record(RECORD_A) + assert record is not None + await clean_history.refresh_detail(record) + assert fake_channel.published_commands == [(B01_Q10_DP.COMMON, {"52": {"op": "select", "id": RECORD_A}})] + + +def _detail() -> Q10CleanRecordDetail: + fixture = Path("tests/map/testdata/b01_q10_map.bin").read_bytes() + return parse_clean_record_detail(b"\x03\x01" + fixture[2:]) + + +async def test_refresh_detail_rejects_record_without_map(clean_history: CleanHistoryTrait) -> None: + record = CleanRecordConverter.parse_record("x_1781226271_1_1_0_0_0_0_2_1_1_0") + assert record is not None + with pytest.raises(RoborockException, match="no saved map detail"): + await clean_history.refresh_detail(record) + + +def test_received_detail_is_serializable_and_path_is_independent(q10_api: Q10PropertiesApi) -> None: + history = q10_api.clean_history + history.update_from_dps(_list_push(RECORD_A)) + detail = _detail() + detail.trace = Q10HistoricalTracePacket(points=[Q10Point(x=1, y=2)], heading=0) + history.update_from_detail(detail) + assert history.detail is detail + assert history.detail_packet is detail.map + assert history.detail_trace is detail.trace + detail.trace.points.clear() + assert history.detail_path == [Q10Point(x=1, y=2)] + assert isinstance(history, CleanHistory) + data = q10_api.as_dict()["clean_history"] + assert data["records"][0]["recordId"] == "abc123def456" + assert data["detailPath"] == [{"x": 1, "y": 2}] + assert "detailImageContent" not in data + assert "detail" not in data + json.dumps(data) + assert history.as_dict(exclude={"records", "detail_path"}) == {} + + +def test_received_detail_without_trace_clears_previous_path(clean_history: CleanHistoryTrait) -> None: + first = _detail() + first.trace = Q10HistoricalTracePacket(points=[Q10Point(x=1, y=2)], heading=0) + clean_history.update_from_detail(first) + assert clean_history.detail_path == [Q10Point(x=1, y=2)] + second = _detail() + clean_history.update_from_detail(second) + assert clean_history.detail is second + assert clean_history.detail_trace is None + assert clean_history.detail_path == [] + + +def test_unsolicited_and_repeated_details_notify_listeners(clean_history: CleanHistoryTrait) -> None: + updates = [] + unsubscribe = clean_history.add_update_listener(lambda: updates.append(clean_history.detail)) + first, second = _detail(), _detail() + clean_history.update_from_detail(first) + clean_history.update_from_detail(second) + assert updates == [first, second] + unsubscribe() + clean_history.update_from_detail(first) + assert updates == [first, second] diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 13e5952ed..0e65afa38 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -1,13 +1,15 @@ """Tests for the Q10 B01 map content trait. -Map list data and map content have independent refresh schedules. Content -requests use a stored map ID, and the device sends the data later in a -``MAP_RESPONSE`` packet. These tests cover that state management. The render -details are tested in ``tests/map/test_b01_q10_render.py``. +Map list data and current map content have independent refresh schedules. +Saved-map previews use a stored map ID, while current content uses a read-only +status request. The device sends data later in a ``MAP_RESPONSE`` packet. +These tests cover state management. The render details are tested in ``tests/map/test_b01_q10_render.py``. """ import asyncio import base64 +import io +import json from collections.abc import AsyncGenerator, Generator from dataclasses import replace from pathlib import Path @@ -15,6 +17,7 @@ from unittest.mock import Mock, patch import pytest +from PIL import Image from vacuum_map_parser_base.config.drawable import Drawable from roborock.cli import _await_q10_map_push, cli @@ -23,14 +26,16 @@ from roborock.devices.traits.b01.q10 import Q10PropertiesApi, create from roborock.devices.traits.b01.q10.command import CommandTrait from roborock.devices.traits.b01.q10.map import MapContentTrait, MapDpsTrait -from roborock.devices.traits.b01.q10.maps import MapsTrait from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import ( B01Q10MapParserConfig, + Q10CleanRecordDetail, Q10HeaderCalibration, + Q10MapPacketKind, Q10Obstacle, Q10Point, Q10TracePacket, + parse_clean_record_detail, parse_map_packet, parse_trace_packet, ) @@ -45,19 +50,9 @@ def _map_trait(map_dps: MapDpsTrait | None = None) -> MapContentTrait: - """Create map content with a stored map ID for tests that do not perform I/O.""" + """Create map content for tests that do not perform I/O.""" command = cast(CommandTrait, Mock(spec=CommandTrait)) - maps = MapsTrait(command) - maps.update_from_dps( - { - B01_Q10_DP.MULTI_MAP: { - "data": [{"id": "12345"}], - "op": "list", - "result": 1, - } - } - ) - return MapContentTrait(map_dps or MapDpsTrait(), maps, command) + return MapContentTrait(map_dps or MapDpsTrait(), command) def _zone_blob() -> str: @@ -118,17 +113,7 @@ def test_q10_position_is_available_as_top_level_cli_command() -> None: class _FakeQ10Properties: def __init__(self) -> None: command = cast(CommandTrait, Mock(spec=CommandTrait)) - self.maps = MapsTrait(command) - self.maps.update_from_dps( - { - B01_Q10_DP.MULTI_MAP: { - "data": [{"id": "12345"}], - "op": "list", - "result": 1, - } - } - ) - self.map = MapContentTrait(MapDpsTrait(), self.maps, command) + self.map = MapContentTrait(MapDpsTrait(), command) self.refresh_count = 0 async def refresh_map() -> None: @@ -322,8 +307,10 @@ async def test_map_content_refresh_does_not_require_stored_map_id( assert mock_channel.published_commands == [(B01_Q10_DP.REQUEST_DPS, {})] -async def test_map_content_refresh_requests_are_not_rate_limited(q10_api: Q10PropertiesApi) -> None: - """The caller controls content cadence; each refresh sends a get request.""" +async def test_map_content_refresh_requests_are_not_rate_limited( + q10_api: Q10PropertiesApi, mock_channel: FakeB01Q10Channel +) -> None: + """The caller controls content cadence; each refresh sends a read-only request.""" q10_api.maps.update_from_dps( { B01_Q10_DP.MULTI_MAP: { @@ -333,11 +320,10 @@ async def test_map_content_refresh_requests_are_not_rate_limited(q10_api: Q10Pro } } ) - with patch.object(q10_api.command, "send") as send: - await q10_api.map.refresh() - await q10_api.map.refresh() + await q10_api.map.refresh() + await q10_api.map.refresh() - assert send.await_count == 2 + assert mock_channel.published_commands == [(B01_Q10_DP.REQUEST_DPS, {}), (B01_Q10_DP.REQUEST_DPS, {})] def test_map_get_ack_does_not_replace_saved_map_list(q10_api: Q10PropertiesApi) -> None: @@ -781,3 +767,212 @@ async def test_set_room_name_rejects_unknown_room_before_publish( async def test_clear_restrictions_without_snapshot(q10_api: Q10PropertiesApi, mock_channel: FakeB01Q10Channel) -> None: await q10_api.map.set_restricted_zones([]) assert mock_channel.published_commands == [(B01_Q10_DP.COMMON, {"54": "AQA="})] + + +async def test_archived_map_pushes_cannot_overwrite_live_map( + q10_api: Q10PropertiesApi, + message_queue: asyncio.Queue[Q10Message], +) -> None: + """03/04 detail packets are isolated from the current live-map trait.""" + current_bytes = FIXTURE.read_bytes() + current = parse_map_packet(current_bytes) + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + clean_record = parse_clean_record_detail(b"\x03\x01" + current_bytes[2:]) + saved_map = parse_map_packet(b"\x04\x01" + current_bytes[2:]) + + message_queue.put_nowait(current) + message_queue.put_nowait(trace) + await _wait_for(lambda: q10_api.map.image_content is not None and bool(q10_api.map.path)) + live_image = q10_api.map.image_content + live_rooms = list(q10_api.map.rooms) + live_path = list(q10_api.map.path) + live_position = q10_api.map.robot_position + live_heading = q10_api.map.robot_heading + live_updates: list[None] = [] + clean_record_updates: list[None] = [] + saved_map_updates: list[None] = [] + q10_api.map.add_update_listener(lambda: live_updates.append(None)) + q10_api.clean_history.add_update_listener(lambda: clean_record_updates.append(None)) + q10_api.maps.add_update_listener(lambda: saved_map_updates.append(None)) + + message_queue.put_nowait(clean_record) + message_queue.put_nowait(saved_map) + await _wait_for(lambda: q10_api.maps.detail_packet is not None) + + assert q10_api.map.image_content == live_image + assert q10_api.map.rooms == live_rooms + assert q10_api.map.path == live_path + assert q10_api.map.robot_position == live_position + assert q10_api.map.robot_heading == live_heading + assert live_updates == [] + assert q10_api.clean_history.detail is clean_record + assert q10_api.clean_history.detail_packet is clean_record.map + assert q10_api.clean_history.detail_packet.kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL + assert q10_api.maps.detail_packet is saved_map + assert q10_api.maps.detail_packet.kind is Q10MapPacketKind.SAVED_MAP_DETAIL + assert clean_record_updates == [None] + assert saved_map_updates == [None] + + +def test_archive_owners_reject_wrong_packet_kinds(q10_api: Q10PropertiesApi) -> None: + """Semantic archive traits reject packets owned by another map stream.""" + current = parse_map_packet(FIXTURE.read_bytes()) + + with pytest.raises(ValueError, match="clean-record detail"): + q10_api.clean_history.update_from_detail(Q10CleanRecordDetail(map=current)) + with pytest.raises(ValueError, match="saved-map detail"): + q10_api.maps.update_from_map_packet(current) + + +@pytest.mark.parametrize("scale", [1, 2, 4]) +async def test_live_q10_map_renders_at_requested_scale(scale: int, fake_channel: FakeB01Q10Channel) -> None: + api = create(fake_channel, map_parser_config=B01Q10MapParserConfig(map_scale=scale)) + payload = FIXTURE.read_bytes() + fake_channel.messages_to_stream = [ + parse_map_packet(payload), + ] + await api.start() + try: + await _wait_for(lambda: api.map.image_content is not None) + assert api.map.image_content is not None + with Image.open(io.BytesIO(api.map.image_content)) as image: + assert image.size == (8 * scale, 6 * scale) + finally: + await api.close() + + +def test_q10_properties_constructor_uses_default_live_map_config(fake_channel: FakeB01Q10Channel) -> None: + api = Q10PropertiesApi(fake_channel) + api.map.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) + assert api.map.image_content is not None + with Image.open(io.BytesIO(api.map.image_content)) as image: + assert image.size == (32, 24) + + +@pytest.mark.parametrize("selected,expected", [(None, "12345"), ("67890", "67890")]) +async def test_saved_map_detail_refresh_publishes_selection( + q10_api: Q10PropertiesApi, + mock_channel: FakeB01Q10Channel, + selected: str | None, + expected: str, +) -> None: + q10_api.maps.update_from_dps( + { + B01_Q10_DP.MULTI_MAP: { + "data": [{"id": "12345"}, {"id": "67890"}], + "op": "list", + "result": 1, + } + } + ) + await q10_api.maps.refresh_detail(selected) + assert mock_channel.published_commands == [(B01_Q10_DP.COMMON, {"61": {"op": "select", "id": expected}})] + assert q10_api.maps.detail_packet is None + + +async def test_saved_map_detail_rejects_unknown_id(q10_api: Q10PropertiesApi, mock_channel: FakeB01Q10Channel) -> None: + q10_api.maps.update_from_dps({B01_Q10_DP.MULTI_MAP: {"data": [{"id": "12345"}], "op": "list", "result": 1}}) + with pytest.raises(RoborockException, match="Unknown Q10 saved-map ID"): + await q10_api.maps.refresh_detail("67890") + assert mock_channel.published_commands == [] + + +async def test_saved_map_details_publish_packet_map_id_and_notify( + q10_api: Q10PropertiesApi, + mock_channel: FakeB01Q10Channel, + message_queue: asyncio.Queue[Q10Message], +) -> None: + updates = [] + q10_api.maps.add_update_listener(lambda: updates.append(q10_api.maps.detail_map_id)) + packet = parse_map_packet(b"\x04\x01" + FIXTURE.read_bytes()[2:]) + message_queue.put_nowait(packet) + await _wait_for(lambda: bool(updates)) + assert updates == [str(packet.map_id)] + assert q10_api.maps.detail_packet is packet + message_queue.put_nowait(replace(packet, map_id=999)) + await _wait_for(lambda: len(updates) == 2) + assert updates == [str(packet.map_id), "999"] + assert q10_api.map.image_content is None + + +async def test_saved_map_detail_refresh_requires_stored_map_id(q10_api: Q10PropertiesApi) -> None: + """Detail cannot be requested until the saved-map list supplies an ID.""" + with pytest.raises(RoborockException, match="map list is available"): + await q10_api.maps.refresh_detail() + + +def test_saved_map_diagnostics_exclude_binary_content(q10_api: Q10PropertiesApi) -> None: + packet = parse_map_packet(b"\x04\x01" + FIXTURE.read_bytes()[2:]) + q10_api.maps.update_from_map_packet(packet) + data = q10_api.as_dict()["maps"] + assert data == {"detailMapId": str(packet.map_id)} + assert q10_api.maps.detail_packet is packet + assert q10_api.maps.as_dict(exclude={"detail_map_id"}) == {} + json.dumps(data) + + +async def test_cli_listener_is_removed_after_timeout(q10_api: Q10PropertiesApi) -> None: + calls = [] + + def predicate() -> bool: + calls.append(True) + return False + + assert not await _await_q10_map_push(q10_api, predicate, timeout=0.01) + calls.clear() + q10_api.map.update_from_trace_packet(Q10TracePacket()) + assert calls == [] + + +def test_live_map_trait_rejects_archived_packet() -> None: + trait = _map_trait() + packet = parse_map_packet(b"\x04\x01" + FIXTURE.read_bytes()[2:]) + with pytest.raises(ValueError, match="current Q10 map packet"): + trait.update_from_map_packet(packet) + assert trait.image_content is None + + +async def test_cli_listener_is_removed_after_cancellation(q10_api: Q10PropertiesApi) -> None: + calls = [] + + def predicate() -> bool: + calls.append(True) + return False + + request = asyncio.create_task(_await_q10_map_push(q10_api, predicate, timeout=1)) + await asyncio.sleep(0) + request.cancel() + with pytest.raises(asyncio.CancelledError): + await request + q10_api.map.update_from_trace_packet(Q10TracePacket()) + assert calls == [] + + +@pytest.mark.parametrize("drawables,expected_base", [([], True), ([Drawable.CHARGER], False)]) +async def test_live_q10_map_honors_drawable_selection( + drawables: list[Drawable], expected_base: bool, fake_channel: FakeB01Q10Channel +) -> None: + api = create(fake_channel, map_parser_config=B01Q10MapParserConfig(drawables=drawables)) + base_packet = parse_map_packet(FIXTURE.read_bytes()) + base = create(FakeB01Q10Channel()) + base.map.update_from_map_packet(base_packet) + assert base.map.image_content is not None + with Image.open(io.BytesIO(base.map.image_content)) as image: + base_pixels = image.tobytes() + packet = replace( + base_packet, + header_calibration=Q10HeaderCalibration( + origin_x=0, origin_y=50, resolution=5, charger_x=30, charger_y=30, charger_phi=90 + ), + ) + fake_channel.messages_to_stream = [ + packet, + ] + await api.start() + try: + await _wait_for(lambda: api.map.image_content is not None) + assert api.map.image_content is not None + with Image.open(io.BytesIO(api.map.image_content)) as image: + assert (image.tobytes() == base_pixels) is expected_base + finally: + await api.close()