File size: 11,104 Bytes
f6aec75 | 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 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 | """Streaming keyword spotting over CTC phoneme posteriors.
Keyword-filler decoding with two robustness measures beyond the textbook
version (both port directly to C on the ESP32):
- minimum phone duration: every phone is expanded to two chained states,
so an alignment must spend >= 2 frames (40 ms) per phone. Kills
spurious single-frame matches.
- duration-normalized scoring: each Viterbi token carries the frame at
which its path entered the keyword, and the detection statistic is
(keyword_score - filler_score) / path_duration -- an average
per-frame deficit, comparable across phrase lengths and speaking
rates. Alignments faster than 2 frames/phone or slower than
MAX_FRAMES_PER_PHONE are rejected outright.
Enrollment of a new wake word is just: phones = text_to_phones(phrase).
State chain per phone i: A_i -> B_i (same label), optional blank after.
"""
import numpy as np
from .phones import BLANK, PHONE_TO_ID, text_to_phones
NEG_INF = -1e30
# 200 ms per phone upper bound: fluent speech runs 40-120 ms/phone, and
# looser caps let garbage alignments crawl across continuous speech
# (observed 1.5-2.3 s "matches" on background TV at 400 ms/phone)
MAX_FRAMES_PER_PHONE = 10
# Highly confusable phone pairs: an alignment may match either member.
# Text pronunciations (CMUdict/G2P) use canonical vowels, but real and TTS
# speech often reduces them -- e.g. "orbit" is listed as AO R B AH T yet
# actually said as AO R B IH T. Without this, a sharper acoustic model
# *punishes* the mismatch harder.
CONFUSABLE = {
"AH": ("AH", "IH", "ER"), # schwa reduces/r-colors: orbit -> orbERt
"IH": ("IH", "AH"),
"ER": ("ER", "AH"),
"AO": ("AO", "AA"), # cot-caught merger and accent variation
"AE": ("AE", "AA"), # trap-father variation (sakura, tanaka)
"UH": ("UH", "UW"), # lax/tense u (book/boot neighbors)
"K": ("K", "G"), # stops voice between vowels: nakuma->naguma
"G": ("G", "K"),
"T": ("T", "D"), # also covers tapped/rolled R heard as D
"D": ("D", "T"),
}
class KeywordSpotter:
def __init__(self, phrase, threshold=-4.4, refractory_frames=50,
strong_margin=1.5, strong_ratio=0.5, phones=None,
wildcards=None, wildcard_penalty=2.2):
"""threshold: average per-frame deficit tolerated along the keyword
path (log-prob units, <= 0; closer to 0 = stricter).
refractory_frames: minimum output frames (20 ms) between fires.
strong_margin/strong_ratio: evidence requirement -- at least
strong_ratio of the path's phone frames must have their phone
within strong_margin log-prob of the frame's best class. Prevents
noisy/mushy audio (where nothing is clearly heard) from firing on
average-score alone.
phones: explicit phoneme sequence (e.g. from voice enrollment);
overrides the dictionary pronunciation of `phrase`."""
if phones is None:
phones = [p for p in text_to_phones(phrase) if p in PHONE_TO_ID]
else:
phones = [p for p in phones if p in PHONE_TO_ID]
if len(phones) < 2:
raise ValueError(f"phrase too short to spot: {phrase!r}")
self.phrase = phrase
self.phones = phones
self.threshold = threshold
self.refractory_frames = refractory_frames
self.strong_margin = strong_margin
self.strong_ratio = strong_ratio
# Mismatch tolerance (default OFF): measured on universal clips
# vs negatives, wildcarding lifted impostor scores as much as
# genuine ones and REDUCED recall at zero-FA operating points
# (vucano negmax -4.14 -> -2.89). Kept only for future
# per-count-lattice experiments.
if wildcards is None:
wildcards = 0
self.max_wild = wildcards * 2 # budget in frames
self.wild_pen = wildcard_penalty
# Build states: leading blank, then per phone A,B (+ trailing blank)
labels = [BLANK]
self.preds = [[0]] # predecessor state ids (self-loop
entry = [True] # implied for every state)
for i, p in enumerate(phones):
pid = PHONE_TO_ID[p]
a = len(labels)
if i == 0:
labels.append(pid); self.preds.append([0]); entry.append(True)
else:
pre = [a - 1] # blank between phones
if labels[a - 2] != pid:
pre.append(a - 2) # skip blank (different phones only)
labels.append(pid); self.preds.append(pre); entry.append(False)
labels.append(pid) # B state: only from A
self.preds.append([a]); entry.append(False)
labels.append(BLANK) # blank after phone
self.preds.append([a + 1]); entry.append(False)
self.labels = labels
# allowed emission ids per state (confusable vowels match either)
self.allowed = []
id_to_phone = {v: k for k, v in PHONE_TO_ID.items()}
for lab in labels:
if lab == BLANK:
self.allowed.append((BLANK,))
else:
ph = id_to_phone[lab]
self.allowed.append(tuple(
PHONE_TO_ID[p] for p in CONFUSABLE.get(ph, (ph,))))
self.entry = entry
self.n_states = len(labels)
self.finals = [self.n_states - 1, self.n_states - 2]
# min: ~60 ms per phone on average (individual phones may be
# shorter); max: 400 ms per phone. Anything outside is not a
# human saying the phrase.
self.min_dur = 3 * len(phones)
self.max_dur = MAX_FRAMES_PER_PHONE * len(phones)
self.reset()
def reset(self):
self.rel = np.full(self.n_states, NEG_INF)
self.start = np.zeros(self.n_states, dtype=np.int64)
self.pframes = np.zeros(self.n_states, dtype=np.int64)
self.strong = np.zeros(self.n_states, dtype=np.int64)
self.wild = np.zeros(self.n_states, dtype=np.int64)
self.cooldown = 0
self.t = 0
def _update(self, log_probs_frame):
"""One Viterbi DP step. Returns the best duration-valid normalized
score at a final state this frame (or None). No side effects on
detection state."""
lp = log_probs_frame
filler = float(lp.max())
prev_rel, prev_start = self.rel, self.start
prev_pf, prev_strong = self.pframes, self.strong
prev_wild = self.wild
cur_rel = np.full(self.n_states, NEG_INF)
cur_start = np.zeros(self.n_states, dtype=np.int64)
cur_pf = np.zeros(self.n_states, dtype=np.int64)
cur_strong = np.zeros(self.n_states, dtype=np.int64)
cur_wild = np.zeros(self.n_states, dtype=np.int64)
for s in range(self.n_states):
best = prev_rel[s]
best_start, best_pf, best_sf, best_w = (
prev_start[s], prev_pf[s], prev_strong[s], prev_wild[s])
for q in self.preds[s]:
if prev_rel[q] > best:
best = prev_rel[q]
best_start, best_pf, best_sf, best_w = (
prev_start[q], prev_pf[q], prev_strong[q],
prev_wild[q])
if self.entry[s] and 0.0 >= best:
# (re)start the keyword here; >= keeps the start time fresh
# while idling in silence, so duration stays meaningful
best, best_start, best_pf, best_sf, best_w = \
0.0, self.t, 0, 0, 0
emit = max(float(lp[i]) for i in self.allowed[s])
is_phone = self.labels[s] != BLANK
used_wild = 0
if is_phone and best_w < self.max_wild:
# mismatch tolerance: accept the frame's best class at a
# penalty when the expected phone isn't there (bounded
# budget converts exact matching into similarity matching)
soft = filler - self.wild_pen
if soft > emit:
emit = soft
used_wild = 1
cur_rel[s] = best + emit - filler
cur_start[s] = best_start
cur_pf[s] = best_pf + (1 if is_phone else 0)
cur_strong[s] = best_sf + (
1 if is_phone and emit >= filler - self.strong_margin else 0)
cur_wild[s] = best_w + used_wild
self.rel, self.start = cur_rel, cur_start
self.pframes, self.strong = cur_pf, cur_strong
self.wild = cur_wild
self.t += 1
norm_best = None
for s in self.finals:
dur = self.t - cur_start[s]
pf = cur_pf[s]
if not (self.min_dur <= dur <= self.max_dur):
continue
if pf < 2 * len(self.phones):
continue
# evidence requirement: the phrase's phones must actually have
# been the near-top hypothesis for enough of the match, not
# merely "not too costly on average" (mushy noisy audio)
if cur_strong[s] < self.strong_ratio * pf:
continue
# normalize by phone frames only: time spent in blank states is
# free in silence, so counting it would let blank-padded paths
# dilute their deficit below any threshold
norm = cur_rel[s] / pf
if norm_best is None or norm > norm_best[0]:
norm_best = (float(norm), int(dur))
return norm_best
def step(self, log_probs_frame):
"""Consume one frame of log-probs (C,). Returns (score, duration)
if the keyword fired this frame, else None. duration is in output
frames (20 ms each), for energy-gating by the caller."""
norm = self._update(log_probs_frame)
if self.cooldown > 0:
self.cooldown -= 1
return None
if norm is not None and norm[0] > self.threshold:
self.cooldown = self.refractory_frames
self.rel = np.full(self.n_states, NEG_INF)
return norm
return None
def run(self, log_probs):
"""Offline helper: (T, C) log-probs -> list of (frame, score)."""
hits = []
for t in range(log_probs.shape[0]):
s = self.step(log_probs[t])
if s is not None:
hits.append((t, s[0]))
return hits
def best_score(self, log_probs):
"""Max normalized final-state score over a clip (pure Viterbi, no
firing/reset side effects) -- used for calibration."""
self.reset()
best = -np.inf
for t in range(log_probs.shape[0]):
s = self._update(log_probs[t])
if s is not None and s[0] > best:
best = s[0]
return best
|