from infer_rvc_python.lib.log_config import logger
import torch
import torch.nn as nn
import gc
import numpy as np
import os
import warnings
import threading
from tqdm import tqdm
from transformers import HubertConfig, HubertModel
from infer_rvc_python.lib.infer_pack.models import (
    SynthesizerTrnMs256NSFsid,
    SynthesizerTrnMs256NSFsid_nono,
    SynthesizerTrnMs768NSFsid,
    SynthesizerTrnMs768NSFsid_nono,
)
from infer_rvc_python.lib.audio import load_audio
import soundfile as sf
from scipy import signal
from time import time as ttime
import faiss
from infer_rvc_python.root_pipe import VC, change_rms, bh, ah
import librosa
from urllib.parse import urlparse
import copy

warnings.filterwarnings("ignore")
SUPPORTED_SAMPLE_RATES = [8000, 12000, 16000, 22050, 24000, 32000, 44100, 48000]  # 96000


class Config:
    def __init__(self, only_cpu=False):
        self.device = "cuda:0"
        self.is_half = True
        self.n_cpu = 0
        self.gpu_name = None
        self.gpu_mem = None
        (
            self.x_pad,
            self.x_query,
            self.x_center,
            self.x_max
        ) = self.device_config(only_cpu)

    def device_config(self, only_cpu) -> tuple:
        if torch.cuda.is_available() and not only_cpu:
            i_device = int(self.device.split(":")[-1])
            self.gpu_name = torch.cuda.get_device_name(i_device)
            if (
                ("16" in self.gpu_name and "V100" not in self.gpu_name.upper())
                or "P40" in self.gpu_name.upper()
                or "1060" in self.gpu_name
                or "1070" in self.gpu_name
                or "1080" in self.gpu_name
            ):
                logger.info(
                    "16/10 Series GPUs and P40 excel "
                    "in single-precision tasks."
                )
                self.is_half = False
            else:
                self.gpu_name = None
            self.gpu_mem = int(
                torch.cuda.get_device_properties(i_device).total_memory
                / 1024
                / 1024
                / 1024
                + 0.4
            )
        elif torch.backends.mps.is_available() and not only_cpu:
            logger.info("Supported N-card not found, using MPS for inference")
            self.device = "mps"
        else:
            logger.info("No supported N-card found, using CPU for inference")
            self.device = "cpu"
            self.is_half = False

        if self.n_cpu == 0:
            self.n_cpu = os.cpu_count()

        if self.is_half:
            # 6GB VRAM configuration
            x_pad = 3
            x_query = 10
            x_center = 60
            x_max = 65
        else:
            # 5GB VRAM configuration
            x_pad = 1
            x_query = 6
            x_center = 38
            x_max = 41

        if self.gpu_mem is not None and self.gpu_mem <= 4:
            x_pad = 1
            x_query = 5
            x_center = 30
            x_max = 32

        logger.info(
            f"Config: Device is {self.device}, "
            f"half precision is {self.is_half}"
        )

        return x_pad, x_query, x_center, x_max


BASE_DOWNLOAD_LINK = "https://huggingface.co/r3gm/sonitranslate_voice_models/resolve/main/"
BASE_MODELS = [
    "rmvpe.pt"
]
BASE_DIR = "."


def load_file_from_url(
    url: str,
    model_dir: str,
    file_name: str | None = None,
    overwrite: bool = False,
    progress: bool = True,
) -> str:
    """Download a file from `url` into `model_dir`,
    using the file present if possible.

    Returns the path to the downloaded file.
    """
    os.makedirs(model_dir, exist_ok=True)
    if not file_name:
        parts = urlparse(url)
        file_name = os.path.basename(parts.path)
    cached_file = os.path.abspath(os.path.join(model_dir, file_name))

    # Overwrite
    if os.path.exists(cached_file):
        if overwrite or os.path.getsize(cached_file) == 0:
            os.remove(cached_file)

    # Download
    if not os.path.exists(cached_file):
        logger.info(f'Downloading: "{url}" to {cached_file}\n')
        from torch.hub import download_url_to_file

        download_url_to_file(url, cached_file, progress=progress)
    else:
        logger.debug(cached_file)

    return cached_file


def friendly_name(file: str):
    if file.startswith("http"):
        file = urlparse(file).path

    file = os.path.basename(file)
    model_name, extension = os.path.splitext(file)
    return model_name, extension


def download_manager(
    url: str,
    path: str,
    extension: str = "",
    overwrite: bool = False,
    progress: bool = True,
):
    url = url.strip()

    name, ext = friendly_name(url)
    name += ext if not extension else f".{extension}"

    if url.startswith("http"):
        filename = load_file_from_url(
            url=url,
            model_dir=path,
            file_name=name,
            overwrite=overwrite,
            progress=progress,
        )
    else:
        filename = path

    return filename


class HubertModelWithFinalProj(HubertModel):
    def __init__(self, config):
        super().__init__(config)
        self.final_proj = nn.Linear(config.hidden_size, config.classifier_proj_size)


class FairseqHubertWrapper(nn.Module):
    def __init__(self, model_path_or_name="r3gm/hubert_base"):
        super().__init__()
        try:
            self.model = HubertModelWithFinalProj.from_pretrained(model_path_or_name)
        except Exception:
            self.model = HubertModel.from_pretrained(model_path_or_name)

    def extract_features(self, source, padding_mask=None, output_layer=12, **kwargs):
        param_dtype = next(self.model.parameters()).dtype
        if source.dtype != param_dtype:
            source = source.to(param_dtype)

        if source.dim() == 1:
            source = source.unsqueeze(0)

        with torch.no_grad():
            outputs = self.model(source, output_hidden_states=True)

        # 9 for v1, 12 for v2/ContentVec
        if output_layer is None or output_layer >= len(outputs.hidden_states):
            hidden_state = outputs.hidden_states[-1]
        else:
            hidden_state = outputs.hidden_states[output_layer]

        return (hidden_state, None)

    def final_proj(self, x):
        if hasattr(self.model, "final_proj"):
            return self.model.final_proj(x)
        return x


def load_hu_bert(config, hubert_path=None):
    if hubert_path and (os.path.exists(hubert_path) or os.path.isdir(hubert_path)):
        target_path = hubert_path
    else:
        target_path = "r3gm/hubert_base"

    hubert_model = FairseqHubertWrapper(target_path)
    hubert_model = hubert_model.to(config.device)

    if config.is_half and torch.device(config.device).type != "cpu":
        hubert_model = hubert_model.half()
    else:
        hubert_model = hubert_model.float()

    hubert_model.eval()
    return hubert_model


def load_trained_model(model_path, config):

    if not model_path:
        raise ValueError("No model found")

    logger.info("Loading %s" % model_path)
    cpt = torch.load(model_path, map_location="cpu")
    if not isinstance(cpt, dict) or "config" not in cpt:
        raise ValueError(
            f"The file '{model_path}' is not a valid RVC model. Missing 'config' dictionary in checkpoint."
        )
    tgt_sr = cpt["config"][-1]
    cpt["config"][-3] = cpt["weight"]["emb_g.weight"].shape[0]  # n_spk
    if_f0 = cpt.get("f0", 1)
    if if_f0 == 0:
        # protect to 0.5 need?
        pass

    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

    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 n_spk, tgt_sr, net_g, vc, cpt, version


