| import os, sys |
| import traceback |
| import logging |
| now_dir = os.getcwd() |
| sys.path.append(now_dir) |
| logger = logging.getLogger(__name__) |
| import numpy as np |
| import soundfile as sf |
| import torch |
| from io import BytesIO |
| from lib.infer_libs.audio import load_audio |
| from lib.infer_libs.audio import wav2 |
| from lib.infer_libs.infer_pack.models import ( |
| SynthesizerTrnMs256NSFsid, |
| SynthesizerTrnMs256NSFsid_nono, |
| SynthesizerTrnMs768NSFsid, |
| SynthesizerTrnMs768NSFsid_nono, |
| ) |
| from lib.pipeline import Pipeline |
| import time |
| import glob |
| from shutil import move |
| from fairseq import checkpoint_utils |
|
|
| sup_audioext = { |
| "wav", |
| "mp3", |
| "flac", |
| "ogg", |
| "opus", |
| "m4a", |
| "mp4", |
| "aac", |
| "alac", |
| "wma", |
| "aiff", |
| "webm", |
| "ac3", |
| } |
|
|
| def note_to_hz(note_name): |
| try: |
| SEMITONES = {'C': -9, 'C#': -8, 'D': -7, 'D#': -6, 'E': -5, 'F': -4, 'F#': -3, 'G': -2, 'G#': -1, 'A': 0, 'A#': 1, 'B': 2} |
| pitch_class, octave = note_name[:-1], int(note_name[-1]) |
| semitone = SEMITONES[pitch_class] |
| note_number = 12 * (octave - 4) + semitone |
| frequency = 440.0 * (2.0 ** (1.0/12)) ** note_number |
| return frequency |
| except: |
| return None |
|
|
| def load_hubert(hubert_model_path, config): |
| models, _, _ = checkpoint_utils.load_model_ensemble_and_task( |
| [hubert_model_path], |
| 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() |
| return hubert_model.eval() |
|
|
| 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_protect): |
| logger.info("Get sid: " + sid) |
|
|
| to_return_protect0 = { |
| "visible": self.if_f0 != 0, |
| "value": to_return_protect[0] |
| if self.if_f0 != 0 and to_return_protect |
| else 0.5, |
| "__type__": "update", |
| } |
| to_return_protect1 = { |
| "visible": self.if_f0 != 0, |
| "value": to_return_protect[1] |
| if self.if_f0 != 0 and to_return_protect |
| else 0.33, |
| "__type__": "update", |
| } |
|
|
| if sid == "" or sid == []: |
| if self.hubert_model is not None: |
| logger.info("Clean model cache") |
| del ( |
| self.net_g, |
| self.n_spk, |
| self.vc, |
| self.hubert_model, |
| self.tgt_sr, |
| ) |
| self.hubert_model = ( |
| self.net_g |
| ) = self.n_spk = self.vc = self.hubert_model = self.tgt_sr = None |
| if torch.cuda.is_available(): |
| torch.cuda.empty_cache() |
| |
| self.if_f0 = self.cpt.get("f0", 1) |
| self.version = self.cpt.get("version", "v1") |
| if self.version == "v1": |
| if self.if_f0 == 1: |
| self.net_g = SynthesizerTrnMs256NSFsid( |
| *self.cpt["config"], is_half=self.config.is_half |
| ) |
| else: |
| self.net_g = SynthesizerTrnMs256NSFsid_nono(*self.cpt["config"]) |
| elif self.version == "v2": |
| if self.if_f0 == 1: |
| self.net_g = SynthesizerTrnMs768NSFsid( |
| *self.cpt["config"], is_half=self.config.is_half |
| ) |
| else: |
| self.net_g = SynthesizerTrnMs768NSFsid_nono(*self.cpt["config"]) |
| del self.net_g, self.cpt |
| if torch.cuda.is_available(): |
| torch.cuda.empty_cache() |
| return ( |
| {"visible": False, "__type__": "update"}, |
| { |
| "visible": True, |
| "value": to_return_protect0, |
| "__type__": "update", |
| }, |
| { |
| "visible": True, |
| "value": to_return_protect1, |
| "__type__": "update", |
| }, |
| "", |
| "", |
| ) |
| |
| person = f'{sid}' |
| |
| logger.info(f"Loading...") |
| 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] |
| self.if_f0 = self.cpt.get("f0", 1) |
| self.version = self.cpt.get("version", "v1") |
|
|
| 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] |
| |
| |
|
|
| return ( |
| ( |
| {"visible": False, "maximum": n_spk, "__type__": "update"}, |
| to_return_protect0, |
| to_return_protect1 |
| ) |
| if to_return_protect |
| else {"visible": False, "maximum": n_spk, "__type__": "update"} |
| ) |
|
|
| def vc_single_dont_save( |
| self, |
| sid, |
| input_audio_path1, |
| f0_up_key, |
| f0_method, |
| file_index, |
| file_index2, |
| index_rate, |
| filter_radius, |
| resample_sr, |
| rms_mix_rate, |
| protect, |
| crepe_hop_length, |
| do_formant, |
| quefrency, |
| timbre, |
| f0_min, |
| f0_max, |
| f0_autotune, |
| hubert_model_path = "assets/hubert/hubert_base.pt" |
| ): |
| """ |
| Performs inference without saving |
| |
| Parameters: |
| - sid (int) |
| - input_audio_path1 (str) |
| - f0_up_key (int) |
| - f0_method (str) |
| - file_index (str) |
| - file_index2 (str) |
| - index_rate (float) |
| - filter_radius (int) |
| - resample_sr (int) |
| - rms_mix_rate (float) |
| - protect (float) |
| - crepe_hop_length (int) |
| - do_formant (bool) |
| - quefrency (float) |
| - timbre (float) |
| - f0_min (str) |
| - f0_max (str) |
| - f0_autotune (bool) |
| - hubert_model_path (str) |
| |
| Returns: |
| Tuple(Tuple(status, index_info, times), Tuple(sr, data)): |
| - Tuple(status, index_info, times): |
| - status (str): either "Success." or an error |
| - index_info (str): index path if used |
| - times (list): [npy_time, f0_time, infer_time, total_time] |
| - Tuple(sr, data): Audio data results. |
| """ |
| global total_time |
| total_time = 0 |
| start_time = time.time() |
| |
| if not input_audio_path1: |
| return "You need to upload an audio", None |
| |
| if not os.path.exists(input_audio_path1): |
| return "Audio was not properly selected or doesn't exist", None |
| |
| f0_up_key = int(f0_up_key) |
| if not f0_min.isdigit(): |
| f0_min = note_to_hz(f0_min) |
| if f0_min: |
| print(f"Converted Min pitch: freq - {f0_min}") |
| else: |
| f0_min = 50 |
| print("Invalid minimum pitch note. Defaulting to 50hz.") |
| else: |
| f0_min = float(f0_min) |
| if not f0_max.isdigit(): |
| f0_max = note_to_hz(f0_max) |
| if f0_max: |
| print(f"Converted Max pitch: freq - {f0_max}") |
| else: |
| f0_max = 1100 |
| print("Invalid maximum pitch note. Defaulting to 1100hz.") |
| else: |
| f0_max = float(f0_max) |
| |
| try: |
| print(f"Attempting to load {input_audio_path1}....") |
| audio = load_audio(file=input_audio_path1, |
| sr=16000, |
| DoFormant=do_formant, |
| Quefrency=quefrency, |
| Timbre=timbre) |
| |
| 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(hubert_model_path, self.config) |
|
|
| try: |
| self.if_f0 = self.cpt.get("f0", 1) |
| except NameError: |
| message = "Model was not properly selected" |
| print(message) |
| return message, None |
| |
| if file_index and not file_index == "" and isinstance(file_index, str): |
| file_index = file_index.strip(" ") \ |
| .strip('"') \ |
| .strip("\n") \ |
| .strip('"') \ |
| .strip(" ") \ |
| .replace("trained", "added") |
| elif file_index2: |
| file_index = file_index2 |
| else: |
| file_index = "" |
|
|
| audio_opt = self.pipeline.pipeline( |
| self.hubert_model, |
| self.net_g, |
| sid, |
| audio, |
| input_audio_path1, |
| 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, |
| crepe_hop_length, |
| f0_autotune, |
| f0_min=f0_min, |
| f0_max=f0_max |
| ) |
|
|
| if self.tgt_sr != resample_sr >= 16000: |
| tgt_sr = resample_sr |
| else: |
| tgt_sr = self.tgt_sr |
| index_info = ( |
| "Index: %s." % file_index |
| if isinstance(file_index, str) and os.path.exists(file_index) |
| else "Index not used." |
| ) |
| end_time = time.time() |
| total_time = end_time - start_time |
| times.append(total_time) |
| return ( |
| ("Success.", index_info, times), |
| (tgt_sr, audio_opt), |
| ) |
| except: |
| info = traceback.format_exc() |
| logger.warn(info) |
| return ( |
| (info, None, [None, None, None, None]), |
| (None, None) |
| ) |
|
|
| def vc_single( |
| self, |
| sid, |
| input_audio_path1, |
| f0_up_key, |
| f0_method, |
| file_index, |
| file_index2, |
| index_rate, |
| filter_radius, |
| resample_sr, |
| rms_mix_rate, |
| protect, |
| format1, |
| crepe_hop_length, |
| do_formant, |
| quefrency, |
| timbre, |
| f0_min, |
| f0_max, |
| f0_autotune, |
| hubert_model_path = "assets/hubert/hubert_base.pt" |
| ): |
| """ |
| Performs inference with saving |
| |
| Parameters: |
| - sid (int) |
| - input_audio_path1 (str) |
| - f0_up_key (int) |
| - f0_method (str) |
| - file_index (str) |
| - file_index2 (str) |
| - index_rate (float) |
| - filter_radius (int) |
| - resample_sr (int) |
| - rms_mix_rate (float) |
| - protect (float) |
| - format1 (str) |
| - crepe_hop_length (int) |
| - do_formant (bool) |
| - quefrency (float) |
| - timbre (float) |
| - f0_min (str) |
| - f0_max (str) |
| - f0_autotune (bool) |
| - hubert_model_path (str) |
| |
| Returns: |
| Tuple(Tuple(status, index_info, times), Tuple(sr, data), output_path): |
| - Tuple(status, index_info, times): |
| - status (str): either "Success." or an error |
| - index_info (str): index path if used |
| - times (list): [npy_time, f0_time, infer_time, total_time] |
| - Tuple(sr, data): Audio data results. |
| - output_path (str): Audio results path |
| """ |
| global total_time |
| total_time = 0 |
| start_time = time.time() |
| |
| if not input_audio_path1: |
| return "You need to upload an audio", None, None |
| |
| if not os.path.exists(input_audio_path1): |
| return "Audio was not properly selected or doesn't exist", None, None |
|
|
| f0_up_key = int(f0_up_key) |
| if not f0_min.isdigit(): |
| f0_min = note_to_hz(f0_min) |
| if f0_min: |
| print(f"Converted Min pitch: freq - {f0_min}") |
| else: |
| f0_min = 50 |
| print("Invalid minimum pitch note. Defaulting to 50hz.") |
| else: |
| f0_min = float(f0_min) |
| if not f0_max.isdigit(): |
| f0_max = note_to_hz(f0_max) |
| if f0_max: |
| print(f"Converted Max pitch: freq - {f0_max}") |
| else: |
| f0_max = 1100 |
| print("Invalid maximum pitch note. Defaulting to 1100hz.") |
| else: |
| f0_max = float(f0_max) |
|
|
| try: |
| print(f"Attempting to load {input_audio_path1}...") |
| audio = load_audio(file=input_audio_path1, |
| sr=16000, |
| DoFormant=do_formant, |
| Quefrency=quefrency, |
| Timbre=timbre) |
| |
| 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(hubert_model_path, self.config) |
|
|
| try: |
| self.if_f0 = self.cpt.get("f0", 1) |
| except NameError: |
| message = "Model was not properly selected" |
| print(message) |
| return message, None |
| if file_index and not file_index == "" and isinstance(file_index, str): |
| file_index = file_index.strip(" ") \ |
| .strip('"') \ |
| .strip("\n") \ |
| .strip('"') \ |
| .strip(" ") \ |
| .replace("trained", "added") |
| elif file_index2: |
| file_index = file_index2 |
| else: |
| file_index = "" |
|
|
| audio_opt = self.pipeline.pipeline( |
| self.hubert_model, |
| self.net_g, |
| sid, |
| audio, |
| input_audio_path1, |
| 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, |
| crepe_hop_length, |
| f0_autotune, |
| f0_min=f0_min, |
| f0_max=f0_max |
| ) |
|
|
| if self.tgt_sr != resample_sr >= 16000: |
| tgt_sr = resample_sr |
| else: |
| tgt_sr = self.tgt_sr |
| index_info = ( |
| "Index: %s." % file_index |
| if isinstance(file_index, str) and os.path.exists(file_index) |
| else "Index not used." |
| ) |
| |
| opt_root = os.path.join(os.getcwd(), "output") |
| os.makedirs(opt_root, exist_ok=True) |
| output_count = 1 |
| |
| while True: |
| opt_filename = f"{os.path.splitext(os.path.basename(input_audio_path1))[0]}{os.path.basename(os.path.dirname(file_index))}{f0_method.capitalize()}_{output_count}.{format1}" |
| current_output_path = os.path.join(opt_root, opt_filename) |
| if not os.path.exists(current_output_path): |
| break |
| output_count += 1 |
| try: |
| if format1 in ["wav", "flac"]: |
| sf.write( |
| current_output_path, |
| audio_opt, |
| self.tgt_sr, |
| ) |
| else: |
| with BytesIO() as wavf: |
| sf.write( |
| wavf, |
| audio_opt, |
| self.tgt_sr, |
| format="wav" |
| ) |
| wavf.seek(0, 0) |
| with open(current_output_path, "wb") as outf: |
| wav2(wavf, outf, format1) |
| except: |
| info = traceback.format_exc() |
| end_time = time.time() |
| total_time = end_time - start_time |
| times.append(total_time) |
| return ( |
| ("Success.", index_info, times), |
| (tgt_sr, audio_opt), |
| current_output_path |
| ) |
| except: |
| info = traceback.format_exc() |
| logger.warn(info) |
| return ( |
| (info, None, [None, None, None, None]), |
| (None, None), |
| None |
| ) |
