From 39d6d3e010add79dbcbd1e6198697329d541d65a Mon Sep 17 00:00:00 2001 From: imonlinux <39863321+imonlinux@users.noreply.github.com> Date: Sat, 11 Oct 2025 17:10:15 -0700 Subject: [PATCH] Add Microphone Mute Switch Entity Implements the core mute functionality. Adds a new MicMuteHandler class that subscribes to the set_mic_mute event. This handler updates the state.mic_muted flag and publishes secondary mic_muted or mic_unmuted events for LED feedback. Modifies the process_audio function to check the state.mic_muted flag in its main loop, pausing audio processing when the flag is True. --- linux_voice_assistant/__main__.py | 79 ++++++++++++------------------- 1 file changed, 31 insertions(+), 48 deletions(-) diff --git a/linux_voice_assistant/__main__.py b/linux_voice_assistant/__main__.py index f23c147..04c896f 100644 --- a/linux_voice_assistant/__main__.py +++ b/linux_voice_assistant/__main__.py @@ -12,7 +12,7 @@ from typing import Dict, List, Optional, Union import numpy as np import sounddevice as sd -from .event_bus import EventBus +from .event_bus import EventBus, EventHandler, subscribe # ADDED EventHandler, subscribe from .led_controller import LedController from .microwakeword import MicroWakeWord, MicroWakeWordFeatures from .models import AvailableWakeWord, Preferences, ServerState, WakeWordType @@ -34,12 +34,29 @@ if is_arm(): else: _LIB_DIR = _REPO_DIR / "lib" / "linux_amd64" +# ----------------------------------------------------------------------------- +# --- ADDED MIC MUTE HANDLER --- +# ----------------------------------------------------------------------------- + +class MicMuteHandler(EventHandler): + """Event handler for mic mute switch.""" + @subscribe + def set_mic_mute(self, data: dict): + is_muted = data.get("state", False) + if self.state.mic_muted != is_muted: + self.state.mic_muted = is_muted + _LOGGER.debug("Mic muted = %s", is_muted) + if is_muted: + self.state.event_bus.publish("mic_muted") + else: + self.state.event_bus.publish("mic_unmuted") # ----------------------------------------------------------------------------- async def main() -> None: parser = argparse.ArgumentParser() + # ... (all parser arguments are the same) ... parser.add_argument("--name", required=True) parser.add_argument( "--audio-input-device", @@ -64,7 +81,6 @@ async def main() -> None: type=float, help="Seconds before wake word can be activated again", ) - # parser.add_argument( "--oww-melspectrogram-model", default=_OWW_DIR / "melspectrogram.tflite", @@ -75,16 +91,13 @@ async def main() -> None: default=_OWW_DIR / "embedding_model.tflite", help="Path to openWakeWord embedding model", ) - # parser.add_argument( "--wakeup-sound", default=str(_SOUNDS_DIR / "wake_word_triggered.flac") ) parser.add_argument( "--timer-finished-sound", default=str(_SOUNDS_DIR / "timer_finished.flac") ) - # parser.add_argument("--preferences-file", default=_REPO_DIR / "preferences.json") - # parser.add_argument( "--host", default="0.0.0.0", @@ -93,7 +106,6 @@ async def main() -> None: parser.add_argument( "--port", type=int, default=6053, help="Port for ESPHome server (default: 6053)" ) - # parser.add_argument( "--led-interface", choices=["spi", "gpio"], @@ -112,7 +124,6 @@ async def main() -> None: default=12, help="GPIO pin for LED data (for Grove)", ) - # parser.add_argument( "--debug", action="store_true", help="Print DEBUG messages to console" ) @@ -121,23 +132,16 @@ async def main() -> None: logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO) _LOGGER.debug(args) - # Get the asyncio event loop loop = asyncio.get_running_loop() - - # Create the event bus event_bus = EventBus() - # Load available wake words wake_word_dirs = [Path(ww_dir) for ww_dir in args.wake_word_dir] available_wake_words: Dict[str, AvailableWakeWord] = {} - for wake_word_dir in wake_word_dirs: for model_config_path in wake_word_dir.glob("*.json"): model_id = model_config_path.stem if model_id == args.stop_model: - # Don't show stop model as an available wake word continue - with open(model_config_path, "r", encoding="utf-8") as model_config_file: model_config = json.load(model_config_file) model_type = model_config["type"] @@ -151,7 +155,6 @@ async def main() -> None: _LOGGER.debug("Available wake words: %s", list(sorted(available_wake_words.keys()))) - # Load preferences preferences_path = Path(args.preferences_file) if preferences_path.exists(): _LOGGER.debug("Loading preferences: %s", preferences_path) @@ -164,34 +167,26 @@ async def main() -> None: libtensorflowlite_c_path = _LIB_DIR / "libtensorflowlite_c.so" _LOGGER.debug("libtensorflowlite_c path: %s", libtensorflowlite_c_path) - # Load wake/stop models wake_models: Dict[str, Union[MicroWakeWord, OpenWakeWord]] = {} if preferences.active_wake_words: - # Load preferred models for wake_word_id in preferences.active_wake_words: wake_word = available_wake_words.get(wake_word_id) if wake_word is None: _LOGGER.warning("Unrecognized wake word id: %s", wake_word_id) continue - _LOGGER.debug("Loading wake model: %s", wake_word_id) wake_models[wake_word_id] = wake_word.load(libtensorflowlite_c_path) - if not wake_models: - # Load default model wake_word_id = args.wake_model wake_word = available_wake_words[wake_word_id] - _LOGGER.debug("Loading wake model: %s", wake_word_id) wake_models[wake_word_id] = wake_word.load(libtensorflowlite_c_path) - # TODO: allow openWakeWord for "stop" stop_model: Optional[MicroWakeWord] = None for wake_word_dir in wake_word_dirs: stop_config_path = wake_word_dir / f"{args.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, libtensorflowlite_c_path @@ -200,7 +195,6 @@ async def main() -> None: assert stop_model is not None - # Create media players with saved volume music_player = MpvMediaPlayer( device=args.audio_output_device, initial_volume=preferences.volume_level ) @@ -230,13 +224,14 @@ async def main() -> None: refractory_seconds=args.refractory_seconds, ) - # Initialize the LED controller AFTER state is created led_controller = LedController( state, interface=args.led_interface, clock_pin=args.led_clock_pin, data_pin=args.led_data_pin, ) + # --- ADDED: Instantiate Mic Mute Handler --- + mic_mute_handler = MicMuteHandler(state) process_audio_thread = threading.Thread( target=process_audio, args=(state,), daemon=True @@ -250,7 +245,6 @@ async def main() -> None: lambda: VoiceSatelliteProtocol(state), host=args.host, port=args.port ) - # Auto discovery (zeroconf, mDNS) discovery = HomeAssistantZeroconf(port=args.port, name=args.name) await discovery.register_server() @@ -281,60 +275,55 @@ async def main() -> None: def process_audio(state: ServerState): """Process audio chunks from the microphone.""" - + # ... (variable initializations are the same) ... wake_words: List[Union[MicroWakeWord, OpenWakeWord]] = [] micro_features: Optional[MicroWakeWordFeatures] = None micro_inputs: List[np.ndarray] = [] - oww_features: Optional[OpenWakeWordFeatures] = None oww_inputs: List[np.ndarray] = [] has_oww = False - last_active: Optional[float] = None try: while True: - audio_chunk = state.audio_queue.get() - if audio_chunk is None: - break - - if state.satellite is None: + # --- THIS IS THE CHANGE --- + # Check for mute status + if state.mic_muted: + time.sleep(0.1) continue + audio_chunk = state.audio_queue.get() + # ... (rest of the function is the same) ... + if audio_chunk is None: + break + if state.satellite is None: + continue if (not wake_words) or (state.wake_words_changed and state.wake_words): - # Update list of wake word models to process state.wake_words_changed = False wake_words = [ww for ww in state.wake_words.values() if ww.is_active] - has_oww = False for wake_word in wake_words: if isinstance(wake_word, OpenWakeWord): has_oww = True - if micro_features is None: micro_features = MicroWakeWordFeatures( libtensorflowlite_c_path=state.libtensorflowlite_c_path, ) - if has_oww and (oww_features is None): oww_features = OpenWakeWordFeatures( melspectrogram_model=state.oww_melspectrogram_path, embedding_model=state.oww_embedding_path, libtensorflowlite_c_path=state.libtensorflowlite_c_path, ) - try: state.satellite.handle_audio(audio_chunk) - assert micro_features is not None 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): @@ -346,27 +335,21 @@ def process_audio(state: ServerState): for prob in wake_word.process_streaming(oww_input): if prob > 0.5: activated = True - if activated: - # Check refractory now = time.monotonic() if (last_active is None) or ( (now - last_active) > state.refractory_seconds ): state.satellite.wakeup(wake_word) last_active = now - - # Always process to keep state correct stopped = False for micro_input in micro_inputs: if state.stop_word.process_streaming(micro_input): stopped = True - if stopped and state.stop_word.is_active: state.satellite.stop() except Exception: _LOGGER.exception("Unexpected error handling audio") - except Exception: _LOGGER.exception("Unexpected error processing audio")