Christian Werner
d359aa4acc
Default output now shows only the received characters; CR/LF become real newlines and LISTEN/UNLISTEN print a clearly visible divider. Pass --debug to restore the previous full annotated trace. Generated by Clanker
97 lines
3.2 KiB
Python
97 lines
3.2 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(
|
|
"--debug", action="store_true",
|
|
help="print the full annotated trace instead of plain received text",
|
|
)
|
|
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)."""
|
|
line_end = "" if fmt.plain else "\n"
|
|
try:
|
|
for line in fmt.format_stream(records):
|
|
print(line, end=line_end, file=out, flush=True)
|
|
if logfile is not None:
|
|
print(line, end=line_end, 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, debug=args.debug)
|
|
|
|
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())
|