219 lines
7.4 KiB
Python
219 lines
7.4 KiB
Python
import traceback
|
|
|
|
import torch
|
|
import soundfile as sf
|
|
|
|
from infer.lib.infer_pack.models import (
|
|
SynthesizerTrnMs256NSFsid,
|
|
SynthesizerTrnMs256NSFsid_nono,
|
|
SynthesizerTrnMs768NSFsid,
|
|
SynthesizerTrnMs768NSFsid_nono,
|
|
)
|
|
from infer.modules.vc.pipeline import Pipeline
|
|
from infer.modules.vc.utils import *
|
|
|
|
|
|
class VC:
|
|
def __init__(self, config):
|
|
self.n_spk = None
|
|
self.tgt_sr = None
|
|
self.net_g = None
|
|
self.pipeline = None
|
|
self.cpt = None
|
|
self.version = None
|
|
self.if_f0 = None
|
|
self.version = None
|
|
self.hubert_model = None
|
|
|
|
self.config = config
|
|
|
|
def get_vc(self, sid, to_return_protect0, to_return_protect1):
|
|
person = f'{os.getenv("weight_root")}/{sid}'
|
|
print(f'loading {person}')
|
|
|
|
self.cpt = torch.load(person, map_location="cpu")
|
|
self.tgt_sr = self.cpt["config"][-1]
|
|
self.cpt["config"][-3] = self.cpt["weight"]["emb_g.weight"].shape[0] # n_spk
|
|
self.if_f0 = self.cpt.get("f0", 1)
|
|
self.version = self.cpt.get("version", "v1")
|
|
|
|
to_return_protect0 = {
|
|
"visible": self.if_f0 != 0,
|
|
"value": to_return_protect0 if self.if_f0 != 0 else 0.5,
|
|
"__type__": "update",
|
|
}
|
|
to_return_protect1 = {
|
|
"visible": self.if_f0 != 0,
|
|
"value": to_return_protect1 if self.if_f0 != 0 else 0.33,
|
|
"__type__": "update",
|
|
}
|
|
|
|
synthesizer_class = {
|
|
("v1", 1): SynthesizerTrnMs256NSFsid,
|
|
("v1", 0): SynthesizerTrnMs256NSFsid_nono,
|
|
("v2", 1): SynthesizerTrnMs768NSFsid,
|
|
("v2", 0): SynthesizerTrnMs768NSFsid_nono
|
|
}
|
|
|
|
self.net_g = synthesizer_class.get((self.version, self.if_f0), SynthesizerTrnMs256NSFsid)(*self.cpt["config"], is_half=self.config.is_half)
|
|
|
|
del self.net_g.enc_q
|
|
|
|
self.net_g.load_state_dict(self.cpt["weight"], strict=False)
|
|
self.net_g.eval().to(self.config.device)
|
|
if self.config.is_half:
|
|
self.net_g = self.net_g.half()
|
|
else:
|
|
self.net_g = self.net_g.float()
|
|
|
|
self.pipeline = Pipeline(self.tgt_sr, self.config)
|
|
n_spk = self.cpt["config"][-3]
|
|
index = {
|
|
"value": get_index_path_from_model(sid),
|
|
"__type__": "update"
|
|
}
|
|
|
|
return (
|
|
{"visible": True, "maximum": n_spk, "__type__": "update"},
|
|
to_return_protect0,
|
|
to_return_protect1,
|
|
index,
|
|
index
|
|
)
|
|
|
|
def vc_single(self, sid, input_audio_path, f0_up_key, f0_file, f0_method, file_index, file_index2, index_rate, filter_radius, resample_sr, rms_mix_rate, protect):
|
|
if input_audio_path is None:
|
|
return "You need to upload an audio", None
|
|
f0_up_key = int(f0_up_key)
|
|
try:
|
|
audio = load_audio(input_audio_path, 16000)
|
|
audio_max = np.abs(audio).max() / 0.95
|
|
if audio_max > 1:
|
|
audio /= audio_max
|
|
times = [0, 0, 0]
|
|
|
|
if self.hubert_model is None:
|
|
self.hubert_model = load_hubert(self.config)
|
|
|
|
file_index = (
|
|
(
|
|
file_index.strip(" ")
|
|
.strip('"')
|
|
.strip("\n")
|
|
.strip('"')
|
|
.strip(" ")
|
|
.replace("trained", "added")
|
|
)
|
|
if file_index != ""
|
|
else file_index2
|
|
) # 防止小白写错，自动帮他替换掉
|
|
|
|
audio_opt = Pipeline.pipeline(
|
|
self.hubert_model,
|
|
self.net_g,
|
|
sid,
|
|
audio,
|
|
input_audio_path,
|
|
times,
|
|
f0_up_key,
|
|
f0_method,
|
|
file_index,
|
|
index_rate,
|
|
self.if_f0,
|
|
filter_radius,
|
|
self.tgt_sr,
|
|
resample_sr,
|
|
rms_mix_rate,
|
|
self.version,
|
|
protect,
|
|
f0_file,
|
|
)
|
|
if self.tgt_sr != resample_sr >= 16000:
|
|
self.tgt_sr = resample_sr
|
|
index_info = (
|
|
"Using index:%s." % file_index
|
|
if os.path.exists(file_index)
|
|
else "Index not used."
|
|
)
|
|
return f"Success.\n {index_info}\nTime:\n npy:{times[0]}s, f0:{times[1]}s, infer:{times[2]}s", (self.tgt_sr, audio_opt)
|
|
except:
|
|
info = traceback.format_exc()
|
|
print(info)
|
|
return info, (None, None)
|
|
|
|
def vc_multi(
|
|
self,
|
|
sid,
|
|
dir_path,
|
|
opt_root,
|
|
paths,
|
|
f0_up_key,
|
|
f0_method,
|
|
file_index,
|
|
file_index2,
|
|
index_rate,
|
|
filter_radius,
|
|
resample_sr,
|
|
rms_mix_rate,
|
|
protect,
|
|
format1):
|
|
try:
|
|
dir_path = (
|
|
dir_path.strip(" ").strip('"').strip("\n").strip('"').strip(" ")
|
|
) # 防止小白拷路径头尾带了空格和"和回车
|
|
opt_root = opt_root.strip(" ").strip('"').strip("\n").strip('"').strip(" ")
|
|
os.makedirs(opt_root, exist_ok=True)
|
|
try:
|
|
if dir_path != "":
|
|
paths = [os.path.join(dir_path, name) for name in os.listdir(dir_path)]
|
|
else:
|
|
paths = [path.name for path in paths]
|
|
except:
|
|
traceback.print_exc()
|
|
paths = [path.name for path in paths]
|
|
infos = []
|
|
for path in paths:
|
|
info, opt = self.vc_single(
|
|
sid,
|
|
path,
|
|
f0_up_key,
|
|
None,
|
|
f0_method,
|
|
file_index,
|
|
file_index2,
|
|
# file_big_npy,
|
|
index_rate,
|
|
filter_radius,
|
|
resample_sr,
|
|
rms_mix_rate,
|
|
protect,
|
|
)
|
|
if "Success" in info:
|
|
try:
|
|
tgt_sr, audio_opt = opt
|
|
if format1 in ["wav", "flac"]:
|
|
sf.write(
|
|
"%s/%s.%s" % (opt_root, os.path.basename(path), format1),
|
|
audio_opt,
|
|
tgt_sr,
|
|
)
|
|
else:
|
|
path = "%s/%s.wav" % (opt_root, os.path.basename(path))
|
|
sf.write(
|
|
path,
|
|
audio_opt,
|
|
tgt_sr,
|
|
)
|
|
if os.path.exists(path):
|
|
os.system(
|
|
"ffmpeg -i %s -vn %s -q:a 2 -y"
|
|
% (path, path[:-4] + ".%s" % format1)
|
|
)
|
|
except:
|
|
info += traceback.format_exc()
|
|
infos.append("%s->%s" % (os.path.basename(path), info))
|
|
yield "\n".join(infos)
|
|
yield "\n".join(infos)
|
|
except:
|
|
yield traceback.format_exc()
|
|
|