import tempfile
import traceback
from pathlib import Path
import gradio as gr
from pipeline import get_pipeline, to_json, to_csv, to_abab_text
UPLOAD_DIR = Path(tempfile.gettempdir()) / "speech_annotation"
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
_last_segments = []
SPEAKER_COLORS = [
("#4F46E5", "#EEF2FF"),
("#059669", "#ECFDF5"),
("#DC2626", "#FEF2F2"),
("#D97706", "#FFFBEB"),
("#7C3AED", "#F5F3FF"),
("#0891B2", "#ECFEFF"),
("#DB2777", "#FDF2F8"),
("#65A30D", "#F7FEE7"),
("#EA580C", "#FFF7ED"),
("#0284C7", "#F0F9FF"),
]
def make_conversation_html(segments):
if not segments:
return ""
speaker_list = list(dict.fromkeys(s.speaker for s in segments))
color_map = {spk: SPEAKER_COLORS[i % len(SPEAKER_COLORS)] for i, spk in enumerate(speaker_list)}
legend_items = "".join(
f""
f""
f"Speaker {spk}"
for spk in speaker_list
)
legend = f"
{legend_items}
"
bubbles = ""
for seg in segments:
fg, bg = color_map[seg.speaker]
align = "flex-end" if speaker_list.index(seg.speaker) % 2 == 1 else "flex-start"
text_align = "text-align:right;" if align == "flex-end" else ""
radius = "4px 16px 16px 16px" if align == "flex-start" else "16px 4px 16px 16px"
bubbles += f"""
Speaker {seg.speaker}
· {seg.start_fmt} → {seg.end_fmt}
{seg.text}
"""
return f"""
"""
def make_table_html(segments):
if not segments:
return ""
speaker_list = list(dict.fromkeys(s.speaker for s in segments))
color_map = {spk: SPEAKER_COLORS[i % len(SPEAKER_COLORS)] for i, spk in enumerate(speaker_list)}
rows = "".join(
f""
f"| {s.speaker} | "
f"{s.start_fmt} | "
f"{s.end_fmt} | "
f"{s.text} | "
f"
"
for s in segments
)
return f"""
| Speaker |
Start |
End |
Transcript |
{rows}
"""
def process_audio(audio_path, num_speakers):
global _last_segments
if audio_path is None:
return "⚠️ Please upload an audio file first.", "", ""
try:
pipeline = get_pipeline()
n = int(num_speakers) if num_speakers and int(num_speakers) > 0 else 0
segments = pipeline.process(audio_path, num_speakers=n)
except Exception as e:
return f"❌ Error: {e}\n{traceback.format_exc()}", "", ""
if not segments:
return "⚠️ No speech detected.", "", ""
_last_segments = segments
unique = len(set(s.speaker for s in segments))
status = f"✅ Done — {len(segments)} segments · {unique} speaker(s) detected"
return status, make_conversation_html(segments), make_table_html(segments)
def export_json():
if not _last_segments:
return None
out = str(UPLOAD_DIR / "annotation.json")
to_json(_last_segments, out)
return out
def export_csv():
if not _last_segments:
return None
out = str(UPLOAD_DIR / "annotation.csv")
to_csv(_last_segments, out)
return out
css = """
.gradio-container { max-width: 1100px !important; margin: auto !important; }
footer { display: none !important; }
"""
with gr.Blocks(title="Speech Annotation Pipeline", css=css) as demo:
gr.Markdown(\"\"\"# 🎙️ Speech Annotation Pipeline
*Upload audio · Detect speakers · Export transcript*\"\"\")
with gr.Row():
with gr.Column(scale=1):
audio_input = gr.Audio(label="Upload Audio (.wav / .mp3 / .flac)", type="filepath")
num_speakers = gr.Slider(minimum=0, maximum=10, step=1, value=0, label="Number of speakers (0 = auto-detect)")
run_btn = gr.Button("▶ Run Annotation", variant="primary", size="lg")
status_box = gr.Textbox(label="Status", value="Ready.", interactive=False)
gr.Markdown("### 📥 Export")
with gr.Row():
json_btn = gr.Button("⬇ JSON", size="sm")
csv_btn = gr.Button("⬇ CSV", size="sm")
json_file = gr.File(label="JSON Download", visible=True)
csv_file = gr.File(label="CSV Download", visible=True)
with gr.Column(scale=2):
gr.Markdown("### 💬 Conversation View")
conversation_html = gr.HTML(
value="Transcript will appear here after processing…
"
)
gr.Markdown("### 📋 Segment Table")
table_html = gr.HTML(value="")
run_btn.click(
fn=process_audio,
inputs=[audio_input, num_speakers],
outputs=[status_box, conversation_html, table_html]
)
json_btn.click(fn=export_json, inputs=[], outputs=[json_file])
csv_btn.click(fn=export_csv, inputs=[], outputs=[csv_file])
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True)