feat: parse Q10 archived map packets

AGENTS.md

@@ -54,7 +54,7 @@ Evaluate every change against the repository's three core layers in strict prior
 ## Detailed Engineering Conventions
 
 ### 1. Typing & Data Models
-- **Subclass `RoborockBase`**: Define structured domain and wire data models as `@dataclass` subclassing `RoborockBase` (`from_dict`, `as_dict`). Avoid `TypedDict` or loose dicts. (Binary protocol packets, transport message envelopes, and map layers are exempt).
+- **Subclass `RoborockBase`**: Define structured domain and wire data models as `@dataclass` subclassing `RoborockBase` (`from_dict`, `as_dict`). Avoid `TypedDict` or loose dicts. (Binary protocol packets, transport message envelopes, and map layers are exempt). The existing frozen `Q10RoborockPoint` coordinate value is also exempt: it must retain immutability and hashability, and Python disallows frozen dataclass inheritance from the non-frozen `RoborockBase`. This exception does not extend to other domain models.
 - **Enum Fallback Resilience**: All enums representing device status, firmware modes, error codes, and wire protocol integer codes MUST inherit from `RoborockEnum` (defining a lowercase `unknown = -1` or `0` member) or `RoborockModeEnum` (using `from_code_optional()`). Internal enums not decoding unknown firmware codes remain standard `Enum`/`StrEnum`.
 - **Strongly Type What You Know; Contain `Any` to the Wire**: Public trait APIs, method signatures, properties, and domain models MUST declare concrete types. `Any` is accepted only where the underlying wire protocol is dynamic or polymorphic (Tuya DPS maps, low-level RPC dispatch, serialization helpers, evolving cloud schemas).
 - **Avoid Forward References & `TYPE_CHECKING`**: Avoid stringified forward references (`"ClassName"`) and `if typing.TYPE_CHECKING:` guards wherever possible. They typically indicate circular dependencies or coupling that should be refactored by extracting shared models.

roborock/map/b01_q10_render.py

@@ -34,6 +34,7 @@
     B01Q10MapParser,
     B01Q10MapParserConfig,
     Q10EraseZone,
+    Q10HistoricalTracePacket,
     Q10MapPacket,
     Q10TracePacket,
     erased_packet,
@@ -62,7 +63,6 @@
 # a much shorter path suffices to confirm it (early in a clean, not just a dense
 # one). See :func:`solve_calibration_with_origin`.
 _MIN_HEADER_CALIBRATION_POINTS = 4
