Dragon-s-Lair-X68k/tools/analysis/16_span_roundtrip.py

#!/usr/bin/env python3
"""GATE for the DLX3 span container: does the reference decoder reproduce the
encoder's own reconstruction, from the emitted bytes?

    python3 tools/analysis/16_span_roundtrip.py [frames_dir] --kbps <KB/s>

Exits non-zero if any frame differs by a single pixel.

WHY THIS EXISTS SEPARATELY FROM 09.  `09_ratectl_drift.py` replays SKIP
semantics in Python against the mode maps the encoder returned; it never reads
a container.  A span breaks exactly that shortcut: a spanned block reads SKIP
in the mode header and is painted by the span section instead, so a replay that
knows only about mode maps reports drift where there is none, and -- far worse
-- a container whose span section is malformed would still pass, because 09
never parses one.  This gate closes that: encode, WRITE THE CONTAINER, read it
back with tools/encoder/dlx.py (the byte-for-byte reference decoder the 68000
is checked against), and compare to what ratectl recorded.

It also has to prove it tested something.  A round-trip over a container with
no spans in it is green by vacuity, which is the failure mode FINDINGS 40.6
named for the snapshot count: a gate must take its expected work from the
generated artefact, not from an assumption.  So the thresholds below are
asserted, not printed.

The `--kbps` default is the BUS rate, not the `scsi` profile's 280: spans are
bought with bytes, and 14_dmac_chain.py scores them against the delivery pipe.
At the profile rate the lam search has already spent the allowance and there is
nothing left to buy a span with -- which is a real finding about the encoder
(FINDINGS 41.2), not a reason for the gate to test nothing.
"""
import argparse, os, pickle, sys, time
sys.path.insert(0, "tools/encoder")
import numpy as np
import vq_hybrid as H, ratectl as RC, encode as E
from dlx import DLX

ap = argparse.ArgumentParser()
ap.add_argument("frames_dir", nargs="?", default="tmp/fr_singe")
ap.add_argument("--kbps", type=float, required=True,
                    help="REQUIRED. There is no default: the delivery rate is a property of the medium and this project has never measured it. FINDINGS 42.1 -- the figure this tool used to default to was a user-supplied '4 Mbps' with no provenance, was a tenth of SCSI-1's asynchronous rating, and was never a bus measurement at all. A default let every table in FINDINGS 30-49 be scored against it without anyone restating it. Pass one explicitly.")
ap.add_argument("--out", default="tmp/s12_roundtrip")
ap.add_argument("--cache", default=None)
a = ap.parse_args()

cache = a.cache or f"tmp/model_{os.path.basename(a.frames_dir.rstrip('/'))}.pkl"
if os.path.exists(cache):
    m = pickle.load(open(cache, "rb"))
    print(f"model from {cache}")
else:
    t = time.time()
    m = H.build(a.frames_dir, k1=256, k4=256, iters=16)
    pickle.dump(m, open(cache, "wb"))
    print(f"built model in {time.time()-t:.0f} s -> {cache}")

bad = 0
for span_mode in ("need", "all"):
    print(f"\n=== spans={span_mode}, {a.kbps:g} KB/s ===")
    m.pop("_sym", None)
    enc = RC.encode_rate_controlled(m, target_kbps=a.kbps, lam_lo=1.0,
                                    cycle_budget=RC.FRAME_CYCLES,
                                    span_mode=span_mode)
    recs = E.build_records(m, enc, span_mode)
    path = f"{a.out}_{span_mode}.dlx"
    total, vid, _ = E.write_container(path, m, recs, 12, m["k1"], m["k4"],
                                      span_mode)

    nsp = sum(len(x) for x in enc["spans"])
    nfr = sum(1 for x in enc["spans"] if x)
    px = sum(len(p) for x in enc["spans"] for _, _, p in x)
    print(f"{path}: {total:,} B, {len(recs)} frames, "
          f"{nsp:,} spans on {nfr} frames, {px:,} pixels painted by one "
          f"({100*px/(len(recs)*m['H']*m['W']):.1f}% of all pixels)")

    d = DLX(path)
    if d.version != 3:
        print(f"FAIL: container is DLX{d.version}, not DLX3"); bad += 1; continue

    # The decoder's own walk of the span section must land exactly where the
    # block payload starts, and blocks() already raises if the payload does not
    # consume the record -- so this reads the spans back through the same code
    # path the 68000 is modelled on rather than trusting the writer.
    got = d.decode_all()
    diff = np.array([(g != r).sum() for g, r in zip(got, enc["recon"])])
    print(f"pixels differing from the encoder's reconstruction: "
          f"{diff.sum()} total, worst frame {diff.max()}, "
          f"frames with any: {int((diff>0).sum())}/{len(diff)}")
    if diff.sum():
        f = int(np.argmax(diff))
        ys, xs = np.where(got[f] != enc["recon"][f])
        print(f"FAIL: frame {f} differs at {diff[f]} px, first (x={xs[0]}, "
              f"y={ys[0]}), block (bx={xs[0]//4}, by={ys[0]//4}), "
              f"mode there = {d.modes(f)[(ys[0]//4)*d.nbx + xs[0]//4]}")
        bad += 1

    # A green round-trip over a container with no spans in it proves nothing.
    if span_mode == "all":
        if nsp < 1000:
            print(f"FAIL: only {nsp} spans emitted -- this gate did not "
                  f"exercise the span path"); bad += 1
        if not (px and max(len(x) for x in enc["spans"]) > 50):
            print(f"FAIL: no frame carries a substantial span table"); bad += 1

print()
if bad:
    print(f"FAILED: {bad} check(s)")
    sys.exit(1)
print("OK  the DLX3 span container round-trips: the reference decoder rebuilds "
      "the\n    encoder's reconstruction exactly, from the emitted bytes.")