Spaces:
Running
on
Zero
Running
on
Zero
| import spaces # PHẢI import TRƯỚC mọi thứ trên HF Spaces ZeroGPU | |
| import os | |
| os.environ['SPACES_ZERO_GPU'] = '1' | |
| import gradio as gr | |
| import soundfile as sf | |
| import tempfile | |
| import torch | |
| from vieneu_tts import VieNeuTTS, FastVieNeuTTS | |
| import time | |
| import numpy as np | |
| import yaml | |
| from utils.core_utils import split_text_into_chunks | |
| import queue | |
| import threading | |
| print("⏳ Đang khởi động VieNeu-TTS...") | |
| # --- LOAD CONFIG --- | |
| CONFIG_PATH = os.path.join(os.path.dirname(__file__), "config.yaml") | |
| try: | |
| with open(CONFIG_PATH, "r", encoding="utf-8") as f: | |
| _config = yaml.safe_load(f) or {} | |
| except Exception as e: | |
| raise RuntimeError(f"Không thể đọc config.yaml: {e}") | |
| BACKBONE_CONFIGS = _config.get("backbone_configs", {}) | |
| CODEC_CONFIGS = _config.get("codec_configs", {}) | |
| VOICE_SAMPLES = _config.get("voice_samples", {}) | |
| _text_settings = _config.get("text_settings", {}) | |
| MAX_CHARS_PER_CHUNK = _text_settings.get("max_chars_per_chunk", 256) | |
| MAX_TOTAL_CHARS_STREAMING = _text_settings.get("max_total_chars_streaming", 3000) | |
| # --- KHỞI TẠO MODEL - LAZY LOADING --- | |
| print("⏳ Chuẩn bị khởi tạo model (lazy loading cho ZeroGPU)...") | |
| tts = None | |
| using_fast_backend = False | |
| model_loaded = False | |
| def initialize_model(): | |
| """Load model với GPU - chỉ gọi khi cần""" | |
| global tts, using_fast_backend, model_loaded | |
| if model_loaded and tts is not None: | |
| return "✅ Model đã được tải sẵn!" | |
| print("📦 Đang tải model với GPU...") | |
| device = "cuda" if torch.cuda.is_available() else "cpu" | |
| print(f"🖥️ Sử dụng thiết bị: {device.upper()}") | |
| try: | |
| backbone_config = BACKBONE_CONFIGS["VieNeu-TTS (GPU)"] | |
| codec_config = CODEC_CONFIGS["NeuCodec (Standard)"] | |
| # Thử dùng FastVieNeuTTS nếu có GPU và đã cài LMDeploy | |
| if device == "cuda": | |
| try: | |
| print("🚀 Thử tải FastVieNeuTTS (LMDeploy backend)...") | |
| tts = FastVieNeuTTS( | |
| backbone_repo=backbone_config["repo"], | |
| backbone_device="cuda", | |
| codec_repo=codec_config["repo"], | |
| codec_device="cuda", | |
| memory_util=0.3, | |
| tp=1, | |
| enable_prefix_caching=True, | |
| quant_policy=8, | |
| enable_triton=True, | |
| max_batch_size=8, | |
| ) | |
| using_fast_backend = True | |
| print("✅ FastVieNeuTTS đã tải thành công!") | |
| # Pre-cache voices | |
| print("📝 Pre-caching voices...") | |
| for voice_name, voice_info in VOICE_SAMPLES.items(): | |
| audio_path = voice_info["audio"] | |
| text_path = voice_info["text"] | |
| if os.path.exists(audio_path) and os.path.exists(text_path): | |
| with open(text_path, "r", encoding="utf-8") as f: | |
| ref_text = f.read() | |
| tts.get_cached_reference(voice_name, audio_path, ref_text) | |
| print(f" ✅ Cached {len(VOICE_SAMPLES)} voices") | |
| except (ImportError, Exception) as e: | |
| print(f"⚠️ LMDeploy không khả dụng ({e}), fallback về VieNeuTTS standard...") | |
| using_fast_backend = False | |
| # Fallback về standard VieNeuTTS | |
| if tts is None: | |
| print("📦 Đang tải VieNeuTTS (Standard backend)...") | |
| tts = VieNeuTTS( | |
| backbone_repo=backbone_config["repo"], | |
| backbone_device=device, | |
| codec_repo=codec_config["repo"], | |
| codec_device=device | |
| ) | |
| using_fast_backend = False | |
| model_loaded = True | |
| print("✅ Model đã tải xong!") | |
| backend = "🚀 LMDeploy (GPU-optimized)" if using_fast_backend else "📦 Standard Backend" | |
| return f"✅ Model đã tải thành công!\n\n**Backend:** {backend}" | |
| except Exception as e: | |
| print(f"⚠️ Lỗi khi tải model: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| return f"❌ Lỗi: {str(e)}" | |
| # Không tải model ngay lúc start - chỉ tải khi user bắt đầu synthesize | |
| print("✅ Sẵn sàng! Model sẽ được tải khi bạn bắt đầu tổng hợp.") | |
| # --- HELPER FUNCTIONS --- | |
| def load_reference_info(voice_choice): | |
| if voice_choice in VOICE_SAMPLES: | |
| audio_path = VOICE_SAMPLES[voice_choice]["audio"] | |
| text_path = VOICE_SAMPLES[voice_choice]["text"] | |
| try: | |
| if os.path.exists(text_path): | |
| with open(text_path, "r", encoding="utf-8") as f: | |
| ref_text = f.read() | |
| return audio_path, ref_text | |
| else: | |
| return audio_path, "⚠️ Không tìm thấy file text mẫu." | |
| except Exception as e: | |
| return None, f"❌ Lỗi: {str(e)}" | |
| return None, "" | |
| def synthesize_speech(text, voice_choice, custom_audio, custom_text, mode_tab, generation_mode, use_batch): | |
| """Tổng hợp giọng nói với GPU acceleration""" | |
| global tts, using_fast_backend, model_loaded | |
| # Auto-load model nếu chưa có | |
| if not model_loaded or tts is None: | |
| yield None, "⏳ Đang tải model lần đầu..." | |
| result = initialize_model() | |
| if "❌" in result: | |
| yield None, result | |
| return | |
| yield None, "✅ Model đã tải! Bắt đầu tổng hợp..." | |
| if not text or text.strip() == "": | |
| yield None, "⚠️ Vui lòng nhập văn bản!" | |
| return | |
| raw_text = text.strip() | |
| # Setup Reference | |
| if mode_tab == "custom_mode": | |
| if custom_audio is None or not custom_text: | |
| yield None, "⚠️ Thiếu Audio hoặc Text mẫu custom." | |
| return | |
| ref_audio_path = custom_audio | |
| ref_text_raw = custom_text | |
| use_cached = False | |
| else: | |
| if voice_choice not in VOICE_SAMPLES: | |
| yield None, "⚠️ Vui lòng chọn giọng mẫu." | |
| return | |
| ref_audio_path = VOICE_SAMPLES[voice_choice]["audio"] | |
| ref_text_path = VOICE_SAMPLES[voice_choice]["text"] | |
| if not os.path.exists(ref_audio_path): | |
| yield None, "❌ Không tìm thấy file audio mẫu." | |
| return | |
| with open(ref_text_path, "r", encoding="utf-8") as f: | |
| ref_text_raw = f.read() | |
| use_cached = True | |
| yield None, "📄 Đang xử lý Reference..." | |
| # Encode reference | |
| try: | |
| if use_cached and using_fast_backend and hasattr(tts, 'get_cached_reference'): | |
| ref_codes = tts.get_cached_reference(voice_choice, ref_audio_path, ref_text_raw) | |
| else: | |
| ref_codes = tts.encode_reference(ref_audio_path) | |
| if isinstance(ref_codes, torch.Tensor): | |
| ref_codes = ref_codes.cpu().numpy() | |
| except Exception as e: | |
| yield None, f"❌ Lỗi xử lý reference: {e}" | |
| return | |
| # Split text | |
| text_chunks = split_text_into_chunks(raw_text, max_chars=MAX_CHARS_PER_CHUNK) | |
| total_chunks = len(text_chunks) | |
| # === STANDARD MODE === | |
| if generation_mode == "Standard (Một lần)": | |
| backend_name = "🚀 LMDeploy" if using_fast_backend else "📦 Standard" | |
| batch_info = " (Batch Mode)" if use_batch and using_fast_backend and total_chunks > 1 else "" | |
| yield None, f"{backend_name} Đang tổng hợp{batch_info} ({total_chunks} đoạn)..." | |
| all_audio_segments = [] | |
| sr = 24000 | |
| silence_pad = np.zeros(int(sr * 0.15), dtype=np.float32) | |
| start_time = time.time() | |
| try: | |
| # Batch processing nếu có FastVieNeuTTS | |
| if use_batch and using_fast_backend and hasattr(tts, 'infer_batch') and total_chunks > 1: | |
| yield None, f"⚡ Xử lý batch ({total_chunks} đoạn cùng lúc)..." | |
| chunk_wavs = tts.infer_batch(text_chunks, ref_codes, ref_text_raw) | |
| for i, chunk_wav in enumerate(chunk_wavs): | |
| if chunk_wav is not None and len(chunk_wav) > 0: | |
| all_audio_segments.append(chunk_wav) | |
| if i < total_chunks - 1: | |
| all_audio_segments.append(silence_pad) | |
| else: | |
| # Sequential processing | |
| for i, chunk in enumerate(text_chunks): | |
| yield None, f"⏳ Đang xử lý đoạn {i+1}/{total_chunks}..." | |
| chunk_wav = tts.infer(chunk, ref_codes, ref_text_raw) | |
| if chunk_wav is not None and len(chunk_wav) > 0: | |
| all_audio_segments.append(chunk_wav) | |
| if i < total_chunks - 1: | |
| all_audio_segments.append(silence_pad) | |
| if not all_audio_segments: | |
| yield None, "❌ Không sinh được audio nào." | |
| return | |
| yield None, "💾 Đang ghép file và lưu..." | |
| final_wav = np.concatenate(all_audio_segments) | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp: | |
| sf.write(tmp.name, final_wav, sr) | |
| output_path = tmp.name | |
| process_time = time.time() - start_time | |
| speed_info = f", Tốc độ: {len(final_wav)/sr/process_time:.2f}x realtime" if process_time > 0 else "" | |
| yield output_path, f"✅ Hoàn tất! (Thời gian: {process_time:.2f}s{speed_info}) {backend_name}" | |
| # Cleanup memory | |
| if using_fast_backend and hasattr(tts, 'cleanup_memory'): | |
| tts.cleanup_memory() | |
| except torch.cuda.OutOfMemoryError as e: | |
| yield None, f"❌ GPU hết VRAM! Hãy thử giảm độ dài văn bản.\n\nChi tiết: {str(e)}" | |
| if torch.cuda.is_available(): | |
| torch.cuda.empty_cache() | |
| return | |
| except Exception as e: | |
| import traceback | |
| traceback.print_exc() | |
| yield None, f"❌ Lỗi: {str(e)}" | |
| return | |
| # === STREAMING MODE === | |
| else: | |
| sr = 24000 | |
| crossfade_samples = int(sr * 0.03) | |
| audio_queue = queue.Queue(maxsize=100) | |
| PRE_BUFFER_SIZE = 3 | |
| end_event = threading.Event() | |
| error_event = threading.Event() | |
| error_msg = "" | |
| def producer_thread(): | |
| nonlocal error_msg | |
| try: | |
| previous_tail = None | |
| for i, chunk_text in enumerate(text_chunks): | |
| stream_gen = tts.infer_stream(chunk_text, ref_codes, ref_text_raw) | |
| for part_idx, audio_part in enumerate(stream_gen): | |
| if audio_part is None or len(audio_part) == 0: | |
| continue | |
| if previous_tail is not None and len(previous_tail) > 0: | |
| overlap = min(len(previous_tail), len(audio_part), crossfade_samples) | |
| if overlap > 0: | |
| fade_out = np.linspace(1.0, 0.0, overlap, dtype=np.float32) | |
| fade_in = np.linspace(0.0, 1.0, overlap, dtype=np.float32) | |
| blended = (audio_part[:overlap] * fade_in + | |
| previous_tail[-overlap:] * fade_out) | |
| processed = np.concatenate([ | |
| previous_tail[:-overlap] if len(previous_tail) > overlap else np.array([]), | |
| blended, | |
| audio_part[overlap:] | |
| ]) | |
| else: | |
| processed = np.concatenate([previous_tail, audio_part]) | |
| tail_size = min(crossfade_samples, len(processed)) | |
| previous_tail = processed[-tail_size:].copy() | |
| output_chunk = processed[:-tail_size] if len(processed) > tail_size else processed | |
| else: | |
| tail_size = min(crossfade_samples, len(audio_part)) | |
| previous_tail = audio_part[-tail_size:].copy() | |
| output_chunk = audio_part[:-tail_size] if len(audio_part) > tail_size else audio_part | |
| if len(output_chunk) > 0: | |
| audio_queue.put((sr, output_chunk)) | |
| if previous_tail is not None and len(previous_tail) > 0: | |
| audio_queue.put((sr, previous_tail)) | |
| except Exception as e: | |
| import traceback | |
| traceback.print_exc() | |
| error_msg = str(e) | |
| error_event.set() | |
| finally: | |
| end_event.set() | |
| audio_queue.put(None) | |
| threading.Thread(target=producer_thread, daemon=True).start() | |
| yield (sr, np.zeros(int(sr * 0.05))), "📄 Đang buffering..." | |
| pre_buffer = [] | |
| while len(pre_buffer) < PRE_BUFFER_SIZE: | |
| try: | |
| item = audio_queue.get(timeout=5.0) | |
| if item is None: | |
| break | |
| pre_buffer.append(item) | |
| except queue.Empty: | |
| if error_event.is_set(): | |
| yield None, f"❌ Lỗi: {error_msg}" | |
| return | |
| break | |
| full_audio_buffer = [] | |
| backend_info = "🚀 LMDeploy" if using_fast_backend else "📦 Standard" | |
| for sr, audio_data in pre_buffer: | |
| full_audio_buffer.append(audio_data) | |
| yield (sr, audio_data), f"🔊 Đang phát ({backend_info})..." | |
| while True: | |
| try: | |
| item = audio_queue.get(timeout=0.05) | |
| if item is None: | |
| break | |
| sr, audio_data = item | |
| full_audio_buffer.append(audio_data) | |
| yield (sr, audio_data), f"🔊 Đang phát ({backend_info})..." | |
| except queue.Empty: | |
| if error_event.is_set(): | |
| yield None, f"❌ Lỗi: {error_msg}" | |
| break | |
| if end_event.is_set() and audio_queue.empty(): | |
| break | |
| continue | |
| if full_audio_buffer: | |
| final_wav = np.concatenate(full_audio_buffer) | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp: | |
| sf.write(tmp.name, final_wav, sr) | |
| yield tmp.name, f"✅ Hoàn tất Streaming! ({backend_info})" | |
| if using_fast_backend and hasattr(tts, 'cleanup_memory'): | |
| tts.cleanup_memory() | |
| # --- UI SETUP --- | |
| theme = gr.themes.Soft( | |
| primary_hue="indigo", | |
| secondary_hue="cyan", | |
| neutral_hue="slate", | |
| font=[gr.themes.GoogleFont('Inter'), 'ui-sans-serif', 'system-ui'], | |
| ).set( | |
| button_primary_background_fill="linear-gradient(90deg, #6366f1 0%, #0ea5e9 100%)", | |
| button_primary_background_fill_hover="linear-gradient(90deg, #4f46e5 0%, #0284c7 100%)", | |
| block_shadow="0 4px 6px -1px rgba(0, 0, 0, 0.1), 0 2px 4px -1px rgba(0, 0, 0, 0.06)", | |
| ) | |
| css = """ | |
| .container { max-width: 1200px; margin: auto; } | |
| .header-box { | |
| text-align: center; | |
| margin-bottom: 25px; | |
| padding: 25px; | |
| background: linear-gradient(135deg, #0f172a 0%, #1e293b 100%); | |
| border-radius: 12px; | |
| border: 1px solid #334155; | |
| box-shadow: 0 10px 15px -3px rgba(0, 0, 0, 0.3); | |
| } | |
| .header-title { | |
| font-size: 2.5rem; | |
| font-weight: 800; | |
| color: white; | |
| background: -webkit-linear-gradient(45deg, #60A5FA, #22D3EE); | |
| -webkit-background-clip: text; | |
| -webkit-text-fill-color: transparent; | |
| margin-bottom: 10px; | |
| } | |
| .header-desc { | |
| font-size: 1.1rem; | |
| color: #cbd5e1; | |
| margin-bottom: 15px; | |
| } | |
| .link-group a { | |
| text-decoration: none; | |
| margin: 0 10px; | |
| font-weight: 600; | |
| color: #94a3b8; | |
| transition: color 0.2s; | |
| } | |
| .link-group a:hover { color: #38bdf8; text-shadow: 0 0 5px rgba(56, 189, 248, 0.5); } | |
| .status-box { font-weight: bold; text-align: center; border: none; background: transparent; } | |
| """ | |
| EXAMPLES_LIST = [ | |
| ["Về miền Tây không chỉ để ngắm nhìn sông nước hữu tình, mà còn để cảm nhận tấm chân tình của người dân nơi đây. Cùng ngồi xuồng ba lá len lỏi qua rặng dừa nước, nghe câu vọng cổ ngọt ngào thì còn gì bằng.", "Vĩnh (nam miền Nam)"], | |
| ["Hà Nội những ngày vào thu mang một vẻ đẹp trầm mặc và cổ kính đến lạ thường. Đi dạo quanh Hồ Gươm vào sáng sớm, hít hà mùi hoa sữa nồng nàn và thưởng thức chút cốm làng Vòng là trải nghiệm khó quên.", "Bình (nam miền Bắc)"], | |
| ["Sự bùng nổ của trí tuệ nhân tạo đang định hình lại cách chúng ta làm việc và sinh sống. Từ xe tự lái đến trợ lý ảo thông minh, công nghệ đang dần xóa nhòa ranh giới giữa thực tại và những bộ phim viễn tưởng.", "Tuyên (nam miền Bắc)"], | |
| ["Ngày xửa ngày xưa, ở một ngôi làng nọ có cô Tấm xinh đẹp, nết na nhưng sớm mồ côi mẹ. Dù bị mẹ kế và Cám hãm hại đủ đường, Tấm vẫn giữ được tấm lòng lương thiện và cuối cùng tìm được hạnh phúc xứng đáng.", "Đoan (nữ miền Nam)"], | |
| ] | |
| with gr.Blocks(theme=theme, css=css, title="VieNeu-TTS Studio") as demo: | |
| with gr.Column(elem_classes="container"): | |
| # Header | |
| gr.HTML(""" | |
| <div class="header-box"> | |
| <div class="header-title">🎙️ VieNeu-TTS Studio</div> | |
| <div class="header-desc"> | |
| Phiên bản: VieNeu-TTS (GPU-optimized với LMDeploy) | |
| </div> | |
| <div class="link-group"> | |
| <a href="https://huggingface.co/pnnbao-ump/VieNeu-TTS" target="_blank">🤗 Model Card</a> • | |
| <a href="https://huggingface.co/datasets/pnnbao-ump/VieNeu-TTS-1000h" target="_blank">📖 Dataset 1000h</a> • | |
| <a href="https://github.com/pnnbao97/VieNeu-TTS" target="_blank">🦜 GitHub</a> | |
| </div> | |
| </div> | |
| """) | |
| # Status info | |
| gr.Markdown("ℹ️ **Model sẽ tự động tải khi bạn bắt đầu tổng hợp giọng nói (lần đầu sẽ mất ~30-60 giây)**") | |
| with gr.Row(elem_classes="container", equal_height=False): | |
| # --- LEFT: INPUT --- | |
| with gr.Column(scale=3, variant="panel"): | |
| gr.Markdown("### 📝 Văn bản đầu vào") | |
| text_input = gr.Textbox( | |
| label="Nhập văn bản", | |
| placeholder="Nhập nội dung tiếng Việt cần chuyển thành giọng nói...", | |
| lines=4, | |
| value="Sự bùng nổ của trí tuệ nhân tạo đang định hình lại cách chúng ta làm việc và sinh sống. Từ xe tự lái đến trợ lý ảo thông minh, công nghệ đang dần xóa nhòa ranh giới giữa thực tại và những bộ phim viễn tưởng.", | |
| show_label=False | |
| ) | |
| gr.Markdown("### 🗣️ Chọn giọng đọc") | |
| with gr.Tabs() as tabs: | |
| with gr.TabItem("👤 Giọng có sẵn (Preset)", id="preset_mode"): | |
| voice_select = gr.Dropdown( | |
| choices=list(VOICE_SAMPLES.keys()), | |
| value=list(VOICE_SAMPLES.keys())[0] if VOICE_SAMPLES else None, | |
| label="Danh sách giọng", | |
| interactive=True | |
| ) | |
| with gr.Accordion("Thông tin giọng mẫu", open=False): | |
| ref_audio_preview = gr.Audio(label="Audio mẫu", interactive=False, type="filepath") | |
| ref_text_preview = gr.Markdown("...") | |
| with gr.TabItem("🎙️ Giọng tùy chỉnh (Custom)", id="custom_mode"): | |
| gr.Markdown("Tải lên giọng của bạn (Zero-shot Cloning)") | |
| custom_audio = gr.Audio(label="File ghi âm (.wav)", type="filepath") | |
| custom_text = gr.Textbox(label="Nội dung ghi âm", placeholder="Nhập chính xác lời thoại...") | |
| gr.Markdown("### ⚙️ Cài đặt tổng hợp") | |
| generation_mode = gr.Radio( | |
| ["Standard (Một lần)", "Streaming (Thời gian thực)"], | |
| value="Standard (Một lần)", | |
| label="Chế độ sinh" | |
| ) | |
| use_batch = gr.Checkbox( | |
| value=True, | |
| label="⚡ Batch Processing (chỉ có hiệu lực khi dùng LMDeploy backend)", | |
| info="Xử lý nhiều đoạn cùng lúc để tăng tốc" | |
| ) | |
| current_mode = gr.Textbox(visible=False, value="preset_mode") | |
| btn_generate = gr.Button("Tổng hợp giọng nói", variant="primary", size="lg") | |
| # --- RIGHT: OUTPUT --- | |
| with gr.Column(scale=2): | |
| gr.Markdown("### 🎧 Kết quả") | |
| with gr.Group(): | |
| audio_output = gr.Audio(label="Audio đầu ra", type="filepath", autoplay=True) | |
| status_output = gr.Textbox(label="Trạng thái", placeholder="Sẵn sàng...") | |
| # --- EXAMPLES --- | |
| with gr.Row(elem_classes="container"): | |
| with gr.Column(): | |
| gr.Markdown("### 📚 Ví dụ mẫu") | |
| gr.Examples(examples=EXAMPLES_LIST, inputs=[text_input, voice_select], label="Thử nghiệm nhanh") | |
| # --- EVENT HANDLERS --- | |
| def update_ref_preview(voice): | |
| audio, text = load_reference_info(voice) | |
| return audio, f"> *\"{text}\"*" | |
| voice_select.change(update_ref_preview, voice_select, [ref_audio_preview, ref_text_preview]) | |
| demo.load(update_ref_preview, voice_select, [ref_audio_preview, ref_text_preview]) | |
| # Tab handling | |
| tabs.children[0].select(fn=lambda: "preset_mode", outputs=current_mode) | |
| tabs.children[1].select(fn=lambda: "custom_mode", outputs=current_mode) | |
| btn_generate.click( | |
| fn=synthesize_speech, | |
| inputs=[text_input, voice_select, custom_audio, custom_text, current_mode, generation_mode, use_batch], | |
| outputs=[audio_output, status_output] | |
| ) | |
| if __name__ == "__main__": | |
| demo.queue().launch( | |
| server_name="0.0.0.0", | |
| server_port=7860 | |
| ) |