# Trace Formatter Implementation Plan > **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. **Goal:** Replace raw `print()` trace output with colored, decoded, column-wrapped trace rendering for sim sessions. **Architecture:** New `trace_fmt.py` module with `TraceFormatter` class + protocol decode functions. `sim_session.py` delegates to it. Terminal width detected dynamically with SIGWINCH support. **Tech Stack:** Python stdlib only (os, signal, sys, struct, shutil). No external dependencies. --- ### Task 1: ISO 15693 request decoder **Files:** - Create: `pm3py/sim/trace_fmt.py` - Test: `tests/test_sim_trace_fmt.py` **Step 1: Write failing tests for 15693 request decoding** ```python """Tests for pm3py.sim.trace_fmt — trace formatting and protocol decoding.""" import os import pytest from pm3py.sim.trace_fmt import decode_15693 class TestDecode15693Request: """Decode reader->tag (direction=0) 15693 commands.""" def test_inventory(self): # flags=0x26, cmd=0x01, mask_len=0 payload = bytes([0x26, 0x01, 0x00]) result = decode_15693(0, payload) assert result == "INVENTORY" def test_inventory_with_mask(self): payload = bytes([0x26, 0x01, 0x08, 0xAB]) result = decode_15693(0, payload) assert result == "INVENTORY mask=8" def test_stay_quiet(self): payload = bytes([0x22, 0x02]) + b"\x01" * 8 result = decode_15693(0, payload) assert result == "STAY QUIET" def test_read_single_block_addressed(self): # flags=0x22 (addressed), cmd=0x20, uid(8), block=0x03 payload = bytes([0x22, 0x20]) + b"\x01" * 8 + bytes([0x03]) result = decode_15693(0, payload) assert result == "READ SINGLE BLOCK #3" def test_read_single_block_unaddressed(self): # flags=0x02 (no address flag), cmd=0x20, block=0x00 payload = bytes([0x02, 0x20, 0x00]) result = decode_15693(0, payload) assert result == "READ SINGLE BLOCK #0" def test_write_single_block(self): payload = bytes([0x22, 0x21]) + b"\x01" * 8 + bytes([0x05]) + b"\xDE\xAD\xBE\xEF" result = decode_15693(0, payload) assert result == "WRITE SINGLE BLOCK #5 [4B]" def test_read_multiple_block(self): # flags=0x22, cmd=0x23, uid(8), start=0x00, count=0x0D payload = bytes([0x22, 0x23]) + b"\x01" * 8 + bytes([0x00, 0x0D]) result = decode_15693(0, payload) assert result == "READ MULTIPLE BLOCK #0+13" def test_reset_to_ready(self): payload = bytes([0x22, 0x26]) + b"\x01" * 8 result = decode_15693(0, payload) assert result == "RESET TO READY" def test_get_system_info(self): payload = bytes([0x22, 0x2B]) + b"\x01" * 8 result = decode_15693(0, payload) assert result == "GET SYSTEM INFO" def test_get_multiple_block_security(self): payload = bytes([0x22, 0x2C]) + b"\x01" * 8 + bytes([0x00, 0x3F]) result = decode_15693(0, payload) assert result == "GET MULTIPLE BLOCK SECURITY #0+63" def test_nxp_read_config(self): # flags=0x22, cmd=0xA1, mfg=0x04, reg=0x02 payload = bytes([0x22, 0xA1, 0x04, 0x02]) result = decode_15693(0, payload) assert result == "NXP READ CONFIG reg=2" def test_nxp_write_config(self): payload = bytes([0x22, 0xA2, 0x04, 0x03, 0xFF]) result = decode_15693(0, payload) assert result == "NXP WRITE CONFIG reg=3 val=0xFF" def test_nxp_get_random(self): payload = bytes([0x22, 0xB2, 0x04]) result = decode_15693(0, payload) assert result == "NXP GET RANDOM" def test_nxp_set_password(self): payload = bytes([0x22, 0xB3, 0x04, 0x01]) + b"\x00" * 4 result = decode_15693(0, payload) assert result == "NXP SET PASSWORD id=1" def test_unknown_command(self): payload = bytes([0x22, 0xFF]) result = decode_15693(0, payload) assert result is None def test_too_short(self): result = decode_15693(0, bytes([0x26])) assert result is None ``` **Step 2: Run tests to verify they fail** Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestDecode15693Request -v` Expected: FAIL — `ModuleNotFoundError: No module named 'pm3py.sim.trace_fmt'` **Step 3: Implement 15693 request decoder** Create `pm3py/sim/trace_fmt.py`: ```python """Trace formatter — colored, decoded, column-wrapped trace output.""" from __future__ import annotations # ---- ISO 15693 flags ---- _15693_FLAG_INVENTORY = 0x04 _15693_FLAG_ADDRESS = 0x20 # ---- ISO 15693 command names ---- _15693_CMDS = { 0x01: "INVENTORY", 0x02: "STAY QUIET", 0x20: "READ SINGLE BLOCK", 0x21: "WRITE SINGLE BLOCK", 0x23: "READ MULTIPLE BLOCK", 0x26: "RESET TO READY", 0x2B: "GET SYSTEM INFO", 0x2C: "GET MULTIPLE BLOCK SECURITY", } # NXP custom commands (manufacturer code 0x04) _15693_NXP_CMDS = { 0xA1: "NXP READ CONFIG", 0xA2: "NXP WRITE CONFIG", 0xB2: "NXP GET RANDOM", 0xB3: "NXP SET PASSWORD", } def _15693_block_offset(flags: int) -> int: """Return the byte offset of the block number field after flags+cmd.""" is_inventory = bool(flags & _15693_FLAG_INVENTORY) if not is_inventory and (flags & _15693_FLAG_ADDRESS): return 10 # flags(1) + cmd(1) + uid(8) return 2 # flags(1) + cmd(1) def decode_15693(direction: int, payload: bytes) -> str | None: """Decode an ISO 15693 frame into a human-readable annotation. Args: direction: 0 = reader->tag, 1 = tag->reader payload: raw frame bytes Returns: Annotation string or None if unrecognized. """ if direction == 0: return _decode_15693_request(payload) else: return _decode_15693_response(payload) def _decode_15693_request(payload: bytes) -> str | None: if len(payload) < 2: return None flags = payload[0] cmd = payload[1] # Standard commands name = _15693_CMDS.get(cmd) if name is not None: return _annotate_15693_request(name, cmd, flags, payload) # NXP custom commands (check manufacturer code) name = _15693_NXP_CMDS.get(cmd) if name is not None: return _annotate_nxp_request(name, cmd, payload) return None def _annotate_15693_request(name: str, cmd: int, flags: int, payload: bytes) -> str: blk_off = _15693_block_offset(flags) if cmd == 0x01: # INVENTORY if len(payload) > 2 and payload[2] > 0: return f"{name} mask={payload[2]}" return name if cmd in (0x20, 0x21): # READ/WRITE SINGLE BLOCK if len(payload) > blk_off: block = payload[blk_off] if cmd == 0x21: data_len = len(payload) - blk_off - 1 return f"{name} #{block} [{data_len}B]" return f"{name} #{block}" return name if cmd in (0x23, 0x2C): # READ MULTIPLE / GET MULTIPLE BLOCK SECURITY if len(payload) > blk_off + 1: start = payload[blk_off] count = payload[blk_off + 1] return f"{name} #{start}+{count}" return name return name def _annotate_nxp_request(name: str, cmd: int, payload: bytes) -> str: if cmd == 0xA1 and len(payload) >= 4: # READ CONFIG return f"{name} reg={payload[3]}" if cmd == 0xA2 and len(payload) >= 5: # WRITE CONFIG return f"{name} reg={payload[3]} val=0x{payload[4]:02X}" if cmd == 0xB3 and len(payload) >= 4: # SET PASSWORD return f"{name} id={payload[3]}" return name def _decode_15693_response(payload: bytes) -> str | None: # Placeholder — implemented in Task 2 return None ``` **Step 4: Run tests to verify they pass** Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestDecode15693Request -v` Expected: all PASS **Step 5: Commit** ```bash cd /home/work/pm3py/.worktrees/sim-framework git add pm3py/sim/trace_fmt.py tests/test_sim_trace_fmt.py git commit --no-gpg-sign -m "feat(trace): add ISO 15693 request decoder" ``` --- ### Task 2: ISO 15693 response decoder **Files:** - Modify: `pm3py/sim/trace_fmt.py` — replace `_decode_15693_response` placeholder - Test: `tests/test_sim_trace_fmt.py` **Step 1: Write failing tests** Append to `tests/test_sim_trace_fmt.py`: ```python class TestDecode15693Response: """Decode tag->reader (direction=1) 15693 responses.""" def test_ok_empty(self): # Just flags=0x00, no data (e.g., write ack) result = decode_15693(1, bytes([0x00])) assert result == "OK" def test_inventory_response(self): # flags=0x00, dsfid=0x00, uid (8 bytes LSB-first) uid_lsb = bytes([0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x04, 0xE0]) payload = bytes([0x00, 0x00]) + uid_lsb result = decode_15693(1, payload) assert result == "OK INVENTORY UID=E004010101010101" def test_error_response(self): payload = bytes([0x01, 0x0F]) result = decode_15693(1, payload) assert result == "ERROR 0x0F" def test_error_block_not_available(self): payload = bytes([0x01, 0x10]) result = decode_15693(1, payload) assert result == "ERROR 0x10 block not available" def test_data_response(self): # flags=0x00 + 4 data bytes (read single block response) payload = bytes([0x00, 0xDE, 0xAD, 0xBE, 0xEF]) result = decode_15693(1, payload) assert result == "OK [4B]" def test_too_short(self): result = decode_15693(1, b"") assert result is None ``` **Step 2: Run to verify failure** Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestDecode15693Response -v` Expected: FAIL — responses decode as `None` **Step 3: Implement response decoder** Replace `_decode_15693_response` in `trace_fmt.py`: ```python _15693_ERRORS = { 0x01: "not supported", 0x02: "not recognized", 0x0F: "unknown error", 0x10: "block not available", 0x11: "block already locked", 0x12: "block locked", 0x13: "block not written", 0x14: "block not locked", } def _decode_15693_response(payload: bytes) -> str | None: if len(payload) < 1: return None flags = payload[0] if flags & 0x01: # Error if len(payload) >= 2: code = payload[1] desc = _15693_ERRORS.get(code, "") if desc: return f"ERROR 0x{code:02X} {desc}" return f"ERROR 0x{code:02X}" return "ERROR" # Success — try to identify the response type data = payload[1:] if len(data) == 0: return "OK" # Inventory response: dsfid(1) + uid(8) = 9 bytes if len(data) == 9: uid_msb = bytes(reversed(data[1:9])) return f"OK INVENTORY UID={uid_msb.hex().upper()}" # Generic data response return f"OK [{len(data)}B]" ``` **Step 4: Run tests** Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestDecode15693Response -v` Expected: all PASS **Step 5: Commit** ```bash cd /home/work/pm3py/.worktrees/sim-framework git add pm3py/sim/trace_fmt.py tests/test_sim_trace_fmt.py git commit --no-gpg-sign -m "feat(trace): add ISO 15693 response decoder" ``` --- ### Task 3: ISO 14443-A decoder **Files:** - Modify: `pm3py/sim/trace_fmt.py` - Test: `tests/test_sim_trace_fmt.py` **Step 1: Write failing tests** Append to test file: ```python from pm3py.sim.trace_fmt import decode_14443a class TestDecode14443aRequest: """Decode reader->tag 14443-A commands.""" def test_reqa(self): assert decode_14443a(0, bytes([0x26])) == "REQA" def test_wupa(self): assert decode_14443a(0, bytes([0x52])) == "WUPA" def test_hlta(self): assert decode_14443a(0, bytes([0x50, 0x00])) == "HLTA" def test_anticol_cl1(self): assert decode_14443a(0, bytes([0x93, 0x20])) == "ANTICOL CL1" def test_select_cl1(self): payload = bytes([0x93, 0x70]) + b"\x01\x02\x03\x04\x04" assert decode_14443a(0, payload) == "SELECT CL1" def test_anticol_cl2(self): assert decode_14443a(0, bytes([0x95, 0x20])) == "ANTICOL CL2" def test_select_cl2(self): payload = bytes([0x95, 0x70]) + b"\x01\x02\x03\x04\x04" assert decode_14443a(0, payload) == "SELECT CL2" def test_anticol_cl3(self): assert decode_14443a(0, bytes([0x97, 0x20])) == "ANTICOL CL3" def test_select_cl3(self): payload = bytes([0x97, 0x70]) + b"\x01\x02\x03\x04\x04" assert decode_14443a(0, payload) == "SELECT CL3" def test_rats(self): assert decode_14443a(0, bytes([0xE0, 0x50])) == "RATS" def test_iblock_even(self): assert decode_14443a(0, bytes([0x02, 0x00, 0xA4])) == "I-BLOCK(0)" def test_iblock_odd(self): assert decode_14443a(0, bytes([0x03, 0x00, 0xA4])) == "I-BLOCK(1)" def test_rack(self): assert decode_14443a(0, bytes([0xA2])) == "R-ACK(0)" assert decode_14443a(0, bytes([0xA3])) == "R-ACK(1)" def test_rnak(self): assert decode_14443a(0, bytes([0xB2])) == "R-NAK(0)" assert decode_14443a(0, bytes([0xB3])) == "R-NAK(1)" def test_deselect(self): assert decode_14443a(0, bytes([0xC2])) == "S(DESELECT)" def test_wtx(self): assert decode_14443a(0, bytes([0xF2, 0x01])) == "S(WTX)" def test_unknown(self): assert decode_14443a(0, bytes([0xFF])) is None def test_empty(self): assert decode_14443a(0, b"") is None class TestDecode14443aResponse: """Decode tag->reader 14443-A responses.""" def test_atqa(self): # 2-byte response to REQA/WUPA assert decode_14443a(1, bytes([0x04, 0x00])) == "ATQA 04 00" def test_sak(self): # 1-byte response to SELECT assert decode_14443a(1, bytes([0x20])) == "SAK 20" def test_ats(self): # ATS: first byte is length ats = bytes([0x05, 0x78, 0x80, 0x70, 0x02]) assert decode_14443a(1, ats) == "ATS [5]" def test_iblock_response(self): assert decode_14443a(1, bytes([0x02, 0x90, 0x00])) == "I-BLOCK(0)" def test_empty(self): assert decode_14443a(1, b"") is None ``` **Step 2: Run to verify failure** Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestDecode14443aRequest tests/test_sim_trace_fmt.py::TestDecode14443aResponse -v` Expected: FAIL — `ImportError: cannot import name 'decode_14443a'` **Step 3: Implement 14443-A decoder** Add to `trace_fmt.py`: ```python # ---- ISO 14443-A constants ---- _14A_CL_MAP = {0x93: "CL1", 0x95: "CL2", 0x97: "CL3"} def decode_14443a(direction: int, payload: bytes) -> str | None: """Decode an ISO 14443-A frame into a human-readable annotation.""" if not payload: return None if direction == 0: return _decode_14443a_request(payload) else: return _decode_14443a_response(payload) def _decode_14443a_request(payload: bytes) -> str | None: b0 = payload[0] # Short frames (single byte) if b0 == 0x26: return "REQA" if b0 == 0x52: return "WUPA" # HLTA if b0 == 0x50 and len(payload) >= 2: return "HLTA" # Anticollision / Select if b0 in _14A_CL_MAP and len(payload) >= 2: cl = _14A_CL_MAP[b0] nvb = payload[1] if nvb == 0x20: return f"ANTICOL {cl}" if nvb == 0x70: return f"SELECT {cl}" return f"ANTICOL {cl} nvb={nvb:02X}" # RATS if b0 == 0xE0: return "RATS" # ISO-DEP I-block if b0 & 0xE2 == 0x02: bn = b0 & 0x01 return f"I-BLOCK({bn})" # R-ACK if b0 & 0xF6 == 0xA2: bn = b0 & 0x01 return f"R-ACK({bn})" # R-NAK if b0 & 0xF6 == 0xB2: bn = b0 & 0x01 return f"R-NAK({bn})" # S(DESELECT) if b0 == 0xC2: return "S(DESELECT)" # S(WTX) if b0 == 0xF2: return "S(WTX)" return None def _decode_14443a_response(payload: bytes) -> str | None: if not payload: return None b0 = payload[0] # I-block response if b0 & 0xE2 == 0x02: bn = b0 & 0x01 return f"I-BLOCK({bn})" # ATQA (2 bytes) if len(payload) == 2 and b0 & 0xF0 == 0x00: return f"ATQA {payload[0]:02X} {payload[1]:02X}" # SAK (1 byte) if len(payload) == 1: return f"SAK {b0:02X}" # ATS (first byte = length, length >= 2) if len(payload) >= 2 and payload[0] == len(payload): return f"ATS [{len(payload)}]" return None ``` **Step 4: Run tests** Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py -v` Expected: all PASS (15693 + 14443-A) **Step 5: Commit** ```bash cd /home/work/pm3py/.worktrees/sim-framework git add pm3py/sim/trace_fmt.py tests/test_sim_trace_fmt.py git commit --no-gpg-sign -m "feat(trace): add ISO 14443-A decoder" ``` --- ### Task 4: TraceFormatter — color, wrapping, terminal width **Files:** - Modify: `pm3py/sim/trace_fmt.py` - Test: `tests/test_sim_trace_fmt.py` **Step 1: Write failing tests** Append to test file: ```python from pm3py.sim.trace_fmt import TraceFormatter class TestTraceFormatterBasic: """Test formatting output (colors stripped for assertion).""" def _strip_ansi(self, s: str) -> str: import re return re.sub(r'\033\[[0-9;]*m', '', s) def test_starts_with_newline(self): fmt = TraceFormatter(mode="sim") result = fmt.format(0, bytes([0x26, 0x01, 0x00])) assert result.startswith("\n") def test_sim_mode_tag(self): fmt = TraceFormatter(mode="sim") result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00]))) assert "[Sim]" in result def test_reader_mode_tag(self): fmt = TraceFormatter(mode="reader") result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00]))) assert "[Rdr]" in result def test_sniff_mode_tag(self): fmt = TraceFormatter(mode="sniff") result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00]))) assert "[Snf]" in result def test_reader_to_tag_arrow(self): fmt = TraceFormatter(mode="sim") result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00]))) assert "Reader \u2192 Tag:" in result def test_tag_to_reader_arrow(self): fmt = TraceFormatter(mode="sim") result = self._strip_ansi(fmt.format(1, bytes([0x00]))) assert "Tag \u2192 Reader:" in result def test_hex_space_separated(self): fmt = TraceFormatter(mode="sim") result = self._strip_ansi(fmt.format(0, bytes([0x22, 0x20, 0x03]))) assert "22 20 03" in result def test_annotation_present(self): fmt = TraceFormatter(mode="sim") result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00]))) assert "INVENTORY" in result def test_no_annotation_for_unknown(self): fmt = TraceFormatter(mode="sim") result = self._strip_ansi(fmt.format(0, bytes([0x22, 0xFF]))) assert "22 FF" in result or "22 ff" in result.lower() class TestTraceFormatterWrapping: """Test column-aligned wrapping for long payloads.""" def _strip_ansi(self, s: str) -> str: import re return re.sub(r'\033\[[0-9;]*m', '', s) def test_short_payload_inline_annotation(self): """Short payload: annotation on same line as hex.""" fmt = TraceFormatter(mode="sim", width=80) result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00]))) lines = result.strip().split("\n") assert len(lines) == 1 assert "INVENTORY" in lines[0] assert "26 01 00" in lines[0] def test_long_payload_wraps(self): """Long payload wraps at prefix column.""" fmt = TraceFormatter(mode="sim", width=60) long_payload = bytes([0x00]) + bytes(40) result = self._strip_ansi(fmt.format(1, long_payload)) lines = result.strip().split("\n") assert len(lines) > 1 # Continuation lines should be indented to prefix column first_hex_col = lines[0].index("00") for line in lines[1:]: if line.strip(): leading = len(line) - len(line.lstrip()) assert leading >= first_hex_col - 1 def test_wrapped_annotation_on_own_line(self): """When hex wraps, annotation goes on its own line.""" fmt = TraceFormatter(mode="sim", width=50) # Inventory response with UID — long enough to wrap at width=50 uid_lsb = bytes([0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x04, 0xE0]) payload = bytes([0x00, 0x00]) + uid_lsb result = self._strip_ansi(fmt.format(1, payload)) lines = result.strip().split("\n") # Annotation should be on a separate line annotation_lines = [l for l in lines if "INVENTORY" in l] hex_lines = [l for l in lines if "00 01" in l.lower() or "04 e0" in l.lower()] if len(hex_lines) > 1: # Wrapped — annotation must be on its own line assert len(annotation_lines) == 1 assert annotation_lines[0] not in hex_lines class TestTraceFormatterNoColor: """Colors suppressed when is_tty=False.""" def test_no_ansi_when_not_tty(self): fmt = TraceFormatter(mode="sim", is_tty=False) result = fmt.format(0, bytes([0x26, 0x01, 0x00])) assert "\033[" not in result ``` **Step 2: Run to verify failure** Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestTraceFormatterBasic tests/test_sim_trace_fmt.py::TestTraceFormatterWrapping tests/test_sim_trace_fmt.py::TestTraceFormatterNoColor -v` Expected: FAIL — `ImportError: cannot import name 'TraceFormatter'` **Step 3: Implement TraceFormatter** Add to `trace_fmt.py`: ```python import os import signal import sys # ---- ANSI colors ---- _C_CYAN = "\033[36m" _C_YELLOW = "\033[33m" _C_MAGENTA = "\033[35m" _C_GREEN = "\033[32m" _C_DIM = "\033[2m" _C_RESET = "\033[0m" _MODE_TAGS = { "sim": ("[Sim]", _C_MAGENTA), "reader": ("[Rdr]", _C_CYAN), "sniff": ("[Snf]", ""), } class TraceFormatter: """Colored, decoded, column-wrapped trace output. Args: mode: "sim", "reader", or "sniff" protocol: "15693" or "14443a" (selects decoder) width: override terminal width (None = auto-detect) is_tty: override TTY detection (None = auto-detect) """ def __init__(self, mode: str = "sim", protocol: str = "15693", width: int | None = None, is_tty: bool | None = None): self._mode = mode self._protocol = protocol self._width = width or self._detect_width() self._is_tty = is_tty if is_tty is not None else sys.stdout.isatty() self._line_count = 0 # Try to register SIGWINCH for dynamic resize if width is None: try: signal.signal(signal.SIGWINCH, self._on_resize) except (OSError, ValueError): pass # not main thread or not Unix def _detect_width(self) -> int: try: return os.get_terminal_size().columns except (OSError, ValueError): return 80 def _on_resize(self, signum, frame): self._width = self._detect_width() def _color(self, code: str, text: str) -> str: if not self._is_tty or not code: return text return f"{code}{text}{_C_RESET}" def format(self, direction: int, payload: bytes) -> str: """Format a trace line with colors, decoding, and wrapping.""" # Re-check width periodically if no SIGWINCH self._line_count += 1 if self._line_count % 10 == 0: try: self._width = self._detect_width() except Exception: pass # Mode tag mode_tag, mode_color = _MODE_TAGS.get(self._mode, ("[???]", "")) # Direction if direction == 0: arrow = "Reader \u2192 Tag:" dir_color = _C_CYAN else: arrow = "Tag \u2192 Reader:" dir_color = _C_YELLOW # Build prefix (uncolored for width calc) prefix_plain = f"{mode_tag} {arrow} " prefix_len = len(prefix_plain) # Colored prefix prefix_colored = self._color(mode_color, mode_tag) + " " + self._color(dir_color, arrow) + " " # Hex bytes (space-separated, uppercase) hex_str = " ".join(f"{b:02X}" for b in payload) # Decode annotation if self._protocol == "15693": annotation = decode_15693(direction, payload) elif self._protocol == "14443a": annotation = decode_14443a(direction, payload) else: annotation = None # Layout: determine if everything fits on one line avail = self._width - prefix_len if annotation: one_line = f"{hex_str} {annotation}" else: one_line = hex_str if len(one_line) <= avail: # Single line hex_colored = self._color(_C_DIM, hex_str) if annotation: ann_colored = self._color(_C_GREEN, annotation) line = f"{prefix_colored}{hex_colored} {ann_colored}" else: line = f"{prefix_colored}{hex_colored}" return f"\n{line}" # Multi-line: wrap hex, annotation on its own line pad = " " * prefix_len hex_lines = self._wrap_hex(hex_str, avail) parts = [f"{prefix_colored}{self._color(_C_DIM, hex_lines[0])}"] for hl in hex_lines[1:]: parts.append(f"{pad}{self._color(_C_DIM, hl)}") if annotation: parts.append(f"{pad}{self._color(_C_GREEN, annotation)}") return "\n" + "\n".join(parts) def _wrap_hex(self, hex_str: str, avail: int) -> list[str]: """Wrap space-separated hex string into lines of at most `avail` chars.""" tokens = hex_str.split(" ") lines = [] current = "" for tok in tokens: candidate = f"{current} {tok}" if current else tok if len(candidate) <= avail: current = candidate else: if current: lines.append(current) current = tok if current: lines.append(current) return lines or [""] def print(self, direction: int, payload: bytes) -> None: """Format and print a trace line to stdout.""" sys.stdout.write(self.format(direction, payload)) sys.stdout.flush() ``` **Step 4: Run tests** Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py -v` Expected: all PASS **Step 5: Commit** ```bash cd /home/work/pm3py/.worktrees/sim-framework git add pm3py/sim/trace_fmt.py tests/test_sim_trace_fmt.py git commit --no-gpg-sign -m "feat(trace): add TraceFormatter with color, wrapping, terminal resize" ``` --- ### Task 5: Wire TraceFormatter into SimSession **Files:** - Modify: `pm3py/sim/sim_session.py:1-2,126-175,177-224` **Step 1: Write failing test** Append to `tests/test_sim_trace_fmt.py`: ```python from unittest.mock import MagicMock, patch import struct from pm3py.sim.sim_session import SimSession, CMD_HF_ISO15693_SIM_TRACE from pm3py.transport import encode_ng_frame, RESP_PREAMBLE_MAGIC, RESP_PREAMBLE_SIZE, RESP_POSTAMBLE_SIZE class TestSimSessionTrace: """Verify SimSession uses TraceFormatter instead of raw print.""" def _make_trace_frame(self, direction: int, payload: bytes) -> bytes: """Build a fake firmware trace response frame.""" from pm3py.transport import RESP_POSTAMBLE_MAGIC data = bytes([direction]) + payload length = len(data) | 0x8000 # NG bit set header = struct.pack("tag inventory frame = self._make_trace_frame(0, bytes([0x26, 0x01, 0x00])) # Mock serial that returns frame then stops mock_serial = MagicMock() mock_serial.in_waiting = len(frame) call_count = 0 def read_side_effect(n): nonlocal call_count call_count += 1 if call_count == 1: return frame session._active = False return b"" mock_serial.read.side_effect = read_side_effect session._trace_reader(mock_serial) mock_fmt.print.assert_called_once_with(0, bytes([0x26, 0x01, 0x00])) ``` **Step 2: Run to verify failure** Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py::TestSimSessionTrace -v` Expected: FAIL — `ImportError` or `AttributeError` (no `TraceFormatter` import in sim_session) **Step 3: Modify sim_session.py** Add import at top of `sim_session.py`: ```python from .trace_fmt import TraceFormatter ``` In `start_15693()`, replace the trace thread creation (around line 167-171) to instantiate the formatter: ```python if trace: self._formatter = TraceFormatter(mode="sim", protocol="15693") self._trace_thread = threading.Thread( target=self._trace_reader, args=(ser,), daemon=True ) self._trace_thread.start() ``` In `_trace_reader()`, replace line 216-217: ```python arrow = "Reader → Tag" if direction == 0 else "Tag → Reader" print(f"[Trace] {arrow}: {payload.hex()}") ``` With: ```python self._formatter.print(direction, payload) ``` **Step 4: Run all tests** Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/test_sim_trace_fmt.py -v` Expected: all PASS Run: `cd /home/work/pm3py/.worktrees/sim-framework && python -m pytest tests/ -v` Expected: all PASS (no regressions) **Step 5: Commit** ```bash cd /home/work/pm3py/.worktrees/sim-framework git add pm3py/sim/sim_session.py pm3py/sim/trace_fmt.py tests/test_sim_trace_fmt.py git commit --no-gpg-sign -m "feat(trace): wire TraceFormatter into SimSession" ```