soothe2-re/verify_bit_exact.py

#!/usr/bin/env python3
"""bit-exact verification — harness out vs plugin *_ref.wav, sample-level report.

Usage:
  verify_bit_exact.py                     # sweep all *.rpp in TEST_ROOT (input=basename.wav)
  verify_bit_exact.py --rpp comb_base     # single case (uses comb_base.rpp + input wav + _ref.wav)
  verify_bit_exact.py --mono              # pipeline operates on channel-average (MEAN)

For each case `name`: read params from `name.rpp` (rpp_allparams), render
`<input>.wav -> /tmp/out_<name>.wav` via dsp/harness, compare samples against
`TEST_ROOT/<name>_ref.wav` (plugin output). Reports samples differing, max |dx|,
RMSE and first mismatching sample.
"""
import subprocess
import sys
import os
import wave
import struct
import glob
import json

HARNESS = "/home/m/re-tools/dsp/build/soothe2_harness"
TEST_ROOT = "/home/m/soothe-bt"
RPPMOD = os.path.join("/home/m/re-tools/handoff", "rpp_allparams.py")
MAX_FRAMES = None  # set to int to compare only first N frames

def read_wav3(path):
    with wave.open(path, 'rb') as w:
        n = w.getnframes()
        nc = w.getnchannels()
        sw = w.getsampwidth()
        extra = w.readframes(n)
    if sw == 3:
        raw = bytearray(extra)
        # make int list
        if len(raw) % 3:
            raw += b'\x00' * (3 - len(raw) % 3)
        vals = []
        for i in range(0, len(raw), 3):
            v = raw[i] | (raw[i+1] << 8) | (raw[i+2] << 16)
            if v & 0x800000:
                v -= 0x1000000
            vals.append(v)
        return vals, nc, sw
    elif sw == 2:
        vals = list(struct.unpack(f'<{len(extra)//2}h', extra))
        return vals, nc, sw
    raise ValueError(f"unsupported sampwidth {sw}")

def rpp_meta(name):
    """Return (input_path, ref_path, conf_path) from RPP project file."""
    rpp = f"{TEST_ROOT}/{name}.rpp"
    if not os.path.exists(rpp):
        return None, None, None
    src = ref = None
    in_src = False
    with open(rpp, errors='replace') as f:
        for line in f:
            if '<SOURCE WAVE' in line and src is None:
                in_src = True
                continue
            if in_src and 'FILE "' in line and src is None:
                i = line.find('FILE "')
                src = line[i + 6:].split('"')[0]
                in_src = False
            if 'RENDER_FILE' in line and ref is None:
                i = line.find('"')
                if i >= 0:
                    ref = line[i + 1:].split('"')[0]
    conf = f"/tmp/{name}.conf"
    with open(conf, 'w') as cf:
        subprocess.run([sys.executable, RPPMOD, "--flat", rpp],
                       stdout=cf, check=False)
    if os.path.getsize(conf) == 0:
        conf = None
    return src, ref, conf

def first_mismatch(a, b, lim=50000):
    for i in range(min(len(a), len(b), lim)):
        if a[i] != b[i]:
            return i
    return -1

def compare(name, in_wav, mono, report=3):
    src, ref, conf = rpp_meta(name)
    if not conf:
        return None, "params decode failed"
    if not src or not os.path.exists(src):
        return None, f"source wav missing: {src}"
    if not ref or not os.path.exists(ref):
        return None, f"ref wav missing: {ref}"
    out = f"/tmp/out_{name}.wav"

    cmd = [HARNESS, in_wav, out, conf]
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    if r.returncode != 0:
        return None, f"harness rc={r.returncode}: {r.stderr[:200]}"

    ra, rc, rw = read_wav3(ref)
    oa, oc, ow = read_wav3(out)
    if not ra or not oa:
        return None, "empty wav"
    if len(ra) < len(oa):
        ra = ra[:len(oa)] + [0] * (len(oa) - len(ra))

    if mono:
        # combine channels into mono reference (L+R)/2 style not exact here:
        # use channel-mean of both, compare only channel 0
        ra = list(ra[0::rc])
        oa = list(oa[0::oc])
        rc = oc = 1

    if MAX_FRAMES:
        cut = MAX_FRAMES * rc
        ra, oa = ra[:cut], oa[:cut]

    n = min(len(ra), len(oa))
    diff = sum(1 for i in range(n) if ra[i] != oa[i])
    mx = max((abs(ra[i] - oa[i]) for i in range(n)), default=0)
    if n:
        rmse = (sum((ra[i] - oa[i]) ** 2 for i in range(n)) / n) ** 0.5
    else:
        rmse = 0.0
    mm = first_mismatch(ra, oa)
    return dict(frames=n, diff=diff, frac=diff / n if n else 1.0,
                mx=mx, rmse=rmse, first=mm), None

def main():
    args = [a for a in sys.argv[1:] if not a.startswith('--')]
    mono = '--mono' in sys.argv[1:]
    single = args[0] if args else None
    if single:
        names = [single]
    else:
        names = sorted(os.path.splitext(os.path.basename(p))[0]
                       for p in glob.glob(f"{TEST_ROOT}/*.rpp"))
    print(f"{len(names)} cases  ('--mono' = single channel)")
    stat = dict(bit=0, nearbit=0, diff=0, skip=0)

    for name in names:
        src, ref, conf = rpp_meta(name)
        if not os.path.exists(src) if src else True:
            stat['skip'] += 1
            continue
        res, err = compare(name, src, mono)
        if res is None:
            print(f"  SKIP {name}: {err}")
            stat['skip'] += 1
            continue
        frac = res['frac']
        if frac == 0.0:
            stat['bit'] += 1
            tag = "BIT-EXACT"
        elif frac < 0.05:
            stat['nearbit'] += 1
            tag = "near"
        else:
            stat['diff'] += 1
            tag = "DIFF"
        print(f"  [{tag:8s}] {name:28s} frames={res['frames']:7d} "
              f"diff={res['diff']:7d} ({res['frac']*100:5.2f}%) "
              f"max|dx|={res['mx']} rmse={res['rmse']:.1f} first@{res['first']}")
    print("\n" + json.dumps(stat, indent=0))

if __name__ == "__main__":
    main()