import os import re import sys import time import tqdm import torch import shutil import librosa import logging import argparse import warnings import parselmouth import logging.handlers import numpy as np import soundfile as sf import torch.nn.functional as F from random import shuffle from multiprocessing import Pool from distutils.util import strtobool from fairseq import checkpoint_utils from functools import partial from concurrent.futures import ThreadPoolExecutor, as_completed sys.path.append(os.getcwd()) from main.configs.config import Config from main.library.predictors.FCPE import FCPE from main.library.predictors.RMVPE import RMVPE from main.library.predictors.WORLD import PYWORLD from main.library.predictors.CREPE import predict, mean, median from main.library.utils import check_predictors, check_embedders, load_audio logger = logging.getLogger(__name__) translations = Config().translations logger.propagate = False warnings.filterwarnings("ignore") for l in ["torch", "faiss", "httpx", "fairseq", "httpcore", "faiss.loader", "numba.core", "urllib3"]: logging.getLogger(l).setLevel(logging.ERROR) def parse_arguments(): parser = argparse.ArgumentParser() parser.add_argument("--model_name", type=str, required=True) parser.add_argument("--rvc_version", type=str, default="v2") parser.add_argument("--f0_method", type=str, default="rmvpe") parser.add_argument("--pitch_guidance", type=lambda x: bool(strtobool(x)), default=True) parser.add_argument("--hop_length", type=int, default=128) parser.add_argument("--cpu_cores", type=int, default=2) parser.add_argument("--gpu", type=str, default="-") parser.add_argument("--sample_rate", type=int, required=True) parser.add_argument("--embedder_model", type=str, default="contentvec_base") return parser.parse_args() def generate_config(rvc_version, sample_rate, model_path): config_save_path = os.path.join(model_path, "config.json") if not os.path.exists(config_save_path): shutil.copy(os.path.join("main", "configs", rvc_version, f"{sample_rate}.json"), config_save_path) def generate_filelist(pitch_guidance, model_path, rvc_version, sample_rate): gt_wavs_dir, feature_dir = os.path.join(model_path, "sliced_audios"), os.path.join(model_path, f"{rvc_version}_extracted") f0_dir, f0nsf_dir = None, None if pitch_guidance: f0_dir, f0nsf_dir = os.path.join(model_path, "f0"), os.path.join(model_path, "f0_voiced") gt_wavs_files, feature_files = set(name.split(".")[0] for name in os.listdir(gt_wavs_dir)), set(name.split(".")[0] for name in os.listdir(feature_dir)) names = gt_wavs_files & feature_files & set(name.split(".")[0] for name in os.listdir(f0_dir)) & set(name.split(".")[0] for name in os.listdir(f0nsf_dir)) if pitch_guidance else gt_wavs_files & feature_files options = [] mute_base_path = os.path.join("assets", "logs", "mute") for name in names: options.append(f"{gt_wavs_dir}/{name}.wav|{feature_dir}/{name}.npy|{f0_dir}/{name}.wav.npy|{f0nsf_dir}/{name}.wav.npy|0" if pitch_guidance else f"{gt_wavs_dir}/{name}.wav|{feature_dir}/{name}.npy|0") mute_audio_path, mute_feature_path = os.path.join(mute_base_path, "sliced_audios", f"mute{sample_rate}.wav"), os.path.join(mute_base_path, f"{rvc_version}_extracted", "mute.npy") for _ in range(2): options.append(f"{mute_audio_path}|{mute_feature_path}|{os.path.join(mute_base_path, 'f0', 'mute.wav.npy')}|{os.path.join(mute_base_path, 'f0_voiced', 'mute.wav.npy')}|0" if pitch_guidance else f"{mute_audio_path}|{mute_feature_path}|0") shuffle(options) with open(os.path.join(model_path, "filelist.txt"), "w") as f: f.write("\n".join(options)) def setup_paths(exp_dir, version = None): wav_path = os.path.join(exp_dir, "sliced_audios_16k") if version: out_path = os.path.join(exp_dir, f"{version}_extracted") os.makedirs(out_path, exist_ok=True) return wav_path, out_path else: output_root1, output_root2 = os.path.join(exp_dir, "f0"), os.path.join(exp_dir, "f0_voiced") os.makedirs(output_root1, exist_ok=True); os.makedirs(output_root2, exist_ok=True) return wav_path, output_root1, output_root2 def read_wave(wav_path, normalize = False): wav, sr = sf.read(wav_path) assert sr == 16000, translations["sr_not_16000"] feats = torch.from_numpy(wav).float() if feats.dim() == 2: feats = feats.mean(-1) feats = feats.view(1, -1) if normalize: feats = F.layer_norm(feats, feats.shape) return feats def get_device(gpu_index): if gpu_index == "cpu": return "cpu" try: index = int(gpu_index) if index < torch.cuda.device_count(): return f"cuda:{index}" else: logger.warning(translations["gpu_not_valid"]) except ValueError: logger.warning(translations["gpu_not_valid"]) return "cpu" class FeatureInput: def __init__(self, sample_rate=16000, hop_size=160, device="cpu"): self.fs = sample_rate self.hop = hop_size self.f0_bin = 256 self.f0_max = 1100.0 self.f0_min = 50.0 self.f0_mel_min = 1127 * np.log(1 + self.f0_min / 700) self.f0_mel_max = 1127 * np.log(1 + self.f0_max / 700) self.device = device def get_providers(self): import onnxruntime ort_providers = onnxruntime.get_available_providers() if "CUDAExecutionProvider" in ort_providers: providers = ["CUDAExecutionProvider"] elif "CoreMLExecutionProvider" in ort_providers: providers = ["CoreMLExecutionProvider"] else: providers = ["CPUExecutionProvider"] return providers def compute_f0_hybrid(self, methods_str, np_arr, hop_length): methods_str = re.search("hybrid\[(.+)\]", methods_str) if methods_str: methods = [method.strip() for method in methods_str.group(1).split("+")] f0_computation_stack, resampled_stack = [], [] logger.debug(translations["hybrid_methods"].format(methods=methods)) for method in methods: f0 = None if method == "pm": f0 = self.get_pm(np_arr) elif method == "dio": f0 = self.get_pyworld(np_arr, "dio") elif method == "mangio-crepe-full": f0 = self.get_mangio_crepe(np_arr, int(hop_length), "full") elif method == "mangio-crepe-full-onnx": f0 = self.get_mangio_crepe(np_arr, int(hop_length), "full", onnx=True) elif method == "mangio-crepe-large": f0 = self.get_mangio_crepe(np_arr, int(hop_length), "large") elif method == "mangio-crepe-large-onnx": f0 = self.get_mangio_crepe(np_arr, int(hop_length), "large", onnx=True) elif method == "mangio-crepe-medium": f0 = self.get_mangio_crepe(np_arr, int(hop_length), "medium") elif method == "mangio-crepe-medium-onnx": f0 = self.get_mangio_crepe(np_arr, int(hop_length), "medium", onnx=True) elif method == "mangio-crepe-small": f0 = self.get_mangio_crepe(np_arr, int(hop_length), "small") elif method == "mangio-crepe-small-onnx": f0 = self.get_mangio_crepe(np_arr, int(hop_length), "small", onnx=True) elif method == "mangio-crepe-tiny": f0 = self.get_mangio_crepe(np_arr, int(hop_length), "tiny") elif method == "mangio-crepe-tiny-onnx": f0 = self.get_mangio_crepe(np_arr, int(hop_length), "tiny", onnx=True) elif method == "crepe-full": f0 = self.get_crepe(np_arr, "full") elif method == "crepe-full-onnx": f0 = self.get_crepe(np_arr, "full", onnx=True) elif method == "crepe-large": f0 = self.get_crepe(np_arr, "large") elif method == "crepe-large-onnx": f0 = self.get_crepe(np_arr, "large", onnx=True) elif method == "crepe-medium": f0 = self.get_crepe(np_arr, "medium") elif method == "crepe-medium-onnx": f0 = self.get_crepe(np_arr, "medium", onnx=True) elif method == "crepe-small": f0 = self.get_crepe(np_arr, "small") elif method == "crepe-small-onnx": f0 = self.get_crepe(np_arr, "small", onnx=True) elif method == "crepe-tiny": f0 = self.get_crepe(np_arr, "tiny") elif method == "crepe-tiny-onnx": f0 = self.get_crepe(np_arr, "tiny", onnx=True) elif method == "fcpe": f0 = self.get_fcpe(np_arr, int(hop_length)) elif method == "fcpe-onnx": f0 = self.get_fcpe(np_arr, int(hop_length), onnx=True) elif method == "fcpe-legacy": f0 = self.get_fcpe(np_arr, int(hop_length), legacy=True) elif method == "fcpe-legacy-onnx": f0 = self.get_fcpe(np_arr, int(hop_length), onnx=True, legacy=True) elif method == "rmvpe": f0 = self.get_rmvpe(np_arr) elif method == "rmvpe-onnx": f0 = self.get_rmvpe(np_arr, onnx=True) elif method == "rmvpe-legacy": f0 = self.get_rmvpe(np_arr, legacy=True) elif method == "rmvpe-legacy-onnx": f0 = self.get_rmvpe(np_arr, legacy=True, onnx=True) elif method == "harvest": f0 = self.get_pyworld(np_arr, "harvest") elif method == "yin": f0 = self.get_yin(np_arr, int(hop_length)) elif method == "pyin": return self.get_pyin(np_arr, int(hop_length)) else: raise ValueError(translations["method_not_valid"]) f0_computation_stack.append(f0) for f0 in f0_computation_stack: resampled_stack.append(np.interp(np.linspace(0, len(f0), (np_arr.size // self.hop)), np.arange(len(f0)), f0)) return resampled_stack[0] if len(resampled_stack) == 1 else np.nanmedian(np.vstack(resampled_stack), axis=0) def compute_f0(self, np_arr, f0_method, hop_length): if f0_method == "pm": return self.get_pm(np_arr) elif f0_method == "dio": return self.get_pyworld(np_arr, "dio") elif f0_method == "mangio-crepe-full": return self.get_mangio_crepe(np_arr, int(hop_length), "full") elif f0_method == "mangio-crepe-full-onnx": return self.get_mangio_crepe(np_arr, int(hop_length), "full", onnx=True) elif f0_method == "mangio-crepe-large": return self.get_mangio_crepe(np_arr, int(hop_length), "large") elif f0_method == "mangio-crepe-large-onnx": return self.get_mangio_crepe(np_arr, int(hop_length), "large", onnx=True) elif f0_method == "mangio-crepe-medium": return self.get_mangio_crepe(np_arr, int(hop_length), "medium") elif f0_method == "mangio-crepe-medium-onnx": return self.get_mangio_crepe(np_arr, int(hop_length), "medium", onnx=True) elif f0_method == "mangio-crepe-small": return self.get_mangio_crepe(np_arr, int(hop_length), "small") elif f0_method == "mangio-crepe-small-onnx": return self.get_mangio_crepe(np_arr, int(hop_length), "small", onnx=True) elif f0_method == "mangio-crepe-tiny": return self.get_mangio_crepe(np_arr, int(hop_length), "tiny") elif f0_method == "mangio-crepe-tiny-onnx": return self.get_mangio_crepe(np_arr, int(hop_length), "tiny", onnx=True) elif f0_method == "crepe-full": return self.get_crepe(np_arr, "full") elif f0_method == "crepe-full-onnx": return self.get_crepe(np_arr, "full", onnx=True) elif f0_method == "crepe-large": return self.get_crepe(np_arr, "large") elif f0_method == "crepe-large-onnx": return self.get_crepe(np_arr, "large", onnx=True) elif f0_method == "crepe-medium": return self.get_crepe(np_arr, "medium") elif f0_method == "crepe-medium-onnx": return self.get_crepe(np_arr, "medium", onnx=True) elif f0_method == "crepe-small": return self.get_crepe(np_arr, "small") elif f0_method == "crepe-small-onnx": return self.get_crepe(np_arr, "small", onnx=True) elif f0_method == "crepe-tiny": return self.get_crepe(np_arr, "tiny") elif f0_method == "crepe-tiny-onnx": return self.get_crepe(np_arr, "tiny", onnx=True) elif f0_method == "fcpe": return self.get_fcpe(np_arr, int(hop_length)) elif f0_method == "fcpe-onnx": return self.get_fcpe(np_arr, int(hop_length), onnx=True) elif f0_method == "fcpe-legacy": return self.get_fcpe(np_arr, int(hop_length), legacy=True) elif f0_method == "fcpe-legacy-onnx": return self.get_fcpe(np_arr, int(hop_length), onnx=True, legacy=True) elif f0_method == "rmvpe": return self.get_rmvpe(np_arr) elif f0_method == "rmvpe-onnx": return self.get_rmvpe(np_arr, onnx=True) elif f0_method == "rmvpe-legacy": return self.get_rmvpe(np_arr, legacy=True) elif f0_method == "rmvpe-legacy-onnx": return self.get_rmvpe(np_arr, legacy=True, onnx=True) elif f0_method == "harvest": return self.get_pyworld(np_arr, "harvest") elif f0_method == "yin": return self.get_yin(np_arr, int(hop_length)) elif f0_method == "pyin": return self.get_pyin(np_arr, int(hop_length)) elif "hybrid" in f0_method: return self.compute_f0_hybrid(f0_method, np_arr, int(hop_length)) else: raise ValueError(translations["method_not_valid"]) def get_pm(self, x): f0 = (parselmouth.Sound(x, self.fs).to_pitch_ac(time_step=(160 / 16000 * 1000) / 1000, voicing_threshold=0.6, pitch_floor=50, pitch_ceiling=1100).selected_array["frequency"]) pad_size = ((x.size // self.hop) - len(f0) + 1) // 2 if pad_size > 0 or (x.size // self.hop) - len(f0) - pad_size > 0: f0 = np.pad(f0, [[pad_size, (x.size // self.hop) - len(f0) - pad_size]], mode="constant") return f0 def get_mangio_crepe(self, x, hop_length, model="full", onnx=False): providers = self.get_providers() if onnx else None audio = torch.from_numpy(x.astype(np.float32)).to(self.device) audio /= torch.quantile(torch.abs(audio), 0.999) audio = audio.unsqueeze(0) source = predict(audio, self.fs, hop_length, self.f0_min, self.f0_max, model=model, batch_size=hop_length * 2, device=self.device, pad=True, providers=providers, onnx=onnx).squeeze(0).cpu().float().numpy() source[source < 0.001] = np.nan return np.nan_to_num(np.interp(np.arange(0, len(source) * (x.size // self.hop), len(source)) / (x.size // self.hop), np.arange(0, len(source)), source)) def get_crepe(self, x, model="full", onnx=False): providers = self.get_providers() if onnx else None f0, pd = predict(torch.tensor(np.copy(x))[None].float(), self.fs, 160, self.f0_min, self.f0_max, model, batch_size=512, device=self.device, return_periodicity=True, providers=providers, onnx=onnx) f0, pd = mean(f0, 3), median(pd, 3) f0[pd < 0.1] = 0 return f0[0].cpu().numpy() def get_fcpe(self, x, hop_length, legacy=False, onnx=False): providers = self.get_providers() if onnx else None model_fcpe = FCPE(os.path.join("assets", "models", "predictors", "fcpe" + (".onnx" if onnx else ".pt")), hop_length=int(hop_length), f0_min=int(self.f0_min), f0_max=int(self.f0_max), dtype=torch.float32, device=self.device, sample_rate=self.fs, threshold=0.03, providers=providers, onnx=onnx) if legacy else FCPE(os.path.join("assets", "models", "predictors", "fcpe" + (".onnx" if onnx else ".pt")), hop_length=160, f0_min=0, f0_max=8000, dtype=torch.float32, device=self.device, sample_rate=self.fs, threshold=0.006, providers=providers, onnx=onnx) f0 = model_fcpe.compute_f0(x, p_len=(x.size // self.hop)) del model_fcpe return f0 def get_rmvpe(self, x, legacy=False, onnx=False): providers = self.get_providers() if onnx else None rmvpe_model = RMVPE(os.path.join("assets", "models", "predictors", "rmvpe" + (".onnx" if onnx else ".pt")), device=self.device, onnx=onnx, providers=providers) f0 = rmvpe_model.infer_from_audio_with_pitch(x, thred=0.03, f0_min=self.f0_min, f0_max=self.f0_max) if legacy else rmvpe_model.infer_from_audio(x, thred=0.03) del rmvpe_model return f0 def get_pyworld(self, x, model="harvest"): pw = PYWORLD() if model == "harvest": f0, t = pw.harvest(x.astype(np.double), fs=self.fs, f0_ceil=self.f0_max, f0_floor=self.f0_min, frame_period=1000 * self.hop / self.fs) elif model == "dio": f0, t = pw.dio(x.astype(np.double), fs=self.fs, f0_ceil=self.f0_max, f0_floor=self.f0_min, frame_period=1000 * self.hop / self.fs) else: raise ValueError(translations["method_not_valid"]) return pw.stonemask(x.astype(np.double), self.fs, t, f0) def get_yin(self, x, hop_length): source = np.array(librosa.yin(x.astype(np.double), sr=self.fs, fmin=self.f0_min, fmax=self.f0_max, hop_length=hop_length)) source[source < 0.001] = np.nan return np.nan_to_num(np.interp(np.arange(0, len(source) * (x.size // self.hop), len(source)) / (x.size // self.hop), np.arange(0, len(source)), source)) def get_pyin(self, x, hop_length): f0, _, _ = librosa.pyin(x.astype(np.double), fmin=self.f0_min, fmax=self.f0_max, sr=self.fs, hop_length=hop_length) source = np.array(f0) source[source < 0.001] = np.nan return np.nan_to_num(np.interp(np.arange(0, len(source) * (x.size // self.hop), len(source)) / (x.size // self.hop), np.arange(0, len(source)), source)) def coarse_f0(self, f0): return np.rint(np.clip(((1127 * np.log(1 + f0 / 700)) - self.f0_mel_min) * (self.f0_bin - 2) / (self.f0_mel_max - self.f0_mel_min) + 1, 1, self.f0_bin - 1)).astype(int) def process_file(self, file_info, f0_method, hop_length): inp_path, opt_path1, opt_path2, np_arr = file_info if os.path.exists(opt_path1 + ".npy") and os.path.exists(opt_path2 + ".npy"): return try: feature_pit = self.compute_f0(np_arr, f0_method, hop_length) np.save(opt_path2, feature_pit, allow_pickle=False) np.save(opt_path1, self.coarse_f0(feature_pit), allow_pickle=False) except Exception as e: raise RuntimeError(f"{translations['extract_file_error']} {inp_path}: {e}") def process_files(self, files, f0_method, hop_length, pbar): for file_info in files: self.process_file(file_info, f0_method, hop_length) pbar.update() def run_pitch_extraction(exp_dir, f0_method, hop_length, num_processes, gpus): input_root, *output_roots = setup_paths(exp_dir) output_root1, output_root2 = output_roots if len(output_roots) == 2 else (output_roots[0], None) paths = [(os.path.join(input_root, name), os.path.join(output_root1, name) if output_root1 else None, os.path.join(output_root2, name) if output_root2 else None, load_audio(os.path.join(input_root, name))) for name in sorted(os.listdir(input_root)) if "spec" not in name] logger.info(translations["extract_f0_method"].format(num_processes=num_processes, f0_method=f0_method)) start_time = time.time() if gpus != "-": gpus = gpus.split("-") process_partials = [] pbar = tqdm.tqdm(total=len(paths), desc=translations["extract_f0"], ncols=100, unit="p") for idx, gpu in enumerate(gpus): feature_input = FeatureInput(device=get_device(gpu)) process_partials.append((feature_input, paths[idx::len(gpus)])) with ThreadPoolExecutor() as executor: for future in as_completed([executor.submit(FeatureInput.process_files, feature_input, part_paths, f0_method, hop_length, pbar) for feature_input, part_paths in process_partials]): pbar.update(1) logger.debug(pbar.format_meter(pbar.n, pbar.total, pbar.format_dict["elapsed"])) future.result() pbar.close() else: with tqdm.tqdm(total=len(paths), desc=translations["extract_f0"], ncols=100, unit="p") as pbar: with Pool(processes=num_processes) as pool: for _ in pool.imap_unordered(partial(FeatureInput(device="cpu").process_file, f0_method=f0_method, hop_length=hop_length), paths): pbar.update(1) logger.debug(pbar.format_meter(pbar.n, pbar.total, pbar.format_dict["elapsed"])) elapsed_time = time.time() - start_time logger.info(translations["extract_f0_success"].format(elapsed_time=f"{elapsed_time:.2f}")) def process_file_embedding(file, wav_path, out_path, model, device, version, saved_cfg): out_file_path = os.path.join(out_path, file.replace("wav", "npy")) if os.path.exists(out_file_path): return feats = read_wave(os.path.join(wav_path, file), normalize=saved_cfg.task.normalize).to(device).float() inputs = {"source": feats, "padding_mask": torch.BoolTensor(feats.shape).fill_(False).to(device), "output_layer": 9 if version == "v1" else 12} with torch.no_grad(): model = model.to(device).float().eval() logits = model.extract_features(**inputs) feats = model.final_proj(logits[0]) if version == "v1" else logits[0] feats = feats.squeeze(0).float().cpu().numpy() if not np.isnan(feats).any(): np.save(out_file_path, feats, allow_pickle=False) else: logger.warning(f"{file} {translations['NaN']}") def run_embedding_extraction(exp_dir, version, gpus, embedder_model): wav_path, out_path = setup_paths(exp_dir, version) logger.info(translations["start_extract_hubert"]) start_time = time.time() try: models, saved_cfg, _ = checkpoint_utils.load_model_ensemble_and_task([os.path.join("assets", "models", "embedders", embedder_model + '.pt')], suffix="") except Exception as e: raise ImportError(translations["read_model_error"].format(e=e)) devices = [get_device(gpu) for gpu in (gpus.split("-") if gpus != "-" else ["cpu"])] paths = sorted([file for file in os.listdir(wav_path) if file.endswith(".wav")]) if not paths: logger.warning(translations["not_found_audio_file"]) sys.exit(1) pbar = tqdm.tqdm(total=len(paths) * len(devices), desc=translations["extract_hubert"], ncols=100, unit="p") for task in [(file, wav_path, out_path, models[0], device, version, saved_cfg) for file in paths for device in devices]: try: process_file_embedding(*task) except Exception as e: raise RuntimeError(f"{translations['process_error']} {task[0]}: {e}") pbar.update(1) logger.debug(pbar.format_meter(pbar.n, pbar.total, pbar.format_dict["elapsed"])) pbar.close() elapsed_time = time.time() - start_time logger.info(translations["extract_hubert_success"].format(elapsed_time=f"{elapsed_time:.2f}")) if __name__ == "__main__": args = parse_arguments() exp_dir = os.path.join("assets", "logs", args.model_name) f0_method, hop_length, num_processes, gpus, version, pitch_guidance, sample_rate, embedder_model = args.f0_method, args.hop_length, args.cpu_cores, args.gpu, args.rvc_version, args.pitch_guidance, args.sample_rate, args.embedder_model check_predictors(f0_method) check_embedders(embedder_model) if logger.hasHandlers(): logger.handlers.clear() else: console_handler = logging.StreamHandler() console_formatter = logging.Formatter(fmt="\n%(asctime)s.%(msecs)03d | %(levelname)s | %(module)s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S") console_handler.setFormatter(console_formatter) console_handler.setLevel(logging.INFO) file_handler = logging.handlers.RotatingFileHandler(os.path.join(exp_dir, "extract.log"), maxBytes=5*1024*1024, backupCount=3, encoding='utf-8') file_formatter = logging.Formatter(fmt="\n%(asctime)s.%(msecs)03d | %(levelname)s | %(module)s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S") file_handler.setFormatter(file_formatter) file_handler.setLevel(logging.DEBUG) logger.addHandler(console_handler) logger.addHandler(file_handler) logger.setLevel(logging.DEBUG) log_data = {translations['modelname']: args.model_name, translations['export_process']: exp_dir, translations['f0_method']: f0_method, translations['pretrain_sr']: sample_rate, translations['cpu_core']: num_processes, "Gpu": gpus, "Hop length": hop_length, translations['training_version']: version, translations['extract_f0']: pitch_guidance, translations['hubert_model']: embedder_model} for key, value in log_data.items(): logger.debug(f"{key}: {value}") pid_path = os.path.join(exp_dir, "extract_pid.txt") with open(pid_path, "w") as pid_file: pid_file.write(str(os.getpid())) try: run_pitch_extraction(exp_dir, f0_method, hop_length, num_processes, gpus) run_embedding_extraction(exp_dir, version, gpus, embedder_model) generate_config(version, sample_rate, exp_dir) generate_filelist(pitch_guidance, exp_dir, version, sample_rate) except Exception as e: logger.error(f"{translations['extract_error']}: {e}") import traceback logger.debug(traceback.format_exc()) if os.path.exists(pid_path): os.remove(pid_path) logger.info(f"{translations['extract_success']} {args.model_name}.")