import os import numpy as np import torch import gradio as gr import spaces from typing import Optional, Tuple try: from funasr import AutoModel HAS_FUNASR = True except ImportError: HAS_FUNASR = False print("Warning: funasr not installed. ASR features will be disabled.") # Dummy class for type hinting class AutoModel: pass from pathlib import Path os.environ["TOKENIZERS_PARALLELISM"] = "false" if os.environ.get("HF_REPO_ID", "").strip() == "": os.environ["HF_REPO_ID"] = "openbmb/VoxCPM1.5" import voxcpm class VoxCPMDemo: def __init__(self) -> None: self.device = "cuda" if torch.cuda.is_available() else "cpu" print(f"🚀 Running on device: {self.device}") # ASR model for prompt text recognition self.asr_model_id = "iic/SenseVoiceSmall" if HAS_FUNASR: self.asr_model: Optional[AutoModel] = AutoModel( model=self.asr_model_id, disable_update=True, log_level='DEBUG', device="cuda:0" if self.device == "cuda" else "cpu", ) else: self.asr_model = None # TTS model (lazy init) self.voxcpm_model: Optional[voxcpm.VoxCPM] = None self.default_local_model_dir = "./models/VoxCPM1.5" # ---------- Model helpers ---------- def _resolve_model_dir(self) -> str: """ Resolve model directory: 1) Use local checkpoint directory if exists 2) If HF_REPO_ID env is set, download into models/{repo} 3) Fallback to 'models' """ if os.path.isdir(self.default_local_model_dir): return self.default_local_model_dir repo_id = os.environ.get("HF_REPO_ID", "").strip() if len(repo_id) > 0: target_dir = os.path.join("models", repo_id.replace("/", "__")) # Check for essential files to ensure download is complete required_files = ["config.json", "audiovae.pth"] has_weights = os.path.exists(os.path.join(target_dir, "model.safetensors")) or \ os.path.exists(os.path.join(target_dir, "pytorch_model.bin")) is_complete = os.path.isdir(target_dir) and \ all(os.path.exists(os.path.join(target_dir, f)) for f in required_files) and \ has_weights if not is_complete: if os.path.isdir(target_dir): print(f"Found incomplete model directory: {target_dir}. Re-downloading...") import shutil shutil.rmtree(target_dir) try: from huggingface_hub import snapshot_download # type: ignore os.makedirs(target_dir, exist_ok=True) print(f"Downloading model from HF repo '{repo_id}' to '{target_dir}' ...") snapshot_download(repo_id=repo_id, local_dir=target_dir, local_dir_use_symlinks=False) except Exception as e: print(f"Warning: HF download failed: {e}. Falling back to 'data'.") return "models" return target_dir return "models" def get_or_load_voxcpm(self) -> voxcpm.VoxCPM: if self.voxcpm_model is not None: return self.voxcpm_model print("Model not loaded, initializing...") model_dir = self._resolve_model_dir() print(f"Using model dir: {model_dir}") self.voxcpm_model = voxcpm.VoxCPM(voxcpm_model_path=model_dir) print("Model loaded successfully.") return self.voxcpm_model # ---------- Functional endpoints ---------- def prompt_wav_recognition(self, prompt_wav: Optional[str]) -> str: if prompt_wav is None: return "" if self.asr_model is None: return "ASR disabled (funasr not installed)" res = self.asr_model.generate(input=prompt_wav, language="auto", use_itn=True) text = res[0]["text"].split('|>')[-1] return text def generate_tts_audio( self, text_input: str, prompt_wav_path_input: Optional[str] = None, prompt_text_input: Optional[str] = None, cfg_value_input: float = 2.0, inference_timesteps_input: int = 10, do_normalize: bool = True, denoise: bool = True, ) -> Tuple[int, np.ndarray]: """ Generate speech from text using VoxCPM; optional reference audio for voice style guidance. Returns (sample_rate, waveform_numpy) """ current_model = self.get_or_load_voxcpm() text = (text_input or "").strip() if len(text) == 0: raise ValueError("Please input text to synthesize.") prompt_wav_path = prompt_wav_path_input if prompt_wav_path_input else None prompt_text = prompt_text_input if prompt_text_input else None print(f"Generating audio for text: '{text[:60]}...'") wav = current_model.generate( text=text, prompt_text=prompt_text, prompt_wav_path=prompt_wav_path, cfg_value=float(cfg_value_input), inference_timesteps=int(inference_timesteps_input), normalize=do_normalize, denoise=denoise, ) return (current_model.tts_model.sample_rate, wav) # ---------- UI Builders ---------- def create_demo_interface(demo: VoxCPMDemo): """Build the Gradio UI for VoxCPM demo.""" # static assets (logo path) gr.set_static_paths(paths=[Path.cwd().absolute()/"assets"]) with gr.Blocks( theme=gr.themes.Soft( primary_hue="blue", secondary_hue="gray", neutral_hue="slate", font=[gr.themes.GoogleFont("Inter"), "Arial", "sans-serif"] ), css=""" .logo-container { text-align: center; margin: 0.5rem 0 1rem 0; } .logo-container img { height: 80px; width: auto; max-width: 200px; display: inline-block; } /* Bold accordion labels */ #acc_quick details > summary, #acc_tips details > summary { font-weight: 600 !important; font-size: 1.1em !important; } /* Bold labels for specific checkboxes */ #chk_denoise label, #chk_denoise span, #chk_normalize label, #chk_normalize span { font-weight: 600; } """ ) as interface: # Header logo gr.HTML('
