iecpoc/log.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 
 from typing import Iterable, Iterator
 
-from . import decode, device
+from . import decode, device, petscii
 
 # bracketed state tag, padded so the content columns line up
 _TAG_WIDTH = 8
@@ -32,12 +32,19 @@ class TraceFormatter:
         the symbolic annotations and state events.
     """
 
-    def __init__(self, my_address: int = 4, raw: bool = False):
+    def __init__(self, my_address: int = 4, raw: bool = False, debug: bool = False):
         self.my_address = my_address
         self.raw = raw
+        self.debug = debug
+
+    @property
+    def plain(self) -> bool:
+        return not self.debug and not self.raw
 
     def format(self, rec: device.IecRecord) -> str | None:
         """Format one record. Returns ``None`` for records to be skipped."""
+        if self.plain:
+            return self._format_plain(rec)
         if self.raw:
             # raw mode: just the byte value of command/data records
             if rec.is_command or rec.is_data:
@@ -60,6 +67,23 @@ class TraceFormatter:
                 yield line
 
     # -- per-kind formatters -------------------------------------------------
+    def _format_plain(self, rec: device.IecRecord) -> str | None:
+        if rec.is_event:
+            return None
+        if rec.is_command:
+            cmd = decode.decode_command(rec.value)
+            if cmd.mnemonic == decode.LISTEN:
+                return f"\n--- LISTEN {cmd.primary} ---\n"
+            if cmd.mnemonic == decode.UNLISTEN:
+                return "\n--- UNLISTEN ---\n"
+            return None
+        if rec.is_data:
+            if rec.value in (0x0D, 0x0A, 0x8D):  # CR, LF, SHIFT-CR
+                return "\n"
+            g = petscii.to_glyph(rec.value)
+            return g.text if g.printable else None
+        return None
+
     def _format_event(self, rec: device.IecRecord) -> str:
         text = decode.describe_event(rec)
         if rec.value == device.EV_IDLE:

iecpoc/main.py

@@ -33,6 +33,10 @@ def build_parser() -> argparse.ArgumentParser:
         "--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",
@@ -47,11 +51,12 @@ def build_parser() -> argparse.ArgumentParser:
 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, file=out, flush=True)
+            print(line, end=line_end, file=out, flush=True)
             if logfile is not None:
-                print(line, file=logfile, flush=True)
+                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.
@@ -61,7 +66,7 @@ def run(records: Iterable[device.IecRecord], fmt: TraceFormatter,
 
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
-    fmt = TraceFormatter(my_address=args.address, raw=args.raw)
+    fmt = TraceFormatter(my_address=args.address, raw=args.raw, debug=args.debug)
 
     logfile = open(args.logfile, "a") if args.logfile else None
     try: