From 382732b881ea3fe68478ee8af24e09f10bbedb8c Mon Sep 17 00:00:00 2001 From: michael Date: Sun, 5 Jul 2026 01:50:39 -0700 Subject: [PATCH] feat(trace): scriptable on_frame callback + quiet mode + SimSession entries SniffSession/SimSession.start_15693 gain on_frame(frame) and quiet kwargs (defaults preserve today's print behavior). Each frame is a plain dict {protocol, direction, timestamp, duration, data, data_hex, decoded[, crc_fail]} with no ANSI. SimSession now accumulates frames (entries property + clear(), mirroring SniffSession) and auto-streams when trace or on_frame is set; printing is gated on trace and not quiet. TraceFormatter.decode() exposes the annotation without color; format()/print() untouched. Co-Authored-By: Claude Opus 4.8 (1M context) --- pm3py/sim/sim_session.py | 53 +++++++++++++++-- pm3py/sniff/session.py | 17 +++++- pm3py/trace/format.py | 15 +++++ tests/test_sim_trace_fmt.py | 1 + tests/test_trace_scripting.py | 104 ++++++++++++++++++++++++++++++++++ 5 files changed, 182 insertions(+), 8 deletions(-) create mode 100644 tests/test_trace_scripting.py diff --git a/pm3py/sim/sim_session.py b/pm3py/sim/sim_session.py index 1072203..c200471 100644 --- a/pm3py/sim/sim_session.py +++ b/pm3py/sim/sim_session.py @@ -50,6 +50,10 @@ class SimSession: self._tag_model: Transponder | None = None self._formatter: TraceFormatter | None = None self.on_field_strength: callable | None = None # callback(adc_mv: int) + self._entries: list[dict] = [] + self._on_frame = None + self._quiet = False + self._print_frames = False @classmethod def open(cls, port: str = DEFAULT_PORT, baudrate: int = 115200) -> "SimSession": @@ -122,7 +126,8 @@ class SimSession: self._active = True self._relay_task = asyncio.create_task(self._relay_loop(tag)) - def start_15693(self, tag, compile: bool = True, trace: bool = False) -> None: + def start_15693(self, tag, compile: bool = True, trace: bool = False, + on_frame=None, quiet: bool = False) -> None: """Start 15693 sim (synchronous). Firmware handles standard commands (inventory, read, write) autonomously. @@ -131,8 +136,16 @@ class SimSession: Args: trace: If True, print live reader↔tag communication to console. + on_frame: optional callback(frame: dict) invoked per decoded frame from + the reader thread — for scripting (see the `entries` property for the + dict shape). Passing it auto-enables trace streaming. + quiet: if True, don't print frames (callback + `entries` still fill). """ self._tag_model = tag + self._on_frame = on_frame + self._quiet = quiet + self._print_frames = trace and not quiet + stream = trace or (on_frame is not None) # Resolve the raw serial port if self._port: @@ -147,7 +160,7 @@ class SimSession: # Start sim with UID and block_size — same format as PM3 client # Client sends UID as-is (firmware reverses internally) - flags = 0x02 if trace else 0x00 + flags = 0x02 if stream else 0x00 block_size = tag._block_size if hasattr(tag, '_block_size') else 4 payload = tag._uid + bytes([block_size, flags]) frame = encode_ng_frame(Cmd.HF_ISO15693_SIMULATE, payload) @@ -168,7 +181,7 @@ class SimSession: if compile: self._compile_and_upload_table(tag, ser) - if trace: + if stream: self._formatter = TraceFormatter(mode="sim", decoder=tag.decode_trace, crc_len=2) self._trace_thread = threading.Thread( target=self._trace_reader, args=(ser,), daemon=True @@ -176,7 +189,7 @@ class SimSession: self._trace_thread.start() print(f"[SimSession] 15693 sim started, UID={tag._uid.hex()}" - + (" (trace ON)" if trace else "")) + + (" (trace ON)" if stream else "")) print(f"[SimSession] Modify tag in REPL, then call tag.sync()") def _compile_and_upload_table(self, tag, ser) -> None: @@ -280,7 +293,23 @@ class SimSession: else: crc_fail = bool(data[0] & 0x80) payload = data[1:] - self._formatter.print(direction, payload, crc_fail=crc_fail) + entry = { + "protocol": 0, # ISO 15693 + "direction": direction, + "timestamp": None, + "duration": None, + "data": payload, + "data_hex": payload.hex(), + "decoded": self._formatter.decode( + direction, payload, crc_fail), + "crc_fail": crc_fail, + } + self._entries.append(entry) + if self._on_frame is not None: + self._on_frame(entry) + if self._print_frames: + self._formatter.print( + direction, payload, crc_fail=crc_fail) except Exception: buf = buf[4:] # skip bad magic, try again @@ -361,6 +390,20 @@ class SimSession: payload = struct.pack(" list[dict]: + """Decoded reader↔tag frames captured while streaming (a copy). + + Each: {'protocol', 'direction', 'timestamp', 'duration', 'data': bytes, + 'data_hex': str, 'decoded': str|None, 'crc_fail': bool}. Populated when + the sim runs with trace=True or an on_frame callback. + """ + return list(self._entries) + + def clear(self) -> None: + """Discard accumulated frames.""" + self._entries.clear() + def stop(self) -> None: """Stop simulation.""" self._active = False diff --git a/pm3py/sniff/session.py b/pm3py/sniff/session.py index 57eb19d..be61d10 100644 --- a/pm3py/sniff/session.py +++ b/pm3py/sniff/session.py @@ -65,6 +65,8 @@ class SniffSession: self._reader_thread: threading.Thread | None = None self._formatter: TraceFormatter | None = None self._state = "idle" + self._on_frame = None + self._quiet = False @classmethod def open(cls, port: str = DEFAULT_PORT, baudrate: int = 115200) -> "SniffSession": @@ -88,14 +90,20 @@ class SniffSession: return cls(ser) - def start_15693(self, live: bool = True, idle_timeout_ms: int = 400) -> None: + def start_15693(self, live: bool = True, idle_timeout_ms: int = 400, + on_frame=None, quiet: bool = False) -> None: """Start 15693 sniff session. live=True (default): streaming with button-toggle pause/resume. live=False: legacy BigBuf batch mode, blocks until button exits. + on_frame: optional callback(frame: dict) invoked per decoded frame from the + reader thread — for scripting (see the `entries` property for the dict shape). + quiet: if True, don't print frames to the console (callback + `entries` still fill). """ if self._active: self.stop() + self._on_frame = on_frame + self._quiet = quiet self._protocol = 0 # 15693 flags = 0 @@ -190,10 +198,13 @@ class SniffSession: "duration": duration, "timestamp": timestamp, "data": payload, + "data_hex": payload.hex(), + "decoded": self._formatter.decode(direction, payload) if self._formatter else None, } self._entries.append(entry) - - if self._formatter: + if self._on_frame is not None: + self._on_frame(entry) + if self._formatter and not self._quiet: self._formatter.print(direction, payload) def _handle_status(self, data: bytes) -> None: diff --git a/pm3py/trace/format.py b/pm3py/trace/format.py index c3aa2a3..7697a71 100644 --- a/pm3py/trace/format.py +++ b/pm3py/trace/format.py @@ -65,6 +65,21 @@ class TraceFormatter: return text return f"{code}{text}{_C_RESET}" + def decode(self, direction: int, payload: bytes, crc_fail: bool = False) -> str | None: + """Return the plain-text decoded annotation for a frame — no ANSI, no hex. + + The same annotation ``format()`` renders (e.g. "INVENTORY", "READ BLOCK 4"), + exposed as structured data for scripting/callbacks. + """ + if self._crc_len > 0 and len(payload) > self._crc_len: + data = payload[:-self._crc_len] + else: + data = payload + annotation = self._decoder(direction, data) if self._decoder is not None else None + if crc_fail: + annotation = f"{annotation} BAD CRC" if annotation else "BAD CRC" + return annotation + def format(self, direction: int, payload: bytes, crc_fail: bool = False) -> str: """Format a trace line with colors, decoding, and wrapping.""" self._line_count += 1 diff --git a/tests/test_sim_trace_fmt.py b/tests/test_sim_trace_fmt.py index 23b5954..b62c73b 100644 --- a/tests/test_sim_trace_fmt.py +++ b/tests/test_sim_trace_fmt.py @@ -561,6 +561,7 @@ class TestSimSessionTraceWiring: session = SimSession() session._formatter = mock_fmt session._active = True + session._print_frames = True # start_15693(trace=True) sets this frame = self._make_trace_frame(0, bytes([0x26, 0x01, 0x00])) mock_serial = MagicMock() diff --git a/tests/test_trace_scripting.py b/tests/test_trace_scripting.py new file mode 100644 index 0000000..6e1accc --- /dev/null +++ b/tests/test_trace_scripting.py @@ -0,0 +1,104 @@ +"""Scriptable trace access: on_frame callback, quiet mode, entries accumulator.""" +import struct +from unittest.mock import MagicMock + +from pm3py.trace.format import TraceFormatter +from pm3py.trace import decode_15693 + +INV = bytes.fromhex("260100f60a") # ISO15693 INVENTORY request + CRC + + +def _fmt(mode="sniff"): + return TraceFormatter(mode=mode, decoder=decode_15693, crc_len=2, is_tty=False) + + +def _sniff_trace(direction, payload): + # CMD_HF_SNIFF_STREAM entry: {protocol,direction,duration,timestamp} + payload + return struct.pack(" no console output + + +def test_sniff_default_still_prints(capsys): + from pm3py.sniff.session import SniffSession + s = SniffSession(MagicMock()) + s._formatter = _fmt() + # defaults preserved: _on_frame=None, _quiet=False + s._handle_trace(_sniff_trace(0, INV)) + assert len(s.entries) == 1 + assert capsys.readouterr().out != "" # prints as before + + +# ---------------- SimSession ---------------- + +def test_sim_on_frame_and_quiet(capsys): + from pm3py.sim.sim_session import SimSession + s = SimSession() + s._formatter = _fmt(mode="sim") + s._active = True + frames = [] + s._on_frame = frames.append + s._print_frames = False # quiet + + _run_sim_reader(s, _sim_trace_frame(0, INV)) + + assert len(frames) == 1 and len(s.entries) == 1 + f = frames[0] + assert f["direction"] == 0 and f["data"] == INV + assert f["data_hex"] == INV.hex() + assert f["decoded"] == "INVENTORY" and f["crc_fail"] is False + assert capsys.readouterr().out == "" + + +def test_sim_entries_accumulate_and_clear(): + from pm3py.sim.sim_session import SimSession + s = SimSession() + s._formatter = _fmt(mode="sim") + s._active = True + _run_sim_reader(s, _sim_trace_frame(0, INV)) + assert len(s.entries) == 1 + s.clear() + assert s.entries == []