92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
"""CLI entry point for the IEC listener PoC userspace (PLAN.md §5).
|
|
|
|
Reads the tagged record stream from the kernel char device (``/dev/iec0``) -- or
|
|
from a captured file via ``--replay`` for host testing -- and prints the
|
|
annotated debug trace. The kernel module owns all timing; this process has none.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from typing import Iterable, TextIO
|
|
|
|
from . import device
|
|
from .log import TraceFormatter
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(
|
|
prog="iecpoc",
|
|
description="Commodore IEC listener PoC -- decode and print what the C64 sends.",
|
|
)
|
|
p.add_argument(
|
|
"-a", "--address", type=int, default=4,
|
|
help="our primary device address (default: 4, a printer)",
|
|
)
|
|
p.add_argument(
|
|
"-d", "--device", default="/dev/iec0",
|
|
help="kernel character device to read (default: /dev/iec0)",
|
|
)
|
|
p.add_argument(
|
|
"--replay", metavar="FILE",
|
|
help="replay a captured record file instead of opening the device",
|
|
)
|
|
p.add_argument(
|
|
"--raw", action="store_true",
|
|
help="dump a bare hex stream instead of the annotated trace",
|
|
)
|
|
p.add_argument(
|
|
"--logfile", metavar="FILE",
|
|
help="also append the trace to this file",
|
|
)
|
|
return p
|
|
|
|
|
|
def run(records: Iterable[device.IecRecord], fmt: TraceFormatter,
|
|
out: TextIO, logfile: TextIO | None = None) -> None:
|
|
"""Format ``records`` through ``fmt`` and write lines to ``out`` (+ logfile)."""
|
|
try:
|
|
for line in fmt.format_stream(records):
|
|
print(line, file=out, flush=True)
|
|
if logfile is not None:
|
|
print(line, file=logfile, flush=True)
|
|
except BrokenPipeError:
|
|
# downstream (e.g. `head`) closed the pipe; redirect stdout to devnull so
|
|
# the interpreter's shutdown flush doesn't re-raise, then exit quietly.
|
|
devnull = os.open(os.devnull, os.O_WRONLY)
|
|
os.dup2(devnull, sys.stdout.fileno())
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
fmt = TraceFormatter(my_address=args.address, raw=args.raw)
|
|
|
|
logfile = open(args.logfile, "a") if args.logfile else None
|
|
try:
|
|
if args.replay:
|
|
with open(args.replay, "rb") as fh:
|
|
run(device.iter_records(fh), fmt, sys.stdout, logfile)
|
|
else:
|
|
try:
|
|
with device.IecDevice(args.device) as dev:
|
|
run(dev, fmt, sys.stdout, logfile)
|
|
except FileNotFoundError:
|
|
print(
|
|
f"error: {args.device} not found -- is the kernel module loaded? "
|
|
f"(use --replay FILE to replay a capture on a host)",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
except KeyboardInterrupt:
|
|
return 0
|
|
finally:
|
|
if logfile is not None:
|
|
logfile.close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|