diff --git a/roborock/data/b01_q10/b01_q10_containers.py b/roborock/data/b01_q10/b01_q10_containers.py index 00384e09..8550bd7c 100644 --- a/roborock/data/b01_q10/b01_q10_containers.py +++ b/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): diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index c2b66f71..5416b7d7 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/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: diff --git a/roborock/devices/traits/b01/q10/goto.py b/roborock/devices/traits/b01/q10/goto.py new file mode 100644 index 00000000..d2c11294 --- /dev/null +++ b/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) diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index 5890e688..a33c2df2 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/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: diff --git a/roborock/devices/traits/b01/q10/vacuum.py b/roborock/devices/traits/b01/q10/vacuum.py index 2747e024..4b846576 100644 --- a/roborock/devices/traits/b01/q10/vacuum.py +++ b/roborock/devices/traits/b01/q10/vacuum.py @@ -1,13 +1,31 @@ """Traits for Q10 B01 devices.""" +import asyncio +import logging +from collections.abc import Callable +from math import hypot + from roborock.data.b01_q10.b01_q10_code_mappings import ( B01_Q10_DP, YXCleanType, YXDeviceCleanTask, YXFanLevel, ) +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint +from roborock.exceptions import RoborockException +from roborock.protocols.b01_q10_protocol import CleanParams, encode_clean_params from .command import CommandTrait +from .goto import GotoAction, GotoActionCommand, GotoSnapshot +from .map import MapContentTrait +from .status import StatusTrait + +_GOTO_HALF_ZONE_SIZE = 200 +_GOTO_TOLERANCE = 200 +_GOTO_TIMEOUT = 300 +_GOTO_RETRY_INTERVAL = 1 + +_LOGGER = logging.getLogger(__name__) class VacuumTrait: @@ -17,9 +35,100 @@ class VacuumTrait: commands to Q10 devices. """ - def __init__(self, command: CommandTrait) -> None: + def __init__( + self, + command: CommandTrait, + status: StatusTrait, + map_content: MapContentTrait, + ) -> None: """Initialize the VacuumTrait.""" self._command = command + self._status = status + self._map = map_content + self._goto_action: GotoAction | None = None + self._goto_action_remove_listener: Callable[[], None] | None = None + self._goto_timeout_task: asyncio.Task[None] | None = None + self._goto_command_task: asyncio.Task[None] | None = None + self._remove_map_listener = self._map.add_update_listener(self._goto_source_updated) + self._remove_status_listener = self._status.add_update_listener(self._goto_source_updated) + + async def close(self) -> None: + """Cancel background work owned by the trait.""" + self.cancel_goto() + self._remove_map_listener() + self._remove_status_listener() + + def cancel_goto(self) -> None: + """Cancel an emulated goto replaced by another command.""" + if self._goto_action is not None: + self._goto_action.complete() + self._goto_action = None + if self._goto_action_remove_listener is not None: + self._goto_action_remove_listener() + self._goto_action_remove_listener = None + current_task = asyncio.current_task() + for task_name in ("_goto_timeout_task", "_goto_command_task"): + if (task := getattr(self, task_name)) is not None: + if task is not current_task: + task.cancel() + setattr(self, task_name, None) + + def _goto_snapshot(self) -> GotoSnapshot: + """Return the latest state used by an active goto action.""" + return GotoSnapshot( + position=self._map.robot_position, + trace_sequence=self._map.trace_sequence, + clean_task_type=self._status.clean_task_type, + status=self._status.status, + ) + + def _goto_source_updated(self) -> None: + """Feed push-derived map or status state to the active goto action.""" + if self._goto_action is not None: + self._goto_action.update(self._goto_snapshot()) + + def _goto_action_updated(self, action: GotoAction, command: GotoActionCommand) -> None: + """Schedule a device command requested by the active goto action.""" + if action is not self._goto_action: + return + if command is GotoActionCommand.COMPLETE: + self.cancel_goto() + return + if self._goto_command_task is None: + self._goto_command_task = asyncio.create_task( + self._async_handle_goto_command(action, command), + name="roborock_q10_goto_command", + ) + + async def _async_handle_goto_command(self, action: GotoAction, command: GotoActionCommand) -> None: + """Perform a pause or stop requested by the active goto action.""" + current_task = asyncio.current_task() + dp_command = B01_Q10_DP.PAUSE if command is GotoActionCommand.PAUSE else B01_Q10_DP.STOP + try: + await self._command.send(command=dp_command, params=0) + except RoborockException as err: + if command is GotoActionCommand.PAUSE: + _LOGGER.warning("Failed to pause completed Q10 goto task; retrying: %s", err) + await asyncio.sleep(_GOTO_RETRY_INTERVAL) + if action is self._goto_action: + self._goto_command_task = None + action.retry() + return + _LOGGER.warning("Failed to stop timed-out Q10 goto task: %s", err) + if action is self._goto_action: + action.complete() + self.cancel_goto() + if self._goto_command_task is current_task: + self._goto_command_task = None + + async def _async_timeout_goto(self, action: GotoAction) -> None: + """Tell the active goto action when its safety timeout expires.""" + try: + await asyncio.sleep(_GOTO_TIMEOUT) + except asyncio.CancelledError: + return + if action is self._goto_action: + action.timeout(self._goto_snapshot()) async def start_clean(self) -> None: """Start a whole-home clean. @@ -34,6 +143,7 @@ async def start_clean(self) -> None: whole-home clean (clean_task_type -> 1). """ await self._command.send(command=B01_Q10_DP.START_CLEAN, params=1) + self.cancel_goto() async def clean_segments(self, segment_ids: list[int]) -> None: """Start a room / segment clean for the given segment (room) ids. @@ -55,6 +165,73 @@ async def clean_segments(self, segment_ids: list[int]) -> None: # "parameters" -- the firmware only accepts that exact key. params={"cmd": YXDeviceCleanTask.ELECTORAL.code, "clean_paramters": segment_ids}, ) + self.cancel_goto() + + async def clean_zone( + self, + first_corner: Q10RoborockPoint, + second_corner: Q10RoborockPoint, + *, + clean_count: int = 1, + ) -> None: + """Clean one rectangular zone in the common Roborock coordinate space.""" + encoded_zone = encode_clean_params(CleanParams(first_corner, second_corner, clean_count)) + await self._command.send( + command=B01_Q10_DP.START_CLEAN, + params={ + "cmd": YXDeviceCleanTask.DIVIDE_AREAS.code, + # "clean_paramters" is the spelling required by the firmware. + "clean_paramters": encoded_zone, + }, + ) + self.cancel_goto() + + async def goto_position(self, target: Q10RoborockPoint) -> None: + """Move to a coordinate using an owned 40 cm zone-clean task.""" + target.to_vector() + snapshot = self._goto_snapshot() + if (position := snapshot.position) is not None and hypot( + position.x - target.x, position.y - target.y + ) <= _GOTO_TOLERANCE: + if self._goto_action is not None and self._goto_action.owns(snapshot): + await self._command.send(command=B01_Q10_DP.PAUSE, params=0) + self.cancel_goto() + return + + encoded_zone = encode_clean_params( + CleanParams( + Q10RoborockPoint( + target.x - _GOTO_HALF_ZONE_SIZE, + target.y - _GOTO_HALF_ZONE_SIZE, + ), + Q10RoborockPoint( + target.x + _GOTO_HALF_ZONE_SIZE, + target.y + _GOTO_HALF_ZONE_SIZE, + ), + ) + ) + await self._command.send( + command=B01_Q10_DP.START_CLEAN, + params={ + "cmd": YXDeviceCleanTask.DIVIDE_AREAS.code, + "clean_paramters": encoded_zone, + }, + ) + self.cancel_goto() + action = GotoAction( + target, + snapshot.trace_sequence, + tolerance=_GOTO_TOLERANCE, + ) + self._goto_action = action + self._goto_action_remove_listener = action.add_update_listener( + lambda command: self._goto_action_updated(action, command) + ) + self._goto_timeout_task = asyncio.create_task( + self._async_timeout_goto(action), + name="roborock_q10_goto_timeout", + ) + action.update(self._goto_snapshot()) async def spot_clean(self) -> None: """Start a spot / part clean around the robot's current position. @@ -62,18 +239,22 @@ async def spot_clean(self) -> None: Verified live: ``{"dps": {"201": 5}}`` (clean_task_type -> 5). """ await self._command.send(command=B01_Q10_DP.START_CLEAN, params=5) + self.cancel_goto() async def pause_clean(self) -> None: """Pause the current task. Verified live: ``{"dps": {"204": 0}}``.""" await self._command.send(command=B01_Q10_DP.PAUSE, params=0) + self.cancel_goto() async def resume_clean(self) -> None: """Resume a paused task. Verified live: ``{"dps": {"205": 0}}``.""" await self._command.send(command=B01_Q10_DP.RESUME, params=0) + self.cancel_goto() async def stop_clean(self) -> None: """Stop / cancel the current task. Verified live: ``{"dps": {"206": 0}}``.""" await self._command.send(command=B01_Q10_DP.STOP, params=0) + self.cancel_goto() async def return_to_dock(self) -> None: """Send the robot back to the dock to charge. @@ -84,6 +265,7 @@ async def return_to_dock(self) -> None: wash mop en route and ``4`` = collect dust en route.) """ await self._command.send(command=B01_Q10_DP.START_BACK, params=5) + self.cancel_goto() async def empty_dustbin(self) -> None: """Empty the dustbin at the dock. diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py index 2f162c64..3a22a8ba 100644 --- a/roborock/map/b01_q10_map_parser.py +++ b/roborock/map/b01_q10_map_parser.py @@ -29,6 +29,7 @@ from vacuum_map_parser_base.config.image_config import ImageConfig from vacuum_map_parser_base.map_data import ImageData, MapData, Point +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint from roborock.data.containers import RoborockBase from roborock.exceptions import RoborockException @@ -222,11 +223,15 @@ def layers(self) -> GridLayers: @dataclass class Q10Point(RoborockBase): - """A single point in Q10 map/trace coordinate space.""" + """A point in the Q10 firmware's dock-relative trace coordinate space.""" x: int y: int + def to_roborock(self) -> Q10RoborockPoint: + """Convert this trace point to common Roborock coordinates.""" + return Q10RoborockPoint.from_trace(self.x, self.y) + @dataclass class Q10TracePacket: diff --git a/roborock/protocols/b01_q10_protocol.py b/roborock/protocols/b01_q10_protocol.py index c0cf6b9b..2eccb15b 100644 --- a/roborock/protocols/b01_q10_protocol.py +++ b/roborock/protocols/b01_q10_protocol.py @@ -2,10 +2,13 @@ import json import logging +from base64 import b64encode 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_containers import Q10RoborockPoint from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import ( Q10MapPacket, @@ -24,6 +27,51 @@ B01_VERSION = b"B01" ParamsType = list | dict | int | None +_Q10_ZONE_NAME_FIELD_LENGTH = 19 + + +@dataclass(frozen=True) +class CleanParams: + """Parameters for one rectangular Q10 zone-clean task.""" + + first_corner: Q10RoborockPoint + second_corner: Q10RoborockPoint + clean_count: int = 1 + + @property + def points(self) -> tuple[Q10RoborockPoint, ...]: + """Return rectangle vertices sorted into canonical wire order.""" + min_x, max_x = sorted((self.first_corner.x, self.second_corner.x)) + min_y, max_y = sorted((self.first_corner.y, self.second_corner.y)) + return ( + Q10RoborockPoint(min_x, min_y), + Q10RoborockPoint(max_x, min_y), + Q10RoborockPoint(max_x, max_y), + Q10RoborockPoint(min_x, max_y), + ) + + +def encode_clean_params(params: CleanParams) -> str: + """Encode Q10 zone-clean parameters for ``dpStartClean`` task type 3.""" + if not isinstance(params, CleanParams): + raise ValueError("params must be CleanParams") + if not isinstance(params.first_corner, Q10RoborockPoint) or not isinstance(params.second_corner, Q10RoborockPoint): + raise ValueError("zone corners must be Q10RoborockPoint values") + if isinstance(params.clean_count, bool) or not 1 <= params.clean_count <= 3: + raise ValueError("clean_count must be between 1 and 3") + if params.first_corner.x == params.second_corner.x or params.first_corner.y == params.second_corner.y: + raise ValueError("zone corners must enclose an area") + + points = params.points + payload = bytearray((1, params.clean_count, 1, len(points))) + for point in points: + payload.extend(pack(">hh", *point.to_vector())) + + # The app protocol reserves a fixed 19-byte UTF-8 name field per zone. An + # unnamed zone is encoded as a zero length followed by zero padding. + payload.append(0) + payload.extend(bytes(_Q10_ZONE_NAME_FIELD_LENGTH)) + return b64encode(payload).decode() def encode_mqtt_payload(command: B01_Q10_DP, params: ParamsType) -> RoborockMessage: diff --git a/tests/data/b01_q10/test_b01_q10_containers.py b/tests/data/b01_q10/test_b01_q10_containers.py new file mode 100644 index 00000000..f355e06e --- /dev/null +++ b/tests/data/b01_q10/test_b01_q10_containers.py @@ -0,0 +1,49 @@ +"""Tests for Q10 data containers.""" + +import pytest + +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint + + +@pytest.mark.parametrize( + ("vector", "expected"), + [ + ((0, 0), Q10RoborockPoint(25500, 25500)), + ((10, 20), Q10RoborockPoint(25550, 25600)), + ((-10, -20), Q10RoborockPoint(25450, 25400)), + ], +) +def test_q10_roborock_point_vector_conversion(vector: tuple[int, int], expected: Q10RoborockPoint) -> None: + """Q10 vector conversion is reversible on the device's 5 mm grid.""" + point = Q10RoborockPoint.from_vector(*vector) + + assert point == expected + assert point.to_vector() == vector + + +@pytest.mark.parametrize( + ("trace", "expected"), + [ + ((0, 0), Q10RoborockPoint(25500, 25500)), + ((276, -1), Q10RoborockPoint(26190, 25498)), + ((-1700, -800), Q10RoborockPoint(21250, 23500)), + ], +) +def test_q10_roborock_point_trace_conversion(trace: tuple[int, int], expected: Q10RoborockPoint) -> None: + """Q10 trace coordinates convert to common millimetre coordinates.""" + assert Q10RoborockPoint.from_trace(*trace) == expected + + +@pytest.mark.parametrize( + "point", + [ + Q10RoborockPoint(25501, 25500), + Q10RoborockPoint(-138345, 25500), + ], +) +def test_q10_roborock_point_rejects_invalid_vector_coordinates( + point: Q10RoborockPoint, +) -> None: + """Outbound vector coordinates must fit the signed wire grid exactly.""" + with pytest.raises(ValueError): + point.to_vector() diff --git a/tests/devices/traits/b01/q10/test_goto.py b/tests/devices/traits/b01/q10/test_goto.py new file mode 100644 index 00000000..ec6d5b2f --- /dev/null +++ b/tests/devices/traits/b01/q10/test_goto.py @@ -0,0 +1,109 @@ +"""Tests for the Q10 goto state machine.""" + +from roborock.data.b01_q10.b01_q10_code_mappings import YXDeviceCleanTask, YXDeviceState +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint +from roborock.devices.traits.b01.q10.goto import ( + GotoAction, + GotoActionCommand, + GotoSnapshot, +) + +TARGET = Q10RoborockPoint(29900, 28650) + + +def _snapshot( + *, + position: Q10RoborockPoint | None = None, + trace_sequence: int | None = 2, + clean_task_type: YXDeviceCleanTask | None = YXDeviceCleanTask.DIVIDE_AREAS, + status: YXDeviceState | None = YXDeviceState.CLEANING, +) -> GotoSnapshot: + return GotoSnapshot(position, trace_sequence, clean_task_type, status) + + +def test_goto_action_requests_pause_at_target() -> None: + """A newly owned zone session requests a pause when it reaches the target.""" + action = GotoAction(TARGET, previous_trace_sequence=1, tolerance=200) + commands: list[GotoActionCommand] = [] + action.add_update_listener(commands.append) + + action.update(_snapshot(position=Q10RoborockPoint(29800, 28650))) + + assert commands == [GotoActionCommand.PAUSE] + + +def test_goto_action_ignores_previous_trace_session() -> None: + """A cached trace from before the goto does not establish ownership.""" + action = GotoAction(TARGET, previous_trace_sequence=1, tolerance=200) + commands: list[GotoActionCommand] = [] + action.add_update_listener(commands.append) + + action.update(_snapshot(position=TARGET, trace_sequence=1)) + + assert commands == [] + + +def test_goto_action_does_not_pause_unconfirmed_task() -> None: + """A new trace alone does not establish ownership of a zone-clean task.""" + action = GotoAction(TARGET, previous_trace_sequence=1, tolerance=200) + commands: list[GotoActionCommand] = [] + action.add_update_listener(commands.append) + + action.update( + _snapshot( + position=TARGET, + clean_task_type=YXDeviceCleanTask.SMART, + ) + ) + + assert commands == [] + + +def test_goto_action_completes_when_owned_session_is_replaced() -> None: + """A later trace sequence is never controlled by the older goto.""" + action = GotoAction(TARGET, previous_trace_sequence=1, tolerance=200) + commands: list[GotoActionCommand] = [] + action.add_update_listener(commands.append) + action.update(_snapshot(position=Q10RoborockPoint(26000, 26000))) + + action.update(_snapshot(position=TARGET, trace_sequence=3)) + + assert commands == [GotoActionCommand.COMPLETE] + + +def test_goto_action_timeout_stops_only_owned_zone() -> None: + """A timeout requests stop only while the owned zone session is current.""" + action = GotoAction(TARGET, previous_trace_sequence=1, tolerance=200) + commands: list[GotoActionCommand] = [] + action.add_update_listener(commands.append) + snapshot = _snapshot(position=Q10RoborockPoint(26000, 26000)) + action.update(snapshot) + + action.timeout(snapshot) + + assert commands == [GotoActionCommand.STOP] + + +def test_goto_action_timeout_completes_without_ownership() -> None: + """A timeout cannot stop a task when no new trace session was observed.""" + action = GotoAction(TARGET, previous_trace_sequence=1, tolerance=200) + commands: list[GotoActionCommand] = [] + action.add_update_listener(commands.append) + + action.timeout(_snapshot(position=TARGET, trace_sequence=1)) + + assert commands == [GotoActionCommand.COMPLETE] + + +def test_goto_action_timeout_supersedes_failed_pause() -> None: + """A failed pause becomes a stop retry once the safety timeout has elapsed.""" + action = GotoAction(TARGET, previous_trace_sequence=1, tolerance=200) + commands: list[GotoActionCommand] = [] + action.add_update_listener(commands.append) + snapshot = _snapshot(position=TARGET) + action.update(snapshot) + + action.timeout(snapshot) + action.retry() + + assert commands == [GotoActionCommand.PAUSE, GotoActionCommand.STOP] diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index ef6aa18f..060553af 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -95,7 +95,8 @@ def test_update_from_trace_packet_populates_path_and_position() -> None: 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) == (276, -1) + assert (trait.robot_position.x, trait.robot_position.y) == (26190, 25498) + assert trait.trace_sequence == trace.sequence assert trait.robot_heading == -34 assert len(updates) == 1 @@ -584,4 +585,4 @@ def test_map_content_trait_as_dict_camelizes_child_keys() -> None: "rawName": "rr_living_room", } assert data["path"] == [{"x": 100, "y": 200}, {"x": 150, "y": 250}] - assert data["robotPosition"] == {"x": 150, "y": 250} + assert data["robotPosition"] == {"x": 25875, "y": 26125} diff --git a/tests/devices/traits/b01/q10/test_vacuum.py b/tests/devices/traits/b01/q10/test_vacuum.py index 9ce448bf..6b6c477e 100644 --- a/tests/devices/traits/b01/q10/test_vacuum.py +++ b/tests/devices/traits/b01/q10/test_vacuum.py @@ -1,11 +1,24 @@ +import asyncio +from base64 import b64decode from collections.abc import Awaitable, Callable from typing import Any +from unittest.mock import AsyncMock import pytest -from roborock.data.b01_q10.b01_q10_code_mappings import YXCleanType, YXFanLevel +from roborock.data.b01_q10.b01_q10_code_mappings import ( + B01_Q10_DP, + YXCleanType, + YXDeviceCleanTask, + YXDeviceState, + YXFanLevel, +) +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint from roborock.devices.traits.b01.q10 import Q10PropertiesApi +from roborock.devices.traits.b01.q10 import vacuum as vacuum_module from roborock.devices.traits.b01.q10.vacuum import VacuumTrait +from roborock.exceptions import RoborockException +from roborock.map.b01_q10_map_parser import Q10Point, Q10TracePacket from .conftest import FakeB01Q10Channel @@ -49,3 +62,177 @@ async def test_vacuum_commands( assert command.code == dp_code assert params == expected_params + + +async def test_clean_zone( + vacuum: VacuumTrait, + fake_channel: FakeB01Q10Channel, +) -> None: + """Test the source-verified Q10 zone payload.""" + await vacuum.clean_zone( + Q10RoborockPoint(25550, 25600), + Q10RoborockPoint(25650, 25700), + clean_count=2, + ) + + command, params = fake_channel.published_commands[0] + assert command.code == 201 + assert params["cmd"] == 3 + assert b64decode(params["clean_paramters"]) == bytes( + ( + 1, + 2, + 1, + 4, + 0, + 10, + 0, + 20, + 0, + 30, + 0, + 20, + 0, + 30, + 0, + 40, + 0, + 10, + 0, + 40, + 0, + *([0] * 19), + ) + ) + + +@pytest.mark.parametrize("clean_count", [0, 4]) +async def test_clean_zone_rejects_invalid_clean_count( + vacuum: VacuumTrait, + clean_count: int, +) -> None: + """Test that the device clean-count range is validated.""" + with pytest.raises(ValueError, match="clean_count must be between 1 and 3"): + await vacuum.clean_zone( + Q10RoborockPoint(25550, 25600), + Q10RoborockPoint(25650, 25700), + clean_count=clean_count, + ) + + +async def test_goto_position_pauses_owned_zone_at_target( + q10_api: Q10PropertiesApi, + fake_channel: FakeB01Q10Channel, +) -> None: + """A goto pauses after its own trace session reaches the target.""" + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(0, 0)], sequence=1)) + q10_api.status.clean_task_type = YXDeviceCleanTask.DIVIDE_AREAS + q10_api.status.status = YXDeviceState.CLEANING + + await q10_api.vacuum.goto_position(Q10RoborockPoint(29900, 28650)) + assert q10_api.vacuum._goto_action is not None + + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1760, 1260)], sequence=2)) + command_task = q10_api.vacuum._goto_command_task + assert command_task is not None + await command_task + + assert [command for command, _ in fake_channel.published_commands] == [ + B01_Q10_DP.START_CLEAN, + B01_Q10_DP.PAUSE, + ] + + +async def test_goto_position_does_not_pause_replacement_session( + q10_api: Q10PropertiesApi, + fake_channel: FakeB01Q10Channel, +) -> None: + """A newer trace session is not controlled by an older goto monitor.""" + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(0, 0)], sequence=1)) + q10_api.status.clean_task_type = YXDeviceCleanTask.DIVIDE_AREAS + q10_api.status.status = YXDeviceState.CLEANING + + await q10_api.vacuum.goto_position(Q10RoborockPoint(29900, 28650)) + assert q10_api.vacuum._goto_action is not None + + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(100, 100)], sequence=2)) + await asyncio.sleep(0) + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1760, 1260)], sequence=3)) + await asyncio.sleep(0) + + assert [command for command, _ in fake_channel.published_commands] == [B01_Q10_DP.START_CLEAN] + assert q10_api.vacuum._goto_action is None + + +async def test_goto_position_retries_pause( + q10_api: Q10PropertiesApi, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A transient pause failure is retried while the goto is still owned.""" + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(0, 0)], sequence=1)) + q10_api.status.clean_task_type = YXDeviceCleanTask.DIVIDE_AREAS + q10_api.status.status = YXDeviceState.CLEANING + send = AsyncMock(side_effect=[None, RoborockException("pause failed"), None]) + monkeypatch.setattr(q10_api.vacuum._command, "send", send) + monkeypatch.setattr(vacuum_module, "_GOTO_RETRY_INTERVAL", 0) + + await q10_api.vacuum.goto_position(Q10RoborockPoint(29900, 28650)) + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1760, 1260)], sequence=2)) + async with asyncio.timeout(1): + while send.await_count < 3: + await asyncio.sleep(0) + + assert [call.kwargs["command"] for call in send.await_args_list] == [ + B01_Q10_DP.START_CLEAN, + B01_Q10_DP.PAUSE, + B01_Q10_DP.PAUSE, + ] + + +async def test_goto_position_stops_owned_zone_after_timeout( + q10_api: Q10PropertiesApi, + fake_channel: FakeB01Q10Channel, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The safety timeout stops only the zone session owned by the goto.""" + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(0, 0)], sequence=1)) + q10_api.status.clean_task_type = YXDeviceCleanTask.DIVIDE_AREAS + q10_api.status.status = YXDeviceState.CLEANING + monkeypatch.setattr(vacuum_module, "_GOTO_TIMEOUT", 0.01) + monkeypatch.setattr(vacuum_module, "_GOTO_RETRY_INTERVAL", 0) + + await q10_api.vacuum.goto_position(Q10RoborockPoint(29900, 28650)) + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(100, 100)], sequence=2)) + async with asyncio.timeout(1): + while len(fake_channel.published_commands) < 2: + await asyncio.sleep(0) + + assert [command for command, _ in fake_channel.published_commands] == [ + B01_Q10_DP.START_CLEAN, + B01_Q10_DP.STOP, + ] + + +async def test_goto_position_at_current_position_pauses_owned_zone( + q10_api: Q10PropertiesApi, + fake_channel: FakeB01Q10Channel, +) -> None: + """An early return pauses an active goto zone instead of orphaning it.""" + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(0, 0)], sequence=1)) + q10_api.status.clean_task_type = YXDeviceCleanTask.DIVIDE_AREAS + q10_api.status.status = YXDeviceState.CLEANING + await q10_api.vacuum.goto_position(Q10RoborockPoint(29900, 28650)) + timeout_task = q10_api.vacuum._goto_timeout_task + assert timeout_task is not None + + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(100, 100)], sequence=2)) + await asyncio.sleep(0) + await q10_api.vacuum.goto_position(Q10RoborockPoint(25750, 25750)) + await asyncio.sleep(0) + + assert q10_api.vacuum._goto_action is None + assert timeout_task.done() + assert [command for command, _ in fake_channel.published_commands] == [ + B01_Q10_DP.START_CLEAN, + B01_Q10_DP.PAUSE, + ] diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py index 5fddc41a..2d6c40f9 100644 --- a/tests/protocols/test_b01_q10_protocol.py +++ b/tests/protocols/test_b01_q10_protocol.py @@ -3,6 +3,7 @@ import json import logging import pathlib +from base64 import b64decode from collections.abc import Generator from typing import Any @@ -11,13 +12,16 @@ from syrupy import SnapshotAssertion from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, YXWaterLevel +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint from roborock.data.code_mappings import completed_warnings from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10TracePacket from roborock.protocols.b01_q10_protocol import ( + CleanParams, Q10DpsUpdate, decode_message, decode_rpc_response, + encode_clean_params, encode_mqtt_payload, ) from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol @@ -30,6 +34,72 @@ TRACE_FIXTURE = pathlib.Path("tests/map/testdata/b01_q10_trace.bin") +def test_encode_clean_params_source_verified_fixture() -> None: + """Zone parameters match a payload verified against ss07 hardware.""" + params = CleanParams( + Q10RoborockPoint(25650, 25700), + Q10RoborockPoint(25550, 25600), + clean_count=2, + ) + + assert b64decode(encode_clean_params(params)) == bytes( + ( + 1, + 2, + 1, + 4, + 0, + 10, + 0, + 20, + 0, + 30, + 0, + 20, + 0, + 30, + 0, + 40, + 0, + 10, + 0, + 40, + 0, + *([0] * 19), + ) + ) + + +@pytest.mark.parametrize("clean_count", [0, 4, True]) +def test_encode_clean_params_rejects_invalid_clean_count(clean_count: int) -> None: + """The protocol validates the device's supported clean-count range.""" + with pytest.raises(ValueError, match="clean_count must be between 1 and 3"): + encode_clean_params( + CleanParams( + Q10RoborockPoint(25550, 25600), + Q10RoborockPoint(25650, 25700), + clean_count, + ) + ) + + +def test_encode_clean_params_rejects_empty_zone() -> None: + """Two corners must enclose a non-empty rectangle.""" + with pytest.raises(ValueError, match="zone corners must enclose an area"): + encode_clean_params( + CleanParams( + Q10RoborockPoint(25550, 25600), + Q10RoborockPoint(25550, 25700), + ) + ) + + +def test_encode_clean_params_rejects_untyped_corners() -> None: + """The wire encoder accepts only common-coordinate point objects.""" + with pytest.raises(ValueError, match="zone corners must be Q10RoborockPoint"): + encode_clean_params(CleanParams((25550, 25600), (25650, 25700))) # type: ignore[arg-type] + + def _message(payload: bytes, protocol: RoborockMessageProtocol) -> RoborockMessage: return RoborockMessage(protocol=protocol, payload=payload, version=b"B01")