vt1200/harness/udpstream/tests/test_tile_replay_check.py

"""Tests for ``tile_check.py --replay``: the MAME replay's pixel verdict.

The replay mode in ``harness/mame/vidstream.lua`` paints a stream produced by
``make_tiles.py`` and dumps the screen after every clip frame.  This checker
decodes ``stream.bin`` itself -- full frames and deltas, with no code shared
with the encoder -- renders each frame through the dictionary, and compares it
with the dump.

The tests here build a replay directory in the *host's* format (the same files
the MAME run writes) and check that the checker passes a good replay and fails
each kind of bad one.  They are the negative controls for the MAME step: a
checker that cannot fail would make the replay evidence worthless.
"""
from pathlib import Path

import pytest

from harness.mame import tile_check
from harness.udpstream import make_tiles

#: Two frames is enough to have one keyframe and one delta.
CLIP_FRAMES = 2


@pytest.fixture(scope="module")
def tiles():
    return make_tiles.build(make_tiles.synthetic_clip(CLIP_FRAMES))


def write_replay(directory: Path, tiles, status_overrides=None):
    """Write the four things a replay run leaves behind."""
    make_tiles.write_cache(directory, tiles, clip="synthetic", every=60,
                           seed=0)
    for frame_no, indices in enumerate(tiles.indices):
        (directory / ("vram-replay-%d.bin" % frame_no)).write_bytes(
            make_tiles.render(indices, tiles.dictionary))
    wire = sum(1 for r in tiles.records if r.kind != make_tiles.KIND_SAME)
    status = {
        "FRAMES": len(tiles.indices),
        "WIRE": wire,
        "RX_OK": wire,
        "RX_BAD": 0,
        "TILES_DROP": 0,
        "DICT_READY": 1,
        "DICT_BYTES": len(tiles.dictionary),
        "DICT_MATCHED": 1,
        "SENTINEL": 0xA5C3,
        "OS_TOUCH": 0,
        "TIMEOUT": 0,
    }
    status.update(status_overrides or {})
    (directory / "replay-status.txt").write_text(
        "".join("%s=%s\n" % (k, v) for k, v in status.items()))
    return directory


def run(directory: Path, capsys):
    code = tile_check.main(["--replay", str(directory)])
    return code, capsys.readouterr().out


def test_a_correct_replay_passes(tmp_path, tiles, capsys):
    write_replay(tmp_path, tiles)
    code, out = run(tmp_path, capsys)
    assert code == 0, out
    assert "replay: %d/%d frames match" % (CLIP_FRAMES, CLIP_FRAMES) in out


def test_one_wrong_pixel_fails_and_names_the_frame(tmp_path, tiles, capsys):
    write_replay(tmp_path, tiles)
    dump = tmp_path / "vram-replay-1.bin"
    data = bytearray(dump.read_bytes())
    data[5 * 160 + 2] ^= 0x01          # one bit, inside tile row 5, tile 0
    dump.write_bytes(data)
    code, out = run(tmp_path, capsys)
    assert code == 1
    assert "frame 1" in out


def test_a_stream_that_paints_something_else_fails(tmp_path, tiles, capsys):
    """The checker must read the stream, not bless indices.bin."""
    write_replay(tmp_path, tiles)
    stream = bytearray((tmp_path / "stream.bin").read_bytes())
    # The last record's payload: flip the last value byte of the delta.
    stream[-1] ^= 0x7F
    (tmp_path / "stream.bin").write_bytes(stream)
    code, out = run(tmp_path, capsys)
    assert code == 1
    assert "frame 1" in out


def test_a_missing_dump_fails(tmp_path, tiles, capsys):
    write_replay(tmp_path, tiles)
    (tmp_path / "vram-replay-1.bin").unlink()
    code, out = run(tmp_path, capsys)
    assert code == 1
    assert "frame 1" in out


@pytest.mark.parametrize("key,value", [("TILES_DROP", 1), ("RX_BAD", 1),
                                       ("TIMEOUT", 1), ("DICT_READY", 0),
                                       ("OS_TOUCH", 1)])
def test_a_bad_status_fails(tmp_path, tiles, capsys, key, value):
    write_replay(tmp_path, tiles, {key: value})
    code, out = run(tmp_path, capsys)
    assert code == 1
    assert key in out


def test_the_clip_comparison_measures_the_replayed_error(tmp_path, tiles,
                                                          capsys):
    """``--clip`` reports BER/PSNR of the *dumped* screens, not the render.

    The screens equal the render (the check above), and the render differs
    from the source by exactly the dictionary's error, so the measurement and
    the encoder's own count have to agree bit for bit.
    """
    frames = make_tiles.synthetic_clip(CLIP_FRAMES)
    clip = tmp_path / "clip"
    clip.mkdir()
    for n, frame in enumerate(frames):
        (clip / ("frame-%04d.bin" % n)).write_bytes(frame)
    write_replay(tmp_path, tiles)

    code = tile_check.main(["--replay", str(tmp_path), "--clip", str(clip)])
    out = capsys.readouterr().out
    assert code == 0, out
    expected = sum(
        sum(make_tiles.tile_errors(frame, indices, tiles.dictionary))
        for frame, indices in zip(frames, tiles.indices))
    assert "reconstruction: %d bit(s)" % expected in out
    assert "PSNR" in out and "BER" in out


def test_a_clip_that_is_not_the_source_shows_up_only_in_the_measurement(
        tmp_path, tiles, capsys):
    """A wrong clip changes the reported error, not the pixel verdict."""
    frames = [bytes(1024 * 160) for _ in range(CLIP_FRAMES)]
    clip = tmp_path / "clip"
    clip.mkdir()
    for n, frame in enumerate(frames):
        (clip / ("frame-%04d.bin" % n)).write_bytes(frame)
    write_replay(tmp_path, tiles)

    code = tile_check.main(["--replay", str(tmp_path), "--clip", str(clip)])
    out = capsys.readouterr().out
    assert code == 0, out          # the screen still equals the render
    assert "reconstruction: 0 bit(s)" not in out