File size: 4,998 Bytes
3f0bbf0 e8b2062 3f0bbf0 29fe3fe 3f0bbf0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """
SmartFile Speaker Diarization — Hugging Face Space
==================================================
Runs pyannote/speaker-diarization-3.1 to detect WHO SPEAKS WHEN in an audio
clip. SmartFile calls this BEFORE transcription so each speaker turn can be
transcribed and labeled (Locuteur 1, Locuteur 2, ...).
Contract (same simple pattern as the STT Space):
INPUT : base64-encoded audio (WAV/etc) in a Textbox.
OUTPUT : a JSON string -> {"segments": [{"speaker": "...", "start": s, "end": s}, ...],
"num_speakers": N}
or {"error": "..."} on failure.
The pipeline is GATED on Hugging Face and requires accepting terms for BOTH:
- pyannote/speaker-diarization-3.1
- pyannote/segmentation-3.0 (dependency)
and an HF_TOKEN secret on this Space with access to them.
Tuning: this client's audio is mostly 2-person interviews/calls, so we hint the
pipeline toward 2 speakers (min 1, max 4) — detected, not hard-forced, so a
stray third voice won't break it. Tiny adjacent same-speaker turns are merged.
NOTE: pyannote loads several models (segmentation + embedding); it is HEAVIER
than the STT Space. Use a paid CPU (or GPU) tier for usable speed.
"""
import base64
import io
import json
import os
import gradio as gr
import torch
from pyannote.audio import Pipeline
HF_TOKEN = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
MODEL_ID = "pyannote/speaker-diarization-3.1"
# Hint for the typical case (2-person interview/call). Detected, not hard-forced.
DEFAULT_MIN_SPEAKERS = 1
DEFAULT_MAX_SPEAKERS = 4
MERGE_GAP_SEC = 0.5 # merge same-speaker segments separated by < this gap
MIN_SEG_SEC = 0.3 # drop micro-segments shorter than this (noise/clicks)
print(f"[diar] loading {MODEL_ID} ...")
pipeline = Pipeline.from_pretrained(MODEL_ID, token=HF_TOKEN)
if torch.cuda.is_available():
pipeline.to(torch.device("cuda"))
print("[diar] using GPU")
print("[diar] pipeline ready")
def _merge_segments(segs):
"""Merge adjacent same-speaker segments with tiny gaps; drop micro-segments."""
if not segs:
return segs
segs = sorted(segs, key=lambda s: s["start"])
merged = [segs[0]]
for s in segs[1:]:
last = merged[-1]
if s["speaker"] == last["speaker"] and s["start"] - last["end"] <= MERGE_GAP_SEC:
last["end"] = max(last["end"], s["end"])
else:
merged.append(s)
return [s for s in merged if (s["end"] - s["start"]) >= MIN_SEG_SEC]
def diarize_b64(audio_b64):
if not audio_b64:
return json.dumps({"error": "empty input"})
try:
if audio_b64.strip().startswith("data:") and "," in audio_b64[:64]:
audio_b64 = audio_b64.split(",", 1)[1]
raw = base64.b64decode(audio_b64)
print(f"[diar] decoded {len(raw)} audio bytes")
# pyannote reads a file path or a waveform; easiest robust path is to
# write the bytes to a temp file (it handles WAV/MP3/etc + resampling).
import tempfile
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
tmp.write(raw)
tmp_path = tmp.name
diarization = pipeline(
tmp_path,
min_speakers=DEFAULT_MIN_SPEAKERS,
max_speakers=DEFAULT_MAX_SPEAKERS,
)
os.unlink(tmp_path)
segs = []
# pyannote 4.x returns an object whose `.speaker_diarization` yields
# (turn, speaker) pairs. pyannote 3.x returns an Annotation with
# `.itertracks(yield_label=True)` yielding (turn, _, speaker). Support
# both so the Space isn't tied to one pyannote version.
if hasattr(diarization, "speaker_diarization"):
for turn, speaker in diarization.speaker_diarization:
segs.append({
"speaker": str(speaker),
"start": round(float(turn.start), 2),
"end": round(float(turn.end), 2),
})
else:
for turn, _, speaker in diarization.itertracks(yield_label=True):
segs.append({
"speaker": str(speaker), # e.g. "SPEAKER_00"
"start": round(float(turn.start), 2),
"end": round(float(turn.end), 2),
})
segs = _merge_segments(segs)
speakers = sorted({s["speaker"] for s in segs})
print(f"[diar] {len(segs)} segments, {len(speakers)} speakers")
return json.dumps({"segments": segs, "num_speakers": len(speakers)})
except Exception as e:
print(f"[diar] error: {e}")
return json.dumps({"error": str(e)})
demo = gr.Interface(
fn=diarize_b64,
inputs=gr.Textbox(label="Base64 audio"),
outputs=gr.Textbox(label="Diarization JSON"),
title="SmartFile Speaker Diarization",
description="Detect who speaks when (pyannote 3.1). Send base64 audio, get speaker segments as JSON.",
)
if __name__ == "__main__":
demo.launch()
|