"""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)