# Copyright (c) Alibaba, Inc. and its affiliates.
|
# Part of the implementation is borrowed from espnet/espnet.
|
|
from typing import Tuple
|
|
import numpy as np
|
import torch
|
import torchaudio.compliance.kaldi as kaldi
|
from funasr.models.frontend.abs_frontend import AbsFrontend
|
from typeguard import check_argument_types
|
from torch.nn.utils.rnn import pad_sequence
|
# import kaldifeat
|
|
def load_cmvn(cmvn_file):
|
with open(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.float)
|
vars = np.array(vars_list).astype(np.float)
|
cmvn = np.array([means, vars])
|
cmvn = torch.as_tensor(cmvn)
|
return cmvn
|
|
|
def apply_cmvn(inputs, cmvn_file): # noqa
|
"""
|
Apply CMVN with mvn data
|
"""
|
|
device = inputs.device
|
dtype = inputs.dtype
|
frame, dim = inputs.shape
|
|
cmvn = load_cmvn(cmvn_file)
|
means = np.tile(cmvn[0:1, :dim], (frame, 1))
|
vars = np.tile(cmvn[1:2, :dim], (frame, 1))
|
inputs += torch.from_numpy(means).type(dtype).to(device)
|
inputs *= torch.from_numpy(vars).type(dtype).to(device)
|
|
return inputs.type(torch.float32)
|
|
|
def apply_lfr(inputs, lfr_m, lfr_n):
|
LFR_inputs = []
|
T = inputs.shape[0]
|
T_lfr = int(np.ceil(T / lfr_n))
|
left_padding = inputs[0].repeat((lfr_m - 1) // 2, 1)
|
inputs = torch.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]).view(1, -1))
|
else: # process last LFR frame
|
num_padding = lfr_m - (T - i * lfr_n)
|
frame = (inputs[i * lfr_n:]).view(-1)
|
for _ in range(num_padding):
|
frame = torch.hstack((frame, inputs[-1]))
|
LFR_inputs.append(frame)
|
LFR_outputs = torch.vstack(LFR_inputs)
|
return LFR_outputs.type(torch.float32)
|
|
|
# class WavFrontend_kaldifeat(AbsFrontend):
|
# """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,
|
# frame_shift: int = 10,
|
# lfr_m: int = 1,
|
# lfr_n: int = 1,
|
# dither: float = 1.0,
|
# snip_edges: bool = True,
|
# upsacle_samples: bool = True,
|
# device: str = 'cpu',
|
# **kwargs,
|
# ):
|
# super().__init__()
|
#
|
# opts = kaldifeat.FbankOptions()
|
# opts.device = device
|
# 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 = snip_edges
|
# opts.mel_opts.debug_mel = False
|
# self.opts = opts
|
# self.fbank_fn = None
|
# self.fbank_beg_idx = 0
|
# self.reset_fbank_status()
|
#
|
# self.lfr_m = lfr_m
|
# self.lfr_n = lfr_n
|
# self.cmvn_file = cmvn_file
|
# self.upsacle_samples = upsacle_samples
|
#
|
# def output_size(self) -> int:
|
# return self.n_mels * self.lfr_m
|
#
|
# def forward_fbank(
|
# self,
|
# input: torch.Tensor,
|
# input_lengths: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
|
# batch_size = input.size(0)
|
# feats = []
|
# feats_lens = []
|
# for i in range(batch_size):
|
# waveform_length = input_lengths[i]
|
# waveform = input[i][:waveform_length]
|
# waveform = waveform * (1 << 15)
|
#
|
# self.fbank_fn.accept_waveform(self.opts.frame_opts.samp_freq, waveform.tolist())
|
# frames = self.fbank_fn.num_frames_ready
|
# frames_cur = frames - self.fbank_beg_idx
|
# mat = torch.empty([frames_cur, self.opts.mel_opts.num_bins], dtype=torch.float32).to(
|
# device=self.opts.device)
|
# for i in range(self.fbank_beg_idx, frames):
|
# mat[i, :] = self.fbank_fn.get_frame(i)
|
# self.fbank_beg_idx += frames_cur
|
#
|
# feat_length = mat.size(0)
|
# feats.append(mat)
|
# feats_lens.append(feat_length)
|
#
|
# feats_lens = torch.as_tensor(feats_lens)
|
# feats_pad = pad_sequence(feats,
|
# batch_first=True,
|
# padding_value=0.0)
|
# return feats_pad, feats_lens
|
#
|
# def reset_fbank_status(self):
|
# self.fbank_fn = kaldifeat.OnlineFbank(self.opts)
|
# self.fbank_beg_idx = 0
|
#
|
# def forward_lfr_cmvn(
|
# self,
|
# input: torch.Tensor,
|
# input_lengths: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
|
# batch_size = input.size(0)
|
# feats = []
|
# feats_lens = []
|
# for i in range(batch_size):
|
# mat = input[i, :input_lengths[i], :]
|
# if self.lfr_m != 1 or self.lfr_n != 1:
|
# mat = apply_lfr(mat, self.lfr_m, self.lfr_n)
|
# if self.cmvn_file is not None:
|
# mat = apply_cmvn(mat, self.cmvn_file)
|
# feat_length = mat.size(0)
|
# feats.append(mat)
|
# feats_lens.append(feat_length)
|
#
|
# feats_lens = torch.as_tensor(feats_lens)
|
# feats_pad = pad_sequence(feats,
|
# batch_first=True,
|
# padding_value=0.0)
|
# return feats_pad, feats_lens
|
