Trim silenzio iniziale/finale (VAD energia RMS, numpy+soundfile): meno audio da processare, meno allucinazioni; timestamp riallineati con offset; TRIM_SILENCE=0 per disattivare, TRIM_THRESHOLD_DB/TRIM_PADDING_MS configurabili; flag --no-trim

transcribe.py

@@ -10,6 +10,84 @@ WHISPER_CLI = os.environ.get('WHISPER_CLI', os.path.expanduser('~/dev/whisper.cp
 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'))
 
+# ---------------------------------------------------------------------------
+# Trim del silenzio iniziale/finale (ottimizzazione)
+# ---------------------------------------------------------------------------
+# Cancella le parti vuote prima della prima voce e dopo l'ultima: meno audio
+# da processare (spettrogramma + pyannote), meno rischio di allucinazioni.
+# VAD a energia (RMS) in puro numpy — nessuna dipendenza aggiuntiva.
+# Configurabile: TRIM_SILENCE=0 per disattivare, TRIM_THRESHOLD_DB (default
+# -40), TRIM_PADDING_MS (default 150). I timestamp restano allineati alla
+# timeline originale (offset somato ai segmenti).
+
+try:
+    import numpy as _np
+    import soundfile as _sf
+    _HAS_NP = True
+except ImportError:
+    _HAS_NP = False
+
+
+def _frame_rms(samples, frame):
+    n = (len(samples) // frame) * frame
+    return _np.sqrt(_np.mean(samples[:n].reshape(-1, frame) ** 2, axis=1))
+
+
+def detect_trim_bounds(samples, sr, frame_ms=30, threshold_db=-40.0,
+                       min_speech_ms=300):
+    """Trova (start_sample, end_sample) del parlato (primo/ultimo frame con
+    RMS >= soglia). None se non c'è parlato sufficiente."""
+    frame = max(1, int(sr * frame_ms / 1000))
+    thr = 10 ** (threshold_db / 20)
+    rms = _frame_rms(samples, frame)
+    speech = rms >= thr
+    if not speech.any():
+        return None
+    min_frames = max(1, int(min_speech_ms / frame_ms))
+    # Indici dei frame sopra soglia
+    idx = _np.where(speech)[0]
+    # Gruppi contigui (per ignorare picchi spuri isolati)
+    groups = _np.split(idx, _np.where(_np.diff(idx) > 1)[0] + 1)
+    long = [g for g in groups if len(g) >= min_frames]
+    if not long:
+        return None
+    start = long[0][0] * frame
+    end = (long[-1][-1] + 1) * frame
+    return int(start), int(end)
+
+
+def trim_silence(audio_path):
+    """Ritorna (path_wav_ripulito, offset_secondi) usando un file temporaneo
+    accanto all'originale; None se il silenzio iniziale/finale è già minimo
+    (o mancano numpy/soundfile, o non c'è parlato)."""
+    if not _HAS_NP:
+        return None
+    try:
+        threshold_db = float(os.environ.get('TRIM_THRESHOLD_DB', '-40'))
+        padding_ms = int(os.environ.get('TRIM_PADDING_MS', '150'))
+        data, sr = _sf.read(audio_path, dtype='float32')
+        if data.ndim > 1:
+            data = data.mean(axis=1)
+        if len(data) == 0:
+            return None
+        bounds = detect_trim_bounds(data, sr, threshold_db=threshold_db)
+        if bounds is None:
+            # Nessun parlato: non trascrivere il vuoto
+            return ('', 0.0)
+        start, end = bounds
+        pad = int(padding_ms / 1000 * sr)
+        start = max(0, start - pad)
+        end = min(len(data), end + pad)
+        # Se il taglio è trascurabile, lascia stare
+        if start < 0.05 * sr and end > len(data) - 0.05 * sr:
+            return None
+        out_path = audio_path + '.trim.wav'
+        _sf.write(out_path, data[start:end], sr)
+        return (out_path, start / sr)
+    except Exception:
+        return None
+
+
 # ---------------------------------------------------------------------------
 # Glossario personale: estensione vocabolario + normalizzazione deterministica
 # ---------------------------------------------------------------------------
@@ -86,7 +164,17 @@ def asr(audio, model, lang, tmp, prompt=None):
     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')
-    cmd = [WHISPER_CLI, '-m', model_path, '-f', audio, '-l', lang,
+    # Trim del silenzio iniziale/finale (ottimizzazione, default attivo)
+    src = audio
+    offset = 0.0
+    if os.environ.get('TRIM_SILENCE', '1') != '0':
+        res = trim_silence(audio)
+        if res is not None:
+            trimmed, offset = res
+            if not trimmed:
+                return []  # nessun parlato: niente da trascrivere
+            src = trimmed
+    cmd = [WHISPER_CLI, '-m', model_path, '-f', src, '-l', lang,
            '-oj', '-of', os.path.join(tmp, 'asr'), '--no-prints',
            '-vm', vad_model, '--vad',
            '--suppress-regex',
@@ -107,8 +195,9 @@ def asr(audio, model, lang, tmp, prompt=None):
                             "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,
+        # Riallinea i timestamp alla timeline originale (dopo il trim)
+        segs.append({'start': s['offsets']['from']/1000.0 + offset,
+                     'end':   s['offsets']['to']/1000.0 + offset,
                      'text':  text})
     return segs
 
@@ -176,7 +265,11 @@ def main():
                     help='initial prompt personale (nomi, termini, gergo)')
     ap.add_argument('--glossary', default=None,
                     help='path a glossario JSON (alias -> forma canonica)')
+    ap.add_argument('--no-trim', action='store_true',
+                    help='disattiva la cancellazione del silenzio iniziale/finale')
     args = ap.parse_args()
+    if args.no_trim:
+        os.environ['TRIM_SILENCE'] = '0'
     base = args.out or os.path.splitext(args.audio)[0]
     flat_map = load_glossary(args.glossary)
     with tempfile.TemporaryDirectory() as tmp: