Save preferences to JSON file

linux_voice_assistant/__main__.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-
 import argparse
 import asyncio
 import json
@@ -12,7 +11,7 @@ from typing import Dict
 import sounddevice as sd
 
 from .microwakeword import MicroWakeWord
-from .models import AvailableWakeWord, ServerState
+from .models import AvailableWakeWord, Preferences, ServerState
 from .mpv_player import MpvMediaPlayer
 from .satellite import VoiceSatelliteProtocol
 from .util import get_mac, is_arm
@@ -57,6 +56,9 @@ async def main() -> None:
     parser.add_argument(
         "--timer-finished-sound", default=str(_SOUNDS_DIR / "timer_finished.flac")
     )
+    #
+    parser.add_argument("--preferences-file", default=_REPO_DIR / "preferences.json")
+    #
     parser.add_argument(
         "--host",
         default="0.0.0.0",
@@ -93,11 +95,28 @@ async def main() -> None:
 
     _LOGGER.debug("Available wake words: %s", list(sorted(available_wake_words.keys())))
 
+    # Load preferences
+    preferences_path = Path(args.preferences_file)
+    if preferences_path.exists():
+        _LOGGER.debug("Loading preferences: %s", preferences_path)
+        with open(preferences_path, "r", encoding="utf-8") as preferences_file:
+            preferences_dict = json.load(preferences_file)
+            preferences = Preferences(**preferences_dict)
+    else:
+        preferences = Preferences()
+
     libtensorflowlite_c_path = _LIB_DIR / "libtensorflowlite_c.so"
     _LOGGER.debug("libtensorflowlite_c path: %s", libtensorflowlite_c_path)
 
     # Load wake/stop models
     wake_config_path = wake_word_dir / f"{args.wake_model}.json"
+    if preferences.active_wake_words:
+        wake_word_id = preferences.active_wake_words[0]
+        maybe_wake_config_path = wake_word_dir / f"{wake_word_id}.json"
+        if maybe_wake_config_path.exists():
+            # Override with last set
+            wake_config_path = maybe_wake_config_path
+
     _LOGGER.debug("Loading wake model: %s", wake_config_path)
     wake_model = MicroWakeWord.from_config(wake_config_path, libtensorflowlite_c_path)
 
@@ -117,6 +136,8 @@ async def main() -> None:
         tts_player=MpvMediaPlayer(device=args.audio_output_device),
         wakeup_sound=args.wakeup_sound,
         timer_finished_sound=args.timer_finished_sound,
+        preferences=preferences,
+        preferences_path=preferences_path,
     )
 
     process_audio_thread = threading.Thread(

linux_voice_assistant/models.py

@@ -1,6 +1,8 @@
 """Shared models."""
 
-from dataclasses import dataclass
+import json
+import logging
+from dataclasses import asdict, dataclass, field
 from pathlib import Path
 from queue import Queue
 from typing import TYPE_CHECKING, Dict, List, Optional
@@ -11,6 +13,8 @@ if TYPE_CHECKING:
     from .mpv_player import MpvMediaPlayer
     from .satellite import VoiceSatelliteProtocol
 
+_LOGGER = logging.getLogger(__name__)
+
 
 @dataclass
 class AvailableWakeWord:
@@ -20,6 +24,11 @@ class AvailableWakeWord:
     config_path: Path
 
 
+@dataclass
+class Preferences:
+    active_wake_words: List[str] = field(default_factory=list)
+
+
 @dataclass
 class ServerState:
     name: str
@@ -33,5 +42,16 @@ class ServerState:
     tts_player: "MpvMediaPlayer"
     wakeup_sound: str
     timer_finished_sound: str
+    preferences: Preferences
+    preferences_path: Path
     media_player_entity: "Optional[MediaPlayerEntity]" = None
     satellite: "Optional[VoiceSatelliteProtocol]" = None
+
+    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
+            )

linux_voice_assistant/satellite.py

@@ -3,7 +3,7 @@
 import logging
 import time
 from collections.abc import Iterable
-from typing import Dict, Optional
+from typing import Dict, Optional, Set
 
 # pylint: disable=no-name-in-module
 from aioesphomeapi.api_pb2 import (  # type: ignore[attr-defined]
@@ -181,9 +181,12 @@ class VoiceSatelliteProtocol(APIServer):
             _LOGGER.info("Connected to Home Assistant")
         elif isinstance(msg, VoiceAssistantSetConfiguration):
             # TODO: support multiple wake words
+            active_wake_words: Set[str] = set()
+
             for wake_word_id in msg.active_wake_words:
                 if wake_word_id == self.state.wake_word.id:
                     # Already active
+                    active_wake_words.add(wake_word_id)
                     break
 
                 model_info = self.state.available_wake_words.get(wake_word_id)
@@ -197,8 +200,12 @@ class VoiceSatelliteProtocol(APIServer):
                 )
 
                 _LOGGER.info("Wake word set: %s", self.state.wake_word.wake_word)
+                active_wake_words.add(wake_word_id)
                 break
 
+            self.state.preferences.active_wake_words = list(active_wake_words)
+            self.state.save_preferences()
+
     def handle_audio(self, audio_chunk: bytes) -> None:
 
         if not self._is_streaming_audio: