import subprocess, torch, os, traceback, sys, warnings, shutil os.environ["no_proxy"] = "localhost, 127.0.0.1, ::1" import threading from time import sleep from subprocess import Popen import faiss 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) 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 edge_tts, asyncio from ilariatts import tts_order_voice 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 pipeline import VC from config import Config from utils import * config = Config() logging.getLogger("numba").setLevel(logging.WARNING) logging.getLogger("fairseq").setLevel(logging.WARNING) logging.getLogger("PIL").setLevel(logging.WARNING) logging.getLogger("imageio").setLevel(logging.WARNING) logging.getLogger("matplotlib").setLevel(logging.WARNING) hubert_model = None language_dict = tts_order_voice ilariavoices = language_dict.keys() 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/buckets/R-Kentaren/rvcmodels-bucket/resolve/main/hubert_base.pt?download=true') 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://huggingface.co/buckets/R-Kentaren/rvcmodels-bucket/resolve/main/rmvpe.pt?download=true') 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() print("\n-------------------------------\nRVC v2 Easy GUI\n-------------------------------\n") 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) 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) 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(language="pt_BR") 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() ): if_gpu_ok = True 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]) 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)) @spaces.GPU def vc_single( sid, input_audio_path, f0_up_key, f0_file, f0_method, file_index, index_rate, filter_radius, resample_sr, rms_mix_rate, protect, crepe_hop_length, ): 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 = 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") ) ) audio_opt = vc.pipeline( hubert_model, net_g, sid, audio, input_audio_path, times, f0_up_key, f0_method, file_index, 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 get_vc(sid): global n_spk, tgt_sr, net_g, vc, cpt, version if sid == "" or sid == []: global hubert_model if hubert_model != None: print("clean_empty_cache") del net_g, n_spk, vc, hubert_model, tgt_sr 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] 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 whethercrepeornah(radio): mango = True if radio == 'mangio-crepe' or radio == 'mangio-crepe-tiny' else False return ({"visible": mango, "__type__": "update"}) from lib.infer_pack.models_onnx import SynthesizerTrnMsNSFsidM def export_onnx(ModelPath, ExportedPath, MoeVS=True): cpt = torch.load(ModelPath, map_location="cpu") cpt["config"][-3] = cpt["weight"]["emb_g.weight"].shape[0] hidden_channels = 256 if cpt.get("version","v1")=="v1"else 768 test_phone = torch.rand(1, 200, hidden_channels) test_phone_lengths = torch.tensor([200]).long() test_pitch = torch.randint(size=(1, 200), low=5, high=255) test_pitchf = torch.rand(1, 200) test_ds = torch.LongTensor([0]) test_rnd = torch.rand(1, 192, 200) device = "cpu" net_g = SynthesizerTrnMsNSFsidM( *cpt["config"], is_half=False,version=cpt.get("version","v1") ) net_g.load_state_dict(cpt["weight"], strict=False) input_names = ["phone", "phone_lengths", "pitch", "pitchf", "ds", "rnd"] output_names = [ "audio", ] torch.onnx.export( net_g, ( test_phone.to(device), test_phone_lengths.to(device), test_pitch.to(device), test_pitchf.to(device), test_ds.to(device), test_rnd.to(device), ), ExportedPath, dynamic_axes={ "phone": [1], "pitch": [1], "pitchf": [1], "rnd": [2], }, do_constant_folding=False, opset_version=16, verbose=False, input_names=input_names, output_names=output_names, ) return "Finished" def change_choices2(): audio_files=[] for filename in os.listdir("./audios"): if filename.endswith(('.wav','.mp3','.ogg','.flac','.m4a','.aac','.mp4')): audio_files.append(os.path.join('./audios',filename).replace('\\', '/')) return {"choices": sorted(audio_files), "__type__": "update"}, {"__type__": "update"} audio_files=[] for filename in os.listdir("./audios"): if filename.endswith(('.wav','.mp3','.ogg','.flac','.m4a','.aac','.mp4')): audio_files.append(os.path.join('./audios',filename).replace('\\', '/')) def get_index(): if check_for_name() != '': chosen_model=sorted(names)[0].split(".")[0] logs_path="./logs/"+chosen_model if os.path.exists(logs_path): for file in os.listdir(logs_path): if file.endswith(".index"): return os.path.join(logs_path, file) return '' else: return '' def get_indexes(): indexes_list=[] for dirpath, dirnames, filenames in os.walk("./logs/"): for filename in filenames: if filename.endswith(".index"): indexes_list.append(os.path.join(dirpath,filename)) if len(indexes_list) > 0: return indexes_list else: return '' def get_name(): if len(audio_files) > 0: return sorted(audio_files)[0] else: return '' def save_to_wav(record_button): if record_button is None: pass else: path_to_file=record_button new_name = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")+'.wav' new_path='./audios/'+new_name shutil.move(path_to_file,new_path) return new_path def save_to_wav2(dropbox): file_path=dropbox.name shutil.move(file_path,'./audios') return os.path.join('./audios',os.path.basename(file_path)) def match_index(sid0): folder=sid0.split(".")[0] parent_dir="./logs/"+folder if os.path.exists(parent_dir): for filename in os.listdir(parent_dir): if filename.endswith(".index"): index_path=os.path.join(parent_dir,filename) return index_path else: return '' def check_for_name(): if len(names) > 0: return sorted(names)[0] else: return '' def ilariaTTS(text, ttsvoice): vo=language_dict[ttsvoice] asyncio.run(edge_tts.Communicate(text, vo).save("./temp_ilaria.mp3")) aud_path = save_to_wav('./temp_ilaria.mp3') return aud_path, aud_path def download_from_url(url, model): if not url: return "The URL cannot be empty." if not model: return "You need to name your model. For example: My model" url = url.strip() zip_dirs = ["zips", "unzips"] for directory in zip_dirs: if os.path.exists(directory): shutil.rmtree(directory) os.makedirs("zips", exist_ok=True) os.makedirs("unzips", exist_ok=True) zipfile = model + '.zip' zipfile_path = './zips/' + zipfile try: if "drive.google.com" in url: subprocess.run(["gdown", url, "--fuzzy", "-O", zipfile_path]) else: subprocess.run(["wget", url, "-O", zipfile_path]) for filename in os.listdir("./zips"): if filename.endswith(".zip"): zipfile_path = os.path.join("./zips/", filename) shutil.unpack_archive(zipfile_path, "./unzips", 'zip') else: return "No zipfile found." for root, dirs, files in os.walk('./unzips'): for file in files: file_path = os.path.join(root, file) if file.endswith(".index"): os.mkdir(f'./logs/{model}') shutil.copy2(file_path, f'./logs/{model}') elif "G_" not in file and "D_" not in file and file.endswith(".pth"): shutil.copy(file_path, f'./weights/{model}.pth') shutil.rmtree("zips") shutil.rmtree("unzips") return "Model downloaded, you can return to the inference page!" except: return "ERROR - download failed. Check if the link is valid." with gr.Blocks(theme=gr.themes.Default(primary_hue="red", secondary_hue="red"), title="RVC - AsukaDev", css=css) as app: gr.Markdown(badges) gr.Markdown(description) gr.HTML("

Easy GUI

") with gr.Tabs(): with gr.TabItem("Inference"): with gr.Row(): sid0 = gr.Dropdown(label="1. Choose your model", choices=sorted(names), value=check_for_name()) refresh_button = gr.Button("To update", variant="primary") if check_for_name() != '': get_vc(sorted(names)[0]) vc_transform0 = gr.Number(label="Change the tone here. If the voice is of the same sex, it is not necessary to change (12 if it is Male to Female, -12 if it is the other way around.", value=0) spk_item = gr.Slider( minimum=0, maximum=2333, step=1, label=i18n("请选择说话人id"), value=0, visible=False, interactive=True, ) sid0.change( fn=get_vc, inputs=[sid0], outputs=[spk_item], ) but0 = gr.Button("Converter", variant="primary") with gr.Row(): with gr.Column(): with gr.Row(): dropbox = gr.File(label="Drag your audio file and click update.") with gr.Row(): record_button=gr.Audio(sources="microphone", label="Or you can use your microphone!", type="filepath") with gr.Row(): input_audio0 = gr.Dropdown( label="2.Choose the audio file", value="./audios/Poema-do-Cume-Arnold", choices=audio_files ) dropbox.upload(fn=save_to_wav2, inputs=[dropbox], outputs=[input_audio0]) dropbox.upload(fn=change_choices2, inputs=[], outputs=[input_audio0]) refresh_button2 = gr.Button("To update", variant="primary", size='sm') record_button.change(fn=save_to_wav, inputs=[record_button], outputs=[input_audio0]) record_button.change(fn=change_choices2, inputs=[], outputs=[input_audio0]) with gr.Column(): vc_output2 = gr.Audio( label="Final result! (Click on the three dots to download the audio)", type='filepath', interactive=False, ) with gr.Accordion('Edge-TTS', open=True): with gr.Column(): ilariaid=gr.Dropdown(label="Voice:", choices=ilariavoices, value="Brazilian-Antonio- (Male)") ilariatext = gr.Textbox(label="Digite o seu Texto", interactive=True, value="Isso é um teste.") ilariatts_button = gr.Button(value="To speak") ilariatts_button.click(fn=ilariaTTS, inputs=[ilariatext, ilariaid], outputs=[record_button, input_audio0]) with gr.Accordion("Index Configuration", open=False): file_index1 = gr.Dropdown( label="3. Choose the index file (if it was not found automatically).", choices=get_indexes(), value=get_index(), interactive=True, ) sid0.change(fn=match_index, inputs=[sid0],outputs=[file_index1]) refresh_button.click( fn=change_choices, inputs=[], outputs=[sid0, file_index1] ) index_rate1 = gr.Slider( minimum=0, maximum=1, label=i18n("Search Feature Ratio"), value=0.66, interactive=True, ) with gr.Accordion("advanced options", open=False): f0method0 = gr.Radio( label="Optional: Change the pitch extraction algorithm. Extraction methods are ranked from “worst quality” to “best quality”. If you don't know what you're doing, quit rmvpe.", choices=["pm", "dio", "crepe-tiny", "mangio-crepe-tiny", "crepe", "harvest", "mangio-crepe", "rmvpe"], value="rmvpe", interactive=True, ) crepe_hop_length = gr.Slider( minimum=1, maximum=512, step=1, label="Mangio-Crepe Hop Length. Higher numbers will reduce the chance of extreme pitch changes but lower numbers will increase accuracy. 64-192 is a good range to experiment with.", value=120, interactive=True, visible=False, ) f0method0.change(fn=whethercrepeornah, inputs=[f0method0], outputs=[crepe_hop_length]) filter_radius0 = gr.Slider( minimum=0, maximum=7, label=i18n(">=3, use median filtering on the harvest tone recognition result, the value is the filter radius, which may weaken the muted sound."), value=3, step=1, interactive=True, ) resample_sr0 = gr.Slider( minimum=0, maximum=48000, label=i18n("后处理重采样至最终采样率,0为不进行重采样"), value=0, step=1, interactive=True, visible=False ) rms_mix_rate0 = gr.Slider( minimum=0, maximum=1, label=i18n("The volume envelope of the input source overrides the fusing rate of the output volume envelope. The closer it is to 1, the more output envelope is used."), value=0.21, interactive=True, ) protect0 = gr.Slider( minimum=0, maximum=0.5, label=i18n("Protect voiceless consonants and breath sounds to avoid artifacts such as electronic sound breakdown. Do not activate when it reaches 0.5. Lower it to increase protection, but may reduce the indexing effect."), value=0.33, step=0.01, interactive=True, ) formanting = gr.Checkbox( value=bool(DoFormant), label="[EXPERIMENTAL] Formant shift inference audio", info="Used for conversions from male to female and vice versa", interactive=True, visible=True, ) formant_preset = gr.Dropdown( value='', choices=get_fshift_presets(), label="browse presets for formanting", visible=bool(DoFormant), ) formant_refresh_button = gr.Button( value='\U0001f504', visible=bool(DoFormant), variant='primary', ) qfrency = gr.Slider( value=Quefrency, info="Default value is 1.0", label="Quefrency for formant shifting", minimum=0.0, maximum=16.0, step=0.1, visible=bool(DoFormant), interactive=True, ) tmbre = gr.Slider( value=Timbre, info="Default value is 1.0", label="Timbre for formant shifting", minimum=0.0, maximum=16.0, step=0.1, visible=bool(DoFormant), interactive=True, ) formant_preset.change(fn=preset_apply, inputs=[formant_preset, qfrency, tmbre], outputs=[qfrency, tmbre]) frmntbut = gr.Button("Apply", variant="primary", visible=bool(DoFormant)) formanting.change(fn=formant_enabled,inputs=[formanting,qfrency,tmbre,frmntbut,formant_preset,formant_refresh_button],outputs=[formanting,qfrency,tmbre,frmntbut,formant_preset,formant_refresh_button]) frmntbut.click(fn=formant_apply,inputs=[qfrency, tmbre], outputs=[qfrency, tmbre]) formant_refresh_button.click(fn=update_fshift_presets,inputs=[formant_preset, qfrency, tmbre],outputs=[formant_preset, qfrency, tmbre]) with gr.Row(): vc_output1 = gr.Textbox("") f0_file = gr.File(label=i18n("F0曲线文件, 可选, 一行一个音高, 代替默认F0及升降调"), visible=False) but0.click( vc_single, [ spk_item, input_audio0, vc_transform0, f0_file, f0method0, file_index1, index_rate1, filter_radius0, resample_sr0, rms_mix_rate0, protect0, crepe_hop_length ], [vc_output1, vc_output2], ) with gr.TabItem("Download new templates"): with gr.Row(): url=gr.Textbox(label="Huggingface Link:") with gr.Row(): model = gr.Textbox(label="Models Name:") download_button=gr.Button("Dowoad") with gr.Row(): status_bar=gr.Textbox(label="Status download") download_button.click(fn=download_from_url, inputs=[url, model], outputs=[status_bar]) gr.Markdown("

Easy GUI (English)

") gr.Markdown( """ original by Rejekts improved by AsukaDev """ ) app.launch(share=True, quiet=True, debug=False)