imonlinux
7f8bdefa39
event_bus: EventBus loop: asyncio.AbstractEventLoop This provides a globally accessible instance of the event bus and the main asyncio event loop to any component that needs them.
100 lines
2.9 KiB
Python
100 lines
2.9 KiB
Python
"""Shared models."""
|
|
|
|
import asyncio
|
|
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, Union
|
|
|
|
from .event_bus import EventBus
|
|
|
|
if TYPE_CHECKING:
|
|
from .entity import ESPHomeEntity, MediaPlayerEntity
|
|
from .microwakeword import MicroWakeWord
|
|
from .mpv_player import MpvMediaPlayer
|
|
from .openwakeword import OpenWakeWord
|
|
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]
|
|
config_path: Path
|
|
|
|
def load(
|
|
self, libtensorflowlite_c_path: Path
|
|
) -> "Union[MicroWakeWord, OpenWakeWord]":
|
|
if self.type == WakeWordType.MICRO_WAKE_WORD:
|
|
from .microwakeword import MicroWakeWord
|
|
|
|
return MicroWakeWord.from_config(
|
|
config_path=self.config_path,
|
|
libtensorflowlite_c_path=libtensorflowlite_c_path,
|
|
)
|
|
|
|
if self.type == WakeWordType.OPEN_WAKE_WORD:
|
|
from .openwakeword import OpenWakeWord
|
|
|
|
return OpenWakeWord.from_config(
|
|
config_path=self.config_path,
|
|
libtensorflowlite_c_path=libtensorflowlite_c_path,
|
|
)
|
|
|
|
raise ValueError(f"Unexpected wake word type: {self.type}")
|
|
|
|
|
|
@dataclass
|
|
class Preferences:
|
|
active_wake_words: List[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class ServerState:
|
|
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]]"
|
|
stop_word: "MicroWakeWord"
|
|
music_player: "MpvMediaPlayer"
|
|
tts_player: "MpvMediaPlayer"
|
|
wakeup_sound: str
|
|
timer_finished_sound: str
|
|
preferences: Preferences
|
|
preferences_path: Path
|
|
libtensorflowlite_c_path: Path
|
|
event_bus: EventBus
|
|
loop: asyncio.AbstractEventLoop
|
|
|
|
# openWakeWord
|
|
oww_melspectrogram_path: Path
|
|
oww_embedding_path: Path
|
|
|
|
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
|
|
)
|