mirror of
https://git.k-space.ee/arti/vt1200.git
synced 2026-10-30 12:03:54 +02:00
86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Replay a captured byte stream — usable as a latd service command.
|
|
|
|
Designed to be handed to latd as the service command, so the VT1200 receives
|
|
exactly the captured bytes:
|
|
|
|
latcp -A -a REPLAY -C 'python3 /path/replay_stream.py stream.bin' -r 100
|
|
|
|
Then point the terminal at the REPLAY service. `--limit` sends only the first
|
|
N bytes, which is how we bisect for the minimal crashing prefix.
|
|
|
|
python3 replay_stream.py stream.bin --limit 4096 --chunk 85 --pace 0.02
|
|
"""
|
|
import argparse
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
DEFAULT_CHUNK = 85 # matches the slot granularity in filter-test2.pcap
|
|
|
|
|
|
def make_raw(fd):
|
|
"""Stop the tty line discipline from rewriting our bytes (\\n -> \\r\\n etc.)."""
|
|
try:
|
|
import termios
|
|
import tty
|
|
tty.setraw(fd, termios.TCSANOW)
|
|
except Exception:
|
|
pass # not a tty: nothing to do
|
|
|
|
|
|
def replay(data, out, chunk=DEFAULT_CHUNK, pace=0.0, sleeper=time.sleep):
|
|
sent = 0
|
|
for i in range(0, len(data), chunk):
|
|
piece = data[i:i + chunk]
|
|
out.write(piece)
|
|
out.flush()
|
|
sent += len(piece)
|
|
if pace:
|
|
sleeper(pace)
|
|
return sent
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("stream")
|
|
ap.add_argument("--limit", type=lambda s: int(s, 0),
|
|
help="send only the first N bytes")
|
|
ap.add_argument("--chunk", type=int, default=DEFAULT_CHUNK)
|
|
ap.add_argument("--pace", type=float, default=0.0,
|
|
help="seconds to sleep between chunks")
|
|
ap.add_argument("--delay", type=float, default=1.0,
|
|
help="initial delay before output (let the circuit settle)")
|
|
ap.add_argument("--no-raw", action="store_true",
|
|
help="do not set the tty to raw mode")
|
|
ap.add_argument("--quiet", action="store_true")
|
|
args = ap.parse_args(argv)
|
|
|
|
with open(args.stream, "rb") as fh:
|
|
data = fh.read()
|
|
if args.limit is not None:
|
|
data = data[:args.limit]
|
|
|
|
if not args.no_raw:
|
|
make_raw(sys.stdout.fileno())
|
|
if args.delay:
|
|
time.sleep(args.delay)
|
|
|
|
if not args.quiet:
|
|
sys.stderr.write(f"replay: {len(data)} bytes, chunk={args.chunk}, "
|
|
f"pace={args.pace}\n")
|
|
sys.stderr.flush()
|
|
|
|
sent = replay(data, sys.stdout.buffer, args.chunk, args.pace)
|
|
if not args.quiet:
|
|
sys.stderr.write(f"replay: sent {sent} bytes\n")
|
|
sys.stderr.flush()
|
|
# keep the pty open briefly so latd flushes the last slots
|
|
time.sleep(1.0)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|