File size: 2,851 Bytes
b8cfa60 6b00c9d b8cfa60 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | """Shared models for Reachy Mini Voice Assistant."""
import json
import logging
from dataclasses import asdict, dataclass, field
from enum import Enum
from pathlib import Path
from queue import Queue
from typing import TYPE_CHECKING, Dict, List, Optional, Set, Union
if TYPE_CHECKING:
from pymicro_wakeword import MicroWakeWord
from pyopen_wakeword import OpenWakeWord
from .entity import ESPHomeEntity, MediaPlayerEntity
from .audio_player import AudioPlayer
from .satellite import VoiceSatelliteProtocol
_LOGGER = logging.getLogger(__name__)
class WakeWordType(str, Enum):
MICRO_WAKE_WORD = "micro"
OPEN_WAKE_WORD = "openWakeWord"
@dataclass
class AvailableWakeWord:
id: str
type: WakeWordType
wake_word: str
trained_languages: List[str]
wake_word_path: Path
def load(self) -> "Union[MicroWakeWord, OpenWakeWord]":
if self.type == WakeWordType.MICRO_WAKE_WORD:
from pymicro_wakeword import MicroWakeWord
return MicroWakeWord.from_config(config_path=self.wake_word_path)
if self.type == WakeWordType.OPEN_WAKE_WORD:
from pyopen_wakeword import OpenWakeWord
oww_model = OpenWakeWord.from_model(model_path=self.wake_word_path)
setattr(oww_model, "wake_word", self.wake_word)
return oww_model
raise ValueError(f"Unexpected wake word type: {self.type}")
@dataclass
class Preferences:
active_wake_words: List[str] = field(default_factory=list)
@dataclass
class ServerState:
"""Global server state."""
name: str
mac_address: str
audio_queue: "Queue[Optional[bytes]]"
entities: "List[ESPHomeEntity]"
available_wake_words: "Dict[str, AvailableWakeWord]"
wake_words: "Dict[str, Union[MicroWakeWord, OpenWakeWord]]"
active_wake_words: Set[str]
stop_word: "MicroWakeWord"
music_player: "AudioPlayer"
tts_player: "AudioPlayer"
wakeup_sound: str
timer_finished_sound: str
preferences: Preferences
preferences_path: Path
download_dir: Path
# Reachy Mini specific
reachy_mini: Optional[object] = None
motion_enabled: bool = True
motion: Optional[object] = None # ReachyMiniMotion instance
media_player_entity: "Optional[MediaPlayerEntity]" = None
satellite: "Optional[VoiceSatelliteProtocol]" = None
wake_words_changed: bool = False
refractory_seconds: float = 2.0
def save_preferences(self) -> None:
"""Save preferences as JSON."""
_LOGGER.debug("Saving preferences: %s", self.preferences_path)
self.preferences_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.preferences_path, "w", encoding="utf-8") as preferences_file:
json.dump(
asdict(self.preferences), preferences_file, ensure_ascii=False, indent=4
)
|