Upload 10 files

setting up main engine

reduction/__init__.py

@@ -0,0 +1 @@
+"""Phonetic-reduction estimator (STT->TTS comparison pipeline)."""

reduction/compare.py

@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+import librosa
+import numpy as np
+
+SR_TARGET = 16000  # common processing rate for MFCCs (also silero-vad's native rate)
+
+
+_VAD_MODEL = None
+
+
+def _vad_model():
+    global _VAD_MODEL
+    if _VAD_MODEL is None:
+        from silero_vad import load_silero_vad  # lazy import — heavy dep
+
+        _VAD_MODEL = load_silero_vad()
+    return _VAD_MODEL
+
+
+def load_audio(path: str, sr: int = SR_TARGET) -> tuple[np.ndarray, int]:
+    y, real_sr = librosa.load(path, sr=sr, mono=True)
+    return y.astype(np.float32), real_sr
+
+
+def slice_word(y: np.ndarray, sr: int, start_s: float, end_s: float) -> np.ndarray:
+    s = max(0, int(round(start_s * sr)))
+    e = min(len(y), int(round(end_s * sr)))
+    return y[s:e]
+
+
+def trim_silence(y: np.ndarray, sr: int, top_db: float = 30.0) -> np.ndarray:
+    """[fallback] Energy-threshold silence trim. Used when VAD returns empty."""
+    if len(y) < int(0.02 * sr):
+        return y
+    yt, _ = librosa.effects.trim(y, top_db=top_db)
+    return yt if len(yt) > 0 else y
+
+
+def trim_silence_vad(
+    y: np.ndarray,
+    sr: int,
+    threshold: float = 0.5,
+    min_speech_ms: int = 30,
+    min_silence_ms: int = 80,
+) -> np.ndarray:
+    """VAD-based trim. Crops leading/trailing non-speech around a word slice.
+
+    Robust to ambient noise, breath, and audible-but-non-phonetic sound that the
+    energy-threshold trim would keep (e.g., the deliberate pause attached to "a"
+    in rec.3 — measured 0.8 s at top_db=30, ~0.05 s with VAD).
+
+    For slices shorter than ~100 ms VAD is unreliable; we fall back to the energy
+    trim (which mostly does nothing on such short slices anyway).
+    """
+    import torch
+
+    if len(y) < int(0.1 * sr):
+        return trim_silence(y, sr)
+
+    # silero-vad operates on 8 kHz or 16 kHz only
+    if sr not in (8000, 16000):
+        y_vad = librosa.resample(y, orig_sr=sr, target_sr=SR_TARGET)
+        scale = sr / SR_TARGET
+        vad_sr = SR_TARGET
+    else:
+        y_vad = y
+        scale = 1.0
+        vad_sr = sr
+
+    from silero_vad import get_speech_timestamps
+
+    audio_t = torch.from_numpy(y_vad.astype(np.float32))
+    speech = get_speech_timestamps(
+        audio_t,
+        _vad_model(),
+        sampling_rate=vad_sr,
+        threshold=threshold,
+        min_speech_duration_ms=min_speech_ms,
+        min_silence_duration_ms=min_silence_ms,
+    )
+    if not speech:
+        return trim_silence(y, sr)  # fallback to energy trim
+
+    # Concatenate speech segments, dropping any inter-segment silence.
+    # Rationale: silero's min_silence_duration_ms=80 ms means any inter-segment gap
+    # is ≥80 ms, which is longer than typical stop closures (40–70 ms in fast Czech),
+    # so a multi-segment split signals a real pause/hesitation that should be excluded.
+    pieces: list[np.ndarray] = []
+    for seg in speech:
+        s = max(0, int(round(seg["start"] * scale)))
+        e = min(len(y), int(round(seg["end"] * scale)))
+        if e > s:
+            pieces.append(y[s:e])
+    if not pieces:
+        return trim_silence(y, sr)
+    return pieces[0] if len(pieces) == 1 else np.concatenate(pieces)
+
+
+def _mfcc_cmn(y: np.ndarray, sr: int, n_mfcc: int = 13, hop_ms: float = 10.0) -> np.ndarray:
+    hop = max(1, int(sr * hop_ms / 1000.0))
+    n_fft = 512 if sr <= 16000 else 1024
+    M = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=n_mfcc, hop_length=hop, n_fft=n_fft)
+    # cepstral mean normalization — removes channel / microphone bias per stream
+    M = M - M.mean(axis=1, keepdims=True)
+    return M
+
+
+def spectral_distance(
+    orig_wav: np.ndarray,
+    orig_sr: int,
+    tts_wav: np.ndarray,
+    tts_sr: int,
+    target_sr: int = SR_TARGET,
+) -> float:
+    """DTW-aligned mean MFCC distance between original word and its TTS rendering.
+
+    Higher = more spectrally distant from canonical TTS = more reduced/different.
+    Returns NaN for too-short slices (cannot compute MFCC reliably).
+    """
+    if orig_sr != target_sr:
+        orig_wav = librosa.resample(orig_wav, orig_sr=orig_sr, target_sr=target_sr)
+    if tts_sr != target_sr:
+        tts_wav = librosa.resample(tts_wav, orig_sr=tts_sr, target_sr=target_sr)
+    # need at least ~30 ms of audio for a meaningful MFCC sequence
+    if len(orig_wav) < int(0.03 * target_sr) or len(tts_wav) < int(0.03 * target_sr):
+        return float("nan")
+    M_orig = _mfcc_cmn(orig_wav, target_sr)
+    M_tts = _mfcc_cmn(tts_wav, target_sr)
+    if M_orig.shape[1] < 2 or M_tts.shape[1] < 2:
+        return float("nan")
+    D, wp = librosa.sequence.dtw(M_orig, M_tts, metric="euclidean")
+    return float(D[-1, -1] / max(1, len(wp)))
+
+
+def center_scores(rows: list[dict]) -> list[dict]:
+    """Add centered scores (mean = 1) to each row.
+
+    Convention: higher score = less reduced (closer to canonical / longer).
+        duration_score = (orig_dur / tts_dur) / mean(orig_dur / tts_dur)
+        spectral_score = mean(spec_dist) / spec_dist   (inverted, same direction as duration)
+        combined_score = arithmetic mean of the two (NaN-safe)
+    """
+    dur_ratios = np.array([r["duration_ratio"] for r in rows], dtype=float)
+    spec_dists = np.array([r["spectral_distance"] for r in rows], dtype=float)
+
+    mean_dr = np.nanmean(dur_ratios) if np.any(~np.isnan(dur_ratios)) else float("nan")
+    mean_sd = np.nanmean(spec_dists) if np.any(~np.isnan(spec_dists)) else float("nan")
+
+    for r in rows:
+        dr = r["duration_ratio"]
+        sd = r["spectral_distance"]
+        r["duration_score"] = (dr / mean_dr) if (mean_dr and not np.isnan(dr)) else float("nan")
+        r["spectral_score"] = (mean_sd / sd) if (sd and not np.isnan(sd) and not np.isnan(mean_sd)) else float("nan")
+        components = [v for v in (r["duration_score"], r["spectral_score"]) if not np.isnan(v)]
+        r["combined_score"] = float(np.mean(components)) if components else float("nan")
+    return rows

