Initial: WhisperX Gradio GUI with 5-min chunking
This commit is contained in:
@@ -0,0 +1,232 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
WhisperX Transcription GUI — Gradio app
|
||||
Real 5-min chunk splitting to avoid hallucination.
|
||||
No Gitea bloat — just transcribe & download.
|
||||
"""
|
||||
|
||||
import os, json, subprocess, tempfile, shutil, re, pathlib, uuid, io
|
||||
import gradio as gr
|
||||
|
||||
# ── Config ──
|
||||
WHISPERX = os.path.expanduser("~/.local/bin/whisperx")
|
||||
OUTPUT_DIR = os.path.expanduser("~/workspace/transcripts/gradio_uploads")
|
||||
CHUNK_SECS = 300 # 5 minutes
|
||||
os.makedirs(OUTPUT_DIR, exist_ok=True)
|
||||
|
||||
LANGUAGES = [
|
||||
("French", "fr"), ("Arabic / Darija", "ar"), ("English", "en"),
|
||||
("German", "de"), ("Spanish", "es"), ("Italian", "it"),
|
||||
("Portuguese", "pt"), ("Dutch", "nl"), ("Japanese", "ja"),
|
||||
("Chinese", "zh"), ("Russian", "ru"), ("Turkish", "tr"),
|
||||
("Polish", "pl"), ("Swedish", "sv"), ("Norwegian", "no"),
|
||||
("Danish", "da"), ("Finnish", "fi"), ("Czech", "cs"),
|
||||
("Romanian", "ro"), ("Hungarian", "hu"), ("Greek", "el"),
|
||||
("Hebrew", "he"), ("Hindi", "hi"), ("Korean", "ko"),
|
||||
("Thai", "th"), ("Vietnamese", "vi"), ("Indonesian", "id"),
|
||||
("Malay", "ms"), ("Auto-detect", ""),
|
||||
]
|
||||
|
||||
MODELS = [
|
||||
("large-v3 (best, ~8GB VRAM)", "large-v3"),
|
||||
("large-v2 (~8GB VRAM)", "large-v2"),
|
||||
("medium (~5GB VRAM)", "medium"),
|
||||
("small (~2GB VRAM)", "small"),
|
||||
("base (~1GB VRAM)", "base"),
|
||||
("tiny (~0.3GB VRAM)", "tiny"),
|
||||
]
|
||||
|
||||
|
||||
def fmt_ts(secs: float) -> str:
|
||||
h = int(secs // 3600)
|
||||
m = int((secs % 3600) // 60)
|
||||
s = secs % 60
|
||||
if h:
|
||||
return f"{h:02d}:{m:02d}:{s:06.3f}"
|
||||
return f"{m:02d}:{s:06.3f}"
|
||||
|
||||
|
||||
def fmt_srt(secs: float) -> str:
|
||||
h = int(secs // 3600)
|
||||
m = int((secs % 3600) // 60)
|
||||
s = int(secs % 60)
|
||||
ms = int((secs - int(secs)) * 1000)
|
||||
return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
|
||||
|
||||
|
||||
def transcribe(audio_path, lang, model):
|
||||
"""Run WhisperX on one file. Returns {segments, error, out_dir}."""
|
||||
out_dir = pathlib.Path(OUTPUT_DIR) / f"{pathlib.Path(audio_path).stem}_{uuid.uuid4().hex[:8]}"
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
cmd = [
|
||||
WHISPERX, audio_path,
|
||||
"--model", model,
|
||||
"--task", "transcribe",
|
||||
"--device", "cuda",
|
||||
"--compute_type", "float16",
|
||||
"--batch_size", "4",
|
||||
"--chunk_size", "30",
|
||||
"--vad_method", "pyannote",
|
||||
"--condition_on_previous_text", "False",
|
||||
"--temperature", "0",
|
||||
"--compression_ratio_threshold", "2.4",
|
||||
"--logprob_threshold", "-1.0",
|
||||
"--no_speech_threshold", "0.6",
|
||||
"--output_dir", str(out_dir),
|
||||
"--output_format", "all",
|
||||
]
|
||||
if lang:
|
||||
cmd.extend(["--language", lang])
|
||||
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=1800)
|
||||
if result.returncode != 0:
|
||||
return {"error": f"WhisperX failed:\n{result.stderr[:2000]}", "segments": [], "out_dir": str(out_dir)}
|
||||
|
||||
# Parse JSON output for segments
|
||||
json_files = list(out_dir.glob("*.json"))
|
||||
segments = []
|
||||
if json_files:
|
||||
with open(json_files[0], encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
segments = data.get("segments", [])
|
||||
|
||||
return {"error": None, "segments": segments, "out_dir": str(out_dir)}
|
||||
|
||||
|
||||
# ── Gradio function ──
|
||||
def do_transcribe(audio_file, lang_code, model_code):
|
||||
if audio_file is None:
|
||||
yield "", gr.Dropdown(), gr.Dropdown(), "⚠️ Upload an audio file first."
|
||||
return
|
||||
|
||||
yield "", gr.Dropdown(), gr.Dropdown(), "⏳ Converting audio to WAV..."
|
||||
|
||||
# Convert to 16kHz mono WAV
|
||||
temp_dir = tempfile.mkdtemp()
|
||||
wav_path = os.path.join(temp_dir, "full.wav")
|
||||
subprocess.run(
|
||||
["ffmpeg", "-y", "-i", audio_file, "-ar", "16000", "-ac", "1", wav_path],
|
||||
capture_output=True, timeout=120,
|
||||
)
|
||||
|
||||
# Get duration
|
||||
probe = subprocess.run(
|
||||
["ffprobe", "-v", "error", "-show_entries", "format=duration",
|
||||
"-of", "csv=p=0", wav_path],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
total_secs = float(probe.stdout.strip() or 0)
|
||||
|
||||
# Split into 5-min chunks
|
||||
yield "", gr.Dropdown(), gr.Dropdown(), f"⏳ Splitting {fmt_ts(total_secs)} into {int(total_secs // CHUNK_SECS) + 1} chunks..."
|
||||
chunks_dir = os.path.join(temp_dir, "chunks")
|
||||
os.makedirs(chunks_dir, exist_ok=True)
|
||||
subprocess.run(
|
||||
["ffmpeg", "-y", "-i", wav_path,
|
||||
"-f", "segment", "-segment_time", str(CHUNK_SECS),
|
||||
"-reset_timestamps", "1", "-c", "copy",
|
||||
f"{chunks_dir}/chunk_%03d.wav"],
|
||||
capture_output=True, timeout=600,
|
||||
)
|
||||
|
||||
chunk_files = sorted(pathlib.Path(chunks_dir).glob("chunk_*.wav"))
|
||||
if not chunk_files:
|
||||
# File shorter than 5 min — use original
|
||||
chunk_files = [pathlib.Path(wav_path)]
|
||||
|
||||
n_chunks = len(chunk_files)
|
||||
all_segments = []
|
||||
|
||||
for idx, chunk_path in enumerate(chunk_files):
|
||||
yield "", gr.Dropdown(), gr.Dropdown(), (
|
||||
f"⏳ Transcribing chunk {idx + 1}/{n_chunks}..."
|
||||
)
|
||||
|
||||
result = transcribe(str(chunk_path), lang_code, model_code)
|
||||
if result["error"]:
|
||||
yield "", gr.Dropdown(), gr.Dropdown(), f"❌ Chunk {idx + 1} failed: {result['error']}"
|
||||
shutil.rmtree(temp_dir, ignore_errors=True)
|
||||
return
|
||||
|
||||
offset = idx * CHUNK_SECS
|
||||
for seg in result["segments"]:
|
||||
seg["start"] += offset
|
||||
seg["end"] += offset
|
||||
all_segments.append(seg)
|
||||
|
||||
all_segments.sort(key=lambda s: s["start"])
|
||||
shutil.rmtree(temp_dir, ignore_errors=True)
|
||||
|
||||
# Build outputs
|
||||
dur = all_segments[-1]["end"] if all_segments else 0
|
||||
|
||||
# Timestamped TXT
|
||||
txt_lines = []
|
||||
for seg in all_segments:
|
||||
txt_lines.append(f"[{fmt_ts(seg['start'])} --> {fmt_ts(seg['end'])}] {seg['text'].strip()}")
|
||||
txt_content = "\n".join(txt_lines)
|
||||
|
||||
# SRT
|
||||
srt_lines = []
|
||||
for i, seg in enumerate(all_segments, 1):
|
||||
srt_lines.append(f"{i}")
|
||||
srt_lines.append(f"{fmt_srt(seg['start'])} --> {fmt_srt(seg['end'])}")
|
||||
srt_lines.append(seg['text'].strip())
|
||||
srt_lines.append("")
|
||||
srt_content = "\n".join(srt_lines)
|
||||
|
||||
# Write files
|
||||
session_dir = pathlib.Path(OUTPUT_DIR) / f"transcript_{uuid.uuid4().hex[:8]}"
|
||||
session_dir.mkdir(parents=True, exist_ok=True)
|
||||
txt_path = str(session_dir / "transcript.txt")
|
||||
srt_path = str(session_dir / "transcript.srt")
|
||||
with open(txt_path, "w") as f:
|
||||
f.write(txt_content)
|
||||
with open(srt_path, "w") as f:
|
||||
f.write(srt_content)
|
||||
|
||||
status = (
|
||||
f"✅ **Done** — {lang_code.upper() if lang_code else 'Auto'} | {model_code} | "
|
||||
f"{fmt_ts(dur)} | {len(all_segments)} segments | {n_chunks} chunk{'s' if n_chunks > 1 else ''}"
|
||||
)
|
||||
|
||||
yield txt_content, gr.Dropdown(value=txt_path, visible=True), gr.Dropdown(value=srt_path, visible=True), status
|
||||
|
||||
|
||||
# ── UI ──
|
||||
with gr.Blocks(title="WhisperX Transcription") as app:
|
||||
gr.Markdown(
|
||||
"# 🎙️ WhisperX Transcription\n"
|
||||
"Upload audio → auto-splits into **5-minute chunks** (anti-hallucination) → downloads TXT + SRT."
|
||||
)
|
||||
|
||||
with gr.Row():
|
||||
with gr.Column(scale=1):
|
||||
audio_input = gr.Audio(type="filepath", label="Upload Audio", sources=["upload", "microphone"])
|
||||
lang_dropdown = gr.Dropdown(choices=LANGUAGES, value="fr", label="Language", info="Auto-detect if unsure")
|
||||
model_dropdown = gr.Dropdown(choices=MODELS, value="large-v3", label="Model Size")
|
||||
transcribe_btn = gr.Button("🚀 Transcribe", variant="primary", size="lg")
|
||||
|
||||
with gr.Column(scale=2):
|
||||
status_text = gr.Markdown("Ready to transcribe.")
|
||||
output_preview = gr.Textbox(label="Transcript Preview", lines=15, max_lines=30, interactive=False)
|
||||
with gr.Row():
|
||||
txt_download = gr.File(label="📄 Download TXT", visible=False)
|
||||
srt_download = gr.File(label="📝 Download SRT", visible=False)
|
||||
|
||||
transcribe_btn.click(
|
||||
fn=do_transcribe,
|
||||
inputs=[audio_input, lang_dropdown, model_dropdown],
|
||||
outputs=[output_preview, txt_download, srt_download, status_text],
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--host", default="127.0.0.1")
|
||||
parser.add_argument("--port", type=int, default=7860)
|
||||
args = parser.parse_args()
|
||||
print(f"🚀 WhisperX Gradio on http://{args.host}:{args.port}")
|
||||
app.launch(server_name=args.host, server_port=args.port, theme=gr.themes.Soft())
|
||||
Reference in New Issue
Block a user