| import os |
| import sys |
| import subprocess |
| import traceback |
| import torch |
|
|
| |
| try: |
| import rvc_python |
| except ImportError: |
| print("rvc-python is not installed or import failed. Performing on-the-fly installation...") |
| try: |
| |
| try: |
| print("Downgrading pip to version 24.0...") |
| subprocess.run([sys.executable, "-m", "pip", "install", "pip==24.0"], check=True) |
| except subprocess.CalledProcessError: |
| print("Standard pip downgrade failed (likely due to permissions). Trying with --user...") |
| subprocess.run([sys.executable, "-m", "pip", "install", "--user", "pip==24.0"], check=True) |
| |
| |
| try: |
| print("Installing rvc-python...") |
| subprocess.run([sys.executable, "-m", "pip", "install", "rvc-python"], check=True) |
| except subprocess.CalledProcessError: |
| print("Standard rvc-python installation failed (likely due to permissions). Trying with --user...") |
| subprocess.run([sys.executable, "-m", "pip", "install", "--user", "rvc-python"], check=True) |
| |
| |
| import rvc_python |
| print("rvc-python successfully installed and imported!") |
| except Exception as e: |
| print(f"Error during on-the-fly rvc-python installation: {e}") |
|
|
| |
| |
| original_torch_load = torch.load |
| def patched_torch_load(*args, **kwargs): |
| if 'weights_only' not in kwargs: |
| kwargs['weights_only'] = False |
| return original_torch_load(*args, **kwargs) |
| torch.load = patched_torch_load |
|
|
| |
| if hasattr(torch, "backends") and hasattr(torch.backends, "mps"): |
| torch.backends.mps.is_available = lambda: False |
|
|
| class RVCEngine: |
| def __init__(self, model_dir="models/rvc"): |
| self.model_dir = model_dir |
| os.makedirs(self.model_dir, exist_ok=True) |
| try: |
| import rvc_python |
| self.is_available = True |
| except ImportError: |
| print("Warning: rvc-python is not installed or import failed. RVC engine will be disabled.") |
| traceback.print_exc() |
| self.is_available = False |
|
|
| def get_available_models(self): |
| """Returns a list of available .pth model files in the model directory.""" |
| if not os.path.exists(self.model_dir): |
| return [] |
| |
| models = [] |
| for file in os.listdir(self.model_dir): |
| if file.endswith(".pth"): |
| models.append(file) |
| return models |
|
|
| def convert(self, input_path: str, output_path: str, model_name: str, pitch: int = 0, in_process: bool = False): |
| """ |
| Convert the voice of an audio file using RVC. |
| If in_process is True, does the conversion in the current process (required for HF ZeroGPU). |
| Otherwise, spawns an isolated subprocess. |
| """ |
| if not self.is_available: |
| raise RuntimeError("RVC is not available because rvc-python is not installed or failed to import.") |
|
|
| model_path = os.path.join(self.model_dir, model_name) |
| if not os.path.exists(model_path): |
| raise FileNotFoundError(f"Model file not found: {model_path}. Please place your RVC models in {self.model_dir}.") |
|
|
| if in_process: |
| print(f"Running in-process RVC conversion for model {model_name}...") |
| base_name = os.path.splitext(model_name)[0] |
| index_path = "" |
| potential_index = os.path.join(self.model_dir, f"{base_name}.index") |
| if os.path.exists(potential_index): |
| index_path = potential_index |
| else: |
| try: |
| index_files = [f for f in os.listdir(self.model_dir) if f.endswith(".index")] |
| if len(index_files) == 1: |
| index_path = os.path.join(self.model_dir, index_files[0]) |
| elif len(index_files) > 1: |
| for idx_file in index_files: |
| idx_base = os.path.splitext(idx_file)[0] |
| if idx_base in base_name or base_name in idx_base: |
| index_path = os.path.join(self.model_dir, idx_file) |
| break |
| except Exception: |
| pass |
|
|
| from rvc_python.infer import RVCInference |
| rvc = RVCInference(models_dir=self.model_dir, device="cuda:0") |
| print(f"Loading RVC model: {model_path} (Index: {index_path})...") |
| rvc.load_model(model_path, index_path=index_path) |
| |
| print(f"Setting pitch shift to {pitch}...") |
| rvc.set_params(f0up_key=pitch, f0method="harvest") |
| |
| print("Running in-process voice conversion...") |
| rvc.infer_file(input_path, output_path) |
| print("In-process voice conversion finished successfully!") |
| return output_path |
|
|
| import subprocess |
| import sys |
| |
| |
| base_dir = os.path.dirname(os.path.abspath(__file__)) |
| python_exe = os.path.join(base_dir, ".venv", "bin", "python") |
| if not os.path.exists(python_exe): |
| python_exe = sys.executable |
| |
| script_path = os.path.join(base_dir, "run_rvc.py") |
| |
| cmd = [ |
| python_exe, |
| script_path, |
| "--input", input_path, |
| "--output", output_path, |
| "--model", model_name, |
| "--pitch", str(pitch) |
| ] |
| |
| print(f"Spawning isolated RVC subprocess: {' '.join(cmd)}") |
| result = subprocess.run(cmd, capture_output=True, text=True) |
| |
| if result.returncode != 0: |
| print(f"RVC subprocess stdout:\n{result.stdout}") |
| print(f"RVC subprocess stderr:\n{result.stderr}") |
| |
| err_msg = result.stderr.strip() |
| if "invalid load key" in err_msg or "pickle" in err_msg or "EOFError" in err_msg or "unserialize" in err_msg: |
| try: |
| with open(model_path, "rb") as f: |
| header = f.read(100) |
| header_info = f" (First 100 bytes of uploaded file: {header!r})" |
| except Exception as read_err: |
| header_info = f" (Could not read file header: {read_err})" |
| raise RuntimeError( |
| f"RVC conversion failed: The uploaded model file is corrupted, empty, or not a valid PyTorch RVC weights file. " |
| f"Ensure you are uploading a valid, non-empty .pth file.{header_info} (Details: {err_msg})" |
| ) |
| raise RuntimeError(f"RVC conversion failed in subprocess (exit code {result.returncode}): {err_msg}") |
| |
| print("RVC subprocess completed successfully.") |
| return output_path |
|
|
