forked from arti/vt1200
483 lines
19 KiB
Python
483 lines
19 KiB
Python
#!/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
|
|
python3 harness/udpstream/vidsend.py --dry-run --frames 200 # no NIC
|
|
|
|
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.frame import ( # noqa: E402
|
|
BROADCAST, ETHERTYPE, HEADER_SIZE, KIND_STATUS, MAGIC, STATUS_SIZE,
|
|
ethernet_frame, unpack_header, unpack_status)
|
|
from harness.udpstream.vidsrc import ( # noqa: E402
|
|
CHUNK, MAX_ROWS, ROW_BYTES, StripSource)
|
|
|
|
__all__ = [
|
|
"CSV_HEADER", "Sender", "bars", "split", "PATTERNS",
|
|
"decode_status_frame", "RawEthernet", "parse_mac",
|
|
]
|
|
|
|
#: One CSV line is printed every this many frames.
|
|
REPORT_EVERY = 30
|
|
|
|
#: The 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"
|
|
|
|
#: 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 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)
|
|
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
|
|
|
|
|
|
# --- 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=REPORT_EVERY,
|
|
help="CSV line every N frames (default %d)"
|
|
% 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")
|
|
args = ap.parse_args(argv)
|
|
|
|
# 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
|
|
|
|
if args.dry_run:
|
|
transport = CountingTransport()
|
|
sender = Sender(transport, src=b"\x02\x00\x00\x00\x00\x01",
|
|
rows=args.rows, stretch=args.stretch, fps=args.fps,
|
|
frames=args.frames or None, pattern=args.pattern,
|
|
chunk=args.chunk, dst=args.dst, receive=None,
|
|
report_every=args.report_every, status_wait=0.0)
|
|
sender.run(judge=False)
|
|
# stdout holds the CSV table only; the verdict of a dry run is prose.
|
|
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 = Sender(net.send, src=src, receive=net.recv, 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=args.report_every,
|
|
status_wait=args.status_wait)
|
|
try:
|
|
return sender.run()
|
|
finally:
|
|
net.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|