reduction/pipeline.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from . import compare, stt, tts
+
+
+def analyze(
+    audio_path: str,
+    whisper_model,
+    voice,
+    language: str = "cs",
+    groundtruth: str | None = None,
+    trim: bool = True,
+) -> tuple[list[dict], str, str | None]:
+    """Run STT -> per-word TTS -> per-word duration & spectral comparison.
+
+    If `groundtruth` is provided, its tokens replace Whisper's word identities
+    (Whisper still drives timing). This isolates the metric from STT errors.
+
+    Returns (rows, whisper_transcript, groundtruth_or_None).
+    """
+    whisper_words, whisper_text = stt.transcribe_words(
+        audio_path, whisper_model, language=language
+    )
+    if groundtruth is not None:
+        gt_tokens = stt.tokenize_transcript(groundtruth)
+        words = stt.align_to_groundtruth(whisper_words, gt_tokens)
+    else:
+        words = whisper_words
+
+    orig_y, orig_sr = compare.load_audio(audio_path)
+
+    rows: list[dict] = []
+    for w in words:
+        orig_slice = compare.slice_word(orig_y, orig_sr, w.start, w.end)
+        if trim:
+            orig_slice = compare.trim_silence_vad(orig_slice, orig_sr)
+        orig_dur_used = len(orig_slice) / orig_sr if len(orig_slice) > 0 else 0.0
+
+        _tts_dur_full, tts_wav, tts_sr = tts.synthesize_duration(voice, w.text)
+        if trim:
+            tts_wav = compare.trim_silence_vad(tts_wav, tts_sr)
+        tts_dur_used = len(tts_wav) / tts_sr if len(tts_wav) > 0 else 0.0
+
+        spec = compare.spectral_distance(orig_slice, orig_sr, tts_wav, tts_sr)
+        dur_ratio = (orig_dur_used / tts_dur_used) if tts_dur_used > 0 else float("nan")
+
+        rows.append(
+            {
+                "word": w.text,
+                "raw": w.raw.strip(),
+                "start": w.start,
+                "end": w.end,
+                "stt_probability": getattr(w, "probability", float("nan")),
+                "orig_duration": orig_dur_used,
+                "tts_duration": tts_dur_used,
+                "duration_ratio": dur_ratio,
+                "spectral_distance": spec,
+            }
+        )
+
+    compare.center_scores(rows)
+    return rows, whisper_text, groundtruth