class BaseLoader:
    def __init__(self, only_cpu=False, hubert_path=None, rmvpe_path=None, preload_models=False):
        self.model_config = {}
        self.cache_model = {}
        self.only_cpu = only_cpu
        self.hubert_path = hubert_path
        self.rmvpe_path = rmvpe_path

        self.config = Config(self.only_cpu)
        self.hu_bert_model = None
        self.model_pitch_estimator = None
        self.model_vc = {}

        if preload_models:
            logger.info("Pre-loading HuBERT and RMVPE models to device...")
            self.hu_bert_model = load_hu_bert(self.config, self.hubert_path)
            self._load_rmvpe()

    def _load_rmvpe(self):
        """Centralized method to load RMVPE."""
        if self.model_pitch_estimator is not None:
            return

        from infer_rvc_python.lib.rmvpe import RMVPE
        logger.info("Loading vocal pitch estimator model")
        
        if self.rmvpe_path is None:
            self.rmvpe_path = ""
        rm_local_path = "rmvpe.pt"
        if os.path.exists(self.rmvpe_path):
            rm_local_path = self.rmvpe_path
        else:
            rm_local_path = download_manager(
                os.path.join(BASE_DOWNLOAD_LINK, "rmvpe.pt"), BASE_DIR
            )
        self.model_pitch_estimator = RMVPE(
            rm_local_path,
            is_half=self.config.is_half,
            device=self.config.device
        )

    def apply_conf(
        self,
        tag="base_model",
        file_model="",
        pitch_algo="pm",
        pitch_lvl=0,
        file_index="",
        index_influence=0.66,
        respiration_median_filtering=3,
        envelope_ratio=0.25,
        consonant_breath_protection=0.33,
        resample_sr=0,
        file_pitch_algo="",
    ):

        if not file_model:
            raise ValueError("Model not found")

        if file_index is None:
            file_index = ""

        if file_pitch_algo is None:
            file_pitch_algo = ""

        self.model_config[tag] = {
            "file_model": file_model,
            "pitch_algo": pitch_algo,
            "pitch_lvl": pitch_lvl,  # no decimal
            "file_index": file_index,
            "index_influence": index_influence,
            "respiration_median_filtering": respiration_median_filtering,
            "envelope_ratio": envelope_ratio,
            "consonant_breath_protection": consonant_breath_protection,
            "resample_sr": resample_sr,
            "file_pitch_algo": file_pitch_algo,
        }
        return f"CONFIGURATION APPLIED FOR {tag}: {file_model}"

    def infer(
        self,
        task_id,
        params,
        # load model
        n_spk,
        tgt_sr,
        net_g,
        pipe,
        cpt,
        version,
        if_f0,
        # load index
        index_rate,
        index,
        big_npy,
        # load f0 file
        inp_f0,
        # audio file
        input_audio_path,
        overwrite,
        type_output,
    ):

        f0_method = params["pitch_algo"]
        f0_up_key = params["pitch_lvl"]
        filter_radius = params["respiration_median_filtering"]
        resample_sr = params["resample_sr"]
        rms_mix_rate = params["envelope_ratio"]
        protect = params["consonant_breath_protection"]
        base_sr = 16000

        if isinstance(input_audio_path, tuple):
            if f0_method == "harvest":
                raise ValueError("Harvest not support from array")
            audio = input_audio_path[0]
            source_sr = input_audio_path[1]
            if source_sr != base_sr:
                audio = librosa.resample(
                    audio.astype(np.float32),
                    orig_sr=source_sr,
                    target_sr=base_sr
                )
            audio = audio.astype(np.float32).flatten()
        elif not os.path.exists(input_audio_path):
            raise ValueError(
                "The audio file was not found or is not "
                f"a valid file: {input_audio_path}"
            )
        else:
            audio = load_audio(input_audio_path, base_sr)

        f0_up_key = int(f0_up_key)

        # Normalize audio
        audio_max = np.abs(audio).max() / 0.95
        if audio_max > 1:
            audio /= audio_max

        times = [0, 0, 0]

        # filters audio signal, pads it, computes sliding window sums,
        # and extracts optimized time indices
        audio = signal.filtfilt(bh, ah, audio)
        audio_pad = np.pad(
            audio, (pipe.window // 2, pipe.window // 2), mode="reflect"
        )
        opt_ts = []
        if audio_pad.shape[0] > pipe.t_max:
            audio_sum = np.zeros_like(audio)
            for i in range(pipe.window):
                audio_sum += audio_pad[i:i - pipe.window]
            for t in range(pipe.t_center, audio.shape[0], pipe.t_center):
                opt_ts.append(
                    t
                    - pipe.t_query
                    + np.where(
                        np.abs(audio_sum[t - pipe.t_query: t + pipe.t_query])
                        == np.abs(audio_sum[t - pipe.t_query: t + pipe.t_query]).min()
                    )[0][0]
                )

        s = 0
        audio_opt = []
        t = None
        t1 = ttime()

        sid_value = 0
        sid = torch.tensor(sid_value, device=pipe.device).unsqueeze(0).long()

        # Pads audio symmetrically, calculates length divided by window size.
        audio_pad = np.pad(audio, (pipe.t_pad, pipe.t_pad), mode="reflect")
        p_len = audio_pad.shape[0] // pipe.window

        # Estimates pitch from audio signal
        pitch, pitchf = None, None
        if if_f0 == 1:
            pitch, pitchf = pipe.get_f0(
                input_audio_path,
                audio_pad,
                p_len,
                f0_up_key,
                f0_method,
                filter_radius,
                inp_f0,
            )
            pitch = pitch[:p_len]
            pitchf = pitchf[:p_len]
            if pipe.device == "mps":
                pitchf = pitchf.astype(np.float32)
            pitch = torch.tensor(
                pitch, device=pipe.device
            ).unsqueeze(0).long()
            pitchf = torch.tensor(
                pitchf, device=pipe.device
            ).unsqueeze(0).float()

        t2 = ttime()
        times[1] += t2 - t1
        for t in opt_ts:
            t = t // pipe.window * pipe.window
            if if_f0 == 1:
                pitch_slice = pitch[
                    :, s // pipe.window: (t + pipe.t_pad2) // pipe.window
                ]
                pitchf_slice = pitchf[
                    :, s // pipe.window: (t + pipe.t_pad2) // pipe.window
                ]
            else:
                pitch_slice = None
                pitchf_slice = None

            audio_slice = audio_pad[s:t + pipe.t_pad2 + pipe.window]
            audio_opt.append(
                pipe.vc(
                    self.hu_bert_model,
                    net_g,
                    sid,
                    audio_slice,
                    pitch_slice,
                    pitchf_slice,
                    times,
                    index,
                    big_npy,
                    index_rate,
                    version,
                    protect,
                )[pipe.t_pad_tgt:-pipe.t_pad_tgt]
            )
            s = t

        pitch_end_slice = pitch[
            :, t // pipe.window:
        ] if t is not None else pitch
        pitchf_end_slice = pitchf[
            :, t // pipe.window:
        ] if t is not None else pitchf

        audio_opt.append(
            pipe.vc(
                self.hu_bert_model,
                net_g,
                sid,
                audio_pad[t:],
                pitch_end_slice,
                pitchf_end_slice,
                times,
                index,
                big_npy,
                index_rate,
                version,
                protect,
            )[pipe.t_pad_tgt:-pipe.t_pad_tgt]
        )

        audio_opt = np.concatenate(audio_opt)
        if rms_mix_rate != 1:
            audio_opt = change_rms(
                audio, 16000, audio_opt, tgt_sr, rms_mix_rate
            )
        if resample_sr >= 16000 and tgt_sr != resample_sr:
            audio_opt = librosa.resample(
                audio_opt, orig_sr=tgt_sr, target_sr=resample_sr
            )
        audio_max = np.abs(audio_opt).max() / 0.99
        max_int16 = 32768
        if audio_max > 1:
            max_int16 /= audio_max
        audio_opt = (audio_opt * max_int16).astype(np.int16)
        del pitch, pitchf, sid
        if torch.cuda.is_available():
            torch.cuda.empty_cache()

        if tgt_sr != resample_sr >= 16000:
            final_sr = resample_sr
        else:
            final_sr = tgt_sr

        """
        "Success.\n %s\nTime:\n npy:%ss, f0:%ss, infer:%ss" % (
            times[0],
            times[1],
            times[2],
        ), (final_sr, audio_opt)

        """

        if type_output == "array":
            return audio_opt, final_sr

        if overwrite:
            output_audio_path = input_audio_path  # Overwrite
            type_output = os.path.splitext(output_audio_path)[1].lstrip(".")
        else:
            basename = os.path.basename(input_audio_path)
            dirname = os.path.dirname(input_audio_path)

            name_, ext_ = os.path.splitext(basename)
            new_basename = f"{name_}_edited{ext_}"
            new_path = os.path.join(dirname, new_basename)

            output_audio_path = new_path

        # Save file
        if not type_output:
            type_output = "wav"
        output_audio_path = os.path.splitext(
            output_audio_path
        )[0]+f".{type_output}"

        try:
            if type_output.lower() == "wav":
                sf.write(file=output_audio_path, samplerate=final_sr, data=audio_opt)
            else:
                target_sr = min(SUPPORTED_SAMPLE_RATES, key=lambda altsr: abs(altsr - final_sr))
                if target_sr != final_sr:
                    logger.debug(f"Resampling from {final_sr} -> {target_sr} for {type_output}")
                    audio_opt = signal.resample_poly(audio_opt, target_sr, final_sr, axis=0).astype(np.int16)
                sf.write(file=output_audio_path, samplerate=target_sr, data=audio_opt, format=type_output)
        except Exception as e:
            logger.error(e)
            logger.error("Error saving file, trying with WAV format")
            output_audio_path = os.path.splitext(output_audio_path)[0]+".wav"
            sf.write(
                file=output_audio_path,
                samplerate=final_sr,
                data=audio_opt
            )

        logger.info(str(output_audio_path))

        self.model_config[task_id]["result"].append(output_audio_path)
        self.output_list.append(output_audio_path)

    def run_threads(self, threads):
        # Start threads
        for thread in threads:
            thread.start()

        # Wait for all threads to finish
        for thread in threads:
            thread.join()

        gc.collect()
        torch.cuda.empty_cache()

    def unload_models(self):
        self.hu_bert_model = None
        self.model_pitch_estimator = None
        self.model_vc = {}
        self.cache_model = {}
        gc.collect()
        torch.cuda.empty_cache()

    def __call__(
        self,
        audio_files=[],
        tag_list=[],
        overwrite=False,
        parallel_workers=1,
        type_output=None,  # ["mp3", "wav", "flac"]
        show_progress=True,
    ):
        logger.info(f"Parallel workers: {str(parallel_workers)}")

        self.output_list = []

        if not self.model_config:
            raise ValueError("No model has been configured for inference")

        if isinstance(audio_files, str):
            audio_files = [audio_files]
        if isinstance(tag_list, str):
            tag_list = [tag_list]

        if not audio_files:
            raise ValueError("No audio found to convert")
        if not tag_list:
            tag_list = [list(self.model_config.keys())[-1]] * len(audio_files)

        if len(audio_files) > len(tag_list):
            logger.info("Extend tag list to match audio files")
            extend_number = len(audio_files) - len(tag_list)
            tag_list.extend([tag_list[0]] * extend_number)

        if len(audio_files) < len(tag_list):
            logger.info("Cut list tags")
            tag_list = tag_list[:len(audio_files)]

        tag_file_pairs = list(zip(tag_list, audio_files))
        sorted_tag_file = sorted(tag_file_pairs, key=lambda x: x[0])

        # Base params
        if not self.hu_bert_model:
            self.hu_bert_model = load_hu_bert(self.config, self.hubert_path)

        cache_params = None
        threads = []
        progress_bar = tqdm(total=len(tag_list), desc="Progress", disable=not show_progress)
        for i, (id_tag, input_audio_path) in enumerate(sorted_tag_file):

            if id_tag not in self.model_config.keys():
                logger.info(
                    f"No configured model for {id_tag} with {input_audio_path}"
                )
                continue

            if (
                len(threads) >= parallel_workers
                or cache_params != id_tag
                and cache_params is not None
            ):

                self.run_threads(threads)
                progress_bar.update(len(threads))

                threads = []

            if cache_params != id_tag:

                self.model_config[id_tag]["result"] = []

                # Unload previous
                (
                    n_spk,
                    tgt_sr,
                    net_g,
                    pipe,
                    cpt,
                    version,
                    if_f0,
                    index_rate,
                    index,
                    big_npy,
                    inp_f0,
                ) = [None] * 11
                gc.collect()
                torch.cuda.empty_cache()

                # Model params
                params = self.model_config[id_tag]

                model_path = params["file_model"]
                f0_method = params["pitch_algo"]
                file_index = params["file_index"]
                index_rate = params["index_influence"]
                f0_file = params["file_pitch_algo"]

                # Load model
                (
                    n_spk,
                    tgt_sr,
                    net_g,
                    pipe,
                    cpt,
                    version
                ) = load_trained_model(model_path, self.config)
                if_f0 = cpt.get("f0", 1)  # pitch data

                # Load index
                if file_index and os.path.isfile(file_index) and index_rate != 0:
                    try:
                        index = faiss.read_index(file_index)
                        if index.ntotal > 0:
                            big_npy = index.reconstruct_n(0, index.ntotal)
                        else:
                            logger.warning(f"Index '{file_index}' has 0 vectors. Disabling index.")
                            index_rate = 0
                            index = big_npy = None
                    except Exception as error:
                        logger.error(f"Index: {str(error)}")
                        index_rate = 0
                        index = big_npy = None
                else:
                    if file_index and not os.path.isfile(file_index):
                        logger.warning("File index not found")
                    index_rate = 0
                    index = big_npy = None

                # Load f0 file
                inp_f0 = None
                if os.path.exists(f0_file):
                    try:
                        with open(f0_file, "r") as f:
                            lines = f.read().strip("\n").split("\n")
                        inp_f0 = []
                        for line in lines:
                            inp_f0.append([float(i) for i in line.split(",")])
                        inp_f0 = np.array(inp_f0, dtype="float32")
                    except Exception as error:
                        logger.error(f"f0 file: {str(error)}")

                if "rmvpe" in f0_method:
                    self._load_rmvpe()
                    pipe.model_rmvpe = self.model_pitch_estimator

                cache_params = id_tag

            # self.infer(
            #     id_tag,
            #     params,
            #     # load model
            #     n_spk,
            #     tgt_sr,
            #     net_g,
            #     pipe,
            #     cpt,
            #     version,
            #     if_f0,
            #     # load index
            #     index_rate,
            #     index,
            #     big_npy,
            #     # load f0 file
            #     inp_f0,
            #     # output file
            #     input_audio_path,
            #     overwrite,
            #     type_output,
            # )

            thread = threading.Thread(
                target=self.infer,
                args=(
                    id_tag,
                    params,
                    # loaded model
                    n_spk,
                    tgt_sr,
                    net_g,
                    pipe,
                    cpt,
                    version,
                    if_f0,
                    # loaded index
                    index_rate,
                    index,
                    big_npy,
                    # loaded f0 file
                    inp_f0,
                    # audio file
                    input_audio_path,
                    overwrite,
                    type_output,
                )
            )

            threads.append(thread)

        # Run last
        if threads:
            self.run_threads(threads)

        progress_bar.update(len(threads))
        progress_bar.close()

        final_result = []
        valid_tags = set(tag_list)
        for tag in valid_tags:
            if (
                tag in self.model_config.keys()
                and "result" in self.model_config[tag].keys()
            ):
                final_result.extend(self.model_config[tag]["result"])

        return final_result

    def generate_from_cache(
        self,
        audio_data=None,  # str or tuple (<array data>,<int sampling rate>)
        tag=None,
        reload=False,
    ):

        if not self.model_config:
            raise ValueError("No model has been configured for inference")

        if not audio_data:
            raise ValueError(
                "An audio file or tuple with "
                "(<numpy data audio>,<sampling rate>) is needed"
            )

        # Base params
        if not self.hu_bert_model:
            self.hu_bert_model = load_hu_bert(self.config, self.hubert_path)

        if tag not in self.model_config.keys():
            raise ValueError(
                f"No configured model for {tag}"
            )

        now_data = self.model_config[tag]
        now_data["tag"] = tag

        if self.cache_model != now_data and not reload:

            # Unload previous
            self.model_vc = {}
            gc.collect()
            torch.cuda.empty_cache()

            model_path = now_data["file_model"]
            f0_method = now_data["pitch_algo"]
            file_index = now_data["file_index"]
            index_rate = now_data["index_influence"]
            f0_file = now_data["file_pitch_algo"]

            # Load model
            (
                self.model_vc["n_spk"],
                self.model_vc["tgt_sr"],
                self.model_vc["net_g"],
                self.model_vc["pipe"],
                self.model_vc["cpt"],
                self.model_vc["version"]
            ) = load_trained_model(model_path, self.config)
            self.model_vc["if_f0"] = self.model_vc["cpt"].get("f0", 1)

            # Load index
            if file_index and os.path.isfile(file_index) and index_rate != 0:
                try:
                    index = faiss.read_index(file_index)
                    if index.ntotal > 0:
                        big_npy = index.reconstruct_n(0, index.ntotal)
                    else:
                        logger.warning(f"Index '{file_index}' has 0 vectors. Disabling index.")
                        index_rate = 0
                        index = big_npy = None
                except Exception as error:
                    logger.error(f"Index: {str(error)}")
                    index_rate = 0
                    index = big_npy = None
            else:
                if file_index and not os.path.isfile(file_index):
                    logger.warning("File index not found")
                index_rate = 0
                index = big_npy = None

            self.model_vc["index_rate"] = index_rate
            self.model_vc["index"] = index
            self.model_vc["big_npy"] = big_npy

            # Load f0 file
            inp_f0 = None
            if os.path.exists(f0_file):
                try:
                    with open(f0_file, "r") as f:
                        lines = f.read().strip("\n").split("\n")
                    inp_f0 = []
                    for line in lines:
                        inp_f0.append([float(i) for i in line.split(",")])
                    inp_f0 = np.array(inp_f0, dtype="float32")
                except Exception as error:
                    logger.error(f"f0 file: {str(error)}")

            self.model_vc["inp_f0"] = inp_f0

            if "rmvpe" in f0_method:
                self._load_rmvpe()
                self.model_vc["pipe"].model_rmvpe = self.model_pitch_estimator

            self.cache_model = copy.deepcopy(now_data)

        return self.infer(
            tag,
            now_data,
            # load model
            self.model_vc["n_spk"],
            self.model_vc["tgt_sr"],
            self.model_vc["net_g"],
            self.model_vc["pipe"],
            self.model_vc["cpt"],
            self.model_vc["version"],
            self.model_vc["if_f0"],
            # load index
            self.model_vc["index_rate"],
            self.model_vc["index"],
            self.model_vc["big_npy"],
            # load f0 file
            self.model_vc["inp_f0"],
            # output file
            audio_data,
            False,
            "array",
        )