stt-server: FastAPI whisper.cpp Vulkan ASR + pyannote diarization, containerizzato

transcribe.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""STT + Speaker Diarization pipeline (frigate.vpn).
+ASR: whisper.cpp (Vulkan) | Diarization: pyannote Community-1 (CPU)
+Uso: transcribe.py <audio> [--lang it] [--model large-v3-turbo] [--out base]
+"""
+import argparse, json, os, subprocess, sys, tempfile
+
+WHISPER_CLI = os.environ.get('WHISPER_CLI', os.path.expanduser('~/dev/whisper.cpp/build/bin/whisper-cli'))
+MODELS_DIR  = os.environ.get('MODELS_DIR', os.path.expanduser('~/dev/whisper.cpp/models'))
+PYTHON      = os.environ.get('STT_PYTHON', os.path.expanduser('~/dev/stt-venv/bin/python'))
+
+def asr(audio, model, lang, tmp):
+    model_path = os.path.join(MODELS_DIR, f'ggml-{model}.bin')
+    out_json = os.path.join(tmp, 'asr.json')
+    vad_model = os.path.join(MODELS_DIR, 'ggml-silero-v6.2.0.bin')
+    # VAD (silero): salta il silenzio/rumore → elimina le allucinazioni
+    # tipo "Grazie a tutti" su audio senza parlato. --suppress-regex come
+    # rete di sicurezza per frasi allucinate residue.
+    subprocess.run([WHISPER_CLI, '-m', model_path, '-f', audio, '-l', lang,
+                    '-oj', '-of', os.path.join(tmp, 'asr'), '--no-prints',
+                    '-vm', vad_model, '--vad',
+                    '--suppress-regex',
+                    r'(Grazie a tutti|Grazie per l.attenzione|Thank you|Thanks for watching|Sottotitoli creati|Sottotitoli)'],
+                   check=True, capture_output=True)
+    data = json.load(open(out_json))
+    segs = []
+    for s in data['transcription']:
+        text = s['text'].strip()
+        # Filtro finale: scarta segmenti che sono solo frasi allucinate
+        # (rete di sicurezza oltre a VAD e suppress-regex)
+        if text.lower() in {'grazie', 'grazie a tutti', 'thank you', 'thanks',
+                            "grazie per l'attenzione", 'sottotitoli',
+                            'sottotitoli creati', 'sottotitoli creati da'}:
+            continue
+        segs.append({'start': s['offsets']['from']/1000.0,
+                     'end':   s['offsets']['to']/1000.0,
+                     'text':  text})
+    return segs
+
+def diarize(audio, tmp):
+    script = f'''
+import json, sys
+import numpy as np, soundfile as sf, torch
+from pyannote.audio import Pipeline
+data, sr = sf.read({audio!r}, dtype='float32')
+wav = torch.from_numpy(data.T if data.ndim > 1 else data[None, :]).float()
+p = Pipeline.from_pretrained('pyannote/speaker-diarization-community-1')
+out = p({{'waveform': wav, 'sample_rate': sr}})
+diar = out.speaker_diarization if hasattr(out, 'speaker_diarization') else out
+turns = [{{'start': t.start, 'end': t.end, 'speaker': sp}}
+         for t, _, sp in diar.itertracks(yield_label=True)]
+json.dump(turns, open({os.path.join(tmp, 'diar.json')!r}, 'w'))
+'''
+    sp = os.path.join(tmp, 'diar.py')
+    open(sp, 'w').write(script)
+    subprocess.run([PYTHON, sp], check=True)
+    return json.load(open(os.path.join(tmp, 'diar.json')))
+
+def overlap(a, b):
+    return max(0.0, min(a['end'], b['end']) - max(a['start'], b['start']))
+
+def merge(segs, turns):
+    for s in segs:
+        best, best_ov = None, 0.0
+        for t in turns:
+            ov = overlap(s, t)
+            if ov > best_ov:
+                best, best_ov = t['speaker'], ov
+        s['speaker'] = best if best_ov > 0.05 else 'UNKNOWN'
+    return segs
+
+def to_srt(segs, path):
+    def ts(x):
+        h, m = int(x//3600), int(x%3600//60)
+        s, ms = int(x%60), int((x-int(x))*1000)
+        return f'{h:02d}:{m:02d}:{s:02d},{ms:03d}'
+    with open(path, 'w') as f:
+        for i, s in enumerate(segs, 1):
+            f.write(f'{i}\n{ts(s["start"])} --> {ts(s["end"])}\n')
+            f.write(f'[{s["speaker"]}] {s["text"]}\n\n')
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument('audio')
+    ap.add_argument('--lang', default='it')
+    ap.add_argument('--model', default='large-v3-turbo')
+    ap.add_argument('--out', default=None)
+    args = ap.parse_args()
+    base = args.out or os.path.splitext(args.audio)[0]
+    with tempfile.TemporaryDirectory() as tmp:
+        print(f'[1/3] ASR whisper.cpp ({args.model}, {args.lang})...', file=sys.stderr)
+        segs = asr(args.audio, args.model, args.lang, tmp)
+        print(f'      {len(segs)} segmenti', file=sys.stderr)
+        print('[2/3] Diarization pyannote Community-1...', file=sys.stderr)
+        turns = diarize(args.audio, tmp)
+        print(f'      {len(turns)} turni parlante', file=sys.stderr)
+        print('[3/3] Merge ASR + diarization...', file=sys.stderr)
+        segs = merge(segs, turns)
+    out_json = base + '.json'
+    out_srt  = base + '.srt'
+    json.dump({'segments': segs}, open(out_json, 'w'), ensure_ascii=False, indent=2)
+    to_srt(segs, out_srt)
+    print(f'OK: {out_json} + {out_srt}', file=sys.stderr)
+    for s in segs:
+        print(f'{s["start"]:6.2f}-{s["end"]:6.2f} [{s["speaker"]}] {s["text"]}')
+
+if __name__ == '__main__':
+    main()