From fd16fa8f90d560b0790ea45291e1e98093f22862 Mon Sep 17 00:00:00 2001 From: michael Date: Wed, 15 Jul 2026 21:25:16 -0700 Subject: [PATCH] feat(trace): live terminal output under bpython via a shared emit() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live sniff/sim stream trace frames from a background thread. bpython replaces sys.stdout with a capture object whose write() does a greenlet switch back to the main thread — which raises off-thread; the text buffers but only repaints on the next keypress, so streamed frames 'pile up until you touch a key'. New trace/output.py emit() buffers via the repl's on_write and fires bpython's threadsafe repaint trigger; a plain write everywhere else, with a safe fallback on any internals mismatch. Route TraceFormatter.print(), SimSession's sim-ended status, and SniffSession's state message through emit(). Flip the trace-line newline convention from leading to trailing so each emitted line self-terminates (format_sniff_line + test updated to match). Co-Authored-By: Claude Fable 5 --- pm3py/sim/sim_session.py | 3 +- pm3py/sniff/session.py | 5 +-- pm3py/trace/format.py | 14 +++---- pm3py/trace/output.py | 75 +++++++++++++++++++++++++++++++++++++ tests/test_sim_trace_fmt.py | 4 +- 5 files changed, 88 insertions(+), 13 deletions(-) create mode 100644 pm3py/trace/output.py diff --git a/pm3py/sim/sim_session.py b/pm3py/sim/sim_session.py index 29b65f4..d7bce53 100644 --- a/pm3py/sim/sim_session.py +++ b/pm3py/sim/sim_session.py @@ -12,6 +12,7 @@ from .frame import RFFrame from .table_compiler import ResponseTable, TableCompiler from pm3py.trace.format import TraceFormatter from pm3py.trace.decode_iso14a import decode_14443a +from pm3py.trace.output import emit from .transponder import Transponder from ..core.protocol import Cmd from ..core.transport import encode_ng_frame, decode_response_frame, RESP_PREAMBLE_MAGIC, RESP_PREAMBLE_SIZE, RESP_POSTAMBLE_SIZE @@ -394,7 +395,7 @@ class SimSession: # HF_ISO14443A_SIMULATE (0x0384) but its teardown reply # is HF_MIFARE_SIMULATE (0x0610) — watch for the latter. self._active = False - print(f"[SimSession] 14a sim ended, status={resp['status']}") + emit(f"[SimSession] 14a sim ended, status={resp['status']}\n") return if resp["cmd"] in _TRACE_CMDS: diff --git a/pm3py/sniff/session.py b/pm3py/sniff/session.py index be61d10..60a35ca 100644 --- a/pm3py/sniff/session.py +++ b/pm3py/sniff/session.py @@ -1,6 +1,5 @@ """Sniff sessions — live streaming with button-toggle, or legacy batch.""" import struct -import sys import threading import time @@ -19,6 +18,7 @@ from ..core.transport import ( from ..trace import parse_tracelog from ..trace.decode_iso15 import decode_15693 from ..trace.format import TraceFormatter, format_sniff_line +from ..trace.output import emit # Default serial port (same as SimSession) DEFAULT_PORT = "/dev/ttyACM0" @@ -220,8 +220,7 @@ class SniffSession: count_str = f" ({count} frames)" if count > 0 else "" msg = f"\n[Snf] {icon} {label}{count_str}\n" - sys.stdout.write(msg) - sys.stdout.flush() + emit(msg) if state == SNIFF_STATE_STOPPED: self._active = False diff --git a/pm3py/trace/format.py b/pm3py/trace/format.py index 7697a71..b9b65fc 100644 --- a/pm3py/trace/format.py +++ b/pm3py/trace/format.py @@ -6,6 +6,7 @@ import signal import sys from .decode_iso15 import decode_15693 +from .output import emit # ---- ANSI colors ---- _C_CYAN = "\033[36m" @@ -150,7 +151,7 @@ class TraceFormatter: line = f"{prefix_colored}{hex_colored}{' ' * gap}{ann_colored}" else: line = f"{prefix_colored}{hex_colored}" - return f"\n{line}" + return f"{line}\n" # Multi-line pad = " " * prefix_len @@ -168,7 +169,7 @@ class TraceFormatter: ann_pad = max(avail - len(al), 0) parts.append(f"{pad}{' ' * ann_pad}{self._color(ann_color, al)}") - return "\n" + "\n".join(parts) + return "\n".join(parts) + "\n" def _color_hex_line(self, line: str, data_hex: str, crc_hex: str) -> str: if not crc_hex: @@ -229,9 +230,8 @@ class TraceFormatter: return lines or [""] def print(self, direction: int, payload: bytes, crc_fail: bool = False) -> None: - """Format and print a trace line to stdout.""" - sys.stdout.write(self.format(direction, payload, crc_fail=crc_fail)) - sys.stdout.flush() + """Format and print a trace line, live even under bpython.""" + emit(self.format(direction, payload, crc_fail=crc_fail)) def format_sniff_line(entry: dict, width: int = 120, is_tty: bool = True) -> str: @@ -243,5 +243,5 @@ def format_sniff_line(entry: dict, width: int = 120, is_tty: bool = True) -> str width=width, is_tty=is_tty, crc_len=2) direction = 1 if entry["is_response"] else 0 result = fmt.format(direction, entry["data"]) - # Strip leading newline that TraceFormatter adds - return result.lstrip("\n") + # Strip the trailing newline that TraceFormatter adds + return result.rstrip("\n") diff --git a/pm3py/trace/output.py b/pm3py/trace/output.py new file mode 100644 index 0000000..a1569ee --- /dev/null +++ b/pm3py/trace/output.py @@ -0,0 +1,75 @@ +"""Terminal output that stays live under interactive REPLs. + +Live sniff and sim stream trace frames from a background thread. Writing those +straight to ``sys.stdout`` is fine in a plain REPL, but bpython replaces stdout +with a capture object whose ``write`` hands control back to the main REPL thread +via a greenlet switch — which raises from any other thread. The text is buffered +first, so it survives, but the display only repaints on the next keypress, so +streamed frames appear to "pile up until you touch a key". + +bpython exposes exactly the escape hatch we need: ``_interrupting_refresh_callback`` +is a curtsies ``threadsafe_event_trigger`` built to wake the input loop from +another thread. So under bpython we buffer the text directly (via the repl's +``on_write``, skipping the greenlet-switching ``write``) and then fire that +trigger to force an immediate repaint. Everywhere else — plain python, IPython — +it is just a direct write. All of the bpython plumbing is internal, so any +mismatch falls back to a plain write and the sniff keeps working (just +keypress-flushed) rather than breaking. + +The detection is intentionally bpython-only: the "wake the render loop" step is +per-REPL and there is no universal hook, so we special-case the one REPL we use +and let every other environment take the direct-write default. +""" +from __future__ import annotations + +import sys + + +def _bpython_sink(): + """Return ``(on_write, wake)`` for the running bpython repl, else ``None``. + + ``sys.stdout`` under bpython is a ``FakeOutput`` whose ``on_write`` is a bound + method of the running repl; that repl carries the thread-safe repaint trigger. + Every access here is bpython-internal, so a miss just yields ``None``. + """ + out = sys.stdout + if type(out).__module__.split(".", 1)[0] != "bpython": + return None + on_write = getattr(out, "on_write", None) + repl = getattr(on_write, "__self__", None) + wake = getattr(repl, "_interrupting_refresh_callback", None) + if callable(on_write) and callable(wake): + return on_write, wake + return None + + +def emit(text: str) -> None: + """Write trace ``text`` to the terminal, live even under bpython. + + Safe to call from a background thread. Under bpython the text is buffered and + an immediate repaint is triggered; otherwise it is a plain stdout write. + """ + sink = None + try: + sink = _bpython_sink() + except Exception: + sink = None + + if sink is not None: + on_write, wake = sink + try: + on_write(text) + except Exception: + pass # fall through to a plain write below + else: + try: + wake() + except Exception: + pass + return + + try: + sys.stdout.write(text) + sys.stdout.flush() + except Exception: + pass diff --git a/tests/test_sim_trace_fmt.py b/tests/test_sim_trace_fmt.py index b62c73b..5b2d6c3 100644 --- a/tests/test_sim_trace_fmt.py +++ b/tests/test_sim_trace_fmt.py @@ -232,10 +232,10 @@ class TestTraceFormatterBasic: import re return re.sub(r'\033\[[0-9;]*m', '', s) - def test_starts_with_newline(self): + def test_ends_with_newline(self): fmt = TraceFormatter(mode="sim", decoder=decode_15693) result = fmt.format(0, bytes([0x26, 0x01, 0x00])) - assert result.startswith("\n") + assert result.endswith("\n") def test_sim_mode_tag(self): fmt = TraceFormatter(mode="sim", decoder=decode_15693)