Create rvc.py

rvc.py

@@ -0,0 +1,556 @@
+import subprocess, torch, os, traceback, sys, warnings, shutil, numpy as np
+from mega import Mega
+os.environ["no_proxy"] = "localhost, 127.0.0.1, ::1"
+import threading
+from time import sleep
+from subprocess import Popen
+import faiss
+from random import shuffle
+import json, datetime, requests
+from gtts import gTTS
+now_dir = os.getcwd()
+sys.path.append(now_dir)
+tmp = os.path.join(now_dir, "TEMP")
+shutil.rmtree(tmp, ignore_errors=True)
+shutil.rmtree("%s/runtime/Lib/site-packages/infer_pack" % (now_dir), ignore_errors=True)
+os.makedirs(tmp, exist_ok=True)
+os.makedirs(os.path.join(now_dir, "logs"), exist_ok=True)
+os.makedirs(os.path.join(now_dir, "weights"), exist_ok=True)
+os.environ["TEMP"] = tmp
+warnings.filterwarnings("ignore")
+torch.manual_seed(114514)
+from i18n import I18nAuto
+
+import signal
+
+import math
+
+from utils import load_audio, CSVutil
+
+global DoFormant, Quefrency, Timbre
+
+if not os.path.isdir('csvdb/'):
+    os.makedirs('csvdb')
+    frmnt, stp = open("csvdb/formanting.csv", 'w'), open("csvdb/stop.csv", 'w')
+    frmnt.close()
+    stp.close()
+
+try:
+    DoFormant, Quefrency, Timbre = CSVutil('csvdb/formanting.csv', 'r', 'formanting')
+    DoFormant = (
+        lambda DoFormant: True if DoFormant.lower() == 'true' else (False if DoFormant.lower() == 'false' else DoFormant)
+    )(DoFormant)
+except (ValueError, TypeError, IndexError):
+    DoFormant, Quefrency, Timbre = False, 1.0, 1.0
+    CSVutil('csvdb/formanting.csv', 'w+', 'formanting', DoFormant, Quefrency, Timbre)
+
+def download_models():
+    # Download hubert base model if not present
+    if not os.path.isfile('./hubert_base.pt'):
+        response = requests.get('https://huggingface.co/lj1995/VoiceConversionWebUI/resolve/main/hubert_base.pt')
+
+        if response.status_code == 200:
+            with open('./hubert_base.pt', 'wb') as f:
+                f.write(response.content)
+            print("Downloaded hubert base model file successfully. File saved to ./hubert_base.pt.")
+        else:
+            raise Exception("Failed to download hubert base model file. Status code: " + str(response.status_code) + ".")
+        
+    # Download rmvpe model if not present
+    if not os.path.isfile('./rmvpe.pt'):
+        response = requests.get('https://drive.usercontent.google.com/download?id=1Hkn4kNuVFRCNQwyxQFRtmzmMBGpQxptI&export=download&authuser=0&confirm=t&uuid=0b3a40de-465b-4c65-8c41-135b0b45c3f7&at=APZUnTV3lA3LnyTbeuduura6Dmi2:1693724254058')
+
+        if response.status_code == 200:
+            with open('./rmvpe.pt', 'wb') as f:
+                f.write(response.content)
+            print("Downloaded rmvpe model file successfully. File saved to ./rmvpe.pt.")
+        else:
+            raise Exception("Failed to download rmvpe model file. Status code: " + str(response.status_code) + ".")
+
+download_models()
+
+
+def formant_apply(qfrency, tmbre):
+    Quefrency = qfrency
+    Timbre = tmbre
+    DoFormant = True
+    CSVutil('csvdb/formanting.csv', 'w+', 'formanting', DoFormant, qfrency, tmbre)
+    
+    return ({"value": Quefrency, "__type__": "update"}, {"value": Timbre, "__type__": "update"})
+
+def get_fshift_presets():
+    fshift_presets_list = []
+    for dirpath, _, filenames in os.walk("./formantshiftcfg/"):
+        for filename in filenames:
+            if filename.endswith(".txt"):
+                fshift_presets_list.append(os.path.join(dirpath,filename).replace('\\','/'))
+                
+    if len(fshift_presets_list) > 0:
+        return fshift_presets_list
+    else:
+        return ''
+
+
+
+def formant_enabled(cbox, qfrency, tmbre, frmntapply, formantpreset, formant_refresh_button):
+    
+    if (cbox):
+
+        DoFormant = True
+        CSVutil('csvdb/formanting.csv', 'w+', 'formanting', DoFormant, qfrency, tmbre)
+        #print(f"is checked? - {cbox}\ngot {DoFormant}")
+        
+        return (
+            {"value": True, "__type__": "update"},
+            {"visible": True, "__type__": "update"},
+            {"visible": True, "__type__": "update"},
+            {"visible": True, "__type__": "update"},
+            {"visible": True, "__type__": "update"},
+            {"visible": True, "__type__": "update"},
+        )
+        
+        
+    else:
+        
+        DoFormant = False
+        CSVutil('csvdb/formanting.csv', 'w+', 'formanting', DoFormant, qfrency, tmbre)
+        
+        #print(f"is checked? - {cbox}\ngot {DoFormant}")
+        return (
+            {"value": False, "__type__": "update"},
+            {"visible": False, "__type__": "update"},
+            {"visible": False, "__type__": "update"},
+            {"visible": False, "__type__": "update"},
+            {"visible": False, "__type__": "update"},
+            {"visible": False, "__type__": "update"},
+            {"visible": False, "__type__": "update"},
+        )
+        
+
+
+def preset_apply(preset, qfer, tmbr):
+    if str(preset) != '':
+        with open(str(preset), 'r') as p:
+            content = p.readlines()
+            qfer, tmbr = content[0].split('\n')[0], content[1]
+            
+            formant_apply(qfer, tmbr)
+    else:
+        pass
+    return ({"value": qfer, "__type__": "update"}, {"value": tmbr, "__type__": "update"})
+
+def update_fshift_presets(preset, qfrency, tmbre):
+    
+    qfrency, tmbre = preset_apply(preset, qfrency, tmbre)
+    
+    if (str(preset) != ''):
+        with open(str(preset), 'r') as p:
+            content = p.readlines()
+            qfrency, tmbre = content[0].split('\n')[0], content[1]
+            
+            formant_apply(qfrency, tmbre)
+    else:
+        pass
+    return (
+        {"choices": get_fshift_presets(), "__type__": "update"},
+        {"value": qfrency, "__type__": "update"},
+        {"value": tmbre, "__type__": "update"},
+    )
+
+i18n = I18nAuto()
+#i18n.print()
+# 判断是否有能用来训练和加速推理的N卡
+ngpu = torch.cuda.device_count()
+gpu_infos = []
+mem = []
+if (not torch.cuda.is_available()) or ngpu == 0:
+    if_gpu_ok = False
+else:
+    if_gpu_ok = False
+    for i in range(ngpu):
+        gpu_name = torch.cuda.get_device_name(i)
+        if (
+            "10" in gpu_name
+            or "16" in gpu_name
+            or "20" in gpu_name
+            or "30" in gpu_name
+            or "40" in gpu_name
+            or "A2" in gpu_name.upper()
+            or "A3" in gpu_name.upper()
+            or "A4" in gpu_name.upper()
+            or "P4" in gpu_name.upper()
+            or "A50" in gpu_name.upper()
+            or "A60" in gpu_name.upper()
+            or "70" in gpu_name
+            or "80" in gpu_name
+            or "90" in gpu_name
+            or "M4" in gpu_name.upper()
+            or "T4" in gpu_name.upper()
+            or "TITAN" in gpu_name.upper()
+        ):  # A10#A100#V100#A40#P40#M40#K80#A4500
+            if_gpu_ok = True  # 至少有一张能用的N卡
+            gpu_infos.append("%s\t%s" % (i, gpu_name))
+            mem.append(
+                int(
+                    torch.cuda.get_device_properties(i).total_memory
+                    / 1024
+                    / 1024
+                    / 1024
+                    + 0.4
+                )
+            )
+if if_gpu_ok == True and len(gpu_infos) > 0:
+    gpu_info = "\n".join(gpu_infos)
+    default_batch_size = min(mem) // 2
+else:
+    gpu_info = i18n("很遗憾您这没有能用的显卡来支持您训练")
+    default_batch_size = 1
+gpus = "-".join([i[0] for i in gpu_infos])
+from lib.infer_pack.models import (
+    SynthesizerTrnMs256NSFsid,
+    SynthesizerTrnMs256NSFsid_nono,
+    SynthesizerTrnMs768NSFsid,
+    SynthesizerTrnMs768NSFsid_nono,
+)
+import soundfile as sf
+from fairseq import checkpoint_utils
+import gradio as gr
+import logging
+from vc_infer_pipeline import VC
+from config import Config
+
+config = Config()
+# from trainset_preprocess_pipeline import PreProcess
+logging.getLogger("numba").setLevel(logging.WARNING)
+
+hubert_model = None
+
+def load_hubert():
+    global hubert_model
+    models, _, _ = checkpoint_utils.load_model_ensemble_and_task(
+        ["hubert_base.pt"],
+        suffix="",
+    )
+    hubert_model = models[0]
+    hubert_model = hubert_model.to(config.device)
+    if config.is_half:
+        hubert_model = hubert_model.half()
+    else:
+        hubert_model = hubert_model.float()
+    hubert_model.eval()
+
+
+weight_root = "weights"
+index_root = "logs"
+names = []
+for name in os.listdir(weight_root):
+    if name.endswith(".pth"):
+        names.append(name)
+index_paths = []
+for root, dirs, files in os.walk(index_root, topdown=False):
+    for name in files:
+        if name.endswith(".index") and "trained" not in name:
+            index_paths.append("%s/%s" % (root, name))
+
+
+
+def vc_single(
+    sid,
+    input_audio_path,
+    f0_up_key,
+    f0_file,
+    f0_method,
+    file_index,
+    #file_index2,
+    # file_big_npy,
+    index_rate,
+    filter_radius,
+    resample_sr,
+    rms_mix_rate,
+    protect,
+    crepe_hop_length,
+):  # spk_item, input_audio0, vc_transform0,f0_file,f0method0
+    global tgt_sr, net_g, vc, hubert_model, version
+    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, DoFormant, Quefrency, Timbre)
+        audio_max = np.abs(audio).max() / 0.95
+        if audio_max > 1:
+            audio /= audio_max
+        times = [0, 0, 0]
+        if hubert_model == None:
+            load_hubert()
+        if_f0 = cpt.get("f0", 1)
+        file_index = (
+            (
+                file_index.strip(" ")
+                .strip('"')
+                .strip("\n")
+                .strip('"')
+                .strip(" ")
+                .replace("trained", "added")
+            )
+        )  # 防止小白写错，自动帮他替换掉
+        # file_big_npy = (
+        #     file_big_npy.strip(" ").strip('"').strip("\n").strip('"').strip(" ")
+        # )
+        audio_opt = vc.pipeline(
+            hubert_model,
+            net_g,
+            sid,
+            audio,
+            input_audio_path,
+            times,
+            f0_up_key,
+            f0_method,
+            file_index,
+            # file_big_npy,
+            index_rate,
+            if_f0,
+            filter_radius,
+            tgt_sr,
+            resample_sr,
+            rms_mix_rate,
+            version,
+            protect,
+            crepe_hop_length,
+            f0_file=f0_file,
+        )
+        if resample_sr >= 16000 and tgt_sr != resample_sr:
+            tgt_sr = resample_sr
+        index_info = (
+            "Using index:%s." % file_index
+            if os.path.exists(file_index)
+            else "Index not used."
+        )
+        return "Success.\n %s\nTime:\n npy:%ss, f0:%ss, infer:%ss" % (
+            index_info,
+            times[0],
+            times[1],
+            times[2],
+        ), (tgt_sr, audio_opt)
+    except:
+        info = traceback.format_exc()
+        print(info)
+        return info, (None, None)
+
+
+def vc_multi(
+    sid,
+    dir_path,
+    opt_root,
+    paths,
+    f0_up_key,
+    f0_method,
+    file_index,
+    file_index2,
+    # file_big_npy,
+    index_rate,
+    filter_radius,
+    resample_sr,
+    rms_mix_rate,
+    protect,
+    format1,
+    crepe_hop_length,
+):
+    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 = vc_single(
+                sid,
+                path,
+                f0_up_key,
+                None,
+                f0_method,
+                file_index,
+                # file_big_npy,
+                index_rate,
+                filter_radius,
+                resample_sr,
+                rms_mix_rate,
+                protect,
+                crepe_hop_length
+            )
+            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()
+
+# 一个选项卡全局只能有一个音色
+def get_vc(sid):
+    global n_spk, tgt_sr, net_g, vc, cpt, version
+    if sid == "" or sid == []:
+        global hubert_model
+        if hubert_model != None:  # 考虑到轮询, 需要加个判断看是否 sid 是由有模型切换到无模型的
+            print("clean_empty_cache")
+            del net_g, n_spk, vc, hubert_model, tgt_sr  # ,cpt
+            hubert_model = net_g = n_spk = vc = hubert_model = tgt_sr = None
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+            ###楼下不这么折腾清理不干净
+            if_f0 = cpt.get("f0", 1)
+            version = cpt.get("version", "v1")
+            if version == "v1":
+                if if_f0 == 1:
+                    net_g = SynthesizerTrnMs256NSFsid(
+                        *cpt["config"], is_half=config.is_half
+                    )
+                else:
+                    net_g = SynthesizerTrnMs256NSFsid_nono(*cpt["config"])
+            elif version == "v2":
+                if if_f0 == 1:
+                    net_g = SynthesizerTrnMs768NSFsid(
+                        *cpt["config"], is_half=config.is_half
+                    )
+                else:
+                    net_g = SynthesizerTrnMs768NSFsid_nono(*cpt["config"])
+            del net_g, cpt
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+            cpt = None
+        return {"visible": False, "__type__": "update"}
+    person = "%s/%s" % (weight_root, sid)
+    print("loading %s" % person)
+    cpt = torch.load(person, map_location="cpu")
+    tgt_sr = cpt["config"][-1]
+    cpt["config"][-3] = cpt["weight"]["emb_g.weight"].shape[0]  # n_spk
+    if_f0 = cpt.get("f0", 1)
+    version = cpt.get("version", "v1")
+    if version == "v1":
+        if if_f0 == 1:
+            net_g = SynthesizerTrnMs256NSFsid(*cpt["config"], is_half=config.is_half)
+        else:
+            net_g = SynthesizerTrnMs256NSFsid_nono(*cpt["config"])
+    elif version == "v2":
+        if if_f0 == 1:
+            net_g = SynthesizerTrnMs768NSFsid(*cpt["config"], is_half=config.is_half)
+        else:
+            net_g = SynthesizerTrnMs768NSFsid_nono(*cpt["config"])
+    del net_g.enc_q
+    print(net_g.load_state_dict(cpt["weight"], strict=False))
+    net_g.eval().to(config.device)
+    if config.is_half:
+        net_g = net_g.half()
+    else:
+        net_g = net_g.float()
+    vc = VC(tgt_sr, config)
+    n_spk = cpt["config"][-3]
+    return {"visible": False, "maximum": n_spk, "__type__": "update"}
+
+
+def change_choices():
+    names = []
+    for name in os.listdir(weight_root):
+        if name.endswith(".pth"):
+            names.append(name)
+    index_paths = []
+    for root, dirs, files in os.walk(index_root, topdown=False):
+        for name in files:
+            if name.endswith(".index") and "trained" not in name:
+                index_paths.append("%s/%s" % (root, name))
+    return {"choices": sorted(names), "__type__": "update"}, {
+        "choices": sorted(index_paths),
+        "__type__": "update",
+    }
+
+
+def clean():
+    return {"value": "", "__type__": "update"}
+
+
+sr_dict = {
+    "32k": 32000,
+    "40k": 40000,
+    "48k": 48000,
+}
+
+
+def if_done(done, p):
+    while 1:
+        if p.poll() == None:
+            sleep(0.5)
+        else:
+            break
+    done[0] = True
+
+
+def if_done_multi(done, ps):
+    while 1:
+        # poll==None代表进程未结束
+        # 只要有一个进程未结束都不停
+        flag = 1
+        for p in ps:
+            if p.poll() == None:
+                flag = 0
+                sleep(0.5)
+                break
+        if flag == 1:
+            break
+    done[0] = True
+
+
+
+global log_interval
+
+
+def set_log_interval(exp_dir, batch_size12):
+    log_interval = 1
+
+    folder_path = os.path.join(exp_dir, "1_16k_wavs")
+
+    if os.path.exists(folder_path) and os.path.isdir(folder_path):
+        wav_files = [f for f in os.listdir(folder_path) if f.endswith(".wav")]
+        if wav_files:
+            sample_size = len(wav_files)
+            log_interval = math.ceil(sample_size / batch_size12)
+            if log_interval > 1:
+                log_interval += 1
+    return log_interval
+
+
+
+
+
+def whethercrepeornah(radio):
+    mango = True if radio == 'mangio-crepe' or radio == 'mangio-crepe-tiny' else False
+    return ({"visible": mango, "__type__": "update"})
+