Perfect short clips into ideal XTTS references (refine.py)

README.md

@@ -62,13 +62,40 @@ For folders or automation, skip the GUI:
 python cli.py /path/to/videos -o out/ --merge
 ```
 
+### Build the *ideal* clone reference (recommended)
+
+`refine.py` goes further than plain extraction — it turns a short, imperfect clip
+into the **best possible reference for XTTS-v2 zero-shot cloning**, which only
+needs ~6 seconds of *clean* speech:
+
+```bash
+python refine.py mom_birthday.mov --out mom_reference.wav
+```
+
+What it does, and why each step matters for cloning a specific person:
+
+1. **Isolates the voice** (Demucs) and **transcribes** it (Whisper) into segments.
+2. **Separates speakers by timbre**, not just pitch — so a mother and a daughter
+   in the same clip aren't merged (the classic failure). Target one with
+   `--speaker A`.
+3. **Scores every segment** (SNR, voiced-ratio, clipping, length) and keeps only
+   the single **cleanest contiguous window** — clean beats long for XTTS.
+4. Renders **24 kHz mono, gently normalized** (XTTS's sweet spot, not broadcast
+   loudness which dulls timbre), and prints a **quality report + warnings** so you
+   know the sample is good before cloning.
+
+It prints JSON (`ok`, `duration`, `snr_db`, `speaker`, `report`, `warnings`) for
+automation, and degrades gracefully if Demucs/Whisper aren't installed.
+
 ---
 
 ## Tips for good cloning results
 
-- Aim for **1–2+ minutes** of clear speech total. More clean audio = better clone.
-- Prefer clips with little background music or noise.
-- Use the **merge** option to combine several short clips into one sample.
+- **Even ~6 seconds works** for zero-shot cloning (XTTS-v2) — *clean* beats
+  *long*. `refine.py` finds the cleanest window for you automatically.
+- Prefer clips with little background music or noise (Demucs handles a lot).
+- Multiple short clips? Refine each and keep the highest-SNR result, or use
+  `cli.py --merge` to combine.
 
 ---
 

analyze.py

@@ -155,7 +155,7 @@ def analyze(src: str, *, use_demucs: Optional[bool] = None,
                                 _mean_pitch(chunk, sr), round(rms, 4)))
 
         log("Clustering speakers by voice…")
-        _assign_speakers(segs)
+        _assign_speakers(segs, data, sr)
         return {"ok": True, "source": src, "duration": round(len(data) / sr, 1),
                 "transcribed": bool(words),
                 "segments": [_seg_json(s) for s in segs],
@@ -169,28 +169,245 @@ def _frange(a: float, b: float, step: float):
         x += step
 
 
-def _assign_speakers(segs: list[Segment]) -> None:
-    """Cluster segments into speaker groups by pitch (simple, transparent: split
-    voiced segments around pitch gaps into up to 3 bands → A/B/C)."""
+def _assign_speakers(segs: list[Segment], data: list[float] | None = None,
+                     sr: int = 16000) -> None:
+    """Group segments by *speaker* using voice timbre, not just pitch.
+
+    Pitch alone merges people who happen to speak in the same range (a mother and
+    a daughter both read as "female-range"). Instead we build a small timbre
+    fingerprint per segment — log-energy across mel-ish frequency bands plus pitch
+    — and cluster those. Two speakers with the same pitch but different vocal
+    tracts still separate, because their spectral shape (formants) differs.
+
+    Falls back to the old pitch-band split when we have no audio to fingerprint.
+    Transparent and dependency-light: numpy is used if present (faster), else a
+    pure-Python path. """
+    voiced = [s for s in segs if s.pitch_hz > 0]
+    if not voiced:
+        return
+    if data is None:
+        _assign_speakers_by_pitch(segs)
+        return
+
+    # 1) fingerprint each voiced segment (spectral-band energies + pitch)
+    feats: list[list[float]] = []
+    for s in voiced:
+        a, b = int(s.start * sr), int(s.end * sr)
+        feats.append(_voice_fingerprint(data[a:b], sr, s.pitch_hz))
+
+    # 2) how many speakers? estimate from feature spread, capped at 3.
+    k = _estimate_speakers(feats)
+    if k <= 1:
+        for s in voiced:
+            s.speaker = "A"
+        return
+
+    # 3) cluster (k-means on standardized features), label by size (A=biggest)
+    labels = _kmeans(feats, k)
+    if labels is None:                     # clustering unavailable → pitch fallback
+        _assign_speakers_by_pitch(segs)
+        return
+    order = _labels_by_frequency(labels)   # remap so A is the most common speaker
+    for s, lab in zip(voiced, labels):
+        s.speaker = order[lab]
+
+
+def _assign_speakers_by_pitch(segs: list[Segment]) -> None:
+    """The original transparent fallback: split voiced segments into up to 3
+    pitch bands (A/B/C). Used when audio fingerprinting isn't available."""
     voiced = [s for s in segs if s.pitch_hz > 0]
     if not voiced:
         return
     pitches = sorted(s.pitch_hz for s in voiced)
-    # crude k≈ up to 3 bands using the overall pitch range
     lo, hi = pitches[0], pitches[-1]
-    if hi - lo < 25:                       # everyone sounds similar → one speaker
-        for s in segs:
-            if s.pitch_hz > 0:
-                s.speaker = "A"
+    if hi - lo < 25:
+        for s in voiced:
+            s.speaker = "A"
         return
     t1 = lo + (hi - lo) / 3
     t2 = lo + 2 * (hi - lo) / 3
-    for s in segs:
-        if s.pitch_hz <= 0:
-            continue
+    for s in voiced:
         s.speaker = "A" if s.pitch_hz < t1 else ("B" if s.pitch_hz < t2 else "C")
 
 
+def _voice_fingerprint(samples: list[float], sr: int, pitch_hz: float) -> list[float]:
+    """A small timbre vector: log energy in a handful of frequency bands (a cheap
+    stand-in for MFCC formant structure) plus normalized pitch. Averaged over the
+    segment's frames so it characterizes the *voice*, not the words."""
+    if not samples:
+        return [0.0] * 9
+    frame = _next_pow2(int(sr * 0.032))     # ~32ms FFT frame
+    hop = frame // 2
+    # mel-ish band edges (Hz) across the vocal range
+    edges = [80, 200, 400, 700, 1100, 1700, 2600, 4000, 6000]
+    bands = [0.0] * (len(edges) - 1)
+    nframes = 0
+    i = 0
+    while i + frame <= len(samples):
+        mag = _fft_mag(samples[i:i + frame])
+        # map FFT bins to Hz and accumulate energy per band
+        for bi in range(len(bands)):
+            f_lo, f_hi = edges[bi], edges[bi + 1]
+            k_lo = max(1, int(f_lo * frame / sr))
+            k_hi = min(len(mag) - 1, int(f_hi * frame / sr))
+            e = sum(mag[k] for k in range(k_lo, k_hi + 1))
+            bands[bi] += e
+        nframes += 1
+        i += hop
+    if nframes:
+        bands = [b / nframes for b in bands]
+    # log-compress (perceptual) and append normalized pitch
+    vec = [math.log(b + 1e-6) for b in bands]
+    vec.append(pitch_hz / 200.0)
+    return vec
+
+
+def _next_pow2(n: int) -> int:
+    p = 1
+    while p < n:
+        p <<= 1
+    return p
+
+
+def _fft_mag(frame: list[float]) -> list[float]:
+    """Magnitude spectrum. Uses numpy if present (fast), else a compact recursive
+    FFT. Applies a Hann window first to reduce leakage."""
+    n = len(frame)
+    win = [0.5 - 0.5 * math.cos(2 * math.pi * i / (n - 1)) for i in range(n)]
+    x = [frame[i] * win[i] for i in range(n)]
+    try:
+        import numpy as np
+        return list(np.abs(np.fft.rfft(np.asarray(x))))
+    except Exception:
+        spec = _fft(x)
+        half = n // 2 + 1
+        return [abs(spec[k]) for k in range(half)]
+
+
+def _fft(x: list[float]) -> list[complex]:
+    """Iterative radix-2 Cooley–Tukey FFT (n must be a power of 2)."""
+    n = len(x)
+    if n <= 1:
+        return [complex(v) for v in x]
+    a = [complex(v) for v in x]
+    # bit-reversal permutation
+    j = 0
+    for i in range(1, n):
+        bit = n >> 1
+        while j & bit:
+            j ^= bit
+            bit >>= 1
+        j |= bit
+        if i < j:
+            a[i], a[j] = a[j], a[i]
+    length = 2
+    while length <= n:
+        ang = -2j * math.pi / length
+        wlen = complex(math.cos(ang.imag), math.sin(ang.imag))
+        for i in range(0, n, length):
+            w = 1 + 0j
+            for k in range(length // 2):
+                u = a[i + k]
+                v = a[i + k + length // 2] * w
+                a[i + k] = u + v
+                a[i + k + length // 2] = u - v
+                w *= wlen
+        length <<= 1
+    return a
+
+
+def _estimate_speakers(feats: list[list[float]]) -> int:
+    """Guess speaker count (1–3) from how spread out the fingerprints are. If the
+    features form one tight blob → 1 speaker; clear separation → 2 or 3."""
+    if len(feats) < 4:
+        return 1
+    # standardized spread: mean pairwise distance vs. within-tolerance
+    import statistics
+    cols = list(zip(*feats))
+    means = [statistics.fmean(c) for c in cols]
+    stds = [statistics.pstdev(c) or 1.0 for c in cols]
+    z = [[(v - means[j]) / stds[j] for j, v in enumerate(f)] for f in feats]
+    # silhouette-ish: try k=2 and k=3, keep whichever separates clearly
+    best_k = 1
+    for k in (2, 3):
+        labels = _kmeans(z, k)
+        if labels is None or len(set(labels)) < k:
+            continue
+        if _cluster_separation(z, labels) > 1.15:   # inter/intra ratio threshold
+            best_k = k
+    return best_k
+
+
+def _cluster_separation(feats: list[list[float]], labels: list[int]) -> float:
+    """Ratio of between-cluster to within-cluster spread. >1 means real clusters."""
+    import statistics
+    groups: dict[int, list[list[float]]] = {}
+    for f, l in zip(feats, labels):
+        groups.setdefault(l, []).append(f)
+    centroids = {l: [statistics.fmean(c) for c in zip(*g)] for l, g in groups.items()}
+    overall = [statistics.fmean(c) for c in zip(*feats)]
+    within = between = 0.0
+    for l, g in groups.items():
+        for f in g:
+            within += _dist(f, centroids[l])
+        between += len(g) * _dist(centroids[l], overall)
+    within = within / len(feats) or 1e-6
+    between = between / len(feats)
+    return between / within
+
+
+def _dist(a: list[float], b: list[float]) -> float:
+    return math.sqrt(sum((x - y) ** 2 for x, y in zip(a, b)))
+
+
+def _kmeans(feats: list[list[float]], k: int, iters: int = 25) -> list[int] | None:
+    """Standardized k-means. numpy if available, else pure-Python. Deterministic
+    init (spread-out seeds) so results are stable run-to-run."""
+    if len(feats) < k:
+        return None
+    import statistics
+    cols = list(zip(*feats))
+    means = [statistics.fmean(c) for c in cols]
+    stds = [statistics.pstdev(c) or 1.0 for c in cols]
+    z = [[(v - means[j]) / stds[j] for j, v in enumerate(f)] for f in feats]
+    # k-means++ style deterministic seeds: first point, then farthest each time
+    centers = [z[0]]
+    while len(centers) < k:
+        far, far_d = z[0], -1.0
+        for p in z:
+            d = min(_dist(p, c) for c in centers)
+            if d > far_d:
+                far, far_d = p, d
+        centers.append(far)
+    labels = [0] * len(z)
+    for _ in range(iters):
+        changed = False
+        for i, p in enumerate(z):
+            best, bd = 0, float("inf")
+            for ci, c in enumerate(centers):
+                d = _dist(p, c)
+                if d < bd:
+                    best, bd = ci, d
+            if labels[i] != best:
+                labels[i] = best
+                changed = True
+        for ci in range(k):
+            members = [z[i] for i in range(len(z)) if labels[i] == ci]
+            if members:
+                centers[ci] = [statistics.fmean(c) for c in zip(*members)]
+        if not changed:
+            break
+    return labels
+
+
+def _labels_by_frequency(labels: list[int]) -> dict[int, str]:
+    """Map raw cluster ids → A/B/C so 'A' is always the most-talked speaker."""
+    from collections import Counter
+    counts = Counter(labels).most_common()
+    names = "ABCDEFG"
+    return {lab: names[i] for i, (lab, _n) in enumerate(counts)}
+
+
 def _speaker_summary(segs: list[Segment]) -> list[dict[str, Any]]:
     out: dict[str, dict[str, Any]] = {}
     for s in segs: