File size: 3,033 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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | import numpy as np
import struct
EPS_F32 = struct.unpack('<f', struct.pack('<I', 0x33D6BF95))[0]
def _hz2mel(hz):
return 2595.0 * np.log10(1.0 + hz / 700.0)
def _mel_filterbank(nfft=512, nfilter=26, low=80, high=7000, sr=16000):
feat_width = nfft // 2 + 1
lowmel = _hz2mel(low)
highmel = _hz2mel(high)
nyquist = sr * 0.5
mel_points = [lowmel + i * (highmel - lowmel) / (nfilter + 1) for i in range(nfilter + 2)]
bin_mels = [_hz2mel(i * nyquist / (nfft // 2)) for i in range(feat_width)]
coeff, bank_pos = [], []
for i in range(nfilter):
start, stop = -1, -1
for j in range(1, feat_width):
lower = (bin_mels[j] - mel_points[i]) / (mel_points[i + 1] - mel_points[i])
upper = (mel_points[i + 2] - bin_mels[j]) / (mel_points[i + 2] - mel_points[i + 1])
temp = min(lower, upper)
if lower > 0 and start == -1:
start = j
if upper <= 0 and stop == -1:
stop = j - 1
if temp > 0:
coeff.append(temp)
bank_pos.append((start, stop))
return np.asarray(coeff, dtype=np.float32), bank_pos
_HANN = np.hanning(513)[:-1].astype(np.float32) # periodic 512 hann == reference window array
_COEFF, _BANK_POS = _mel_filterbank()
_OFFS = []
_off = 0
for _s, _e in _BANK_POS:
_OFFS.append((_s, _e, _off, _off + _e - _s + 1))
_off += _e - _s + 1
def frame_fbank(samples: np.ndarray) -> np.ndarray:
"""单帧 (512,) int16 -> (26,) log-fbank (Q6.10 值，float)。"""
x = samples.astype(np.float32) * (1.0 / 32768.0) * _HANN
spec = np.abs(np.fft.rfft(x, n=512))
out = np.zeros(26, dtype=np.float32)
for f, (s, e, c0, c1) in enumerate(_OFFS):
out[f] = np.dot(spec[s : e + 1], _COEFF[c0:c1])
return np.log(out + EPS_F32)
class FbankStream:
"""流式 fbank：每帧 512 个 int16 样本 -> 维护 64 帧窗口，输出 (1,26,64)。"""
def __init__(self, window: int = 64):
self.window = window
self.buf = np.zeros((window, 26), dtype=np.float32)
def push(self, samples: np.ndarray) -> np.ndarray:
"""喂入一帧 int16 (512,)，返回当前窗口 (1,26,64) 模型输入。"""
f = frame_fbank(samples)
q = np.trunc(np.clip(f, -32, 32) * 1024.0 + 0.5) / 1024.0
self.buf = np.roll(self.buf, -1, axis=0)
self.buf[-1] = q
return self.buf.T[np.newaxis].astype(np.float32)
def reset(self):
self.buf[:] = 0.0
def preprocess(*arrays):
"""兼容通用入口：输入 int16 PCM (N,) 或 (1,N)，返回全部滑窗 (M,1,26,64)。"""
if len(arrays) != 1:
raise ValueError('wakeup 需要单个 int16 PCM 数组')
pcm = np.asarray(arrays[0]).reshape(-1).astype(np.int16)
n_frames = (pcm.shape[0] - 512) // 512 + 1
if n_frames < 1:
raise ValueError('PCM 太短')
fs = FbankStream()
wins = []
for i in range(n_frames):
wins.append(fs.push(pcm[i * 512 : i * 512 + 512]))
return [np.stack(wins)] |