# -*- encoding: utf-8 -*-
|
from pathlib import Path
|
from typing import Any, Dict, Iterable, List, NamedTuple, Set, Tuple, Union
|
|
import numpy as np
|
from typeguard import check_argument_types
|
import kaldi_native_fbank as knf
|
|
root_dir = Path(__file__).resolve().parent
|
|
logger_initialized = {}
|
|
|
class WavFrontend():
|
"""Conventional frontend structure for ASR.
|
"""
|
|
def __init__(
|
self,
|
cmvn_file: str = None,
|
fs: int = 16000,
|
window: str = 'hamming',
|
n_mels: int = 80,
|
frame_length: int = 25.0,
|
frame_shift: int = 10,
|
filter_length_min: int = -1,
|
filter_length_max: float = -1,
|
lfr_m: int = 1,
|
lfr_n: int = 1,
|
dither: float = 1.0
|
) -> None:
|
check_argument_types()
|
|
opts = knf.FbankOptions()
|
opts.frame_opts.samp_freq = fs
|
opts.frame_opts.dither = dither
|
opts.frame_opts.window_type = window
|
opts.frame_opts.frame_shift_ms = float(frame_shift)
|
opts.frame_opts.frame_length_ms = float(frame_length)
|
opts.mel_opts.num_bins = n_mels
|
opts.energy_floor = 0
|
opts.frame_opts.snip_edges = True
|
opts.mel_opts.debug_mel = False
|
self.opts = opts
|
|
self.filter_length_min = filter_length_min
|
self.filter_length_max = filter_length_max
|
self.lfr_m = lfr_m
|
self.lfr_n = lfr_n
|
self.cmvn_file = cmvn_file
|
|
if self.cmvn_file:
|
self.cmvn = self.load_cmvn()
|
|
def fbank(self,
|
waveform: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
|
waveform = waveform * (1 << 15)
|
fbank_fn = knf.OnlineFbank(self.opts)
|
fbank_fn.accept_waveform(self.opts.frame_opts.samp_freq, waveform.tolist())
|
frames = fbank_fn.num_frames_ready
|
mat = np.empty([frames, self.opts.mel_opts.num_bins])
|
for i in range(frames):
|
mat[i, :] = fbank_fn.get_frame(i)
|
feat = mat.astype(np.float32)
|
feat_len = np.array(mat.shape[0]).astype(np.int32)
|
return feat, feat_len
|
|
def lfr_cmvn(self, feat: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
|
if self.lfr_m != 1 or self.lfr_n != 1:
|
feat = self.apply_lfr(feat, self.lfr_m, self.lfr_n)
|
|
if self.cmvn_file:
|
feat = self.apply_cmvn(feat)
|
|
feat_len = np.array(feat.shape[0]).astype(np.int32)
|
return feat, feat_len
|
|
@staticmethod
|
def apply_lfr(inputs: np.ndarray, lfr_m: int, lfr_n: int) -> np.ndarray:
|
LFR_inputs = []
|
|
T = inputs.shape[0]
|
T_lfr = int(np.ceil(T / lfr_n))
|
left_padding = np.tile(inputs[0], ((lfr_m - 1) // 2, 1))
|
inputs = np.vstack((left_padding, inputs))
|
T = T + (lfr_m - 1) // 2
|
for i in range(T_lfr):
|
if lfr_m <= T - i * lfr_n:
|
LFR_inputs.append(
|
(inputs[i * lfr_n:i * lfr_n + lfr_m]).reshape(1, -1))
|
else:
|
# process last LFR frame
|
num_padding = lfr_m - (T - i * lfr_n)
|
frame = inputs[i * lfr_n:].reshape(-1)
|
for _ in range(num_padding):
|
frame = np.hstack((frame, inputs[-1]))
|
|
LFR_inputs.append(frame)
|
LFR_outputs = np.vstack(LFR_inputs).astype(np.float32)
|
return LFR_outputs
|
|
def apply_cmvn(self, inputs: np.ndarray) -> np.ndarray:
|
"""
|
Apply CMVN with mvn data
|
"""
|
frame, dim = inputs.shape
|
means = np.tile(self.cmvn[0:1, :dim], (frame, 1))
|
vars = np.tile(self.cmvn[1:2, :dim], (frame, 1))
|
inputs = (inputs + means) * vars
|
return inputs
|
|
def load_cmvn(self,) -> np.ndarray:
|
with open(self.cmvn_file, 'r', encoding='utf-8') as f:
|
lines = f.readlines()
|
|
means_list = []
|
vars_list = []
|
for i in range(len(lines)):
|
line_item = lines[i].split()
|
if line_item[0] == '<AddShift>':
|
line_item = lines[i + 1].split()
|
if line_item[0] == '<LearnRateCoef>':
|
add_shift_line = line_item[3:(len(line_item) - 1)]
|
means_list = list(add_shift_line)
|
continue
|
elif line_item[0] == '<Rescale>':
|
line_item = lines[i + 1].split()
|
if line_item[0] == '<LearnRateCoef>':
|
rescale_line = line_item[3:(len(line_item) - 1)]
|
vars_list = list(rescale_line)
|
continue
|
|
means = np.array(means_list).astype(np.float64)
|
vars = np.array(vars_list).astype(np.float64)
|
cmvn = np.array([means, vars])
|
return cmvn
|
