From 2c937e3cb34c3001ea9575dcdf1b1874a70d8a10 Mon Sep 17 00:00:00 2001 From: Michael Hansen Date: Mon, 15 Sep 2025 11:06:09 -0500 Subject: [PATCH] Save preferences to JSON file --- .gitignore | 1 + CHANGELOG.md | 5 +++++ linux_voice_assistant/__main__.py | 25 +++++++++++++++++++++++-- linux_voice_assistant/models.py | 22 +++++++++++++++++++++- linux_voice_assistant/satellite.py | 9 ++++++++- 5 files changed, 58 insertions(+), 4 deletions(-) diff --git a/.gitignore b/.gitignore index a8f5b64..3c1e29b 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,4 @@ venv/ *.egg-info/ dist/ +preferences.json diff --git a/CHANGELOG.md b/CHANGELOG.md index c6dc197..588bdb7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # Changelog +## Unreleased + +- Save active wake words to preferences JSON file +- Refactor main into separate files + ## 1.0.0 - Initial release diff --git a/linux_voice_assistant/__main__.py b/linux_voice_assistant/__main__.py index badf80e..61c3bd6 100644 --- a/linux_voice_assistant/__main__.py +++ b/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( diff --git a/linux_voice_assistant/models.py b/linux_voice_assistant/models.py index 1cc48a4..4a47a7c 100644 --- a/linux_voice_assistant/models.py +++ b/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 + ) diff --git a/linux_voice_assistant/satellite.py b/linux_voice_assistant/satellite.py index 47fec95..ebee376 100644 --- a/linux_voice_assistant/satellite.py +++ b/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: