Add pipeline code

_pipeline/acx_check.py

@@ -0,0 +1,106 @@
+"""Deterministic ACX checker (R13). Measures each MP3 and prints a pass/fail table. Exit code 1 if anything fails.
+Checks per file: MP3 CBR bitrate, sample rate, channels, RMS, peak, noise floor, room tone head and tail, duration.
+Checks per set: all files mono or all stereo; each script has an MP3; chapter scripts start with a spoken header."""
+import glob
+import json
+import math
+import os
+import subprocess
+
+import audiomeasure as am
+
+
+def packet_sizes(path):
+    out = subprocess.run(["ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", "packet=size", "-of", "default=nw=1:nk=1", path],
+                         capture_output=True, text=True, check=True).stdout.split()
+    return [int(x) for x in out]
+
+
+def measure(cfg, path):
+    """All numbers for one file, as a dict."""
+    a = cfg["acx"]
+    info = am.probe(path)
+    w = am.windows(path)
+    rms = [r for r, _ in w]
+    win_s = 0.05
+    loud = [i for i, r in enumerate(rms) if r > a["speech_window_db"]]
+    sizes = packet_sizes(path)
+    blocks = [am.db_mean_power(rms[i:i + 10]) for i in range(0, len(rms) - 9, 10)]
+    return {
+        "codec": info["codec_name"],
+        "sample_rate": int(info["sample_rate"]),
+        "channels": int(info["channels"]),
+        "bitrate_kbps": round(int(info["bit_rate"]) / 1000),
+        "cbr": max(sizes) - min(sizes) <= 1,
+        "duration_s": float(info["duration"]),
+        "rms_db": am.db_mean_power(rms),
+        "peak_db": max(p for _, p in w),
+        "noise_floor_db": min(blocks) if blocks else float("nan"),
+        "head_s": loud[0] * win_s if loud else float("nan"),
+        "tail_s": (len(rms) - 1 - loud[-1]) * win_s if loud else float("nan"),
+    }
+
+
+def judge(cfg, name, m):
+    """List of (check name, value text, passed)."""
+    a = cfg["acx"]
+    retail = os.path.basename(name).startswith("retail-sample")
+    dur_ok = a["retail_min_s"] <= m["duration_s"] <= a["retail_max_s"] if retail else m["duration_s"] < a["max_minutes"] * 60
+    ok = lambda v, lo, hi: (not math.isnan(v)) and lo <= v <= hi
+    return [
+        ("format", "%s %s %dk %d Hz" % (m["codec"], "CBR" if m["cbr"] else "VBR", m["bitrate_kbps"], m["sample_rate"]),
+         m["codec"] == "mp3" and m["cbr"] and m["bitrate_kbps"] >= a["min_bitrate_kbps"] and m["sample_rate"] == a["sample_rate"]),
+        ("channels", str(m["channels"]), m["channels"] in (1, 2)),
+        ("rms dB", "%.1f" % m["rms_db"], ok(m["rms_db"], a["rms_min_db"], a["rms_max_db"])),
+        ("peak dB", "%.1f" % m["peak_db"], m["peak_db"] <= a["peak_max_db"]),
+        ("noise floor dB", "%.1f" % m["noise_floor_db"], m["noise_floor_db"] <= a["noise_floor_max_db"]),
+        ("head s", "%.2f" % m["head_s"], ok(m["head_s"], a["head_min_s"], a["head_max_s"])),
+        ("tail s", "%.2f" % m["tail_s"], ok(m["tail_s"], a["tail_min_s"], a["tail_max_s"])),
+        ("length", "%.1f s" % m["duration_s"], dur_ok),
+    ]
+
+
+def set_checks(cfg, folder, measured):
+    """Checks across files; folder is an audio/ folder next to script/ (or None)."""
+    rows = []
+    chans = {m["channels"] for m in measured.values()}
+    rows.append(("all files mono or all stereo", ",".join(map(str, sorted(chans))), len(chans) <= 1))
+    scripts = os.path.join(os.path.dirname(folder), "script") if folder else None
+    if scripts and os.path.isdir(scripts):
+        for sp in sorted(glob.glob(os.path.join(scripts, "*.json"))):
+            stem = os.path.splitext(os.path.basename(sp))[0]
+            rows.append(("mp3 exists for script " + stem, "yes" if stem in {os.path.splitext(os.path.basename(n))[0] for n in measured} else "no",
+                         stem in {os.path.splitext(os.path.basename(n))[0] for n in measured}))
+            if stem.startswith("chapter-"):
+                with open(sp, encoding="utf-8") as f:
+                    first = json.load(f)[0]["kind"]
+                rows.append(("spoken section header in " + stem, first, first == "header"))
+    return rows
+
+
+def run(cfg, paths):
+    files, folder = [], None
+    for p in paths:
+        if os.path.isdir(p):
+            folder = os.path.abspath(p)
+            files += sorted(glob.glob(os.path.join(p, "*.mp3")))
+        else:
+            files.append(p)
+    if not files:
+        print("no MP3 files found")
+        return 1
+    measured, failed = {}, False
+    for f in files:
+        m = measure(cfg, f)
+        measured[f] = m
+        print("\n%s" % f)
+        print("  %-16s %-26s %s" % ("check", "value", "result"))
+        for name, value, passed in judge(cfg, f, m):
+            failed |= not passed
+            print("  %-16s %-26s %s" % (name, value, "PASS" if passed else "FAIL"))
+    print("\nSet checks")
+    for name, value, passed in set_checks(cfg, folder, measured):
+        failed |= not passed
+        print("  %-40s %-10s %s" % (name, value, "PASS" if passed else "FAIL"))
+    print("\nOVERALL: %s" % ("FAIL" if failed else "PASS"))
+    return 1 if failed else 0