diodiogod
12ce9be0c0
Address issue #319 by deferring automatic HuBERT selection until the RVC checkpoint dimension is known. - Match 768-dimensional voices with ContentVec automatically - Match verified 1024-dimensional voices with HuBERT Large automatically - Warn that HuBERT Large training lacks a compatible pretrained generator - Recommend ContentVec 768 for reliable RVC training and inference
651 lines
27 KiB
Python
651 lines
27 KiB
Python
"""
|
|
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()
|