|
| 1 | +"""State management for an emulated Q10 goto action.""" |
| 2 | + |
| 3 | +import logging |
| 4 | +from collections.abc import Callable |
| 5 | +from dataclasses import dataclass |
| 6 | +from enum import StrEnum |
| 7 | +from math import hypot |
| 8 | + |
| 9 | +from roborock.callbacks import CallbackList |
| 10 | +from roborock.data.b01_q10.b01_q10_code_mappings import YXDeviceCleanTask, YXDeviceState |
| 11 | +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint |
| 12 | + |
| 13 | +_LOGGER = logging.getLogger(__name__) |
| 14 | +_TERMINAL_STATES = { |
| 15 | + YXDeviceState.IDLE, |
| 16 | + YXDeviceState.PAUSED, |
| 17 | + YXDeviceState.RETURNING_HOME, |
| 18 | + YXDeviceState.CHARGING, |
| 19 | +} |
| 20 | + |
| 21 | + |
| 22 | +class GotoActionCommand(StrEnum): |
| 23 | + """A command requested by a Q10 goto action.""" |
| 24 | + |
| 25 | + PAUSE = "pause" |
| 26 | + STOP = "stop" |
| 27 | + COMPLETE = "complete" |
| 28 | + |
| 29 | + |
| 30 | +@dataclass(frozen=True) |
| 31 | +class GotoSnapshot: |
| 32 | + """Device state needed to advance a goto action.""" |
| 33 | + |
| 34 | + position: Q10RoborockPoint | None |
| 35 | + trace_sequence: int | None |
| 36 | + clean_task_type: YXDeviceCleanTask | None |
| 37 | + status: YXDeviceState | None |
| 38 | + |
| 39 | + |
| 40 | +class GotoAction: |
| 41 | + """Decide how one emulated goto should react to device updates. |
| 42 | +
|
| 43 | + The action owns no tasks and sends no device commands. ``VacuumTrait`` feeds |
| 44 | + it push-derived snapshots and performs commands requested by its callbacks. |
| 45 | + """ |
| 46 | + |
| 47 | + def __init__( |
| 48 | + self, |
| 49 | + target: Q10RoborockPoint, |
| 50 | + previous_trace_sequence: int | None, |
| 51 | + *, |
| 52 | + tolerance: int, |
| 53 | + ) -> None: |
| 54 | + """Initialize a goto action waiting for a new trace session.""" |
| 55 | + self._target = target |
| 56 | + self._previous_trace_sequence = previous_trace_sequence |
| 57 | + self._tolerance = tolerance |
| 58 | + self._owned_trace_sequence: int | None = None |
| 59 | + self._owned_task_seen = False |
| 60 | + self._command_pending = False |
| 61 | + self._timeout_requested = False |
| 62 | + self._finished = False |
| 63 | + self._latest_snapshot: GotoSnapshot | None = None |
| 64 | + self._callbacks: CallbackList[GotoActionCommand] = CallbackList(logger=_LOGGER) |
| 65 | + |
| 66 | + def add_update_listener(self, callback: Callable[[GotoActionCommand], None]) -> Callable[[], None]: |
| 67 | + """Register a callback for the next command requested by the action.""" |
| 68 | + return self._callbacks.add_callback(callback) |
| 69 | + |
| 70 | + def update(self, snapshot: GotoSnapshot) -> None: |
| 71 | + """Process the latest push-derived device state.""" |
| 72 | + self._latest_snapshot = snapshot |
| 73 | + self._evaluate(snapshot) |
| 74 | + |
| 75 | + def retry(self) -> None: |
| 76 | + """Re-evaluate the latest state after a requested command failed.""" |
| 77 | + if self._finished or self._latest_snapshot is None: |
| 78 | + return |
| 79 | + self._command_pending = False |
| 80 | + if self._timeout_requested: |
| 81 | + self._evaluate_timeout(self._latest_snapshot) |
| 82 | + else: |
| 83 | + self._evaluate(self._latest_snapshot) |
| 84 | + |
| 85 | + def timeout(self, snapshot: GotoSnapshot) -> None: |
| 86 | + """Request a stop only if this action still owns the current zone task.""" |
| 87 | + if self._finished: |
| 88 | + return |
| 89 | + self._latest_snapshot = snapshot |
| 90 | + self._timeout_requested = True |
| 91 | + if self._command_pending: |
| 92 | + return |
| 93 | + self._evaluate_timeout(snapshot) |
| 94 | + |
| 95 | + def _evaluate_timeout(self, snapshot: GotoSnapshot) -> None: |
| 96 | + """Derive the safe timeout command from the latest device state.""" |
| 97 | + if self.owns(snapshot): |
| 98 | + self._emit(GotoActionCommand.STOP) |
| 99 | + else: |
| 100 | + self._emit(GotoActionCommand.COMPLETE) |
| 101 | + |
| 102 | + def complete(self) -> None: |
| 103 | + """Mark the action complete after its requested command succeeds.""" |
| 104 | + self._finished = True |
| 105 | + self._command_pending = False |
| 106 | + |
| 107 | + def owns(self, snapshot: GotoSnapshot) -> bool: |
| 108 | + """Return whether this action owns the current Q10 zone-clean session.""" |
| 109 | + return ( |
| 110 | + self._owned_trace_sequence is not None |
| 111 | + and snapshot.trace_sequence == self._owned_trace_sequence |
| 112 | + and snapshot.clean_task_type is YXDeviceCleanTask.DIVIDE_AREAS |
| 113 | + ) |
| 114 | + |
| 115 | + def _evaluate(self, snapshot: GotoSnapshot) -> None: |
| 116 | + """Derive the next command from the latest device state.""" |
| 117 | + if self._finished or self._command_pending: |
| 118 | + return |
| 119 | + |
| 120 | + if self._owned_trace_sequence is None: |
| 121 | + if snapshot.trace_sequence is not None and snapshot.trace_sequence != self._previous_trace_sequence: |
| 122 | + self._owned_trace_sequence = snapshot.trace_sequence |
| 123 | + elif snapshot.trace_sequence != self._owned_trace_sequence: |
| 124 | + _LOGGER.debug("Q10 goto task was replaced by another cleaning session") |
| 125 | + self._emit(GotoActionCommand.COMPLETE) |
| 126 | + return |
| 127 | + |
| 128 | + if ( |
| 129 | + self._owned_trace_sequence is not None |
| 130 | + and snapshot.clean_task_type is YXDeviceCleanTask.DIVIDE_AREAS |
| 131 | + and snapshot.status not in _TERMINAL_STATES |
| 132 | + ): |
| 133 | + self._owned_task_seen = True |
| 134 | + |
| 135 | + if self._owned_task_seen and ( |
| 136 | + snapshot.clean_task_type is not YXDeviceCleanTask.DIVIDE_AREAS or snapshot.status in _TERMINAL_STATES |
| 137 | + ): |
| 138 | + self._emit(GotoActionCommand.COMPLETE) |
| 139 | + return |
| 140 | + |
| 141 | + if ( |
| 142 | + self.owns(snapshot) |
| 143 | + and snapshot.position is not None |
| 144 | + and hypot( |
| 145 | + snapshot.position.x - self._target.x, |
| 146 | + snapshot.position.y - self._target.y, |
| 147 | + ) |
| 148 | + <= self._tolerance |
| 149 | + ): |
| 150 | + self._emit(GotoActionCommand.PAUSE) |
| 151 | + |
| 152 | + def _emit(self, command: GotoActionCommand) -> None: |
| 153 | + """Publish a requested command once until it is handled.""" |
| 154 | + if command is GotoActionCommand.COMPLETE: |
| 155 | + self._finished = True |
| 156 | + else: |
| 157 | + self._command_pending = True |
| 158 | + self._callbacks(command) |
0 commit comments