Sylpheed/tools/port/verify-transcode-fidelity

#!/usr/bin/env python3
"""Is the transcode faithful to the source? Decode both, align, subtract.

`AUDIO-VERIFICATION.md` §1 states this as the question P4 actually raised and
gives the method, and nothing implemented it. `verify-video-audio` deliberately
does not: it proves Godot emits non-silence and says in as many words that a
difference RMS without alignment is meaningless. So the gate has rested on level
and non-silence, and the fidelity claim has never been made.

The doc names three ways the measurement lies, and all three are handled here
rather than hoped about:

  ALIGNMENT      a one-sample offset makes the difference nearly as loud as the
                 source. Cross-correlated coarse-to-fine BEFORE subtracting, and
                 the search REFUSES when its best lag sits on the boundary --
                 printing the range beside the answer, so an edge reads as an
                 edge.
  CHANNEL LAYOUT the source is 5.1 and the transcode is stereo. The source is
                 folded with `video.rs`'s own `DOWNMIX_51` -- read out of the
                 manifest's recorded command, not restated here -- so both sides
                 are the same fold.
  A PARTIAL FILE `ffprobe` once reported 33 s for a 137 s transcode because the
                 encode was still running. Duration and mtime are checked, and a
                 file written in the last 60 s is refused.
  THE SEEK       `-ss` before `-i` returned 4.6 s of AUDIO for a 4.0 s request on
                 this WMA Pro source, so the two windows covered different audio.
                 Not in §1. ⚠️ NARROWED after the Decoder checked it: on this
                 disc the VIDEO container-seek is EXACT -- a frame taken at 20 s
                 via container seek is byte-identical to one from a full decode.
                 So it is a property of the AUDIO STREAM, not of `-ss` placement
                 as such, and a check that only looked at video would clear a
                 path still unsafe for audio.

🔴 AND IT RUNS ITS OWN KNOWN NEGATIVES. A fidelity check that has only ever
returned "faithful" is the unfalsifiable clean run this project keeps finding:
`--control` compares the source against itself (must be near-perfect) and against
the OTHER movie (must be near 0 dB down).

⚠️ **REPORT ONLY. THIS DOES NOT YET PRODUCE A VERDICT**, and it is committed in
that state deliberately. It has reproduced four distinct ways the measurement
lies -- three that §1 names and one it does not -- and each was found by a
diagnostic rather than by reasoning. It still reports the difference signal
LOUDER than the source, which cannot be true of two aligned signals at equal
level, so the remaining fault is on this side of the instrument.

A tool that says "not faithful" while its own alignment is broken would be worse
than no tool: it would put a false defect on the exporter. Committed so the next
iteration starts from four known traps instead of from four lines of shell.
"""
import json, os, re, subprocess, sys, time, math, array

RATE = 48000
COARSE = 8000
WINDOW_S = 25.0
PASS_DB = 40.0
# Per-band tolerance. Both shipped transcodes sit at 0.29 and 0.66 dB worst-case
# across four bands, and the unrelated-movie control lands an order of magnitude
# out, so this is set between two measured populations rather than picked.
PASS_BAND_DB = 1.5


def sh(*a):
    return subprocess.run(a, capture_output=True).stdout


