72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
"""Pet state machine — pure logic, no Qt/audio dependencies, so it's cheap
|
|
to unit test. The UI layer (ui/pet_window.py) reacts to state changes by
|
|
swapping the active sprite animation; the worker thread (ui/app.py) drives
|
|
transitions as the mic/wake/converse/tts pipeline progresses.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from enum import Enum
|
|
from typing import Callable, Optional
|
|
|
|
|
|
class PetState(str, Enum):
|
|
IDLE = "idle" # waiting for the wake phrase (or a click)
|
|
LISTENING = "listening" # actively recording an utterance
|
|
THINKING = "thinking" # waiting on the server (STT done, converse in flight)
|
|
TALKING = "talking" # playing back the TTS reply
|
|
ERROR = "error" # brief flash state on failure, then back to idle
|
|
|
|
|
|
# States it's valid to move to from each state. Keeps ad-hoc bugs (e.g.
|
|
# firing TALKING before a reply exists) from silently passing through.
|
|
#
|
|
# IDLE -> TALKING is legal (not just IDLE -> LISTENING) because of proactive
|
|
# announcements: the heartbeat poll (controller.py's _maybe_heartbeat) can
|
|
# make the pet speak unprompted — a reminder firing, a nudge from the server
|
|
# — without the user having said anything first, so there's no preceding
|
|
# LISTENING/THINKING leg for that turn.
|
|
_TRANSITIONS: dict[PetState, set[PetState]] = {
|
|
PetState.IDLE: {PetState.LISTENING, PetState.TALKING, PetState.ERROR},
|
|
PetState.LISTENING: {PetState.THINKING, PetState.IDLE, PetState.ERROR},
|
|
PetState.THINKING: {PetState.TALKING, PetState.IDLE, PetState.ERROR},
|
|
PetState.TALKING: {PetState.IDLE, PetState.ERROR},
|
|
PetState.ERROR: {PetState.IDLE},
|
|
}
|
|
|
|
|
|
class InvalidTransition(Exception):
|
|
pass
|
|
|
|
|
|
class PetStateMachine:
|
|
def __init__(self, on_change: Optional[Callable[[PetState, PetState], None]] = None):
|
|
self._state = PetState.IDLE
|
|
self._on_change = on_change
|
|
|
|
@property
|
|
def state(self) -> PetState:
|
|
return self._state
|
|
|
|
def transition(self, new_state: PetState) -> None:
|
|
if new_state == self._state:
|
|
return
|
|
allowed = _TRANSITIONS.get(self._state, set())
|
|
if new_state not in allowed:
|
|
raise InvalidTransition(f"{self._state} -> {new_state} is not allowed")
|
|
old_state = self._state
|
|
self._state = new_state
|
|
if self._on_change is not None:
|
|
self._on_change(old_state, new_state)
|
|
|
|
def force(self, new_state: PetState) -> None:
|
|
"""Bypass the transition table — used only for recovering to IDLE
|
|
from an unexpected/edge-case state (e.g. after an exception mid
|
|
pipeline). Prefer transition() everywhere else."""
|
|
old_state = self._state
|
|
if new_state == old_state:
|
|
return
|
|
self._state = new_state
|
|
if self._on_change is not None:
|
|
self._on_change(old_state, new_state)
|