pm3py/docs/plans/2026-03-17-trace-formatter-plan.md

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

"""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:

"""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

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:

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:

_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

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:

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:

# ---- 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

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:

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:

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

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:

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("<IHHI",
                             RESP_PREAMBLE_MAGIC,
                             length,
                             0,  # status
                             CMD_HF_ISO15693_SIM_TRACE)
        postamble = struct.pack("<H", RESP_POSTAMBLE_MAGIC)
        return header + data + postamble

    @patch("pm3py.sim.sim_session.TraceFormatter")
    def test_trace_reader_uses_formatter(self, MockFormatter):
        """_trace_reader should call formatter.print() not raw print()."""
        mock_fmt = MockFormatter.return_value
        session = SimSession()
        session._formatter = mock_fmt
        session._active = True

        # Build a trace frame for: reader->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:

from .trace_fmt import TraceFormatter

In start_15693(), replace the trace thread creation (around line 167-171) to instantiate the formatter:

        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:

                                arrow = "Reader → Tag" if direction == 0 else "Tag → Reader"
                                print(f"[Trace] {arrow}: {payload.hex()}")

With:

                                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

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"