Dragon-s-Lair-X68k/tools/bench/verify_packed_audio.py

#!/usr/bin/env python3
"""The container's own audio, READ BACK OFF THE SPEAKER.  ROADMAP P6c.

    python3 tools/bench/verify_packed_audio.py <in.dlxp> <capture.wav> [counters.json]

WHY THIS READS THE CAPTURE AND NOT THE PLAYER'S COUNTERS.  src/player/packed.s
reports how many lumps it armed and how many payload bytes it handed the chip,
and every one of those numbers can be right while the sound is wrong.  Nothing
in this format parses anything (FINDINGS 67.4): a lump fetched one sector out is
not an error, it is 7,168 B of noise played at the right length; a payload one
byte long is not an error either, it is a rate.  The only instrument that can
tell those apart from a correct run is the stream the chip actually produced.

WHAT IS CHECKED, and it is the whole scene rather than a sample of it:

  1. every lump's payload, decoded with the FOUR AXES OUT OF THE CONTAINER'S OWN
     HEADER (FINDINGS 66/67.3), appears in the capture SAMPLE-EXACT and in
     order.  Not "close": the recursion is exact arithmetic and MAME's okim6258
     puts `signal << 4` into a stream the machine routes to the speaker at gain
     0.50, so a chip sample is `signal * 8` and recovering it is a division and
     not a rounding.  The residual is asserted.

  2. the PAYLOAD lengths are the accumulator's and not the lump's.  A player
     that fed the chip the whole A*512 B lump runs 0.09% fast -- 1.25 s of
     lip-sync over the game (67.2) -- and the difference between the two is 6.54
     B a group, which is 13 samples.  So this is checked by LENGTH: lump k's run
     of matched samples must be exactly 2*lump_bytes(k), and that alternates
     14,322 / 14,324 rather than being 14,336 every time.

  3. THE SEAMS, measured rather than assumed.  Channel 3 counts out at the end
     of a lump and the chip has no FIFO and no starvation state -- it goes on
     decoding whatever byte its data register still holds, alternating that
     byte's low and high nibbles, until the CPU arms the next lump.  Those
     samples are NOT silence, they are the recursion running on a repeated byte,
     and the state they leave behind is what lump k+1 decodes from.  So the
     search below carries the state across the seam and reports its LENGTH,
     which is the audible cost of every design decision on the video path.

THE NEGATIVE CONTROL IS BUILT IN.  A seam is found by searching for the repeat
count that makes the next lump match; if the player had fed the wrong bytes, no
repeat count would make it match and the run fails rather than sliding.
"""
import json, os, struct, sys, wave

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)),
                                "..", "encoder"))
import adpcm
from dlxp import DLXP, lump_bytes as _lump_bytes


def dlxp_lump_bytes(d, k):
    """What the CADENCE gives lump k, before the stream's end
    truncates it -- so a lump that is short because the scene ran
    out can be told from one that is short because the remainder
    arithmetic said so."""
    return _lump_bytes(k, d.cad_f, d.fps, d.aud_hz)

SCALE = 8            # okim6258's `signal << 4` into a 32768 stream, times the
                     # machine's 0.50 speaker route.  verify_adpcm_chip.py's.
RESID = 2            # counts of slack on that recovery, as 66 measured it
LOOK = 32            # samples of continuation a candidate run has to survive
MAXRUN = 60000       # nibbles one delivered byte may be stretched over: 3.8 s
                     # at 15,625 Hz, far past any seam a working player makes.
                     # A bound is what makes a failure say "not found" rather
                     # than run until the host is bored.
LEAD = 400000        # samples of silence before PLAY


def stepper(dec):
    """One sample of the recursion, as a closure over the container's own four
    axes.  adpcm.decode_state is the same arithmetic and is what the whole-lump
    paths use; this exists because the walk below needs it ONE NIBBLE AT A TIME
    and a function call per sample over 78,125 bytes is the difference between
    a gate that runs in seconds and one that does not."""
    lo, hi = adpcm.clamp_bounds(dec["bits"])
    variant, step, adj = dec["variant"], adpcm.STEP, adpcm.INDEX_ADJUST

    def one(st, n):
        sig, idx = st
        sig += adpcm.delta(n, step[idx], variant)
        sig = lo if sig < lo else (hi if sig > hi else sig)
        idx += adj[n & 7]
        idx = 0 if idx < 0 else (48 if idx > 48 else idx)
        return sig, (sig, idx)
    return one


