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


BANDS = [(0, 500), (500, 2000), (2000, 6000), (6000, 16000)]


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):
    """Per-band level, source against transcode. No alignment involved."""
    print(f"  {label}")
    worst = 0.0
    for lo, hi in BANDS:
        a = band_db(src, af, lo, hi)
        b = band_db(dst, None, 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 main():
    man = json.load(open("export/manifest.json"))
    names = [v["name"] for v in man.get("videos", [])]
    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 (needs no alignment)")
        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.
        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")
        down = compare(src, dst, af, name)
        if down is None or down < PASS_DB:
            fail += 1
        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.")
    return 0


sys.exit(main())