diff --git a/linux_voice_assistant/models.py b/linux_voice_assistant/models.py index 2b585d0..ca2feb5 100644 --- a/linux_voice_assistant/models.py +++ b/linux_voice_assistant/models.py @@ -1,385 +1,101 @@ -#!/usr/bin/env python3 -import argparse +"""Shared models.""" + import asyncio import json import logging -import sys -import threading -import time +from dataclasses import asdict, dataclass, field +from enum import Enum from pathlib import Path -from typing import Dict, List, Optional, Set, Union +from typing import TYPE_CHECKING, Dict, List, Optional, Set, Union -import numpy as np -import soundcard as sc -from pymicro_wakeword import MicroWakeWord, MicroWakeWordFeatures -from pyopen_wakeword import OpenWakeWord, OpenWakeWordFeatures +from .event_bus import EventBus -from .config import Config, load_config_from_json -from .event_bus import EventBus, EventHandler, subscribe -from .led_controller import LedController -from .mqtt_controller import MqttController -from .models import AvailableWakeWord, Preferences, ServerState, WakeWordType -from .mpv_player import MpvMediaPlayer -from .satellite import VoiceSatelliteProtocol -from .util import get_mac -from .zeroconf import HomeAssistantZeroconf +if TYPE_CHECKING: + from pymicro_wakeword import MicroWakeWord + from pyopen_wakeword import OpenWakeWord + from .entity import ESPHomeEntity, MediaPlayerEntity + from .mpv_player import MpvMediaPlayer + from .satellite import VoiceSatelliteProtocol _LOGGER = logging.getLogger(__name__) -_MODULE_DIR = Path(__file__).parent -_REPO_DIR = _MODULE_DIR.parent -# ----------------------------------------------------------------------------- -class MicMuteHandler(EventHandler): - """Event handler for mic mute switch.""" - def __init__(self, state, mqtt_controller: Optional[MqttController]): - super().__init__(state) - self.mqtt_controller = mqtt_controller +class SatelliteState(str, Enum): + """Voice satellite state.""" + STARTING = "starting" + IDLE = "idle" + LISTENING = "listening" + THINKING = "thinking" + RESPONDING = "responding" + ERROR = "error" - @subscribe - def set_mic_mute(self, data: dict): - is_muted = data.get("state", False) - if self.state.mic_muted != is_muted: - self.state.mic_muted = is_muted - _LOGGER.debug("Mic muted = %s", is_muted) + +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 - if self.mqtt_controller: - self.mqtt_controller.publish_mute_state(is_muted) - - if is_muted: - self.state.event_bus.publish("mic_muted") - else: - self.state.event_bus.publish("mic_unmuted") + 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}") -async def main() -> None: - parser = argparse.ArgumentParser() - parser.add_argument( - "-c", - "--config", - type=Path, - required=False, - default=_MODULE_DIR / "config.json", - help="Path to configuration.json file (default: linux_voice_assistant/config.json)", - ) - parser.add_argument( - "--debug", - action="store_true", - help="Enable debug logging (overrides config file)", - ) - parser.add_argument( - "--list-input-devices", - action="store_true", - help="List audio input devices and exit", - ) - parser.add_argument( - "--list-output-devices", - action="store_true", - help="List audio output devices and exit", - ) - args = parser.parse_args() - - # --- Handle List Devices --- - if args.list_input_devices: - print("Input devices") - print("=" * 13) - for idx, mic in enumerate(sc.all_microphones(include_loopback=False)): - print(f"[{idx}]", mic.name) - return - - if args.list_output_devices: - player = MpvMediaPlayer(loop=None) - print("Output devices") - print("=" * 14) - try: - for speaker in player.player.audio_device_list: - print(speaker["name"] + ":", speaker["description"]) - except Exception as e: - _LOGGER.error("Failed to list output devices: %s", e) - return - - # --- Load Configuration --- - config_path = args.config - - if not config_path.is_absolute(): - config_path = _REPO_DIR / config_path - - config = load_config_from_json(config_path) - - if args.debug: - config.app.debug = True - - logging.basicConfig(level=logging.DEBUG if config.app.debug else logging.INFO) - _LOGGER.info("Loading configuration from: %s", config_path) - _LOGGER.debug("Configuration loaded: %s", config) - - loop = asyncio.get_running_loop() - event_bus = EventBus() - - # --- Create Download Dir --- - download_dir = _REPO_DIR / config.wake_word.download_dir - download_dir.mkdir(parents=True, exist_ok=True) - _LOGGER.debug("Download directory: %s", download_dir) - - # --- Load Preferences --- - preferences_path = _REPO_DIR / config.app.preferences_file - if preferences_path.exists(): - with open(preferences_path, "r", encoding="utf-8") as preferences_file: - preferences_dict = json.load(preferences_file) - preferences = Preferences(**preferences_dict) - else: - preferences = Preferences() - - preferences.num_leds = getattr(preferences, 'num_leds', config.led.num_leds) - - # --- Load Wake Words --- - available_wake_words: Dict[str, AvailableWakeWord] = {} - - if not config.wake_word.directories: - config.wake_word.directories = ["wakewords", "wakewords/openWakeWord"] - - # Add download directory to search path - wake_word_dirs = [_REPO_DIR / d for d in config.wake_word.directories] - wake_word_dirs.append(download_dir / "external_wake_words") # <-- ADDED - - for wake_word_dir in wake_word_dirs: - if not wake_word_dir.exists(): - continue - - for model_config_path in wake_word_dir.glob("*.json"): - model_id = model_config_path.stem - if model_id == config.wake_word.stop_model: - continue - with open(model_config_path, "r", encoding="utf-8") as model_config_file: - model_config = json.load(model_config_file) - model_type = WakeWordType(model_config["type"]) - - if model_type == WakeWordType.OPEN_WAKE_WORD: - wake_word_path = model_config_path.parent / model_config["model"] - else: - wake_word_path = model_config_path - - available_wake_words[model_id] = AvailableWakeWord( - id=model_id, - type=model_type, - wake_word=model_config["wake_word"], - trained_languages=model_config.get("trained_languages", []), - wake_word_path=wake_word_path, - ) - - _LOGGER.debug("Available wake words: %s", list(sorted(available_wake_words.keys()))) - - # --- Load Active Wake Word Models --- - active_wake_words: Set[str] = set() - wake_models: Dict[str, Union[MicroWakeWord, OpenWakeWord]] = {} - if preferences.active_wake_words: - for wake_word_id in preferences.active_wake_words: - wake_word = available_wake_words.get(wake_word_id) - if wake_word is None: continue - _LOGGER.debug("Loading wake model: %s", wake_word_id) - wake_models[wake_word_id] = wake_word.load() - active_wake_words.add(wake_word_id) - - if not wake_models: - wake_word_id = config.wake_word.model - wake_word = available_wake_words.get(wake_word_id) - if wake_word: - _LOGGER.debug("Loading wake model: %s", wake_word_id) - wake_models[wake_word_id] = wake_word.load() - active_wake_words.add(wake_word_id) - - # --- Load Stop Model --- - stop_model: Optional[MicroWakeWord] = None - for ww_dir_str in config.wake_word.directories: - wake_word_dir = _REPO_DIR / ww_dir_str - stop_config_path = wake_word_dir / f"{config.wake_word.stop_model}.json" - if not stop_config_path.exists(): continue - _LOGGER.debug("Loading stop model: %s", stop_config_path) - stop_model = MicroWakeWord.from_config(stop_config_path) - break - assert stop_model is not None - - # --- Resolve Microphone --- - if config.audio.input_device is not None: - try: - input_device_idx = int(config.audio.input_device) - mic = sc.all_microphones(include_loopback=False)[input_device_idx] - except (ValueError, IndexError): - mic = sc.get_microphone(config.audio.input_device) - else: - mic = sc.default_microphone() - - if mic is None: - _LOGGER.critical("No microphone found.") - sys.exit(1) - - _LOGGER.info("Using audio input device: %s", mic.name) - - # --- Initialize Media Players --- - music_player = MpvMediaPlayer( - loop=loop, - device=config.audio.output_device, - initial_volume=preferences.volume_level - ) - tts_player = MpvMediaPlayer( - loop=loop, - device=config.audio.output_device, - initial_volume=preferences.volume_level - ) - - # --- Create Global Server State --- - state = ServerState( - name=config.app.name, - mac_address=get_mac(), - entities=[], - available_wake_words=available_wake_words, - wake_words=wake_models, - active_wake_words=active_wake_words, - stop_word=stop_model, - music_player=music_player, - tts_player=tts_player, - wakeup_sound=str(_REPO_DIR / config.app.wakeup_sound), - timer_finished_sound=str(_REPO_DIR / config.app.timer_finished_sound), - preferences=preferences, - preferences_path=preferences_path, - event_bus=event_bus, - loop=loop, - download_dir=download_dir, # <-- ADDED - refractory_seconds=config.wake_word.refractory_seconds, - ) - - # --- Initialize Controllers --- - led_controller = LedController( - state, - led_type=config.led.led_type, - interface=config.led.interface, - clock_pin=config.led.clock_pin, - data_pin=config.led.data_pin, - num_leds=state.preferences.num_leds, - ) - - mqtt_controller: Optional[MqttController] = None - if config.mqtt.enabled: - mqtt_controller = MqttController( - state=state, - host=config.mqtt.host, - port=config.mqtt.port, - username=config.mqtt.username, - password=config.mqtt.password - ) - mqtt_controller.start() - - mic_mute_handler = MicMuteHandler(state, mqtt_controller) - - # --- Start Audio Processing Thread --- - process_audio_thread = threading.Thread( - target=process_audio, - args=(state, mic, config.audio.input_block_size), - daemon=True - ) - process_audio_thread.start() - - # --- Start ESPHome Server --- - server = await loop.create_server( - lambda: VoiceSatelliteProtocol(state), - host=config.esphome.host, - port=config.esphome.port - ) - discovery = HomeAssistantZeroconf(port=config.esphome.port, name=config.app.name) - await discovery.register_server() - - # --- Run Forever --- - try: - async with server: - _LOGGER.info("Server started (host=%s, port=%s)", config.esphome.host, config.esphome.port) - await server.serve_forever() - except KeyboardInterrupt: - pass - finally: - if mqtt_controller: - mqtt_controller.stop() - - state.mic_muted = True - process_audio_thread.join() - - _LOGGER.debug("Server stopped") - -def process_audio(state: ServerState, mic, block_size: int): - """Process audio chunks from the microphone in a separate thread.""" - wake_words: List[Union[MicroWakeWord, OpenWakeWord]] = [] - micro_features: Optional[MicroWakeWordFeatures] = None - micro_inputs: List[np.ndarray] = [] - oww_features: Optional[OpenWakeWordFeatures] = None - oww_inputs: List[np.ndarray] = [] - has_oww = False - last_active: Optional[float] = None - try: - _LOGGER.debug("Opening audio input device: %s", mic.name) - with mic.recorder(samplerate=16000, channels=1, blocksize=block_size) as mic_in: - while True: - if state.mic_muted: - time.sleep(0.1) - continue - - audio_chunk_array = mic_in.record(block_size).reshape(-1) - audio_chunk = ( - (np.clip(audio_chunk_array, -1.0, 1.0) * 32767.0) - .astype(" 0.5 for oi in oww_inputs for p in wake_word.process_streaming(oi)): activated = True - - if activated: - now = time.monotonic() - if (last_active is None) or ((now - last_active) > state.refractory_seconds): - state.satellite.wakeup(wake_word) - last_active = now - - stopped = False - for micro_input in micro_inputs: - if state.stop_word.process_streaming(micro_input): - stopped = True - - if stopped and (state.stop_word.id in state.active_wake_words): - state.satellite.stop() - except Exception: _LOGGER.exception("Unexpected error handling audio") - except Exception as e: - _LOGGER.critical("Error in audio processing thread: %s", e) - state.loop.call_soon_threadsafe(state.loop.stop) +@dataclass +class Preferences: + active_wake_words: List[str] = field(default_factory=list) + volume_level: float = 1.0 + num_leds: int = 3 -if __name__ == "__main__": - print("--- __main__ block executing ---") - asyncio.run(main()) +@dataclass +class ServerState: + # --- Fields WITHOUT default values --- + name: str + mac_address: str + 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: "MpvMediaPlayer" + tts_player: "MpvMediaPlayer" + wakeup_sound: str + timer_finished_sound: str + preferences: Preferences + preferences_path: Path + event_bus: EventBus + loop: asyncio.AbstractEventLoop + download_dir: Path # <-- ADDED + + # --- Fields WITH default values --- + satellite: "Optional[VoiceSatelliteProtocol]" = None + wake_words_changed: bool = False + refractory_seconds: float = 2.0 + mic_muted: bool = False + + 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 + )