#!/usr/bin/env python3 """VoiceForge Studio — Professional TTS & ASR Platform""" import os, json, time, tempfile, warnings, traceback from pathlib import Path from typing import List, Dict, Any import gradio as gr import numpy as np import soundfile as sf import torch import requests warnings.filterwarnings("ignore") HF_TOKEN = os.environ.get("HF_TOKEN", "") DEVICE = "cuda:0" if torch.cuda.is_available() else "cpu" DTYPE = torch.float16 if DEVICE == "cuda:0" else torch.float32 TEMP_DIR = Path(tempfile.gettempdir()) / "voiceforge" TEMP_DIR.mkdir(exist_ok=True) HF_API = "https://api-inference.huggingface.co/models" TTS_MODELS = { "Qwen3-TTS 1.7B (Voice Design + Clone)": "Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign", "VibeVoice 1.5B (Long-form)": "microsoft/VibeVoice-1.5B", "VibeVoice Realtime 0.5B": "microsoft/VibeVoice-Realtime-0.5B", "Voxtral 4B TTS": "mistralai/Voxtral-4B-TTS-2603", "IndexTTS-2 (Emotion)": "IndexTeam/IndexTTS-2", "OmniVoice": "zardus-ai/omnivoice-tts", } ASR_MODELS = { "Whisper Large v3 Turbo (Word Timestamps)": "openai/whisper-large-v3-turbo", "Qwen3-ASR 1.7B (Line Timestamps)": "Qwen/Qwen3-ASR-1.7B", "Qwen3-ASR 0.6B (Fast)": "Qwen/Qwen3-ASR-0.6B", "VibeVoice-ASR (Diarization)": "microsoft/VibeVoice-ASR", } # Smaller TTS models that work via Inference API fallback TTS_FALLBACKS = [ "suno/bark", "facebook/mms-tts-eng", ] _MODEL_CACHE: Dict[str, Any] = {} _AVAIL = {} def check_avail(): global _AVAIL _AVAIL["cuda"] = torch.cuda.is_available() _AVAIL["gpu"] = torch.cuda.get_device_name(0) if _AVAIL["cuda"] else "N/A" _AVAIL["hf_token"] = bool(HF_TOKEN) try: import transformers _AVAIL["transformers"] = True except: _AVAIL["transformers"] = False return _AVAIL def fmt_ts(s): if s is None: return "--:--:--.---" return f"{int(s//3600):02d}:{int((s%3600)//60):02d}:{s%60:06.3f}" def spk_color(sid): c={"SPEAKER_00":"#FF6B6B","SPEAKER_01":"#4ECDC4","SPEAKER_02":"#45B7D1","SPEAKER_03":"#96CEB4","SPEAKER_04":"#FFEAA7","SPEAKER_05":"#DDA0DD"} return c.get(sid,"#BDC3C7") def merge_dia(chunks, dia): if dia is None or not chunks: return chunks merged=[] for ch in chunks: st=ch.get("start",0) or 0; en=ch.get("end",st) or st spk="UNKNOWN"; mo=0 try: for turn,_,s in dia.itertracks(yield_label=True): ov=max(0,min(en,turn.end)-max(st,turn.start)) if ov>mo: mo=ov; spk=s except: pass merged.append({**ch,"speaker":spk}) return merged def render_html(merged): if not merged: return "

No transcription.

" h="""
""" for it in merged: spk=it.get("speaker","UNKNOWN"); txt=it.get("text",""); s=it.get("start",0); e=it.get("end",0); col=spk_color(spk) h+=f"""
{spk}[{fmt_ts(s)} → {fmt_ts(e)}]

{txt}

