diff --git a/linux_voice_assistant/__main__.py b/linux_voice_assistant/__main__.py index 2b585d0..764e2be 100644 --- a/linux_voice_assistant/__main__.py +++ b/linux_voice_assistant/__main__.py @@ -6,15 +6,16 @@ import logging import sys import threading import time +from dataclasses import dataclass from pathlib import Path -from typing import Dict, List, Optional, Set, Union +from typing import Dict, List, Optional, Set, Tuple, Union import numpy as np import soundcard as sc from pymicro_wakeword import MicroWakeWord, MicroWakeWordFeatures from pyopen_wakeword import OpenWakeWord, OpenWakeWordFeatures -from .config import Config, load_config_from_json +from .config import Config, LedConfig, MqttConfig, load_config_from_json from .event_bus import EventBus, EventHandler, subscribe from .led_controller import LedController from .mqtt_controller import MqttController @@ -28,16 +29,47 @@ _LOGGER = logging.getLogger(__name__) _MODULE_DIR = Path(__file__).parent _REPO_DIR = _MODULE_DIR.parent +# ----------------------------------------------------------------------------- +# Helper dataclasses +# ----------------------------------------------------------------------------- + +@dataclass +class WakeWordData: + """Dataclass to hold all wake word models and settings.""" + available: Dict[str, AvailableWakeWord] + models: Dict[str, Union[MicroWakeWord, OpenWakeWord]] + active: Set[str] + stop_model: MicroWakeWord + +@dataclass +class MediaPlayers: + """Dataclass to hold media player instances.""" + music: MpvMediaPlayer + tts: MpvMediaPlayer + +# ----------------------------------------------------------------------------- +# Mic Mute Handler # ----------------------------------------------------------------------------- class MicMuteHandler(EventHandler): - """Event handler for mic mute switch.""" - def __init__(self, state, mqtt_controller: Optional[MqttController]): - super().__init__(state) + """ + Manages the mic_muted state and saves preferences. + This is the only controller that writes to ServerState. + """ + def __init__( + self, + event_bus: EventBus, + state: ServerState, + mqtt_controller: Optional[MqttController], + ): + super().__init__(event_bus) + self.state = state self.mqtt_controller = mqtt_controller + self._subscribe_all_methods() @subscribe def set_mic_mute(self, data: dict): + """Event handler for mic mute commands.""" is_muted = data.get("state", False) if self.state.mic_muted != is_muted: self.state.mic_muted = is_muted @@ -47,177 +79,224 @@ class MicMuteHandler(EventHandler): self.mqtt_controller.publish_mute_state(is_muted) if is_muted: - self.state.event_bus.publish("mic_muted") + self.event_bus.publish("mic_muted") else: - self.state.event_bus.publish("mic_unmuted") + self.event_bus.publish("mic_unmuted") + + @subscribe + def set_num_leds(self, data: dict): + """Event handler to save num_leds to preferences.""" + num_leds = data.get("num_leds") + if (num_leds is not None) and (self.state.preferences.num_leds != num_leds): + self.state.preferences.num_leds = num_leds + self.state.save_preferences() # ----------------------------------------------------------------------------- - +# Main Application +# ----------------------------------------------------------------------------- async def main() -> None: + # --- 1. Load Basics --- + config, loop, event_bus = _init_basics() + + # --- 2. Load Preferences --- + preferences = _load_preferences(config) + + # --- 3. Find Microphone --- + mic = _get_microphone(config) + + # --- 4. Load Wake Words --- + wake_word_data = _load_wake_words(config, preferences) + + # --- 5. Initialize Media Players --- + media_players = _init_media_players(loop, config, preferences) + + # --- 6. Create Server State --- + state = _create_server_state( + config, loop, event_bus, preferences, + wake_word_data, media_players + ) + + # --- 7. Initialize Controllers --- + _init_controllers(loop, event_bus, state, config, preferences) + + # --- 8. Start Audio Thread --- + audio_thread = threading.Thread( + target=_process_audio, + args=(state, mic, config.audio.input_block_size), + daemon=True + ) + audio_thread.start() + + # --- 9. Run Server --- + try: + await _run_server(state, config) + finally: + # --- 10. Cleanup --- + _LOGGER.debug("Shutting down...") + state.shutdown = True # Signal audio thread to stop + audio_thread.join() + + if hasattr(state, "mqtt_controller") and state.mqtt_controller: + _LOGGER.debug("Stopping MQTT controller...") + state.mqtt_controller.stop() + +# ----------------------------------------------------------------------------- +# Helper Functions +# ----------------------------------------------------------------------------- + +def _init_basics() -> Tuple[Config, asyncio.AbstractEventLoop, EventBus]: + """Loads config, sets up logging, and creates loop/event bus.""" parser = argparse.ArgumentParser() parser.add_argument( - "-c", - "--config", - type=Path, - required=False, + "-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", + help="Path to configuration.json file" ) + parser.add_argument("--debug", action="store_true", help="Enable debug logging") + parser.add_argument("--list-input-devices", action="store_true", help="List audio input devices") + parser.add_argument("--list-output-devices", action="store_true", help="List audio output devices") args = parser.parse_args() - # --- Handle List Devices --- + # --- List devices and exit --- if args.list_input_devices: - print("Input devices") - print("=" * 13) + print("Input devices\n" + "=" * 13) for idx, mic in enumerate(sc.all_microphones(include_loopback=False)): print(f"[{idx}]", mic.name) - return + sys.exit(0) if args.list_output_devices: - player = MpvMediaPlayer(loop=None) - print("Output devices") - print("=" * 14) + print("Output devices\n" + "=" * 14) try: + player = MpvMediaPlayer(loop=None) 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 + sys.exit(1) - # --- Load Configuration --- + # --- Load config --- 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 + # --- Setup logging --- 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() + + return config, loop, event_bus - # --- 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 --- +def _load_preferences(config: Config) -> Preferences: + """Loads preferences.json file.""" 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) + with open(preferences_path, "r", encoding="utf-8") as f: + preferences_dict = json.load(f) preferences = Preferences(**preferences_dict) else: preferences = Preferences() preferences.num_leds = getattr(preferences, 'num_leds', config.led.num_leds) + return preferences + +def _get_microphone(config: Config): + """Finds and returns the microphone specified in the config.""" + mic = None + 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, include_loopback=False) + else: + mic = sc.default_microphone() - # --- Load Wake Words --- - available_wake_words: Dict[str, AvailableWakeWord] = {} + if mic is None: + _LOGGER.critical("No microphone found. Use --list-input-devices to see options.") + sys.exit(1) + + _LOGGER.info("Using audio input device: %s", mic.name) + return mic + +def _load_wake_words(config: Config, preferences: Preferences) -> WakeWordData: + """Loads all available and active wake word models.""" if not config.wake_word.directories: config.wake_word.directories = ["wakewords", "wakewords/openWakeWord"] + + download_dir = _REPO_DIR / config.wake_word.download_dir + download_dir.mkdir(parents=True, exist_ok=True) + _LOGGER.debug("Download directory: %s", download_dir) - # 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 + wake_word_dirs.append(download_dir / "external_wake_words") + available: Dict[str, AvailableWakeWord] = {} 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 + for config_path in wake_word_dir.glob("*.json"): + model_id = 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) + with open(config_path, "r", encoding="utf-8") as f: + model_config = json.load(f) 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 + wake_word_path = ( + config_path.parent / model_config["model"] + if model_type == WakeWordType.OPEN_WAKE_WORD + else config_path + ) - available_wake_words[model_id] = AvailableWakeWord( + available[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.keys()))) - _LOGGER.debug("Available wake words: %s", list(sorted(available_wake_words.keys()))) + active: Set[str] = set() + models: Dict[str, Union[MicroWakeWord, OpenWakeWord]] = {} - # --- 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) + for ww_id in preferences.active_wake_words: + if ww_id in available: + _LOGGER.debug("Loading wake model (from preferences): %s", ww_id) + models[ww_id] = available[ww_id].load() + active.add(ww_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 --- + if not models: + ww_id = config.wake_word.model + if ww_id in available: + _LOGGER.debug("Loading wake model (from config): %s", ww_id) + models[ww_id] = available[ww_id].load() + active.add(ww_id) + 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 + stop_config_path = _REPO_DIR / ww_dir_str / f"{config.wake_word.stop_model}.json" + if stop_config_path.exists(): + _LOGGER.debug("Loading stop model: %s", stop_config_path) + stop_model = MicroWakeWord.from_config(stop_config_path) + break + assert stop_model is not None, "Stop model not found" - # --- 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) + return WakeWordData(available, models, active, stop_model) - # --- Initialize Media Players --- +def _init_media_players(loop: asyncio.AbstractEventLoop, config: Config, preferences: Preferences) -> MediaPlayers: + """Initializes the music and TTS media players.""" + _LOGGER.debug("Initializing media players") music_player = MpvMediaPlayer( loop=loop, device=config.audio.output_device, @@ -228,61 +307,78 @@ async def main() -> None: device=config.audio.output_device, initial_volume=preferences.volume_level ) - - # --- Create Global Server State --- - state = ServerState( + return MediaPlayers(music=music_player, tts=tts_player) + +def _create_server_state( + config: Config, + loop: asyncio.AbstractEventLoop, + event_bus: EventBus, + preferences: Preferences, + wake_word_data: WakeWordData, + media_players: MediaPlayers, +) -> ServerState: + """Creates the global ServerState object.""" + _LOGGER.debug("Creating server state") + return ServerState( name=config.app.name, mac_address=get_mac(), + event_bus=event_bus, + loop=loop, 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, + music_player=media_players.music, + tts_player=media_players.tts, + available_wake_words=wake_word_data.available, + wake_words=wake_word_data.models, + active_wake_words=wake_word_data.active, + stop_word=wake_word_data.stop_model, 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 + preferences_path=_REPO_DIR / config.app.preferences_file, + download_dir=_REPO_DIR / config.wake_word.download_dir, refractory_seconds=config.wake_word.refractory_seconds, ) + +def _init_controllers( + loop: asyncio.AbstractEventLoop, + event_bus: EventBus, + state: ServerState, + config: Config, + preferences: Preferences, +): + """Initializes all decoupled controllers.""" + _LOGGER.debug("Initializing controllers") - # --- 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, + loop=loop, + event_bus=event_bus, + config=config.led, + preferences=preferences, ) - + 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 + loop=loop, + event_bus=event_bus, + config=config.mqtt, + app_name=config.app.name, + mac_address=state.mac_address, + preferences=preferences, ) + setattr(state, "mqtt_controller", mqtt_controller) 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 + + mic_mute_handler = MicMuteHandler( + event_bus=event_bus, + state=state, + mqtt_controller=mqtt_controller, ) - process_audio_thread.start() - - # --- Start ESPHome Server --- - server = await loop.create_server( + +async def _run_server(state: ServerState, config: Config): + """Starts the ESPHome server and ZeroConf discovery.""" + _LOGGER.debug("Starting ESPHome server") + server = await state.loop.create_server( lambda: VoiceSatelliteProtocol(state), host=config.esphome.host, port=config.esphome.port @@ -290,96 +386,121 @@ async def main() -> None: 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() + async with server: + _LOGGER.info("Server started (host=%s, port=%s)", config.esphome.host, config.esphome.port) + await server.serve_forever() - _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] = [] +def _process_audio(state: ServerState, mic, block_size: int): + """Main audio processing loop. Runs in a separate thread.""" + + micro_features: MicroWakeWordFeatures = MicroWakeWordFeatures() oww_features: Optional[OpenWakeWordFeatures] = None + + wake_words: List[Union[MicroWakeWord, OpenWakeWord]] = [] + micro_inputs: List[np.ndarray] = [] 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: + while True: + # --- THIS IS THE FIX --- + # Wait while muted. This loop will spin until unmuted. + if state.mic_muted: + _LOGGER.debug("Audio thread muted.") + while state.mic_muted: + if state.shutdown: + _LOGGER.debug("Shutdown signal received, stopping audio thread.") + return # Exit thread 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 (not wake_words) or (state.wake_words_changed): + state.wake_words_changed = False + wake_words = [ + ww for ww in state.wake_words.values() + if ww.id in state.active_wake_words + ] + has_oww = any(isinstance(ww, OpenWakeWord) for ww in wake_words) - 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 + if has_oww and (oww_features is None): + _LOGGER.debug("Initializing OpenWakeWord features...") + oww_features = OpenWakeWordFeatures.from_builtin() - stopped = False - for micro_input in micro_inputs: - if state.stop_word.process_streaming(micro_input): - stopped = True + try: + micro_inputs.clear(); micro_inputs.extend(micro_features.process_streaming(audio_chunk)) + + if has_oww: + assert oww_features is not None + oww_inputs.clear(); oww_inputs.extend(oww_features.process_streaming(audio_chunk)) + + for wake_word in wake_words: + activated = False + if isinstance(wake_word, MicroWakeWord): + if any(wake_word.process_streaming(mi) for mi in micro_inputs): activated = True + elif isinstance(wake_word, OpenWakeWord): + if any(p > 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.loop.call_soon_threadsafe(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") + if stopped and (state.stop_word.id in state.active_wake_words): + state.loop.call_soon_threadsafe(state.satellite.stop) + + state.loop.call_soon_threadsafe(state.satellite.handle_audio, audio_chunk) + + except Exception: + _LOGGER.exception("Unexpected error handling audio") + except Exception as e: - _LOGGER.critical("Error in audio processing thread: %s", e) + _LOGGER.critical("A soundcard error occurred: %s", e) state.loop.call_soon_threadsafe(state.loop.stop) +# ----------------------------------------------------------------------------- + if __name__ == "__main__": print("--- __main__ block executing ---") asyncio.run(main()) diff --git a/linux_voice_assistant/event_bus.py b/linux_voice_assistant/event_bus.py index 55f2d9c..24330ae 100644 --- a/linux_voice_assistant/event_bus.py +++ b/linux_voice_assistant/event_bus.py @@ -1,3 +1,5 @@ +"""A simple synchronous publish/subscribe event bus.""" + import logging from typing import Any, Callable, Dict, List, Optional @@ -8,34 +10,33 @@ class EventBus: """A simple synchronous publish/subscribe event bus.""" def __init__(self): - # A dictionary to hold listeners for specific string topics self.topics: Dict[str, List[Callable[[Any], None]]] = {} def subscribe(self, topic: str, listener: Callable[[Any], None]) -> None: - """ - Subscribes a listener to a topic. - """ + """Subscribes a listener to a topic.""" if topic not in self.topics: self.topics[topic] = [] self.topics[topic].append(listener) + _LOGGER.debug(f"Subscribed listener to topic '{topic}'") def publish(self, topic: str, data: Optional[Dict[str, Any]] = None) -> None: - """ - Publishes an event to all subscribed listeners. - """ + """Publishes an event to all subscribed listeners.""" if data is None: data = {} data['__topic'] = topic listeners = self.topics.get(topic, []) + _LOGGER.debug(f"Publishing event to {len(listeners)} listeners on topic '{topic}'") for listener in listeners: try: listener(data) except Exception: _LOGGER.exception("Error in event listener for topic %s", topic) +# ----------------------------------------------------------------------------- # Client helpers for subscriptions +# ----------------------------------------------------------------------------- def subscribe(func: Callable) -> Callable: """Decorator to mark a method for event bus subscription.""" @@ -45,10 +46,13 @@ def subscribe(func: Callable) -> Callable: class EventHandler: """ A base class for components that subscribe to events. + + Subclasses must call `self._subscribe_all_methods()` in their __init__. """ - def __init__(self, state: Any): - self.state = state - self._subscribe_all_methods() + def __init__(self, event_bus: EventBus): + self.event_bus = event_bus + # Note: Subclasses must call self._subscribe_all_methods() + # after their own __init__ is complete. def _subscribe_all_methods(self): """Finds and subscribes all methods decorated with @subscribe.""" @@ -57,5 +61,4 @@ class EventHandler: if hasattr(method, '_event_bus_subscribe'): # The topic is the name of the method itself. - self.state.event_bus.subscribe(method_name, method) - _LOGGER.debug(f"Subscribed method '{method_name}' to topic '{method_name}'") + self.event_bus.subscribe(method_name, method) diff --git a/linux_voice_assistant/led_controller.py b/linux_voice_assistant/led_controller.py index ce20c8d..29be8af 100644 --- a/linux_voice_assistant/led_controller.py +++ b/linux_voice_assistant/led_controller.py @@ -1,10 +1,12 @@ import asyncio import logging -from typing import Any, Tuple +from typing import Any, Callable, Dict, Optional, Tuple import board -from .event_bus import EventHandler, subscribe +from .config import LedConfig +from .event_bus import EventBus, EventHandler, subscribe +from .models import Preferences _LOGGER = logging.getLogger(__name__) @@ -21,17 +23,15 @@ _PURPLE = (128, 0, 255) class LedController(EventHandler): def __init__( self, - state, - led_type: str = "dotstar", - interface: str = "spi", - clock_pin: int = 13, - data_pin: int = 12, - num_leds: int = 3, + loop: asyncio.AbstractEventLoop, + event_bus: EventBus, + config: LedConfig, + preferences: Preferences, ): - super().__init__(state) - self.loop = state.loop - self.num_leds = num_leds - self.current_task = None + super().__init__(event_bus) + self.loop = loop + self.num_leds = preferences.num_leds # Get num_leds from preferences + self.current_task: Optional[asyncio.Task] = None self._is_ready = False self.leds = None @@ -44,9 +44,8 @@ class LedController(EventHandler): } try: - # Universal Hardware Initialization - if led_type == "neopixel": - if interface == "spi": + if config.led_type == "neopixel": + if config.interface == "spi": import busio import neopixel_spi _LOGGER.debug(f"Initializing {self.num_leds} NeoPixel LEDs on hardware SPI") @@ -54,27 +53,30 @@ class LedController(EventHandler): self.leds = neopixel_spi.NeoPixel_SPI(spi, self.num_leds, auto_write=False) else: # GPIO import neopixel - _LOGGER.debug(f"Initializing {self.num_leds} NeoPixel LEDs on GPIO data={data_pin}") - pin_object = getattr(board, f"D{data_pin}") + _LOGGER.debug(f"Initializing {self.num_leds} NeoPixel LEDs on GPIO data={config.data_pin}") + pin_object = getattr(board, f"D{config.data_pin}") self.leds = neopixel.NeoPixel(pin_object, self.num_leds, auto_write=False) else: # dotstar import adafruit_dotstar - if interface == "gpio": - _LOGGER.debug(f"Initializing {self.num_leds} DotStar LEDs on GPIO data={data_pin}, clock={clock_pin}") - data_pin_obj = getattr(board, f"D{data_pin}") - clock_pin_obj = getattr(board, f"D{clock_pin}") + if config.interface == "gpio": + _LOGGER.debug(f"Initializing {self.num_leds} DotStar LEDs on GPIO data={config.data_pin}, clock={config.clock_pin}") + data_pin_obj = getattr(board, f"D{config.data_pin}") + clock_pin_obj = getattr(board, f"D{config.clock_pin}") self.leds = adafruit_dotstar.DotStar(clock_pin_obj, data_pin_obj, self.num_leds, auto_write=False) else: # SPI _LOGGER.debug(f"Initializing {self.num_leds} DotStar LEDs on hardware SPI") self.leds = adafruit_dotstar.DotStar(board.SCLK, board.MOSI, self.num_leds, auto_write=False) self._is_ready = True - _LOGGER.info(f"LED Controller initialized for {self.num_leds} {led_type} LEDs.") + _LOGGER.info(f"LED Controller initialized for {self.num_leds} {config.led_type} LEDs.") self.run_action("startup_sequence") except Exception: _LOGGER.exception("Failed to initialize LED controller. LEDs will be disabled.") + # Subscribe to events *after* __init__ is complete + self._subscribe_all_methods() + def run_action(self, action_method_name: str, *args: Any) -> None: if not self._is_ready: return if self.current_task and not self.current_task.done(): self.current_task.cancel() @@ -89,7 +91,7 @@ class LedController(EventHandler): if publish_state: config["state_name"] = state_name - self.state.event_bus.publish("publish_state_to_mqtt", config) + self.event_bus.publish("publish_state_to_mqtt", config) async def startup_sequence(self, color=None, brightness=None): await self.blink(_GREEN, 1.0) async def off(self, color, brightness): self.leds.fill(_OFF); self.leds.show() @@ -134,33 +136,23 @@ class LedController(EventHandler): self.leds.fill(_OFF); self.leds[i % self.num_leds] = bright_color; self.leds.show(); i += 1; await asyncio.sleep(speed) except asyncio.CancelledError: self.leds.fill(_OFF); self.leds.show() - # --- MODIFIED STATE SUBSCRIPTIONS --- - + # --- State Subscriptions --- @subscribe def voice_idle(self, data: dict): self._apply_state_effect("idle") - @subscribe def voice_listen(self, data: dict): self._apply_state_effect("listening") - @subscribe def voice_thinking(self, data: dict): self._apply_state_effect("thinking") - @subscribe def voice_responding(self, data: dict): self._apply_state_effect("responding") - @subscribe def voice_error(self, data: dict): self._apply_state_effect("error") - - # --- REMOVED REDUNDANT SUBSCRIPTIONS --- - # (e.g., voice_wakeword, voice_stt_end, etc.) - @subscribe def mic_muted(self, data: dict): self.run_action("solid", _DIM_RED, 1.0) - @subscribe def mic_unmuted(self, data: dict): self._apply_state_effect("idle") - # --- MQTT CONFIG SUBSCRIPTIONS (Unchanged) --- + # --- MQTT Config Subscriptions --- def _update_config(self, state_name: str, data: dict, apply: bool): config = self.configs[state_name]; changed = False; is_retained = data.get("retained", False) @@ -179,7 +171,7 @@ class LedController(EventHandler): return if changed: if apply: self._apply_state_effect(state_name) - else: config["state_name"] = state_name; self.state.event_bus.publish("publish_state_to_mqtt", config) + else: config["state_name"] = state_name; self.event_bus.publish("publish_state_to_mqtt", config) @subscribe def set_idle_effect(self, data: dict): self._update_config("idle", data, True) @@ -201,6 +193,7 @@ class LedController(EventHandler): def set_error_effect(self, data: dict): self._update_config("error", data, False) @subscribe def set_error_color(self, data: dict): self._update_config("error", data, False) + @subscribe def set_num_leds(self, data: dict): num_leds = data.get("num_leds") diff --git a/linux_voice_assistant/models.py b/linux_voice_assistant/models.py index ca2feb5..97ae49e 100644 --- a/linux_voice_assistant/models.py +++ b/linux_voice_assistant/models.py @@ -67,24 +67,25 @@ class Preferences: @dataclass class ServerState: + """A simple dataclass to hold core application state.""" # --- Fields WITHOUT default values --- name: str mac_address: str + event_bus: EventBus + loop: asyncio.AbstractEventLoop entities: "List[ESPHomeEntity]" + music_player: "MpvMediaPlayer" + tts_player: "MpvMediaPlayer" 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 - + download_dir: Path + preferences: Preferences # <-- MOVED UP + # --- Fields WITH default values --- satellite: "Optional[VoiceSatelliteProtocol]" = None wake_words_changed: bool = False diff --git a/linux_voice_assistant/mqtt_controller.py b/linux_voice_assistant/mqtt_controller.py index e26d297..19dff7a 100644 --- a/linux_voice_assistant/mqtt_controller.py +++ b/linux_voice_assistant/mqtt_controller.py @@ -1,10 +1,13 @@ +import asyncio import json import logging -from typing import TYPE_CHECKING, List +from typing import TYPE_CHECKING, List, Optional import paho.mqtt.client as mqtt -from .event_bus import EventHandler, subscribe +from .config import MqttConfig +from .event_bus import EventBus, EventHandler, subscribe +from .models import Preferences if TYPE_CHECKING: from .models import ServerState @@ -12,16 +15,29 @@ if TYPE_CHECKING: _LOGGER = logging.getLogger(__name__) class MqttController(EventHandler): - def __init__(self, state: "ServerState", host: str, port: int, username: str, password: str): - super().__init__(state) - self._host = host - self._port = port - self._username = username - self._password = password - - self._device_name = self.state.name - self._device_id = self.state.name.lower().replace(" ", "_") + def __init__( + self, + loop: asyncio.AbstractEventLoop, + event_bus: EventBus, + config: MqttConfig, + app_name: str, + mac_address: str, + preferences: Preferences, + ): + super().__init__(event_bus) + self.loop = loop + self.preferences = preferences + + self._host = config.host + self._port = config.port + self._username = config.username + self._password = config.password + + self._device_name = app_name + self._device_id = app_name.lower().replace(" ", "_") self._topic_prefix = f"lva/{self._device_id}" + + self._is_muted = False # Internal state self.CONFIGURABLE_STATES: List[str] = ["idle", "listening", "thinking", "responding", "error"] self.topics = { @@ -40,6 +56,9 @@ class MqttController(EventHandler): self._client.on_connect = self._on_connect self._client.on_message = self._on_message + # Subscribe to events *after* __init__ is complete + self._subscribe_all_methods() + def start(self): try: if self._username: @@ -59,49 +78,33 @@ class MqttController(EventHandler): def _on_connect(self, client, userdata, flags, rc): if rc == 0: _LOGGER.info("Connected to MQTT broker") - # Subscribe to all command topics - client.subscribe(f"{self._topic_prefix}/+/set") # Wildcard for all commands - - # Subscribe to retained state topics to sync on startup + client.subscribe(f"{self._topic_prefix}/+/set") client.subscribe(f"{self._topic_prefix}/+/state") - self._publish_discovery_configs() else: _LOGGER.error("Failed to connect to MQTT, return code %d", rc) def _on_message(self, client, userdata, msg): - """ - Handles incoming MQTT messages (runs in MQTT thread). - This must not block or modify shared state. - It schedules the real handler to run on the main asyncio loop. - """ try: topic = msg.topic payload = msg.payload.decode() _LOGGER.debug("Received MQTT message on topic %s (from MQTT thread)", msg.topic) - - # Schedule the actual processing on the main event loop - self.state.loop.call_soon_threadsafe(self._handle_message_on_loop, topic, payload) + self.loop.call_soon_threadsafe(self._handle_message_on_loop, topic, payload) except Exception: _LOGGER.exception("Error in _on_message") def _handle_message_on_loop(self, topic: str, payload: str): - """ -This method runs on the main asyncio loop and safely handles -the MQTT message logic and state changes. -""" _LOGGER.debug("Handling message for topic %s on main loop", topic) if topic == self.topics["mute"]["command"]: - self.state.event_bus.publish("set_mic_mute", {"state": payload.upper() == "ON"}) + # Publish event for MicMuteHandler to process + self.event_bus.publish("set_mic_mute", {"state": payload.upper() == "ON"}) elif topic == self.topics["num_leds"]["command"]: try: num_leds = int(payload) - self.state.event_bus.publish("set_num_leds", {"num_leds": num_leds}) - if self.state.preferences.num_leds != num_leds: - self.state.preferences.num_leds = num_leds - self.state.save_preferences() + # Publish event for LedController and MicMuteHandler (to save) + self.event_bus.publish("set_num_leds", {"num_leds": num_leds}) self.publish_num_leds_state(num_leds) except ValueError: pass @@ -109,32 +112,29 @@ the MQTT message logic and state changes. state_topics = self.topics[state_name] if topic == state_topics["effect_command"]: effect_id = payload.lower().replace(" ", "_") - self.state.event_bus.publish(f"set_{state_name}_effect", {"effect": effect_id}) + self.event_bus.publish(f"set_{state_name}_effect", {"effect": effect_id}) elif topic == state_topics["light_command"]: try: data = json.loads(payload) - if "state" in data: if data["state"].upper() == "OFF": - self.state.event_bus.publish(f"set_{state_name}_effect", {"effect": "off"}) + self.event_bus.publish(f"set_{state_name}_effect", {"effect": "off"}) else: # ON command - self.state.event_bus.publish(f"turn_on_{state_name}") + self.event_bus.publish(f"turn_on_{state_name}") if "color" in data or "brightness" in data: - self.state.event_bus.publish(f"set_{state_name}_color", data) - + self.event_bus.publish(f"set_{state_name}_color", data) except json.JSONDecodeError: pass - # Handle retained state messages on startup elif topic == state_topics["effect_state"]: effect_id = payload.lower().replace(" ", "_") - self.state.event_bus.publish(f"set_{state_name}_effect", {"effect": effect_id, "retained": True}) + self.event_bus.publish(f"set_{state_name}_effect", {"effect": effect_id, "retained": True}) elif topic == state_topics["light_state"]: try: data = json.loads(payload) data["retained"] = True - self.state.event_bus.publish(f"set_{state_name}_color", data) + self.event_bus.publish(f"set_{state_name}_color", data) except json.JSONDecodeError: pass def _publish_discovery_configs(self): @@ -160,10 +160,12 @@ the MQTT message logic and state changes. _LOGGER.debug("Published all MQTT discovery configs") self._client.publish(availability_topic, "online", retain=True) - self.publish_mute_state(self.state.mic_muted) - self.publish_num_leds_state(self.state.preferences.num_leds) + # Publish initial state from internal values + self.publish_mute_state(self._is_muted) + self.publish_num_leds_state(self.preferences.num_leds) def publish_mute_state(self, is_muted: bool): + self._is_muted = is_muted self._client.publish(self.topics["mute"]["state"], "ON" if is_muted else "OFF", retain=True) def publish_num_leds_state(self, num_leds: int): @@ -184,3 +186,13 @@ the MQTT message logic and state changes. "color": { "r": data.get("color")[0], "g": data.get("color")[1], "b": data.get("color")[2] } } self._client.publish(state_topics["light_state"], json.dumps(light_state), retain=True) + + # --- Listen for state changes from other components --- + + @subscribe + def mic_muted(self, data: dict): + self.publish_mute_state(True) + + @subscribe + def mic_unmuted(self, data: dict): + self.publish_mute_state(False) diff --git a/linux_voice_assistant/satellite.py b/linux_voice_assistant/satellite.py index 3fe1ccc..e9bb763 100644 --- a/linux_voice_assistant/satellite.py +++ b/linux_voice_assistant/satellite.py @@ -11,7 +11,6 @@ from typing import Dict, Optional, Set, Union from urllib.parse import urlparse, urlunparse from urllib.request import urlopen -# pylint: disable=no-name-in-module from aioesphomeapi.api_pb2 import ( DeviceInfoRequest, DeviceInfoResponse, @@ -25,7 +24,7 @@ from aioesphomeapi.api_pb2 import ( VoiceAssistantConfigurationRequest, VoiceAssistantConfigurationResponse, VoiceAssistantEventResponse, - VoiceAssistantExternalWakeWord, # <-- ADDED + VoiceAssistantExternalWakeWord, VoiceAssistantRequest, VoiceAssistantSetConfiguration, VoiceAssistantTimerEventResponse, @@ -42,7 +41,7 @@ from pyopen_wakeword import OpenWakeWord from .api_server import APIServer from .entity import MediaPlayerEntity -from .models import AvailableWakeWord, ServerState, SatelliteState, WakeWordType # <-- ADDED +from .models import AvailableWakeWord, ServerState, SatelliteState, WakeWordType from .util import call_all _LOGGER = logging.getLogger(__name__) @@ -51,9 +50,8 @@ _LOGGER = logging.getLogger(__name__) class VoiceSatelliteProtocol(APIServer): def __init__(self, state: ServerState) -> None: super().__init__(state.name) - self.state = state - self.state.satellite = self + self.state.satellite = self # <-- ADDED BACK self.media_player_entity = MediaPlayerEntity( server=self, @@ -70,16 +68,14 @@ class VoiceSatelliteProtocol(APIServer): self._state: SatelliteState = SatelliteState.STARTING self._is_streaming_audio: bool = False - # Announce/TTS state self._tts_url: Optional[str] = None self._continue_conversation: bool = False self._timer_finished: bool = False - # Flags set by events, processed in _determine_final_state self._run_end_received: bool = False self._tts_end_received: bool = False - self._external_wake_words: Dict[str, VoiceAssistantExternalWakeWord] = {} # <-- ADDED + self._external_wake_words: Dict[str, VoiceAssistantExternalWakeWord] = {} def _set_state(self, new_state: SatelliteState): if self._state == new_state: @@ -210,7 +206,6 @@ class VoiceSatelliteProtocol(APIServer): if isinstance(msg, ListEntitiesRequest): yield ListEntitiesDoneResponse() elif isinstance(msg, VoiceAssistantConfigurationRequest): - # --- MODIFIED --- available_wake_words = [ VoiceAssistantWakeWord( id=ww.id, @@ -220,7 +215,6 @@ class VoiceSatelliteProtocol(APIServer): for ww in self.state.available_wake_words.values() ] - # Add external wake words self._external_wake_words.clear() for eww in msg.external_wake_words: if eww.model_type != "micro": @@ -244,7 +238,6 @@ class VoiceSatelliteProtocol(APIServer): ], max_active_wake_words=2, ) - # --- END MODIFIED --- _LOGGER.info("Connected to Home Assistant") self._set_state(SatelliteState.IDLE) @@ -258,9 +251,7 @@ class VoiceSatelliteProtocol(APIServer): model_info = self.state.available_wake_words.get(wake_word_id) - # --- ADDED --- if not model_info: - # Check external wake words (may require download) external_wake_word = self._external_wake_words.get(wake_word_id) if not external_wake_word: _LOGGER.warning("Unknown wake word: %s", wake_word_id) @@ -268,10 +259,9 @@ class VoiceSatelliteProtocol(APIServer): model_info = self._download_external_wake_word(external_wake_word) if not model_info: - continue # download failed + continue self.state.available_wake_words[wake_word_id] = model_info - # --- END ADDED --- _LOGGER.debug("Loading wake word: %s", model_info.wake_word_path) self.state.wake_words[wake_word_id] = model_info.load() @@ -288,9 +278,8 @@ class VoiceSatelliteProtocol(APIServer): self.state.wake_words_changed = True def handle_audio(self, audio_chunk: bytes) -> None: - if not self._is_streaming_audio: - return - self.send_messages([VoiceAssistantAudio(data=audio_chunk)]) + if self._is_streaming_audio: + self.send_messages([VoiceAssistantAudio(data=audio_chunk)]) def wakeup(self, wake_word: Union[MicroWakeWord, OpenWakeWord]) -> None: if self._state not in (SatelliteState.IDLE, SatelliteState.STARTING): @@ -372,7 +361,6 @@ class VoiceSatelliteProtocol(APIServer): ), ) - # --- ADDED --- def _download_external_wake_word( self, external_wake_word: VoiceAssistantExternalWakeWord ) -> Optional[AvailableWakeWord]: @@ -382,7 +370,6 @@ class VoiceSatelliteProtocol(APIServer): config_path = eww_dir / f"{external_wake_word.id}.json" should_download_config = not config_path.exists() - # Check if we need to download the model file model_path = eww_dir / f"{external_wake_word.id}.tflite" should_download_model = True if model_path.exists(): @@ -399,7 +386,6 @@ class VoiceSatelliteProtocol(APIServer): ) if should_download_config or should_download_model: - # Download config _LOGGER.debug("Downloading %s to %s", external_wake_word.url, config_path) with urlopen(external_wake_word.url) as request: if request.status != 200: @@ -414,7 +400,6 @@ class VoiceSatelliteProtocol(APIServer): shutil.copyfileobj(request, model_file) if should_download_model: - # Download model file parsed_url = urlparse(external_wake_word.url) parsed_url = parsed_url._replace( path=posixpath.join(posixpath.dirname(parsed_url.path), model_path.name) @@ -437,9 +422,8 @@ class VoiceSatelliteProtocol(APIServer): type=WakeWordType.MICRO_WAKE_WORD, wake_word=external_wake_word.wake_word, trained_languages=external_wake_word.trained_languages, - wake_word_path=config_path, # micro-wakeword loads from config + wake_word_path=config_path, ) - # --- END ADDED --- def connection_lost(self, exc): super().connection_lost(exc)