71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
"""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)
|