Plan 3: state machine (IDLE -> WAKE_PENDING -> LISTENING <-> ASSISTANT_SPEAKING -> IDLE_PENDING -> IDLE)

2026-06-11 22:47:23 +00:00
parent 908e3f4907
commit d75129c4f9
2 changed files with 172 additions and 0 deletions
@@ -0,0 +1,70 @@
+"""Assistant client state machine.
+
+States flow: IDLE -> WAKE_PENDING -> LISTENING <-> ASSISTANT_SPEAKING
+                                       \\         /
+                                        session_ended
+                                                 \\
+                                                  IDLE_PENDING -> IDLE
+"""
+from enum import Enum, auto
+
+from client.log import get_logger
+
+_log = get_logger("state")
+
+
+class State(Enum):
+    IDLE = auto()
+    WAKE_PENDING = auto()
+    LISTENING = auto()
+    ASSISTANT_SPEAKING = auto()
+    IDLE_PENDING = auto()
+
+
+class StateMachine:
+    def __init__(self) -> None:
+        self._state = State.IDLE
+
+    @property
+    def state(self) -> State:
+        return self._state
+
+    @property
+    def wakeword_enabled(self) -> bool:
+        return self._state is State.IDLE
+
+    @property
+    def uplink_enabled(self) -> bool:
+        return self._state is State.LISTENING
+
+    def _transition(self, target: State) -> None:
+        if target is self._state:
+            return
+        _log.info("state %s -> %s", self._state.name, target.name)
+        self._state = target
+
+    def wake_fired(self) -> bool:
+        if self._state is not State.IDLE:
+            return False
+        self._transition(State.WAKE_PENDING)
+        return True
+
+    def session_started(self) -> None:
+        if self._state in (State.WAKE_PENDING, State.IDLE):
+            self._transition(State.LISTENING)
+
+    def assistant_audio_arrived(self) -> None:
+        if self._state is State.LISTENING:
+            self._transition(State.ASSISTANT_SPEAKING)
+
+    def assistant_done(self) -> None:
+        if self._state is State.ASSISTANT_SPEAKING:
+            self._transition(State.LISTENING)
+
+    def session_ended(self) -> None:
+        if self._state in (State.LISTENING, State.ASSISTANT_SPEAKING, State.WAKE_PENDING):
+            self._transition(State.IDLE_PENDING)
+
+    def sleep_played(self) -> None:
+        if self._state is State.IDLE_PENDING:
+            self._transition(State.IDLE)