File size: 2,508 Bytes
139f25e | 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 | """流式唤醒词检测器：fbank -> 64 帧窗口 -> 模型 -> 3 帧滑动和触发。"""
from __future__ import annotations
import numpy as np
from .inference import ModelSession
from .preprocess import FbankStream
from .postprocess import wake_scores, detect, DET_THRESHOLD_95, DET_THRESHOLD_90
class WakeWordDetector:
"""对齐 reference 触发语义：每 32ms 帧喂入 512 个 int16 样本。
process(frame) 返回 (wake_score, window_sum, triggered)。
window_sum 为最近 det_win 帧 wake 得分之和；triggered = window_sum > threshold。
"""
def __init__(self, model_path: str, det_win: int = 3,
threshold: float = DET_THRESHOLD_95, providers=None):
self.session = ModelSession(model_path, providers=providers)
self.fbank = FbankStream()
self.det_win = det_win
self.threshold = threshold
self.scores: list[float] = []
def process(self, frame: np.ndarray) -> dict:
"""frame: int16 (512,) 32ms PCM。返回 {wake_score, window_sum, triggered}。"""
if frame.shape[0] != 512:
raise ValueError(f"帧长必须为 512（32ms@16k），got {frame.shape[0]}")
win = self.fbank.push(frame)
logits = self.session.run_named([win])[0]
score = float(wake_scores(logits).reshape(-1)[0])
self.scores.append(score)
if len(self.scores) > self.det_win:
self.scores = self.scores[-self.det_win:]
window_sum = float(np.sum(self.scores))
return {
"wake_score": score,
"window_sum": window_sum,
"triggered": bool(window_sum > self.threshold),
"threshold": self.threshold,
}
def reset(self):
self.fbank.reset()
self.scores.clear()
def detect_stream(session, pcm: np.ndarray, det_win: int = 3,
threshold: float = DET_THRESHOLD_95) -> dict:
"""离线处理整段 int16 PCM：返回逐帧得分与触发结果。"""
det = WakeWordDetector.__new__(WakeWordDetector)
det.session = session
det.fbank = FbankStream()
det.det_win = det_win
det.threshold = threshold
det.scores = []
pcm = np.asarray(pcm, dtype=np.int16).reshape(-1)
frames = (pcm.shape[0] - 512) // 512 + 1
per_frame = []
for i in range(frames):
r = det.process(pcm[i * 512 : i * 512 + 512])
per_frame.append(r)
scores = np.array([r["wake_score"] for r in per_frame])
return detect(scores, det_win, threshold)
|