forked from arti/vt1200
294 lines
12 KiB
Python
294 lines
12 KiB
Python
"""Tests for the clip -> tile dictionary -> cached stream pipeline.
|
|
|
|
Task 6's acceptance is "a 30-frame synthetic clip quantised and replayed in
|
|
MAME must reproduce each frame's rendered bitmap within the dictionary's error
|
|
bound". The MAME half is ``harness/mame/tile_check.py --replay``; this file
|
|
pins the host half: the clip generator is deterministic, the plan is Task 5's
|
|
cadence rule, the cache round-trips byte for byte, and the render the terminal
|
|
is asked to paint differs from the source only by what the dictionary cannot
|
|
express.
|
|
|
|
The error is a *bound*, not a threshold dressed up as one. For every tile the
|
|
plan stores the nearest dictionary entry under Hamming distance; that distance
|
|
is exactly the number of pixels the screen will differ by, and it is the
|
|
minimum any encoder using this dictionary could achieve. One test checks the
|
|
encoder against a brute-force search, so the reported error is a property of
|
|
the dictionary and the clip rather than of a lucky sample; another bounds the
|
|
whole clip's bit-error rate, which is what a regression would move.
|
|
"""
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from harness.udpstream import make_tiles
|
|
from harness.udpstream.tilecost import (MAX_TILES, TILE_BYTES, TILE_COLS,
|
|
TILES_PER_SCREEN, _frame_tiles)
|
|
from harness.udpstream.tiledict import KEYFRAME_EVERY, unpack_delta
|
|
|
|
#: The repo root, so the committed cache can be found from anywhere.
|
|
REPO = Path(__file__).resolve().parents[3]
|
|
|
|
#: Where the demo's cache lives (``make_tiles.py``'s default output).
|
|
CACHE = REPO / "harness" / "udpstream" / "tiles"
|
|
|
|
#: The bit-error-rate the synthetic clip must stay under. It is an acceptance
|
|
#: bound, not a measurement: the measured value is ~0.1% and is reported by
|
|
#: the CLI and NETWORK_REPORT.md, so this only has to catch a regression that
|
|
#: makes the dictionary meaningfully worse.
|
|
MAX_BER = 0.02
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def clip30():
|
|
"""The 30-frame synthetic clip, built once for the whole module."""
|
|
return make_tiles.synthetic_clip(30)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def tiles30(clip30):
|
|
"""The dictionary, quantised indices and plan for that clip."""
|
|
return make_tiles.build(clip30)
|
|
|
|
|
|
# --- the clip -------------------------------------------------------------
|
|
|
|
def test_synthetic_clip_is_deterministic_and_full_screen(clip30):
|
|
assert clip30 == make_tiles.synthetic_clip(30)
|
|
assert len(clip30) == 30
|
|
assert all(len(f) == 1024 * 160 for f in clip30)
|
|
|
|
|
|
def test_synthetic_clip_moves(clip30):
|
|
assert len(set(clip30)) == len(clip30), "every frame must differ"
|
|
# and the difference is real motion, not a one-pixel flicker
|
|
first, second = clip30[0], clip30[1]
|
|
differing = sum((a ^ b).bit_count() for a, b in zip(first, second))
|
|
assert differing > 400
|
|
changed = sum(1 for a, b in zip(_frame_tiles(first), _frame_tiles(second))
|
|
if a != b)
|
|
assert changed >= 5, "a delta needs changed tiles to carry"
|
|
|
|
|
|
def test_synthetic_clip_is_richer_than_the_dictionary(clip30):
|
|
"""k = 256 must actually be a constraint, or the PSNR says nothing."""
|
|
tiles = set()
|
|
for frame in clip30:
|
|
tiles |= set(_frame_tiles(frame))
|
|
assert len(tiles) > MAX_TILES
|
|
|
|
|
|
# --- the plan -------------------------------------------------------------
|
|
|
|
def test_build_shapes(clip30, tiles30):
|
|
assert len(tiles30.dictionary) == MAX_TILES * TILE_BYTES
|
|
assert len(tiles30.indices) == len(clip30)
|
|
assert all(len(i) == TILES_PER_SCREEN for i in tiles30.indices)
|
|
assert len(tiles30.records) == len(clip30)
|
|
|
|
|
|
def test_plan_starts_with_a_full_frame_and_keeps_the_cadence(tiles30):
|
|
records = tiles30.records
|
|
assert records[0].kind == make_tiles.KIND_FULL
|
|
assert records[0].payload == tiles30.indices[0]
|
|
assert len(records[0].payload) == TILES_PER_SCREEN
|
|
|
|
# A 30-frame clip is shorter than the 60-frame cadence, so frame 0 is the
|
|
# only full frame; the rule itself is exercised below with a shorter
|
|
# interval and again in the drop test.
|
|
assert [r.frame_no for r in records if r.kind == make_tiles.KIND_FULL] == [0]
|
|
|
|
fast = make_tiles.plan(tiles30.indices, every=10)
|
|
assert [r.frame_no for r in fast if r.kind == make_tiles.KIND_FULL] == \
|
|
[0, 10, 20]
|
|
|
|
|
|
def test_delta_payloads_carry_exactly_the_changed_tiles(tiles30):
|
|
previous = None
|
|
for record, frame in zip(tiles30.records, tiles30.indices):
|
|
if previous is not None:
|
|
changed = sum(a != b for a, b in zip(previous, frame))
|
|
if record.kind == make_tiles.KIND_SAME:
|
|
assert changed == 0
|
|
assert record.payload == b""
|
|
else:
|
|
assert record.kind == make_tiles.KIND_DELTA
|
|
assert record.changed == changed
|
|
assert len(record.payload) == 2 + 3 * changed
|
|
assert changed > 0
|
|
# positions are strictly ascending and name the changed tiles
|
|
pairs = unpack_delta(record.payload)
|
|
assert [p for p, _ in pairs] == \
|
|
[i for i in range(TILES_PER_SCREEN) if previous[i] != frame[i]]
|
|
assert bytes(v for _, v in pairs) == bytes(
|
|
frame[i] for i, _ in pairs)
|
|
previous = frame
|
|
|
|
|
|
def test_the_stream_decodes_back_to_the_quantiser_output(tiles30):
|
|
"""The cache's plan, read through the receiver's own decoder, is lossless.
|
|
|
|
This is the property the MAME replay checks on the terminal: the stream
|
|
alone (not the indices file) determines the picture.
|
|
"""
|
|
state = None
|
|
for record, expected in zip(tiles30.records, tiles30.indices):
|
|
if record.kind == make_tiles.KIND_FULL:
|
|
state = bytearray(record.payload)
|
|
elif record.kind == make_tiles.KIND_DELTA:
|
|
assert state is not None
|
|
for position, value in unpack_delta(record.payload):
|
|
state[position] = value
|
|
elif record.kind != make_tiles.KIND_SAME:
|
|
raise AssertionError("unexpected kind %r" % (record.kind,))
|
|
assert bytes(state) == expected
|
|
|
|
|
|
def test_a_drop_forces_a_keyframe_at_the_next_frame(tiles30):
|
|
records = make_tiles.plan(tiles30.indices, drops={7: 3})
|
|
kinds = {r.frame_no: r.kind for r in records}
|
|
assert kinds[7] == make_tiles.KIND_FULL
|
|
assert kinds[1] == make_tiles.KIND_DELTA
|
|
assert kinds[8] == make_tiles.KIND_DELTA
|
|
# and the cumulative counter is consumed once, not on every later frame
|
|
assert [n for n, k in kinds.items() if k == make_tiles.KIND_FULL] == [0, 7]
|
|
|
|
|
|
def test_an_unchanged_frame_is_not_sent():
|
|
black = bytes(TILES_PER_SCREEN)
|
|
records = make_tiles.plan([black, black, black], every=60)
|
|
assert [r.kind for r in records] == \
|
|
[make_tiles.KIND_FULL, make_tiles.KIND_SAME, make_tiles.KIND_SAME]
|
|
assert [len(r.payload) for r in records] == [TILES_PER_SCREEN, 0, 0]
|
|
|
|
|
|
# --- the error bound ------------------------------------------------------
|
|
|
|
def test_encoder_attains_the_dictionary_minimum(tiles30, clip30):
|
|
"""Every stored index is the nearest entry, checked by brute force.
|
|
|
|
``quantise`` caches the answer per distinct tile, so a wrong cache would
|
|
pass a self-consistency check; this compares a sample of tiles against an
|
|
independent search over all 256 entries.
|
|
"""
|
|
dictionary = tiles30.dictionary
|
|
entries = [dictionary[i * TILE_BYTES:(i + 1) * TILE_BYTES]
|
|
for i in range(MAX_TILES)]
|
|
for frame_no in (0, 7, 29):
|
|
tiles = _frame_tiles(clip30[frame_no])
|
|
indices = tiles30.indices[frame_no]
|
|
for position in range(0, len(tiles), 97):
|
|
tile = tiles[position].to_bytes(TILE_BYTES, "big")
|
|
got = indices[position]
|
|
want_distance = min(sum((a ^ b).bit_count()
|
|
for a, b in zip(tile, entry))
|
|
for entry in entries)
|
|
got_distance = sum((a ^ b).bit_count()
|
|
for a, b in zip(tile, entries[got]))
|
|
assert got_distance == want_distance, (
|
|
"tile %d of frame %d is %d bits from its entry, but entry %d "
|
|
"is %d bits away" % (position, frame_no, got_distance,
|
|
got, want_distance))
|
|
|
|
|
|
def test_reconstruction_is_within_the_dictionary_bound(tiles30, clip30):
|
|
assert len(tiles30.bit_errors) == len(clip30)
|
|
pixels = 1024 * 1280
|
|
for frame_no, (bits, frame) in enumerate(zip(tiles30.bit_errors, clip30)):
|
|
# bit_errors is the sum of the per-tile distances, so it is exactly
|
|
# the number of pixels that will differ.
|
|
assert bits == sum(make_tiles.tile_errors(
|
|
frame, tiles30.indices[frame_no], tiles30.dictionary))
|
|
ber = bits / pixels
|
|
assert ber <= MAX_BER, (frame_no, ber)
|
|
assert make_tiles.psnr_db(bits, pixels) == pytest.approx(
|
|
-10.0 * __import__("math").log10(ber))
|
|
|
|
|
|
def test_a_perfect_dictionary_reports_no_error():
|
|
assert make_tiles.psnr_db(0, 1024 * 1280) is None
|
|
assert make_tiles.ber(0, 1024 * 1280) == 0.0
|
|
|
|
|
|
def test_render_follows_the_tile_geometry(tiles30):
|
|
screen = make_tiles.render(tiles30.indices[0], tiles30.dictionary)
|
|
assert len(screen) == 1024 * 160
|
|
for tx, ty in ((0, 0), (TILE_COLS - 1, 31), (17, 9)):
|
|
entry = tiles30.indices[0][ty * TILE_COLS + tx]
|
|
tile = tiles30.dictionary[entry * TILE_BYTES:(entry + 1) * TILE_BYTES]
|
|
for row in (0, 13, 31):
|
|
offset = (ty * 32 + row) * 160 + tx * 4
|
|
assert screen[offset:offset + 4] == \
|
|
tile[row * 4:(row + 1) * 4]
|
|
|
|
|
|
def test_rendered_frames_differ_from_the_source_only_by_the_error(tiles30,
|
|
clip30):
|
|
"""The full-screen render is the source with exactly the error bits flipped."""
|
|
for frame_no in (0, 15, 29):
|
|
screen = make_tiles.render(tiles30.indices[frame_no],
|
|
tiles30.dictionary)
|
|
source = clip30[frame_no]
|
|
differing = 0
|
|
for row in range(1024):
|
|
left = row * 160
|
|
right = left + 160
|
|
differing += sum((a ^ b).bit_count()
|
|
for a, b in zip(source[left:right],
|
|
screen[left:right]))
|
|
assert differing == tiles30.bit_errors[frame_no]
|
|
|
|
|
|
# --- the cache ------------------------------------------------------------
|
|
|
|
def test_cache_round_trips(tiles30, tmp_path):
|
|
make_tiles.write_cache(tmp_path, tiles30, clip="synthetic",
|
|
every=KEYFRAME_EVERY, seed=0)
|
|
cache = make_tiles.read_cache(tmp_path)
|
|
assert cache.dictionary == tiles30.dictionary
|
|
assert cache.indices == tiles30.indices
|
|
assert cache.records == tiles30.records
|
|
assert cache.manifest["frames"] == len(tiles30.indices)
|
|
assert cache.manifest["dictionary_bytes"] == len(tiles30.dictionary)
|
|
assert cache.manifest["keyframes"] == [0]
|
|
assert (tmp_path / "stream.bin").read_bytes().startswith(
|
|
make_tiles.STREAM_MAGIC)
|
|
|
|
|
|
def test_read_cache_rejects_a_stream_that_does_not_match_the_indices(tiles30,
|
|
tmp_path):
|
|
make_tiles.write_cache(tmp_path, tiles30, clip="synthetic",
|
|
every=KEYFRAME_EVERY, seed=0)
|
|
stream = bytearray((tmp_path / "stream.bin").read_bytes())
|
|
# Flip one index byte of the first full frame: the stream now paints a
|
|
# different picture than indices.bin says.
|
|
stream[10 + 4 + 5] ^= 0xFF
|
|
(tmp_path / "stream.bin").write_bytes(stream)
|
|
with pytest.raises(ValueError):
|
|
make_tiles.read_cache(tmp_path)
|
|
|
|
|
|
def test_committed_cache_matches_the_generator():
|
|
"""The demo's cache is the generator's output, not a stale leftover."""
|
|
cache = make_tiles.read_cache(CACHE)
|
|
tiles = make_tiles.build(make_tiles.synthetic_clip(30))
|
|
assert cache.dictionary == tiles.dictionary
|
|
assert cache.indices == tiles.indices
|
|
assert cache.records == tiles.records
|
|
assert json.loads((CACHE / "manifest.json").read_text()) == cache.manifest
|
|
|
|
|
|
def test_load_clip_reads_sorted_raw_frames(tmp_path):
|
|
frames = [bytes(32 * 160) for _ in range(3)]
|
|
for name, frame in zip(("b.bin", "a.bin", "c.bin"),
|
|
(frames[1], frames[0], frames[2])):
|
|
(tmp_path / name).write_bytes(frame)
|
|
(tmp_path / "notes.txt").write_text("ignored")
|
|
assert make_tiles.load_clip(tmp_path) == frames
|
|
|
|
|
|
def test_load_clip_rejects_a_partial_tile_band(tmp_path):
|
|
(tmp_path / "a.bin").write_bytes(bytes(16 * 160)) # 16 rows, not 32
|
|
with pytest.raises(ValueError):
|
|
make_tiles.load_clip(tmp_path)
|