|
import json |
|
import os |
|
import time |
|
from io import BytesIO |
|
from pathlib import Path |
|
|
|
import librosa |
|
import numpy as np |
|
import soundfile |
|
import torch |
|
|
|
import utils |
|
from infer_tools.f0_static import compare_pitch, static_f0_time |
|
from modules.diff.diffusion import GaussianDiffusion |
|
from modules.diff.net import DiffNet |
|
from modules.vocoders.nsf_hifigan import NsfHifiGAN |
|
from preprocessing.hubertinfer import HubertEncoder |
|
from preprocessing.process_pipeline import File2Batch, get_pitch_parselmouth |
|
from utils.hparams import hparams, set_hparams |
|
from utils.pitch_utils import denorm_f0, norm_interp_f0 |
|
|
|
|
|
def timeit(func): |
|
def run(*args, **kwargs): |
|
t = time.time() |
|
res = func(*args, **kwargs) |
|
print('executing \'%s\' costed %.3fs' % (func.__name__, time.time() - t)) |
|
return res |
|
|
|
return run |
|
|
|
|
|
def format_wav(audio_path): |
|
if Path(audio_path).suffix == '.wav': |
|
return |
|
raw_audio, raw_sample_rate = librosa.load(audio_path, mono=True, sr=None) |
|
soundfile.write(Path(audio_path).with_suffix(".wav"), raw_audio, raw_sample_rate) |
|
|
|
|
|
def fill_a_to_b(a, b): |
|
if len(a) < len(b): |
|
for _ in range(0, len(b) - len(a)): |
|
a.append(a[0]) |
|
|
|
|
|
def get_end_file(dir_path, end): |
|
file_lists = [] |
|
for root, dirs, files in os.walk(dir_path): |
|
files = [f for f in files if f[0] != '.'] |
|
dirs[:] = [d for d in dirs if d[0] != '.'] |
|
for f_file in files: |
|
if f_file.endswith(end): |
|
file_lists.append(os.path.join(root, f_file).replace("\\", "/")) |
|
return file_lists |
|
|
|
|
|
def mkdir(paths: list): |
|
for path in paths: |
|
if not os.path.exists(path): |
|
os.mkdir(path) |
|
|
|
|
|
class Svc: |
|
def __init__(self, project_name, config_name, hubert_gpu, model_path, onnx=False): |
|
self.project_name = project_name |
|
self.DIFF_DECODERS = { |
|
'wavenet': lambda hp: DiffNet(hp['audio_num_mel_bins']), |
|
} |
|
|
|
self.model_path = model_path |
|
self.dev = torch.device("cpu") |
|
|
|
self._ = set_hparams(config=config_name, exp_name=self.project_name, infer=True, |
|
reset=True, hparams_str='', print_hparams=False) |
|
|
|
hparams['hubert_gpu'] = hubert_gpu |
|
self.hubert = HubertEncoder(hparams['hubert_path'], onnx=onnx) |
|
self.model = GaussianDiffusion( |
|
phone_encoder=self.hubert, |
|
out_dims=hparams['audio_num_mel_bins'], |
|
denoise_fn=self.DIFF_DECODERS[hparams['diff_decoder_type']](hparams), |
|
timesteps=hparams['timesteps'], |
|
K_step=hparams['K_step'], |
|
loss_type=hparams['diff_loss_type'], |
|
spec_min=hparams['spec_min'], spec_max=hparams['spec_max'], |
|
) |
|
utils.load_ckpt(self.model, self.model_path, 'model', force=True, strict=True) |
|
self.model.to(self.dev) |
|
self.vocoder = NsfHifiGAN() |
|
|
|
def infer(self, in_path, key, acc, spk_id=0, use_crepe=True, singer=False): |
|
batch = self.pre(in_path, acc, spk_id, use_crepe) |
|
batch['f0'] = batch['f0'] + (key / 12) |
|
batch['f0'][batch['f0'] > np.log2(hparams['f0_max'])] = 0 |
|
|
|
@timeit |
|
def diff_infer(): |
|
spk_embed = batch.get('spk_embed') if not hparams['use_spk_id'] else batch.get('spk_ids') |
|
energy = batch.get('energy').cpu() if batch.get('energy') else None |
|
if spk_embed is None: |
|
spk_embed = torch.LongTensor([0]).cpu() |
|
diff_outputs = self.model( |
|
hubert=batch['hubert'].cpu(), spk_embed_id=spk_embed.cpu(), mel2ph=batch['mel2ph'].cpu(), |
|
f0=batch['f0'].cpu(), energy=energy, ref_mels=batch["mels"].cpu(), infer=True) |
|
return diff_outputs |
|
|
|
outputs = diff_infer() |
|
batch['outputs'] = outputs['mel_out'] |
|
batch['mel2ph_pred'] = outputs['mel2ph'] |
|
batch['f0_gt'] = denorm_f0(batch['f0'], batch['uv'], hparams) |
|
batch['f0_pred'] = outputs.get('f0_denorm') |
|
return self.after_infer(batch, singer, in_path) |
|
|
|
@timeit |
|
def after_infer(self, prediction, singer, in_path): |
|
for k, v in prediction.items(): |
|
if type(v) is torch.Tensor: |
|
prediction[k] = v.cpu().numpy() |
|
|
|
|
|
mel_gt = prediction["mels"] |
|
mel_gt_mask = np.abs(mel_gt).sum(-1) > 0 |
|
|
|
mel_pred = prediction["outputs"] |
|
mel_pred_mask = np.abs(mel_pred).sum(-1) > 0 |
|
mel_pred = mel_pred[mel_pred_mask] |
|
mel_pred = np.clip(mel_pred, hparams['mel_vmin'], hparams['mel_vmax']) |
|
|
|
f0_gt = prediction.get("f0_gt") |
|
f0_pred = prediction.get("f0_pred") |
|
if f0_pred is not None: |
|
f0_gt = f0_gt[mel_gt_mask] |
|
if len(f0_pred) > len(mel_pred_mask): |
|
f0_pred = f0_pred[:len(mel_pred_mask)] |
|
f0_pred = f0_pred[mel_pred_mask] |
|
torch.cuda.is_available() and torch.cuda.empty_cache() |
|
|
|
if singer: |
|
data_path = in_path.replace("batch", "singer_data") |
|
mel_path = data_path[:-4] + "_mel.npy" |
|
f0_path = data_path[:-4] + "_f0.npy" |
|
np.save(mel_path, mel_pred) |
|
np.save(f0_path, f0_pred) |
|
wav_pred = self.vocoder.spec2wav(mel_pred, f0=f0_pred) |
|
return f0_gt, f0_pred, wav_pred |
|
|
|
def pre(self, wav_fn, accelerate, spk_id=0, use_crepe=True): |
|
if isinstance(wav_fn, BytesIO): |
|
item_name = self.project_name |
|
else: |
|
song_info = wav_fn.split('/') |
|
item_name = song_info[-1].split('.')[-2] |
|
temp_dict = {'wav_fn': wav_fn, 'spk_id': spk_id, 'id': 0} |
|
|
|
temp_dict = File2Batch.temporary_dict2processed_input(item_name, temp_dict, self.hubert, infer=True, |
|
use_crepe=use_crepe) |
|
hparams['pndm_speedup'] = accelerate |
|
batch = File2Batch.processed_input2batch([getitem(temp_dict)]) |
|
return batch |
|
|
|
def evaluate_key(self, wav_path, key, auto_key): |
|
if "f0_static" in hparams.keys(): |
|
f0_static = json.loads(hparams['f0_static']) |
|
wav, mel = self.vocoder.wav2spec(wav_path) |
|
input_f0 = get_pitch_parselmouth(wav, mel, hparams)[0] |
|
pitch_time_temp = static_f0_time(input_f0) |
|
eval_dict = {} |
|
for trans_key in range(-12, 12): |
|
eval_dict[trans_key] = compare_pitch(f0_static, pitch_time_temp, trans_key=trans_key) |
|
sort_key = sorted(eval_dict, key=eval_dict.get, reverse=True)[:5] |
|
print(f"推荐移调:{sort_key}") |
|
if auto_key: |
|
print(f"自动变调已启用，您的输入key被{sort_key[0]}key覆盖，控制参数为auto_key") |
|
return sort_key[0] |
|
else: |
|
print("config缺少f0_staic，无法使用自动变调，可通过infer_tools/data_static添加") |
|
return key |
|
|
|
|
|
def getitem(item): |
|
max_frames = hparams['max_frames'] |
|
spec = torch.Tensor(item['mel']).cpu()[:max_frames] |
|
mel2ph = torch.LongTensor(item['mel2ph']).cpu()[:max_frames] if 'mel2ph' in item else None |
|
f0, uv = norm_interp_f0(item["f0"][:max_frames], hparams) |
|
hubert = torch.Tensor(item['hubert'][:hparams['max_input_tokens']]).cpu() |
|
pitch = torch.LongTensor(item.get("pitch")).cpu()[:max_frames] |
|
sample = { |
|
"id": item['id'], |
|
"spk_id": item['spk_id'], |
|
"item_name": item['item_name'], |
|
"hubert": hubert, |
|
"mel": spec, |
|
"pitch": pitch, |
|
"f0": f0, |
|
"uv": uv, |
|
"mel2ph": mel2ph, |
|
"mel_nonpadding": spec.abs().sum(-1) > 0, |
|
} |
|
if hparams['use_energy_embed']: |
|
sample['energy'] = item['energy'] |
|
return sample |
|
|
