Merge pull request #36 from imonlinux/decouple-controllers-breakup-main

Decouple controllers breakup main

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]):
+    Manages the mic_muted state and saves preferences.
-        super().__init__(state)
+    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,70 +79,112 @@ 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",
+        "-c", "--config", type=Path, required=False,
-        "--config",
-        type=Path,
-        required=False,
         default=_MODULE_DIR / "config.json",
-        help="Path to configuration.json file (default: linux_voice_assistant/config.json)",
+        help="Path to configuration.json file"
-    )
-    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",
     )
+    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("Input devices\n" + "=" * 13)
-        print("=" * 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\n" + "=" * 14)
-        print("Output devices")
-        print("=" * 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)
@@ -118,106 +192,111 @@ async def main() -> None:
     loop = asyncio.get_running_loop()
     event_bus = EventBus()
     
-    # --- Create Download Dir ---
+    return config, loop, event_bus
-    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:
+        with open(preferences_path, "r", encoding="utf-8") as f:
-            preferences_dict = json.load(preferences_file)
+            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
 
-    # --- Load Wake Words ---
+def _get_microphone(config: Config):
-    available_wake_words: Dict[str, AvailableWakeWord] = {}
+    """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()
+    
+    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"]
         
-    # Add download directory to search path
+    download_dir = _REPO_DIR / config.wake_word.download_dir
-    wake_word_dirs = [_REPO_DIR / d for d in config.wake_word.directories]
+    download_dir.mkdir(parents=True, exist_ok=True)
-    wake_word_dirs.append(download_dir / "external_wake_words") # <-- ADDED
+    _LOGGER.debug("Download directory: %s", download_dir)
 
+    wake_word_dirs = [_REPO_DIR / d for d in config.wake_word.directories]
+    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 config_path in wake_word_dir.glob("*.json"):
-        for model_config_path in wake_word_dir.glob("*.json"):
+            model_id = config_path.stem
-            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:
+            with open(config_path, "r", encoding="utf-8") as f:
-                model_config = json.load(model_config_file)
+                model_config = json.load(f)
                 model_type = WakeWordType(model_config["type"])
                 
-                if model_type == WakeWordType.OPEN_WAKE_WORD:
+                wake_word_path = (
-                    wake_word_path = model_config_path.parent / model_config["model"]
+                    config_path.parent / model_config["model"]
-                else:
+                    if model_type == WakeWordType.OPEN_WAKE_WORD
-                    wake_word_path = model_config_path
+                    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:
+        for ww_id in preferences.active_wake_words:
-            wake_word = available_wake_words.get(wake_word_id)
+            if ww_id in available:
-            if wake_word is None: continue
+                _LOGGER.debug("Loading wake model (from preferences): %s", ww_id)
-            _LOGGER.debug("Loading wake model: %s", wake_word_id)
+                models[ww_id] = available[ww_id].load()
-            wake_models[wake_word_id] = wake_word.load()
+                active.add(ww_id)
-            active_wake_words.add(wake_word_id)
     
-    if not wake_models:
+    if not models:
-        wake_word_id = config.wake_word.model
+        ww_id = config.wake_word.model
-        wake_word = available_wake_words.get(wake_word_id)
+        if ww_id in available:
-        if wake_word:
+            _LOGGER.debug("Loading wake model (from config): %s", ww_id)
-            _LOGGER.debug("Loading wake model: %s", wake_word_id)
+            models[ww_id] = available[ww_id].load()
-            wake_models[wake_word_id] = wake_word.load()
+            active.add(ww_id)
-            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 = _REPO_DIR / ww_dir_str / f"{config.wake_word.stop_model}.json"
-        stop_config_path = wake_word_dir / f"{config.wake_word.stop_model}.json"
+        if stop_config_path.exists():
-        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
+    assert stop_model is not None, "Stop model not found"
 
-    # --- Resolve Microphone ---
+    return WakeWordData(available, models, active, stop_model)
-    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:
+def _init_media_players(loop: asyncio.AbstractEventLoop, config: Config, preferences: Preferences) -> MediaPlayers:
-        _LOGGER.critical("No microphone found.")
+    """Initializes the music and TTS media players."""
-        sys.exit(1)
+    _LOGGER.debug("Initializing media players")
-        
-    _LOGGER.info("Using audio input device: %s", mic.name)
-
-    # --- Initialize 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
     )
+    return MediaPlayers(music=music_player, tts=tts_player)
 
-    # --- Create Global Server State ---
+def _create_server_state(
-    state = ServerState(
+    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,
+        music_player=media_players.music,
-        wake_words=wake_models,
+        tts_player=media_players.tts,
-        active_wake_words=active_wake_words,
+        available_wake_words=wake_word_data.available,
-        stop_word=stop_model,
+        wake_words=wake_word_data.models,
-        music_player=music_player,
+        active_wake_words=wake_word_data.active,
-        tts_player=tts_player,
+        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,
+        preferences_path=_REPO_DIR / config.app.preferences_file,
-        event_bus=event_bus,
+        download_dir=_REPO_DIR / config.wake_word.download_dir,
-        loop=loop,
-        download_dir=download_dir, # <-- ADDED
         refractory_seconds=config.wake_word.refractory_seconds,
     )
 
-    # --- Initialize Controllers ---
+def _init_controllers(
+    loop: asyncio.AbstractEventLoop,
+    event_bus: EventBus,
+    state: ServerState,
+    config: Config,
+    preferences: Preferences,
+):
+    """Initializes all decoupled controllers."""
+    _LOGGER.debug("Initializing controllers")
+    
     led_controller = LedController(
-        state,
+        loop=loop,
-        led_type=config.led.led_type,
+        event_bus=event_bus,
-        interface=config.led.interface,
+        config=config.led,
-        clock_pin=config.led.clock_pin,
+        preferences=preferences,
-        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,
+            loop=loop,
-            host=config.mqtt.host,
+            event_bus=event_bus,
-            port=config.mqtt.port,
+            config=config.mqtt,
-            username=config.mqtt.username,
+            app_name=config.app.name,
-            password=config.mqtt.password
+            mac_address=state.mac_address,
+            preferences=preferences,
         )
+        setattr(state, "mqtt_controller", mqtt_controller)
         mqtt_controller.start()
 
-    mic_mute_handler = MicMuteHandler(state, mqtt_controller)
+    mic_mute_handler = MicMuteHandler(
-    
+        event_bus=event_bus,
-    # --- Start Audio Processing Thread ---
+        state=state,
-    process_audio_thread = threading.Thread(
+        mqtt_controller=mqtt_controller,
-        target=process_audio,
-        args=(state, mic, config.audio.input_block_size),
-        daemon=True
     )
-    process_audio_thread.start()
 
-    # --- Start ESPHome Server ---
+async def _run_server(state: ServerState, config: Config):
-    server = await loop.create_server(
+    """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,38 +386,57 @@ 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()
 
-    _LOGGER.debug("Server stopped")
+def _process_audio(state: ServerState, mic, block_size: int):
+    """Main audio processing loop. Runs in a separate thread."""
     
-def process_audio(state: ServerState, mic, block_size: int):
+    micro_features: MicroWakeWordFeatures = MicroWakeWordFeatures()
-    """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
+    
+    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:
+        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)
+
+                # --- Just unmuted ---
+                _LOGGER.debug("Audio thread unmuted, clearing buffers.")
+                micro_features.reset()
+                if oww_features is not None:
+                    oww_features.reset()
+                last_active = time.monotonic() # Start refractory period
+                
+                # Re-open microphone to flush OS buffer
+                _LOGGER.debug("Re-opening microphone to flush OS buffers...")
+                with mic.recorder(samplerate=16000, channels=1, blocksize=block_size) as mic_in:
+                    mic_in.flush() # Discard any stale data
+                    _LOGGER.debug("Microphone flushed.")
+                    # Fall through to the main recording loop
+            # --- END FIX ---
+            
             _LOGGER.debug("Opening audio input device: %s", mic.name)
             with mic.recorder(samplerate=16000, channels=1, blocksize=block_size) as mic_in:
-            while True:
+                # --- This is now the main recording loop ---
-                if state.mic_muted:
+                while not state.mic_muted: 
-                    time.sleep(0.1)
+                    if state.shutdown:
-                    continue
+                        _LOGGER.debug("Shutdown signal received, stopping audio thread.")
+                        return # Exit thread
                     
                     audio_chunk_array = mic_in.record(block_size).reshape(-1)
                     audio_chunk = (
@@ -331,23 +446,22 @@ def process_audio(state: ServerState, mic, block_size: int):
                     )
 
                     if state.satellite is None:
+                        time.sleep(0.01)
                         continue
                     
-                if (not wake_words) or (state.wake_words_changed and state.wake_words):
+                    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 micro_features is None:
-                        micro_features = MicroWakeWordFeatures()
-                    if has_oww and (oww_features is None):
-                        oww_features = OpenWakeWordFeatures.from_builtin()
-                try:
-                    state.satellite.handle_audio(audio_chunk)
                         
-                    assert micro_features is not None
+                        if has_oww and (oww_features is None):
+                            _LOGGER.debug("Initializing OpenWakeWord features...")
+                            oww_features = OpenWakeWordFeatures.from_builtin()
+                    
+                    try:
                         micro_inputs.clear(); micro_inputs.extend(micro_features.process_streaming(audio_chunk))
                         
                         if has_oww:
@@ -364,7 +478,7 @@ def process_audio(state: ServerState, mic, block_size: int):
                             if activated:
                                 now = time.monotonic()
                                 if (last_active is None) or ((now - last_active) > state.refractory_seconds):
-                                state.satellite.wakeup(wake_word)
+                                    state.loop.call_soon_threadsafe(state.satellite.wakeup, wake_word)
                                     last_active = now
                         
                         stopped = False
@@ -373,13 +487,20 @@ def process_audio(state: ServerState, mic, block_size: int):
                                 stopped = True
 
                         if stopped and (state.stop_word.id in state.active_wake_words):
-                        state.satellite.stop()
+                            state.loop.call_soon_threadsafe(state.satellite.stop)
-                except Exception: _LOGGER.exception("Unexpected error handling audio")
+                            
+                        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())

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):
+    def __init__(self, event_bus: EventBus):
-        self.state = state
+        self.event_bus = event_bus
-        self._subscribe_all_methods()
+        # 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)
+                self.event_bus.subscribe(method_name, method)
-                _LOGGER.debug(f"Subscribed method '{method_name}' to topic '{method_name}'")

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,
+        loop: asyncio.AbstractEventLoop,
-        led_type: str = "dotstar",
+        event_bus: EventBus,
-        interface: str = "spi",
+        config: LedConfig,
-        clock_pin: int = 13,
+        preferences: Preferences,
-        data_pin: int = 12,
-        num_leds: int = 3,
     ):
-        super().__init__(state)
+        super().__init__(event_bus)
-        self.loop = state.loop
+        self.loop = loop
-        self.num_leds = num_leds
+        self.num_leds = preferences.num_leds  # Get num_leds from preferences
-        self.current_task = None
+        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 config.led_type == "neopixel":
-            if led_type == "neopixel":
+                if config.interface == "spi":
-                if 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}")
+                    _LOGGER.debug(f"Initializing {self.num_leds} NeoPixel LEDs on GPIO data={config.data_pin}")
-                    pin_object = getattr(board, f"D{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":
+                if config.interface == "gpio":
-                    _LOGGER.debug(f"Initializing {self.num_leds} DotStar LEDs on GPIO data={data_pin}, clock={clock_pin}")
+                    _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{data_pin}")
+                    data_pin_obj = getattr(board, f"D{config.data_pin}")
-                    clock_pin_obj = getattr(board, f"D{clock_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")

linux_voice_assistant/models.py

@@ -67,23 +67,24 @@ 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
+    download_dir: Path
-    loop: asyncio.AbstractEventLoop
+    preferences: Preferences  # <-- MOVED UP
-    download_dir: Path  # <-- ADDED
     
     # --- Fields WITH default values ---
     satellite: "Optional[VoiceSatelliteProtocol]" = None

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,17 +15,30 @@ if TYPE_CHECKING:
 _LOGGER = logging.getLogger(__name__)
 
 class MqttController(EventHandler):
-    def __init__(self, state: "ServerState", host: str, port: int, username: str, password: str):
+    def __init__(
-        super().__init__(state)
+        self,
-        self._host = host
+        loop: asyncio.AbstractEventLoop,
-        self._port = port
+        event_bus: EventBus,
-        self._username = username
+        config: MqttConfig,
-        self._password = password
+        app_name: str,
+        mac_address: str,
+        preferences: Preferences,
+    ):
+        super().__init__(event_bus)
+        self.loop = loop
+        self.preferences = preferences
         
-        self._device_name = self.state.name
+        self._host = config.host
-        self._device_id = self.state.name.lower().replace(" ", "_")
+        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 = {
             "mute": { "command": f"{self._topic_prefix}/mute/set", "state": f"{self._topic_prefix}/mute/state" },
@@ -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")
-            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}/+/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)
-            
+            self.loop.call_soon_threadsafe(self._handle_message_on_loop, topic, payload)
-            # Schedule the actual processing on the main event loop
-            self.state.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})
+                # Publish event for LedController and MicMuteHandler (to save)
-                if self.state.preferences.num_leds != num_leds:
+                self.event_bus.publish("set_num_leds", {"num_leds": num_leds})
-                    self.state.preferences.num_leds = num_leds
-                    self.state.save_preferences()
                 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)
+        # Publish initial state from internal values
-        self.publish_num_leds_state(self.state.preferences.num_leds)
+        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)

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,8 +278,7 @@ class VoiceSatelliteProtocol(APIServer):
             self.state.wake_words_changed = True
 
     def handle_audio(self, audio_chunk: bytes) -> None:
-        if not self._is_streaming_audio:
+        if self._is_streaming_audio:
-            return
             self.send_messages([VoiceAssistantAudio(data=audio_chunk)])
 
     def wakeup(self, wake_word: Union[MicroWakeWord, OpenWakeWord]) -> None:
@@ -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)