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 at 8 kHz, then refined sample-exact at
                 the full rate, BEFORE subtracting.
  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.

🔴 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


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:
                return None, c          # pinned: refuse rather than report it
            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


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)
        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("  🔴 REPORT ONLY -- NO FIDELITY VERDICT IS CLAIMED, 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())