try: import spaces except ImportError: pass import asyncio import functools import os def _run_with_unpatch(target, *args, **kwargs): """Module-level helper for multiprocessing spawn compatibility. Called inside every child process created by nanovllm-voxcpm to restore real CUDA functions that ZeroGPU patches to stubs. """ try: import torch.utils._python_dispatch as _pydisp import torch.overrides as _overrides try: stack = _pydisp._get_dispatch_mode_stack() stack.clear() except Exception: pass try: stack = _overrides._get_function_overrides_stack() stack.clear() except Exception: pass except Exception: pass try: import spaces.zero.torch.patching as _patching try: _patching.unpatch() except Exception: pass except Exception: pass import torch as _torch _real_torch_load = getattr(_torch, "_real_load", _torch.load) _torch._real_load = _real_torch_load def _patched_load(f, map_location=None, *a, **kw): if map_location is None: map_location = "cpu" return _real_torch_load(f, map_location=map_location, *a, **kw) _torch.load = _patched_load return target(*args, **kwargs) def _is_spaces_target(target): actual = target while isinstance(actual, functools.partial): actual = actual.func return getattr(actual, "__module__", "").startswith("spaces.") # Suppress harmless Python 3.10 asyncio __del__ noise when file descriptors # are already invalidated during event-loop garbage collection. _original_loop_del = asyncio.base_events.BaseEventLoop.__del__ def _safe_loop_del(self): try: _original_loop_del(self) except Exception: pass asyncio.base_events.BaseEventLoop.__del__ = _safe_loop_del # ZeroGPU workers are daemon processes by default, but nanovllm-voxcpm needs # to spawn its own child processes inside the worker. Patch spaces to use # non-daemon workers so multiprocessing.Process.start() succeeds. # Also patch every multiprocessing child to call torch.unpatch() so # nanovllm-voxcpm's nested workers get real CUDA instead of ZeroGPU stubs. if os.environ.get("SPACES_ZERO_GPU", "").lower() in ("1", "t", "true"): try: import spaces.zero.wrappers as _spaces_wrappers _orig_process_cls = _spaces_wrappers.Process class _NonDaemonProcess(_orig_process_cls): def __init__(self, *args, **kwargs): kwargs["daemon"] = False super().__init__(*args, **kwargs) _spaces_wrappers.Process = _NonDaemonProcess except Exception: pass try: import multiprocessing.process as _mp_process _original_base_process_init = _mp_process.BaseProcess.__init__ def _patched_base_process_init(self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None): if kwargs is None: kwargs = {} if target is not None and not _is_spaces_target(target): target = functools.partial(_run_with_unpatch, target) return _original_base_process_init( self, group=group, target=target, name=name, args=args, kwargs=kwargs, daemon=daemon ) _mp_process.BaseProcess.__init__ = _patched_base_process_init except Exception: pass import atexit import io import json import logging import os import queue import sys import tempfile from datetime import datetime, timezone from pathlib import Path from threading import Lock, Semaphore, Thread from typing import Optional, Tuple import gradio as gr import numpy as np os.environ["TOKENIZERS_PARALLELISM"] = "false" os.environ["OPENBLAS_NUM_THREADS"] = "4" os.environ["OMP_NUM_THREADS"] = "4" os.environ["MKL_NUM_THREADS"] = "4" DEFAULT_MODEL_REF = "openbmb/VoxCPM2" if ( os.environ.get("NANOVLLM_MODEL", "").strip() == "" and os.environ.get("NANOVLLM_MODEL_PATH", "").strip() == "" and os.environ.get("HF_REPO_ID", "").strip() == "" ): os.environ["HF_REPO_ID"] = DEFAULT_MODEL_REF logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) logger = logging.getLogger(__name__) DEFAULT_ASR_MODEL_REF = "FunAudioLLM/SenseVoiceSmall" DEFAULT_ZIPENHANCER_MODEL = "iic/speech_zipenhancer_ans_multiloss_16k_base" MAX_REFERENCE_AUDIO_SECONDS = 50.0 _persistent_root = None _request_log_dir = None def _configure_cache_dirs() -> None: global _persistent_root, _request_log_dir persistent_root = Path(os.environ.get("SPACE_PERSISTENT_ROOT", "/data")).expanduser() if not persistent_root.exists(): logger.info("Persistent storage not detected. Request logs disabled.") return logs_dir = Path( os.environ.get("REQUEST_LOG_DIR", str(persistent_root / "logs")) ).expanduser() logs_dir.mkdir(parents=True, exist_ok=True) _persistent_root = persistent_root _request_log_dir = logs_dir logger.info(f"Persistent storage detected at {persistent_root}") logger.info(f"Request logs will be written to daily files under {_request_log_dir}") _configure_cache_dirs() _asr_model = None _voxcpm_server = None _model_info = None _denoiser = None _asr_lock = Lock() _server_lock = Lock() _prewarm_lock = Lock() _denoiser_lock = Lock() _denoise_semaphore = Semaphore(int(os.environ.get("DENOISE_MAX_CONCURRENT", "1"))) _prewarm_started = False _runtime_diag_logged = False _active_generation_requests = 0 _active_generation_lock = Lock() def _get_int_env(name: str, default: int) -> int: value = os.environ.get(name, "").strip() if not value: return default return int(value) def _get_float_env(name: str, default: float) -> float: value = os.environ.get(name, "").strip() if not value: return default return float(value) def _get_bool_env(name: str, default: bool) -> bool: value = os.environ.get(name, "").strip().lower() if not value: return default if value in {"1", "true", "yes", "on"}: return True if value in {"0", "false", "no", "off"}: return False raise ValueError(f"Invalid boolean env: {name}={value!r}") def _get_devices_env() -> list[int]: raw = os.environ.get("NANOVLLM_SERVERPOOL_DEVICES", "0").strip() values = [part.strip() for part in raw.split(",") if part.strip()] if not values: return [0] return [int(part) for part in values] def _resolve_model_ref() -> str: for env_name in ("NANOVLLM_MODEL", "NANOVLLM_MODEL_PATH", "HF_REPO_ID"): value = os.environ.get(env_name, "").strip() if value: return value return DEFAULT_MODEL_REF def _resolve_asr_model_ref() -> str: return DEFAULT_ASR_MODEL_REF def _resolve_zipenhancer_model_ref() -> str: for env_name in ("ZIPENHANCER_MODEL_ID", "ZIPENHANCER_MODEL_PATH"): value = os.environ.get(env_name, "").strip() if value: return value return DEFAULT_ZIPENHANCER_MODEL def _log_runtime_diagnostics_once() -> None: global _runtime_diag_logged if _runtime_diag_logged: return import torch info = { "python": sys.version.split()[0], "torch": torch.__version__, "cuda": torch.version.cuda, "cuda_available": torch.cuda.is_available(), "cuda_device_count": torch.cuda.device_count(), "cxx11abi": bool(torch._C._GLIBCXX_USE_CXX11_ABI), "model_ref": _resolve_model_ref(), "devices": _get_devices_env(), } logger.info(f"Runtime diagnostics: {info}") _runtime_diag_logged = True class _ZipEnhancer: def __init__(self, model_ref: str): import torchaudio from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks self._torchaudio = torchaudio self.model_ref = model_ref self._pipeline = pipeline(Tasks.acoustic_noise_suppression, model=model_ref) def _normalize_loudness(self, wav_path: str) -> None: audio, sr = self._torchaudio.load(wav_path) loudness = self._torchaudio.functional.loudness(audio, sr) normalized_audio = self._torchaudio.functional.gain(audio, -20 - loudness) self._torchaudio.save(wav_path, normalized_audio, sr) def enhance(self, input_path: str) -> str: with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp_file: output_path = tmp_file.name try: self._pipeline(input_path, output_path=output_path) self._normalize_loudness(output_path) return output_path except Exception: if os.path.exists(output_path): try: os.unlink(output_path) except OSError: pass raise def get_denoiser(): global _denoiser if _denoiser is not None: return _denoiser with _denoiser_lock: if _denoiser is not None: return _denoiser model_ref = _resolve_zipenhancer_model_ref() logger.info(f"Loading ZipEnhancer denoiser from {model_ref} ...") _denoiser = _ZipEnhancer(model_ref) logger.info("ZipEnhancer denoiser loaded.") return _denoiser def _extract_asr_text(asr_result) -> str: if not asr_result: return "" first_item = asr_result[0] if isinstance(first_item, dict): return str(first_item.get("text", "")).split("|>")[-1].strip() return "" def _read_audio_bytes(audio_path: Optional[str]) -> tuple[bytes | None, str | None]: if audio_path is None or not audio_path.strip(): return None, None path = Path(audio_path) audio_format = path.suffix.lstrip(".").lower() or "wav" if audio_format == "wav": return path.read_bytes(), audio_format import torchaudio waveform, sr = torchaudio.load(str(path)) if waveform.shape[0] > 1: waveform = waveform.mean(dim=0, keepdim=True) buf = io.BytesIO() torchaudio.save(buf, waveform, sr, format="wav") return buf.getvalue(), "wav" def _get_audio_duration_seconds(audio_path: str) -> float: import warnings import torchaudio with warnings.catch_warnings(): warnings.simplefilter("ignore") info = torchaudio.info(audio_path) return float(info.num_frames) / float(info.sample_rate) def _begin_generation_request() -> None: global _active_generation_requests with _active_generation_lock: _active_generation_requests += 1 def _end_generation_request() -> None: global _active_generation_requests with _active_generation_lock: _active_generation_requests = max(0, _active_generation_requests - 1) def _get_active_generation_requests() -> int: with _active_generation_lock: return _active_generation_requests def _trim_audio_to_seconds(audio_path: str, max_seconds: float) -> str: import torchaudio waveform, sr = torchaudio.load(audio_path) max_frames = int(max_seconds * sr) if waveform.shape[1] <= max_frames: return audio_path trimmed = waveform[:, :max_frames] tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) tmp.close() torchaudio.save(tmp.name, trimmed, sr) return tmp.name def _validate_reference_audio_duration( audio_path: str, request: Optional[gr.Request] = None ) -> str: duration_seconds = _get_audio_duration_seconds(audio_path) if duration_seconds > MAX_REFERENCE_AUDIO_SECONDS: logger.info(f"Reference audio ({duration_seconds:.1f}s) exceeds limit, trimming to {MAX_REFERENCE_AUDIO_SECONDS - 1:.0f}s") return _trim_audio_to_seconds(audio_path, MAX_REFERENCE_AUDIO_SECONDS - 1.0) return audio_path def _prepare_audio_for_encoding( audio_path: Optional[str], *, denoise: bool, request: Optional[gr.Request] = None, ) -> tuple[bytes | None, str | None, Optional[str]]: if audio_path is None or not audio_path.strip(): return None, None, None original_path = audio_path audio_path = _validate_reference_audio_duration(audio_path, request) source_path = audio_path temp_path = audio_path if audio_path != original_path else None if denoise: logger.info("Applying ZipEnhancer denoising to reference audio ...") acquired = _denoise_semaphore.acquire(timeout=30) if not acquired: raise gr.Error(_get_i18n_text("denoise_busy_error", request)) try: temp_path = get_denoiser().enhance(audio_path) source_path = temp_path except Exception as exc: logger.exception("ZipEnhancer denoising failed") raise gr.Error(_get_i18n_text("denoise_failed_error", request)) from exc finally: _denoise_semaphore.release() audio_bytes, audio_format = _read_audio_bytes(source_path) return audio_bytes, audio_format, temp_path def _safe_prompt_wav_recognition( use_prompt_text: bool, prompt_wav: Optional[str], request: Optional[gr.Request] = None ) -> str: try: return prompt_wav_recognition(use_prompt_text, prompt_wav) except Exception as exc: logger.warning(f"ASR recognition failed: {exc}") raise gr.Error(_get_i18n_text("asr_failed_error", request)) from exc def _stop_server_if_needed() -> None: global _voxcpm_server, _model_info if _voxcpm_server is None: return if isinstance(_voxcpm_server, _AsyncServerBridge): _voxcpm_server.stop() else: stop = getattr(_voxcpm_server, "stop", None) if callable(stop): try: stop() except Exception as exc: logger.warning(f"Failed to stop nano-vLLM server cleanly: {exc}") _voxcpm_server = None _model_info = None atexit.register(_stop_server_if_needed) # ---------- Inline i18n (en + zh-CN only) ---------- _USAGE_INSTRUCTIONS_EN = "" _EXAMPLES_FOOTER_EN = "" _USAGE_INSTRUCTIONS_ZH = "" _EXAMPLES_FOOTER_ZH = "" _I18N_TRANSLATIONS = { "en": { "reference_audio_label": "🎤 Reference Audio (optional — upload for cloning)", "show_prompt_text_label": "🎙️ Ultimate Cloning Mode (transcript-guided cloning)", "show_prompt_text_info": "Auto-transcribes reference audio for every vocal nuance reproduced. Control Instruction will be disabled when active.", "prompt_text_label": "Transcript of Reference Audio (auto-filled via ASR, editable)", "prompt_text_placeholder": "The transcript of your reference audio will appear here …", "control_label": "🎛️ Control Instruction (optional — supports Chinese & English)", "control_placeholder": "e.g. A warm young woman / 年轻女性，温柔甜美 / Excited and fast-paced", "target_text_label": "✍️ Target Text — the content to speak", "generate_btn": "🔊 Generate Speech", "generated_audio_label": "Generated Audio", "advanced_settings_title": "⚙️ Advanced Settings", "ref_denoise_label": "Reference audio enhancement", "ref_denoise_info": "Apply ZipEnhancer denoising to the reference audio before cloning", "normalize_label": "Text normalization", "normalize_info": "Normalize numbers, dates, and abbreviations via wetext", "cfg_label": "CFG (guidance scale)", "cfg_info": "Higher → closer to the prompt / reference; lower → more creative variation", "reference_audio_too_long_error": "Reference audio is too long. Please upload audio no longer than 50 seconds.", "denoise_busy_error": "Too many reference-audio enhancement requests are running. Please try again in a moment.", "denoise_failed_error": "Reference audio enhancement failed. Please try disabling denoise or use a cleaner clip.", "backend_retry_error": "The backend is temporarily unstable. Please try again in a moment.", "asr_failed_error": "ASR failed. Please fill the transcript manually or try another reference audio.", "usage_instructions": _USAGE_INSTRUCTIONS_EN, "examples_footer": _EXAMPLES_FOOTER_EN, }, "zh-CN": { "reference_audio_label": "🎤 参考音频（可选 — 上传后用于克隆）", "show_prompt_text_label": "🎙️ 极致克隆模式（基于文本引导的极致克隆）", "show_prompt_text_info": "自动识别参考音频文本，完整还原音色、节奏、情感等全部声音细节。开启后 Control Instruction 将暂时禁用", "prompt_text_label": "参考音频内容文本（ASR 自动填充，可手动编辑）", "prompt_text_placeholder": "参考音频的文字内容将自动识别并显示在此处 …", "control_label": "🎛️ Control Instruction（可选 — 支持中英文描述）", "control_placeholder": "如：年轻女性，温柔甜美 / A warm young woman / 暴躁老哥，语速飞快", "target_text_label": "✍️ Target Text — 要合成的目标文本", "generate_btn": "🔊 开始生成", "generated_audio_label": "生成结果", "advanced_settings_title": "⚙️ 高级设置", "ref_denoise_label": "参考音频降噪增强", "ref_denoise_info": "克隆前使用 ZipEnhancer 对参考音频进行降噪处理", "normalize_label": "文本规范化", "normalize_info": "自动规范化数字、日期及缩写（基于 wetext）", "cfg_label": "CFG（引导强度）", "cfg_info": "数值越高 → 越贴合提示/参考音色；数值越低 → 生成风格更自由", "reference_audio_too_long_error": "参考音频太长了，请上传不超过 50 秒的音频。", "denoise_busy_error": "当前参考音频降噪请求过多，请稍后再试。", "denoise_failed_error": "参考音频降噪失败，请尝试关闭降噪或更换更干净的音频。", "backend_retry_error": "后端暂时不稳定，请稍后再试。", "asr_failed_error": "ASR 识别失败，请手动填写参考音频文本，或更换一段参考音频后重试。", "usage_instructions": _USAGE_INSTRUCTIONS_ZH, "examples_footer": _EXAMPLES_FOOTER_ZH, }, "zh-Hans": None, "zh": None, } _I18N_TRANSLATIONS["zh-Hans"] = _I18N_TRANSLATIONS["zh-CN"] _I18N_TRANSLATIONS["zh"] = _I18N_TRANSLATIONS["zh-CN"] for _d in _I18N_TRANSLATIONS.values(): if _d is not None: for _k, _v in _I18N_TRANSLATIONS["en"].items(): _d.setdefault(_k, _v) I18N = gr.I18n(**_I18N_TRANSLATIONS) def _resolve_ui_language(request: Optional[gr.Request] = None) -> str: if request is None: return "en" accept_language = str(request.headers.get("accept-language", "")).lower() if accept_language.startswith("zh"): return "zh-CN" return "en" def _get_i18n_text(key: str, request: Optional[gr.Request] = None) -> str: locale = _resolve_ui_language(request) return _I18N_TRANSLATIONS.get(locale, _I18N_TRANSLATIONS["en"]).get( key, _I18N_TRANSLATIONS["en"].get(key, key) ) def _append_request_log(payload: dict) -> None: if _request_log_dir is None: return now = datetime.now(timezone.utc) record = {"timestamp": now.isoformat(), **payload} log_path = _request_log_dir / f"{now.date().isoformat()}.jsonl" with log_path.open("a", encoding="utf-8") as fp: fp.write(json.dumps(record, ensure_ascii=False) + "\n") DEFAULT_TARGET_TEXT = ( "VoxCPM2 is a creative multilingual TTS model from ModelBest, " "designed to generate highly realistic speech." ) _CUSTOM_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; } /* Toggle switch style */ .switch-toggle { padding: 8px 12px; border-radius: 8px; background: var(--block-background-fill); } .switch-toggle input[type="checkbox"] { appearance: none; -webkit-appearance: none; width: 44px; height: 24px; background: #ccc; border-radius: 12px; position: relative; cursor: pointer; transition: background 0.3s ease; flex-shrink: 0; } .switch-toggle input[type="checkbox"]::after { content: ""; position: absolute; top: 2px; left: 2px; width: 20px; height: 20px; background: white; border-radius: 50%; transition: transform 0.3s ease; box-shadow: 0 1px 3px rgba(0,0,0,0.2); } .switch-toggle input[type="checkbox"]:checked { background: var(--color-accent); } .switch-toggle input[type="checkbox"]:checked::after { transform: translateX(20px); } """ _APP_THEME = gr.themes.Soft( primary_hue="blue", secondary_hue="gray", neutral_hue="slate", font=[gr.themes.GoogleFont("Inter"), "Arial", "sans-serif"], ) def get_asr_model(): global _asr_model if _asr_model is not None: return _asr_model with _asr_lock: if _asr_model is not None: return _asr_model from funasr import AutoModel from huggingface_hub import snapshot_download device = os.environ.get("ASR_DEVICE", "cpu").strip() or "cpu" asr_model_ref = _resolve_asr_model_ref() logger.info(f"Downloading ASR model from Hugging Face: {asr_model_ref}") asr_model_path = snapshot_download(repo_id=asr_model_ref) logger.info(f"Loading ASR model on {device} ...") _asr_model = AutoModel( model=asr_model_path, disable_update=True, log_level="INFO", device=device, ) logger.info("ASR model loaded.") return _asr_model class _AsyncServerBridge: """Thread-safe bridge to AsyncVoxCPM2ServerPool running in a dedicated event loop.""" def __init__(self): self._loop: Optional[asyncio.AbstractEventLoop] = None self._thread: Optional[Thread] = None self._server_pool = None self._model_info: Optional[dict] = None self._closed = False def _run_loop(self) -> None: assert self._loop is not None asyncio.set_event_loop(self._loop) try: self._loop.run_forever() finally: try: self._loop.close() except Exception: pass def start(self) -> None: _log_runtime_diagnostics_once() model_ref = _resolve_model_ref() logger.info(f"Loading nano-vLLM VoxCPM async server from {model_ref} ...") self._loop = asyncio.new_event_loop() self._thread = Thread(target=self._run_loop, name="nanovllm-event-loop", daemon=True) self._thread.start() try: async def _init(): from nanovllm_voxcpm import VoxCPM pool = VoxCPM.from_pretrained( model=model_ref, max_num_batched_tokens=_get_int_env("NANOVLLM_SERVERPOOL_MAX_NUM_BATCHED_TOKENS", 8192), max_num_seqs=_get_int_env("NANOVLLM_SERVERPOOL_MAX_NUM_SEQS", 16), max_model_len=_get_int_env("NANOVLLM_SERVERPOOL_MAX_MODEL_LEN", 4096), gpu_memory_utilization=_get_float_env("NANOVLLM_SERVERPOOL_GPU_MEMORY_UTILIZATION", 0.95), enforce_eager=_get_bool_env("NANOVLLM_SERVERPOOL_ENFORCE_EAGER", False), devices=_get_devices_env(), ) await pool.wait_for_ready() return pool future = asyncio.run_coroutine_threadsafe(_init(), self._loop) self._server_pool = future.result() info_future = asyncio.run_coroutine_threadsafe( self._server_pool.get_model_info(), self._loop ) self._model_info = info_future.result() logger.info(f"nano-vLLM async server loaded: {self._model_info}") except Exception: self.stop() raise def get_model_info(self) -> dict: assert self._model_info is not None return self._model_info def encode_latents(self, wav: bytes, wav_format: str, timeout: float = 120) -> bytes: if self._closed: raise RuntimeError("nano-vLLM bridge is closed") assert self._loop is not None and self._server_pool is not None future = asyncio.run_coroutine_threadsafe( self._server_pool.encode_latents(wav, wav_format), self._loop ) try: return future.result(timeout=timeout) finally: if not future.done(): future.cancel() def generate(self, timeout: float = 300, **kwargs): if self._closed: raise RuntimeError("nano-vLLM bridge is closed") assert self._loop is not None and self._server_pool is not None result_queue: queue.Queue = queue.Queue() import time as _time async def _drain(): try: async for chunk in self._server_pool.generate(**kwargs): result_queue.put(chunk) result_queue.put(None) except Exception as exc: result_queue.put(exc) deadline = _time.monotonic() + timeout future = asyncio.run_coroutine_threadsafe(_drain(), self._loop) try: while True: remaining = deadline - _time.monotonic() if remaining <= 0: raise TimeoutError(f"Generation exceeded {timeout}s timeout") try: item = result_queue.get(timeout=min(0.5, remaining)) except queue.Empty: if future.done(): exc = future.exception() if exc is not None: raise exc continue if item is None: break if isinstance(item, Exception): raise item yield item finally: if not future.done(): future.cancel() def stop(self) -> None: if self._closed: return self._closed = True try: if self._loop is not None and self._server_pool is not None: future = asyncio.run_coroutine_threadsafe(self._server_pool.stop(), self._loop) future.result(timeout=10) except Exception as exc: logger.warning(f"Failed to stop async server pool cleanly: {exc}") finally: try: if self._loop is not None: self._loop.call_soon_threadsafe(self._loop.stop) except Exception: pass if self._thread is not None: self._thread.join(timeout=10) self._server_pool = None self._model_info = None self._thread = None self._loop = None def get_voxcpm_server() -> _AsyncServerBridge: global _voxcpm_server, _model_info if _voxcpm_server is not None: return _voxcpm_server with _server_lock: if _voxcpm_server is not None: return _voxcpm_server bridge = _AsyncServerBridge() bridge.start() _voxcpm_server = bridge _model_info = bridge.get_model_info() return _voxcpm_server def get_model_info() -> dict: global _model_info if _model_info is None: get_voxcpm_server() assert _model_info is not None return _model_info def _prewarm_backend() -> None: try: logger.info("Starting backend prewarm ...") get_voxcpm_server() logger.info("Backend prewarm completed.") except Exception as exc: logger.warning(f"Backend prewarm failed: {exc}") def _start_background_prewarm() -> None: global _prewarm_started if not _get_bool_env("NANOVLLM_PREWARM", True): return # ZeroGPU: model must be loaded inside the @spaces.GPU worker process, # so avoid initializing CUDA in the main process. if os.environ.get("SPACES_ZERO_GPU", "").lower() in ("1", "t", "true"): logger.info("ZeroGPU detected via env: disabling background prewarm to avoid main-process CUDA init") return with _prewarm_lock: if _prewarm_started: return _prewarm_started = True Thread(target=_prewarm_backend, name="nanovllm-prewarm", daemon=True).start() # ---------- GPU-accelerated inference ---------- def prompt_wav_recognition(use_prompt_text: bool, prompt_wav: Optional[str]) -> str: if not use_prompt_text or prompt_wav is None or not prompt_wav.strip(): return "" asr_model = get_asr_model() res = asr_model.generate(input=prompt_wav, language="auto", use_itn=True) return _extract_asr_text(res) def _float_audio_to_int16(wav: np.ndarray) -> np.ndarray: clipped = np.clip(wav, -1.0, 1.0) return (clipped * 32767.0).astype(np.int16, copy=False) def _generate_tts_audio_once( text_input: str, control_instruction: str = "", reference_wav_path_input: Optional[str] = None, use_prompt_text: bool = False, prompt_text_input: str = "", cfg_value_input: float = 2.0, do_normalize: bool = True, denoise: bool = True, request: Optional[gr.Request] = None, ) -> Tuple[int, np.ndarray]: temp_audio_path = None try: server = get_voxcpm_server() model_info = get_model_info() text = (text_input or "").strip() if len(text) == 0: raise ValueError("Please input text to synthesize.") control = (control_instruction or "").strip() final_text = f"({control}){text}" if control and not use_prompt_text else text audio_bytes, audio_format, temp_audio_path = _prepare_audio_for_encoding( reference_wav_path_input, denoise=bool(denoise), request=request, ) prompt_text_clean = (prompt_text_input or "").strip() if use_prompt_text and audio_bytes is None: raise ValueError("Ultimate Cloning Mode requires a reference audio clip.") if use_prompt_text and not prompt_text_clean: raise ValueError( "Ultimate Cloning Mode requires a transcript. Please wait for ASR or fill it in manually." ) if not use_prompt_text: prompt_text_clean = "" if do_normalize: logger.info( "Ignoring normalize option: nano-vLLM backend does not support per-request text normalization." ) prompt_latents = None ref_audio_latents = None if audio_bytes is not None and audio_format is not None and use_prompt_text: logger.info(f"[Ultimate Cloning] encoding prompt audio as {audio_format}") prompt_latents = server.encode_latents(audio_bytes, audio_format) elif audio_bytes is not None and audio_format is not None: logger.info(f"[Controllable Cloning] encoding reference audio as {audio_format}") ref_audio_latents = server.encode_latents(audio_bytes, audio_format) if prompt_latents is not None: logger.info("[Ultimate Cloning] reference audio + transcript") elif ref_audio_latents is not None: logger.info("[Controllable Cloning] reference audio only") else: logger.info(f"[Voice Design] control: {control[:50] if control else 'None'}") chunks: list[np.ndarray] = [] logger.info(f"Generating: '{final_text[:80]}...'") for chunk in server.generate( target_text=final_text, prompt_latents=prompt_latents, prompt_text=prompt_text_clean if prompt_latents is not None else "", max_generate_length=_get_int_env("NANOVLLM_MAX_GENERATE_LENGTH", 2000), temperature=_get_float_env("NANOVLLM_TEMPERATURE", 1.0), cfg_value=float(cfg_value_input), ref_audio_latents=ref_audio_latents, ): chunks.append(chunk) if not chunks: raise RuntimeError("The model returned no audio chunks.") wav = np.concatenate(chunks, axis=0).astype(np.float32, copy=False) wav = _float_audio_to_int16(wav) return (int(model_info["sample_rate"]), wav) finally: if temp_audio_path and os.path.exists(temp_audio_path): try: os.unlink(temp_audio_path) except OSError: pass # ZeroGPU wrapper: GPU work must run inside a @spaces.GPU worker process. # The main process on ZeroGPU has fake CUDA; real GPU is only available in workers. # Workers are reused, so model loading happens once on the first call. @spaces.GPU(duration=600) def _gpu_generate_tts_audio_once( text_input: str, control_instruction: str = "", reference_wav_path_input: Optional[str] = None, use_prompt_text: bool = False, prompt_text_input: str = "", cfg_value_input: float = 2.0, do_normalize: bool = True, denoise: bool = True, ) -> Tuple[int, np.ndarray]: return _generate_tts_audio_once( text_input=text_input, control_instruction=control_instruction, reference_wav_path_input=reference_wav_path_input, use_prompt_text=use_prompt_text, prompt_text_input=prompt_text_input, cfg_value_input=cfg_value_input, do_normalize=do_normalize, denoise=denoise, request=None, ) def generate_tts_audio( text_input: str, control_instruction: str = "", reference_wav_path_input: Optional[str] = None, use_prompt_text: bool = False, prompt_text_input: str = "", cfg_value_input: float = 2.0, do_normalize: bool = True, denoise: bool = True, request: Optional[gr.Request] = None, ) -> Tuple[int, np.ndarray]: _begin_generation_request() request_payload = { "event": "tts_request", "ui_language": _resolve_ui_language(request), "text": (text_input or "").strip(), "control_instruction": (control_instruction or "").strip(), "use_prompt_text": bool(use_prompt_text), "prompt_text": (prompt_text_input or "").strip(), "cfg_value": float(cfg_value_input), "do_normalize": bool(do_normalize), "denoise": bool(denoise), "has_reference_audio": bool(reference_wav_path_input and reference_wav_path_input.strip()), } if request_payload["has_reference_audio"]: try: request_payload["reference_audio_duration_seconds"] = round( _get_audio_duration_seconds(reference_wav_path_input), 3 ) except Exception as exc: request_payload["reference_audio_duration_error"] = str(exc) try: try: result = _gpu_generate_tts_audio_once( text_input=text_input, control_instruction=control_instruction, reference_wav_path_input=reference_wav_path_input, use_prompt_text=use_prompt_text, prompt_text_input=prompt_text_input, cfg_value_input=cfg_value_input, do_normalize=do_normalize, denoise=denoise, ) try: _append_request_log({**request_payload, "status": "success"}) except Exception as exc: logger.warning(f"Failed to append request log: {exc}") return result except (ValueError, gr.Error) as exc: try: _append_request_log( {**request_payload, "status": "rejected", "error": str(exc)} ) except Exception as log_exc: logger.warning(f"Failed to append request log: {log_exc}") if isinstance(exc, gr.Error): raise raise gr.Error(str(exc)) from exc except Exception as exc: logger.exception("Generation failed") try: _append_request_log({**request_payload, "status": "error", "error": str(exc)}) except Exception as log_exc: logger.warning(f"Failed to append request log: {log_exc}") active_requests = _get_active_generation_requests() if active_requests > 1: logger.warning( "Generation failed with %s active requests; skipping shared backend restart: %s", active_requests, exc, ) raise gr.Error(_get_i18n_text("backend_retry_error", request)) from exc logger.warning(f"Generation failed, restarting backend and retrying once: {exc}") with _server_lock: _stop_server_if_needed() try: result = _gpu_generate_tts_audio_once( text_input=text_input, control_instruction=control_instruction, reference_wav_path_input=reference_wav_path_input, use_prompt_text=use_prompt_text, prompt_text_input=prompt_text_input, cfg_value_input=cfg_value_input, do_normalize=do_normalize, denoise=denoise, ) try: _append_request_log({**request_payload, "status": "success_after_retry"}) except Exception as log_exc: logger.warning(f"Failed to append request log: {log_exc}") return result except Exception as retry_exc: logger.exception("Retry failed") try: _append_request_log( {**request_payload, "status": "retry_failed", "error": str(retry_exc)} ) except Exception as log_exc: logger.warning(f"Failed to append request log: {log_exc}") raise gr.Error(_get_i18n_text("backend_retry_error", request)) from retry_exc finally: _end_generation_request() # ---------- UI ---------- def create_demo_interface(): assets_dir = Path.cwd().absolute() / "assets" if assets_dir.exists(): gr.set_static_paths(paths=[assets_dir]) def _on_toggle_instant(checked): if checked: return ( gr.update(visible=True, value="", placeholder="Recognizing reference audio..."), gr.update(visible=False), ) return ( gr.update(visible=False), gr.update(visible=True, interactive=True), ) def _run_asr_if_needed(checked, audio_path, request: gr.Request = None): if not checked or not audio_path: return gr.update() logger.info("Running ASR on reference audio...") asr_text = _safe_prompt_wav_recognition(True, audio_path, request=request) logger.info(f"ASR result: {asr_text[:60]}...") return gr.update( value=asr_text, placeholder=_get_i18n_text("prompt_text_placeholder", request), ) with gr.Blocks() as interface: if (assets_dir / "voxcpm_logo.png").exists(): gr.HTML( 'VoxCPM2' ) gr.Markdown(I18N("usage_instructions")) with gr.Row(): with gr.Column(): reference_wav = gr.Audio( sources=["upload", "microphone"], type="filepath", label=I18N("reference_audio_label"), ) show_prompt_text = gr.Checkbox( value=False, label=I18N("show_prompt_text_label"), info=I18N("show_prompt_text_info"), elem_classes=["switch-toggle"], ) prompt_text = gr.Textbox( value="", label=I18N("prompt_text_label"), placeholder=I18N("prompt_text_placeholder"), lines=2, visible=False, ) control_instruction = gr.Textbox( value="", label=I18N("control_label"), placeholder=I18N("control_placeholder"), lines=2, ) text = gr.Textbox( value=DEFAULT_TARGET_TEXT, label=I18N("target_text_label"), lines=3, ) with gr.Accordion(I18N("advanced_settings_title"), open=False): DoDenoisePromptAudio = gr.Checkbox( value=False, label=I18N("ref_denoise_label"), elem_classes=["switch-toggle"], info=I18N("ref_denoise_info"), ) DoNormalizeText = gr.Checkbox( value=False, label=I18N("normalize_label"), elem_classes=["switch-toggle"], info=I18N("normalize_info"), ) cfg_value = gr.Slider( minimum=1.0, maximum=3.0, value=2.0, step=0.1, label=I18N("cfg_label"), info=I18N("cfg_info"), ) run_btn = gr.Button(I18N("generate_btn"), variant="primary", size="lg") with gr.Column(): audio_output = gr.Audio(label=I18N("generated_audio_label")) gr.Markdown(I18N("examples_footer")) show_prompt_text.change( fn=_on_toggle_instant, inputs=[show_prompt_text], outputs=[prompt_text, control_instruction], ).then( fn=_run_asr_if_needed, inputs=[show_prompt_text, reference_wav], outputs=[prompt_text], ) run_btn.click( fn=generate_tts_audio, inputs=[ text, control_instruction, reference_wav, show_prompt_text, prompt_text, cfg_value, DoNormalizeText, DoDenoisePromptAudio, ], outputs=[audio_output], show_progress=True, api_name="generate", ) return interface def run_demo( server_name: str = "0.0.0.0", server_port: int = 7860, show_error: bool = True ): interface = create_demo_interface() _start_background_prewarm() interface.queue( max_size=_get_int_env("GRADIO_QUEUE_MAX_SIZE", 10), default_concurrency_limit=_get_int_env("GRADIO_DEFAULT_CONCURRENCY_LIMIT", 4), ).launch( server_name=server_name, server_port=int(os.environ.get("PORT", server_port)), show_error=show_error, i18n=I18N, theme=_APP_THEME, css=_CUSTOM_CSS, ssr_mode=_get_bool_env("GRADIO_SSR_MODE", False), ) if __name__ == "__main__": run_demo()