supotato6
8 months ago
parent
fe2db3292f
commit
03f6d9ed6e
@ -0,0 +1,130 @@
|
||||
from paddlespeech.t2s.models.CosyVoice.cosyvoice import CosyVoice2
|
||||
import sys
|
||||
from paddlenlp.transformers import AutoTokenizer, AutoModelForCausalLM
|
||||
from pathlib import Path
|
||||
import paddle
|
||||
import torch
|
||||
paddle.seed(42)
|
||||
from paddlespeech.t2s.frontend.CosyVoiceFrontEnd.frontend import CosyVoiceFrontEnd
|
||||
from paddlespeech.t2s.models.CosyVoice.llm import Qwen2LM,Qwen2Encoder
|
||||
from paddlespeech.t2s.models.CosyVoice.common import ras_sampling
|
||||
from paddlespeech.t2s.frontend.CosyVoiceFrontEnd.tokenizer import get_qwen_tokenizer
|
||||
from hyperpyyaml import load_hyperpyyaml
|
||||
hyper_yaml_path = "/root/paddlejob/workspace/zhangjinghong/CosyVoice/pretrained_models/CosyVoice2-0.5B_paddle/cosyvoice2.yaml"
|
||||
with open(hyper_yaml_path, 'r') as f:
|
||||
configs = load_hyperpyyaml(f)
|
||||
|
||||
# frontend = CosyVoiceFrontEnd(
|
||||
# lambda:get_qwen_tokenizer('/root/paddlejob/workspace/zhangjinghong/CosyVoice/pretrained_models/CosyVoice2-0.5B/CosyVoice-BlankEN',skip_special_tokens=True),
|
||||
# configs['feat_extractor'],
|
||||
# "/root/paddlejob/workspace/zhangjinghong/CosyVoice/pretrained_models/CosyVoice2-0.5B/campplus.onnx",
|
||||
# "/root/paddlejob/workspace/zhangjinghong/CosyVoice/pretrained_models/CosyVoice2-0.5B/speech_tokenizer_v2.onnx",
|
||||
# "/root/paddlejob/workspace/zhangjinghong/CosyVoice/pretrained_models/CosyVoice2-0.5B/spk2info.pt",
|
||||
# configs['allowed_special']
|
||||
# )
|
||||
# prompt_wav = "../CosyVoice/gaoziyuan_10.wav"
|
||||
# tts_text = frontend.text_normalize("定位解决了云存储服务和 data loader 等多个环节的性能波动问题", split=True, text_frontend=True)
|
||||
# prompt_text = frontend.text_normalize('清晨的阳光透过树叶洒在地面上，微风轻轻吹过，带来花草的香气。街边的咖啡店刚开门，传来阵阵烘焙的香味，让人感到放松与愉快。', split=True, text_frontend=True)
|
||||
# model_input = frontend.frontend_zero_shot(tts_text, prompt_text, prompt_wav, 24000,'')
|
||||
# paddle.save(model_input,'model_input.pdparams')
|
||||
# # cosyvoice_model = CosyVoice2("../CosyVoice/pretrained_models/CosyVoice2-0.5B_paddle")
|
||||
# model = AutoModelForCausalLM.from_pretrained('/root/paddlejob/workspace/zhangjinghong/test/pretrained/Qwen/Qwen2-0.5B')
|
||||
# print(type(model))
|
||||
# llm = Qwen2Encoder(model)
|
||||
# qwen_lm = Qwen2LM(896,896,6561,llm,ras_sampling)
|
||||
# state_dict = paddle.load("/root/paddlejob/workspace/zhangjinghong/CosyVoice/pretrained_models/CosyVoice2-0.5B_paddle/llm.pdparams")
|
||||
# qwen_lm.set_state_dict(state_dict)
|
||||
|
||||
|
||||
# new_dict = torch.load("/root/paddlejob/workspace/zhangjinghong/CosyVoice/data.pt")
|
||||
# text = new_dict['text']
|
||||
# text_len = new_dict['text_len']
|
||||
# prompt_text = new_dict['prompt_text']
|
||||
# prompt_text_len = new_dict['prompt_text_len']
|
||||
# prompt_speech_token = new_dict['prompt_speech_token']
|
||||
# prompt_speech_token_len = new_dict['prompt_speech_token_len']
|
||||
# embedding = new_dict['embedding']
|
||||
# uuid = new_dict['uuid']
|
||||
|
||||
|
||||
# text = model_input['text']
|
||||
# text_len = model_input['text_len']
|
||||
# prompt_text = model_input['prompt_text']
|
||||
# prompt_text_len = model_input['prompt_text_len']
|
||||
# prompt_speech_token = model_input['llm_prompt_speech_token']
|
||||
# prompt_speech_token_len = model_input['llm_prompt_speech_token_len']
|
||||
# embedding = model_input['llm_embedding']
|
||||
# uuid = new_dict['uuid']
|
||||
|
||||
# # 统一设备并转换为Paddle张量
|
||||
# device = paddle.CUDAPlace(0) # 使用GPU设备
|
||||
# text_tensor = paddle.to_tensor(text).cuda()
|
||||
# prompt_text_tensor = paddle.to_tensor(prompt_text).cuda()
|
||||
# prompt_speech_token_tensor = paddle.to_tensor(prompt_speech_token).cuda()
|
||||
# embedding_tensor = paddle.to_tensor(embedding, dtype='float32').cuda()
|
||||
# # 确保长度张量也统一设备并正确转换
|
||||
# text_len_tensor = text_len.cuda() if hasattr(text_len, 'cuda') else paddle.to_tensor(text_len).cuda()
|
||||
# prompt_text_len_tensor = prompt_text_len.cuda() if hasattr(prompt_text_len, 'cuda') else paddle.to_tensor(prompt_text_len).cuda()
|
||||
# prompt_speech_token_len_tensor = prompt_speech_token_len.cuda() if hasattr(prompt_speech_token_len, 'cuda') else paddle.to_tensor(prompt_speech_token_len).cuda()
|
||||
# token=[]
|
||||
# for i in qwen_lm.inference(text=text_tensor,
|
||||
# text_len=text_len_tensor,
|
||||
# prompt_text=prompt_text_tensor,
|
||||
# prompt_text_len=prompt_text_len_tensor,
|
||||
# prompt_speech_token=prompt_speech_token_tensor,
|
||||
# prompt_speech_token_len=prompt_speech_token_len_tensor,
|
||||
# embedding=embedding_tensor,
|
||||
# uuid=uuid):
|
||||
# token.append(i)
|
||||
# # print(text)
|
||||
# print("token: ",i)
|
||||
|
||||
############################################################################################################################
|
||||
|
||||
|
||||
flow = configs['flow']
|
||||
flow_state_dict = paddle.load("/root/paddlejob/workspace/zhangjinghong/CosyVoice/pretrained_models/CosyVoice2-0.5B_paddle/flow.pdparams")
|
||||
flow.set_state_dict(flow_state_dict)
|
||||
input_dict = torch.load("/root/paddlejob/workspace/zhangjinghong/test/CosyVoice/data.pt")
|
||||
flow.eval()
|
||||
tts_mel, _ = flow.inference(
|
||||
token = paddle.to_tensor(input_dict['token']),
|
||||
token_len = paddle.to_tensor(input_dict['token_len']),
|
||||
prompt_token = paddle.to_tensor(input_dict['prompt_token'].cpu().numpy(), dtype = 'int32'),
|
||||
prompt_token_len = paddle.to_tensor(input_dict['prompt_token_len'].cpu().numpy()),
|
||||
prompt_feat = paddle.to_tensor(input_dict['prompt_feat'].cpu().numpy()),
|
||||
prompt_feat_len = paddle.to_tensor(input_dict['prompt_feat_len'].cpu().numpy()),
|
||||
embedding = paddle.to_tensor(input_dict['embedding'].cpu().numpy()),
|
||||
streaming = input_dict['streaming'],
|
||||
finalize = input_dict['finalize']
|
||||
)
|
||||
paddle.save(tts_mel,"tts_mel.pdparams")
|
||||
|
||||
############################################################################################################################
|
||||
|
||||
from paddlespeech.t2s.models.hifigan.cosy_hifigan import HiFTGenerator
|
||||
from paddlespeech.t2s.models.hifigan.f0_predictor import ConvRNNF0Predictor
|
||||
hift_state_dict = paddle.load("/root/paddlejob/workspace/zhangjinghong/CosyVoice/pretrained_models/CosyVoice2-0.5B_paddle/hift.pdparams")
|
||||
input_mel = paddle.to_tensor(torch.load("../CosyVoice/tts_mel.pt").detach().cpu().numpy()).cuda()
|
||||
hift_cache_source= paddle.to_tensor(torch.load("../CosyVoice/hift_cache_source.pt").detach().cpu().numpy()).cuda()
|
||||
# hift_cache_source = paddle.zeros([1, 1, 0])
|
||||
hift_configs = configs['hift']
|
||||
f0_config = configs['f0_predictor']
|
||||
f0_predictor = ConvRNNF0Predictor(**f0_config)
|
||||
|
||||
hift_configs['f0_predictor'] = f0_predictor
|
||||
hift = HiFTGenerator(**hift_configs)
|
||||
hift.set_state_dict(hift_state_dict)
|
||||
# for k,v in hift.state_dict().items():
|
||||
# print(k,v.shape)
|
||||
# print("---"*40)
|
||||
for k,v in hift_state_dict.items():
|
||||
print(k,v.shape)
|
||||
tts_speech, tts_source = hift.inference(speech_feat=input_mel, cache_source=hift_cache_source)
|
||||
paddle.save(tts_speech,"speech.pdparams")
|
||||
# tts_speech,_ = hift.inference(input_dict['tts_mel'],input_dict['cache_source'])
|
||||
|
||||
import torchaudio
|
||||
import torch
|
||||
torchaudio.save("paddle.wav",torch.tensor(tts_speech.numpy()),24000)
|
||||
# sf.write("paddle.wav",tts_speech[0],24000)
|
||||
@ -0,0 +1,110 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
|
||||
import paddle
|
||||
import torchaudio
|
||||
import paddlespeech
|
||||
|
||||
logging.getLogger("matplotlib").setLevel(logging.WARNING)
|
||||
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s %(levelname)s %(message)s")
|
||||
|
||||
|
||||
def read_lists(list_file):
|
||||
lists = []
|
||||
with open(list_file, "r", encoding="utf8") as fin:
|
||||
for line in fin:
|
||||
lists.append(line.strip())
|
||||
return lists
|
||||
|
||||
|
||||
def read_json_lists(list_file):
|
||||
lists = read_lists(list_file)
|
||||
results = {}
|
||||
for fn in lists:
|
||||
with open(fn, "r", encoding="utf8") as fin:
|
||||
results.update(json.load(fin))
|
||||
return results
|
||||
|
||||
|
||||
def load_wav(wav, target_sr, min_sr=16000):
|
||||
speech, sample_rate = torchaudio.load(wav, backend="soundfile")
|
||||
speech = speech.mean(dim=0, keepdim=True)
|
||||
if sample_rate != target_sr:
|
||||
assert (
|
||||
sample_rate >= min_sr
|
||||
), "wav sample rate {} must be greater than {}".format(sample_rate, target_sr)
|
||||
speech = torchaudio.transforms.Resample(
|
||||
orig_freq=sample_rate, new_freq=target_sr
|
||||
)(speech)
|
||||
return speech
|
||||
|
||||
|
||||
def convert_onnx_to_trt(trt_model, trt_kwargs, onnx_model, fp16):
|
||||
import tensorrt as trt
|
||||
|
||||
logging.info("Converting onnx to trt...")
|
||||
network_flags = 1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
|
||||
logger = trt.Logger(trt.Logger.INFO)
|
||||
builder = trt.Builder(logger)
|
||||
network = builder.create_network(network_flags)
|
||||
parser = trt.OnnxParser(network, logger)
|
||||
config = builder.create_builder_config()
|
||||
config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 << 32)
|
||||
if fp16:
|
||||
config.set_flag(trt.BuilderFlag.FP16)
|
||||
profile = builder.create_optimization_profile()
|
||||
with open(onnx_model, "rb") as f:
|
||||
if not parser.parse(f.read()):
|
||||
for error in range(parser.num_errors):
|
||||
print(parser.get_error(error))
|
||||
raise ValueError("failed to parse {}".format(onnx_model))
|
||||
for i in range(len(trt_kwargs["input_names"])):
|
||||
profile.set_shape(
|
||||
trt_kwargs["input_names"][i],
|
||||
trt_kwargs["min_shape"][i],
|
||||
trt_kwargs["opt_shape"][i],
|
||||
trt_kwargs["max_shape"][i],
|
||||
)
|
||||
tensor_dtype = trt.DataType.HALF if fp16 else trt.DataType.FLOAT
|
||||
for i in range(network.num_inputs):
|
||||
input_tensor = network.get_input(i)
|
||||
input_tensor.dtype = tensor_dtype
|
||||
for i in range(network.num_outputs):
|
||||
output_tensor = network.get_output(i)
|
||||
output_tensor.dtype = tensor_dtype
|
||||
config.add_optimization_profile(profile)
|
||||
engine_bytes = builder.build_serialized_network(network, config)
|
||||
with open(trt_model, "wb") as f:
|
||||
f.write(engine_bytes)
|
||||
logging.info("Succesfully convert onnx to trt...")
|
||||
|
||||
|
||||
def export_cosyvoice2_vllm(model, model_path, device):
|
||||
if os.path.exists(model_path):
|
||||
return
|
||||
dtype = paddle.bfloat16
|
||||
use_bias = True if model.llm_decoder.bias is not None else False
|
||||
model.llm.model.lm_head = model.llm_decoder
|
||||
embed_tokens = model.llm.model.model.embed_tokens
|
||||
model.llm.model.set_input_embeddings(model.speech_embedding)
|
||||
model.llm.model.to(device)
|
||||
model.llm.model.to(dtype)
|
||||
tmp_vocab_size = model.llm.model.config.vocab_size
|
||||
tmp_tie_embedding = model.llm.model.config.tie_word_embeddings
|
||||
del model.llm.model.generation_config.eos_token_id
|
||||
del model.llm.model.config.bos_token_id
|
||||
del model.llm.model.config.eos_token_id
|
||||
model.llm.model.config.vocab_size = model.speech_embedding.num_embeddings
|
||||
model.llm.model.config.tie_word_embeddings = False
|
||||
model.llm.model.config.use_bias = use_bias
|
||||
model.llm.model.save_pretrained(model_path)
|
||||
if use_bias is True:
|
||||
os.system(
|
||||
"sed -i s@Qwen2ForCausalLM@CosyVoice2ForCausalLM@g {}/config.json".format(
|
||||
os.path.abspath(model_path)
|
||||
)
|
||||
)
|
||||
model.llm.model.config.vocab_size = tmp_vocab_size
|
||||
model.llm.model.config.tie_word_embeddings = tmp_tie_embedding
|
||||
model.llm.model.set_input_embeddings(embed_tokens)
|
||||
@ -0,0 +1,747 @@
|
||||
import math
|
||||
from typing import Tuple
|
||||
|
||||
import paddle
|
||||
|
||||
import paddlespeech
|
||||
|
||||
__all__ = [
|
||||
"get_mel_banks",
|
||||
"inverse_mel_scale",
|
||||
"inverse_mel_scale_scalar",
|
||||
"mel_scale",
|
||||
"mel_scale_scalar",
|
||||
"spectrogram",
|
||||
"fbank",
|
||||
"mfcc",
|
||||
"vtln_warp_freq",
|
||||
"vtln_warp_mel_freq",
|
||||
]
|
||||
EPSILON = paddle.to_tensor(paddle.finfo(paddle.float32).eps)
|
||||
MILLISECONDS_TO_SECONDS = 0.001
|
||||
HAMMING = "hamming"
|
||||
HANNING = "hanning"
|
||||
POVEY = "povey"
|
||||
RECTANGULAR = "rectangular"
|
||||
BLACKMAN = "blackman"
|
||||
WINDOWS = [HAMMING, HANNING, POVEY, RECTANGULAR, BLACKMAN]
|
||||
|
||||
|
||||
def _get_epsilon(device, dtype):
|
||||
return EPSILON.to(device=device, dtype=dtype)
|
||||
|
||||
|
||||
def _next_power_of_2(x: int) -> int:
|
||||
"""Returns the smallest power of 2 that is greater than x"""
|
||||
return 1 if x == 0 else 2 ** (x - 1).bit_length()
|
||||
|
||||
|
||||
def _get_strided(
|
||||
waveform: paddle.Tensor, window_size: int, window_shift: int, snip_edges: bool
|
||||
) -> paddle.Tensor:
|
||||
"""Given a waveform (1D tensor of size ``num_samples``), it returns a 2D tensor (m, ``window_size``)
|
||||
representing how the window is shifted along the waveform. Each row is a frame.
|
||||
|
||||
Args:
|
||||
waveform (Tensor): Tensor of size ``num_samples``
|
||||
window_size (int): Frame length
|
||||
window_shift (int): Frame shift
|
||||
snip_edges (bool): If True, end effects will be handled by outputting only frames that completely fit
|
||||
in the file, and the number of frames depends on the frame_length. If False, the number of frames
|
||||
depends only on the frame_shift, and we reflect the data at the ends.
|
||||
|
||||
Returns:
|
||||
Tensor: 2D tensor of size (m, ``window_size``) where each row is a frame
|
||||
"""
|
||||
assert waveform.dim() == 1
|
||||
num_samples = waveform.shape[0]
|
||||
strides = window_shift * waveform.strides[0], waveform.strides[0]
|
||||
if snip_edges:
|
||||
if num_samples < window_size:
|
||||
return paddle.empty((0, 0), dtype=waveform.dtype, device=waveform.device)
|
||||
else:
|
||||
m = 1 + (num_samples - window_size) // window_shift
|
||||
else:
|
||||
reversed_waveform = paddle.flip(x=waveform, axis=[0])
|
||||
m = (num_samples + window_shift // 2) // window_shift
|
||||
pad = window_size // 2 - window_shift // 2
|
||||
pad_right = reversed_waveform
|
||||
if pad > 0:
|
||||
pad_left = reversed_waveform[-pad:]
|
||||
waveform = paddle.cat((pad_left, waveform, pad_right), axis=0)
|
||||
else:
|
||||
waveform = paddle.cat((waveform[-pad:], pad_right), axis=0)
|
||||
sizes = m, window_size
|
||||
return waveform.as_strided(shape=sizes, stride=strides)
|
||||
|
||||
|
||||
def _feature_window_function(
|
||||
window_type: str,
|
||||
window_size: int,
|
||||
blackman_coeff: float,
|
||||
device: paddle.device,
|
||||
dtype: int,
|
||||
) -> paddle.Tensor:
|
||||
"""Returns a window function with the given type and size"""
|
||||
if window_type == HANNING:
|
||||
return paddle.audio.functional.get_window(
|
||||
win_length=window_size, fftbins=False, dtype=dtype, window="hann"
|
||||
)
|
||||
elif window_type == HAMMING:
|
||||
return paddle.hamming_window(
|
||||
window_size,
|
||||
periodic=False,
|
||||
alpha=0.54,
|
||||
beta=0.46,
|
||||
device=device,
|
||||
dtype=dtype,
|
||||
)
|
||||
elif window_type == POVEY:
|
||||
return paddle.audio.functional.get_window(
|
||||
win_length=window_size, fftbins=False, dtype=dtype, window="hann"
|
||||
).pow(0.85)
|
||||
elif window_type == RECTANGULAR:
|
||||
return paddle.ones(window_size, device=device, dtype=dtype)
|
||||
elif window_type == BLACKMAN:
|
||||
a = 2 * math.pi / (window_size - 1)
|
||||
window_function = paddle.arange(window_size, device=device, dtype=dtype)
|
||||
return (
|
||||
blackman_coeff
|
||||
- 0.5 * paddle.cos(a * window_function)
|
||||
+ (0.5 - blackman_coeff) * paddle.cos(2 * a * window_function)
|
||||
).to(device=device, dtype=dtype)
|
||||
else:
|
||||
raise Exception("Invalid window type " + window_type)
|
||||
|
||||
|
||||
def _get_log_energy(
|
||||
strided_input: paddle.Tensor, epsilon: paddle.Tensor, energy_floor: float
|
||||
) -> paddle.Tensor:
|
||||
"""Returns the log energy of size (m) for a strided_input (m,*)"""
|
||||
place, dtype = strided_input.place, strided_input.dtype
|
||||
log_energy = paddle.maximum(strided_input.pow(2).sum(axis=1), epsilon).log()
|
||||
if energy_floor == 0.0:
|
||||
return log_energy
|
||||
return paddle.maximum(
|
||||
log_energy, paddle.to_tensor(math.log(energy_floor), place=place, dtype=dtype)
|
||||
)
|
||||
|
||||
|
||||
def _get_waveform_and_window_properties(
|
||||
waveform: paddle.Tensor,
|
||||
channel: int,
|
||||
sample_frequency: float,
|
||||
frame_shift: float,
|
||||
frame_length: float,
|
||||
round_to_power_of_two: bool,
|
||||
preemphasis_coefficient: float,
|
||||
) -> Tuple[paddle.Tensor, int, int, int]:
|
||||
"""Gets the waveform and window properties"""
|
||||
channel = max(channel, 0)
|
||||
assert channel < waveform.shape[0], "Invalid channel {} for size {}".format(
|
||||
channel, waveform.shape[0]
|
||||
)
|
||||
waveform = waveform[channel, :]
|
||||
window_shift = int(sample_frequency * frame_shift * MILLISECONDS_TO_SECONDS)
|
||||
window_size = int(sample_frequency * frame_length * MILLISECONDS_TO_SECONDS)
|
||||
padded_window_size = (
|
||||
_next_power_of_2(window_size) if round_to_power_of_two else window_size
|
||||
)
|
||||
assert (
|
||||
2 <= window_size <= len(waveform)
|
||||
), "choose a window size {} that is [2, {}]".format(window_size, len(waveform))
|
||||
assert 0 < window_shift, "`window_shift` must be greater than 0"
|
||||
assert (
|
||||
padded_window_size % 2 == 0
|
||||
), "the padded `window_size` must be divisible by two. use `round_to_power_of_two` or change `frame_length`"
|
||||
assert (
|
||||
0.0 <= preemphasis_coefficient <= 1.0
|
||||
), "`preemphasis_coefficient` must be between [0,1]"
|
||||
assert sample_frequency > 0, "`sample_frequency` must be greater than zero"
|
||||
return waveform, window_shift, window_size, padded_window_size
|
||||
|
||||
|
||||
def _get_window(
|
||||
waveform: paddle.Tensor,
|
||||
padded_window_size: int,
|
||||
window_size: int,
|
||||
window_shift: int,
|
||||
window_type: str,
|
||||
blackman_coeff: float,
|
||||
snip_edges: bool,
|
||||
raw_energy: bool,
|
||||
energy_floor: float,
|
||||
dither: float,
|
||||
remove_dc_offset: bool,
|
||||
preemphasis_coefficient: float,
|
||||
) -> Tuple[paddle.Tensor, paddle.Tensor]:
|
||||
"""Gets a window and its log energy
|
||||
|
||||
Returns:
|
||||
(Tensor, Tensor): strided_input of size (m, ``padded_window_size``) and signal_log_energy of size (m)
|
||||
"""
|
||||
|
||||
place, dtype = waveform.place, waveform.dtype
|
||||
epsilon = _get_epsilon(place, dtype)
|
||||
strided_input = _get_strided(waveform, window_size, window_shift, snip_edges)
|
||||
if dither != 0.0:
|
||||
rand_gauss = paddle.randn(strided_input.shape, place=place, dtype=dtype)
|
||||
strided_input = strided_input + rand_gauss * dither
|
||||
if remove_dc_offset:
|
||||
row_means = paddle.mean(strided_input, axis=1).unsqueeze(1)
|
||||
strided_input = strided_input - row_means
|
||||
if raw_energy:
|
||||
signal_log_energy = _get_log_energy(strided_input, epsilon, energy_floor)
|
||||
if preemphasis_coefficient != 0.0:
|
||||
offset_strided_input = paddle.nn.functional.pad(
|
||||
strided_input.unsqueeze(0), (1, 0), mode="replicate"
|
||||
).squeeze(0)
|
||||
strided_input = (
|
||||
strided_input - preemphasis_coefficient * offset_strided_input[:, :-1]
|
||||
)
|
||||
window_function = _feature_window_function(
|
||||
window_type, window_size, blackman_coeff, place, dtype
|
||||
).unsqueeze(0)
|
||||
strided_input = strided_input * window_function
|
||||
if padded_window_size != window_size:
|
||||
padding_right = padded_window_size - window_size
|
||||
strided_input = paddle.nn.functional.pad(
|
||||
strided_input.unsqueeze(0), (0, padding_right), mode="constant", value=0
|
||||
).squeeze(0)
|
||||
if not raw_energy:
|
||||
signal_log_energy = _get_log_energy(strided_input, epsilon, energy_floor)
|
||||
return strided_input, signal_log_energy
|
||||
|
||||
|
||||
def _subtract_column_mean(tensor: paddle.Tensor, subtract_mean: bool) -> paddle.Tensor:
|
||||
if subtract_mean:
|
||||
col_means = paddle.mean(tensor, axis=0).unsqueeze(0)
|
||||
tensor = tensor - col_means
|
||||
return tensor
|
||||
|
||||
|
||||
def spectrogram(
|
||||
waveform: paddle.Tensor,
|
||||
blackman_coeff: float = 0.42,
|
||||
channel: int = -1,
|
||||
dither: float = 0.0,
|
||||
energy_floor: float = 1.0,
|
||||
frame_length: float = 25.0,
|
||||
frame_shift: float = 10.0,
|
||||
min_duration: float = 0.0,
|
||||
preemphasis_coefficient: float = 0.97,
|
||||
raw_energy: bool = True,
|
||||
remove_dc_offset: bool = True,
|
||||
round_to_power_of_two: bool = True,
|
||||
sample_frequency: float = 16000.0,
|
||||
snip_edges: bool = True,
|
||||
subtract_mean: bool = False,
|
||||
window_type: str = POVEY,
|
||||
) -> paddle.Tensor:
|
||||
"""Create a spectrogram from a raw audio signal. This matches the input/output of Kaldi's
|
||||
compute-spectrogram-feats.
|
||||
|
||||
Args:
|
||||
waveform (Tensor): Tensor of audio of size (c, n) where c is in the range [0,2)
|
||||
blackman_coeff (float, optional): Constant coefficient for generalized Blackman window. (Default: ``0.42``)
|
||||
channel (int, optional): Channel to extract (-1 -> expect mono, 0 -> left, 1 -> right) (Default: ``-1``)
|
||||
dither (float, optional): Dithering constant (0.0 means no dither). If you turn this off, you should set
|
||||
the energy_floor option, e.g. to 1.0 or 0.1 (Default: ``0.0``)
|
||||
energy_floor (float, optional): Floor on energy (absolute, not relative) in Spectrogram computation. Caution:
|
||||
this floor is applied to the zeroth component, representing the total signal energy. The floor on the
|
||||
individual spectrogram elements is fixed at std::numeric_limits<float>::epsilon(). (Default: ``1.0``)
|
||||
frame_length (float, optional): Frame length in milliseconds (Default: ``25.0``)
|
||||
frame_shift (float, optional): Frame shift in milliseconds (Default: ``10.0``)
|
||||
min_duration (float, optional): Minimum duration of segments to process (in seconds). (Default: ``0.0``)
|
||||
preemphasis_coefficient (float, optional): Coefficient for use in signal preemphasis (Default: ``0.97``)
|
||||
raw_energy (bool, optional): If True, compute energy before preemphasis and windowing (Default: ``True``)
|
||||
remove_dc_offset (bool, optional): Subtract mean from waveform on each frame (Default: ``True``)
|
||||
round_to_power_of_two (bool, optional): If True, round window size to power of two by zero-padding input
|
||||
to FFT. (Default: ``True``)
|
||||
sample_frequency (float, optional): Waveform data sample frequency (must match the waveform file, if
|
||||
specified there) (Default: ``16000.0``)
|
||||
snip_edges (bool, optional): If True, end effects will be handled by outputting only frames that completely fit
|
||||
in the file, and the number of frames depends on the frame_length. If False, the number of frames
|
||||
depends only on the frame_shift, and we reflect the data at the ends. (Default: ``True``)
|
||||
subtract_mean (bool, optional): Subtract mean of each feature file [CMS]; not recommended to do
|
||||
it this way. (Default: ``False``)
|
||||
window_type (str, optional): Type of window ('hamming'|'hanning'|'povey'|'rectangular'|'blackman')
|
||||
(Default: ``'povey'``)
|
||||
|
||||
Returns:
|
||||
Tensor: A spectrogram identical to what Kaldi would output. The shape is
|
||||
(m, ``padded_window_size // 2 + 1``) where m is calculated in _get_strided
|
||||
"""
|
||||
device, dtype = waveform.device, waveform.dtype
|
||||
epsilon = _get_epsilon(device, dtype)
|
||||
(
|
||||
waveform,
|
||||
window_shift,
|
||||
window_size,
|
||||
padded_window_size,
|
||||
) = _get_waveform_and_window_properties(
|
||||
waveform,
|
||||
channel,
|
||||
sample_frequency,
|
||||
frame_shift,
|
||||
frame_length,
|
||||
round_to_power_of_two,
|
||||
preemphasis_coefficient,
|
||||
)
|
||||
if len(waveform) < min_duration * sample_frequency:
|
||||
return paddle.empty(0)
|
||||
strided_input, signal_log_energy = _get_window(
|
||||
waveform,
|
||||
padded_window_size,
|
||||
window_size,
|
||||
window_shift,
|
||||
window_type,
|
||||
blackman_coeff,
|
||||
snip_edges,
|
||||
raw_energy,
|
||||
energy_floor,
|
||||
dither,
|
||||
remove_dc_offset,
|
||||
preemphasis_coefficient,
|
||||
)
|
||||
fft = paddle.fft.rfft(strided_input)
|
||||
power_spectrum = paddle.maximum(fft.abs().pow(2.0), epsilon).log()
|
||||
power_spectrum[:, 0] = signal_log_energy
|
||||
power_spectrum = _subtract_column_mean(power_spectrum, subtract_mean)
|
||||
return power_spectrum
|
||||
|
||||
|
||||
def inverse_mel_scale_scalar(mel_freq: float) -> float:
|
||||
return 700.0 * (math.exp(mel_freq / 1127.0) - 1.0)
|
||||
|
||||
|
||||
def inverse_mel_scale(mel_freq: paddle.Tensor) -> paddle.Tensor:
|
||||
return 700.0 * ((mel_freq / 1127.0).exp() - 1.0)
|
||||
|
||||
|
||||
def mel_scale_scalar(freq: float) -> float:
|
||||
return 1127.0 * math.log(1.0 + freq / 700.0)
|
||||
|
||||
|
||||
def mel_scale(freq: paddle.Tensor) -> paddle.Tensor:
|
||||
return 1127.0 * (1.0 + freq / 700.0).log()
|
||||
|
||||
|
||||
def vtln_warp_freq(
|
||||
vtln_low_cutoff: float,
|
||||
vtln_high_cutoff: float,
|
||||
low_freq: float,
|
||||
high_freq: float,
|
||||
vtln_warp_factor: float,
|
||||
freq: paddle.Tensor,
|
||||
) -> paddle.Tensor:
|
||||
"""This computes a VTLN warping function that is not the same as HTK's one,
|
||||
but has similar inputs (this function has the advantage of never producing
|
||||
empty bins).
|
||||
|
||||
This function computes a warp function F(freq), defined between low_freq
|
||||
and high_freq inclusive, with the following properties:
|
||||
F(low_freq) == low_freq
|
||||
F(high_freq) == high_freq
|
||||
The function is continuous and piecewise linear with two inflection
|
||||
points.
|
||||
The lower inflection point (measured in terms of the unwarped
|
||||
frequency) is at frequency l, determined as described below.
|
||||
The higher inflection point is at a frequency h, determined as
|
||||
described below.
|
||||
If l <= f <= h, then F(f) = f/vtln_warp_factor.
|
||||
If the higher inflection point (measured in terms of the unwarped
|
||||
frequency) is at h, then max(h, F(h)) == vtln_high_cutoff.
|
||||
Since (by the last point) F(h) == h/vtln_warp_factor, then
|
||||
max(h, h/vtln_warp_factor) == vtln_high_cutoff, so
|
||||
h = vtln_high_cutoff / max(1, 1/vtln_warp_factor).
|
||||
= vtln_high_cutoff * min(1, vtln_warp_factor).
|
||||
If the lower inflection point (measured in terms of the unwarped
|
||||
frequency) is at l, then min(l, F(l)) == vtln_low_cutoff
|
||||
This implies that l = vtln_low_cutoff / min(1, 1/vtln_warp_factor)
|
||||
= vtln_low_cutoff * max(1, vtln_warp_factor)
|
||||
Args:
|
||||
vtln_low_cutoff (float): Lower frequency cutoffs for VTLN
|
||||
vtln_high_cutoff (float): Upper frequency cutoffs for VTLN
|
||||
low_freq (float): Lower frequency cutoffs in mel computation
|
||||
high_freq (float): Upper frequency cutoffs in mel computation
|
||||
vtln_warp_factor (float): Vtln warp factor
|
||||
freq (Tensor): given frequency in Hz
|
||||
|
||||
Returns:
|
||||
Tensor: Freq after vtln warp
|
||||
"""
|
||||
assert (
|
||||
vtln_low_cutoff > low_freq
|
||||
), "be sure to set the vtln_low option higher than low_freq"
|
||||
assert (
|
||||
vtln_high_cutoff < high_freq
|
||||
), "be sure to set the vtln_high option lower than high_freq [or negative]"
|
||||
l = vtln_low_cutoff * max(1.0, vtln_warp_factor)
|
||||
h = vtln_high_cutoff * min(1.0, vtln_warp_factor)
|
||||
scale = 1.0 / vtln_warp_factor
|
||||
Fl = scale * l
|
||||
Fh = scale * h
|
||||
assert l > low_freq and h < high_freq
|
||||
scale_left = (Fl - low_freq) / (l - low_freq)
|
||||
scale_right = (high_freq - Fh) / (high_freq - h)
|
||||
res = paddle.empty_like(freq)
|
||||
outside_low_high_freq = paddle.lt(freq, low_freq) | paddle.gt(freq, high_freq)
|
||||
before_l = paddle.lt(freq, l)
|
||||
before_h = paddle.lt(freq, h)
|
||||
after_h = paddle.ge(freq, h)
|
||||
res[after_h] = high_freq + scale_right * (freq[after_h] - high_freq)
|
||||
res[before_h] = scale * freq[before_h]
|
||||
res[before_l] = low_freq + scale_left * (freq[before_l] - low_freq)
|
||||
res[outside_low_high_freq] = freq[outside_low_high_freq]
|
||||
return res
|
||||
|
||||
|
||||
def vtln_warp_mel_freq(
|
||||
vtln_low_cutoff: float,
|
||||
vtln_high_cutoff: float,
|
||||
low_freq,
|
||||
high_freq: float,
|
||||
vtln_warp_factor: float,
|
||||
mel_freq: paddle.Tensor,
|
||||
) -> paddle.Tensor:
|
||||
"""
|
||||
Args:
|
||||
vtln_low_cutoff (float): Lower frequency cutoffs for VTLN
|
||||
vtln_high_cutoff (float): Upper frequency cutoffs for VTLN
|
||||
low_freq (float): Lower frequency cutoffs in mel computation
|
||||
high_freq (float): Upper frequency cutoffs in mel computation
|
||||
vtln_warp_factor (float): Vtln warp factor
|
||||
mel_freq (Tensor): Given frequency in Mel
|
||||
|
||||
Returns:
|
||||
Tensor: ``mel_freq`` after vtln warp
|
||||
"""
|
||||
return mel_scale(
|
||||
vtln_warp_freq(
|
||||
vtln_low_cutoff,
|
||||
vtln_high_cutoff,
|
||||
low_freq,
|
||||
high_freq,
|
||||
vtln_warp_factor,
|
||||
inverse_mel_scale(mel_freq),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def get_mel_banks(
|
||||
num_bins: int,
|
||||
window_length_padded: int,
|
||||
sample_freq: float,
|
||||
low_freq: float,
|
||||
high_freq: float,
|
||||
vtln_low: float,
|
||||
vtln_high: float,
|
||||
vtln_warp_factor: float,
|
||||
) -> Tuple[paddle.Tensor, paddle.Tensor]:
|
||||
"""
|
||||
Returns:
|
||||
(Tensor, Tensor): The tuple consists of ``bins`` (which is
|
||||
melbank of size (``num_bins``, ``num_fft_bins``)) and ``center_freqs`` (which is
|
||||
center frequencies of bins of size (``num_bins``)).
|
||||
"""
|
||||
assert num_bins > 3, "Must have at least 3 mel bins"
|
||||
assert window_length_padded % 2 == 0
|
||||
num_fft_bins = window_length_padded / 2
|
||||
nyquist = 0.5 * sample_freq
|
||||
if high_freq <= 0.0:
|
||||
high_freq += nyquist
|
||||
assert (
|
||||
0.0 <= low_freq < nyquist
|
||||
and 0.0 < high_freq <= nyquist
|
||||
and low_freq < high_freq
|
||||
), "Bad values in options: low-freq {} and high-freq {} vs. nyquist {}".format(
|
||||
low_freq, high_freq, nyquist
|
||||
)
|
||||
fft_bin_width = sample_freq / window_length_padded
|
||||
mel_low_freq = mel_scale_scalar(low_freq)
|
||||
mel_high_freq = mel_scale_scalar(high_freq)
|
||||
mel_freq_delta = (mel_high_freq - mel_low_freq) / (num_bins + 1)
|
||||
if vtln_high < 0.0:
|
||||
vtln_high += nyquist
|
||||
assert (
|
||||
vtln_warp_factor == 1.0
|
||||
or low_freq < vtln_low < high_freq
|
||||
and 0.0 < vtln_high < high_freq
|
||||
and vtln_low < vtln_high
|
||||
), "Bad values in options: vtln-low {} and vtln-high {}, versus low-freq {} and high-freq {}".format(
|
||||
vtln_low, vtln_high, low_freq, high_freq
|
||||
)
|
||||
bin = paddle.arange(num_bins).unsqueeze(1)
|
||||
left_mel = mel_low_freq + bin * mel_freq_delta
|
||||
center_mel = mel_low_freq + (bin + 1.0) * mel_freq_delta
|
||||
right_mel = mel_low_freq + (bin + 2.0) * mel_freq_delta
|
||||
if vtln_warp_factor != 1.0:
|
||||
left_mel = vtln_warp_mel_freq(
|
||||
vtln_low, vtln_high, low_freq, high_freq, vtln_warp_factor, left_mel
|
||||
)
|
||||
center_mel = vtln_warp_mel_freq(
|
||||
vtln_low, vtln_high, low_freq, high_freq, vtln_warp_factor, center_mel
|
||||
)
|
||||
right_mel = vtln_warp_mel_freq(
|
||||
vtln_low, vtln_high, low_freq, high_freq, vtln_warp_factor, right_mel
|
||||
)
|
||||
center_freqs = inverse_mel_scale(center_mel)
|
||||
mel = mel_scale(fft_bin_width * paddle.arange(num_fft_bins)).unsqueeze(0)
|
||||
up_slope = (mel - left_mel) / (center_mel - left_mel)
|
||||
down_slope = (right_mel - mel) / (right_mel - center_mel)
|
||||
if vtln_warp_factor == 1.0:
|
||||
bins = paddle.maximum(
|
||||
paddle.zeros(1), paddle.minimum(up_slope, down_slope)
|
||||
)
|
||||
else:
|
||||
bins = paddle.zeros_like(up_slope)
|
||||
up_idx = paddle.gt(mel, left_mel) & paddle.le(mel, center_mel)
|
||||
down_idx = paddle.gt(mel, center_mel) & paddle.lt(mel, right_mel)
|
||||
bins[up_idx] = up_slope[up_idx]
|
||||
bins[down_idx] = down_slope[down_idx]
|
||||
return bins, center_freqs
|
||||
|
||||
|
||||
def fbank(
|
||||
waveform: paddle.Tensor,
|
||||
blackman_coeff: float = 0.42,
|
||||
channel: int = -1,
|
||||
dither: float = 0.0,
|
||||
energy_floor: float = 1.0,
|
||||
frame_length: float = 25.0,
|
||||
frame_shift: float = 10.0,
|
||||
high_freq: float = 0.0,
|
||||
htk_compat: bool = False,
|
||||
low_freq: float = 20.0,
|
||||
min_duration: float = 0.0,
|
||||
num_mel_bins: int = 23,
|
||||
preemphasis_coefficient: float = 0.97,
|
||||
raw_energy: bool = True,
|
||||
remove_dc_offset: bool = True,
|
||||
round_to_power_of_two: bool = True,
|
||||
sample_frequency: float = 16000.0,
|
||||
snip_edges: bool = True,
|
||||
subtract_mean: bool = False,
|
||||
use_energy: bool = False,
|
||||
use_log_fbank: bool = True,
|
||||
use_power: bool = True,
|
||||
vtln_high: float = -500.0,
|
||||
vtln_low: float = 100.0,
|
||||
vtln_warp: float = 1.0,
|
||||
window_type: str = POVEY,
|
||||
) -> paddle.Tensor:
|
||||
device, dtype = waveform.place, waveform.dtype
|
||||
(
|
||||
waveform,
|
||||
window_shift,
|
||||
window_size,
|
||||
padded_window_size,
|
||||
) = _get_waveform_and_window_properties(
|
||||
waveform,
|
||||
channel,
|
||||
sample_frequency,
|
||||
frame_shift,
|
||||
frame_length,
|
||||
round_to_power_of_two,
|
||||
preemphasis_coefficient,
|
||||
)
|
||||
if len(waveform) < min_duration * sample_frequency:
|
||||
return paddle.empty(0, place=place, dtype=dtype)
|
||||
strided_input, signal_log_energy = _get_window(
|
||||
waveform,
|
||||
padded_window_size,
|
||||
window_size,
|
||||
window_shift,
|
||||
window_type,
|
||||
blackman_coeff,
|
||||
snip_edges,
|
||||
raw_energy,
|
||||
energy_floor,
|
||||
dither,
|
||||
remove_dc_offset,
|
||||
preemphasis_coefficient,
|
||||
)
|
||||
spectrum = paddle.fft.rfft(strided_input).abs()
|
||||
if use_power:
|
||||
spectrum = spectrum.pow(2.0)
|
||||
mel_energies, _ = get_mel_banks(
|
||||
num_mel_bins,
|
||||
padded_window_size,
|
||||
sample_frequency,
|
||||
low_freq,
|
||||
high_freq,
|
||||
vtln_low,
|
||||
vtln_high,
|
||||
vtln_warp,
|
||||
)
|
||||
mel_energies = mel_energies.to(device=device, dtype=dtype)
|
||||
mel_energies = paddle.nn.functional.pad(
|
||||
mel_energies, (0, 1), mode="constant", value=0
|
||||
)
|
||||
mel_energies = paddle.mm(input=spectrum, mat2=mel_energies.T)
|
||||
if use_log_fbank:
|
||||
mel_energies = paddle.maximum(
|
||||
mel_energies, _get_epsilon(device, dtype)
|
||||
).log()
|
||||
if use_energy:
|
||||
signal_log_energy = signal_log_energy.unsqueeze(1)
|
||||
if htk_compat:
|
||||
mel_energies = paddle.cat((mel_energies, signal_log_energy), axis=1)
|
||||
else:
|
||||
mel_energies = paddle.cat((signal_log_energy, mel_energies), axis=1)
|
||||
mel_energies = _subtract_column_mean(mel_energies, subtract_mean)
|
||||
return mel_energies
|
||||
|
||||
def _get_dct_matrix(num_ceps: int, num_mel_bins: int) -> paddle.Tensor:
|
||||
dct_matrix = paddle.audio.functional.create_dct(
|
||||
n_mfcc=num_mel_bins,
|
||||
n_mels=num_mel_bins,
|
||||
norm='ortho'
|
||||
)
|
||||
first_col = paddle.full(
|
||||
shape=[num_mel_bins, 1],
|
||||
fill_value=math.sqrt(1 / float(num_mel_bins))
|
||||
)
|
||||
if num_ceps > 1:
|
||||
other_cols = dct_matrix[:, 1:num_ceps]
|
||||
dct_matrix = paddle.concat([first_col, other_cols], axis=1)
|
||||
else:
|
||||
dct_matrix = first_col
|
||||
return dct_matrix
|
||||
|
||||
|
||||
def _get_lifter_coeffs(num_ceps: int, cepstral_lifter: float) -> paddle.Tensor:
|
||||
i = paddle.arange(num_ceps)
|
||||
return 1.0 + 0.5 * cepstral_lifter * paddle.sin(math.pi * i / cepstral_lifter)
|
||||
|
||||
|
||||
def mfcc(
|
||||
waveform: paddle.Tensor,
|
||||
blackman_coeff: float = 0.42,
|
||||
cepstral_lifter: float = 22.0,
|
||||
channel: int = -1,
|
||||
dither: float = 0.0,
|
||||
energy_floor: float = 1.0,
|
||||
frame_length: float = 25.0,
|
||||
frame_shift: float = 10.0,
|
||||
high_freq: float = 0.0,
|
||||
htk_compat: bool = False,
|
||||
low_freq: float = 20.0,
|
||||
num_ceps: int = 13,
|
||||
min_duration: float = 0.0,
|
||||
num_mel_bins: int = 23,
|
||||
preemphasis_coefficient: float = 0.97,
|
||||
raw_energy: bool = True,
|
||||
remove_dc_offset: bool = True,
|
||||
round_to_power_of_two: bool = True,
|
||||
sample_frequency: float = 16000.0,
|
||||
snip_edges: bool = True,
|
||||
subtract_mean: bool = False,
|
||||
use_energy: bool = False,
|
||||
vtln_high: float = -500.0,
|
||||
vtln_low: float = 100.0,
|
||||
vtln_warp: float = 1.0,
|
||||
window_type: str = POVEY,
|
||||
) -> paddle.Tensor:
|
||||
"""Create a mfcc from a raw audio signal. This matches the input/output of Kaldi's
|
||||
compute-mfcc-feats.
|
||||
|
||||
Args:
|
||||
waveform (Tensor): Tensor of audio of size (c, n) where c is in the range [0,2)
|
||||
blackman_coeff (float, optional): Constant coefficient for generalized Blackman window. (Default: ``0.42``)
|
||||
cepstral_lifter (float, optional): Constant that controls scaling of MFCCs (Default: ``22.0``)
|
||||
channel (int, optional): Channel to extract (-1 -> expect mono, 0 -> left, 1 -> right) (Default: ``-1``)
|
||||
dither (float, optional): Dithering constant (0.0 means no dither). If you turn this off, you should set
|
||||
the energy_floor option, e.g. to 1.0 or 0.1 (Default: ``0.0``)
|
||||
energy_floor (float, optional): Floor on energy (absolute, not relative) in Spectrogram computation. Caution:
|
||||
this floor is applied to the zeroth component, representing the total signal energy. The floor on the
|
||||
individual spectrogram elements is fixed at std::numeric_limits<float>::epsilon(). (Default: ``1.0``)
|
||||
frame_length (float, optional): Frame length in milliseconds (Default: ``25.0``)
|
||||
frame_shift (float, optional): Frame shift in milliseconds (Default: ``10.0``)
|
||||
high_freq (float, optional): High cutoff frequency for mel bins (if <= 0, offset from Nyquist)
|
||||
(Default: ``0.0``)
|
||||
htk_compat (bool, optional): If true, put energy last. Warning: not sufficient to get HTK compatible
|
||||
features (need to change other parameters). (Default: ``False``)
|
||||
low_freq (float, optional): Low cutoff frequency for mel bins (Default: ``20.0``)
|
||||
num_ceps (int, optional): Number of cepstra in MFCC computation (including C0) (Default: ``13``)
|
||||
min_duration (float, optional): Minimum duration of segments to process (in seconds). (Default: ``0.0``)
|
||||
num_mel_bins (int, optional): Number of triangular mel-frequency bins (Default: ``23``)
|
||||
preemphasis_coefficient (float, optional): Coefficient for use in signal preemphasis (Default: ``0.97``)
|
||||
raw_energy (bool, optional): If True, compute energy before preemphasis and windowing (Default: ``True``)
|
||||
remove_dc_offset (bool, optional): Subtract mean from waveform on each frame (Default: ``True``)
|
||||
round_to_power_of_two (bool, optional): If True, round window size to power of two by zero-padding input
|
||||
to FFT. (Default: ``True``)
|
||||
sample_frequency (float, optional): Waveform data sample frequency (must match the waveform file, if
|
||||
specified there) (Default: ``16000.0``)
|
||||
snip_edges (bool, optional): If True, end effects will be handled by outputting only frames that completely fit
|
||||
in the file, and the number of frames depends on the frame_length. If False, the number of frames
|
||||
depends only on the frame_shift, and we reflect the data at the ends. (Default: ``True``)
|
||||
subtract_mean (bool, optional): Subtract mean of each feature file [CMS]; not recommended to do
|
||||
it this way. (Default: ``False``)
|
||||
use_energy (bool, optional): Add an extra dimension with energy to the FBANK output. (Default: ``False``)
|
||||
vtln_high (float, optional): High inflection point in piecewise linear VTLN warping function (if
|
||||
negative, offset from high-mel-freq (Default: ``-500.0``)
|
||||
vtln_low (float, optional): Low inflection point in piecewise linear VTLN warping function (Default: ``100.0``)
|
||||
vtln_warp (float, optional): Vtln warp factor (only applicable if vtln_map not specified) (Default: ``1.0``)
|
||||
window_type (str, optional): Type of window ('hamming'|'hanning'|'povey'|'rectangular'|'blackman')
|
||||
(Default: ``"povey"``)
|
||||
|
||||
Returns:
|
||||
Tensor: A mfcc identical to what Kaldi would output. The shape is (m, ``num_ceps``)
|
||||
where m is calculated in _get_strided
|
||||
"""
|
||||
assert (
|
||||
num_ceps <= num_mel_bins
|
||||
), "num_ceps cannot be larger than num_mel_bins: %d vs %d" % (
|
||||
num_ceps,
|
||||
num_mel_bins,
|
||||
)
|
||||
device, dtype = waveform.device, waveform.dtype
|
||||
feature = fbank(
|
||||
waveform=waveform,
|
||||
blackman_coeff=blackman_coeff,
|
||||
channel=channel,
|
||||
dither=dither,
|
||||
energy_floor=energy_floor,
|
||||
frame_length=frame_length,
|
||||
frame_shift=frame_shift,
|
||||
high_freq=high_freq,
|
||||
htk_compat=htk_compat,
|
||||
low_freq=low_freq,
|
||||
min_duration=min_duration,
|
||||
num_mel_bins=num_mel_bins,
|
||||
preemphasis_coefficient=preemphasis_coefficient,
|
||||
raw_energy=raw_energy,
|
||||
remove_dc_offset=remove_dc_offset,
|
||||
round_to_power_of_two=round_to_power_of_two,
|
||||
sample_frequency=sample_frequency,
|
||||
snip_edges=snip_edges,
|
||||
subtract_mean=False,
|
||||
use_energy=use_energy,
|
||||
use_log_fbank=True,
|
||||
use_power=True,
|
||||
vtln_high=vtln_high,
|
||||
vtln_low=vtln_low,
|
||||
vtln_warp=vtln_warp,
|
||||
window_type=window_type,
|
||||
)
|
||||
if use_energy:
|
||||
signal_log_energy = feature[:, num_mel_bins if htk_compat else 0]
|
||||
mel_offset = int(not htk_compat)
|
||||
feature = feature[:, mel_offset : num_mel_bins + mel_offset]
|
||||
dct_matrix = _get_dct_matrix(num_ceps, num_mel_bins).to(dtype=dtype, device=device)
|
||||
feature = feature.matmul(dct_matrix)
|
||||
if cepstral_lifter != 0.0:
|
||||
lifter_coeffs = _get_lifter_coeffs(num_ceps, cepstral_lifter).unsqueeze(0)
|
||||
feature *= lifter_coeffs.to(device=device, dtype=dtype)
|
||||
if use_energy:
|
||||
feature[:, 0] = signal_log_energy
|
||||
if htk_compat:
|
||||
energy = feature[:, 0].unsqueeze(1)
|
||||
feature = feature[:, 1:]
|
||||
if not use_energy:
|
||||
energy *= math.sqrt(2)
|
||||
feature = paddle.cat((feature, energy), axis=1)
|
||||
feature = _subtract_column_mean(feature, subtract_mean)
|
||||
return feature
|
||||
@ -0,0 +1,578 @@
|
||||
import base64
|
||||
import os
|
||||
from functools import lru_cache
|
||||
from typing import Optional
|
||||
|
||||
import paddle
|
||||
|
||||
import tiktoken
|
||||
from whisper.tokenizer import Tokenizer
|
||||
from paddlenlp.transformers import AutoTokenizer
|
||||
LANGUAGES = {
|
||||
"en": "english",
|
||||
"zh": "chinese",
|
||||
"de": "german",
|
||||
"es": "spanish",
|
||||
"ru": "russian",
|
||||
"ko": "korean",
|
||||
"fr": "french",
|
||||
"ja": "japanese",
|
||||
"pt": "portuguese",
|
||||
"tr": "turkish",
|
||||
"pl": "polish",
|
||||
"ca": "catalan",
|
||||
"nl": "dutch",
|
||||
"ar": "arabic",
|
||||
"sv": "swedish",
|
||||
"it": "italian",
|
||||
"id": "indonesian",
|
||||
"hi": "hindi",
|
||||
"fi": "finnish",
|
||||
"vi": "vietnamese",
|
||||
"he": "hebrew",
|
||||
"uk": "ukrainian",
|
||||
"el": "greek",
|
||||
"ms": "malay",
|
||||
"cs": "czech",
|
||||
"ro": "romanian",
|
||||
"da": "danish",
|
||||
"hu": "hungarian",
|
||||
"ta": "tamil",
|
||||
"no": "norwegian",
|
||||
"th": "thai",
|
||||
"ur": "urdu",
|
||||
"hr": "croatian",
|
||||
"bg": "bulgarian",
|
||||
"lt": "lithuanian",
|
||||
"la": "latin",
|
||||
"mi": "maori",
|
||||
"ml": "malayalam",
|
||||
"cy": "welsh",
|
||||
"sk": "slovak",
|
||||
"te": "telugu",
|
||||
"fa": "persian",
|
||||
"lv": "latvian",
|
||||
"bn": "bengali",
|
||||
"sr": "serbian",
|
||||
"az": "azerbaijani",
|
||||
"sl": "slovenian",
|
||||
"kn": "kannada",
|
||||
"et": "estonian",
|
||||
"mk": "macedonian",
|
||||
"br": "breton",
|
||||
"eu": "basque",
|
||||
"is": "icelandic",
|
||||
"hy": "armenian",
|
||||
"ne": "nepali",
|
||||
"mn": "mongolian",
|
||||
"bs": "bosnian",
|
||||
"kk": "kazakh",
|
||||
"sq": "albanian",
|
||||
"sw": "swahili",
|
||||
"gl": "galician",
|
||||
"mr": "marathi",
|
||||
"pa": "punjabi",
|
||||
"si": "sinhala",
|
||||
"km": "khmer",
|
||||
"sn": "shona",
|
||||
"yo": "yoruba",
|
||||
"so": "somali",
|
||||
"af": "afrikaans",
|
||||
"oc": "occitan",
|
||||
"ka": "georgian",
|
||||
"be": "belarusian",
|
||||
"tg": "tajik",
|
||||
"sd": "sindhi",
|
||||
"gu": "gujarati",
|
||||
"am": "amharic",
|
||||
"yi": "yiddish",
|
||||
"lo": "lao",
|
||||
"uz": "uzbek",
|
||||
"fo": "faroese",
|
||||
"ht": "haitian creole",
|
||||
"ps": "pashto",
|
||||
"tk": "turkmen",
|
||||
"nn": "nynorsk",
|
||||
"mt": "maltese",
|
||||
"sa": "sanskrit",
|
||||
"lb": "luxembourgish",
|
||||
"my": "myanmar",
|
||||
"bo": "tibetan",
|
||||
"tl": "tagalog",
|
||||
"mg": "malagasy",
|
||||
"as": "assamese",
|
||||
"tt": "tatar",
|
||||
"haw": "hawaiian",
|
||||
"ln": "lingala",
|
||||
"ha": "hausa",
|
||||
"ba": "bashkir",
|
||||
"jw": "javanese",
|
||||
"su": "sundanese",
|
||||
"yue": "cantonese",
|
||||
"minnan": "minnan",
|
||||
"wuyu": "wuyu",
|
||||
"dialect": "dialect",
|
||||
"zh/en": "zh/en",
|
||||
"en/zh": "en/zh",
|
||||
}
|
||||
TO_LANGUAGE_CODE = {
|
||||
**{language: code for code, language in LANGUAGES.items()},
|
||||
"burmese": "my",
|
||||
"valencian": "ca",
|
||||
"flemish": "nl",
|
||||
"haitian": "ht",
|
||||
"letzeburgesch": "lb",
|
||||
"pushto": "ps",
|
||||
"panjabi": "pa",
|
||||
"moldavian": "ro",
|
||||
"moldovan": "ro",
|
||||
"sinhalese": "si",
|
||||
"castilian": "es",
|
||||
"mandarin": "zh",
|
||||
}
|
||||
AUDIO_EVENT = {
|
||||
"ASR": "ASR",
|
||||
"AED": "AED",
|
||||
"SER": "SER",
|
||||
"Speech": "Speech",
|
||||
"/Speech": "/Speech",
|
||||
"BGM": "BGM",
|
||||
"/BGM": "/BGM",
|
||||
"Laughter": "Laughter",
|
||||
"/Laughter": "/Laughter",
|
||||
"Applause": "Applause",
|
||||
"/Applause": "/Applause",
|
||||
}
|
||||
EMOTION = {"HAPPY": "HAPPY", "SAD": "SAD", "ANGRY": "ANGRY", "NEUTRAL": "NEUTRAL"}
|
||||
TTS_Vocal_Token = {
|
||||
"TTS/B": "TTS/B",
|
||||
"TTS/O": "TTS/O",
|
||||
"TTS/Q": "TTS/Q",
|
||||
"TTS/A": "TTS/A",
|
||||
"TTS/CO": "TTS/CO",
|
||||
"TTS/CL": "TTS/CL",
|
||||
"TTS/H": "TTS/H",
|
||||
**{f"TTS/SP{i:02d}": f"TTS/SP{i:02d}" for i in range(1, 14)},
|
||||
}
|
||||
|
||||
|
||||
@lru_cache(maxsize=None)
|
||||
def get_encoding(name: str = "gpt2", num_languages: int = 99):
|
||||
vocab_path = os.path.join(os.path.dirname(__file__), "assets", f"{name}.tiktoken")
|
||||
ranks = {
|
||||
base64.b64decode(token): int(rank)
|
||||
for token, rank in (line.split() for line in open(vocab_path) if line)
|
||||
}
|
||||
n_vocab = len(ranks)
|
||||
special_tokens = {}
|
||||
specials = [
|
||||
"<|endoftext|>",
|
||||
"<|startoftranscript|>",
|
||||
*[f"<|{lang}|>" for lang in list(LANGUAGES.keys())[:num_languages]],
|
||||
*[f"<|{audio_event}|>" for audio_event in list(AUDIO_EVENT.keys())],
|
||||
*[f"<|{emotion}|>" for emotion in list(EMOTION.keys())],
|
||||
"<|translate|>",
|
||||
"<|transcribe|>",
|
||||
"<|startoflm|>",
|
||||
"<|startofprev|>",
|
||||
"<|nospeech|>",
|
||||
"<|notimestamps|>",
|
||||
*[f"<|SPECIAL_TOKEN_{i}|>" for i in range(1, 31)],
|
||||
*[f"<|{tts}|>" for tts in list(TTS_Vocal_Token.keys())],
|
||||
*[f"<|{i * 0.02:.2f}|>" for i in range(1501)],
|
||||
]
|
||||
for token in specials:
|
||||
special_tokens[token] = n_vocab
|
||||
n_vocab += 1
|
||||
return tiktoken.Encoding(
|
||||
name=os.path.basename(vocab_path),
|
||||
explicit_n_vocab=n_vocab,
|
||||
pat_str="'s|'t|'re|'ve|'m|'ll|'d| ?\\p{L}+| ?\\p{N}+| ?[^\\s\\p{L}\\p{N}]+|\\s+(?!\\S)|\\s+",
|
||||
mergeable_ranks=ranks,
|
||||
special_tokens=special_tokens,
|
||||
)
|
||||
|
||||
|
||||
@lru_cache(maxsize=None)
|
||||
def get_tokenizer(
|
||||
multilingual: bool,
|
||||
*,
|
||||
num_languages: int = 99,
|
||||
language: Optional[str] = None,
|
||||
task: Optional[str] = None,
|
||||
) -> Tokenizer:
|
||||
if language is not None:
|
||||
language = language.lower()
|
||||
if language not in LANGUAGES:
|
||||
if language in TO_LANGUAGE_CODE:
|
||||
language = TO_LANGUAGE_CODE[language]
|
||||
else:
|
||||
raise ValueError(f"Unsupported language: {language}")
|
||||
if multilingual:
|
||||
encoding_name = "multilingual_zh_ja_yue_char_del"
|
||||
language = language or "en"
|
||||
task = task or "transcribe"
|
||||
else:
|
||||
encoding_name = "gpt2"
|
||||
language = None
|
||||
task = None
|
||||
encoding = get_encoding(name=encoding_name, num_languages=num_languages)
|
||||
return Tokenizer(
|
||||
encoding=encoding, num_languages=num_languages, language=language, task=task
|
||||
)
|
||||
|
||||
|
||||
class CosyVoice2Tokenizer:
|
||||
def __init__(self, token_path, skip_special_tokens=True):
|
||||
super().__init__()
|
||||
special_tokens = {
|
||||
"eos_token": "<|endoftext|>",
|
||||
"pad_token": "<|endoftext|>",
|
||||
"additional_special_tokens": [
|
||||
"<|im_start|>",
|
||||
"<|im_end|>",
|
||||
"<|endofprompt|>",
|
||||
"[breath]",
|
||||
"<strong>",
|
||||
"</strong>",
|
||||
"[noise]",
|
||||
"[laughter]",
|
||||
"[cough]",
|
||||
"[clucking]",
|
||||
"[accent]",
|
||||
"[quick_breath]",
|
||||
"<laughter>",
|
||||
"</laughter>",
|
||||
"[hissing]",
|
||||
"[sigh]",
|
||||
"[vocalized-noise]",
|
||||
"[lipsmack]",
|
||||
"[mn]",
|
||||
],
|
||||
}
|
||||
self.special_tokens = special_tokens
|
||||
self.tokenizer = AutoTokenizer.from_pretrained(token_path)
|
||||
self.tokenizer.add_special_tokens(special_tokens)
|
||||
self.skip_special_tokens = skip_special_tokens
|
||||
|
||||
def encode(self, text, **kwargs):
|
||||
tokens = self.tokenizer(text)
|
||||
tokens = tokens["input_ids"][0]
|
||||
return tokens
|
||||
|
||||
def decode(self, tokens):
|
||||
tokens = paddle.tensor(tokens, dtype=paddle.int64)
|
||||
text = self.tokenizer.batch_decode(
|
||||
[tokens], skip_special_tokens=self.skip_special_tokens
|
||||
)[0]
|
||||
return text
|
||||
|
||||
|
||||
class CosyVoice3Tokenizer(CosyVoice2Tokenizer):
|
||||
def __init__(self, token_path, skip_special_tokens=True):
|
||||
special_tokens = {
|
||||
"eos_token": "<|endoftext|>",
|
||||
"pad_token": "<|endoftext|>",
|
||||
"additional_special_tokens": [
|
||||
"<|im_start|>",
|
||||
"<|im_end|>",
|
||||
"<|endofprompt|>",
|
||||
"[breath]",
|
||||
"<strong>",
|
||||
"</strong>",
|
||||
"[noise]",
|
||||
"[laughter]",
|
||||
"[cough]",
|
||||
"[clucking]",
|
||||
"[accent]",
|
||||
"[quick_breath]",
|
||||
"<laughter>",
|
||||
"</laughter>",
|
||||
"[hissing]",
|
||||
"[sigh]",
|
||||
"[vocalized-noise]",
|
||||
"[lipsmack]",
|
||||
"[mn]",
|
||||
"<|endofsystem|>",
|
||||
"[AA]",
|
||||
"[AA0]",
|
||||
"[AA1]",
|
||||
"[AA2]",
|
||||
"[AE]",
|
||||
"[AE0]",
|
||||
"[AE1]",
|
||||
"[AE2]",
|
||||
"[AH]",
|
||||
"[AH0]",
|
||||
"[AH1]",
|
||||
"[AH2]",
|
||||
"[AO]",
|
||||
"[AO0]",
|
||||
"[AO1]",
|
||||
"[AO2]",
|
||||
"[AW]",
|
||||
"[AW0]",
|
||||
"[AW1]",
|
||||
"[AW2]",
|
||||
"[AY]",
|
||||
"[AY0]",
|
||||
"[AY1]",
|
||||
"[AY2]",
|
||||
"[B]",
|
||||
"[CH]",
|
||||
"[D]",
|
||||
"[DH]",
|
||||
"[EH]",
|
||||
"[EH0]",
|
||||
"[EH1]",
|
||||
"[EH2]",
|
||||
"[ER]",
|
||||
"[ER0]",
|
||||
"[ER1]",
|
||||
"[ER2]",
|
||||
"[EY]",
|
||||
"[EY0]",
|
||||
"[EY1]",
|
||||
"[EY2]",
|
||||
"[F]",
|
||||
"[G]",
|
||||
"[HH]",
|
||||
"[IH]",
|
||||
"[IH0]",
|
||||
"[IH1]",
|
||||
"[IH2]",
|
||||
"[IY]",
|
||||
"[IY0]",
|
||||
"[IY1]",
|
||||
"[IY2]",
|
||||
"[JH]",
|
||||
"[K]",
|
||||
"[L]",
|
||||
"[M]",
|
||||
"[N]",
|
||||
"[NG]",
|
||||
"[OW]",
|
||||
"[OW0]",
|
||||
"[OW1]",
|
||||
"[OW2]",
|
||||
"[OY]",
|
||||
"[OY0]",
|
||||
"[OY1]",
|
||||
"[OY2]",
|
||||
"[P]",
|
||||
"[R]",
|
||||
"[S]",
|
||||
"[SH]",
|
||||
"[T]",
|
||||
"[TH]",
|
||||
"[UH]",
|
||||
"[UH0]",
|
||||
"[UH1]",
|
||||
"[UH2]",
|
||||
"[UW]",
|
||||
"[UW0]",
|
||||
"[UW1]",
|
||||
"[UW2]",
|
||||
"[V]",
|
||||
"[W]",
|
||||
"[Y]",
|
||||
"[Z]",
|
||||
"[ZH]",
|
||||
"[a]",
|
||||
"[ai]",
|
||||
"[an]",
|
||||
"[ang]",
|
||||
"[ao]",
|
||||
"[b]",
|
||||
"[c]",
|
||||
"[ch]",
|
||||
"[d]",
|
||||
"[e]",
|
||||
"[ei]",
|
||||
"[en]",
|
||||
"[eng]",
|
||||
"[f]",
|
||||
"[g]",
|
||||
"[h]",
|
||||
"[i]",
|
||||
"[ian]",
|
||||
"[in]",
|
||||
"[ing]",
|
||||
"[iu]",
|
||||
"[ià]",
|
||||
"[iàn]",
|
||||
"[iàng]",
|
||||
"[iào]",
|
||||
"[iá]",
|
||||
"[ián]",
|
||||
"[iáng]",
|
||||
"[iáo]",
|
||||
"[iè]",
|
||||
"[ié]",
|
||||
"[iòng]",
|
||||
"[ióng]",
|
||||
"[iù]",
|
||||
"[iú]",
|
||||
"[iā]",
|
||||
"[iān]",
|
||||
"[iāng]",
|
||||
"[iāo]",
|
||||
"[iē]",
|
||||
"[iě]",
|
||||
"[iōng]",
|
||||
"[iū]",
|
||||
"[iǎ]",
|
||||
"[iǎn]",
|
||||
"[iǎng]",
|
||||
"[iǎo]",
|
||||
"[iǒng]",
|
||||
"[iǔ]",
|
||||
"[j]",
|
||||
"[k]",
|
||||
"[l]",
|
||||
"[m]",
|
||||
"[n]",
|
||||
"[o]",
|
||||
"[ong]",
|
||||
"[ou]",
|
||||
"[p]",
|
||||
"[q]",
|
||||
"[r]",
|
||||
"[s]",
|
||||
"[sh]",
|
||||
"[t]",
|
||||
"[u]",
|
||||
"[uang]",
|
||||
"[ue]",
|
||||
"[un]",
|
||||
"[uo]",
|
||||
"[uà]",
|
||||
"[uài]",
|
||||
"[uàn]",
|
||||
"[uàng]",
|
||||
"[uá]",
|
||||
"[uái]",
|
||||
"[uán]",
|
||||
"[uáng]",
|
||||
"[uè]",
|
||||
"[ué]",
|
||||
"[uì]",
|
||||
"[uí]",
|
||||
"[uò]",
|
||||
"[uó]",
|
||||
"[uā]",
|
||||
"[uāi]",
|
||||
"[uān]",
|
||||
"[uāng]",
|
||||
"[uē]",
|
||||
"[uě]",
|
||||
"[uī]",
|
||||
"[uō]",
|
||||
"[uǎ]",
|
||||
"[uǎi]",
|
||||
"[uǎn]",
|
||||
"[uǎng]",
|
||||
"[uǐ]",
|
||||
"[uǒ]",
|
||||
"[vè]",
|
||||
"[w]",
|
||||
"[x]",
|
||||
"[y]",
|
||||
"[z]",
|
||||
"[zh]",
|
||||
"[à]",
|
||||
"[ài]",
|
||||
"[àn]",
|
||||
"[àng]",
|
||||
"[ào]",
|
||||
"[á]",
|
||||
"[ái]",
|
||||
"[án]",
|
||||
"[áng]",
|
||||
"[áo]",
|
||||
"[è]",
|
||||
"[èi]",
|
||||
"[èn]",
|
||||
"[èng]",
|
||||
"[èr]",
|
||||
"[é]",
|
||||
"[éi]",
|
||||
"[én]",
|
||||
"[éng]",
|
||||
"[ér]",
|
||||
"[ì]",
|
||||
"[ìn]",
|
||||
"[ìng]",
|
||||
"[í]",
|
||||
"[ín]",
|
||||
"[íng]",
|
||||
"[ò]",
|
||||
"[òng]",
|
||||
"[òu]",
|
||||
"[ó]",
|
||||
"[óng]",
|
||||
"[óu]",
|
||||
"[ù]",
|
||||
"[ùn]",
|
||||
"[ú]",
|
||||
"[ún]",
|
||||
"[ā]",
|
||||
"[āi]",
|
||||
"[ān]",
|
||||
"[āng]",
|
||||
"[āo]",
|
||||
"[ē]",
|
||||
"[ēi]",
|
||||
"[ēn]",
|
||||
"[ēng]",
|
||||
"[ě]",
|
||||
"[ěi]",
|
||||
"[ěn]",
|
||||
"[ěng]",
|
||||
"[ěr]",
|
||||
"[ī]",
|
||||
"[īn]",
|
||||
"[īng]",
|
||||
"[ō]",
|
||||
"[ōng]",
|
||||
"[ōu]",
|
||||
"[ū]",
|
||||
"[ūn]",
|
||||
"[ǎ]",
|
||||
"[ǎi]",
|
||||
"[ǎn]",
|
||||
"[ǎng]",
|
||||
"[ǎo]",
|
||||
"[ǐ]",
|
||||
"[ǐn]",
|
||||
"[ǐng]",
|
||||
"[ǒ]",
|
||||
"[ǒng]",
|
||||
"[ǒu]",
|
||||
"[ǔ]",
|
||||
"[ǔn]",
|
||||
"[ǘ]",
|
||||
"[ǚ]",
|
||||
"[ǜ]",
|
||||
],
|
||||
}
|
||||
self.special_tokens = special_tokens
|
||||
self.tokenizer = transformers.AutoTokenizer.from_pretrained(token_path)
|
||||
self.tokenizer.add_special_tokens(special_tokens)
|
||||
self.skip_special_tokens = skip_special_tokens
|
||||
|
||||
|
||||
@lru_cache(maxsize=None)
|
||||
def get_qwen_tokenizer(
|
||||
token_path: str, skip_special_tokens: bool, version: str = "cosyvoice2"
|
||||
):
|
||||
if version == "cosyvoice2":
|
||||
return CosyVoice2Tokenizer(
|
||||
token_path=token_path, skip_special_tokens=skip_special_tokens
|
||||
)
|
||||
elif version == "cosyvoice3":
|
||||
return CosyVoice3Tokenizer(
|
||||
token_path=token_path, skip_special_tokens=skip_special_tokens
|
||||
)
|
||||
else:
|
||||
raise ValueError
|
||||
Loading…
Reference in new issue