From f5242a4252a8f62ddd7a89b5051e3415304216c1 Mon Sep 17 00:00:00 2001 From: Ankur Sinha <14087196+sinhaankur@users.noreply.github.com> Date: Sat, 11 Jul 2026 17:22:54 -0400 Subject: [PATCH] Perfect short clips into ideal XTTS references (refine.py) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The insight: XTTS-v2 is zero-shot — it clones from ~6s of *clean* speech. So when you only have a little audio of a loved one, the win is making those seconds immaculate, not gathering more. - refine.py: score every segment (SNR/voiced/clipping/length), pick the single cleanest contiguous window, render 24kHz mono gently-normalized (XTTS's sweet spot), and emit a quality report + warnings. Returns JSON. - analyze.py: separate speakers by voice TIMBRE (log-mel band energies + pitch, k-means) instead of 3 fixed pitch bands — so a mother and a daughter at the same pitch no longer merge. Pure-Python FFT with a numpy fast path; pitch-band fallback kept. - test_refine.py: 7 offline synthetic tests (scoring, window selection, same-pitch timbre separation, single-speaker, FFT correctness). - README: document refine.py and the 'clean beats long' guidance. Verified end-to-end: an 11s three-speaker mix → a 24kHz mono reference of just the dominant speaker, with an honest quality report. --- README.md | 33 ++++- analyze.py | 241 ++++++++++++++++++++++++++++++-- refine.py | 366 +++++++++++++++++++++++++++++++++++++++++++++++++ test_refine.py | 150 ++++++++++++++++++++ 4 files changed, 775 insertions(+), 15 deletions(-) create mode 100644 refine.py create mode 100644 test_refine.py diff --git a/README.md b/README.md index f3db489..1013a16 100644 --- a/README.md +++ b/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. --- diff --git a/analyze.py b/analyze.py index de2a5b1..f3ed235 100644 --- a/analyze.py +++ b/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: diff --git a/refine.py b/refine.py new file mode 100644 index 0000000..a35bcda --- /dev/null +++ b/refine.py @@ -0,0 +1,366 @@ +""" +refine — turn a short, possibly-noisy clip into the *ideal* XTTS reference. + +The insight this module is built on: XTTS-v2 is a **zero-shot** cloner — it clones +from ~6 seconds of *clean* speech. So when you only have a little audio (even a +few seconds of a loved one), the win isn't gathering more — it's making those +seconds immaculate and handing XTTS exactly the reference it wants: + + * her voice only (no music, no other speakers) ← engine.py Demucs + analyze.py + * the single cleanest window (no silence/noise/clipping/overlap) ← scoring here + * 24 kHz mono, gently normalized, 6–20s long ← XTTS's sweet spot + +This module does the scoring + selection + XTTS-tuned rendering, and emits a +small **quality report** so you can trust the sample before cloning. + +Everything is local and stdlib-only for the analysis (numpy used only if present, +with a pure-Python fallback). It reuses engine.py for ffmpeg/Demucs and analyze.py +for transcription + speaker grouping. +""" + +from __future__ import annotations + +import array +import math +import os +import sys +import tempfile +import wave +from dataclasses import dataclass, asdict, field +from typing import Any, Callable, Optional + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import engine # noqa: E402 +import analyze # noqa: E402 + +# XTTS-v2 renders at 24 kHz; giving it a 24 kHz mono reference avoids an internal +# resample and matches the model's sweet spot. +XTTS_SR = 24000 +# XTTS clones well from ~6–20s. Below MIN it gets unstable; above MAX adds nothing +# and can dilute with weaker moments. +REF_MIN_SEC = 6.0 +REF_TARGET_SEC = 12.0 +REF_MAX_SEC = 20.0 + + +# ---------------------------------------------------------------------------- +# audio helpers (mono float samples) +# ---------------------------------------------------------------------------- +def _read_wav_mono(path: str) -> tuple[list[float], int]: + w = wave.open(path, "rb") + sr, ch, n = w.getframerate(), w.getnchannels(), w.getnframes() + raw = w.readframes(n) + w.close() + a = array.array("h") + a.frombytes(raw) + data = list(a) + if ch == 2: + data = [(data[i] + data[i + 1]) * 0.5 for i in range(0, len(data) - 1, 2)] + return [x / 32768.0 for x in data], sr + + +def _rms(xs: list[float]) -> float: + if not xs: + return 0.0 + return math.sqrt(sum(x * x for x in xs) / len(xs)) + + +def _clip_ratio(xs: list[float], thresh: float = 0.98) -> float: + """Fraction of samples riding the rails — clipped audio clones badly.""" + if not xs: + return 1.0 + return sum(1 for x in xs if abs(x) >= thresh) / len(xs) + + +def _voiced_ratio(xs: list[float], sr: int) -> float: + """Fraction of short frames that carry real energy (a proxy for how much of + the window is actual speech vs. silence/room tone).""" + if not xs: + return 0.0 + frame = max(1, int(sr * 0.03)) + total = voiced = 0 + noise = _noise_floor(xs, sr) + gate = max(noise * 3.0, 0.01) + for i in range(0, len(xs) - frame, frame): + total += 1 + if _rms(xs[i:i + frame]) > gate: + voiced += 1 + return voiced / total if total else 0.0 + + +def _noise_floor(xs: list[float], sr: int) -> float: + """Estimate the noise floor as the 10th-percentile frame RMS (quiet frames + are room tone/hiss). Used for a rough SNR.""" + frame = max(1, int(sr * 0.03)) + rms_frames = [_rms(xs[i:i + frame]) for i in range(0, len(xs) - frame, frame)] + if not rms_frames: + return 0.0 + rms_frames.sort() + return rms_frames[max(0, int(len(rms_frames) * 0.10))] + + +def _snr_db(xs: list[float], sr: int) -> float: + """Rough SNR in dB: speech RMS (75th pct frame) over noise floor (10th pct).""" + frame = max(1, int(sr * 0.03)) + rms_frames = sorted(_rms(xs[i:i + frame]) for i in range(0, len(xs) - frame, frame)) + if len(rms_frames) < 4: + return 0.0 + noise = rms_frames[max(0, int(len(rms_frames) * 0.10))] or 1e-6 + speech = rms_frames[min(len(rms_frames) - 1, int(len(rms_frames) * 0.75))] + return round(20.0 * math.log10(max(speech, 1e-6) / noise), 1) + + +# ---------------------------------------------------------------------------- +# per-segment scoring +# ---------------------------------------------------------------------------- +@dataclass +class ScoredSegment: + index: int + start: float + end: float + text: str + speaker: str + duration: float + snr_db: float + voiced: float # 0..1 fraction that's speech + clip: float # 0..1 fraction clipped (lower is better) + score: float # overall 0..1 (higher is better) + + +def _score_segment(xs: list[float], sr: int, dur: float) -> tuple[float, dict]: + """Score a chunk for cloning suitability. Rewards clean, voiced, unclipped + speech of a useful length; punishes noise, silence, clipping, and too-short.""" + snr = _snr_db(xs, sr) + voiced = _voiced_ratio(xs, sr) + clip = _clip_ratio(xs) + # component scores, each 0..1 + s_snr = max(0.0, min(1.0, (snr - 5.0) / 25.0)) # 5 dB→0, 30 dB→1 + s_voiced = max(0.0, min(1.0, (voiced - 0.4) / 0.5)) # want mostly speech + s_clip = max(0.0, 1.0 - clip * 40.0) # any real clipping hurts + s_len = max(0.0, min(1.0, dur / 4.0)) # up to ~4s per seg is fine + score = 0.42 * s_snr + 0.28 * s_voiced + 0.20 * s_clip + 0.10 * s_len + return round(score, 3), {"snr_db": snr, "voiced": round(voiced, 2), + "clip": round(clip, 4)} + + +# ---------------------------------------------------------------------------- +# window selection: find the best contiguous run of segments +# ---------------------------------------------------------------------------- +def _best_window(scored: list[ScoredSegment]) -> list[ScoredSegment]: + """Choose the contiguous run of segments (same speaker, in time order) whose + total duration lands in [REF_MIN, REF_MAX] with the highest mean score. A + contiguous window keeps natural prosody, which XTTS prefers over stitched + scraps. Falls back to the top-scoring segments if nothing hits REF_MIN.""" + if not scored: + return [] + ordered = sorted(scored, key=lambda s: s.start) + best: list[ScoredSegment] = [] + best_val = -1.0 + for i in range(len(ordered)): + run: list[ScoredSegment] = [] + dur = 0.0 + for j in range(i, len(ordered)): + seg = ordered[j] + # break the run on a speaker change (don't blend two people) + if run and seg.speaker != run[0].speaker: + break + run.append(seg) + dur += seg.duration + if dur >= REF_MIN_SEC: + # mean score, lightly favoring windows near REF_TARGET_SEC + mean = sum(s.score for s in run) / len(run) + closeness = 1.0 - min(1.0, abs(dur - REF_TARGET_SEC) / REF_TARGET_SEC) + val = mean * (0.85 + 0.15 * closeness) + if val > best_val and dur <= REF_MAX_SEC + 4: + best_val, best = val, list(run) + if dur >= REF_MAX_SEC: + break + if best: + return best + # Nothing reached REF_MIN. Fall back to the best segments of a SINGLE speaker + # (never blend two people): pick the speaker with the most total duration, + # then take their highest-scoring segments. + by_speaker: dict[str, list[ScoredSegment]] = {} + for s in ordered: + by_speaker.setdefault(s.speaker, []).append(s) + top_speaker = max(by_speaker.values(), key=lambda g: sum(s.duration for s in g)) + return sorted(top_speaker, key=lambda s: -s.score)[:6] + + +# ---------------------------------------------------------------------------- +# the main entry point +# ---------------------------------------------------------------------------- +@dataclass +class RefineResult: + ok: bool + out: Optional[str] = None + error: Optional[str] = None + duration: float = 0.0 + snr_db: float = 0.0 + speaker: str = "?" + picked_segments: int = 0 + report: list[str] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + + +def build_reference( + src: str, + out_path: str, + *, + speaker: Optional[str] = None, + use_demucs: Optional[bool] = None, + log: Callable[[str], None] = lambda m: None, +) -> RefineResult: + """From a video/audio file, build the ideal XTTS reference WAV: + + 1. isolate the voice (Demucs) + transcribe + group speakers (analyze.py) + 2. score every segment for cloning suitability + 3. pick the single cleanest contiguous window (optionally for one speaker) + 4. render it to 24 kHz mono, gently normalized, and report the quality + + Returns a RefineResult with the output path, a quality report, and warnings. + """ + if not engine.have_ffmpeg(): + return RefineResult(ok=False, error="ffmpeg not found") + if not os.path.isfile(src): + return RefineResult(ok=False, error=f"file not found: {src}") + if os.path.splitext(src)[1].lower() not in engine.SUPPORTED_EXTS: + return RefineResult(ok=False, error=f"unsupported file type: {src}") + + log("Analyzing (isolate → transcribe → group speakers)…") + try: + res = analyze.analyze(src, use_demucs=use_demucs, log=log) + except engine.ProcessingError as e: + return RefineResult(ok=False, error=str(e)) + if not res.get("ok"): + return RefineResult(ok=False, error=res.get("error", "analysis failed")) + + speakers = res.get("speakers", []) + # default to the dominant speaker (most seconds) unless one is named + if speaker is None and speakers: + speaker = speakers[0]["speaker"] + + with tempfile.TemporaryDirectory() as tmp: + # re-derive the isolated, mono 16k voice track once for scoring + cutting + raw = os.path.join(tmp, "raw.wav") + try: + engine._extract_audio(src, raw) + except engine.ProcessingError as e: + return RefineResult(ok=False, error=str(e)) + voice = raw + if (engine.have_demucs() if use_demucs is None else use_demucs): + try: + voice = engine._isolate_with_demucs(raw, tmp) + except engine.ProcessingError: + voice = raw + work = os.path.join(tmp, "work.wav") + engine._run(["ffmpeg", "-y", "-i", voice, "-ar", "16000", "-ac", "1", + "-c:a", "pcm_s16le", work]) + data, sr = _read_wav_mono(work) + + # score each segment (optionally filtered to the chosen speaker) + scored: list[ScoredSegment] = [] + for s in res["segments"]: + if speaker is not None and s["speaker"] not in (speaker, "?"): + continue + a, b = int(s["start"] * sr), int(s["end"] * sr) + chunk = data[a:b] + if len(chunk) < sr * 0.25: # ignore <0.25s scraps + continue + sc, m = _score_segment(chunk, sr, s["duration"]) + scored.append(ScoredSegment( + index=s["index"], start=s["start"], end=s["end"], + text=s.get("text", ""), speaker=s["speaker"], duration=s["duration"], + snr_db=m["snr_db"], voiced=m["voiced"], clip=m["clip"], score=sc)) + + if not scored: + return RefineResult(ok=False, error="no usable speech found in the clip") + + window = _best_window(scored) + picked_dur = round(sum(s.duration for s in window), 1) + log(f"Picked {len(window)} segment(s) · {picked_dur}s of the cleanest speech.") + + # cut + concatenate the chosen window from the ISOLATED voice, then render + # the XTTS-tuned reference (24k mono, gentle normalization). + parts = [] + for i, s in enumerate(sorted(window, key=lambda x: x.start)): + p = os.path.join(tmp, f"p{i}.wav") + engine._run(["ffmpeg", "-y", "-i", voice, "-ss", str(s.start), + "-to", str(s.end), "-ar", str(XTTS_SR), "-ac", "1", + "-c:a", "pcm_s16le", p]) + parts.append(p) + merged = os.path.join(tmp, "merged.wav") + if len(parts) == 1: + os.replace(parts[0], merged) + else: + engine.merge_wavs(parts, merged, gap_seconds=0.25) + _xtts_normalize(merged, out_path) + + # measure the finished reference for the report + fin, fsr = _read_wav_mono(out_path) + dur = round(len(fin) / fsr, 1) + snr = _snr_db(fin, fsr) + report, warnings = _build_report(dur, snr, window, speakers, speaker) + return RefineResult( + ok=True, out=out_path, duration=dur, snr_db=snr, speaker=speaker or "?", + picked_segments=len(window), report=report, warnings=warnings) + + +def _xtts_normalize(in_wav: str, out_wav: str) -> None: + """XTTS-tuned finish: 24 kHz mono, a light high-pass to kill rumble, gentle + peak-safe normalization (NOT broadcast loudnorm, which over-compresses and + hurts cloned timbre).""" + cp = engine._run([ + "ffmpeg", "-y", "-i", in_wav, + "-af", "highpass=f=70,dynaudnorm=f=200:g=8:p=0.9,alimiter=limit=0.95", + "-ar", str(XTTS_SR), "-ac", "1", "-c:a", "pcm_s16le", out_wav, + ]) + if cp.returncode != 0 or not os.path.exists(out_wav): + # fall back to a plain resample so we still produce a usable file + engine._run(["ffmpeg", "-y", "-i", in_wav, "-ar", str(XTTS_SR), + "-ac", "1", "-c:a", "pcm_s16le", out_wav]) + + +def _build_report(dur: float, snr: float, window: list[ScoredSegment], + speakers: list[dict], chosen: Optional[str] + ) -> tuple[list[str], list[str]]: + report = [ + f"reference: {dur}s of clean speech @ {XTTS_SR//1000} kHz mono", + f"est. SNR: {snr} dB " + + ("(excellent)" if snr >= 20 else "(good)" if snr >= 12 + else "(usable)" if snr >= 7 else "(noisy — consider a cleaner clip)"), + f"segments used: {len(window)}", + ] + if len(speakers) > 1: + report.append( + f"speakers detected: {len(speakers)} — used '{chosen}' " + f"({next((s['seconds'] for s in speakers if s['speaker']==chosen), '?')}s, " + f"{next((s.get('likely','') for s in speakers if s['speaker']==chosen), '')})") + warnings = [] + if dur < REF_MIN_SEC: + warnings.append( + f"only {dur}s — XTTS works from ~6s; this may clone but add another " + f"clip for a truer voice.") + if snr < 7: + warnings.append("low SNR — background noise may bleed into the clone; " + "try a cleaner source or install demucs.") + if len(speakers) > 1: + warnings.append("multiple speakers were present — verify the chosen voice " + "is the right person (listen to the file).") + return report, warnings + + +if __name__ == "__main__": + import argparse + import json + ap = argparse.ArgumentParser(description="Build the ideal XTTS reference from a clip.") + ap.add_argument("src") + ap.add_argument("--out", default="reference.wav") + ap.add_argument("--speaker", help="speaker label to keep (default: dominant)") + ap.add_argument("--no-demucs", action="store_true") + a = ap.parse_args() + r = build_reference(a.src, a.out, + speaker=a.speaker, + use_demucs=False if a.no_demucs else None, + log=lambda m: print(m, file=sys.stderr)) + print(json.dumps(asdict(r), indent=2)) diff --git a/test_refine.py b/test_refine.py new file mode 100644 index 0000000..160755e --- /dev/null +++ b/test_refine.py @@ -0,0 +1,150 @@ +""" +Tests for the refine/analyze upgrades — segment scoring, best-window selection, +and timbre-based speaker separation. All offline & synthetic (no ffmpeg, no +model, no real audio files needed), so they run anywhere in ~a second. + +Run: python3 test_refine.py (or: python3 -m pytest test_refine.py) +""" + +from __future__ import annotations + +import math +import os +import random +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import refine +import analyze +from refine import ScoredSegment as S +from analyze import Segment + + +SR = 16000 + + +def _speechlike(voiced_pattern, noise=0.002, seg_sec=0.5): + """Build a mono signal: bursts of a 150 Hz tone with quiet gaps + noise floor. + voiced_pattern is a list of bools, one per 0.5s segment.""" + random.seed(1) + xs = [] + for voiced in voiced_pattern: + for i in range(int(SR * seg_sec)): + base = 0.3 * math.sin(2 * math.pi * 150 * i / SR) if voiced else 0.0 + xs.append(base + random.uniform(-noise, noise)) + return xs + + +def test_scoring_ranks_clean_above_noisy(): + speech = _speechlike([True, False] * 5) + noisy = [x + random.uniform(-0.15, 0.15) for x in speech] + clean_score, cm = refine._score_segment(speech, SR, 5.0) + noisy_score, nm = refine._score_segment(noisy, SR, 5.0) + assert clean_score > noisy_score, (clean_score, noisy_score) + assert cm["snr_db"] > nm["snr_db"] + print(f"✓ scoring: clean {clean_score} > noisy {noisy_score} (SNR {cm['snr_db']} vs {nm['snr_db']})") + + +def test_scoring_penalizes_clipping(): + clean = _speechlike([True, True, True, True]) + clipped = [max(-1.0, min(1.0, x * 6.0)) for x in clean] # drive into the rails + cs, _ = refine._score_segment(clean, SR, 2.0) + ps, pm = refine._score_segment(clipped, SR, 2.0) + assert pm["clip"] > 0.0 + assert cs > ps + print(f"✓ scoring: clipping penalized ({cs} > {ps}, clip={pm['clip']})") + + +def test_best_window_picks_highest_scoring_run(): + # segment 1 is weak; the best 6s run should be [2,3] + segs = [ + S(0, 0, 3, "a", "A", 3, 25, 0.9, 0.0, 0.90), + S(1, 3, 6, "b", "A", 3, 8, 0.6, 0.0, 0.40), + S(2, 6, 9, "c", "A", 3, 24, 0.9, 0.0, 0.88), + S(3, 9, 12, "d", "A", 3, 26, 0.92, 0.0, 0.91), + ] + win = refine._best_window(segs) + dur = sum(s.duration for s in win) + assert dur >= refine.REF_MIN_SEC + assert 1 not in [s.index for s in win], [s.index for s in win] + print(f"✓ window: picked {[s.index for s in win]} ({dur:.0f}s), skipped the weak seg") + + +def test_best_window_never_blends_two_speakers(): + segs = [ + S(0, 0, 4, "a", "A", 4, 25, 0.9, 0.0, 0.9), + S(1, 4, 8, "b", "B", 4, 25, 0.9, 0.0, 0.9), # different speaker + S(2, 8, 12, "c", "A", 4, 25, 0.9, 0.0, 0.9), + ] + win = refine._best_window(segs) + speakers = set(s.speaker for s in win) + assert len(speakers) == 1, speakers + print(f"✓ window: stays within one speaker ({speakers})") + + +def _voice(pitch, formant_boost, dur=1.0): + """A harmonic 'voice' at `pitch` with energy emphasized near `formant_boost` + Hz — fakes a distinct vocal tract at the same pitch.""" + random.seed(0) + xs = [] + for i in range(int(SR * dur)): + t = i / SR + s = 0.0 + for h in range(1, 15): + f = pitch * h + amp = 0.3 / h + if abs(f - formant_boost) < 400: + amp *= 2.5 + s += amp * math.sin(2 * math.pi * f * t) + xs.append(s * 0.3 + random.uniform(-0.002, 0.002)) + return xs + + +def test_speaker_separation_by_timbre_not_just_pitch(): + """The case the old pitch-band code failed: two speakers at the SAME pitch + but different formants must NOT be merged.""" + data, segs, t = [], [], 0.0 + for n in range(8): + formant = 900 if n % 2 == 0 else 2400 + data += _voice(180, formant, 1.0) + segs.append(Segment(n, t, t + 1.0, "t", 180.0, 0.3)) + t += 1.0 + analyze._assign_speakers(segs, data, SR) + labels = [s.speaker for s in segs] + even = set(labels[i] for i in range(0, 8, 2)) + odd = set(labels[i] for i in range(1, 8, 2)) + assert len(set(labels)) >= 2, labels + assert even.isdisjoint(odd), (even, odd) + print(f"✓ speakers: same-pitch voices separated by timbre ({labels})") + + +def test_single_speaker_stays_one_group(): + """A clip of one voice should not be over-split into phantom speakers.""" + data, segs, t = [], [], 0.0 + for n in range(6): + data += _voice(180, 1200, 1.0) # same voice throughout + segs.append(Segment(n, t, t + 1.0, "t", 180.0, 0.3)) + t += 1.0 + analyze._assign_speakers(segs, data, SR) + labels = set(s.speaker for s in segs) + assert labels == {"A"}, labels + print(f"✓ speakers: one voice stays a single group ({labels})") + + +def test_fft_matches_numpy_when_available(): + """The pure-Python FFT magnitude should track numpy's rfft (correctness).""" + frame = [math.sin(2 * math.pi * 220 * i / SR) for i in range(512)] + mag = analyze._fft_mag(frame) + # peak bin should correspond to ~220 Hz + peak = max(range(1, len(mag)), key=lambda k: mag[k]) + freq = peak * SR / 512 + assert abs(freq - 220) < SR / 512 * 2, freq + print(f"✓ fft: dominant bin ≈ {freq:.0f} Hz (expected 220)") + + +if __name__ == "__main__": + fns = [v for k, v in sorted(globals().items()) + if k.startswith("test_") and callable(v)] + for fn in fns: + fn() + print(f"\nall {len(fns)} refine/analyze tests passed.")