feat: add Q10 zone, position, and goto support

roborock/data/b01_q10/b01_q10_containers.py

@@ -28,6 +28,62 @@
     YXWaterLevel,
 )
 
+_ROBOROCK_COORDINATE_OFFSET_MM = 25_500
+_Q10_TRACE_UNIT_MM = 2.5
+_Q10_VECTOR_UNIT_MM = 5
+
+
+@dataclass(frozen=True)
+class Q10RoborockPoint:
+    """A point in the common Roborock millimetre coordinate space.
+
+    Q10 trace and vector coordinates are firmware details. Public Q10 APIs use
+    this coordinate system, matching other Roborock devices and placing the dock
+    at ``(25500, 25500)``.
+    """
+
+    x: int
+    y: int
+
+    @classmethod
+    def from_trace(cls, x: int, y: int) -> "Q10RoborockPoint":
+        """Convert Q10 trace coordinates to common Roborock coordinates."""
+        for value in (x, y):
+            if isinstance(value, bool) or not isinstance(value, int):
+                raise ValueError("trace coordinates must be integers")
+        return cls(
+            x=round(_ROBOROCK_COORDINATE_OFFSET_MM + x * _Q10_TRACE_UNIT_MM),
+            y=round(_ROBOROCK_COORDINATE_OFFSET_MM + y * _Q10_TRACE_UNIT_MM),
+        )
+
+    @classmethod
+    def from_vector(cls, x: int, y: int) -> "Q10RoborockPoint":
+        """Convert Q10 vector coordinates to common Roborock coordinates."""
+        for value in (x, y):
+            if isinstance(value, bool) or not isinstance(value, int):
+                raise ValueError("vector coordinates must be integers")
+            if not -(2**15) <= value < 2**15:
+                raise ValueError("vector coordinates are outside the Q10 map range")
+        return cls(
+            x=_ROBOROCK_COORDINATE_OFFSET_MM + x * _Q10_VECTOR_UNIT_MM,
+            y=_ROBOROCK_COORDINATE_OFFSET_MM + y * _Q10_VECTOR_UNIT_MM,
+        )
+
+    def to_vector(self) -> tuple[int, int]:
+        """Convert common Roborock coordinates to the Q10 vector grid."""
+        coordinates: list[int] = []
+        for value in (self.x, self.y):
+            if isinstance(value, bool) or not isinstance(value, int):
+                raise ValueError("coordinates must be integers")
+            relative_mm = value - _ROBOROCK_COORDINATE_OFFSET_MM
+            if relative_mm % _Q10_VECTOR_UNIT_MM:
+                raise ValueError("coordinates must align to the Q10 5 mm grid")
+            coordinate = relative_mm // _Q10_VECTOR_UNIT_MM
+            if not -(2**15) <= coordinate < 2**15:
+                raise ValueError("coordinates are outside the Q10 map range")
+            coordinates.append(coordinate)
+        return coordinates[0], coordinates[1]
+
 
 @dataclass
 class dpCleanRecord(RoborockBase):

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

@@ -96,7 +96,6 @@ def __init__(self, channel: B01Q10Channel) -> None:
         """Initialize the B01Props API."""
         self._channel = channel
         self.command = CommandTrait(channel)
-        self.vacuum = VacuumTrait(self.command)
         self.remote = RemoteTrait(self.command)
         self.status = StatusTrait()
         self.volume = SoundVolumeTrait(self.command)
@@ -109,6 +108,7 @@ def __init__(self, channel: B01Q10Channel) -> None:
         self._map_dps = MapDpsTrait()
         self.maps = MapsTrait(self.command)
         self.map = MapContentTrait(self._map_dps, self.maps, self.command)
