Files

575 lines
25 KiB
Python
Executable File

#!/usr/bin/env python3
"""tile_check.py -- did the 32x32 tile blit put every dictionary entry on screen?
Task 4's pixel check. The Lua side (harness/mame/vidstream.lua,
VT1200_MODE=tiles) uploads a 32 KB dictionary through the payload's own RX
ring, sends one TILES_FULL frame (1,280 index bytes) and then dumps four files:
vram-tiles.bin the screen as memory: 1024 rows of 256 bytes from bit
0x04000000
dict-tiles.bin the 32,768 dictionary bytes that were uploaded (the Lua
read them back from POOL_PTR + 2 * SLOT_BYTES before sending the frame, so
an upload mismatch fails the mode instead of being blamed on the blit)
indices-tiles.bin the 1,280 index bytes of the TILES_FULL frame
tiles-status.txt key=value lines (RX_OK, TILES_DROP, DICT_READY, SENTINEL,
OS_TOUCH, TIMEOUT, ...)
This script renders the screen the indices and the dictionary say must be
there and compares it with the dump, tile by tile:
for each tile (tx, ty): screen rows ty*32 .. ty*32+31,
bytes tx*4 .. tx*4+3 of each 256-byte row
<- dictionary[index(ty, tx)] row-major, 32 rows of four bytes.
That is an independent reading of the geometry (tilecost.py's TILE_* constants
and the design's section 8.3), not a copy of the payload's loop. A source tile
is 32 px = two 16-bit words = four bytes a row, 128 bytes in all; the screen
row pitch is 2048 bits = 256 bytes, the real unit's (harness/c-hello/README.md).
Only the first 160 bytes of a screen row are pixels the blitter writes (40
tiles x 32 px = 1280 px); the remaining 96 bytes are outside every tile and are
not compared -- they belong to the resident OS and no tile blit touches them.
A timed-out run writes TIMEOUT=1 and is reported as a timeout, not as a screen
full of wrong pixels: blaming the blit for a frame the payload never accepted
would be the wrong verdict (ruling R15). The VRAM canary is asserted too, so a
payload that stopped arming it cannot pass as a correct screen.
The check is deliberately sensitive to a wrong SPTCH (source pitch), which is
the negative control Task 4 asks for: every dictionary entry has 32 distinct
rows, and this script refuses to call a dictionary usable unless that holds, so
a blit that re-reads row 0 for every row cannot match.
``--replay DIR`` is Task 6's check: ``make_tiles.py`` writes a cache
(``dictionary.bin``, ``stream.bin``), the MAME replay mode paints its frames in
order and dumps the 160 active bytes of every screen row after each one
(``vram-replay-<frame>.bin``), and this script decodes ``stream.bin`` itself --
a second reading of the full-frame and delta layouts, sharing no code with the
encoder -- renders every frame through the dictionary and compares. The
stream, not ``indices.bin``, decides the picture: a cache whose stream paints
something else fails here, which is the point of re-deriving it. With
``--clip DIR`` the dumps are also compared with the source frames and the
reconstruction error (BER/PSNR) is printed -- a measurement of what the
terminal painted, which does not change the verdict.
usage:
tile_check.py [--vram PATH] [--dict PATH] [--indices PATH] [--status PATH]
tile_check.py --replay DIR
Exit status 0 only if every check passes.
"""
import argparse
import math
import os
import struct
import sys
HEADER_SIZE = 16
TILE_PIXELS = 32 # px per tile side, and rows per tile
TILE_ROW_BYTES = 4 # 32 px, 1 bpp
TILE_BYTES = TILE_PIXELS * TILE_ROW_BYTES # 128
TILE_COLS = 40 # 1280 px / 32 px
TILE_ROWS = 32 # 1024 px / 32 px
TILES_PER_SCREEN = TILE_COLS * TILE_ROWS # 1,280
DICT_ENTRIES = 256
DICT_BYTES = DICT_ENTRIES * TILE_BYTES # 32,768
ACTIVE_ROW_BYTES = TILE_COLS * TILE_ROW_BYTES # 160 bytes = 1280 pixels
VRAM_PITCH = 256 # bytes per screen row (2048 bits)
SCREEN_ROWS = 1024
CANARY = 0xA5C3 # SENT_VALUE in vidstream.c, at row 1023 / column 1280
# The replay stream (make_tiles.STREAM_MAGIC and record layout), decoded here
# independently of the encoder. A SAME record (kind 0) carries no payload and
# is not sent: the screen must not change for it.
STREAM_MAGIC = b"TS01"
STREAM_HEADER = struct.Struct("<4sH")
RECORD_HEADER = struct.Struct("<BHH")
KIND_SAME, KIND_FULL, KIND_DELTA = 0, 2, 3
ACTIVE_ROW_BYTES = TILE_COLS * 4 # 160
# The four tiles the brief calls out, plus the order this script reports them.
CORNERS = ((0, 0, "top-left"), (TILE_COLS - 1, 0, "top-right"),
(0, TILE_ROWS - 1, "bottom-left"),
(TILE_COLS - 1, TILE_ROWS - 1, "bottom-right"))
failures = 0
def check(cond, what):
global failures
if not cond:
failures += 1
print("FAIL: " + what)
return cond
def read_file(path):
"""The whole file, or None with a diagnosable line instead of a traceback."""
try:
with open(path, "rb") as f:
return f.read()
except OSError as e:
print("tile_check: cannot read %s: %s" % (path, e))
return None
def read_status(path):
st = {}
try:
with open(path, "r", encoding="ascii", errors="replace") as f:
for line in f:
if "=" in line:
k, v = line.strip().split("=", 1)
try:
st[k] = int(v, 0)
except ValueError:
st[k] = v
except OSError as e:
print("tile_check: cannot read %s: %s" % (path, e))
return None
return st
def tile_dict(dictionary, entry):
return dictionary[entry * TILE_BYTES:(entry + 1) * TILE_BYTES]
# --- Task 6: the replay of a make_tiles.py stream -------------------------
def read_stream(path):
"""Parse ``stream.bin``: ``(frame count, [(kind, frame_no, payload), ...])``.
Raises ``ValueError`` on anything the writer could not have produced, so a
truncated cache is reported as such instead of being decoded into a
different picture.
"""
with open(path, "rb") as f:
data = f.read()
if len(data) < STREAM_HEADER.size:
raise ValueError("shorter than the %d-byte header"
% STREAM_HEADER.size)
magic, frames = STREAM_HEADER.unpack_from(data, 0)
if magic != STREAM_MAGIC:
raise ValueError("magic %r, expected %r" % (magic, STREAM_MAGIC))
records = []
offset = STREAM_HEADER.size
while offset < len(data):
if offset + RECORD_HEADER.size > len(data):
raise ValueError("truncated in a record header")
kind, frame_no, length = RECORD_HEADER.unpack_from(data, offset)
offset += RECORD_HEADER.size
if offset + length > len(data):
raise ValueError("truncated in frame %d's payload" % frame_no)
records.append((kind, frame_no, data[offset:offset + length]))
offset += length
if len(records) != frames:
raise ValueError("%d records for %d frames" % (len(records), frames))
for n, (kind, frame_no, payload) in enumerate(records):
if frame_no != n:
raise ValueError("record %d names frame %d" % (n, frame_no))
if kind not in (KIND_SAME, KIND_FULL, KIND_DELTA):
raise ValueError("frame %d has unknown kind %d" % (n, kind))
return frames, records
def apply_delta(state, payload):
"""Apply one delta payload to ``state``; raise ``ValueError`` on bad input.
The same rules the payload's C decoder enforces, written out here from the
layout rather than imported: exact length, a positive count, on-grid
positions and strictly ascending order.
"""
if len(payload) < 2:
raise ValueError("a delta opens with a 2-byte count, got %d bytes"
% len(payload))
count = struct.unpack_from("<H", payload, 0)[0]
if not 1 <= count <= TILES_PER_SCREEN:
raise ValueError("count %d is not 1..%d" % (count, TILES_PER_SCREEN))
if len(payload) != 2 + 3 * count:
raise ValueError("count %d needs %d bytes, got %d"
% (count, 2 + 3 * count, len(payload)))
previous = -1
for k in range(count):
position = struct.unpack_from("<H", payload, 2 + 2 * k)[0]
if position >= TILES_PER_SCREEN:
raise ValueError("position %d is off the grid" % position)
if position <= previous:
raise ValueError("positions are not ascending at %d" % position)
previous = position
state[position] = payload[2 + 2 * count + k]
def load_clip(directory):
"""Read the source clip's raw frames, sorted by name, or raise ValueError."""
paths = sorted(p for p in os.listdir(directory) if p.endswith(".bin"))
if not paths:
raise ValueError("no *.bin frames in %s" % (directory,))
frames = []
for name in paths:
with open(os.path.join(directory, name), "rb") as f:
data = f.read()
if not data or len(data) % ACTIVE_ROW_BYTES:
raise ValueError("%s is not whole %d-byte rows"
% (name, ACTIVE_ROW_BYTES))
if len(data) > SCREEN_ROWS * ACTIVE_ROW_BYTES:
raise ValueError("%s is taller than the screen" % name)
frames.append(data)
return frames
def check_replay(directory, clip_dir=None):
"""Compare every replayed frame with the stream's own render.
With ``clip_dir`` the *dumped* screens are also compared with the source
frames and the reconstruction error is reported (BER and PSNR) -- a
measurement of what the terminal actually painted, not of what the host
thought it would paint. It does not change the verdict: the verdict is
"the screen equals the render", which is checked with or without a clip.
"""
dictionary = read_file(os.path.join(directory, "dictionary.bin"))
stream_path = os.path.join(directory, "stream.bin")
status = read_status(os.path.join(directory, "replay-status.txt"))
if dictionary is None or status is None:
return 2
try:
frames, records = read_stream(stream_path)
except (OSError, ValueError) as exc:
print("tile_check: cannot read %s: %s" % (stream_path, exc))
return 2
print("tile_check: replay %s: %d frames, dictionary %d B"
% (directory, frames, len(dictionary)))
clip = None
if clip_dir:
try:
clip = load_clip(clip_dir)
except (OSError, ValueError) as exc:
print("tile_check: cannot read the clip in %s: %s"
% (clip_dir, exc))
return 2
if check(len(clip) >= frames,
"the clip has %d frames, the stream %d"
% (len(clip), frames)):
clip = clip[:frames]
else:
clip = None
check(len(dictionary) == DICT_BYTES,
"dictionary is %d B, expected %d" % (len(dictionary), DICT_BYTES))
# --- the payload's own record ------------------------------------------
wire = sum(1 for kind, _frame_no, _payload in records
if kind != KIND_SAME)
if check(status.get("TIMEOUT", 0) == 0,
"TIMEOUT=%r: the replay did not finish" % (status.get("TIMEOUT"),)):
check(status.get("FRAMES") == frames,
"FRAMES=%r, the stream has %d" % (status.get("FRAMES"), frames))
check(status.get("WIRE") == wire,
"WIRE=%r, the stream sends %d frames" % (status.get("WIRE"), wire))
check(status.get("RX_OK") == wire,
"RX_OK=%r, expected %d accepted frames"
% (status.get("RX_OK"), wire))
check(status.get("RX_BAD") == 0,
"RX_BAD=%r, expected 0 rejected frames" % (status.get("RX_BAD"),))
check(status.get("TILES_DROP") == 0,
"TILES_DROP=%r: every replayed frame describes the same "
"dictionary the terminal holds" % (status.get("TILES_DROP"),))
check(status.get("DICT_MATCHED") == 1,
"DICT_MATCHED=%r: the payload's dictionary is not the file "
"uploaded" % (status.get("DICT_MATCHED"),))
check(status.get("DICT_READY") == 1 and
status.get("DICT_BYTES") == DICT_BYTES,
"DICT_READY=%r DICT_BYTES=%r, expected 1 and %d"
% (status.get("DICT_READY"), status.get("DICT_BYTES"), DICT_BYTES))
check(status.get("OS_TOUCH") == 0,
"OS_TOUCH=%r: something else wrote the sentinel corner of VRAM"
% (status.get("OS_TOUCH"),))
check(status.get("SENTINEL") == CANARY,
"canary word is %r, expected 0x%04X"
% (status.get("SENTINEL"), CANARY))
# --- every frame, rendered from the stream alone -----------------------
state = None
matched = 0
clip_bits = 0
for frame_no, (kind, _named, payload) in enumerate(records):
try:
if kind == KIND_FULL:
if len(payload) != TILES_PER_SCREEN:
raise ValueError("full frame is %d bytes, not %d"
% (len(payload), TILES_PER_SCREEN))
state = bytearray(payload)
elif kind == KIND_DELTA:
if state is None:
raise ValueError("a delta before any full frame")
apply_delta(state, payload)
elif kind != KIND_SAME:
raise ValueError("unknown kind %d" % kind)
except ValueError as exc:
check(False, "frame %d: stream does not decode: %s"
% (frame_no, exc))
break
if state is None:
check(False, "frame %d: the stream does not begin with a full "
"frame" % frame_no)
break
if kind == KIND_SAME and payload:
check(False, "frame %d: a SAME record carries %d payload bytes"
% (frame_no, len(payload)))
path = os.path.join(directory, "vram-replay-%d.bin" % frame_no)
dump = read_file(path)
if dump is None:
check(False, "frame %d: no dump at %s" % (frame_no, path))
continue
if len(dump) != SCREEN_ROWS * ACTIVE_ROW_BYTES:
check(False, "frame %d: dump is %d B, expected %d"
% (frame_no, len(dump), SCREEN_ROWS * ACTIVE_ROW_BYTES))
continue
bad_rows = 0
for y in range(SCREEN_ROWS):
want = rendered_row(dictionary, state, y // TILE_PIXELS,
y % TILE_PIXELS)
got = dump[y * ACTIVE_ROW_BYTES:(y + 1) * ACTIVE_ROW_BYTES]
if got != want:
bad_rows += 1
if bad_rows == 1:
at = next(i for i in range(ACTIVE_ROW_BYTES)
if got[i] != want[i])
print("tile_check: FAIL: frame %d: screen row %d byte %d "
"(%d differing of %d): expected %02X, got %02X"
% (frame_no, y, at, sum(a != b for a, b in
zip(got, want)),
ACTIVE_ROW_BYTES, want[at], got[at]))
if check(bad_rows == 0,
"frame %d: %d of %d active screen rows do not match the "
"render" % (frame_no, bad_rows, SCREEN_ROWS)):
matched += 1
if clip is not None and len(clip[frame_no]) == len(dump):
bits = sum((a ^ b).bit_count()
for a, b in zip(dump, clip[frame_no]))
clip_bits += bits
print("tile_check: frame %d reconstruction: %d bit(s) (BER %.6f)"
% (frame_no, bits, bits / (SCREEN_ROWS * 1280)))
if failures > 8:
print("tile_check: stopping after the first failures")
break
check(matched == frames,
"%d of %d frames match their render" % (matched, frames))
print("tile_check: replay: %d/%d frames match" % (matched, frames))
if clip is not None:
pixels = frames * SCREEN_ROWS * 1280
psnr = ("inf" if clip_bits == 0 else
"%.2f dB" % (-10.0 * math.log10(clip_bits / pixels)))
print("tile_check: reconstruction: %d bit(s) over %d frame(s), "
"BER %.6f, PSNR %s"
% (clip_bits, frames, clip_bits / pixels, psnr))
if failures:
print("tile_check: FAIL (%d checks)" % failures)
return 1
print("tile_check: PASS")
return 0
def tile_index(indices, tx, ty):
return indices[ty * TILE_COLS + tx]
def rendered_row(dictionary, indices, ty, row):
"""The 160 active bytes of screen row ty*32+row, from the indices."""
out = bytearray()
for tx in range(TILE_COLS):
entry = tile_index(indices, tx, ty)
t = tile_dict(dictionary, entry)
out += t[row * TILE_ROW_BYTES:(row + 1) * TILE_ROW_BYTES]
return bytes(out)
def main(argv=None):
# The failure counter is module state, and the tests call main() more than
# once in one process; a run's verdict must not inherit the last one's.
global failures
failures = 0
ap = argparse.ArgumentParser()
d = os.environ.get("VID_DUMP_DIR", "/tmp/opencode")
ap.add_argument("--replay", metavar="DIR",
help="check a make_tiles.py replay directory instead of "
"one tile frame")
ap.add_argument("--clip", metavar="DIR",
help="with --replay: the source frames, for the "
"reconstruction error (BER/PSNR)")
ap.add_argument("--vram", default=os.path.join(d, "vram-tiles.bin"))
ap.add_argument("--dict", default=os.path.join(d, "dict-tiles.bin"))
ap.add_argument("--indices", default=os.path.join(d, "indices-tiles.bin"))
ap.add_argument("--status", default=os.path.join(d, "tiles-status.txt"))
args = ap.parse_args(argv)
if args.replay:
return check_replay(args.replay, args.clip)
vram = read_file(args.vram)
dictionary = read_file(args.dict)
indices = read_file(args.indices)
status = read_status(args.status)
if vram is None or dictionary is None or indices is None or status is None:
return 2
print("tile_check: vram %s (%d B) dict %s (%d B) indices %s (%d B)"
% (args.vram, len(vram), args.dict, len(dictionary),
args.indices, len(indices)))
# --- a timed-out wait is not a pixel verdict (ruling R15) ---------------
if status.get("TIMEOUT", 0) != 0:
print("FAIL: timed out waiting for the frame (RX_OK=%r, "
"TILES_DROP=%r: the payload did not finish with the tile frame)"
% (status.get("RX_OK"), status.get("TILES_DROP")))
print("tile_check: FAIL (timeout)")
return 1
# --- the payload's own record -------------------------------------------
check(status.get("RX_OK") == 1,
"RX_OK=%r, expected exactly 1 accepted TILES_FULL frame"
% (status.get("RX_OK"),))
check(status.get("RX_BAD") == 0,
"RX_BAD=%r, expected 0 rejected frames" % (status.get("RX_BAD"),))
check(status.get("TILES_DROP") == 0,
"TILES_DROP=%r: a tile frame with a complete dictionary must paint"
% (status.get("TILES_DROP"),))
check(status.get("DICT_READY") == 1,
"DICT_READY=%r, expected 1" % (status.get("DICT_READY"),))
check(status.get("DICT_BYTES") == DICT_BYTES,
"DICT_BYTES=%r, expected %d" % (status.get("DICT_BYTES"), DICT_BYTES))
check(status.get("DICT_MATCHED") == 1,
"DICT_MATCHED=%r: the dictionary the payload holds is not the one "
"that was uploaded" % (status.get("DICT_MATCHED"),))
check(status.get("OS_TOUCH") == 0,
"OS_TOUCH=%r: something else wrote the sentinel corner of VRAM"
% (status.get("OS_TOUCH"),))
check(status.get("SENTINEL") == CANARY,
"canary word is %r, expected 0x%04X (SENT_VALUE in vidstream.c)"
% (status.get("SENTINEL"), CANARY))
print("tile_check: RX_OK=%r RX_BAD=%r TILES_DROP=%r DICT_READY=%r "
"SENTINEL=%r"
% (status.get("RX_OK"), status.get("RX_BAD"),
status.get("TILES_DROP"), status.get("DICT_READY"),
status.get("SENTINEL")))
# --- the inputs have the size the geometry needs -----------------------
if not check(len(dictionary) == DICT_BYTES,
"dictionary is %d B, expected %d"
% (len(dictionary), DICT_BYTES)):
return 1
if not check(len(indices) == TILES_PER_SCREEN,
"index frame is %d B, expected %d"
% (len(indices), TILES_PER_SCREEN)):
return 1
if not check(len(vram) == SCREEN_ROWS * VRAM_PITCH,
"vram dump is %d B, expected %d"
% (len(vram), SCREEN_ROWS * VRAM_PITCH)):
return 1
# --- the test data can tell a right blit from a broken one -------------
#
# A dictionary whose entries all look alike, or whose rows are all the
# same, would let a wrong tile index or a wrong source pitch pass. The
# checks below are about the test DATA, not the payload: if they fail the
# run has nothing to say about the blitter either way.
used = sorted(set(indices))
print("tile_check: %d distinct dictionary entries used by the frame"
% len(used))
data_failures = failures
check(len(used) >= 100,
"the index frame uses %d distinct entries; a frame with few distinct "
"tiles cannot detect a wrong tile mapping" % len(used))
corner_idx = []
for tx, ty, name in CORNERS:
corner_idx.append(tile_index(indices, tx, ty))
check(len(set(corner_idx)) == len(corner_idx),
"the four corner indices are not distinct: %r" % (corner_idx,))
for entry in used:
t = tile_dict(dictionary, entry)
rows = [t[r * TILE_ROW_BYTES:(r + 1) * TILE_ROW_BYTES]
for r in range(TILE_PIXELS)]
if not check(len(set(rows)) > 1,
"dictionary entry %d has 32 identical rows: a wrong "
"SPTCH could not be detected" % entry):
break
if not check(any(b != 0 for b in t) and any(b != 0xFF for b in t),
"dictionary entry %d is a solid field" % entry):
break
if failures > data_failures:
# The test data cannot tell a right blit from a broken one; comparing
# pixels would report a mismatch that says nothing about the payload.
return 1
# --- the screen, tile by tile ------------------------------------------
matched = 0
bad = []
for ty in range(TILE_ROWS):
for tx in range(TILE_COLS):
entry = tile_index(indices, tx, ty)
t = tile_dict(dictionary, entry)
ok = True
for r in range(TILE_PIXELS):
y = ty * TILE_PIXELS + r
want = t[r * TILE_ROW_BYTES:(r + 1) * TILE_ROW_BYTES]
got = vram[y * VRAM_PITCH + tx * TILE_ROW_BYTES:
y * VRAM_PITCH + (tx + 1) * TILE_ROW_BYTES]
if got != want:
ok = False
break
if ok:
matched += 1
elif len(bad) < 8:
bad.append((tx, ty, entry, r, want, got))
for tx, ty, entry, r, want, got in bad:
print("tile_check: tile (%d,%d) entry %d, row %d: expected %s, got %s"
% (tx, ty, entry, r, want.hex(), got.hex()))
print("tile_check: %d/%d tiles match their dictionary entry"
% (matched, TILES_PER_SCREEN))
check(matched == TILES_PER_SCREEN,
"%d tiles do not carry their dictionary entry"
% (TILES_PER_SCREEN - matched))
# --- the four corners, named ------------------------------------------
for tx, ty, name in CORNERS:
entry = tile_index(indices, tx, ty)
t = tile_dict(dictionary, entry)
ok = True
for r in range(TILE_PIXELS):
y = ty * TILE_PIXELS + r
got = vram[y * VRAM_PITCH + tx * TILE_ROW_BYTES:
y * VRAM_PITCH + (tx + 1) * TILE_ROW_BYTES]
if got != t[r * TILE_ROW_BYTES:(r + 1) * TILE_ROW_BYTES]:
ok = False
break
print("tile_check: %s corner (%d,%d) entry %d: %s"
% (name, tx, ty, entry, "PASS" if ok else "FAIL"))
check(ok, "%s corner tile (%d,%d) is not dictionary entry %d"
% (name, tx, ty, entry))
# --- the full-screen row view (one comparison per row) -----------------
row_bad = 0
for ty in range(TILE_ROWS):
for row in range(TILE_PIXELS):
y = ty * TILE_PIXELS + row
want = rendered_row(dictionary, indices, ty, row)
got = vram[y * VRAM_PITCH:y * VRAM_PITCH + ACTIVE_ROW_BYTES]
if got != want:
row_bad += 1
if row_bad <= 8:
print("tile_check: screen row %d: expected %s..., got %s..."
% (y, want[:16].hex(), got[:16].hex()))
check(row_bad == 0,
"%d of %d active screen rows do not match the rendered frame"
% (row_bad, SCREEN_ROWS))
if failures:
print("tile_check: FAIL (%d checks)" % failures)
return 1
print("tile_check: PASS")
return 0
if __name__ == "__main__":
sys.exit(main())