#!/usr/bin/env python3
"""vidsend.py -- the host sender: paced strip frames and the status back-channel.

One frame *n* leaves at ``t0 + n/fps`` (an absolute schedule, so a slow frame is
caught up on rather than shifting every later frame), the frame is chunked at
1450 bytes and carried in Ethernet frames of the stream's own type
(:data:`harness.udpstream.frame.ETHERTYPE`), and every 30 frames one CSV line is
printed.  In the other direction the script listens for the terminal's
``KIND_STATUS`` frames and decodes them.

    sudo python3 harness/udpstream/vidsend.py --iface vt1200 --frames 200
    sudo python3 harness/udpstream/vidsend.py --iface vt1200 --tiles --frames 300
    python3 harness/udpstream/vidsend.py --dry-run --frames 200      # no NIC

``--tiles`` sends the cached tile stream instead of strip frames: the 32 KB
dictionary in chunks (``tiledict.dict_chunks``), then one ``TILES_FULL`` or
``TILES_DELTA`` per clip frame from the cache ``make_tiles.py`` wrote.  The
keyframe decision is live, not a replay of the cache: ``keyframe_due(frame_no,
drops, seen_drops)`` sees the terminal's *cumulative* ``drops`` from each status
frame, so a reported loss forces a full frame at the next opportunity rather
than waiting for the 60-frame cadence.  A clip shorter than ``--frames`` loops,
and the cadence keeps counting across the loop, which is what makes a 6,300-
frame soak from a 30-frame cache a real run and not 210 restarts.

Exit status is 0 only when a status frame was received and the last one reports
``drops == 0`` and ``crc_err == 0``.  "No status" is a failure: the two
conditions were never observed, and a sender that cannot see the terminal must
not claim the terminal is happy.

WHAT THIS CANNOT PROVE HERE.  The MAME driver creates no
``device_network_interface`` (ruling R11), so the emulated LANCE is not attached
to a host NIC: this script cannot be pointed at the emulator.  It is exercised
three ways instead, and none of them is a network run:

  * the pacing, chunking, padding and status decode are unit-tested through an
    injected ``send`` callable and an injected clock
    (harness/udpstream/tests/test_vidsend.py);
  * ``--dry-run`` runs the real 200-frame, 30 fps schedule against a counting
    transport, which is what makes a wall-clock pacing check possible without a
    socket -- it prints ``RESULT DRY-RUN`` and never claims a delivery;
  * the terminal's own status emission is checked in MAME
    (``VT1200_MODE=stream harness/mame/run-vidstream.sh``), where the payload's
    LANCE transmit path is exercised and the bytes are decoded by
    ``harness/mame/vidstream_status_check.py`` with the same wire format this
    script parses.
"""
from __future__ import annotations

import argparse
import os
import socket
import struct
import sys
import time

if __package__ in (None, ""):                   # run as a script, not a module
    sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(
        os.path.abspath(__file__)))))

from harness.udpstream import make_tiles                               # noqa: E402
from harness.udpstream.frame import (                               # noqa: E402
    BROADCAST, ETHERTYPE, HEADER_SIZE, KIND_STATUS, MAGIC, STATUS_SIZE,
    ethernet_frame, pack_header, unpack_header, unpack_status)
from harness.udpstream.tiledict import (                            # noqa: E402
    DELTA_HDR, KEYFRAME_EVERY, TILES_PER_SCREEN, delta_frame,
    dict_chunks, keyframe_due)
from harness.udpstream.vidsrc import (                               # noqa: E402
    CHUNK, MAX_ROWS, RING_DATAGRAMS, ROW_BYTES, StripSource)

__all__ = [
    "CSV_HEADER", "CSV_HEADER_TILES", "Sender", "TileSender", "bars", "split",
    "PATTERNS", "decode_status_frame", "RawEthernet", "parse_mac",
]

#: One CSV line is printed every this many frames.
REPORT_EVERY = 30

#: The strip CSV columns, in this order: frame number, elapsed seconds, achieved
#: fps, bytes handed to the transport so far.  The CSV goes to stdout and the
#: prose (status lines, the verdict) to stderr, so a run's stdout is a clean
#: table.
CSV_HEADER = "frame,elapsed_s,fps,bytes"

