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) <noreply@anthropic.com>
This commit is contained in:
michael
2026-07-05 01:50:39 -07:00
parent 3aab155004
commit 382732b881
5 changed files with 182 additions and 8 deletions

View File

@@ -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("<IH", offset, len(data)) + data
await self._t.send_ng_no_response(Cmd.HF_ISO15693_EML_SETMEM, payload)
@property
def entries(self) -> 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