Spaces:
Running
Running
File size: 3,098 Bytes
a31f556 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | # 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
|