-
 _Q10_DRAWABLE_TYPES = {
     Drawable.CHARGER,
     Drawable.NO_GO_AREAS,
@@ -84,7 +84,7 @@ class Q10MapOverlays:
 
 def render_q10_map(
     packet: Q10MapPacket,
-    trace: Q10TracePacket | None,
+    trace: Q10TracePacket | Q10HistoricalTracePacket | None,
     overlays: Q10MapOverlays,
     *,
     config: B01Q10MapParserConfig,
@@ -133,7 +133,7 @@ def render_q10_map(
 
 def solve_q10_calibration(
     packet: Q10MapPacket,
-    trace: Q10TracePacket | None,
+    trace: Q10TracePacket | Q10HistoricalTracePacket | None,
 ) -> GridCalibration | None:
     """Derive world-to-pixel calibration from a map and its current trace.
 
@@ -232,7 +232,7 @@ def _erased_cells(
 def _place_trace(
     map_data: MapData,
     calibration: GridCalibration,
-    trace: Q10TracePacket,
+    trace: Q10TracePacket | Q10HistoricalTracePacket,
     *,
     charger_heading: int | None = None,
 ) -> None:

roborock/protocols/b01_q10_protocol.py

@@ -11,10 +11,11 @@
 from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint
 from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import (
+    Q10CleanRecordDetail,
     Q10MapPacket,
+    Q10MapPacketKind,
     Q10TracePacket,
-    is_map_packet,
-    is_trace_packet,
+    parse_clean_record_detail,
     parse_map_packet,
     parse_trace_packet,
 )
@@ -155,24 +156,27 @@ class Q10DpsUpdate:
 # A single decoded message from a Q10 device: a DPS status update, a full map
 # packet, or a live cleaning-path (trace) packet. Map/trace packets arrive as
 # protocol-301 ``MAP_RESPONSE`` pushes; everything else is a DPS update.
-Q10Message = Q10DpsUpdate | Q10MapPacket | Q10TracePacket
+Q10Message = Q10DpsUpdate | Q10MapPacket | Q10TracePacket | Q10CleanRecordDetail
 
 
 def decode_message(message: RoborockMessage) -> Q10Message | None:
     """Decode a pushed Q10 ``RoborockMessage`` into a typed message.
 
-    ``MAP_RESPONSE`` (protocol 301) payloads carry the binary map (``01 01``) or
-    trace (``02 01``) packets, which are parsed by the map parser; any other
-    ``MAP_RESPONSE`` marker is unrecognized and yields ``None``. Every other
-    protocol is treated as a DPS status update.
+    ``MAP_RESPONSE`` (protocol 301) payloads carry binary current-map (``01
+    01``), trace (``02 01``), clean-record detail (``03 01``), or saved-map
+    detail (``04 01``) packets. Any other marker is unrecognized and yields
+    ``None``. Every other protocol is treated as a DPS status update.
 
     Raises ``RoborockException`` if a recognized payload fails to parse.
     """
     if message.protocol == RoborockMessageProtocol.MAP_RESPONSE:
         payload = message.payload or b""
-        if is_map_packet(payload):
-            return parse_map_packet(payload)
-        if is_trace_packet(payload):
+        kind = Q10MapPacketKind.from_payload(payload)
+        if kind is Q10MapPacketKind.TRACE:
             return parse_trace_packet(payload)
+        if kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL:
+            return parse_clean_record_detail(payload)
+        if kind is not None:
+            return parse_map_packet(payload)
         return None
     return Q10DpsUpdate(dps=decode_rpc_response(message))

tests/conformance/test_model_conformance.py

@@ -10,6 +10,7 @@
 import pytest
 
 import roborock.data
+from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint
 from roborock.data.containers import RoborockBase
 from tests.conformance.discovery import discover_dataclasses, to_pytest_params
 
@@ -20,6 +21,12 @@
 )
 def test_data_model_subclasses_roborock_base(model_cls: type) -> None:
     """All domain dataclasses in roborock.data must inherit from RoborockBase."""
+    # This immutable coordinate value predates the conformance suite. Frozen
+    # dataclasses cannot inherit from the non-frozen RoborockBase dataclass.
+    # Keep this exception explicit; other models must satisfy the normal rule.
+    if model_cls is Q10RoborockPoint:
+        assert model_cls.__dataclass_params__.frozen  # type: ignore[attr-defined]
+        return
     assert issubclass(model_cls, RoborockBase), (
         f"{model_cls.__module__}.{model_cls.__name__} is a dataclass but does not inherit from RoborockBase. "
         "Per AGENTS.md, domain containers must subclass RoborockBase for serialization."

tests/data/b01_q10/test_b01_q10_containers.py

@@ -1,5 +1,7 @@
 """Tests for Q10 data containers."""
 
+from dataclasses import FrozenInstanceError
+
 import pytest
 
 from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint
@@ -47,3 +49,15 @@ def test_q10_roborock_point_rejects_invalid_vector_coordinates(
     """Outbound vector coordinates must fit the signed wire grid exactly."""
     with pytest.raises(ValueError):
         point.to_vector()
+
+
+@pytest.mark.parametrize("field", ["x", "y"])
+def test_roborock_point_preserves_immutable_value_contract(field: str) -> None:
+    point = Q10RoborockPoint(25500, 25500)
+    original_hash = hash(point)
+    with pytest.raises(FrozenInstanceError):
+        setattr(point, field, 0)
+    with pytest.raises(FrozenInstanceError):
+        delattr(point, field)
+    assert point == Q10RoborockPoint(25500, 25500)
+    assert hash(point) == original_hash

tests/map/test_b01_q10_map_parser.py

@@ -2,6 +2,7 @@
 
 import io
 from pathlib import Path
+from typing import Any
 
 import pytest
 from PIL import Image
@@ -10,11 +11,14 @@
 from roborock.map.b01_grid_layers import LAYER_BACKGROUND, LAYER_FLOOR, LAYER_WALL
 from roborock.map.b01_q10_map_parser import (
     B01Q10MapParser,
+    Q10MapPacketKind,
+    Q10Point,
     Q10Room,
     classify_q10_cell,
     is_map_packet,
     is_trace_packet,
     lz4_block_decompress,
+    parse_clean_record_detail,
     parse_map_packet,
     parse_trace_packet,
 )
@@ -96,15 +100,22 @@ def test_lz4_block_roundtrip_all_literals() -> None:
     block.append(0x0F << 4)
     block.append(len(original) - 15)
     block += original
-    assert lz4_block_decompress(bytes(block)) == original
+    assert lz4_block_decompress(bytes(block), max_output_size=len(original)) == original
 
 
 def test_lz4_block_back_reference() -> None:
     """Back-references expand runs (e.g. RLE-style repeats)."""
     # seq1: 1 literal 'A', then match (offset 1, length 4+4=8) -> 'A' x9.
     # seq2: final literals-only token (0 literals) ends the block per LZ4 spec.
     block = bytes([0x14, ord("A"), 0x01, 0x00, 0x00])
-    assert lz4_block_decompress(block) == b"A" * 9
+    assert lz4_block_decompress(block, max_output_size=9) == b"A" * 9
+
+
+def test_lz4_block_rejects_output_over_limit() -> None:
+    block = bytes([0x14, ord("A"), 0x01, 0x00, 0x00])
+
+    with pytest.raises(RoborockException, match="maximum output size"):
+        lz4_block_decompress(block, max_output_size=8)
 
 
 def test_is_map_packet() -> None:
@@ -356,13 +367,32 @@ def test_parse_trace_rejects_misaligned_points() -> None:
         parse_trace_packet(b"\x02\x01" + b"\x00" * 12 + b"\x01\x02\x03")
 
 
+@pytest.mark.parametrize("declared_count", [0, 2])
+def test_parse_trace_rejects_declared_count_mismatch(declared_count: int) -> None:
+    """An aligned body cannot silently disagree with the firmware count."""
+    payload = bytearray(_trace_payload([(10, 20)]))
+    payload[8:10] = declared_count.to_bytes(2, "big")
+
+    with pytest.raises(RoborockException, match="point count"):
+        parse_trace_packet(bytes(payload))
+
+
 def test_parse_rejects_bad_layout_length() -> None:
     payload = bytearray(_payload())
     payload[27:29] = (0xFFFF).to_bytes(2, "big")  # compressed length past the buffer
     with pytest.raises(RoborockException, match="invalid layout block length"):
         parse_map_packet(bytes(payload))
 
 
+def test_parse_rejects_unreasonable_header_dimensions() -> None:
+    payload = bytearray(_payload())
+    payload[7:9] = (65535).to_bytes(2, "big")
+    payload[9:11] = (65535).to_bytes(2, "big")
+
+    with pytest.raises(RoborockException, match="supported grid size"):
+        parse_map_packet(bytes(payload))
+
+
 def test_parse_erase_zones_from_map_packet_tail() -> None:
     """Erase rectangles appended after the grid decode to world polygons."""
     rects = [
@@ -387,6 +417,32 @@ def _carpet_tail(width: int, height: int, carpet: bytes, erase: bytes = bytes([0
     return erase + (width * height).to_bytes(4, "big") + len(block).to_bytes(2, "big") + block
 
 
+def _map_detail_payload(
+    marker: bytes,
+    points: list[tuple[int, int]],
+    *,
+    version: int = 1,
+    opaque_value: int = 2,
+    heading: int = 3,
+    reserved: int = 0,
+    prefix: int = 0,
+    trailing: bytes = b"",
+) -> bytes:
+    """Build a neutral synthetic detail packet from the existing map fixture."""
+    header = (
+        version.to_bytes(2, "big")
+        + opaque_value.to_bytes(4, "big")
+        + len(points).to_bytes(4, "big")
+        + heading.to_bytes(2, "big", signed=True)
+        + reserved.to_bytes(2, "big")
+    )
+    point_table = b"".join(x.to_bytes(2, "big", signed=True) + y.to_bytes(2, "big", signed=True) for x, y in points)
+    history = bytes([prefix]) + header + point_table
+    payload = bytearray(FIXTURE.read_bytes() + _carpet_tail(8, 6, bytes(48)) + history + trailing)
+    payload[:2] = marker
+    return bytes(payload)
+
+
 def test_parse_carpet_mask_from_map_packet_tail() -> None:
     """A carpet mask after the erase section decodes to a same-dims grid.
 
@@ -415,6 +471,82 @@ def test_parse_map_packet_without_carpet() -> None:
     assert parse_map_packet(FIXTURE.read_bytes()).carpet_mask is None
 
 
+def test_classify_current_clean_record_and_saved_map_packets() -> None:
+    """All known map markers retain an explicit semantic kind."""
+    current = FIXTURE.read_bytes()
+    clean_record = _map_detail_payload(b"\x03\x01", [(10, -20)])
+    saved_map = _map_detail_payload(b"\x04\x01", [(10, -20)])
+
+    assert is_map_packet(current)
+    assert parse_map_packet(current).kind is Q10MapPacketKind.CURRENT
+    assert parse_map_packet(clean_record).kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL
+    assert parse_map_packet(saved_map).kind is Q10MapPacketKind.SAVED_MAP_DETAIL
+
+
+def test_parse_clean_record_historical_trace_with_unknown_tail() -> None:
+    """The bounded historical path is decoded without interpreting later bytes."""
+    packet = parse_clean_record_detail(
+        _map_detail_payload(b"\x03\x01", [(10, -20), (-30, 40)], trailing=b"future-section")
+    )
+
+    assert packet.trace is not None
+    assert [(point.x, point.y) for point in packet.trace.points] == [(10, -20), (-30, 40)]
+    assert packet.trace.heading == 3
+    assert packet.trace.robot_position == Q10Point(-30, 40)
+
+
+def test_zero_point_historical_trace_with_following_section() -> None:
+    """A zero-point path remains valid when a later section follows it."""
+    packet = parse_clean_record_detail(_map_detail_payload(b"\x03\x01", [], trailing=b"recorded-path"))
+
+    assert packet.trace is not None
+    assert packet.trace.points == []
+
+
+def test_historical_trace_is_not_inferred_for_other_packet_kinds() -> None:
+    """The validated ``03 01`` layout is not assumed for current or saved maps."""
+    current = parse_map_packet(_map_detail_payload(b"\x01\x01", [(10, -20)]))
+    saved_map = parse_map_packet(_map_detail_payload(b"\x04\x01", [(10, -20)]))
+
+    assert current.kind is Q10MapPacketKind.CURRENT
+    assert saved_map.kind is Q10MapPacketKind.SAVED_MAP_DETAIL
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"version": 2},
+        {"reserved": 1},
+        {"prefix": 1},
+    ],
+)
+def test_unsupported_historical_trace_header_is_ignored(kwargs: dict[str, Any]) -> None:
+    payload = _map_detail_payload(b"\x03\x01", [(10, -20)], **kwargs)
+    packet = parse_clean_record_detail(payload)
+
+    assert packet.trace is None
+
+
+def test_truncated_historical_trace_is_ignored() -> None:
+    payload = _map_detail_payload(b"\x03\x01", [(10, -20)])[:-2]
+    packet = parse_clean_record_detail(payload)
+
+    assert packet.trace is None
+
+
+def test_invalid_erase_section_is_ignored() -> None:
+    """An invalid erase header cannot become an anchor for later sections."""
+    tail = b"\x01\xffopaque-tail"
+    payload = bytearray(FIXTURE.read_bytes() + tail)
+    payload[:2] = b"\x03\x01"
+
+    packet = parse_clean_record_detail(bytes(payload))
+
+    assert packet.map.erase_zones == []
+    assert packet.map.carpet_mask is None
+    assert packet.trace is None
+
+
 def test_carpet_mask_ignored_when_uncompressed_len_mismatches() -> None:
     """If the section doesn't line up (uncompressed_len != w*h) carpet is dropped."""
     carpet = bytes([4] * 48)
