vt1200/harness/udpstream/send.py

#!/usr/bin/env python3
"""Host sender for the VT1200 udpstream spike.

Sends full 1280x1024 frames as 8-row datagrams:

    +0  u16 start row (LE)      +2  u16 row count      +4  rows*160 bytes

Usage:
    python3 harness/udpstream/send.py --host 100.103.100.103 --fps 5
    python3 harness/udpstream/send.py --host 100.103.100.103 --fps 0 --seconds 5

The frame is a black field with a white bar (encoding the frame number), a
moving box, and a border, so a screenshot proves the whole frame arrived and
which frame it was.
"""
from __future__ import annotations

import argparse
import socket
import struct
import sys
import time

W, H = 1280, 1024
ROW_BYTES = W // 8          # 160
ROWS_PER_PKT = 8
PORT = 7000


def frame(n: int) -> bytes:
    """One 1-bpp frame, packed rows of ROW_BYTES, LSB-first."""
    rows = bytearray(H * ROW_BYTES)
    # border: rows 0,1,H-2,H-1 and columns 0..3 / W-4..W-1 white
    for y in (0, 1, H - 2, H - 1):
        rows[y * ROW_BYTES:(y + 1) * ROW_BYTES] = b"\xff" * ROW_BYTES
    for y in range(H):
        r = rows[y * ROW_BYTES:(y + 1) * ROW_BYTES]
        r[0] = 0xFF
        r[ROW_BYTES - 1] = 0xFF
        rows[y * ROW_BYTES:(y + 1) * ROW_BYTES] = r

    # moving 128x128 box
    x0 = (n * 37) % (W - 128)
    y0 = (n * 23) % (H - 128)
    for y in range(y0, y0 + 128):
        base = y * ROW_BYTES
        for xb in range(x0 // 8, (x0 + 128) // 8):
            rows[base + xb] = 0xFF

    # frame-number bar on row 64: 16 pixels of white per 16-frame step
    y = 64
    base = y * ROW_BYTES
    for i in range(n % 64):
        x = 32 + i * 16
        rows[base + (x // 8)] |= 0xFF
        rows[base + (x // 8) + 1] |= 0xFF

    return bytes(rows)


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--host", default="100.103.100.103")
    ap.add_argument("--port", type=int, default=PORT)
    ap.add_argument("--fps", type=float, default=5.0,
                    help="target frames/s; 0 = as fast as possible")
    ap.add_argument("--pkt-gap-ms", type=float, default=0.0,
                    help="sleep between datagrams; gives the cooperative OS a "
                         "chance to run (a continuous burst starves it)")
    ap.add_argument("--seconds", type=float, default=0,
                    help="stop after N seconds (0 = run until Ctrl-C)")
    a = ap.parse_args(argv[1:])

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1 << 20)
    dst = (a.host, a.port)

    n = sent_bytes = sent_pkts = 0
    t0 = last = time.monotonic()
    frames_at_last = 0

    try:
        while True:
            f = frame(n)
            for r in range(0, H, ROWS_PER_PKT):
                payload = f[r * ROW_BYTES:(r + ROWS_PER_PKT) * ROW_BYTES]
                pkt = struct.pack("<HH", r, ROWS_PER_PKT) + payload
                sock.sendto(pkt, dst)
                sent_bytes += len(pkt)
                sent_pkts += 1
                if a.pkt_gap_ms:
                    time.sleep(a.pkt_gap_ms / 1000.0)
            n += 1

            now = time.monotonic()
            if now - last >= 1.0:
                rate = (n - frames_at_last) / (now - last)
                print(f"{n:6d} frames  {rate:5.2f} fps  "
                      f"{sent_bytes / (now - t0) / 1e6:5.3f} MB/s  "
                      f"{sent_pkts / (now - t0):6.0f} pkt/s", flush=True)
                last = now
                frames_at_last = n
            if a.seconds and now - t0 >= a.seconds:
                break
            if a.fps > 0:
                sleep = t0 + n / a.fps - time.monotonic()
                if sleep > 0:
                    time.sleep(sleep)
    except KeyboardInterrupt:
        pass
    finally:
        dt = time.monotonic() - t0
        print(f"sent {n} frames, {sent_bytes} bytes in {dt:.1f}s "
              f"({sent_bytes / dt / 1e6:.2f} MB/s, {sent_pkts / dt:.0f} pkt/s)",
              flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))