""" return h+"
" def render_txt(merged,fmt="full"): if not merged: return "" lines=[] for it in merged: spk=it.get("speaker","UNKNOWN"); txt=it.get("text",""); s=it.get("start",0); e=it.get("end",0) if fmt=="full": lines.append(f"[{fmt_ts(s)} → {fmt_ts(e)}] {spk}: {txt}") elif fmt=="speaker_only": lines.append(f"{spk}: {txt}") elif fmt=="text_only": lines.append(txt) elif fmt=="srt": i=len(lines)+1; ss=fmt_ts(s).replace(".",","); ee=fmt_ts(e).replace(".",","); lines.append(f"{i}\n{ss} --> {ee}\n{spk}: {txt}\n") return "\n".join(lines) # ── Whisper ASR ── def load_whisper(): k="whisper" if k in _MODEL_CACHE: return _MODEL_CACHE[k] from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline m=AutoModelForSpeechSeq2Seq.from_pretrained("openai/whisper-large-v3-turbo",torch_dtype=DTYPE,low_cpu_mem_usage=True,use_safetensors=True) m.to(DEVICE) p=AutoProcessor.from_pretrained("openai/whisper-large-v3-turbo") pipe=pipeline("automatic-speech-recognition",model=m,tokenizer=p.tokenizer,feature_extractor=p.feature_extractor,torch_dtype=DTYPE,device=DEVICE) _MODEL_CACHE[k]=pipe return pipe def transcribe_whisper(path,ts="word",lang=None): pipe=load_whisper() gk={} if lang and lang!="auto": gk["language"]=lang.lower() ts_set="word" if ts=="word" else (True if ts=="chunk" else False) r=pipe(path,return_timestamps=ts_set,generate_kwargs=gk) chunks=[] for ch in r.get("chunks",[]): t=ch.get("timestamp",(0,0)) if t is None or t==(None,None): t=(0,0) s=t[0] if t[0] is not None else 0 e=t[1] if len(t)>1 and t[1] is not None else s chunks.append({"text":ch.get("text",""),"start":s,"end":e}) return r.get("text",""),chunks # ── Diarization (optional) ── def load_dia(): k="dia" if k in _MODEL_CACHE: return _MODEL_CACHE[k] from pyannote.audio import Pipeline p=Pipeline.from_pretrained("pyannote/speaker-diarization-3.1",use_auth_token=HF_TOKEN) if DEVICE=="cuda:0": p.to(torch.device("cuda")) _MODEL_CACHE[k]=p return p def run_dia(path,ns=None,mn=None,mx=None): try: p=load_dia(); kw={} if ns: kw["num_speakers"]=int(ns) if mn: kw["min_speakers"]=int(mn) if mx: kw["max_speakers"]=int(mx) return p(path,**kw) except Exception as e: print(f"[dia err] {e}"); return None # ── TTS via HF Inference API ── def hf_tts(model_id,text): if not HF_TOKEN: return None,"HF_TOKEN not set" try: r=requests.post(f"{HF_API}/{model_id}",headers={"Authorization":f"Bearer {HF_TOKEN}"},json={"inputs":text},timeout=120) if r.status_code==200: with tempfile.NamedTemporaryFile(suffix=".wav",delete=False) as f: f.write(r.content); d,sr=sf.read(f.name,dtype="float32"); return d,sr return None,f"API {r.status_code}: {r.text[:200]}" except Exception as e: return None,str(e) def do_tts(model,mode,text,lang,vd,ref_a,ref_t): if not text or not text.strip(): return None,"❌ Enter text.",{} st=time.time() mid=TTS_MODELS.get(model) errors=[] # Try requested model audio,sr=hf_tts(mid,text) if audio is not None: dur=len(audio)/sr; el=time.time()-st info={"model":model,"mode":mode,"lang":lang,"dur_sec":round(dur,2),"sr":sr,"gen_sec":round(el,2),"rtf":round(el/dur,3) if dur else 0,"device":DEVICE,"source":"HF Inference API"} return (sr,audio),f"✅ {dur:.1f}s in {el:.1f}s (RTF={el/dur:.2f})" if dur else "✅ Done",info errors.append(f"Primary model ({model}): API unavailable") # Try fallbacks for fb in TTS_FALLBACKS: audio,sr=hf_tts(fb,text) if audio is not None: dur=len(audio)/sr; el=time.time()-st info={"model":model,"fallback":fb,"mode":mode,"lang":lang,"dur_sec":round(dur,2),"sr":sr,"gen_sec":round(el,2),"rtf":round(el/dur,3) if dur else 0,"device":DEVICE,"source":"HF Inference API (fallback)"} return (sr,audio),f"✅ Fallback {fb}: {dur:.1f}s in {el:.1f}s",info errors.append(f"Fallback {fb}: unavailable") # All failed err_msg="\n".join(errors) return None,f"❌ TTS unavailable on free tier.\n{err_msg}\n\n💡 Upgrade to a GPU Space for local inference with voice cloning.",{"errors":errors} def do_asr(model,path,lang,ts_mode,en_dia,ns,mn,mx,out_fmt): if path is None: return "","

❌ Upload audio.

",{} lg=None if lang=="Auto-detect" else lang.lower() ts="word" if ts_mode=="Word-level" else ("chunk" if ts_mode=="Line-level (chunk)" else False) st=time.time() try: txt,chunks=transcribe_whisper(path,ts,lg) dia=None if en_dia: try: dia=run_dia(path,ns,mn,mx) except Exception as e: print(f"[dia] {e}") merged=merge_dia(chunks,dia) fm={"Full (with timestamps & speakers)":"full","Speaker only":"speaker_only","Text only":"text_only","SRT subtitles":"srt"} text_out=render_txt(merged,fm.get(out_fmt,"full")) html_out=render_html(merged) el=time.time()-st ad=0 try: ad=sf.info(path).duration except: pass spks=set(it.get("speaker","UNKNOWN") for it in merged) info={"model":model,"lang":lang,"audio_sec":round(ad,2),"trans_sec":round(el,2),"rtf":round(el/ad,3) if ad else 0,"segments":len(chunks),"speakers":len(spks),"spk_list":sorted(list(spks)),"ts":ts_mode,"dia":"pyannote" if dia else "None","device":DEVICE} return text_out,html_out,info except Exception as e: tb=traceback.format_exc() return "",f"

❌ {str(e)}

{tb[:2000]}
",{} def do_batch(files,model,lang,dia): if not files: return [],None lg=None if lang=="Auto-detect" else lang.lower() res=[]; data=[] for f in files: try: fp=f.name if hasattr(f,'name') else str(f) txt,chunks=transcribe_whisper(fp,"chunk",lg) d=None if dia: try: d=run_dia(fp) except: pass mg=merge_dia(chunks,d) spks=set(it.get("speaker","UNKNOWN") for it in mg) dur=0 try: dur=sf.info(fp).duration except: pass res.append([Path(fp).name,f"{dur:.1f}s",str(len(spks)),txt[:100]+"..." if len(txt)>100 else txt,"✅"]) data.append({"file":Path(fp).name,"txt":txt,"segs":mg,"spks":sorted(list(spks)),"dur":dur}) except Exception as e: res.append([str(f),"?","?",str(e)[:100],"❌"]) jp=str(TEMP_DIR/"batch.json") with open(jp,"w") as f: json.dump(data,f,indent=2,default=str) return res,jp # ── UI ── def build_ui(): check_avail() badges=[] if _AVAIL.get("transformers"): badges.append("✅ transformers") if _AVAIL.get("cuda"): badges.append(f"🎮 {_AVAIL.get('gpu','GPU')}") else: badges.append("💻 CPU") if _AVAIL.get("hf_token"): badges.append("🔑 HF Token") css=""".vf-header{text-align:center;padding:32px 20px;background:linear-gradient(135deg,#6366f1,#8b5cf6,#f472b6);border-radius:16px;margin-bottom:24px;color:white;}.vf-header h1{font-size:2.5em;font-weight:800;margin:0;letter-spacing:-1px;}.vf-header p{font-size:1.1em;opacity:0.9;margin:8px 0 0 0;}.vf-info{background:rgba(99,102,241,0.08)!important;border-left:3px solid #6366f1!important;border-radius:0 8px 8px 0!important;padding:12px 16px!important;margin:8px 0!important;color:#e2e8f0!important;}""" with gr.Blocks(css=css,theme=gr.themes.Base(primary_hue="indigo",secondary_hue="violet",neutral_hue="slate").set(body_background_fill="*neutral_950",body_text_color="*neutral_100",background_fill_primary="*neutral_900",background_fill_secondary="*neutral_800",block_background_fill="*neutral_800",block_border_color="*neutral_700",input_background_fill="*neutral_800",button_primary_background_fill="*primary_600",button_primary_text_color="white"),title="VoiceForge Studio",analytics_enabled=False) as app: gr.HTML("""

🎙️ VoiceForge Studio

Professional TTS & ASR — Voice Cloning · Timestamps · Diarization · Emotion Control

""") with gr.Row(): gr.HTML(f"""
{' · '.join(badges)}
""") with gr.Tab("🗣️ Text-to-Speech"): gr.Markdown("## Generate Natural Speech with Voice Cloning & Emotion Control") with gr.Row(): with gr.Column(scale=1): tts_model=gr.Dropdown(choices=list(TTS_MODELS.keys()),value=list(TTS_MODELS.keys())[0],label="🎛️ TTS Model") tts_info=gr.HTML("""
Qwen3-TTS 1.7B — Voice Design & Voice Clone. 12 languages. Requires GPU for local inference.
""") tts_mode=gr.Radio(choices=["Voice Design","Voice Clone"],value="Voice Design",label="🎯 Mode") tts_text=gr.Textbox(label="📝 Text to Speak",lines=4,value="Hello! Welcome to VoiceForge Studio.") tts_lang=gr.Dropdown(choices=["English","Chinese","Spanish","French","German","Japanese","Korean","Arabic","Hindi","Portuguese","Italian","Russian"],value="English",label="🌐 Language") with gr.Group(visible=True) as vd_g: voice_desc=gr.Textbox(label="🎨 Voice Description",placeholder="e.g., warm deep male voice",value="A natural, clear speaking voice.",lines=2) with gr.Group(visible=False) as vc_g: ref_audio=gr.Audio(label="🎤 Reference Audio",type="filepath",sources=["upload","microphone"]) ref_text=gr.Textbox(label="📄 Reference Transcript",placeholder="Text of reference audio",lines=2) with gr.Group(visible=False) as emo_g: gr.Markdown("### 🎭 Emotion Control") emo_mode=gr.Radio(choices=["Neutral","Emotion Vector","Emotion Audio","Text-Guided"],value="Neutral",label="Emotion Mode") with gr.Row(): emo_h=gr.Slider(0,1,0,0.1,label="😊 Happy") emo_s=gr.Slider(0,1,0,0.1,label="😢 Sad") with gr.Row(): emo_a=gr.Slider(0,1,0,0.1,label="😠 Angry") emo_f=gr.Slider(0,1,0,0.1,label="😨 Fear") tts_btn=gr.Button("🚀 Generate Speech",variant="primary") with gr.Column(scale=1): tts_out_a=gr.Audio(label="🔊 Generated Audio",type="numpy",autoplay=False) tts_out_s=gr.Textbox(label="📊 Status",value="Ready — TTS uses HF Inference API or requires GPU upgrade for local models.",interactive=False) tts_out_i=gr.JSON(label="ℹ️ Info",value={}) def sw_tts_mode(m,md): ic=m=="Voice Clone"; ie="IndexTTS" in md; return {vd_g:gr.Group(visible=not ic),vc_g:gr.Group(visible=ic),emo_g:gr.Group(visible=ie)} tts_mode.change(sw_tts_mode,inputs=[tts_mode,tts_model],outputs=[vd_g,vc_g,emo_g]) def sw_tts_model(m): im={"Qwen3-TTS 1.7B (Voice Design + Clone)":"""
Qwen3-TTS 1.7B — Voice Design & Clone. 12 languages. 12kHz. Requires GPU Space for local inference.
""","VibeVoice 1.5B (Long-form)":"""
VibeVoice 1.5B — Long-form up to 90min, 4 speakers. Requires GPU.
""","VibeVoice Realtime 0.5B":"""
VibeVoice Realtime — Streaming, low latency. Requires GPU.
""","Voxtral 4B TTS":"""
Voxtral 4B — vLLM, OpenAI-compatible. Requires GPU.
""","IndexTTS-2 (Emotion)":"""
IndexTTS-2 — 8-dimension emotion control! Requires GPU.
""","OmniVoice":"""
OmniVoice — OpenAI-compatible API. Requires GPU.
"""} return {tts_info:im.get(m,""),emo_g:gr.Group(visible="IndexTTS" in m)} tts_model.change(sw_tts_model,inputs=tts_model,outputs=[tts_info,emo_g]) tts_btn.click(do_tts,inputs=[tts_model,tts_mode,tts_text,tts_lang,voice_desc,ref_audio,ref_text],outputs=[tts_out_a,tts_out_s,tts_out_i]) with gr.Tab("🎙️ Speech-to-Text (ASR)"): gr.Markdown("## Transcribe Audio with Timestamps, Diarization & Speaker ID") with gr.Row(): with gr.Column(scale=1): asr_model=gr.Dropdown(choices=list(ASR_MODELS.keys()),value=list(ASR_MODELS.keys())[0],label="🎛️ ASR Model") asr_info=gr.HTML("""
Whisper Large v3 Turbo — Fast word-level timestamps. 99 languages. Runs locally on CPU or GPU.
""") asr_audio=gr.Audio(label="🎤 Upload Audio",type="filepath",sources=["upload","microphone"]) asr_lang=gr.Dropdown(choices=["Auto-detect","English","Chinese","Spanish","French","German","Japanese","Korean","Arabic","Hindi","Portuguese","Italian","Russian"],value="Auto-detect",label="🌐 Language") ts_mode=gr.Radio(choices=["Word-level","Line-level (chunk)","None"],value="Word-level",label="⏱️ Timestamp Granularity") with gr.Accordion("🔊 Speaker Diarization (pyannote 3.1)",open=False): en_dia=gr.Checkbox(label="Enable Diarization",value=False) ns=gr.Number(label="Number of Speakers (opt)",value=None,minimum=1,maximum=20,precision=0) mn=gr.Number(label="Min Speakers",value=None,minimum=1,maximum=20,precision=0) mx=gr.Number(label="Max Speakers",value=None,minimum=1,maximum=20,precision=0) with gr.Accordion("⚙️ Advanced Options",open=False): out_fmt=gr.Dropdown(choices=["Full (with timestamps & speakers)","Speaker only","Text only","SRT subtitles"],value="Full (with timestamps & speakers)",label="Output Format") asr_btn=gr.Button("🚀 Transcribe",variant="primary") with gr.Column(scale=1): asr_out_t=gr.Textbox(label="📝 Transcription",lines=10) asr_out_h=gr.HTML(label="🎨 Visual Transcript") asr_out_i=gr.JSON(label="ℹ️ Info",value={}) def sw_asr(m): im={"Whisper Large v3 Turbo (Word Timestamps)":"""
Whisper Turbo — Word timestamps, 99 langs. Runs locally.
""","Qwen3-ASR 1.7B (Line Timestamps)":"""
Qwen3-ASR 1.7B — Forced alignment, 30+ langs. Requires GPU.
""","Qwen3-ASR 0.6B (Fast)":"""
Qwen3-ASR 0.6B — Lightweight, fast. Requires GPU.
""","VibeVoice-ASR (Diarization)":"""
VibeVoice-ASR — Native diarization + timestamps. Requires GPU.
"""} return im.get(m,"") asr_model.change(sw_asr,inputs=asr_model,outputs=asr_info) asr_btn.click(do_asr,inputs=[asr_model,asr_audio,asr_lang,ts_mode,en_dia,ns,mn,mx,out_fmt],outputs=[asr_out_t,asr_out_h,asr_out_i]) with gr.Tab("📦 Batch Processing"): gr.Markdown("## Process Multiple Files") with gr.Row(): with gr.Column(): bf=gr.File(label="📁 Upload Audio Files",file_count="multiple",file_types=[".wav",".mp3",".flac",".m4a",".ogg"]) bm=gr.Dropdown(choices=list(ASR_MODELS.keys()),value=list(ASR_MODELS.keys())[0],label="🎛️ ASR Model") bl=gr.Dropdown(choices=["Auto-detect","English","Chinese","Spanish","French"],value="Auto-detect",label="🌐 Language") bd=gr.Checkbox(label="Enable Diarization",value=False) bb=gr.Button("🚀 Process Batch",variant="primary") with gr.Column(): br=gr.Dataframe(headers=["File","Duration","Speakers","Text Preview","Status"],label="📊 Results",wrap=True) bdwn=gr.File(label="📥 Download Results (JSON)") bb.click(do_batch,inputs=[bf,bm,bl,bd],outputs=[br,bdwn]) with gr.Tab("ℹ️ About & Models"): gr.Markdown(f""" ## 🎙️ VoiceForge Studio Professional TTS & ASR platform. ### TTS Models | Model | Features | Languages | Local? | |-------|----------|-----------|--------| | **Qwen3-TTS 1.7B** | Voice Design + Clone | 12 | ❌ GPU needed | | **VibeVoice 1.5B** | Long-form, 4 speakers | 50+ | ❌ GPU needed | | **VibeVoice Realtime** | Streaming | 50+ | ❌ GPU needed | | **Voxtral 4B** | vLLM, OpenAI-compatible | 9 | ❌ GPU needed | | **IndexTTS-2** | Emotion control | EN, ZH | ❌ GPU needed | | **OmniVoice** | OpenAI-compatible API | Multi | ❌ GPU needed | ### ASR Models | Model | Features | Languages | Local? | |-------|----------|-----------|--------| | **Whisper Turbo** | Word timestamps | 99 | ✅ Runs on CPU/GPU | | **Qwen3-ASR 1.7B** | Forced alignment | 30+ | ❌ GPU needed | | **Qwen3-ASR 0.6B** | Fast | 30+ | ❌ GPU needed | | **VibeVoice-ASR** | Native diarization | 50+ | ❌ GPU needed | ### Device - **{DEVICE.upper()}** — {_AVAIL.get('gpu','N/A')} - Badges: {' · '.join(badges)} ### 💡 Recommendation For full voice cloning TTS with all models, upgrade this Space to a **GPU tier** (e.g., NVIDIA A10G or T4). On the free CPU tier, ASR (Whisper) works perfectly. TTS large models require GPU or HF Inference API Pro. Built with ❤️ using Gradio, Transformers, PyTorch, Hugging Face """) return app if __name__=="__main__": app=build_ui() app.launch(server_name="0.0.0.0",server_port=int(os.environ.get("GRADIO_SERVER_PORT",7860)),share=False,show_error=True)