reduction/stt.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+import difflib
+import re
+from dataclasses import dataclass
+
+from faster_whisper import WhisperModel
+
+
+_STRIP_RE = re.compile(r"^[\s\.,!?;:\"'„“”‚‘’\(\)\[\]…—–-]+|[\s\.,!?;:\"'„“”‚‘’\(\)\[\]…—–-]+$")
+
+
+@dataclass
+class Word:
+    text: str       # normalized form for TTS (lowercased, punctuation stripped)
+    raw: str        # original Whisper token (kept for debugging / display)
+    start: float    # seconds in original audio
+    end: float
+    probability: float = float("nan")  # Whisper per-word probability (NaN if unknown)
+
+    @property
+    def duration(self) -> float:
+        return self.end - self.start
+
+
+def _normalize(token: str) -> str:
+    return _STRIP_RE.sub("", token).lower()
+
+
+def transcribe_words(
+    audio_path: str,
+    model: WhisperModel,
+    language: str = "cs",
+) -> tuple[list[Word], str]:
+    """Transcribe `audio_path` and return word-level segmentation + full text.
+
+    Empty / pure-punctuation tokens are skipped.
+    """
+    segments, _info = model.transcribe(
+        audio_path, language=language, word_timestamps=True
+    )
+    words: list[Word] = []
+    full_text: list[str] = []
+    for seg in segments:
+        full_text.append(seg.text)
+        if not seg.words:
+            continue
+        for w in seg.words:
+            normalized = _normalize(w.word)
+            if not normalized:
+                continue
+            prob = getattr(w, "probability", float("nan"))
+            words.append(
+                Word(
+                    text=normalized,
+                    raw=w.word,
+                    start=w.start,
+                    end=w.end,
+                    probability=float(prob) if prob is not None else float("nan"),
+                )
+            )
+    return words, "".join(full_text).strip()
+
+
+def load_model(
+    size: str = "large-v3",
+    device: str = "cpu",
+    compute_type: str = "int8",
+    download_root: str = "models/whisper",
+) -> WhisperModel:
+    return WhisperModel(
+        size, device=device, compute_type=compute_type, download_root=download_root
+    )
+
+
+def tokenize_transcript(text: str) -> list[str]:
+    """Split a free-form transcript into normalized word tokens (lowercase, no punct)."""
+    return [t for t in (_normalize(tok) for tok in text.split()) if t]
+
+
+def _char_dist(a: str, b: str) -> float:
+    """Normalized character-level edit distance in [0, 1]. 0 = identical."""
+    if not a and not b:
+        return 0.0
+    return 1.0 - difflib.SequenceMatcher(a=a, b=b, autojunk=False).ratio()
+
+
+def align_to_groundtruth(
+    whisper_words: list[Word], gt_tokens: list[str], gap_penalty: float = 0.7
+) -> list[Word]:
+    """Replace Whisper word identities with ground-truth tokens, keeping Whisper timing.
+
+    Uses Needleman–Wunsch alignment with character-level edit distance as substitution cost.
+    Behavior:
+      - Aligned (W_i, GT_j): emit Word(text=GT_j, start/end from W_i).
+      - GT_j unmatched: try to interpolate timing from neighboring aligned tokens; if
+        impossible (start/end of utterance with no anchor), drop with a stderr warning.
+      - W_i unmatched: drop (Whisper inserted a phantom token).
+    """
+    if not whisper_words:
+        return []
+    if not gt_tokens:
+        return list(whisper_words)
+
+    n, m = len(whisper_words), len(gt_tokens)
+    dp = [[0.0] * (m + 1) for _ in range(n + 1)]
+    bt = [[""] * (m + 1) for _ in range(n + 1)]
+    for i in range(1, n + 1):
+        dp[i][0] = i * gap_penalty
+        bt[i][0] = "D"
+    for j in range(1, m + 1):
+        dp[0][j] = j * gap_penalty
+        bt[0][j] = "I"
+    for i in range(1, n + 1):
+        for j in range(1, m + 1):
+            sub = dp[i - 1][j - 1] + _char_dist(whisper_words[i - 1].text, gt_tokens[j - 1])
+            dele = dp[i - 1][j] + gap_penalty
+            ins = dp[i][j - 1] + gap_penalty
+            best = min(sub, dele, ins)
+            dp[i][j] = best
+            bt[i][j] = "M" if best == sub else ("D" if best == dele else "I")
+
+    # backtrack
+    pairs: list[tuple[int | None, int | None]] = []
+    i, j = n, m
+    while i > 0 or j > 0:
+        op = bt[i][j]
+        if op == "M":
+            pairs.append((i - 1, j - 1))
+            i -= 1
+            j -= 1
+        elif op == "D":
+            pairs.append((i - 1, None))
+            i -= 1
+        else:
+            pairs.append((None, j - 1))
+            j -= 1
+    pairs.reverse()
+
+    # First pass: emit matched + skip Whisper-only; collect GT-only with neighbor anchors.
+    out: list[Word | None] = []
+    pending_gt: list[int] = []  # GT indices waiting for next anchor
+    last_end: float | None = None
+    for w_idx, g_idx in pairs:
+        if w_idx is not None and g_idx is not None:
+            ww = whisper_words[w_idx]
+            # if there are pending GT tokens, distribute them between last_end and ww.start
+            if pending_gt:
+                if last_end is not None:
+                    span_start, span_end = last_end, ww.start
+                    n_pend = len(pending_gt)
+                    for k, gi in enumerate(pending_gt):
+                        ts = span_start + (span_end - span_start) * k / n_pend
+                        te = span_start + (span_end - span_start) * (k + 1) / n_pend
+                        out.append(Word(text=gt_tokens[gi], raw=gt_tokens[gi], start=ts, end=te))
+                else:
+                    # at start of utterance with no anchor — assign 0..ww.start span
+                    n_pend = len(pending_gt)
+                    for k, gi in enumerate(pending_gt):
+                        ts = ww.start * k / n_pend
+                        te = ww.start * (k + 1) / n_pend
+                        out.append(Word(text=gt_tokens[gi], raw=gt_tokens[gi], start=ts, end=te))
+                pending_gt = []
+            out.append(Word(text=gt_tokens[g_idx], raw=gt_tokens[g_idx], start=ww.start, end=ww.end))
+            last_end = ww.end
+        elif w_idx is not None and g_idx is None:
+            # Whisper-only — skip, but advance time anchor
+            last_end = whisper_words[w_idx].end
+        else:  # GT-only
+            pending_gt.append(g_idx)  # type: ignore[arg-type]
+
+    # Trailing pending GT tokens (no right-anchor): assign small constant duration
+    if pending_gt:
+        if last_end is not None:
+            span_start = last_end
+            span_end = last_end + 0.3 * len(pending_gt)  # rough fallback
+            n_pend = len(pending_gt)
+            for k, gi in enumerate(pending_gt):
+                ts = span_start + (span_end - span_start) * k / n_pend
+                te = span_start + (span_end - span_start) * (k + 1) / n_pend
+                out.append(Word(text=gt_tokens[gi], raw=gt_tokens[gi], start=ts, end=te))
+
+    return [w for w in out if w is not None]