#: The tile CSV adds what plan step 3 asks for: which kind went out, how many
#: payload bytes it carried, how many tiles changed, and whether the frame was
#: a keyframe.  The first four columns are the strip table's, so a run can be
#: processed by the same tools.
CSV_HEADER_TILES = ("frame,elapsed_s,fps,bytes,kind,payload_bytes,"
                    "changed_tiles,keyframe")

#: Bars per screen: a source row is ink while ``(r + n) % 16 < 8``.
BAR_PERIOD = 16


# --- patterns -------------------------------------------------------------

def bars(rows: int, frame_no: int) -> bytes:
    """Packed rows of horizontal bars, drifting up one row per frame.

    Two row classes per frame (ink and blank), so a decoder that reused one
    source row cannot produce the picture, and the pattern moves between
    frames, so a frozen stream is distinguishable from a live one.
    """
    out = bytearray(rows * ROW_BYTES)
    for r in range(rows):
        if (r + frame_no) % BAR_PERIOD < BAR_PERIOD // 2:
            start = r * ROW_BYTES
            out[start:start + ROW_BYTES] = b"\xff" * ROW_BYTES
    return bytes(out)


def split(rows: int, frame_no: int) -> bytes:
    """Packed rows of a split field with a vertical edge that moves per frame.

    A second pattern whose answer is known analytically: the left 640 pixels of
    every row are ink, and the right side is ink only above a moving horizon.
    """
    out = bytearray(rows * ROW_BYTES)
    horizon = (frame_no * 3) % max(rows, 1)
    for r in range(rows):
        start = r * ROW_BYTES
        out[start:start + ROW_BYTES // 2] = b"\xff" * (ROW_BYTES // 2)
        if r < horizon:
            out[start + ROW_BYTES // 2:start + ROW_BYTES] = \
                b"\xff" * (ROW_BYTES // 2)
    return bytes(out)


PATTERNS = {"bars": bars, "split": split}

#: The largest VS frame the tile path can put on the wire: a delta that names
#: every tile is ``count u16`` plus 3 bytes per entry, the most the format can
#: carry.  The dictionary goes out as one-datagram chunks and never uses
#: ``--chunk``.
MAX_TILE_FRAME = HEADER_SIZE + DELTA_HDR + 3 * TILES_PER_SCREEN


def min_chunk(frame_bytes: int) -> int:
    """Smallest ``--chunk`` whose frame fits the terminal's RX ring.

    The ring holds :data:`RING_DATAGRAMS` descriptors and cannot receive a frame
    that needs more; the payload drops it and counts ``RX_RING_DROPS``.  One
    frame of ``frame_bytes`` split at ``chunk`` needs ``ceil(frame_bytes /
    chunk)`` datagrams, so the smallest legal chunk is the ceiling of the frame
    size over the ring size.
    """
    return (frame_bytes + RING_DATAGRAMS - 1) // RING_DATAGRAMS


# --- the status back-channel ---------------------------------------------

def decode_status_frame(data: bytes):
    """Decode one received Ethernet frame, or return None.

    None means "not a status frame of ours": a different ethertype, too short,
    not a ``VS`` frame, or a different kind.  A status frame whose header CRC is
    wrong is *returned* with ``crc_ok`` False, so the caller can count it
    instead of silently treating it as noise.
    """
    if len(data) < 14 + HEADER_SIZE + STATUS_SIZE:
        return None
    if data[12:14] != struct.pack(">H", ETHERTYPE):
        return None
    body = data[14:]
    try:
        h = unpack_header(body)
    except ValueError:
        return None
    if h["magic"] != MAGIC or h["kind"] != KIND_STATUS:
        return None
    if h["payload_len"] != STATUS_SIZE:
        return None
    out = unpack_status(body[HEADER_SIZE:HEADER_SIZE + STATUS_SIZE])
    out["kind"] = h["kind"]
    out["frame_no"] = h["frame_no"]
    out["crc_ok"] = h["crc_ok"]
    return out


# --- pacing ---------------------------------------------------------------

class Sender:
    """Send ``rows``-row strip frames at a fixed rate and collect status frames.

    Everything that would touch the world is injected, so the schedule and the
    verdict can be tested without a socket:

    ``transport(data: bytes)``  one Ethernet frame, header included
    ``receive()``               zero or more received Ethernet frames
    ``clock()`` / ``sleep(s)``  the time base the schedule is measured on
    """

    def __init__(self, transport, src, receive=None, rows=128, stretch=8,
                 fps=30.0, frames=None, pattern="bars", chunk=CHUNK,
                 dst=BROADCAST, clock=time.monotonic, sleep=time.sleep,
                 report_every=REPORT_EVERY, status_wait=1.0, out=None,
                 err=None):
        if pattern not in PATTERNS:
            raise ValueError("unknown pattern %r; choose from %s"
                             % (pattern, ", ".join(sorted(PATTERNS))))
        if chunk < HEADER_SIZE + 1:
            raise ValueError("chunk %d is too small for a frame" % chunk)
        if fps < 0:
            raise ValueError("fps must be >= 0 (0 = unpaced): %r" % (fps,))
        self.source = StripSource(rows, stretch)
        need = min_chunk(HEADER_SIZE + rows * ROW_BYTES)
        if chunk < need:
            raise ValueError(
                "chunk %d would split a %d-row frame (%d bytes) across more "
                "than the terminal's %d-descriptor RX ring; chunk must be at "
                "least %d"
                % (chunk, rows, HEADER_SIZE + rows * ROW_BYTES,
                   RING_DATAGRAMS, need))
        self.pattern = pattern
        self.rows = rows
        self.stretch = stretch
        self.fps = float(fps)
        self.frames = frames
        self.chunk = chunk
        self.dst = dst
        self.src = src
        self.transport = transport
        self.receive = receive
        self.clock = clock
        self.sleep = sleep
        self.report_every = report_every
        self.status_wait = status_wait
        self.out = out if out is not None else sys.stdout
        self.err = err if err is not None else sys.stderr
        self.frame_offset = 0
        self.sent_frames = 0
        self.sent_datagrams = 0
        self.sent_bytes = 0
        self.statuses = []
        self.bad_status = 0
        self.t0 = 0.0

    # -- output --------------------------------------------------------
    def say(self, text):
        """One CSV line: stdout, because it is a table."""
        print(text, file=self.out, flush=True)

    def log(self, text):
        """Prose: stderr, so it cannot be mistaken for a CSV row."""
        print(text, file=self.err, flush=True)

    # -- one frame -----------------------------------------------------
    def frame_bytes(self, n: int) -> bytes:
        """The complete wire frame (header + body) for frame *n*."""
        body = PATTERNS[self.pattern](self.rows, n)
        no = (self.frame_offset + n) & 0xFFFF
        return b"".join(self.source.encode(no, body))

    def datagrams(self, n: int) -> list[bytes]:
        """The frame split into datagrams, only the last padded to the minimum.

        Padding matters: the 60-byte Ethernet minimum must be reached at the
        *end* of a frame.  Padding a middle chunk would insert bytes the sender
        never meant between two chunks of one frame, and the receiver would see
        a body shifted by however many pads were inserted.
        """
        frame = self.frame_bytes(n)
        chunks = [frame[i:i + self.chunk]
                  for i in range(0, len(frame), self.chunk)]
        out = []
        for i, chunk in enumerate(chunks):
            out.append(ethernet_frame(chunk, src=self.src, dst=self.dst,
                                      pad=(i == len(chunks) - 1)))
        return out

    def send_frame(self, n: int):
        for data in self.datagrams(n):
            self.transport(data)
            self.sent_datagrams += 1
            self.sent_bytes += len(data)
        self.sent_frames += 1

    # -- status --------------------------------------------------------
    def poll(self):
        if self.receive is None:
            return
        for data in (self.receive() or [])[:64]:
            st = decode_status_frame(data)
            if st is None:
                continue
            if not st["crc_ok"]:
                self.bad_status += 1
                self.log("status: header CRC bad, ignored (frames_rx=%d)"
                         % st["frames_rx"])
                continue
            self.statuses.append(st)
            self.log("status %d: frame_no=%d frames_rx=%d crc_err=%d drops=%d "
                     "csr0=0x%04X heap_free_kb=%d"
                     % (len(self.statuses), st["frame_no"], st["frames_rx"],
                        st["crc_err"], st["drops"], st["csr0"],
                        st["heap_free_kb"]))

    # -- the run -------------------------------------------------------
    def run(self, judge=True) -> int:
        """Send the run; ``judge`` False skips the verdict (the dry run's case)."""
        self.t0 = self.clock()
        if self.report_every:
            self.say(CSV_HEADER)
        n = 0
        try:
            while self.frames is None or n < self.frames:
                self.wait_for(n)
                self.send_frame(n)
                n += 1
                self.poll()
                if self.report_every and n % self.report_every == 0:
                    self.report(n)
        except KeyboardInterrupt:
            self.log("interrupted after %d frames" % n)
        if judge:
            self.drain()
            return self.verdict()
        return 0

    def wait_for(self, n: int):
        """Sleep until ``t0 + n/fps``; the first frame does not sleep."""
        if self.fps <= 0:
            return
        delta = self.t0 + n / self.fps - self.clock()
        if delta > 0:
            self.sleep(delta)

    def report(self, n: int):
        elapsed = self.clock() - self.t0
        rate = n / elapsed if elapsed > 0 else 0.0
        self.say("%d,%.3f,%.2f,%d" % (n, elapsed, rate, self.sent_bytes))

    def drain(self, limit=None):
        """Keep listening after the last frame, for a status still in flight.

        The terminal reports every 64 accepted frames, so the report for the
        last few frames of a short run may never come; the wait is bounded and
        ends early only when a report covers everything that was sent.
        """
        if self.receive is None or self.status_wait <= 0:
            return
        deadline = self.clock() + (self.status_wait if limit is None else limit)
        while self.clock() < deadline:
            self.poll()
            if (self.statuses and self.frames is not None
                    and self.statuses[-1]["frames_rx"] >= self.frames):
                return
            self.sleep(0.02)

    def verdict(self) -> int:
        elapsed = self.clock() - self.t0
        self.log("sent %d frames in %d datagrams, %d bytes, %.1f fps"
                 % (self.sent_frames, self.sent_datagrams, self.sent_bytes,
                    (self.sent_frames / elapsed) if elapsed > 0 else 0.0))
        if not self.statuses:
            self.log("FAIL: no status frame received (the terminal said "
                     "nothing, so drops and crc_err are unknown)")
            return 1
        last = self.statuses[-1]
        if self.bad_status:
            self.log("FAIL: %d status frame(s) had a bad header CRC"
                     % self.bad_status)
            return 1
        if last["drops"] or last["crc_err"]:
            self.log("FAIL: terminal reports drops=%d crc_err=%d"
                     % (last["drops"], last["crc_err"]))
            return 1
        lag = (self.sent_frames - last["frames_rx"]
               if self.frames is not None else 0)
        self.log("PASS: %d status frames, last frames_rx=%d "
                 "(drops=0 crc_err=0), %d frame(s) sent since that report"
                 % (len(self.statuses), last["frames_rx"], max(lag, 0)))
        return 0


# --- tile mode: the cached stream -----------------------------------------

class TileSender:
    """Send a ``make_tiles.py`` cache: the dictionary, then full/delta frames.

    Everything the world touches is injected exactly as in :class:`Sender`, so
    the plan, the wire bytes and the verdict are testable without a socket.

    The frames come from the cache's ``indices`` rather than from its
    ``stream`` records: the sender has to *decide* the keyframe at send time,
    because only it can see the terminal's status frame.  With no drops the
    decision reproduces the cached plan byte for byte -- ``test_vidsend.py``
    pins that -- and with a drop it inserts a full frame the cache never
    planned, which is the whole point of carrying ``drops`` into
    :func:`tiledict.keyframe_due`.

    A clip shorter than the run loops.  ``frame_no`` keeps counting, so the
    cadence and the loop are one timeline: frame 60 is a keyframe whether it is
    clip frame 0 or clip frame 30.
    """

    def __init__(self, cache, transport, src, receive=None, dst=BROADCAST,
                 fps=30.0, frames=None, chunk=CHUNK, clock=time.monotonic,
                 sleep=time.sleep, report_every=1, status_wait=1.0,
                 every=KEYFRAME_EVERY, out=None, err=None):
        if chunk < HEADER_SIZE + 1:
            raise ValueError("chunk %d is too small for a frame" % chunk)
        need = min_chunk(MAX_TILE_FRAME)
        if chunk < need:
            raise ValueError(
                "chunk %d would split a full-screen delta (%d bytes) across "
                "more than the terminal's %d-descriptor RX ring; chunk must "
                "be at least %d"
                % (chunk, MAX_TILE_FRAME, RING_DATAGRAMS, need))
        if fps < 0:
            raise ValueError("fps must be >= 0 (0 = unpaced): %r" % (fps,))
        if not cache.indices:
            raise ValueError("the tile cache has no frames")
        self.cache = cache
        self.transport = transport
        self.receive = receive
        self.src = src
        self.dst = dst
        self.fps = float(fps)
        self.frames = frames
        self.chunk = chunk
        self.every = every
        self.clock = clock
        self.sleep = sleep
        self.report_every = report_every
        self.status_wait = status_wait
        self.out = out if out is not None else sys.stdout
        self.err = err if err is not None else sys.stderr

        self.t0 = 0.0
        self.frames_done = 0        # clip frames the schedule has produced
        self.sent_frames = 0        # frames actually put on the wire
        self.full_frames = 0
        self.delta_frames = 0
        self.same_frames = 0
        self.sent_datagrams = 0
        self.sent_bytes = 0
        self.dictionary_sent = 0     # datagrams the dictionary upload took
        self.keyframes = []         # the clip frame numbers sent as TILES_FULL
        self.previous = None        # the last index frame the terminal has
        self.drops = 0              # the terminal's cumulative drops, latest
        self.seen_drops = 0         # the value the last decision consumed
        self.statuses = []
        self.bad_status = 0
        self.interrupted = False

    # -- output --------------------------------------------------------
    def say(self, text):
        print(text, file=self.out, flush=True)

    def log(self, text):
        print(text, file=self.err, flush=True)

    # -- the wire ------------------------------------------------------
    def dictionary_datagrams(self) -> list[bytes]:
        """The 32 KB dictionary as Ethernet frames, offsets ascending.

        A chunk's own framing is the protocol's (``tiledict.MAX_PAYLOAD``), not
        this sender's ``--chunk``: the two are different sizes for different
        reasons.  Every chunk is one datagram, so each is padded to the
        Ethernet minimum on its own -- none is ever in the middle of a frame.
        """
        return [ethernet_frame(packet, src=self.src, dst=self.dst, pad=True)
                for packet in dict_chunks(self.cache.dictionary)]

    def frame_plan(self, n: int):
        """``(kind, payload, changed, keyframe)`` for clip frame ``n``.

        ``keyframe`` is the decision, ``kind`` the consequence: a full frame is
        always a keyframe, a delta is never one, and SAME (nothing changed
        since the previous frame in the loop) carries no payload at all.
        """
        current = self.cache.indices[n % len(self.cache.indices)]
        changed = 0 if self.previous is None else sum(
            1 for a, b in zip(self.previous, current) if a != b)
        if keyframe_due(n, self.drops, self.seen_drops, self.every):
            return make_tiles.KIND_FULL, bytes(current), changed, True
        payload = delta_frame(self.previous, current)
        if payload is None:
            return make_tiles.KIND_SAME, b"", 0, False
        return make_tiles.KIND_DELTA, payload, changed, False

    def datagrams(self, n: int, kind: int, payload: bytes) -> list[bytes]:
        """``payload`` as one VS frame, chunked and wrapped for the transport."""
        frame = pack_header(kind, n & 0xFFFF, len(payload), 0, 0,
                            payload=payload) + payload
        chunks = [frame[i:i + self.chunk]
                  for i in range(0, len(frame), self.chunk)]
        return [ethernet_frame(chunk, src=self.src, dst=self.dst,
                               pad=(i == len(chunks) - 1))
                for i, chunk in enumerate(chunks)]

    def send_frame(self, n: int, kind: int, payload: bytes):
        for data in self.datagrams(n, kind, payload):
            self.transport(data)
            self.sent_datagrams += 1
            self.sent_bytes += len(data)
        self.sent_frames += 1

    def send_dictionary(self):
        for data in self.dictionary_datagrams():
            self.transport(data)
            self.sent_datagrams += 1
            self.sent_bytes += len(data)
            self.dictionary_sent += 1

    # -- status --------------------------------------------------------
    def poll(self):
        if self.receive is None:
            return
        for data in (self.receive() or [])[:64]:
            st = decode_status_frame(data)
            if st is None:
                continue
            if not st["crc_ok"]:
                self.bad_status += 1
                self.log("status: header CRC bad, ignored (frames_rx=%d)"
                         % st["frames_rx"])
                continue
            self.statuses.append(st)
            self.drops = max(self.drops, st["drops"])
            self.log("status %d: frame_no=%d frames_rx=%d crc_err=%d drops=%d "
                     "csr0=0x%04X heap_free_kb=%d"
                     % (len(self.statuses), st["frame_no"], st["frames_rx"],
                        st["crc_err"], st["drops"], st["csr0"],
                        st["heap_free_kb"]))

    # -- the run -------------------------------------------------------
    def run(self, judge=True) -> int:
        self.t0 = self.clock()
        self.send_dictionary()
        if self.report_every:
            self.say(CSV_HEADER_TILES)
        n = 0
        try:
            while self.frames is None or n < self.frames:
                self.wait_for(n)
                # The status is read *before* the decision: a drop reported
                # now is a drop this frame can repair.
                self.poll()
                kind, payload, changed, keyframe = self.frame_plan(n)
                self.seen_drops = self.drops
                if kind == make_tiles.KIND_SAME:
                    self.same_frames += 1
                else:
                    self.send_frame(n, kind, payload)
                    if kind == make_tiles.KIND_FULL:
                        self.full_frames += 1
                        self.keyframes.append(n)
                    else:
                        self.delta_frames += 1
                self.previous = self.cache.indices[n % len(self.cache.indices)]
                self.frames_done += 1
                n += 1
                if self.report_every and n % self.report_every == 0:
                    # The row names the frame it describes (0-based, the same
                    # number that went in the wire header), not the count.
                    self.report(n - 1, kind, payload, changed, keyframe)
        except KeyboardInterrupt:
            self.interrupted = True
            self.log("interrupted after %d frames" % n)
        if judge:
            self.drain()
            return self.verdict()
        return 0

    def wait_for(self, n: int):
        """Sleep until ``t0 + n/fps``; the first frame does not sleep."""
        if self.fps <= 0:
            return
        delta = self.t0 + n / self.fps - self.clock()
        if delta > 0:
            self.sleep(delta)

    def report(self, n: int, kind: int, payload: bytes, changed: int,
               keyframe: bool):
        elapsed = self.clock() - self.t0
        rate = (n + 1) / elapsed if elapsed > 0 else 0.0
        names = {make_tiles.KIND_FULL: "full", make_tiles.KIND_DELTA: "delta",
                 make_tiles.KIND_SAME: "same"}
        self.say("%d,%.3f,%.2f,%d,%s,%d,%d,%d"
                 % (n, elapsed, rate, self.sent_bytes, names[kind],
                    len(payload), changed, 1 if keyframe else 0))

    def drain(self, limit=None):
        if self.receive is None or self.status_wait <= 0:
            return
        deadline = self.clock() + (self.status_wait if limit is None else limit)
        while self.clock() < deadline:
            self.poll()
            if (self.statuses and self.frames is not None
                    and self.statuses[-1]["frames_rx"] >= self.sent_frames):
                return
            self.sleep(0.02)

    def verdict(self) -> int:
        elapsed = self.clock() - self.t0
        self.log("sent %d frame(s) in %d datagrams, %d bytes, %.1f fps "
                 "(%d full, %d delta, %d unchanged, %d dictionary datagrams)"
                 % (self.sent_frames, self.sent_datagrams, self.sent_bytes,
                    (self.sent_frames / elapsed) if elapsed > 0 else 0.0,
                    self.full_frames, self.delta_frames, self.same_frames,
                    self.dictionary_sent))
        if self.frames is not None and not self.interrupted \
                and self.frames_done < self.frames:
            self.log("FAIL: stopped after %d of %d frame(s)"
                     % (self.frames_done, self.frames))
            return 1
        if not self.statuses:
            self.log("FAIL: no status frame received (the terminal said "
                     "nothing, so drops and crc_err are unknown)")
            return 1
        last = self.statuses[-1]
        if self.bad_status:
            self.log("FAIL: %d status frame(s) had a bad header CRC"
                     % self.bad_status)
            return 1
        if last["drops"] or last["crc_err"]:
            self.log("FAIL: terminal reports drops=%d crc_err=%d"
                     % (last["drops"], last["crc_err"]))
            return 1
        lag = (self.sent_frames - last["frames_rx"]
               if self.frames is not None else 0)
        self.log("PASS: %d status frames, last frames_rx=%d "
                 "(drops=0 crc_err=0), %d frame(s) sent since that report"
                 % (len(self.statuses), last["frames_rx"], max(lag, 0)))
        return 0


# --- transports -----------------------------------------------------------

def parse_mac(text: str) -> bytes:
    parts = text.split(":")
    if len(parts) != 6:
        raise argparse.ArgumentTypeError("expected aa:bb:cc:dd:ee:ff, got %r"
                                         % text)
    try:
        return bytes(int(p, 16) for p in parts)
    except ValueError:
        raise argparse.ArgumentTypeError("not a MAC address: %r" % text)


def mac_of(iface: str) -> bytes:
    with open("/sys/class/net/%s/address" % iface) as f:
        return parse_mac(f.read().strip())


class RawEthernet:
    """AF_PACKET transport: send on ``iface`` and receive the stream's type.

    Two sockets, because they answer two questions: the TX socket sends a frame
    the caller built (header included), the RX socket only queues frames whose
    ethertype is the stream's, so the status frames arrive without the host
    having to sift the interface's whole traffic.  Both need CAP_NET_RAW.
    """

    def __init__(self, iface: str):
        self.iface = iface
        self.tx = socket.socket(socket.AF_PACKET, socket.SOCK_RAW)
        self.tx.bind((iface, 0))
        self.rx = socket.socket(socket.AF_PACKET, socket.SOCK_RAW,
                                socket.htons(ETHERTYPE))
        self.rx.bind((iface, 0))
        self.rx.setblocking(False)
        self.src = mac_of(iface)

    def send(self, data: bytes):
        self.tx.send(data)

    def recv(self):
        out = []
        while True:
            try:
                out.append(self.rx.recv(2048))
            except BlockingIOError:
                return out
            if len(out) >= 64:
                return out

    def close(self):
        self.tx.close()
        self.rx.close()


class CountingTransport:
    """``--dry-run``: count what the schedule would send, touch no hardware."""

    def __init__(self):
        self.datagrams = 0
        self.bytes = 0

    def __call__(self, data: bytes):
        self.datagrams += 1
        self.bytes += len(data)


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--iface", help="interface to send on (needs CAP_NET_RAW)")
    ap.add_argument("--rows", type=int, default=128,
                    help="unique source rows a frame carries (default 128; "
                         "128x8 fills the screen, ruling R3)")
    ap.add_argument("--stretch", type=int, default=8,
                    help="screen rows each source row paints (default 8)")
    ap.add_argument("--fps", type=float, default=30.0,
                    help="frames per second; 0 = as fast as possible")
    ap.add_argument("--frames", type=int, default=0,
                    help="stop after N frames (0 = until interrupted)")
    ap.add_argument("--pattern", default="bars", choices=sorted(PATTERNS),
                    help="test picture (default bars)")
    ap.add_argument("--chunk", type=int, default=CHUNK,
                    help="bytes of frame per datagram (default %d)" % CHUNK)
    ap.add_argument("--dst", type=parse_mac, default=BROADCAST,
                    help="destination MAC (default broadcast: the terminal's "
                         "LANCE accepts broadcast without a filter entry)")
    ap.add_argument("--src", type=parse_mac, default=None,
                    help="source MAC (default: the interface's own)")
    ap.add_argument("--report-every", type=int, default=None,
                    help="CSV line every N frames (default %d, or 1 with "
                         "--tiles: the plan asks for one row per frame)"
                         % REPORT_EVERY)
    ap.add_argument("--status-wait", type=float, default=1.0,
                    help="seconds to keep listening for a status after the "
                         "last frame (default 1.0)")
    ap.add_argument("--dry-run", action="store_true",
                    help="run the schedule against a counter, no socket")
    ap.add_argument("--tiles", action="store_true",
                    help="send the cached tile stream (dictionary + full/delta "
                         "frames) instead of strip frames")
    ap.add_argument("--tiles-dir", default=make_tiles.DEFAULT_CACHE,
                    help="the cache make_tiles.py wrote (default %(default)s)")
    args = ap.parse_args(argv)

    cache = None
    if args.tiles:
        try:
            cache = make_tiles.read_cache(args.tiles_dir)
        except ValueError as exc:
            print("vidsend.py: %s" % exc, file=sys.stderr)
            print("  build one first: python3 harness/udpstream/make_tiles.py "
                  "--out %s" % args.tiles_dir, file=sys.stderr)
            return 2
        # Tile mode has no strip geometry: the screen is the whole 40x32 grid.
        report_every = args.report_every if args.report_every is not None else 1
    else:
        # The geometry is vidsrc's, not this script's: `stretch` must divide
        # the 1024-row screen or the encoder refuses the frame, and --rows 80
        # --stretch 12 (the plan's own example) is such a pair -- 80*12 = 960
        # rows, the other 64 scanlines unpainted (ruling R3).  Report that as
        # a usage error instead of a traceback.
        try:
            StripSource(args.rows, args.stretch)
        except ValueError as e:
            print("vidsend.py: bad geometry: %s\n"
                  "  for a full screen rows*stretch must be 1024 (128x8 is the "
                  "ruled default, R3) and rows may not exceed %d, the "
                  "terminal's ring ceiling." % (e, MAX_ROWS), file=sys.stderr)
            return 2
        report_every = (args.report_every if args.report_every is not None
                        else REPORT_EVERY)

    # A frame split across more datagrams than the terminal's RX ring holds can
    # never be received: the ring runs out of descriptors and the payload drops
    # the partial frame.  This is worst in tile mode (a delta naming every tile
    # is 3,858 bytes) and for a wide strip frame at a small --chunk; reject it
    # here, where the limit can be named, rather than on the wire.
    if cache is not None:
        frame_bytes = MAX_TILE_FRAME
        what = "a full-screen delta"
    else:
        frame_bytes = HEADER_SIZE + args.rows * ROW_BYTES
        what = "a %d-row frame" % args.rows
    need = min_chunk(frame_bytes)
    if args.chunk < need:
        print("vidsend.py: --chunk %d would split %s (%d bytes) across %d "
              "datagrams; the terminal's RX ring holds %d, so --chunk must be "
              "at least %d"
              % (args.chunk, what, frame_bytes,
                 (frame_bytes + args.chunk - 1) // args.chunk,
                 RING_DATAGRAMS, need), file=sys.stderr)
        return 2

    def build_sender(transport, receive, src):
        if cache is not None:
            return TileSender(cache, transport, src=src, receive=receive,
                              fps=args.fps, frames=args.frames or None,
                              chunk=args.chunk, dst=args.dst,
                              report_every=report_every,
                              status_wait=args.status_wait)
        return Sender(transport, src=src, receive=receive, rows=args.rows,
                      stretch=args.stretch, fps=args.fps,
                      frames=args.frames or None, pattern=args.pattern,
                      chunk=args.chunk, dst=args.dst,
                      report_every=report_every,
                      status_wait=args.status_wait)

    if args.dry_run:
        transport = CountingTransport()
        sender = build_sender(transport, None, b"\x02\x00\x00\x00\x00\x01")
        sender.status_wait = 0.0
        sender.run(judge=False)
        # stdout holds the CSV table only; the verdict of a dry run is prose.
        if cache is not None:
            print("RESULT DRY-RUN: %d frames (%d full, %d delta, %d unchanged), "
                  "%d dictionary datagrams, %d datagrams, %d bytes counted; "
                  "nothing left the host and no status was received"
                  % (sender.sent_frames, sender.full_frames,
                     sender.delta_frames, sender.same_frames,
                     sender.dictionary_sent, sender.sent_datagrams,
                     sender.sent_bytes), file=sys.stderr)
        else:
            print("RESULT DRY-RUN: %d frames, %d datagrams, %d bytes counted; "
                  "nothing left the host and no status was received"
                  % (sender.sent_frames, sender.sent_datagrams,
                     sender.sent_bytes), file=sys.stderr)
        return 0

    if not args.iface:
        ap.error("--iface is required unless --dry-run is given")
    try:
        net = RawEthernet(args.iface)
    except PermissionError:
        print("vidsend.py: %s needs CAP_NET_RAW (run as root, or use "
              "--dry-run)" % args.iface, file=sys.stderr)
        return 2
    except OSError as e:
        print("vidsend.py: cannot open %s: %s" % (args.iface, e), file=sys.stderr)
        return 2

    src = args.src if args.src is not None else net.src
    sender = build_sender(net.send, net.recv, src)
    try:
        return sender.run()
    finally:
        net.close()


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