def walk(rec, pos, st, data, dec):
    """Account for EVERY byte the player handed the chip, in order.

    WHY A WALK AND NOT A COMPARISON.  A whole-stream `decode(data) == capture`
    is the check this wanted to be and it does not survive contact with the
    machine.  The MSM6258 has no FIFO and no handshake at all: the DMA channel
    writes a byte into the data register whenever #DRQ3 asks, and the chip
    decodes a nibble out of whatever is in there on every sample tick.  Those
    are two clocks -- 7,812.5 B/s and 15,625 Hz -- and MAME's okim6258 data_w
    RESETS the nibble select on every write.  So a byte is normally played as
    two nibbles, and near a boundary it can be played as one (the high nibble
    dropped) or as three or more (the low/high pair repeated) purely from where
    the write lands inside a sound-stream slice.

    THE MODEL IS THEREFORE ONE LINE: byte b was played as `c` nibbles taken from
    the cycle (b&15, b>>4), c >= 1.  This walk finds `c` for every byte, and the
    HISTOGRAM of c is the result -- c=2 everywhere is a chip being fed exactly
    at its own rate, and every c>2 is the chip replaying a byte while the 68000
    was somewhere else, which is what a SEAM is.

    It is not a loose check.  Every one of the `c` samples has to be exactly
    right, the run has to be followed by LOOK samples that are exactly right,
    and a byte the player never sent leaves no c at all.  Returns
    (fail_index, pos, runs); fail_index is None on success.
    """
    one = stepper(dec)
    runs = []
    n = len(data)
    pos_ = pos
    for i in range(n):
        b = data[i]
        pair = (b & 15, b >> 4)
        nxt = data[i + 1] & 15 if i + 1 < n else None
        # DEEP FIRST, THEN SHALLOW.  LOOK samples of continuation is what tells
        # a real seam from a coincidence -- a repeated pair can agree with the
        # next lump's first nibble for one sample and does not for thirty-two.
        # But the window is 16 bytes wide and a SECOND stretched byte inside it
        # rejects the true answer as well as the false ones, so a byte that no
        # candidate survives is retried with a shallow window rather than
        # failing -- and the retry window is TWO samples, the next byte's own
        # pair, because a four-sample one reaches into the stretched byte
        # itself.  The cost of resolving it wrong is a failed walk, not a pass.
        got = _pick(rec, pos_, st, pair, nxt, data, i, one, LOOK)
        if got is None:
            got = _pick(rec, pos_, st, pair, nxt, data, i, one, 2)
        if got is None:
            return i, pos_, runs
        c, st = got
        pos_ += c
        runs.append(c)
    return None, pos_, runs


def _pick(rec, pos, st, pair, nxt, data, i, one, look):
    """The run length for one byte: every candidate `c` whose samples are exact
    and whose continuation survives `look`, resolved to c=2 where c=2 is one of
    them.  Two nibbles a byte is what the two clocks agree on; anything else is
    an event and an event needs the evidence, which is what `look` is."""
    cands = []
    s2, c = st, 0
    while c < MAXRUN and pos + c < len(rec):
        smp, s3 = one(s2, pair[c & 1])
        if rec[pos + c] != smp:
            break
        c += 1
        s2 = s3
        if nxt is None:
            cands.append((c, s2))
            break
        smp2, _ = one(s2, nxt)
        if pos + c < len(rec) and rec[pos + c] == smp2 \
           and _look(rec, pos + c, s2, data, i + 1, one, look):
            cands.append((c, s2))
            if c > 2 and len(cands) >= 2:
                break
    if not cands:
        return None
    for cc in cands:
        if cc[0] == 2:
            return cc
    return cands[0]


