| """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: Optional[object] = None |
| motion_enabled: bool = True |
| motion: Optional[object] = None |
|
|
| 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 |
| ) |
|
|
