File size: 4,505 Bytes
c0395f4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | """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")
|