@@ -473,3 +605,36 @@ def test_real_fixture_header_calibration_is_keepalive() -> None:
     """The synthetic fixture carries no header origin, so callers fall back to a fit."""
     cal = parse_map_packet(FIXTURE.read_bytes()).header_calibration
     assert cal is not None and cal.is_keepalive
+
+
+@pytest.mark.parametrize("payload", [b"", b"\x01", b"\x00\x01", b"\xff\x01", b"\x01\x02"])
+def test_unknown_map_markers_are_not_current_maps(payload: bytes) -> None:
+    assert Q10MapPacketKind.from_payload(payload) is None
+    with pytest.raises(RoborockException):
+        parse_map_packet(payload)
+
+
+def test_map_kind_unknown_fallback() -> None:
+    assert Q10MapPacketKind(255) is Q10MapPacketKind.unknown
+    assert Q10MapPacketKind.unknown.marker == b""
+
+
+@pytest.mark.parametrize(
+    "points, expected",
+    [
+        ([(0, 0), (1000, 1000), (1001, 1001), (1002, 1002)], [(1000, 1000), (1001, 1001), (1002, 1002)]),
+        ([(0, 0), (1, 1), (2, 2)], [(0, 0), (1, 1), (2, 2)]),
+        ([(0, 0), (1000, 1000)], [(0, 0), (1000, 1000)]),
+    ],
+)
+def test_clean_record_path_stray_point_handling(points: list[tuple[int, int]], expected: list[tuple[int, int]]) -> None:
+    detail = parse_clean_record_detail(_map_detail_payload(b"\x03\x01", points))
+    assert detail.trace is not None
+    assert [(point.x, point.y) for point in detail.trace.points] == expected
+    assert detail.map.grid
+
+
+@pytest.mark.parametrize("marker", [b"\x01\x01", b"\x02\x01", b"\x04\x01"])
+def test_clean_record_parser_rejects_other_packet_kinds(marker: bytes) -> None:
+    with pytest.raises(RoborockException, match="not a Q10 clean-record"):
+        parse_clean_record_detail(marker + FIXTURE.read_bytes()[2:])