reduction/tts.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import io
+import wave
+
+import numpy as np
+from piper import PiperVoice
+
+
+def load_voice(voice_path: str = "models/piper/cs_CZ-jirka-medium.onnx") -> PiperVoice:
+    return PiperVoice.load(voice_path)
+
+
+def synthesize(voice: PiperVoice, text: str) -> tuple[np.ndarray, int]:
+    """Synthesize `text` and return (mono float32 waveform in [-1,1], sample_rate)."""
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as wf:
+        voice.synthesize_wav(text, wf)
+    buf.seek(0)
+    with wave.open(buf, "rb") as wf:
+        sr = wf.getframerate()
+        n_channels = wf.getnchannels()
+        sampwidth = wf.getsampwidth()
+        raw = wf.readframes(wf.getnframes())
+    if sampwidth != 2:
+        raise RuntimeError(f"Unexpected Piper sample width: {sampwidth} bytes")
+    pcm = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0
+    if n_channels > 1:
+        pcm = pcm.reshape(-1, n_channels).mean(axis=1)
+    return pcm, sr
+
+
+def synthesize_duration(voice: PiperVoice, text: str) -> tuple[float, np.ndarray, int]:
+    """Convenience: returns (duration_seconds, waveform, sample_rate)."""
+    wav, sr = synthesize(voice, text)
+    return len(wav) / sr, wav, sr