infer/modules/vc/modules.py

import os, sys
import traceback
import logging

now_dir = os.getcwd()
sys.path.append(now_dir)
logger = logging.getLogger(__name__)

logger.info(f"Current working directory: {now_dir}")
logger.info(f"Python path updated with: {now_dir}")

import lib.globals.globals as rvc_globals
import numpy as np
import soundfile as sf
import torch
from io import BytesIO
from infer.lib.audio import load_audio
from infer.lib.audio import wav2
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 *
import time
import scipy.io.wavfile as wavfile

logger.info("All imports completed successfully")

def note_to_hz(note_name):
    logger.info(f"Converting note to Hz: {note_name}")
    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])
    logger.debug(f"Parsed note - pitch_class: {pitch_class}, octave: {octave}")
    
    semitone = SEMITONES[pitch_class]
    note_number = 12 * (octave - 4) + semitone
    frequency = 440.0 * (2.0 ** (1.0/12)) ** note_number
    
    logger.debug(f"Note conversion result - semitone: {semitone}, note_number: {note_number}, frequency: {frequency}")
    return frequency

class VC:
    def __init__(self, config):
        logger.info("Initializing VC class")
        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
        logger.info(f"VC initialized with config: {config}")

    def get_vc(self, sid, *to_return_protect):
        logger.info(f"Starting get_vc with sid: {sid}")
        logger.debug(f"to_return_protect arguments: {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",
        }
        
        logger.debug(f"to_return_protect0: {to_return_protect0}")
        logger.debug(f"to_return_protect1: {to_return_protect1}")

        if not sid:
            logger.info("No sid provided, cleaning model cache")
            if self.hubert_model is not None:
                logger.info("Hubert model exists, proceeding with cleanup")
                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():
                    logger.info("CUDA available, emptying cache")
                    torch.cuda.empty_cache()
                else:
                    logger.info("CUDA not available")

            logger.info("Extracting f0 and version from checkpoint")
            self.if_f0 = self.cpt.get("f0", 1)
            self.version = self.cpt.get("version", "v1")
            logger.info(f"if_f0: {self.if_f0}, version: {self.version}")
            
            if self.version == "v1":
                logger.info("Using v1 model")
                if self.if_f0 == 1:
                    logger.info("Creating SynthesizerTrnMs256NSFsid with f0")
                    self.net_g = SynthesizerTrnMs256NSFsid(
                        *self.cpt["config"], is_half=self.config.is_half
                    )
                else:
                    logger.info("Creating SynthesizerTrnMs256NSFsid_nono without f0")
                    self.net_g = SynthesizerTrnMs256NSFsid_nono(*self.cpt["config"])
            elif self.version == "v2":
                logger.info("Using v2 model")
                if self.if_f0 == 1:
                    logger.info("Creating SynthesizerTrnMs768NSFsid with f0")
                    self.net_g = SynthesizerTrnMs768NSFsid(
                        *self.cpt["config"], is_half=self.config.is_half
                    )
                else:
                    logger.info("Creating SynthesizerTrnMs768NSFsid_nono without f0")
                    self.net_g = SynthesizerTrnMs768NSFsid_nono(*self.cpt["config"])

            logger.info("Deleting net_g and cpt, clearing CUDA cache")
            del self.net_g, self.cpt
            if torch.cuda.is_available():
                torch.cuda.empty_cache()
            
            logger.info("Returning early with cleanup complete")
            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 model from: {person}")
        logger.info(f"Loading...")
        
        try:
            logger.info("Loading checkpoint with torch.load")
            self.cpt = torch.load(person, map_location="cpu")
            logger.info("Checkpoint loaded successfully")
            logger.info(f"cpt type:{type(self.cpt['model'])}")
            logger.info(f"cpt details:{self.cpt['model'].keys()}")
        except Exception as e:
            logger.error(f"Failed to load checkpoint: {e}")
            raise

        self.tgt_sr = self.cpt["config"][-1]
        logger.info(f"Target sample rate: {self.tgt_sr}")
        
        self.cpt["config"][-3] = self.cpt["weight"]["emb_g.weight"].shape[0]  # n_spk
        logger.info(f"Number of speakers: {self.cpt['config'][-3]}")
        
        self.if_f0 = self.cpt.get("f0", 1)
        self.version = self.cpt.get("version", "v1")
        logger.info(f"Model configuration - if_f0: {self.if_f0}, version: {self.version}")
        
        synthesizer_class = {
            ("v1", 1): SynthesizerTrnMs256NSFsid,
            ("v1", 0): SynthesizerTrnMs256NSFsid_nono,
            ("v2", 1): SynthesizerTrnMs768NSFsid,
            ("v2", 0): SynthesizerTrnMs768NSFsid_nono,
        }
        
        selected_class = synthesizer_class.get(
            (self.version, self.if_f0), SynthesizerTrnMs256NSFsid
        )
        logger.info(f"Selected synthesizer class: {selected_class.__name__}")
        
        try:
            self.net_g = selected_class(*self.cpt["config"], is_half=self.config.is_half)
            logger.info("Network generator created successfully")
        except Exception as e:
            logger.error(f"Failed to create network generator: {e}")
            raise

        logger.info("Deleting encoder query module")
        del self.net_g.enc_q
        
        logger.info("Loading state dict")
        self.net_g.load_state_dict(self.cpt["weight"], strict=False)
        
        logger.info(f"Setting network to eval mode and moving to device: {self.config.device}")
        self.net_g.eval().to(self.config.device)
        
        if self.config.is_half:
            logger.info("Converting network to half precision")
            self.net_g = self.net_g.half()
        else:
            logger.info("Keeping network in float precision")
            self.net_g = self.net_g.float()

        logger.info("Creating pipeline")
        self.pipeline = Pipeline(self.tgt_sr, self.config)
        
        n_spk = self.cpt["config"][-3]
        logger.info(f"Final n_spk value: {n_spk}")
        
        index = {"value": get_index_path_from_model(sid), "__type__": "update"}
        logger.info(f"Index path: {index['value']}")
        logger.info("Select index: " + index["value"])
        
        result = (
            (
                {"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"}
        )
        logger.debug(f"get_vc returning: {result}")
        return result

    def vc_single(
        self,
        sid,
        input_audio_path0,
        input_audio_path1,
        f0_up_key,
        f0_file,
        f0_method,
        file_index,
        file_index2,
        index_rate,
        filter_radius,
        resample_sr,
        rms_mix_rate,
        protect,
        crepe_hop_length,
        f0_min,
        note_min,
        f0_max,
        note_max,
        f0_autotune,
    ):
        logger.info("Starting vc_single with parameters:")
        logger.info(f"sid: {sid}")
        logger.info(f"input_audio_path0: {input_audio_path0}")
        logger.info(f"input_audio_path1: {input_audio_path1}")
        logger.info(f"f0_up_key: {f0_up_key}")
        logger.info(f"f0_method: {f0_method}")
        logger.info(f"file_index: {file_index}")
        logger.debug(f"index_rate: {index_rate}")
        logger.debug(f"filter_radius: {filter_radius}")
        logger.debug(f"resample_sr: {resample_sr}")
        logger.debug(f"rms_mix_rate: {rms_mix_rate}")
        logger.debug(f"protect: {protect}")
        
        global total_time
        total_time = 0
        start_time = time.time()
        logger.info(f"Start time recorded: {start_time}")

        if not input_audio_path0 and not input_audio_path1:
            logger.error("No audio input provided")
            return "You need to upload an audio", None

        if (not os.path.exists(input_audio_path0)) and (not os.path.exists(os.path.join(now_dir, input_audio_path0))):
            logger.error(f"Audio file doesn't exist: {input_audio_path0}")
            return "Audio was not properly selected or doesn't exist", None

        input_audio_path1 = input_audio_path1 or input_audio_path0
        logger.info(f"Final input audio path: {input_audio_path1}")
        print(f"\nStarting inference for '{os.path.basename(input_audio_path1)}'")
        print("-------------------")

        f0_up_key = int(f0_up_key)
        logger.debug(f"f0_up_key converted to int: {f0_up_key}")

        if rvc_globals.NotesOrHertz and f0_method != 'rmvpe':
            logger.info("Using note-based pitch conversion")
            f0_min = note_to_hz(note_min) if note_min else 50
            f0_max = note_to_hz(note_max) if note_max else 1100
            print(f"Converted Min pitch: freq - {f0_min}\n"
                  f"Converted Max pitch: freq - {f0_max}")
        else:
            logger.info("Using frequency-based pitch")
            f0_min = f0_min or 50
            f0_max = f0_max or 1100
            logger.debug(f"f0_min: {f0_min}, f0_max: {f0_max}")

        try:
            input_audio_path1 = input_audio_path1 or input_audio_path0
            print(f"Attempting to load {input_audio_path1}....")
            
            logger.info("Loading audio with parameters:")
            logger.debug(f"DoFormant: {rvc_globals.DoFormant}")
            logger.debug(f"Quefrency: {rvc_globals.Quefrency}")
            logger.debug(f"Timbre: {rvc_globals.Timbre}")
            
            audio = load_audio(file=input_audio_path1, sr=16000, 
                             DoFormant=rvc_globals.DoFormant, 
                             Quefrency=rvc_globals.Quefrency, 
                             Timbre=rvc_globals.Timbre)
            logger.info(f"Audio loaded successfully, shape: {audio.shape}")
            
            audio_max = np.abs(audio).max() / 0.95
            logger.debug(f"Audio max value: {audio_max}")
            
            if audio_max > 1:
                logger.info("Normalizing audio")
                audio /= audio_max
                logger.debug(f"Audio normalized, new max: {np.abs(audio).max()}")

            times = [0, 0, 0]
            logger.debug(f"Times array initialized: {times}")

            if self.hubert_model is None:
                logger.info("Loading Hubert model")
                self.hubert_model = load_hubert(self.config)
                logger.info("Hubert model loaded successfully")
            else:
                logger.debug("Hubert model already loaded")

            try:
                self.if_f0 = self.cpt.get("f0", 1)
                logger.debug(f"if_f0 value: {self.if_f0}")
            except NameError:
                message = "Model was not properly selected"
                logger.error(message)
                return message, None

            logger.debug(f"Processing file_index: {file_index}")
            file_index = (
                (
                    file_index.strip(" ")
                    .strip('"')
                    .strip("\n")
                    .strip('"')
                    .strip(" ")
                    .replace("trained", "added")
                )
                if file_index != ""
                else file_index2
            )
            logger.debug(f"Processed file_index: {file_index}")

            logger.info("Starting pipeline processing")
            try:
                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_file=f0_file,
                    f0_min=f0_min,
                    f0_max=f0_max
                )
                logger.info(f"Pipeline processing completed, output shape: {audio_opt.shape}")
                logger.info(f"Processing times: {times}")
                
            except AssertionError as e:
                message = "Mismatching index version detected (v1 with v2, or v2 with v1)."
                logger.error(f"AssertionError: {message}")
                logger.error(f"Exception details: {e}")
                print(message)
                return message, None
            except NameError as e:
                message = "RVC libraries are still loading. Please try again in a few seconds."
                logger.error(f"NameError: {message}")
                logger.error(f"Exception details: {e}")
                print(message)
                return message, None

            if self.tgt_sr != resample_sr >= 16000:
                logger.info(f"Updating target sample rate from {self.tgt_sr} to {resample_sr}")
                self.tgt_sr = resample_sr

            index_info = (
                "Index:\n%s." % file_index
                if os.path.exists(file_index)
                else "Index not used."
            )
            logger.debug(f"Index info: {index_info}")

            end_time = time.time()
            total_time = end_time - start_time
            logger.info(f"End time: {end_time}, Total processing time: {total_time}")

            output_folder = "audio-outputs"
            os.makedirs(output_folder, exist_ok=True)
            logger.info(f"Output folder created/verified: {output_folder}")
            
            output_filename = "generated_audio_{}.wav"
            output_count = 1
            while True:
                current_output_path = os.path.join(output_folder, output_filename.format(output_count))
                if not os.path.exists(current_output_path):
                    break
                output_count += 1
            
            logger.info(f"Saving audio to: {current_output_path}")
            wavfile.write(current_output_path, self.tgt_sr, audio_opt)
            print(f"Generated audio saved to: {current_output_path}")

            success_message = f"Success.\n {index_info}\nTime:\n npy:{times[0]}, f0:{times[1]}, infer:{times[2]}\nTotal Time: {total_time} seconds"
            logger.info(f"Returning success: {success_message}")
            return success_message, (self.tgt_sr, audio_opt)

        except:
            info = traceback.format_exc()
            logger.error(f"Exception in vc_single: {info}")
            logger.warn(info)
            return info, (None, None)

    def vc_single_dont_save(
        self,
        sid,
        input_audio_path0,
        input_audio_path1,
        f0_up_key,
        f0_file,
        f0_method,
        file_index,
        file_index2,
        index_rate,
        filter_radius,
        resample_sr,
        rms_mix_rate,
        protect,
        crepe_hop_length,
        f0_min,
        note_min,
        f0_max,
        note_max,
        f0_autotune,
    ):
        logger.info("Starting vc_single_dont_save (no file saving)")
        logger.debug("Parameters same as vc_single")
        
        global total_time
        total_time = 0
        start_time = time.time()

        if not input_audio_path0 and not input_audio_path1:
            logger.error("No audio input provided")
            return "You need to upload an audio", None

        if (not os.path.exists(input_audio_path0)) and (not os.path.exists(os.path.join(now_dir, input_audio_path0))):
            logger.error(f"Audio file doesn't exist: {input_audio_path0}")
            return "Audio was not properly selected or doesn't exist", None

        input_audio_path1 = input_audio_path1 or input_audio_path0
        logger.info(f"Final input audio path: {input_audio_path1}")
        print(f"\nStarting inference for '{os.path.basename(input_audio_path1)}'")
        print("-------------------")

        f0_up_key = int(f0_up_key)
        logger.debug(f"f0_up_key converted to int: {f0_up_key}")

        if rvc_globals.NotesOrHertz and f0_method != 'rmvpe':
            logger.info("Using note-based pitch conversion")
            f0_min = note_to_hz(note_min) if note_min else 50
            f0_max = note_to_hz(note_max) if note_max else 1100
            print(f"Converted Min pitch: freq - {f0_min}\n"
                  f"Converted Max pitch: freq - {f0_max}")
        else:
            logger.info("Using frequency-based pitch")
            f0_min = f0_min or 50
            f0_max = f0_max or 1100

        try:
            input_audio_path1 = input_audio_path1 or input_audio_path0
            print(f"Attempting to load {input_audio_path1}....")
            
            audio = load_audio(file=input_audio_path1, sr=16000, 
                             DoFormant=rvc_globals.DoFormant, 
                             Quefrency=rvc_globals.Quefrency, 
                             Timbre=rvc_globals.Timbre)
            logger.info(f"Audio loaded successfully, shape: {audio.shape}")
            
            audio_max = np.abs(audio).max() / 0.95
            if audio_max > 1:
                audio /= audio_max
                logger.info("Audio normalized")

            times = [0, 0, 0]

            if self.hubert_model is None:
                logger.info("Loading Hubert model")
                self.hubert_model = load_hubert(self.config)

            try:
                self.if_f0 = self.cpt.get("f0", 1)
            except NameError:
                message = "Model was not properly selected"
                logger.error(message)
                return message, None

            file_index = (
                (
                    file_index.strip(" ")
                    .strip('"')
                    .strip("\n")
                    .strip('"')
                    .strip(" ")
                    .replace("trained", "added")
                )
                if file_index != ""
                else file_index2
            )

            logger.info("Starting pipeline processing (dont_save version)")
            try:
                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_file=f0_file,
                    f0_min=f0_min,
                    f0_max=f0_max
                )
                logger.info(f"Pipeline processing completed, output shape: {audio_opt.shape}")
                
            except AssertionError:
                message = "Mismatching index version detected (v1 with v2, or v2 with v1)."
                logger.error(message)
                print(message)
                return message, None
            except NameError:
                message = "RVC libraries are still loading. Please try again in a few seconds."
                logger.error(message)
                print(message)
                return message, None

            if self.tgt_sr != resample_sr >= 16000:
                self.tgt_sr = resample_sr

            index_info = (
                "Index:\n%s." % file_index
                if os.path.exists(file_index)
                else "Index not used."
            )

            end_time = time.time()
            total_time = end_time - start_time
            logger.info(f"Processing completed, total time: {total_time}")
            logger.info("NOT saving file (dont_save version)")

            return f"Success.\n {index_info}\nTime:\n npy:{times[0]}, f0:{times[1]}, infer:{times[2]}\nTotal Time: {total_time} seconds", (self.tgt_sr, audio_opt)

        except:
            info = traceback.format_exc()
            logger.error(f"Exception in vc_single_dont_save: {info}")
            logger.warn(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,
        crepe_hop_length,
        f0_min,
        note_min,
        f0_max,
        note_max,
        f0_autotune,
    ):
        logger.info("Starting vc_multi batch processing")
        logger.info(f"sid: {sid}")
        logger.info(f"dir_path: {dir_path}")
        logger.info(f"opt_root: {opt_root}")
        logger.debug(f"paths: {paths}")
        logger.info(f"format1: {format1}")
        
        if rvc_globals.NotesOrHertz and f0_method != 'rmvpe':
            logger.info("Using note-based pitch conversion for batch")
            f0_min = note_to_hz(note_min) if note_min else 50
            f0_max = note_to_hz(note_max) if note_max else 1100
            print(f"Converted Min pitch: freq - {f0_min}\n"
                  f"Converted Max pitch: freq - {f0_max}")
        else:
            logger.info("Using frequency-based pitch for batch")
            f0_min = f0_min or 50
            f0_max = f0_max or 1100

        try:
            dir_path = (
                dir_path.strip(" ").strip('"').strip("\n").strip('"').strip(" ")
            )
            opt_root = opt_root.strip(" ").strip('"').strip("\n").strip('"').strip(" ")
            
            logger.info(f"Cleaned dir_path: {dir_path}")
            logger.info(f"Cleaned opt_root: {opt_root}")
            
            os.makedirs(opt_root, exist_ok=True)
            logger.info(f"Output directory created/verified: {opt_root}")

            try:
                if dir_path != "":
                    logger.info(f"Reading files from directory: {dir_path}")
                    paths = [
                        os.path.join(dir_path, name) for name in os.listdir(dir_path)
                    ]
                    logger.info(f"Found {len(paths)} files in directory")
                else:
                    logger.info("Using provided paths")
                    paths = [path.name for path in paths]
            except:
                logger.error("Error processing paths")
                traceback.print_exc()
                paths = [path.name for path in paths]

            logger.info(f"Processing {len(paths)} files:")
            for i, path in enumerate(paths):
                logger.debug(f"File {i+1}: {path}")

            infos = []
            for i, path in enumerate(paths):
                logger.info(f"Processing file {i+1}/{len(paths)}: {os.path.basename(path)}")
                
                info, opt = self.vc_single(
                    sid,
                    path,
                    f0_up_key,
                    None,
                    f0_method,
                    file_index,
                    file_index2,
                    index_rate,
                    filter_radius,
                    resample_sr,
                    rms_mix_rate,
                    protect,
                )
                
                logger.debug(f"vc_single result for {os.path.basename(path)}: {info[:100]}...")
                
                if "Success" in info:
                    logger.info(f"Processing successful for {os.path.basename(path)}")
                    try:
                        tgt_sr, audio_opt = opt
                        output_path = "%s/%s.%s" % (opt_root, os.path.basename(path), format1)
                        logger.info(f"Saving to: {output_path}")
                        
                        if format1 in ["wav", "flac"]:
                            logger.debug(f"Using soundfile to save {format1} format")
                            sf.write(output_path, audio_opt, tgt_sr)
                        else:
                            logger.debug(f"Using wav2 conversion for {format1} format")
                            with BytesIO() as wavf:
                                sf.write(wavf, audio_opt, tgt_sr, format="wav")
                                wavf.seek(0, 0)
                                with open(output_path, "wb") as outf:
                                    wav2(wavf, outf, format1)
                        
                        logger.info(f"File saved successfully: {output_path}")
                        
                    except:
                        logger.error(f"Failed to save {os.path.basename(path)}")
                        info += traceback.format_exc()
                else:
                    logger.warning(f"Processing failed for {os.path.basename(path)}")
                
                result_info = "%s->%s" % (os.path.basename(path), info)
                infos.append(result_info)
                logger.debug(f"Added to results: {result_info[:100]}...")
                
                # Yield progress update
                progress_info = "\n".join(infos)
                logger.info(f"Yielding progress update for {i+1}/{len(paths)} files")
                yield progress_info

            final_result = "\n".join(infos)
            logger.info(f"Batch processing completed. Final result length: {len(final_result)}")
            logger.info("Yielding final result")
            yield final_result
            
        except:
            error_info = traceback.format_exc()
            logger.error(f"Exception in vc_multi: {error_info}")
            yield error_info