Spaces:
Running
Running
| # backend/voice/wake_word.py | |
| # Real wake word detection using openWakeWord | |
| # In CLOUD_ENV mode this is a no-op stub — HF Spaces has no microphone. | |
| # On local PC/device, it auto-downloads tflite models and listens for both personas. | |
| import os | |
| import logging | |
| import numpy as np | |
| from typing import Optional | |
| IS_CLOUD = os.environ.get("CLOUD_ENV", "false").lower() == "true" | |
| class WakeWordDetector: | |
| """ | |
| Dual-persona wake word detector. | |
| - Local mode : listens for 'hey jarvis' AND 'hey friday', auto-downloads models. | |
| - Cloud mode : no-op stub (HF Spaces has no microphone — mobile/PC handles wake word). | |
| """ | |
| def __init__(self, keywords: list[str] = None): | |
| self.mute_wake_word_during_tts = False | |
| self._enabled = False | |
| self.model = None | |
| if IS_CLOUD: | |
| logging.info("[WakeWord] Cloud mode — wake word detection disabled (no mic on server).") | |
| return | |
| if keywords is None: | |
| keywords = ["hey jarvis", "hey friday"] | |
| try: | |
| import openwakeword | |
| from openwakeword.model import Model | |
| # Download all built-in pre-trained models on first run | |
| openwakeword.utils.download_models() | |
| # Load only the keywords that have matching tflite files | |
| import openwakeword.utils as oww_utils | |
| available = oww_utils.get_pretrained_model_paths() | |
| available_names = {os.path.splitext(os.path.basename(p))[0].replace("_", " ").lower(): p for p in available} | |
| loadable = [] | |
| for kw in keywords: | |
| clean = kw.lower().replace(" ", "_") | |
| matched = [p for name, p in available_names.items() if clean in name or kw.lower() in name] | |
| if matched: | |
| loadable.append(matched[0]) | |
| else: | |
| logging.warning(f"[WakeWord] No pretrained model found for '{kw}' — skipping.") | |
| if loadable: | |
| self.model = Model(wakeword_models=loadable, inference_framework="tflite") | |
| self._enabled = True | |
| logging.info(f"[WakeWord] Loaded {len(loadable)} wake word model(s): {loadable}") | |
| else: | |
| logging.warning("[WakeWord] No wake word models loaded — all keywords unsupported.") | |
| except Exception as e: | |
| logging.warning(f"[WakeWord] Failed to load wake word models: {e}. Continuing without wake word.") | |
| def process_frame(self, audio_frame: np.ndarray) -> Optional[str]: | |
| if not self._enabled or self.model is None: | |
| return None | |
| if getattr(self, 'mute_wake_word_during_tts', False): | |
| return None | |
| try: | |
| predictions = self.model.predict(audio_frame) | |
| for keyword, score in predictions.items(): | |
| if score > 0.5: | |
| return keyword | |
| except Exception as e: | |
| import logging; logging.getLogger(__name__).error(f"Swallowed exception: {e}") | |
| return None | |