sirmrak/linux-voice-assistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Microphone Mute Switch Entity

Browse Source 
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.
This commit is contained in:
imonlinux
2025-10-11 17:10:15 -07:00
committed by GitHub
parent 4d12149dae
commit 39d6d3e010
1 changed files with 31 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
linux_voice_assistant/__main__.py
+31 -48
View File
@@ -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")