From e487c86b87a1799b7beb63ab7c8a3c02bfbd809b Mon Sep 17 00:00:00 2001 From: Hmmbob <33529490+hmmbob@users.noreply.github.com> Date: Fri, 31 Jul 2026 16:39:00 +0200 Subject: [PATCH 1/4] feat: add Q10 zone cleaning and position coordinates --- .../devices/traits/b01/q10/coordinates.py | 18 ++++++ roborock/devices/traits/b01/q10/vacuum.py | 58 +++++++++++++++++++ tests/devices/traits/b01/q10/test_map.py | 2 + tests/devices/traits/b01/q10/test_vacuum.py | 49 ++++++++++++++++ 4 files changed, 127 insertions(+) create mode 100644 roborock/devices/traits/b01/q10/coordinates.py diff --git a/roborock/devices/traits/b01/q10/coordinates.py b/roborock/devices/traits/b01/q10/coordinates.py new file mode 100644 index 00000000..7c48e932 --- /dev/null +++ b/roborock/devices/traits/b01/q10/coordinates.py @@ -0,0 +1,18 @@ +"""Coordinate conversion helpers for Q10 B01 devices.""" + +# Q10 trace coordinates are relative to the dock and use 2.5 mm units. The +# public Roborock actions use millimetres with the dock at (25500, 25500). +ROBOROCK_COORDINATE_OFFSET = 25500 +Q10_TRACE_UNIT_MM = 2.5 +# Zone and restriction vectors use 5 mm units in the same dock-relative frame. +Q10_VECTOR_UNIT_MM = 5 + + +def trace_to_roborock_coordinate(value: int) -> int: + """Convert a Q10 trace value to the common Roborock coordinate space.""" + return round(ROBOROCK_COORDINATE_OFFSET + value * Q10_TRACE_UNIT_MM) + + +def roborock_to_vector_coordinate(value: int) -> int: + """Convert a common Roborock coordinate to the Q10 vector format.""" + return round((value - ROBOROCK_COORDINATE_OFFSET) / Q10_VECTOR_UNIT_MM) diff --git a/roborock/devices/traits/b01/q10/vacuum.py b/roborock/devices/traits/b01/q10/vacuum.py index 2747e024..c1dc9610 100644 --- a/roborock/devices/traits/b01/q10/vacuum.py +++ b/roborock/devices/traits/b01/q10/vacuum.py @@ -1,5 +1,9 @@ """Traits for Q10 B01 devices.""" +from base64 import b64encode +from struct import error as StructError +from struct import pack + from roborock.data.b01_q10.b01_q10_code_mappings import ( B01_Q10_DP, YXCleanType, @@ -8,6 +12,41 @@ ) from .command import CommandTrait +from .coordinates import roborock_to_vector_coordinate + +_ZONE_NAME_FIELD_LENGTH = 19 + + +def _encode_zone(x1: int, y1: int, x2: int, y2: int, clean_count: int) -> str: + """Encode one rectangular Q10 cleaning zone.""" + if not 1 <= clean_count <= 3: + raise ValueError("clean_count must be between 1 and 3") + + min_x, max_x = sorted((x1, x2)) + min_y, max_y = sorted((y1, y2)) + points = ( + (min_x, min_y), + (max_x, min_y), + (max_x, max_y), + (min_x, max_y), + ) + payload = bytearray((1, clean_count, 1, len(points))) + try: + for point_x, point_y in points: + payload.extend( + pack( + ">hh", + roborock_to_vector_coordinate(point_x), + roborock_to_vector_coordinate(point_y), + ) + ) + except StructError as err: + raise ValueError("zone coordinates are outside the supported range") from err + + # The app protocol reserves a fixed 19-byte UTF-8 name field per zone. + payload.append(0) + payload.extend(bytes(_ZONE_NAME_FIELD_LENGTH)) + return b64encode(payload).decode() class VacuumTrait: @@ -56,6 +95,25 @@ async def clean_segments(self, segment_ids: list[int]) -> None: params={"cmd": YXDeviceCleanTask.ELECTORAL.code, "clean_paramters": segment_ids}, ) + async def clean_zone( + self, + x1: int, + y1: int, + x2: int, + y2: int, + *, + clean_count: int = 1, + ) -> None: + """Clean one rectangular zone in the common Roborock coordinate space.""" + 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": _encode_zone(x1, y1, x2, y2, clean_count), + }, + ) + async def spot_clean(self) -> None: """Start a spot / part clean around the robot's current position. diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index ef6aa18f..ea13144a 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -96,6 +96,8 @@ def test_update_from_trace_packet_populates_path_and_position() -> None: 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.roborock_position is not None + assert (trait.roborock_position.x, trait.roborock_position.y) == (26190, 25498) assert trait.robot_heading == -34 assert len(updates) == 1 diff --git a/tests/devices/traits/b01/q10/test_vacuum.py b/tests/devices/traits/b01/q10/test_vacuum.py index 9ce448bf..240eeb2f 100644 --- a/tests/devices/traits/b01/q10/test_vacuum.py +++ b/tests/devices/traits/b01/q10/test_vacuum.py @@ -1,3 +1,4 @@ +from base64 import b64decode from collections.abc import Awaitable, Callable from typing import Any @@ -49,3 +50,51 @@ 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(25550, 25600, 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(25550, 25600, 25650, 25700, clean_count=clean_count) From 6269118b1ecd74c7bf53d139531974e0a703f236 Mon Sep 17 00:00:00 2001 From: Hmmbob <33529490+hmmbob@users.noreply.github.com> Date: Fri, 31 Jul 2026 17:07:04 +0200 Subject: [PATCH 2/4] feat: add safe Q10 goto lifecycle --- roborock/devices/traits/b01/q10/__init__.py | 3 +- roborock/devices/traits/b01/q10/map.py | 16 ++ roborock/devices/traits/b01/q10/vacuum.py | 181 +++++++++++++++++++- tests/devices/traits/b01/q10/test_map.py | 1 + tests/devices/traits/b01/q10/test_vacuum.py | 130 +++++++++++++- 5 files changed, 327 insertions(+), 4 deletions(-) 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/map.py b/roborock/devices/traits/b01/q10/map.py index 5890e688..5850d1f6 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -33,6 +33,7 @@ from .command import CommandTrait from .common import UpdatableTrait +from .coordinates import trace_to_roborock_coordinate from .maps import MapsTrait _LOGGER = logging.getLogger(__name__) @@ -135,6 +136,21 @@ 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 + @property + def roborock_position(self) -> Q10Point | None: + """Current position in the common Roborock millimetre coordinate space.""" + if (position := self.robot_position) is None: + return None + return Q10Point( + x=trace_to_roborock_coordinate(position.x), + y=trace_to_roborock_coordinate(position.y), + ) + + @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: """Current heading for orienting a robot marker on a caller-rendered map.""" diff --git a/roborock/devices/traits/b01/q10/vacuum.py b/roborock/devices/traits/b01/q10/vacuum.py index c1dc9610..bd312eec 100644 --- a/roborock/devices/traits/b01/q10/vacuum.py +++ b/roborock/devices/traits/b01/q10/vacuum.py @@ -1,6 +1,9 @@ """Traits for Q10 B01 devices.""" +import asyncio +import logging from base64 import b64encode +from math import hypot from struct import error as StructError from struct import pack @@ -8,13 +11,23 @@ B01_Q10_DP, YXCleanType, YXDeviceCleanTask, + YXDeviceState, YXFanLevel, ) +from roborock.exceptions import RoborockException from .command import CommandTrait from .coordinates import roborock_to_vector_coordinate +from .map import MapContentTrait +from .status import StatusTrait _ZONE_NAME_FIELD_LENGTH = 19 +_GOTO_HALF_ZONE_SIZE = 200 +_GOTO_TOLERANCE = 200 +_GOTO_TIMEOUT = 300 +_GOTO_RETRY_INTERVAL = 1 + +_LOGGER = logging.getLogger(__name__) def _encode_zone(x1: int, y1: int, x2: int, y2: int, clean_count: int) -> str: @@ -56,9 +69,128 @@ 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_monitor_task: asyncio.Task[None] | None = None + self._goto_trace_sequence: int | None = None + + async def close(self) -> None: + """Cancel background work owned by the trait.""" + if (task := self._goto_monitor_task) is None: + return + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self._goto_monitor_task = None + self._goto_trace_sequence = None + + def cancel_goto(self) -> None: + """Cancel monitoring for an emulated goto replaced by another command.""" + if self._goto_monitor_task is not None: + self._goto_monitor_task.cancel() + self._goto_monitor_task = None + self._goto_trace_sequence = None + + async def _async_monitor_goto_target( + self, + x: int, + y: int, + previous_trace_sequence: int | None, + ) -> None: + """Pause the owned mini-zone task after it reaches the target.""" + current_task = asyncio.current_task() + owned_trace_sequence: int | None = None + owned_task_seen = False + update_event = asyncio.Event() + remove_map_listener = self._map.add_update_listener(update_event.set) + remove_status_listener = self._status.add_update_listener(update_event.set) + try: + async with asyncio.timeout(_GOTO_TIMEOUT): + while True: + trace_sequence = self._map.trace_sequence + if owned_trace_sequence is None: + if trace_sequence is not None and trace_sequence != previous_trace_sequence: + owned_trace_sequence = trace_sequence + self._goto_trace_sequence = trace_sequence + elif trace_sequence != owned_trace_sequence: + _LOGGER.debug("Q10 goto task was replaced by another cleaning session") + return + + if ( + owned_trace_sequence is not None + and self._status.clean_task_type is YXDeviceCleanTask.DIVIDE_AREAS + and self._status.status + not in { + YXDeviceState.IDLE, + YXDeviceState.PAUSED, + YXDeviceState.RETURNING_HOME, + YXDeviceState.CHARGING, + } + ): + owned_task_seen = True + + if owned_task_seen and self._status.clean_task_type is not YXDeviceCleanTask.DIVIDE_AREAS: + _LOGGER.debug("Q10 goto task was replaced by another task type") + return + + if owned_task_seen and self._status.status in { + YXDeviceState.IDLE, + YXDeviceState.PAUSED, + YXDeviceState.RETURNING_HOME, + YXDeviceState.CHARGING, + }: + return + + if ( + owned_trace_sequence is not None + and (position := self._map.roborock_position) is not None + and hypot(position.x - x, position.y - y) <= _GOTO_TOLERANCE + ): + try: + await self._command.send(command=B01_Q10_DP.PAUSE, params=0) + except RoborockException as err: + _LOGGER.warning("Failed to pause completed Q10 goto task; retrying: %s", err) + else: + return + + update_event.clear() + try: + async with asyncio.timeout(_GOTO_RETRY_INTERVAL): + await update_event.wait() + except TimeoutError: + pass + except TimeoutError: + if ( + owned_trace_sequence is not None + and self._map.trace_sequence == owned_trace_sequence + and self._status.clean_task_type is YXDeviceCleanTask.DIVIDE_AREAS + ): + _LOGGER.warning( + "Q10 vacuum did not reach goto target (%s, %s) within %s seconds; stopping zone task", + x, + y, + _GOTO_TIMEOUT, + ) + try: + await self._command.send(command=B01_Q10_DP.STOP, params=0) + except RoborockException as err: + _LOGGER.warning("Failed to stop timed-out Q10 goto task: %s", err) + finally: + remove_map_listener() + remove_status_listener() + if self._goto_monitor_task is current_task: + self._goto_monitor_task = None + self._goto_trace_sequence = None async def start_clean(self) -> None: """Start a whole-home clean. @@ -73,6 +205,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. @@ -94,6 +227,7 @@ 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, @@ -105,14 +239,52 @@ async def clean_zone( clean_count: int = 1, ) -> None: """Clean one rectangular zone in the common Roborock coordinate space.""" + encoded_zone = _encode_zone(x1, y1, x2, y2, 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": _encode_zone(x1, y1, x2, y2, clean_count), + "clean_paramters": encoded_zone, + }, + ) + self.cancel_goto() + + async def goto_position(self, x: int, y: int) -> None: + """Move to a coordinate using an owned 40 cm zone-clean task.""" + if (position := self._map.roborock_position) is not None and hypot( + position.x - x, position.y - y + ) <= _GOTO_TOLERANCE: + if ( + self._goto_monitor_task is not None + and self._goto_trace_sequence is not None + and self._map.trace_sequence == self._goto_trace_sequence + and self._status.clean_task_type is YXDeviceCleanTask.DIVIDE_AREAS + ): + await self._command.send(command=B01_Q10_DP.PAUSE, params=0) + self.cancel_goto() + return + + previous_trace_sequence = self._map.trace_sequence + encoded_zone = _encode_zone( + x - _GOTO_HALF_ZONE_SIZE, + y - _GOTO_HALF_ZONE_SIZE, + x + _GOTO_HALF_ZONE_SIZE, + y + _GOTO_HALF_ZONE_SIZE, + 1, + ) + await self._command.send( + command=B01_Q10_DP.START_CLEAN, + params={ + "cmd": YXDeviceCleanTask.DIVIDE_AREAS.code, + "clean_paramters": encoded_zone, }, ) + self.cancel_goto() + self._goto_monitor_task = asyncio.create_task( + self._async_monitor_goto_target(x, y, previous_trace_sequence), + name="roborock_q10_goto", + ) async def spot_clean(self) -> None: """Start a spot / part clean around the robot's current position. @@ -120,18 +292,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. @@ -142,6 +318,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/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index ea13144a..204a3e07 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -98,6 +98,7 @@ def test_update_from_trace_packet_populates_path_and_position() -> None: assert (trait.robot_position.x, trait.robot_position.y) == (276, -1) assert trait.roborock_position is not None assert (trait.roborock_position.x, trait.roborock_position.y) == (26190, 25498) + assert trait.trace_sequence == trace.sequence assert trait.robot_heading == -34 assert len(updates) == 1 diff --git a/tests/devices/traits/b01/q10/test_vacuum.py b/tests/devices/traits/b01/q10/test_vacuum.py index 240eeb2f..b7f53371 100644 --- a/tests/devices/traits/b01/q10/test_vacuum.py +++ b/tests/devices/traits/b01/q10/test_vacuum.py @@ -1,12 +1,23 @@ +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.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 @@ -98,3 +109,120 @@ async def test_clean_zone_rejects_invalid_clean_count( """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(25550, 25600, 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(29900, 28650) + monitor = q10_api.vacuum._goto_monitor_task + assert monitor is not None + + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1760, 1260)], sequence=2)) + await monitor + + 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(29900, 28650) + monitor = q10_api.vacuum._goto_monitor_task + assert monitor 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 monitor + + assert [command for command, _ in fake_channel.published_commands] == [B01_Q10_DP.START_CLEAN] + + +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]) + q10_api.vacuum._command.send = send + monkeypatch.setattr(vacuum_module, "_GOTO_RETRY_INTERVAL", 0) + + await q10_api.vacuum.goto_position(29900, 28650) + monitor = q10_api.vacuum._goto_monitor_task + assert monitor is not None + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1760, 1260)], sequence=2)) + await monitor + + 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(29900, 28650) + monitor = q10_api.vacuum._goto_monitor_task + assert monitor is not None + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(100, 100)], sequence=2)) + await monitor + + 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(29900, 28650) + monitor = q10_api.vacuum._goto_monitor_task + assert monitor 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(25750, 25750) + await asyncio.sleep(0) + + assert q10_api.vacuum._goto_monitor_task is None + assert monitor.cancelled() + assert [command for command, _ in fake_channel.published_commands] == [ + B01_Q10_DP.START_CLEAN, + B01_Q10_DP.PAUSE, + ] From 34e3a43f7a96b4b7501fd946492ad949bedc4d87 Mon Sep 17 00:00:00 2001 From: Hmmbob <33529490+hmmbob@users.noreply.github.com> Date: Wed, 9 Sep 2026 17:38:14 +0200 Subject: [PATCH 3/4] refactor: apply Q10 API review feedback Co-authored-by: Harry Coureau --- roborock/data/b01_q10/b01_q10_containers.py | 56 ++++ .../devices/traits/b01/q10/coordinates.py | 18 -- roborock/devices/traits/b01/q10/goto.py | 158 ++++++++++ roborock/devices/traits/b01/q10/map.py | 22 +- roborock/devices/traits/b01/q10/vacuum.py | 279 +++++++----------- roborock/map/b01_q10_map_parser.py | 7 +- roborock/protocols/b01_q10_protocol.py | 48 +++ tests/data/b01_q10/test_b01_q10_containers.py | 49 +++ tests/devices/traits/b01/q10/test_goto.py | 93 ++++++ tests/devices/traits/b01/q10/test_map.py | 6 +- tests/devices/traits/b01/q10/test_vacuum.py | 60 ++-- tests/protocols/test_b01_q10_protocol.py | 70 +++++ 12 files changed, 638 insertions(+), 228 deletions(-) delete mode 100644 roborock/devices/traits/b01/q10/coordinates.py create mode 100644 roborock/devices/traits/b01/q10/goto.py create mode 100644 tests/data/b01_q10/test_b01_q10_containers.py create mode 100644 tests/devices/traits/b01/q10/test_goto.py 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/coordinates.py b/roborock/devices/traits/b01/q10/coordinates.py deleted file mode 100644 index 7c48e932..00000000 --- a/roborock/devices/traits/b01/q10/coordinates.py +++ /dev/null @@ -1,18 +0,0 @@ -"""Coordinate conversion helpers for Q10 B01 devices.""" - -# Q10 trace coordinates are relative to the dock and use 2.5 mm units. The -# public Roborock actions use millimetres with the dock at (25500, 25500). -ROBOROCK_COORDINATE_OFFSET = 25500 -Q10_TRACE_UNIT_MM = 2.5 -# Zone and restriction vectors use 5 mm units in the same dock-relative frame. -Q10_VECTOR_UNIT_MM = 5 - - -def trace_to_roborock_coordinate(value: int) -> int: - """Convert a Q10 trace value to the common Roborock coordinate space.""" - return round(ROBOROCK_COORDINATE_OFFSET + value * Q10_TRACE_UNIT_MM) - - -def roborock_to_vector_coordinate(value: int) -> int: - """Convert a common Roborock coordinate to the Q10 vector format.""" - return round((value - ROBOROCK_COORDINATE_OFFSET) / Q10_VECTOR_UNIT_MM) diff --git a/roborock/devices/traits/b01/q10/goto.py b/roborock/devices/traits/b01/q10/goto.py new file mode 100644 index 00000000..371b8c35 --- /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._owned_trace_sequence is not None + 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 5850d1f6..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 ( @@ -33,7 +34,6 @@ from .command import CommandTrait from .common import UpdatableTrait -from .coordinates import trace_to_roborock_coordinate from .maps import MapsTrait _LOGGER = logging.getLogger(__name__) @@ -128,23 +128,15 @@ 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 - - @property - def roborock_position(self) -> Q10Point | None: + def robot_position(self) -> Q10RoborockPoint | None: """Current position in the common Roborock millimetre coordinate space.""" - if (position := self.robot_position) is None: + if self._trace_packet is None or (position := self._trace_packet.robot_position) is None: return None - return Q10Point( - x=trace_to_roborock_coordinate(position.x), - y=trace_to_roborock_coordinate(position.y), - ) + return position.to_roborock() @property def trace_sequence(self) -> int | None: @@ -197,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 bd312eec..4b846576 100644 --- a/roborock/devices/traits/b01/q10/vacuum.py +++ b/roborock/devices/traits/b01/q10/vacuum.py @@ -2,26 +2,24 @@ import asyncio import logging -from base64 import b64encode +from collections.abc import Callable from math import hypot -from struct import error as StructError -from struct import pack 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.exceptions import RoborockException +from roborock.protocols.b01_q10_protocol import CleanParams, encode_clean_params from .command import CommandTrait -from .coordinates import roborock_to_vector_coordinate +from .goto import GotoAction, GotoActionCommand, GotoSnapshot from .map import MapContentTrait from .status import StatusTrait -_ZONE_NAME_FIELD_LENGTH = 19 _GOTO_HALF_ZONE_SIZE = 200 _GOTO_TOLERANCE = 200 _GOTO_TIMEOUT = 300 @@ -30,38 +28,6 @@ _LOGGER = logging.getLogger(__name__) -def _encode_zone(x1: int, y1: int, x2: int, y2: int, clean_count: int) -> str: - """Encode one rectangular Q10 cleaning zone.""" - if not 1 <= clean_count <= 3: - raise ValueError("clean_count must be between 1 and 3") - - min_x, max_x = sorted((x1, x2)) - min_y, max_y = sorted((y1, y2)) - points = ( - (min_x, min_y), - (max_x, min_y), - (max_x, max_y), - (min_x, max_y), - ) - payload = bytearray((1, clean_count, 1, len(points))) - try: - for point_x, point_y in points: - payload.extend( - pack( - ">hh", - roborock_to_vector_coordinate(point_x), - roborock_to_vector_coordinate(point_y), - ) - ) - except StructError as err: - raise ValueError("zone coordinates are outside the supported range") from err - - # The app protocol reserves a fixed 19-byte UTF-8 name field per zone. - payload.append(0) - payload.extend(bytes(_ZONE_NAME_FIELD_LENGTH)) - return b64encode(payload).decode() - - class VacuumTrait: """Trait for sending vacuum commands. @@ -79,118 +45,90 @@ def __init__( self._command = command self._status = status self._map = map_content - self._goto_monitor_task: asyncio.Task[None] | None = None - self._goto_trace_sequence: int | None = None + 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.""" - if (task := self._goto_monitor_task) is None: - return - task.cancel() - try: - await task - except asyncio.CancelledError: - pass - self._goto_monitor_task = None - self._goto_trace_sequence = None + self.cancel_goto() + self._remove_map_listener() + self._remove_status_listener() def cancel_goto(self) -> None: - """Cancel monitoring for an emulated goto replaced by another command.""" - if self._goto_monitor_task is not None: - self._goto_monitor_task.cancel() - self._goto_monitor_task = None - self._goto_trace_sequence = 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, + ) - async def _async_monitor_goto_target( - self, - x: int, - y: int, - previous_trace_sequence: int | None, - ) -> None: - """Pause the owned mini-zone task after it reaches the target.""" + 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() - owned_trace_sequence: int | None = None - owned_task_seen = False - update_event = asyncio.Event() - remove_map_listener = self._map.add_update_listener(update_event.set) - remove_status_listener = self._status.add_update_listener(update_event.set) + 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: - async with asyncio.timeout(_GOTO_TIMEOUT): - while True: - trace_sequence = self._map.trace_sequence - if owned_trace_sequence is None: - if trace_sequence is not None and trace_sequence != previous_trace_sequence: - owned_trace_sequence = trace_sequence - self._goto_trace_sequence = trace_sequence - elif trace_sequence != owned_trace_sequence: - _LOGGER.debug("Q10 goto task was replaced by another cleaning session") - return - - if ( - owned_trace_sequence is not None - and self._status.clean_task_type is YXDeviceCleanTask.DIVIDE_AREAS - and self._status.status - not in { - YXDeviceState.IDLE, - YXDeviceState.PAUSED, - YXDeviceState.RETURNING_HOME, - YXDeviceState.CHARGING, - } - ): - owned_task_seen = True - - if owned_task_seen and self._status.clean_task_type is not YXDeviceCleanTask.DIVIDE_AREAS: - _LOGGER.debug("Q10 goto task was replaced by another task type") - return - - if owned_task_seen and self._status.status in { - YXDeviceState.IDLE, - YXDeviceState.PAUSED, - YXDeviceState.RETURNING_HOME, - YXDeviceState.CHARGING, - }: - return - - if ( - owned_trace_sequence is not None - and (position := self._map.roborock_position) is not None - and hypot(position.x - x, position.y - y) <= _GOTO_TOLERANCE - ): - try: - await self._command.send(command=B01_Q10_DP.PAUSE, params=0) - except RoborockException as err: - _LOGGER.warning("Failed to pause completed Q10 goto task; retrying: %s", err) - else: - return - - update_event.clear() - try: - async with asyncio.timeout(_GOTO_RETRY_INTERVAL): - await update_event.wait() - except TimeoutError: - pass - except TimeoutError: - if ( - owned_trace_sequence is not None - and self._map.trace_sequence == owned_trace_sequence - and self._status.clean_task_type is YXDeviceCleanTask.DIVIDE_AREAS - ): - _LOGGER.warning( - "Q10 vacuum did not reach goto target (%s, %s) within %s seconds; stopping zone task", - x, - y, - _GOTO_TIMEOUT, - ) - try: - await self._command.send(command=B01_Q10_DP.STOP, params=0) - except RoborockException as err: - _LOGGER.warning("Failed to stop timed-out Q10 goto task: %s", err) - finally: - remove_map_listener() - remove_status_listener() - if self._goto_monitor_task is current_task: - self._goto_monitor_task = None - self._goto_trace_sequence = None + 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. @@ -231,15 +169,13 @@ async def clean_segments(self, segment_ids: list[int]) -> None: async def clean_zone( self, - x1: int, - y1: int, - x2: int, - y2: int, + first_corner: Q10RoborockPoint, + second_corner: Q10RoborockPoint, *, clean_count: int = 1, ) -> None: """Clean one rectangular zone in the common Roborock coordinate space.""" - encoded_zone = _encode_zone(x1, y1, x2, y2, clean_count) + encoded_zone = encode_clean_params(CleanParams(first_corner, second_corner, clean_count)) await self._command.send( command=B01_Q10_DP.START_CLEAN, params={ @@ -250,28 +186,29 @@ async def clean_zone( ) self.cancel_goto() - async def goto_position(self, x: int, y: int) -> None: + async def goto_position(self, target: Q10RoborockPoint) -> None: """Move to a coordinate using an owned 40 cm zone-clean task.""" - if (position := self._map.roborock_position) is not None and hypot( - position.x - x, position.y - y + 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_monitor_task is not None - and self._goto_trace_sequence is not None - and self._map.trace_sequence == self._goto_trace_sequence - and self._status.clean_task_type is YXDeviceCleanTask.DIVIDE_AREAS - ): + 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 - previous_trace_sequence = self._map.trace_sequence - encoded_zone = _encode_zone( - x - _GOTO_HALF_ZONE_SIZE, - y - _GOTO_HALF_ZONE_SIZE, - x + _GOTO_HALF_ZONE_SIZE, - y + _GOTO_HALF_ZONE_SIZE, - 1, + 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, @@ -281,10 +218,20 @@ async def goto_position(self, x: int, y: int) -> None: }, ) self.cancel_goto() - self._goto_monitor_task = asyncio.create_task( - self._async_monitor_goto_target(x, y, previous_trace_sequence), - name="roborock_q10_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. 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..afc6131c 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 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..fc8b31cc --- /dev/null +++ b/tests/devices/traits/b01/q10/test_goto.py @@ -0,0 +1,93 @@ +"""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_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 204a3e07..060553af 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -95,9 +95,7 @@ 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.roborock_position is not None - assert (trait.roborock_position.x, trait.roborock_position.y) == (26190, 25498) + 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 @@ -587,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 b7f53371..6b6c477e 100644 --- a/tests/devices/traits/b01/q10/test_vacuum.py +++ b/tests/devices/traits/b01/q10/test_vacuum.py @@ -13,6 +13,7 @@ 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 @@ -68,7 +69,11 @@ async def test_clean_zone( fake_channel: FakeB01Q10Channel, ) -> None: """Test the source-verified Q10 zone payload.""" - await vacuum.clean_zone(25550, 25600, 25650, 25700, clean_count=2) + await vacuum.clean_zone( + Q10RoborockPoint(25550, 25600), + Q10RoborockPoint(25650, 25700), + clean_count=2, + ) command, params = fake_channel.published_commands[0] assert command.code == 201 @@ -108,7 +113,11 @@ async def test_clean_zone_rejects_invalid_clean_count( ) -> 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(25550, 25600, 25650, 25700, clean_count=clean_count) + await vacuum.clean_zone( + Q10RoborockPoint(25550, 25600), + Q10RoborockPoint(25650, 25700), + clean_count=clean_count, + ) async def test_goto_position_pauses_owned_zone_at_target( @@ -120,12 +129,13 @@ async def test_goto_position_pauses_owned_zone_at_target( q10_api.status.clean_task_type = YXDeviceCleanTask.DIVIDE_AREAS q10_api.status.status = YXDeviceState.CLEANING - await q10_api.vacuum.goto_position(29900, 28650) - monitor = q10_api.vacuum._goto_monitor_task - assert monitor is not None + 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)) - await monitor + 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, @@ -142,16 +152,16 @@ async def test_goto_position_does_not_pause_replacement_session( q10_api.status.clean_task_type = YXDeviceCleanTask.DIVIDE_AREAS q10_api.status.status = YXDeviceState.CLEANING - await q10_api.vacuum.goto_position(29900, 28650) - monitor = q10_api.vacuum._goto_monitor_task - assert monitor is not None + 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 monitor + 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( @@ -163,14 +173,14 @@ async def test_goto_position_retries_pause( q10_api.status.clean_task_type = YXDeviceCleanTask.DIVIDE_AREAS q10_api.status.status = YXDeviceState.CLEANING send = AsyncMock(side_effect=[None, RoborockException("pause failed"), None]) - q10_api.vacuum._command.send = send + monkeypatch.setattr(q10_api.vacuum._command, "send", send) monkeypatch.setattr(vacuum_module, "_GOTO_RETRY_INTERVAL", 0) - await q10_api.vacuum.goto_position(29900, 28650) - monitor = q10_api.vacuum._goto_monitor_task - assert monitor is not None + await q10_api.vacuum.goto_position(Q10RoborockPoint(29900, 28650)) q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1760, 1260)], sequence=2)) - await monitor + 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, @@ -191,11 +201,11 @@ async def test_goto_position_stops_owned_zone_after_timeout( monkeypatch.setattr(vacuum_module, "_GOTO_TIMEOUT", 0.01) monkeypatch.setattr(vacuum_module, "_GOTO_RETRY_INTERVAL", 0) - await q10_api.vacuum.goto_position(29900, 28650) - monitor = q10_api.vacuum._goto_monitor_task - assert monitor is not None + await q10_api.vacuum.goto_position(Q10RoborockPoint(29900, 28650)) q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(100, 100)], sequence=2)) - await monitor + 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, @@ -211,17 +221,17 @@ async def test_goto_position_at_current_position_pauses_owned_zone( 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(29900, 28650) - monitor = q10_api.vacuum._goto_monitor_task - assert monitor is not None + 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(25750, 25750) + await q10_api.vacuum.goto_position(Q10RoborockPoint(25750, 25750)) await asyncio.sleep(0) - assert q10_api.vacuum._goto_monitor_task is None - assert monitor.cancelled() + 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..bb6b494f 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]) +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") From 38c2c52d0a7a77adf8f4e6b81f51f1a130a3e2c9 Mon Sep 17 00:00:00 2001 From: Hmmbob <33529490+hmmbob@users.noreply.github.com> Date: Wed, 9 Sep 2026 17:59:46 +0200 Subject: [PATCH 4/4] fix: require confirmed Q10 goto ownership --- roborock/devices/traits/b01/q10/goto.py | 2 +- roborock/protocols/b01_q10_protocol.py | 2 +- tests/devices/traits/b01/q10/test_goto.py | 16 ++++++++++++++++ tests/protocols/test_b01_q10_protocol.py | 2 +- 4 files changed, 19 insertions(+), 3 deletions(-) diff --git a/roborock/devices/traits/b01/q10/goto.py b/roborock/devices/traits/b01/q10/goto.py index 371b8c35..d2c11294 100644 --- a/roborock/devices/traits/b01/q10/goto.py +++ b/roborock/devices/traits/b01/q10/goto.py @@ -139,7 +139,7 @@ def _evaluate(self, snapshot: GotoSnapshot) -> None: return if ( - self._owned_trace_sequence is not None + self.owns(snapshot) and snapshot.position is not None and hypot( snapshot.position.x - self._target.x, diff --git a/roborock/protocols/b01_q10_protocol.py b/roborock/protocols/b01_q10_protocol.py index afc6131c..2eccb15b 100644 --- a/roborock/protocols/b01_q10_protocol.py +++ b/roborock/protocols/b01_q10_protocol.py @@ -57,7 +57,7 @@ def encode_clean_params(params: CleanParams) -> str: 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 not 1 <= params.clean_count <= 3: + 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") diff --git a/tests/devices/traits/b01/q10/test_goto.py b/tests/devices/traits/b01/q10/test_goto.py index fc8b31cc..ec6d5b2f 100644 --- a/tests/devices/traits/b01/q10/test_goto.py +++ b/tests/devices/traits/b01/q10/test_goto.py @@ -43,6 +43,22 @@ def test_goto_action_ignores_previous_trace_session() -> None: 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) diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py index bb6b494f..2d6c40f9 100644 --- a/tests/protocols/test_b01_q10_protocol.py +++ b/tests/protocols/test_b01_q10_protocol.py @@ -70,7 +70,7 @@ def test_encode_clean_params_source_verified_fixture() -> None: ) -@pytest.mark.parametrize("clean_count", [0, 4]) +@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"):