scripts/dataset/check_clips.py

"""Direct integrity sweep of the acoustic-space dataset.

Reads WAV headers (no ffprobe subprocess) for every file actually used in
training, plus the sliced source set. Checks for truncation, duration spread,
format drift, and per-pair reference/target mismatch.
"""
import csv
import wave
from collections import Counter, defaultdict
from pathlib import Path

SLICED = Path("/workspace/AUDIO-LTX-LORA/Dataset/Sliced")
DATA = Path("/workspace/Demos/data/acoustic-space-ableton/audio")


def probe(p):
    """(duration_s, sample_rate, channels, n_frames) or None if unreadable."""
    try:
        with wave.open(str(p), "rb") as w:
            n, sr, ch = w.getnframes(), w.getframerate(), w.getnchannels()
            return (n / sr if sr else -1.0, sr, ch, n)
    except Exception as e:
        return ("ERR:" + type(e).__name__, None, None, None)


def sweep(label, root):
    files = sorted(root.rglob("*.wav"))
    print(f"\n=== {label}: {len(files)} wav files ===")
    if not files:
        return {}
    durs, fmts, bad = Counter(), Counter(), []
    info = {}
    for f in files:
        d, sr, ch, n = probe(f)
        if isinstance(d, str):
            bad.append((f.name, d))
            continue
        info[f.name] = (d, sr, ch, n)
        durs[round(d, 3)] += 1
        fmts[(sr, ch)] += 1
    print("  durations:")
    for d, c in sorted(durs.items()):
        print(f"    {d:>7.3f}s  x{c}")
    print(f"  formats: {dict(fmts)}")
    if bad:
        print(f"  !! UNREADABLE ({len(bad)}):")
        for name, err in bad:
            print(f"     {name}  {err}")
    else:
        print("  all files readable")
    return info


# --- the two sets that matter -------------------------------------------
sliced = sweep("SLICED sources", SLICED)
refs = sweep("TRAINING references (dry inputs)", DATA / "references")
tgts = sweep("TRAINING targets (wet outputs)", DATA / "targets")

# --- per-pair reference vs target ---------------------------------------
print("\n=== PER-PAIR reference vs target duration ===")
mismatch = []
for name, (d_t, *_rest) in sorted(tgts.items()):
    if name not in refs:
        mismatch.append((name, "NO MATCHING REFERENCE", ""))
        continue
    d_r = refs[name][0]
    if abs(d_r - d_t) > 0.01:
        mismatch.append((name, f"ref {d_r:.3f}s", f"tgt {d_t:.3f}s"))
only_ref = set(refs) - set(tgts)
for name in sorted(only_ref):
    mismatch.append((name, "NO MATCHING TARGET", ""))
if mismatch:
    print(f"  !! {len(mismatch)} mismatched pairs:")
    for row in mismatch:
        print(f"     {row[0]:<46} {row[1]} {row[2]}")
else:
    print(f"  all {len(tgts)} pairs matched in duration")

# --- the suspect cathedral clip -----------------------------------------
print("\n=== SUSPECT: claps_rhythm / cathedral files ===")
for label, d in (("sliced", SLICED), ("references", DATA / "references"),
                 ("targets", DATA / "targets")):
    hits = sorted(p for p in d.rglob("*.wav")
                  if "claps" in p.name.lower() or "claps" in str(p.parent).lower())
    for p in hits:
        dur, sr, ch, n = probe(p)
        size = p.stat().st_size
        durs = f"{dur:.3f}s" if not isinstance(dur, str) else dur
        print(f"  [{label:<10}] {p.name:<44} {durs:>9}  {sr}Hz {ch}ch  frames={n}  {size:,}B")

# --- manifest cross-check -----------------------------------------------
print("\n=== slices.csv (parsed properly) ===")
csv_path = SLICED / "slices.csv"
if csv_path.exists():
    with open(csv_path, newline="") as fh:
        rows = list(csv.DictReader(fh))
    print(f"  rows: {len(rows)}")
    dur_counts = Counter(r.get("duration_s") for r in rows)
    print(f"  duration_s values: {dict(dur_counts)}")
    warned = [r for r in rows if (r.get("warnings") or "").strip()]
    print(f"  rows with warnings: {len(warned)}")
    for r in warned[:10]:
        print(f"    {r.get('output')}: {r.get('warnings')}")
    print("  claps_rhythm rows:")
    for r in rows:
        if "claps" in (r.get("source_id") or "").lower():
            print(f"    {r.get('environment'):<24} {r.get('output'):<40} "
                  f"dur={r.get('duration_s')} peak={r.get('peak_dbfs')} "
                  f"rms={r.get('rms_dbfs')} tail={r.get('tail_rms_dbfs')}")
    # per-space source coverage
    print("  coverage (sources per space):")
    cov = defaultdict(set)
    for r in rows:
        cov[r.get("environment")].add(r.get("source_id"))
    for env, srcs in sorted(cov.items()):
        print(f"    {env:<24} {len(srcs)} sources")