109 lines
4.7 KiB
Python
Executable File
109 lines
4.7 KiB
Python
Executable File
#!/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()
|