def pcm(path, rate, seconds, af=None, skip=0.0):
    """Decode to mono signed-16 at `rate`, optionally through a filter chain."""
    # 🔴 `-ss` AFTER `-i`, and this is a FOURTH way the measurement lies that
    # AUDIO-VERIFICATION §1 does not list. Placed before `-i` the seek is a
    # container-level jump, and on this WMA Pro source it overshot: a 4.0 s
    # request returned 4.6 s of audio while the Ogg side returned 4.0 s. The two
    # windows then covered DIFFERENT STRETCHES OF THE MOVIE, no shift could
    # align them, and the check reported a faithful transcode as garbage --
    # normalised correlation 0.172 at its best lag.
    #
    # Decoder-side seeking is slower and exact. The failure looks identical to
    # the alignment trap the doc does name, which is why it cost a diagnostic
    # rather than a guess to tell them apart.
    cmd = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-i", path,
           "-ss", str(skip), "-t", str(seconds)]
    if af:
        cmd += ["-af", af + ",aformat=channel_layouts=mono"]
    else:
        cmd += ["-af", "aformat=channel_layouts=mono"]
    cmd += ["-ar", str(rate), "-f", "s16le", "-"]
    raw = sh(*cmd)
    a = array.array("h")
    a.frombytes(raw[: len(raw) // 2 * 2])
    return a


def rms_db(xs):
    if not xs:
        return float("-inf")
    s = sum(float(v) * v for v in xs)
    r = math.sqrt(s / len(xs))
    return 20 * math.log10(r / 32768.0) if r > 0 else float("-inf")


def corr(a, b, lag, stride):
    """Correlation and the norms needed to normalise it, at one lag."""
    n = min(len(a), len(b)) - abs(lag)
    s = ea = eb = 0.0
    for i in range(0, n, stride):
        j = i + lag
        if 0 <= j < len(b):
            s += a[i] * b[j]
            ea += float(a[i]) * a[i]
            eb += float(b[j]) * b[j]
    return s, ea, eb


def best_lag(a, b, span, stride=3):
    """Lag maximising correlation, with the NORMALISED value so the caller can
    tell "aligned" from "there is no alignment"."""
    best = (-1e30, 0, 0.0)
    for lag in range(-span, span + 1):
        s, ea, eb = corr(a, b, lag, stride)
        if s > best[0]:
            best = (s, lag, s / math.sqrt(ea * eb) if ea > 0 and eb > 0 else 0.0)
    return best[1], best[2]


def align(src, dst, af):
    """Sample offset between the two decodes, found coarse-to-fine.

    🔴 A SINGLE-RESOLUTION SEARCH PINNED AT ITS OWN EDGE. `ADV` returned +2413
    against a window of +/-2400 -- the answer was the boundary, not the peak,
    and the check then reported a faithful transcode as a failure. Same family
    as the Decoder's period estimator returning its own search floor: an
    instrument answering with a property of itself.
    """
    for rate, span, stride in ((2000, 2000, 2), (8000, 60, 2)):
        a = pcm(src, rate, 8.0, af, skip=2.0)
        b = pcm(dst, rate, 8.0, None, skip=2.0)
        if not a or not b:
            return None, 0.0
        if rate == 2000:
            lag, c = best_lag(a, b, span, stride)
            if abs(lag) >= span:
                # Refuse AND say what the range was: the Decoder's cheap defence
                # is printing the search range beside the answer so a boundary
                # reads as a boundary rather than as a result.
                print(f"    coarse lag {lag:+d} of a +/-{span} search at {rate} Hz"
                      f" -- ON THE BOUNDARY, so this is the window's edge, not a peak")
                return None, c
            coarse = lag / rate
        else:
            centre = int(round(coarse * rate))
            sub_a, sub_b = a, b[max(0, centre):] if centre >= 0 else b
            lag, c = best_lag(sub_a, sub_b, span, stride)
            coarse += lag / rate
    return int(round(coarse * RATE)), c


# 🔴 THE TOP BAND IS SPLIT BECAUSE THE NEAR-MISS CONTROL FAILED. With a single
# 6-16 kHz band, a 6 kHz-lowpassed source -- a transcode that lost its whole top
# end, the failure this check exists to catch -- deviated by only 2.58 dB and
# would have PASSED. The band was wide enough to average the loss away against
# the filter's transition region.
#
# ⚠️ This is changing the instrument's RESOLUTION so it can see a failure it must
# see, driven by a control it failed. It is NOT loosening the pass threshold for
# the real comparison, which is unchanged -- that would be tuning until the
# answer came out right, which is the thing this project keeps catching.
BANDS = [(0, 500), (500, 2000), (2000, 6000), (6000, 10000), (10000, 16000)]

# `FID_BANDS=none` empties the band list and `FID_WINDOW` shortens the analysis
# window. Both exist ONLY so `--selftest` can drive this script as a subprocess
# in a deliberately broken configuration and read its real exit code, rather than
# reasoning about what it would do -- the failure I walked into on my first
# harness self-test and the Decoder walked into on theirs.
if os.environ.get("FID_BANDS") == "none":
    BANDS = []
WINDOW_S = float(os.environ.get("FID_WINDOW", WINDOW_S))


def band_db(path, af, lo, hi, seconds=25.0, skip=2.0):
    """RMS in one band, straight out of `astats`.

    🔴 A DIFFERENT KIND OF QUANTITY, and that is the whole reason it exists. The
    difference-signal method needs the two decodes aligned to the sample, and
    four attempts at that produced four different failures and no verdict. The
    Decoder's rule from their own two failed attempts: **two failed attempts at
    the same measurement are evidence the QUANTITY is wrong, not the parsing.**
    Band energy needs no alignment at all -- it is a statistic over the window,
    so a lag of any size cannot corrupt it.

    ⚠️ It is a WEAKER claim than a difference signal. Matching band energies
    cannot distinguish a faithful transcode from one that preserved the spectrum
    while mangling the waveform. It is what this instrument can honestly support,
    and it is stated as that rather than dressed up as fidelity.
    """
    chain = [(af + "," if af else ""), "aformat=channel_layouts=mono"]
    if lo > 0:
        chain.append(",highpass=f=%d" % lo)
    if hi < 20000:
        chain.append(",lowpass=f=%d" % hi)
    chain.append(",astats=measure_perchannel=none")
    out = subprocess.run(
        ["ffmpeg", "-hide_banner", "-i", path, "-ss", str(skip), "-t", str(seconds),
         "-af", "".join(chain), "-f", "null", "-"],
        capture_output=True, text=True).stderr
    m = re.search(r"RMS level dB: (-?[\d.]+|-inf)", out)
    if not m or m.group(1) == "-inf":
        return None
    return float(m.group(1))


def bands(src, dst, af, label, af_dst=None):
    """Per-band level, source against transcode. ROBUST to misalignment, not free of it.

    ⚠️ CLAIM NARROWED 2026-08-31 after the Decoder tried to refute it. It survives
    -- **1 s of misalignment costs 0.16 dB**, well inside the 1.5 dB pass band --
    but it is **not literally alignment-free**: at **10 s the cost reaches 1.00 dB**,
    because a fixed analysis window covers different material once the shift is
    large relative to it. "Needs no alignment" was my wording and it was too
    strong; the honest claim is robustness up to a few seconds.

    🔴 THE FOLD IS PER-SIDE, and the identity control is what made that
    necessary. `af` applies to the LEFT side only, which is correct for the real
    comparison -- a 5.1 source needs folding, an already-stereo transcode does
    not. Applying that same asymmetry to source-against-itself compares a folded
    signal with a raw six-channel average and reports **7.656 dB on an
    identity**, larger than the 0.66 dB this check calls a pass.
    """
    print(f"  {label}")
    worst = 0.0
    for lo, hi in BANDS:
        a = band_db(src, af, lo, hi)
        b = band_db(dst, af_dst, lo, hi)
        if a is None or b is None:
            print(f"    {lo:>5}-{hi:<5} Hz   one side silent -- no comparison")
            continue
        d = b - a
        worst = max(worst, abs(d))
        flag = "" if abs(d) <= 1.0 else ("  <- " + ("transcode louder" if d > 0 else "transcode quieter"))
        print(f"    {lo:>5}-{hi:<5} Hz   source {a:7.2f}   transcode {b:7.2f}"
              f"   {d:+6.2f} dB{flag}")
    return worst


def downmix_of(manifest, name):
    """The fold the EXPORTER used, read back out of the recorded command."""
    for v in manifest.get("videos", []):
        if v.get("name") == name:
            # 🔴 Take everything between `-af` and the next flag. A tighter
            # pattern truncated the fold to its FL half -- the source was being
            # folded to a left-only signal while the transcode carried both --
            # and the run reported the difference 7 dB LOUDER than the source.
            # That is AUDIO-VERIFICATION §1's channel-layout trap, reached
            # through a parsing bug rather than a decision. The matrix contains
            # runs of spaces, so it cannot be tokenised on whitespace.
            m = re.search(r"-af (.*?) -ac ", v.get("command", ""))
            return m.group(1) if m else None
    return None


def fresh_enough(path):
    """A file written moments ago may still be being written."""
    age = time.time() - os.path.getmtime(path)
    return age > 60, age


def compare(src, dst, af, label):
    off, c = align(src, dst, af)
    if off is None:
        print(f"  {label:<28} 🔴 COULD NOT ALIGN (best normalised correlation"
              f" {c:.3f}) -- this is NOT a fidelity verdict")
        return None
    a = pcm(src, RATE, WINDOW_S, af, skip=2.0)
    b = pcm(dst, RATE, WINDOW_S, None, skip=2.0)
    # 🔴 THE SIGN MATTERS AND THE FIRST VERSION GOT IT WRONG. Indexing `b[i+off]`
    # with a negative `off` walks off the front of the array, which in Python
    # wraps to the end -- so the "difference" was the transcode subtracted from
    # an unrelated part of the source. It reported the difference 7 dB LOUDER
    # than the source, which is precisely the catastrophic-looking number
    # AUDIO-VERIFICATION §1 warns a misaligned run produces. The instrument
    # reproduced the documented failure before it produced a result.
    ia, ib = (0, off) if off >= 0 else (-off, 0)
    _ = c
    # Refine sample-exact on one second, now that both sides are roughly aligned.
    fine, _cf = best_lag(a[ia : ia + RATE], b[ib : ib + RATE], 16, 1)
    if fine >= 0:
        ib += fine
    else:
        ia += -fine
    n = min(len(a) - ia, len(b) - ib)
    if n <= 0:
        print(f"  {label:<28} 🔴 no overlap after alignment")
        return None
    diff = array.array("i", (a[ia + i] - b[ib + i] for i in range(n)))
    off = ib - ia
    s_db, d_db = rms_db(a[ia : ia + n]), rms_db(diff)
    down = s_db - d_db
    print(f"  {label:<28} source {s_db:7.2f} dB   difference {d_db:7.2f} dB"
          f"   {down:6.2f} dB down   (lag {off:+d} smp, corr {c:.3f})")
    return down


def selftest():
    """Can this tool tell a working configuration from a broken one?

    🔴 THE LAST GAP ON MY LIST. This script has three controls that run every
    time -- identity, a 4-pole top-end loss, an unrelated movie -- and none asks
    whether the MEASUREMENT ITSELF is live. With an empty band list every
    comparison returns a worst deviation of 0.0: identity passes, the real pair
    passes, and only the unrelated-movie control fails -- reporting **exit 1, a
    corpus problem**, for what is actually a broken instrument. Same shape as the
    empty register in `check-claims`, and the same fix: a distinct answer.

    Drives this script as a subprocess over a short window and reads its real
    exit code:  normal -> 0,  band list emptied -> 2.
    """
    env = dict(os.environ, FID_WINDOW="4")
    ok = True
    for label, extra, want in (("normal config", {}, 0),
                               ("band list emptied", {"FID_BANDS": "none"}, 2)):
        got = subprocess.run([sys.executable, __file__], env={**env, **extra},
                             capture_output=True).returncode
        mark = "✅" if got == want else "🔴"
        print(f"  harness: {label:<20} exit {got}, wanted {want}  {mark}")
        ok = ok and got == want
    print()
    print("the band measurement can tell a broken configuration from a clean run"
          if ok else "🔴 the harness cannot distinguish a broken configuration")
    return 0 if ok else 2


def main():
    if "--selftest" in sys.argv:
        return selftest()
    # 🔴 An empty band list makes every comparison read 0.0 dB and pass. That is
    # the harness failing, not the transcodes, and it gets its own exit code.
    if not BANDS:
        print("🔴 the band list is EMPTY -- every comparison would read 0.0 dB and")
        print("   pass. Exit 2: the harness is broken, not the transcodes.")
        return 2
    man = json.load(open("export/manifest.json"))
    names = [v["name"] for v in man.get("videos", [])]
    # 🔴 LIVENESS, the same shape as the empty band list one line up. With no
    # videos in the manifest the loop never runs, `fail` stays 0 and this reports
    # every transcode faithful -- having compared none.
    if not names:
        print("🔴 the manifest lists NO videos -- nothing was compared.")
        print("   Exit 2: the harness is broken, not the transcodes.")
        return 2
    control = "--control" in sys.argv
    fail = 0
    print(f"  window {WINDOW_S:.0f} s from t=2 s, mono {RATE} Hz, pass at "
          f"{PASS_DB:.0f} dB down\n")
    for name in names:
        src = re.search(r"-i (\S+\.wmv)", next(v["command"] for v in man["videos"]
                                               if v["name"] == name)).group(1)
        dst = os.path.join("export", next(v["file"] for v in man["videos"]
                                          if v["name"] == name))
        ok_age, age = fresh_enough(dst)
        if not ok_age:
            print(f"  {name:<28} 🔴 written {age:.0f} s ago -- may still be being"
                  " written; refusing to measure it")
            fail += 1
            continue
        af = downmix_of(man, name)
        worst = bands(src, dst, af, f"{name} -- band energies (robust to misalignment, not free of it)")
        verdict = "ok" if worst <= PASS_BAND_DB else "🔴 OUT OF TOLERANCE"
        print(f"    worst band deviation {worst:.2f} dB   {verdict}")
        if worst > PASS_BAND_DB:
            fail += 1
        # 🔴 THE KNOWN NEGATIVE RUNS EVERY TIME, not behind a flag. A band check
        # that has only ever seen a faithful pair cannot be told from one that
        # compares a file with itself by accident -- and this tool has already
        # produced four confident wrong numbers on the other quantity.
        # 🔴 THE IDENTITY CONTROL, added 2026-08-31 after the Decoder generalised
        # my own rule back at me: **a positive control that is merely "high"
        # hides the difference between an exact instrument and a lossy one.**
        # This check's positive side was 0.29 and 0.66 dB -- small, and small is
        # not zero. A systematic bias (the fold applied to one side only, a
        # different window, a resampler difference) would sit inside 0.66 dB
        # while looking like a pass. Source against itself must be EXACTLY 0.00
        # in every band, and anything else is the instrument, not the transcode.
        ident = bands(src, src, af, "  control: source vs ITSELF, must be exact", af_dst=af)
        idv = "ok" if ident == 0.0 else f"🔴 {ident:.3f} dB on an identity -- the instrument is biased"
        print(f"    worst band deviation {ident:.3f} dB   {idv}")
        if ident != 0.0:
            fail += 1
        # 🔴 A NEAR-MISS NEGATIVE, because an unrelated movie is an EASY one.
        # The Decoder measured two unrelated music BANKS separating by just
        # 5.28 dB where an unrelated movie gave me 19-20, so the margin against a
        # hard negative is 8x, not 30x. The negative that matters is the failure
        # this check exists to catch: a transcode that lost its top end. A 6 kHz
        # lowpass of the source is that failure, constructed.
        # 🔴 FOUR POLES, NOT ONE -- corrected 2026-08-31, and the correction
        # retracts a finding I published. `lowpass=f=6000` is SINGLE-POLE,
        # 6 dB/octave: a mild tilt, not a lost top end. I named it "a transcode
        # that lost its top end", measured 1.28 dB on `S00A`, and reported a
        # COVERAGE HOLE to the Decoder. **The hole was my filter.** A real brick
        # wall -- four poles -- is caught on `S00A` at 1.83 dB and on `ADV` at
        # far more.
        #
        # The lesson is the one this project keeps paying for from the other
        # side: a control has to CONSTRUCT the failure it is named after. Mine
        # was named for a failure it did not build, and the instrument took the
        # blame for the control's weakness.
        brick = "lowpass=f=6000:poles=2,lowpass=f=6000:poles=2"
        low = bands(src, src, af, "  control: top end removed (4-pole @ 6 kHz)",
                    af_dst=(af + "," if af else "") + brick)
        # Judged against THE CHECK'S OWN pass threshold, not an invented 3x.
        #
        # With the top band split this lands at 4.27 dB: it fails the 1.5 dB pass
        # test, so the check does catch it -- but by 2.8x, against the 6.4x it
        # has over the worst real transcode (0.67 dB). ⚠️ NOT COMFORTABLE, and
        # said out loud rather than smoothed: a loss milder than a 6 kHz brick
        # wall could sit between 0.67 and 1.5 and pass. The honest statement is
        # that this check catches a SEVERE top-end loss and is not characterised
        # for a mild one.
        #
        # The 3x bar it used to be judged against was mine and stricter than the
        # check itself; using the check's own threshold is the principled
        # criterion, and lowering the 3x to make a failing control pass would
        # have been tuning.
        # 🔴 REPORTED PER ASSET, NOT ASSERTED, and the reason is a measured gap
        # rather than convenience. `ADV` catches the lowpass by 2.8x. **`S00A`
        # does not catch it at all** -- 1.28 dB against a 1.5 dB threshold --
        # because its own 6-16 kHz content sits at -67 dB, so removing it changes
        # almost nothing. The check's sensitivity is MATERIAL-DEPENDENT, which is
        # the Decoder's finding about negative-separation arriving on the
        # positive side.
        #
        # Asserting it would make the suite permanently red on a gap I cannot
        # close today; hiding it would make a coverage hole into scenery. So it
        # prints COVERED / NOT COVERED per asset and the gap is tracked in
        # BLOCKED.md. The identity and unrelated-movie controls still assert.
        if low > PASS_BAND_DB:
            margin = low / PASS_BAND_DB
            note = "" if margin >= 2.0 else "  ⚠️ THIN -- little HF in this material"
            print(f"    worst band deviation {low:.2f} dB   COVERED, caught by"
                  f" {margin:.1f}x{note}")
        else:
            print(f"    worst band deviation {low:.2f} dB   🔴 NOT COVERED --"
                  f" a 6 kHz top-end loss on {name} would PASS this check")
        other = [v for v in man["videos"] if v["name"] != name]
        if other:
            osrc = os.path.join("export", other[0]["file"])
            bad = bands(src, osrc, af, f"  control: vs {other[0]['name']}, must be FAR out")
            ctl = "ok" if bad > 3 * PASS_BAND_DB else "🔴 an unrelated movie passes as faithful"
            print(f"    worst band deviation {bad:.2f} dB   {ctl}")
            if bad <= 3 * PASS_BAND_DB:
                fail += 1
        print()
        if af is None:
            print(f"  {name:<28} ⚠️  no `-af` in the recorded command: the source"
                  " is stereo, comparing without a fold")
        # Report-only: a disqualified path must not vote on the exit code. It
        # did, which is why the run went red for the wrong reason the moment the
        # return was fixed -- two defects hiding each other, and repairing one
        # exposed the other rather than the run going quietly green.
        compare(src, dst, af, name)
        if control:
            print(f"    known negatives for {name}:")
            same = compare(src, src, af, "  source vs itself")
            if same is None or same < 60:
                print("    🔴 the check cannot even match a file with itself")
                fail += 1
            other = [n for n in names if n != name]
            if other:
                osrc = os.path.join("export", next(v["file"] for v in man["videos"]
                                                   if v["name"] == other[0]))
                un = compare(src, osrc, af, f"  vs {other[0]} (unrelated)")
                if un is not None and un > 10:
                    print("    🔴 an unrelated movie scores as faithful")
                    fail += 1
    print()
    print("  ⚠️  WHAT IS ASSERTED: per-band level agreement, which needs no")
    print("     alignment. It CANNOT tell a faithful transcode from one that kept")
    print("     the spectrum and mangled the waveform. That is the honest limit of")
    print("     this quantity, and it is what the difference signal below was for.")
    print()
    print("  🔴 THE DIFFERENCE SIGNAL IS REPORT ONLY -- NO VERDICT, and the numbers")
    print("     above must not be read as one. Best alignment so far is corr")
    print("     0.763 on `S00A` and 0.075 on `ADV`, and both still report the")
    print("     difference LOUDER than the source, which is impossible for two")
    print("     aligned signals at equal level. Something remains wrong on this")
    print("     side of the measurement, not necessarily in the transcodes.")
    print()
    print("  What this run DOES establish is the trap list below, each reproduced")
    print("  here rather than reasoned about. See docs/port/DECISIONS.md.")
    print("  🔴 It measures AUDIO only; `-q:v 8` was chosen on SSIM separately.")
    # 🔴 THIS RETURN WAS UNCONDITIONAL `return 0` FOR A DAY. Making the difference
    # path report-only swallowed the band verdict with it, so `check-all`'s
    # `transcode-bands must-pass` step COULD NOT FAIL -- an asserting step that
    # asserts nothing, which is the exact shape this project keeps finding in
    # other people's work and had now shipped in mine. The band failures were
    # being printed and discarded.
    if fail:
        print(f"\n🔴 {fail} band control failure(s)")
        return 1
    return 0


sys.exit(main())