""" Minimal RVC Reference Implementation Wrapper Calls the original reference code directly with minimal modifications """ import os import sys import importlib import types import weakref import numpy as np # CRITICAL FIX for Python 3.13 + numba + librosa compatibility # 🔬 NUMBA WORKAROUND: Commented out - testing if still needed with numba 0.61.2+ and librosa 0.11.0+ # Only disable numba JIT on Python 3.13+ where it causes compatibility issues # if sys.version_info >= (3, 13): # os.environ['NUMBA_DISABLE_JIT'] = '1' # print("🔧 RVC: Disabled numba JIT for Python 3.13+ compatibility") # Additional librosa compatibility monkey-patching (keeping this active since it's useful) def _normalize(S, norm=np.inf, axis=-1, threshold=None, fill=None): S = np.asarray(S) if norm is None: return S if norm == np.inf: length = np.max(np.abs(S), axis=axis, keepdims=True) elif norm == -np.inf: length = np.min(np.abs(S), axis=axis, keepdims=True) else: length = np.sum(np.abs(S) ** norm, axis=axis, keepdims=True) ** (1.0 / norm) if threshold is not None: length = np.maximum(length, threshold) if fill is not None: length = np.where(length == 0, fill, length) else: length = np.where(length == 0, 1.0, length) return S / length def _pad_center(data, size, axis=-1, **kwargs): data = np.asarray(data) n = data.shape[axis] lpad = int((size - n) // 2) rpad = int(size - n - lpad) pad_widths = [(0, 0)] * data.ndim pad_widths[axis] = (lpad, rpad) return np.pad( data, pad_widths, mode=kwargs.get("mode", "constant"), constant_values=kwargs.get("constant_values", 0), ) def _tiny(x): x = np.asarray(x) return np.finfo(x.dtype if np.issubdtype(x.dtype, np.floating) else np.float32).tiny def _fill_off_diagonal(x, radius, value=0): x = np.asarray(x) if x.ndim == 1: return x.copy() result = x.copy() n = min(result.shape) for i in range(n): for j in range(n): if abs(i - j) <= radius and i != j: result[i, j] = value return result def _is_positive_int(x): try: return isinstance(x, int) and x > 0 except (TypeError, ValueError): return False def _expand_to(x, *, ndim=None, axes=None): x = np.asarray(x) if ndim is not None: while x.ndim < ndim: x = np.expand_dims(x, axis=-1) if axes is not None: axes_list = [axes] if isinstance(axes, int) else axes for axis in sorted(axes_list): if axis >= x.ndim: x = np.expand_dims(x, axis=axis) return x def _fix_length(data, *, size, axis=-1, **kwargs): data = np.asarray(data) n = data.shape[axis] if n > size: slices = [slice(None)] * data.ndim slices[axis] = slice(0, size) return data[tuple(slices)] if n < size: pad_widths = [(0, 0)] * data.ndim pad_widths[axis] = (0, size - n) return np.pad(data, pad_widths, mode=kwargs.get("mode", "constant")) return data def _stack(arrays, axis=0): return np.stack(arrays, axis=axis) def _deprecated(*args, **kwargs): def decorator(func): return func if args and callable(args[0]) and len(args) == 1 and not kwargs: return args[0] return decorator _LIBROSA_UTIL_FALLBACKS = { "normalize": _normalize, "pad_center": _pad_center, "tiny": _tiny, "fill_off_diagonal": _fill_off_diagonal, "is_positive_int": _is_positive_int, "expand_to": _expand_to, "fix_length": _fix_length, "stack": _stack, } def _build_librosa_decorators_fallback_module(): decorators_module = types.ModuleType("librosa.util.decorators") decorators_module.deprecated = _deprecated return decorators_module def _build_librosa_util_fallback_module(): util_module = types.ModuleType("librosa.util") util_module.__path__ = [] util_module.__dict__.update(_LIBROSA_UTIL_FALLBACKS) util_module.decorators = _build_librosa_decorators_fallback_module() return util_module def _ensure_librosa_util_helpers(util_module): for name, fallback in _LIBROSA_UTIL_FALLBACKS.items(): if not hasattr(util_module, name): setattr(util_module, name, fallback) def apply_librosa_compatibility_patches(): """Patch librosa.util only when the real module is missing or broken.""" try: import librosa except Exception as e: print(f"⚠️ RVC: Could not import librosa for compatibility patches: {e}") return False try: util_module = importlib.import_module("librosa.util") _ensure_librosa_util_helpers(util_module) librosa.util = util_module return True except Exception as e: util_module = _build_librosa_util_fallback_module() sys.modules["librosa.util.decorators"] = util_module.decorators sys.modules["librosa.util"] = util_module librosa.util = util_module print(f"🔧 RVC: Using bundled librosa.util fallback ({e})") return True import torch from typing import Tuple, Optional class MinimalRVCWrapper: """ Minimal wrapper that directly calls the working reference implementation Uses direct imports from the reference directory without copying code """ def __init__(self): self.hubert_model = None self.reference_path = None self._model_cache = {} # Cache loaded RVC models to prevent VRAM spikes self._hubert_cache = {} # Cache Hubert models separately self._setup_reference_path() def clear_model_cache(self): """Clear cached models to free VRAM""" import torch import gc freed_count = len(self._model_cache) + len(self._hubert_cache) # Remove from ComfyUI's model list before clearing cache try: import comfy.model_management as model_management if hasattr(model_management, 'current_loaded_models'): # Remove RVC models from ComfyUI's tracking models_to_remove = [] for wrapper in model_management.current_loaded_models: if hasattr(wrapper, 'model_info') and wrapper.model_info.engine == "rvc": models_to_remove.append(wrapper) for wrapper in models_to_remove: model_management.current_loaded_models.remove(wrapper) if models_to_remove: print(f"🗑️ Removed {len(models_to_remove)} RVC models from ComfyUI tracking") except Exception as e: print(f"⚠️ Could not remove from ComfyUI tracking: {e}") self._model_cache.clear() self._hubert_cache.clear() if torch.cuda.is_available(): torch.cuda.empty_cache() gc.collect() print(f"🗑️ RVC minimal wrapper: Cleared {freed_count} cached models") def _register_rvc_model_with_comfyui(self, model_data, model_path): """Register RVC model with ComfyUI model management""" try: import comfy.model_management as model_management from utils.models.comfyui_model_wrapper.base_wrapper import ComfyUIModelWrapper, ModelInfo from engines.rvc.rvc_engine import RVCModelWrapper # Just register the model with ComfyUI - let ComfyUI's free_memory() handle unloading when needed # (follows the same pattern as other TTS engines like ChatterBox, F5-TTS) # Estimate model size for VRAM tracking model_size = 0 if 'net_g' in model_data and hasattr(model_data['net_g'], 'parameters'): for param in model_data['net_g'].parameters(): model_size += param.numel() * param.element_size() # Wrap the model_data dict with RVCModelWrapper so it supports weakref and .to() from utils.device import resolve_torch_device device = resolve_torch_device("auto") rvc_wrapped = RVCModelWrapper(model_data, device) # Create ModelInfo for the RVC model model_info = ModelInfo( model=rvc_wrapped, model_type="rvc_voice", engine="rvc", device=device, memory_size=model_size, load_device=device ) # Wrap for ComfyUI model management wrapper = ComfyUIModelWrapper(rvc_wrapped, model_info) # Register with ComfyUI's model list - this makes Clear VRAM work if hasattr(model_management, 'current_loaded_models'): if hasattr(model_management, 'LoadedModel'): loaded_model = model_management.LoadedModel(wrapper) loaded_model.real_model = weakref.ref(rvc_wrapped) loaded_model._tts_wrapper_ref = wrapper # prevent GC loaded_model.model_finalizer = weakref.finalize(wrapper, lambda: None) model_management.current_loaded_models.insert(0, loaded_model) else: model_management.current_loaded_models.append(wrapper) print(f"✅ Registered new RVC model with ComfyUI model management") except Exception as e: print(f"⚠️ Could not register RVC with ComfyUI management: {e}") def _register_hubert_model_with_comfyui(self, hubert_model, hubert_path): """Register Hubert model with ComfyUI model management""" try: import comfy.model_management as model_management from utils.models.comfyui_model_wrapper.base_wrapper import ComfyUIModelWrapper, ModelInfo, SimpleModelWrapper # Just register the model with ComfyUI - let ComfyUI's free_memory() handle unloading when needed # (follows the same pattern as other TTS engines like ChatterBox, F5-TTS) # Estimate model size for VRAM tracking model_size = 0 if hasattr(hubert_model, 'parameters'): for param in hubert_model.parameters(): model_size += param.numel() * param.element_size() # Wrap Hubert with SimpleModelWrapper so it has .model attribute for ComfyUI logging hubert_wrapped = SimpleModelWrapper(hubert_model) # Create ModelInfo for the Hubert model from utils.device import resolve_torch_device hubert_device = resolve_torch_device("auto") model_info = ModelInfo( model=hubert_wrapped, model_type="hubert", engine="rvc", device=hubert_device, memory_size=model_size, load_device=hubert_device ) # Wrap the model so ComfyUI can manage it wrapper = ComfyUIModelWrapper(hubert_wrapped, model_info) # Register with ComfyUI's model list - this makes Clear VRAM work if hasattr(model_management, 'current_loaded_models'): if hasattr(model_management, 'LoadedModel'): loaded_model = model_management.LoadedModel(wrapper) loaded_model.real_model = weakref.ref(hubert_wrapped) loaded_model._tts_wrapper_ref = wrapper # prevent GC loaded_model.model_finalizer = weakref.finalize(wrapper, lambda: None) model_management.current_loaded_models.insert(0, loaded_model) else: model_management.current_loaded_models.append(wrapper) print(f"✅ Registered new Hubert model with ComfyUI model management") except Exception as e: print(f"⚠️ Could not register Hubert with ComfyUI management: {e}") def _setup_reference_path(self): """Setup path to implementation (moved from docs to proper engine location)""" current_dir = os.path.dirname(__file__) self.reference_path = os.path.join(current_dir, "impl") self.lib_path = os.path.join(self.reference_path, "lib") # Add all necessary paths for reference implementation self.infer_pack_path = os.path.join(self.lib_path, "infer_pack") self.text_path = os.path.join(self.infer_pack_path, "text") # Add paths in order of priority if self.text_path not in sys.path: sys.path.insert(0, self.text_path) # For symbols if self.infer_pack_path not in sys.path: sys.path.insert(0, self.infer_pack_path) # For modules, attentions, commons if self.lib_path not in sys.path: sys.path.insert(0, self.lib_path) # For infer_pack, utils, etc. if self.reference_path not in sys.path: sys.path.insert(0, self.reference_path) # For config, vc_infer_pipeline, etc. @staticmethod def _get_rvc_feature_dim(model_data) -> int: """Read the HuBERT feature width required by a loaded RVC checkpoint.""" cpt = model_data.get("cpt", {}) if isinstance(model_data, dict) else {} input_weight = cpt.get("weight", {}).get("enc_p.emb_phone.weight") if input_weight is not None and getattr(input_weight, "ndim", 0) == 2: return int(input_weight.shape[1]) return int(cpt.get("feature_dim", 768) or 768) def _resolve_auto_hubert(self, model_data) -> Optional[str]: """Select/download the encoder matching the loaded RVC voice model.""" feature_dim = self._get_rvc_feature_dim(model_data) if feature_dim not in (768, 1024): raise ValueError( f"Unsupported RVC HuBERT feature dimension: {feature_dim}. " "Expected a 768- or 1024-dimensional checkpoint." ) model_key = "hubert-large" if feature_dim == 1024 else "content-vec-best" from engines.rvc.hubert_downloader import ensure_hubert_model hubert_path = ensure_hubert_model(model_key) if hubert_path: print( f"✅ Auto-selected HuBERT for {feature_dim}-dimensional RVC model: " f"{os.path.basename(hubert_path)}" ) return hubert_path def convert_voice(self, audio: np.ndarray, sample_rate: int, model_path: str, index_path: Optional[str] = None, f0_up_key: int = 0, f0_method: str = "rmvpe", index_rate: float = 0.75, protect: float = 0.33, rms_mix_rate: float = 0.25, resample_sr: int = 0, f0_autotune: bool = False, crepe_hop_length: int = 160, filter_radius: int = 3, use_cache: bool = True, batch_size: int = 1, **kwargs) -> Optional[Tuple[np.ndarray, int]]: """ Perform voice conversion using direct reference calls """ try: print(f"🎵 Minimal wrapper RVC conversion: {f0_method} method, pitch: {f0_up_key}") # Apply before importing RVC modules; some Python 3.13 numba stacks crash # while lazy-loading librosa.util. if sys.version_info >= (3, 13): apply_librosa_compatibility_patches() # Import reference functions using absolute imports to avoid package issues from engines.rvc.impl.vc_infer_pipeline import get_vc, vc_single from engines.rvc.impl.lib.model_utils import load_hubert from engines.rvc.impl.config import config from utils.device import resolve_torch_device # CRITICAL FIX: Reload models to correct device if they were offloaded # Use intelligent device detection with MPS support target_device = resolve_torch_device("auto") # Load RVC model (with caching to prevent VRAM spikes) cache_key = f"{model_path}:{index_path}" if use_cache and cache_key in self._model_cache: print(f"♻️ Using cached RVC model: {os.path.basename(model_path)}") model_data = self._model_cache[cache_key] # Check if model was offloaded to CPU and needs to be reloaded if 'net_g' in model_data and hasattr(model_data['net_g'], 'parameters'): try: first_param = next(model_data['net_g'].parameters()) current_device = str(first_param.device) if current_device != target_device: print(f"🔄 Reloading cached RVC model from {current_device} to {target_device}") if hasattr(model_data['net_g'], 'to'): model_data['net_g'] = model_data['net_g'].to(target_device) if 'vc' in model_data and hasattr(model_data['vc'], 'to'): model_data['vc'] = model_data['vc'].to(target_device) # Re-register with ComfyUI after reload self._register_rvc_model_with_comfyui(model_data, model_path) except StopIteration: pass else: print(f"🔄 Loading RVC model via minimal wrapper: {os.path.basename(model_path)}") model_data = get_vc(model_path, index_path) if not model_data: print("❌ Failed to load RVC model") return None if use_cache: self._model_cache[cache_key] = model_data print(f"💾 Cached RVC model for reuse") # CRITICAL: Register with ComfyUI model management so Clear VRAM button can see it self._register_rvc_model_with_comfyui(model_data, model_path) # Load Hubert model (with caching). Auto-selection is deferred until # now because the separate RVC checkpoint determines 768 vs 1024. hubert_path = kwargs.get('hubert_path') if not hubert_path: hubert_path = self._resolve_auto_hubert(model_data) if not hubert_path: print("❌ Hubert model not found") return None if use_cache and hubert_path in self._hubert_cache: print(f"♻️ Using cached Hubert model") hubert_model = self._hubert_cache[hubert_path] # Check if Hubert was offloaded to CPU and needs to be reloaded if hasattr(hubert_model, 'parameters'): try: first_param = next(hubert_model.parameters()) current_device = str(first_param.device) if current_device != target_device: print(f"🔄 Reloading cached Hubert model from {current_device} to {target_device}") hubert_model = hubert_model.to(target_device) self._hubert_cache[hubert_path] = hubert_model # Re-register with ComfyUI after reload self._register_hubert_model_with_comfyui(hubert_model, hubert_path) except StopIteration: pass else: print(f"🔄 Loading Hubert model: {os.path.basename(hubert_path)}") hubert_model = load_hubert(hubert_path, config) if not hubert_model: print("❌ Failed to load Hubert model") return None if use_cache: self._hubert_cache[hubert_path] = hubert_model print(f"💾 Cached Hubert model for reuse") # CRITICAL: Register with ComfyUI model management so Clear VRAM button can see it self._register_hubert_model_with_comfyui(hubert_model, hubert_path) # Prepare input audio input_audio = (audio, sample_rate) # Ensure RMVPE model is available for reference implementation if f0_method in ["rmvpe", "rmvpe+", "rmvpe_onnx"]: print(f"🔧 RVC: {f0_method} method requires RMVPE model, checking availability...") try: from utils.downloads.model_downloader import download_rmvpe_for_reference rmvpe_path = download_rmvpe_for_reference() if rmvpe_path: print(f"✅ RMVPE model ready at: {rmvpe_path}") else: print("❌ RMVPE model download failed, RVC may fail with this f0_method") except Exception as e: print(f"❌ RMVPE download error: {e}") print("⚠️ RMVPE model not available, continuing anyway...") # Call reference vc_single function result = vc_single( cpt=model_data["cpt"], net_g=model_data["net_g"], vc=model_data["vc"], hubert_model=hubert_model, sid=0, # speaker id input_audio=input_audio, f0_up_key=f0_up_key, f0_method=f0_method, file_index=model_data["file_index"], index_rate=index_rate, filter_radius=filter_radius, resample_sr=resample_sr, protect=protect, rms_mix_rate=rms_mix_rate, crepe_hop_length=crepe_hop_length, f0_autotune=f0_autotune, batch_size=batch_size, **kwargs ) if result: output_audio, output_sr = result print(f"✅ Minimal wrapper RVC conversion completed") # CRITICAL FIX for issue #158: Comprehensive cleanup to prevent VRAM accumulation try: import gc # Step 1: Clear pitch extraction models from VC instance if hasattr(model_data['vc'], 'model_rmvpe'): print("🗑️ Clearing RMVPE pitch model...") del model_data['vc'].model_rmvpe if hasattr(model_data['vc'], 'model_fcpe'): print("🗑️ Clearing FCPE pitch model...") del model_data['vc'].model_fcpe if hasattr(model_data['vc'], 'model_crepe'): print("🗑️ Clearing Crepe pitch model...") del model_data['vc'].model_crepe # Step 2: Move output audio to CPU and clear any GPU tensors if isinstance(output_audio, torch.Tensor): output_audio = output_audio.detach().cpu().numpy() # Step 3: Clean up pitch extractors only (they accumulate the most) # RVC and Hubert models stay cached - ComfyUI's free_memory() will unload them # as needed when VRAM is actually tight (follows other TTS engines' approach) print("✅ Pitch extractors cleaned, RVC/Hubert models remain cached") # Step 4: Force garbage collection and CUDA cleanup gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() torch.cuda.synchronize() torch.cuda.empty_cache() print("✅ VRAM cleanup completed") except Exception as e: print(f"⚠️ Warning during cleanup: {e}") return (output_audio, output_sr) else: print("❌ RVC conversion returned None") return None except Exception as e: print(f"❌ Minimal wrapper conversion error: {e}") import traceback traceback.print_exc() return None def _temporary_cwd(self): """Context manager to temporarily change working directory for imports""" class TempCWD: def __init__(self, path): self.path = path self.old_cwd = None def __enter__(self): self.old_cwd = os.getcwd() os.chdir(self.path) return self def __exit__(self, exc_type, exc_val, exc_tb): os.chdir(self.old_cwd) return TempCWD(self.reference_path) def _find_hubert_model(self) -> Optional[str]: """Find available Hubert model.""" try: import folder_paths models_dir = folder_paths.models_dir # Common Hubert model names and locations - RVC compatible first hubert_candidates = [ "content-vec-best.safetensors", # RVC library expects this specifically "hubert_base_jp.safetensors", # Japanese HuBERT (safetensors) "hubert_base_kr.safetensors", # Korean HuBERT (safetensors) "hubert_base.pt", "chinese-hubert-base.pt", "hubert_base_jp.pt", "hubert_base_kr.pt", "chinese-wav2vec2-base.pt" ] for model_name in hubert_candidates: # Try TTS path first, then legacy locations search_paths = [ os.path.join(models_dir, "TTS", "hubert", model_name), os.path.join(models_dir, "TTS", model_name), os.path.join(models_dir, "hubert", model_name), # Legacy os.path.join(models_dir, model_name) # Legacy - direct in models/ ] for model_path in search_paths: if os.path.exists(model_path): print(f"📄 Found Hubert model: {model_name} at {model_path}") return model_path # If no model found, try to download content-vec-best as fallback print("❌ No compatible Hubert model found locally") print("📥 Attempting to download RVC-compatible model as fallback...") try: from engines.rvc.hubert_downloader import find_or_download_hubert fallback_path = find_or_download_hubert("content-vec-best", models_dir) if fallback_path: print(f"✅ Downloaded RVC-compatible fallback: {fallback_path}") return fallback_path else: print("❌ Failed to download fallback model") except Exception as e: print(f"❌ Fallback download failed: {e}") return None except Exception as e: print(f"Error finding Hubert model: {e}") return None # Global wrapper instance minimal_wrapper = MinimalRVCWrapper()