def _look(rec, pos, st, data, i, one, look=LOOK):
    """`look` samples of continuation, assuming two nibbles a byte from here.

    This is what tells a real seam from a coincidence.  At a lump boundary the
    repeated pair can happen to agree with the next lump's first nibble for one
    sample; it does not go on agreeing for thirty-two.
    """
    n, k = len(data), 0
    while k < look and i < n:
        b = data[i]
        for nib in (b & 15, b >> 4):
            if pos >= len(rec):
                return True
            smp, st = one(st, nib)
            if rec[pos] != smp:
                return False
            pos += 1
            k += 1
        i += 1
    return True


def main():
    if len(sys.argv) < 3:
        sys.exit(__doc__)
    d = DLXP(sys.argv[1])
    counters = json.load(open(sys.argv[3])) if len(sys.argv) > 3 else None
    if not d.has_audio:
        sys.exit("this container is silent -- there is nothing to have heard")

    dec = d.decoder()
    fails = []
    def ck(ok, msg):
        print(("OK   " if ok else "FAIL ") + msg)
        if not ok:
            fails.append(msg)

    w = wave.open(sys.argv[2])
    n, ch, rate = w.getnframes(), w.getnchannels(), w.getframerate()
    s = struct.unpack("<%dh" % (n * ch), w.readframes(n))
    left, right = s[0::ch], s[1::ch]
    ck(rate == d.aud_hz,
       f"the capture is at {rate:,} Hz and the chip's stream is {d.aud_hz:,} -- "
       f"equal rates are what keep MAME's resampler out of the measurement")
    ck(list(left) == list(right), "both speakers carry the same samples (pan 00)")
    ck(any(left), "the capture contains a signal at all")
    if not any(left) or rate != d.aud_hz:
        return 1
    rec = [round(v / SCALE) for v in left]
    start = next(i for i, v in enumerate(rec) if v)
    ck(start < LEAD, f"the chip starts playing {start/rate:.2f} s in")

    print(f"--- {sys.argv[1]}: {d.n_lumps} lumps, decoder {dec['variant']}/"
          f"{dec['order']}, {dec['bits']}-bit clamp, accumulator {dec['init']} "
          f"at PLAY -- ALL FOUR out of the header (67.3)")

    # ---- THE STREAM, ACCOUNTED FOR BYTE BY BYTE.  d.audio() is the container's
    # own lumps reassembled BY PAYLOAD -- 67.2's accumulator, not the padded
    # sector runs -- so a player that fed the chip whole lumps does not merely
    # score worse here, it fails to walk: the 6.54 B of zero at the end of a
    # lump are nibbles that are not in this stream.
    data = d.audio()
    bad, pos, runs = walk(rec, start, (dec["init"], 0), data, dec)
    ck(bad is None,
       f"all {len(data):,} bytes of the container's audio reached the chip, in "
       f"order, and every sample the chip produced from them is exact"
       + ("" if bad is None else f" -- byte {bad:,} of {len(data):,} does not "
          f"account for capture sample {pos:,}"))
    if bad is not None:
        k = bad * 2 * d.fps // (d.cad_f * d.aud_hz)
        print(f"     that is inside lump {k}, {bad - sum(len(d.lump(j)) for j in range(k)):,} B in")
        return 1

    matched = pos - start
    worst = max(abs(v - SCALE * round(v / SCALE))
                for v in left[start:start + matched])
    ck(worst <= RESID,
       f"every one of {matched:,} matched samples is within {worst} of a "
       f"multiple of {SCALE} -- so `signal = sample/{SCALE}` recovers the chip's "
       f"own stream rather than rounding to the nearest story")

    # ---- 1. THE FEED.  c=2 is a chip being fed at exactly its own rate.
    hist = {}
    for c in runs:
        hist[c] = hist.get(c, 0) + 1
    two = hist.get(2, 0)
    ck(two * 1000 >= len(runs) * 999,
       f"{two:,} of {len(runs):,} bytes ({two*100/len(runs):.3f}%) were played "
       f"as exactly two nibbles -- the chip was paced by its own #DRQ3 and not "
       f"by the CPU")
    print(f"     nibbles per delivered byte: "
          + ", ".join(f"{c}x{v:,}" for c, v in sorted(hist.items())))

    # ---- 2. THE SEAMS, which is what every c > 2 is.  A lump's channel counts
    # out and the chip goes on replaying the last byte until the CPU arms the
    # next one; the excess nibbles ARE that interval, measured in the only place
    # it exists, which is the sound.
    seams = [(i, c - 2) for i, c in enumerate(runs) if c > 2]
    print(f"--- THE SEAMS: {len(seams)} byte(s) were stretched, out of "
          f"{d.n_lumps - 1} lump boundaries")
    if seams:
        ex = sum(c for _, c in seams)
        print(f"     worst {max(c for _, c in seams)} samples = "
              f"{max(c for _, c in seams)/rate*1000:.2f} ms; total {ex} samples "
              f"= {ex/rate*1000:.2f} ms of replayed byte over "
              f"{matched/rate:.2f} s of audio ({ex*100/matched:.4f}%)")
    # and every stretched byte must BE a lump boundary -- a stretch anywhere
    # else is the CPU losing the chip in the middle of a buffer.
    ends = set()
    off = 0
    for k in range(d.n_lumps):
        off += len(d.lump(k))
        ends.add(off - 1)
    stray = [i for i, _ in seams if i not in ends]
    ck(not stray,
       f"every stretched byte is the LAST byte of a lump ({len(stray)} were not)"
       + ("" if not stray else f" -- first at byte {stray[0]:,}, which is the "
          f"chip running dry in the middle of a buffer"))

    # ---- 3. THE PAYLOAD IS THE ACCUMULATOR'S (FINDINGS 67.2).  The walk
    # already proves it -- a whole-lump player's stream contains the padding and
    # would not walk -- so what is left is to price what was avoided.  The LAST
    # lump is left out: 120 frames is not a multiple of F=11, so it carries ten
    # frames of audio and is short for an arithmetic reason and not a rate one.
    full = [k for k in range(d.n_lumps)
            if len(d.lump(k)) == dlxp_lump_bytes(d, k)]
    ck(len(full) >= d.n_lumps - 1,
       f"{len(full)} of {d.n_lumps} lumps carry a whole group of audio")
    if len(full) > 1:
        pad = d.cad_a * 512
        got = sum(len(d.lump(k)) for k in full)
        over = pad * len(full) - got
        secs = got * 2 / rate
        print(f"--- THE PADDING IS DRIFT (FINDINGS 67.2), over the {len(full)} "
              f"lumps that carry a whole group")
        print(f"     payload {got:,} B against {pad*len(full):,} B of lump "
              f"space: {over:,} B more, {over*100/got:.3f}%, "
              f"{over*2/rate*1000:.2f} ms over {secs:.2f} s of audio")
        print(f"     -> {over*2/rate/secs*22.8*60:.2f} s of lip-sync over the "
              f"game's 22.8 min, and the accumulator in pg_apay is the three "
              f"lines that do not spend it")

    if counters:
        ck(counters["late"] == 0,
           f"the player never re-armed a channel that still had bytes to send "
           f"({counters['late']} did)")
        ck(counters["starve"] == 0,
           f"the player never found the channel counted out with no lump ready "
           f"({counters['starve']} times it did)")
        ck(counters["bytes"] == len(data),
           f"the player's own byte count ({counters['bytes']:,}) is the "
           f"container's whole payload ({len(data):,})")
        print(f"     the player: {counters['armed']} lumps armed, "
              f"{counters['fetched']} fetched, {counters['serv']:,} service "
              f"calls over {counters['shown']} frames "
              f"({'BUS HELD' if counters['held'] else 'CYCLE STEALING'})")

    print("PACKED AUDIO GATE " + ("GREEN" if not fails
                                  else f"RED: {len(fails)} failed"))
    return 1 if fails else 0


if __name__ == "__main__":
    sys.exit(main())