Support openWakeWord

linux_voice_assistant/models.py

@@ -3,26 +3,55 @@
 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
+from typing import TYPE_CHECKING, Dict, List, Optional, Union
 
 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:
@@ -36,7 +65,7 @@ class ServerState:
     audio_queue: "Queue[Optional[bytes]]"
     entities: "List[ESPHomeEntity]"
     available_wake_words: "Dict[str, AvailableWakeWord]"
-    wake_words: "Dict[str, MicroWakeWord]"
+    wake_words: "Dict[str, Union[MicroWakeWord, OpenWakeWord]]"
     stop_word: "MicroWakeWord"
     music_player: "MpvMediaPlayer"
     tts_player: "MpvMediaPlayer"
@@ -45,9 +74,15 @@ class ServerState:
     preferences: Preferences
     preferences_path: Path
     libtensorflowlite_c_path: Path
+
+    # 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."""