+        self.vacuum = VacuumTrait(self.command, self.status, self.map)
         self.clean_history = CleanHistoryTrait(self.command)
         # Read-model traits updated from the device's DPS push stream.
         self._updatable_traits = [
@@ -131,6 +131,7 @@ async def start(self) -> None:
 
     async def close(self) -> None:
         """Close any resources held by the trait."""
+        await self.vacuum.close()
         if self._subscribe_task is not None:
             self._subscribe_task.cancel()
             try:

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

@@ -0,0 +1,158 @@
+"""State management for an emulated Q10 goto action."""
+
+import logging
+from collections.abc import Callable
+from dataclasses import dataclass
+from enum import StrEnum
+from math import hypot
+
+from roborock.callbacks import CallbackList
+from roborock.data.b01_q10.b01_q10_code_mappings import YXDeviceCleanTask, YXDeviceState
+from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint
+
+_LOGGER = logging.getLogger(__name__)
+_TERMINAL_STATES = {
+    YXDeviceState.IDLE,
+    YXDeviceState.PAUSED,
+    YXDeviceState.RETURNING_HOME,
+    YXDeviceState.CHARGING,
+}
+
+
+class GotoActionCommand(StrEnum):
+    """A command requested by a Q10 goto action."""
+
+    PAUSE = "pause"
+    STOP = "stop"
+    COMPLETE = "complete"
+
+
+@dataclass(frozen=True)
+class GotoSnapshot:
+    """Device state needed to advance a goto action."""
+
+    position: Q10RoborockPoint | None
+    trace_sequence: int | None
+    clean_task_type: YXDeviceCleanTask | None
+    status: YXDeviceState | None
+
+
+class GotoAction:
+    """Decide how one emulated goto should react to device updates.
+
+    The action owns no tasks and sends no device commands. ``VacuumTrait`` feeds
+    it push-derived snapshots and performs commands requested by its callbacks.
+    """
+
+    def __init__(
+        self,
+        target: Q10RoborockPoint,
+        previous_trace_sequence: int | None,
+        *,
+        tolerance: int,
+    ) -> None:
+        """Initialize a goto action waiting for a new trace session."""
+        self._target = target
+        self._previous_trace_sequence = previous_trace_sequence
+        self._tolerance = tolerance
+        self._owned_trace_sequence: int | None = None
+        self._owned_task_seen = False
+        self._command_pending = False
+        self._timeout_requested = False
+        self._finished = False
+        self._latest_snapshot: GotoSnapshot | None = None
+        self._callbacks: CallbackList[GotoActionCommand] = CallbackList(logger=_LOGGER)
+
+    def add_update_listener(self, callback: Callable[[GotoActionCommand], None]) -> Callable[[], None]:
+        """Register a callback for the next command requested by the action."""
+        return self._callbacks.add_callback(callback)
+
+    def update(self, snapshot: GotoSnapshot) -> None:
+        """Process the latest push-derived device state."""
+        self._latest_snapshot = snapshot
+        self._evaluate(snapshot)
+
+    def retry(self) -> None:
+        """Re-evaluate the latest state after a requested command failed."""
+        if self._finished or self._latest_snapshot is None:
+            return
+        self._command_pending = False
+        if self._timeout_requested:
+            self._evaluate_timeout(self._latest_snapshot)
+        else:
+            self._evaluate(self._latest_snapshot)
+
+    def timeout(self, snapshot: GotoSnapshot) -> None:
+        """Request a stop only if this action still owns the current zone task."""
+        if self._finished:
+            return
+        self._latest_snapshot = snapshot
+        self._timeout_requested = True
+        if self._command_pending:
+            return
+        self._evaluate_timeout(snapshot)
+
+    def _evaluate_timeout(self, snapshot: GotoSnapshot) -> None:
+        """Derive the safe timeout command from the latest device state."""
+        if self.owns(snapshot):
+            self._emit(GotoActionCommand.STOP)
+        else:
+            self._emit(GotoActionCommand.COMPLETE)
+
+    def complete(self) -> None:
+        """Mark the action complete after its requested command succeeds."""
+        self._finished = True
+        self._command_pending = False
+
+    def owns(self, snapshot: GotoSnapshot) -> bool:
+        """Return whether this action owns the current Q10 zone-clean session."""
+        return (
+            self._owned_trace_sequence is not None
+            and snapshot.trace_sequence == self._owned_trace_sequence
+            and snapshot.clean_task_type is YXDeviceCleanTask.DIVIDE_AREAS
+        )
+
+    def _evaluate(self, snapshot: GotoSnapshot) -> None:
+        """Derive the next command from the latest device state."""
+        if self._finished or self._command_pending:
+            return
+
+        if self._owned_trace_sequence is None:
+            if snapshot.trace_sequence is not None and snapshot.trace_sequence != self._previous_trace_sequence:
+                self._owned_trace_sequence = snapshot.trace_sequence
+        elif snapshot.trace_sequence != self._owned_trace_sequence:
+            _LOGGER.debug("Q10 goto task was replaced by another cleaning session")
+            self._emit(GotoActionCommand.COMPLETE)
+            return
+
+        if (
+            self._owned_trace_sequence is not None
+            and snapshot.clean_task_type is YXDeviceCleanTask.DIVIDE_AREAS
+            and snapshot.status not in _TERMINAL_STATES
+        ):
+            self._owned_task_seen = True
+
+        if self._owned_task_seen and (
+            snapshot.clean_task_type is not YXDeviceCleanTask.DIVIDE_AREAS or snapshot.status in _TERMINAL_STATES
+        ):
+            self._emit(GotoActionCommand.COMPLETE)
+            return
+
+        if (
+            self.owns(snapshot)
+            and snapshot.position is not None
+            and hypot(
+                snapshot.position.x - self._target.x,
+                snapshot.position.y - self._target.y,
+            )
+            <= self._tolerance
+        ):
+            self._emit(GotoActionCommand.PAUSE)
+
+    def _emit(self, command: GotoActionCommand) -> None:
+        """Publish a requested command once until it is handled."""
+        if command is GotoActionCommand.COMPLETE:
+            self._finished = True
+        else:
+            self._command_pending = True
+        self._callbacks(command)

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

@@ -19,6 +19,7 @@
 
 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_containers import Q10RoborockPoint
 from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
 from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import (
@@ -127,13 +128,20 @@ def rooms(self) -> list[Q10Room]:
 
     @property
     def path(self) -> list[Q10Point]:
-        """Full path for live status and callers drawing their own map overlay."""
+        """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 robot_position(self) -> Q10Point | None:
-        """Current position for live status and caller-rendered map overlays."""
-        return self._trace_packet.robot_position if self._trace_packet else None
+    def robot_position(self) -> Q10RoborockPoint | None:
+        """Current position in the common Roborock millimetre coordinate space."""
+        if self._trace_packet is None or (position := self._trace_packet.robot_position) is None:
+            return None
+        return position.to_roborock()
+
+    @property
+    def trace_sequence(self) -> int | None:
+        """Current cleaning-session sequence from the trace stream."""
+        return self._trace_packet.sequence if self._trace_packet else None
 
     @property
     def robot_heading(self) -> int | None:
@@ -181,7 +189,9 @@ def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:
         data = {
             "rooms": [room.as_dict() for room in self.rooms],
             "path": [point.as_dict() for point in self.path],
-            "robotPosition": self.robot_position.as_dict() if self.robot_position is not None else None,
+            "robotPosition": (
+                {"x": position.x, "y": position.y} if (position := self.robot_position) is not None else None
+            ),
             "robotHeading": self.robot_heading,
         }
         for key in exclude_set: