feat: add Q10 map management controls

roborock/data/b01_q10/b01_q10_code_mappings.py

@@ -1,6 +1,6 @@
 from enum import IntEnum
 
-from ..code_mappings import RoborockModeEnum
+from ..code_mappings import RoborockEnum, RoborockModeEnum
 
 
 class B01_Q10_DP(RoborockModeEnum):
@@ -354,3 +354,12 @@ class RemoteCommand(IntEnum):
     RIGHT = 3
     STOP = 4
     EXIT = 5
+
+
+class Q10RestrictionType(RoborockEnum):
+    """Restriction types supported by the Q10 map editor."""
+
+    unknown = -1
+    NO_GO = 0
+    NO_MOP = 2
+    THRESHOLD = 3

roborock/devices/traits/b01/q10/map.py

@@ -14,11 +14,12 @@
 """
 
 import logging
+from collections.abc import Sequence
 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, YXDeviceState
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, Q10RestrictionType, YXDeviceState
 from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint
 from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
 from roborock.exceptions import RoborockException
@@ -29,8 +30,9 @@
     Q10Room,
     Q10TracePacket,
 )
-from roborock.map.b01_q10_overlays import parse_virtual_wall_blob, parse_zone_blob
+from roborock.map.b01_q10_overlays import Q10RestrictedZone, Q10VirtualWall, parse_virtual_wall_blob, parse_zone_blob
 from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map
+from roborock.protocols.b01_q10_protocol import encode_restricted_zones, encode_room_name, encode_virtual_walls
 
 from .command import CommandTrait
 from .common import UpdatableTrait
@@ -40,6 +42,10 @@
 _DOCKED_STATES = {YXDeviceState.CHARGING, YXDeviceState.EMPTYING_THE_BIN}
 
 
+def _to_roborock_point(point: tuple[int, int]) -> Q10RoborockPoint:
+    return Q10RoborockPoint.from_vector(*point)
+
+
 @dataclass
 class MapDps(RoborockBase):
     """Low-level map values delivered in the Q10 DPS stream."""
@@ -131,6 +137,64 @@ def path(self) -> list[Q10Point]:
         """Full path in the Q10 trace coordinate space used by the map renderer."""
         return self._trace_packet.points if self._trace_packet else []
 
+    @property
+    def restricted_zones(self) -> tuple[Q10RestrictedZone, ...]:
+        """Restrictions reported by the device in common Roborock coordinates."""
+        restrictions: list[Q10RestrictedZone] = []
+        for zone in self._map_dps.overlays.zones:
+            if len(zone.vertices) != 4:
+                continue
+            vertices = (
+                _to_roborock_point(zone.vertices[0]),
+                _to_roborock_point(zone.vertices[1]),
+                _to_roborock_point(zone.vertices[2]),
+                _to_roborock_point(zone.vertices[3]),
+            )
+            restriction_type = Q10RestrictionType(zone.type)
+            restrictions.append(Q10RestrictedZone(restriction_type, vertices))
+        return tuple(restrictions)
+
+    @property
+    def virtual_walls(self) -> tuple[Q10VirtualWall, ...]:
+        """Virtual walls reported by the device in common Roborock coordinates."""
+        walls: list[Q10VirtualWall] = []
+        for wall in self._map_dps.overlays.virtual_walls:
+            if len(wall.vertices) != 2:
+                continue
+            walls.append(
+                Q10VirtualWall(
+                    start=_to_roborock_point(wall.vertices[0]),
+                    end=_to_roborock_point(wall.vertices[1]),
+                )
+            )
+        return tuple(walls)
+
+    async def set_restricted_zones(self, zones: Sequence[Q10RestrictedZone]) -> None:
+        """Replace the complete restricted-zone collection."""
+        payload = encode_restricted_zones(zones)
+        await self._command.send(
+            B01_Q10_DP.COMMON,
+            {str(B01_Q10_DP.RESTRICTED_ZONE.code): payload},
+        )
+
+    async def set_virtual_walls(self, walls: Sequence[Q10VirtualWall]) -> None:
+        """Replace the complete virtual-wall collection."""
+        payload = encode_virtual_walls(walls)
+        await self._command.send(
+            B01_Q10_DP.COMMON,
+            {str(B01_Q10_DP.VIRTUAL_WALL.code): payload},
+        )
+
+    async def set_room_name(self, room_id: int, name: str) -> None:
+        """Set the name of a room in the current map."""
+        if room_id not in {room.id for room in self.rooms}:
+            raise RoborockException(f"Unknown Q10 room ID: {room_id}")
+        payload = encode_room_name(room_id, name)
+        await self._command.send(
+            B01_Q10_DP.COMMON,
+            {str(B01_Q10_DP.RESET_ROOM_NAME.code): payload},
+        )
+
     @property
     def robot_position(self) -> Q10RoborockPoint | None:
         """Current position in the common Roborock millimetre coordinate space."""

roborock/devices/traits/b01/q10/maps.py

@@ -6,8 +6,9 @@
 
 from roborock.data import RoborockBase
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
-from roborock.data.b01_q10.b01_q10_containers import dpMultiMap
+from roborock.data.b01_q10.b01_q10_containers import Q10MapInfo, dpMultiMap
 from roborock.devices.traits.common import DpsDataConverter
+from roborock.exceptions import RoborockException
 
 from .command import CommandTrait
 from .common import UpdatableTrait
@@ -28,6 +29,13 @@ def current_map_id(self) -> str | None:
             return None
         return self.multi_map.current_map_id
 
+    @property
+    def map_list(self) -> list[Q10MapInfo]:
+        """Return a copy of the successfully reported saved-map list."""
+        if self.multi_map is None or self.multi_map.op != "list" or self.multi_map.result != 1:
+            return []
+        return list(self.multi_map.data)
+
 
 class MapsTrait(Maps, UpdatableTrait):
     """Request and store the Q10 saved-map list."""
@@ -48,6 +56,20 @@ async def refresh(self) -> None:
             {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}},
         )
 
+    async def set_current_map(self, map_id: str) -> None:
+        """Request a saved map without optimistically changing cached state."""
+        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": "apply",
+                    "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)

roborock/map/b01_q10_overlays.py

@@ -31,6 +31,10 @@
 import binascii
 from dataclasses import dataclass, field
 
+from roborock.data import RoborockBase
+from roborock.data.b01_q10.b01_q10_code_mappings import Q10RestrictionType
+from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint
+
 _DEFAULT_RECORD_SIZE = 38  # 2-byte record header + up to 9 (x, y) int16 pairs
 
 
@@ -42,6 +46,22 @@ class Q10Zone:
     vertices: list[tuple[int, int]] = field(default_factory=list)
 
 
+@dataclass
+class Q10RestrictedZone(RoborockBase):
+    """A four-point restriction in common Roborock coordinates."""
+
+    type: Q10RestrictionType
+    vertices: tuple[Q10RoborockPoint, Q10RoborockPoint, Q10RoborockPoint, Q10RoborockPoint]
+
+
+@dataclass
+class Q10VirtualWall(RoborockBase):
+    """A virtual wall in common Roborock coordinates."""
+
+    start: Q10RoborockPoint
+    end: Q10RoborockPoint
+
+
 def _decode_blob(data: str | None) -> bytes:
     if data is None:
         return b""
@@ -55,8 +75,8 @@ def parse_zone_blob(data: str | None) -> list[Q10Zone]:
     """Decode a Q10 zone/wall overlay blob into a list of :class:`Q10Zone`.
 
     Accepts the base64 string straight from the data point. Returns ``[]`` for
-    empty/absent/unparsable blobs (the device sends a single ``0x00`` byte when
-    there are none).
+    empty/absent/unparsable blobs. Empty DP 55 reports contain version ``1``
+    and count ``0``; some firmware appends one padding byte.
     """
     raw = _decode_blob(data)
     if len(raw) < 2:
@@ -156,7 +176,7 @@ def parse_virtual_wall_blob(data: str | None) -> list[Q10Zone]:
 # Corrected from an earlier reading that treated type 3 as no-mop -- 3 is the
 # door-threshold rectangle; the no-mop area reads back as type 2. Reported and
 # verified by @andrewlyeats (ss07 read-backs + the ioBroker Q10 parser).
-ZONE_TYPE_NO_GO = 0
+ZONE_TYPE_NO_GO = Q10RestrictionType.NO_GO.value
 ZONE_TYPE_VIRTUAL_WALL = 1
-ZONE_TYPE_NO_MOP = 2
-ZONE_TYPE_THRESHOLD = 3
+ZONE_TYPE_NO_MOP = Q10RestrictionType.NO_MOP.value
+ZONE_TYPE_THRESHOLD = Q10RestrictionType.THRESHOLD.value

roborock/protocols/b01_q10_protocol.py

@@ -1,13 +1,16 @@
 """Roborock B01 Protocol encoding and decoding."""
 
+import base64
 import json
 import logging
+import struct
 from base64 import b64encode
+from collections.abc import Sequence
 from dataclasses import dataclass
 from struct import pack
 from typing import Any
 
-from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, Q10RestrictionType
 from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint
 from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import (
@@ -19,6 +22,7 @@
     parse_map_packet,
     parse_trace_packet,
 )
+from roborock.map.b01_q10_overlays import Q10RestrictedZone, Q10VirtualWall
 from roborock.roborock_message import (
     RoborockMessage,
     RoborockMessageProtocol,
@@ -75,6 +79,118 @@ def encode_clean_params(params: CleanParams) -> str:
     return b64encode(payload).decode()
 
 
+_Q10_ZONE_RECORD_SIZE = 38
+_Q10_ROOM_NAME_FIELD_SIZE = 19
+
+
+def _orientation(a: tuple[int, int], b: tuple[int, int], c: tuple[int, int]) -> int:
+    return (b[0] - a[0]) * (c[1] - a[1]) - (b[1] - a[1]) * (c[0] - a[0])
+
+
+def _on_segment(a: tuple[int, int], b: tuple[int, int], point: tuple[int, int]) -> bool:
+    return min(a[0], b[0]) <= point[0] <= max(a[0], b[0]) and min(a[1], b[1]) <= point[1] <= max(a[1], b[1])
+
+
+def _segments_cross(a: tuple[int, int], b: tuple[int, int], c: tuple[int, int], d: tuple[int, int]) -> bool:
+    orientations = (
+        _orientation(a, b, c),
+        _orientation(a, b, d),
+        _orientation(c, d, a),
+        _orientation(c, d, b),
+    )
+    if orientations[0] * orientations[1] < 0 and orientations[2] * orientations[3] < 0:
+        return True
+    return any(
+        orientation == 0 and _on_segment(start, end, point)
+        for orientation, start, end, point in (
+            (orientations[0], a, b, c),
+            (orientations[1], a, b, d),
+            (orientations[2], c, d, a),
+            (orientations[3], c, d, b),
+        )
+    )
+
+
+def _validate_quadrilateral(vertices: list[tuple[int, int]]) -> None:
+    area_twice = sum(
+        x1 * y2 - x2 * y1 for (x1, y1), (x2, y2) in zip(vertices, (*vertices[1:], vertices[0]), strict=True)
+    )
+    if area_twice == 0:
+        raise ValueError("restricted-zone vertices must enclose an area")
+    if _segments_cross(vertices[0], vertices[1], vertices[2], vertices[3]) or _segments_cross(
+        vertices[1], vertices[2], vertices[3], vertices[0]
+    ):
+        raise ValueError("restricted-zone edges must not cross")
+
+
+def encode_restricted_zones(zones: Sequence[Q10RestrictedZone]) -> str:
+    """Encode the complete Q10 restricted-zone collection."""
+    if len(zones) > 255:
+        raise ValueError("zones must be a sequence of at most 255 restrictions")
+    if not zones:
+        return base64.b64encode(b"\x01\x00").decode()
+
+    payload = bytearray((1, len(zones)))
+    for zone in zones:
+        if zone.type is Q10RestrictionType.unknown:
+            raise ValueError("unknown restricted-zone types cannot be written")
+        if len(zone.vertices) != 4:
+            raise ValueError("restricted zones must contain exactly four vertices")
+        if len(set(zone.vertices)) != 4:
+            raise ValueError("restricted-zone vertices must be distinct")
+        record = bytearray((zone.type, len(zone.vertices)))
+        vector_vertices = [point.to_vector() for point in zone.vertices]
+        _validate_quadrilateral(vector_vertices)
+        for x, y in vector_vertices:
+            record.extend(
+                struct.pack(
+                    ">hh",
+                    x,
+                    y,
+                )
+            )
+        record.extend(bytes(_Q10_ZONE_RECORD_SIZE - len(record)))
+        payload.extend(record)
+    return base64.b64encode(payload).decode()
+
+
+def encode_virtual_walls(walls: Sequence[Q10VirtualWall]) -> str:
+    """Encode the complete Q10 virtual-wall collection."""
+    if len(walls) > 255:
+        raise ValueError("walls must be a sequence of at most 255 virtual walls")
+
+    payload = bytearray((len(walls),))
+    for wall in walls:
+        if wall.start == wall.end:
+            raise ValueError("virtual-wall endpoints must be distinct")
+        for point in (wall.start, wall.end):
+            x, y = point.to_vector()
+            payload.extend(
+                struct.pack(
+                    ">hh",
+                    x,
+                    y,
+                )
+            )
+    return base64.b64encode(payload).decode()
+
+
+def encode_room_name(room_id: int, name: str) -> str:
+    """Encode a Q10 room-name update."""
+    if not 0 <= room_id <= 255:
+        raise ValueError("room_id must be an integer between 0 and 255")
+    if not name or "\x00" in name:
+        raise ValueError("name must be a non-empty string without null bytes")
+    encoded_name = name.encode("utf-8")
+    if len(encoded_name) > _Q10_ROOM_NAME_FIELD_SIZE:
+        raise ValueError("name must be at most 19 UTF-8 bytes")
+
+    payload = bytearray((1, room_id, 0, len(encoded_name)))
+    payload.extend(encoded_name)
+    payload.extend(bytes(_Q10_ROOM_NAME_FIELD_SIZE - len(encoded_name)))
+    return base64.b64encode(payload).decode()
+
+
 def encode_mqtt_payload(command: B01_Q10_DP, params: ParamsType) -> RoborockMessage:
     """Encode payload for B01 Q10 commands over MQTT.