WIP: sniff infrastructure, transport additions, test coverage

In-progress work — committing before refactor merge so git rename
detection can carry changes to new file locations.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

.gitignore

@@ -4,3 +4,7 @@ __pycache__/
 .pytest_cache/
 venv/
 .worktrees/
+*.pdf
+
+# Capture data (sniffs, sims, taginfo scans)
+data/

pm3py/hf_15.py

@@ -1,6 +1,7 @@
 """ISO 15693 commands: hf.15.*"""
 import struct
-from .protocol import Cmd
+import sys
+from .protocol import Cmd, PM3_CMD_DATA_SIZE
 from .transport import PM3Transport, PM3Error
 
 # ISO15 PM3 flags
@@ -10,6 +11,461 @@ ISO15_RAW = 0x04
 ISO15_APPEND_CRC = 0x08
 ISO15_READ_RESPONSE = 0x10
 
+# tracelog_hdr_t size: timestamp(4) + duration(2) + data_len_flags(2) = 8
+TRACELOG_HDR_SIZE = 8
+
+
+def parse_tracelog(raw: bytes) -> list[dict]:
+    """Parse PM3 trace buffer into a list of trace entries.
+
+    Each entry: {timestamp, duration, is_response, data, parity}
+    """
+    entries = []
+    offset = 0
+    while offset + TRACELOG_HDR_SIZE <= len(raw):
+        timestamp, duration, data_len_flags = struct.unpack_from("<IHH", raw, offset)
+        is_response = bool(data_len_flags & 0x8000)
+        data_len = data_len_flags & 0x7FFF
+        offset += TRACELOG_HDR_SIZE
+
+        if data_len == 0:
+            break  # end of trace / sentinel
+
+        if offset + data_len > len(raw):
+            break
+
+        data = raw[offset:offset + data_len]
+        offset += data_len
+
+        parity_len = (data_len + 7) // 8 if data_len > 0 else 0
+        parity = raw[offset:offset + parity_len] if parity_len else b""
+        offset += parity_len
+
+        entries.append({
+            "timestamp": timestamp,
+            "duration": duration,
+            "is_response": is_response,
+            "data": data,
+            "parity": parity,
+        })
+    return entries
+
+
+# ---- NDEF decode (for trace annotation) ----
+
+_NDEF_MAGIC = 0xE1
+
+_URI_PREFIXES = {
+    0x00: "", 0x01: "http://www.", 0x02: "https://www.",
+    0x03: "http://", 0x04: "https://", 0x05: "tel:", 0x06: "mailto:",
+}
+
+_TNF_EMPTY = 0x00
+_TNF_WELL_KNOWN = 0x01
+_TNF_MEDIA = 0x02
+_TNF_URI = 0x03
+_TNF_EXTERNAL = 0x04
+_MAX_ANN_TEXT = 40
+
+
+def decode_ndef_annotation(data: bytes) -> str | None:
+    """Decode NDEF TLV data into a human-readable annotation string.
+
+    Handles CC (capability container) blocks and NDEF message TLVs.
+    Returns None if the data doesn't contain recognizable NDEF content.
+    """
+    if len(data) < 3:
+        return None
+
+    parts = []
+    offset = 0
+
+    if data[0] == _NDEF_MAGIC and len(data) >= 4:
+        ver_major = (data[1] >> 6) & 0x03
+        ver_minor = (data[1] >> 4) & 0x03
+        mlen = data[2]
+        features = data[3]
+        cc_str = f"CC v{ver_major}.{ver_minor} MLEN={mlen}"
+        if features & 0x01:
+            cc_str += " MBREAD"
+        parts.append(cc_str)
+        offset = 4
+        if offset >= len(data):
+            return cc_str
+
+    while offset < len(data):
+        tlv_type = data[offset]
+        offset += 1
+        if tlv_type == 0x00:
+            continue
+        if tlv_type == 0xFE:
+            break
+        if tlv_type != 0x03:
+            break
+
+        if offset >= len(data):
+            break
+        tlv_len = data[offset]
+        offset += 1
+        if tlv_len == 0xFF:
+            if offset + 2 > len(data):
+                break
+            tlv_len = (data[offset] << 8) | data[offset + 1]
+            offset += 2
+
+        if tlv_len == 0:
+            parts.append("NDEF empty")
+            continue
+
+        msg_end = min(offset + tlv_len, len(data))
+        records = _parse_ndef_records(data[offset:msg_end])
+        for rec in records:
+            parts.append(rec)
+        offset = msg_end
+
+    return " | ".join(parts) if parts else None
+
+
+def _parse_ndef_records(msg: bytes) -> list[str]:
+    """Parse NDEF records from a message payload."""
+    records = []
+    offset = 0
+    while offset < len(msg):
+        if offset + 3 > len(msg):
+            break
+        flags = msg[offset]
+        tnf = flags & 0x07
+        sr = bool(flags & 0x10)
+        il = bool(flags & 0x08)
+        type_len = msg[offset + 1]
+        offset += 2
+
+        if sr:
+            if offset >= len(msg):
+                break
+            payload_len = msg[offset]
+            offset += 1
+        else:
+            if offset + 4 > len(msg):
+                break
+            payload_len = int.from_bytes(msg[offset:offset + 4], "big")
+            offset += 4
+
+        id_len = 0
+        if il:
+            if offset >= len(msg):
+                break
+            id_len = msg[offset]
+            offset += 1
+
+        if offset + type_len > len(msg):
+            break
+        rec_type = msg[offset:offset + type_len]
+        offset += type_len
+        offset += id_len
+
+        if offset + payload_len > len(msg):
+            payload = msg[offset:]
+        else:
+            payload = msg[offset:offset + payload_len]
+        offset += payload_len
+
+        records.append(_decode_ndef_record(tnf, rec_type, payload))
+    return records
+
+
+def _decode_ndef_record(tnf: int, rec_type: bytes, payload: bytes) -> str:
+    """Decode a single NDEF record into an annotation string."""
+    if tnf == _TNF_WELL_KNOWN:
+        if rec_type == b"T" and len(payload) >= 1:
+            lang_len = payload[0] & 0x3F
+            lang = payload[1:1 + lang_len].decode("ascii", errors="replace")
+            text = payload[1 + lang_len:].decode("utf-8", errors="replace")
+            if len(text) > _MAX_ANN_TEXT:
+                text = text[:_MAX_ANN_TEXT] + "..."
+            return f'NDEF Text "{lang}" "{text}"'
+        if rec_type == b"U" and len(payload) >= 1:
+            prefix = _URI_PREFIXES.get(payload[0], "")
+            suffix = payload[1:].decode("utf-8", errors="replace")
+            uri = prefix + suffix
+            if len(uri) > _MAX_ANN_TEXT:
+                uri = uri[:_MAX_ANN_TEXT] + "..."
+            return f"NDEF URI {uri}"
+        return f"NDEF WK type={rec_type!r} [{len(payload)}B]"
+    if tnf == _TNF_MEDIA:
+        mime = rec_type.decode("ascii", errors="replace")
+        if len(mime) > 30:
+            mime = mime[:30] + "..."
+        return f"NDEF MIME {mime} [{len(payload)}B]"
+    if tnf == _TNF_EXTERNAL:
+        ext_type = rec_type.decode("ascii", errors="replace")
+        return f"NDEF EXT {ext_type} [{len(payload)}B]"
+    if tnf == _TNF_EMPTY:
+        return "NDEF empty"
+    return f"NDEF TNF={tnf} [{len(payload)}B]"
+
+
+# ---- ISO 15693 command names (for sniff annotation) ----
+_15693_CMDS = {
+    0x01: "INVENTORY",
+    0x02: "STAY QUIET",
+    0x20: "READ SINGLE BLOCK",
+    0x21: "WRITE SINGLE BLOCK",
+    0x22: "LOCK BLOCK",
+    0x23: "READ MULTIPLE BLOCKS",
+    0x24: "WRITE MULTIPLE BLOCKS",
+    0x25: "SELECT",
+    0x26: "RESET TO READY",
+    0x27: "WRITE AFI",
+    0x28: "LOCK AFI",
+    0x29: "WRITE DSFID",
+    0x2A: "LOCK DSFID",
+    0x2B: "GET SYSTEM INFO",
+    0x2C: "GET MULTIPLE BLOCK SECURITY",
+}
+
+_15693_NXP_CMDS = {
+    0xA0: "NXP INVENTORY READ",
+    0xA1: "NXP FAST INVENTORY READ",
+    0xA2: "NXP SET EAS",
+    0xA3: "NXP RESET EAS",
+    0xA4: "NXP LOCK EAS",
+    0xA5: "NXP EAS ALARM",
+    0xA6: "NXP PASSWORD PROTECT EAS/AFI",
+    0xA7: "NXP WRITE EAS ID",
+    0xAB: "NXP GET SYSTEM INFO",
+    0xB2: "NXP GET RANDOM",
+    0xB3: "NXP SET PASSWORD",
+    0xB4: "NXP WRITE PASSWORD",
+    0xB5: "NXP LOCK PASSWORD",
+    0xB6: "NXP PROTECT PAGE",
+    0xB7: "NXP LOCK PAGE PROTECTION",
+    0xB9: "NXP DESTROY",
+    0xBA: "NXP ENABLE PRIVACY",
+    0xBB: "NXP 64-BIT PASSWORD PROTECTION",
+    0xBD: "NXP READ SIGNATURE",
+}
+
+_15693_ERRORS = {
+    0x01: "not supported",
+    0x02: "not recognized",
+    0x10: "block not available",
+    0x11: "block already locked",
+    0x12: "block locked",
+    0x13: "block not written",
+    0x14: "block not locked",
+}
+
+_FLAG_INVENTORY = 0x04
+_FLAG_ADDRESS = 0x20
+
+
+def _block_offset(flags: int) -> int:
+    """Byte offset of block number after flags+cmd."""
+    if not (flags & _FLAG_INVENTORY) and (flags & _FLAG_ADDRESS):
+        return 10  # flags(1) + cmd(1) + uid(8)
+    return 2  # flags(1) + cmd(1)
+
+
+def decode_15693_request(payload: bytes) -> str | None:
+    """Decode an ISO 15693 request frame into annotation."""
+    if len(payload) < 2:
+        return None
+
+    flags = payload[0]
+    cmd = payload[1]
+
+    name = _15693_CMDS.get(cmd) or _15693_NXP_CMDS.get(cmd)
+    if name is None:
+        return f"UNKNOWN CMD 0x{cmd:02X}"
+
+    blk_off = _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
+        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
+
+    if cmd == 0x2B:  # GET SYSTEM INFO
+        return name
+
+    if cmd == 0xB3 and len(payload) >= 4:  # NXP SET PASSWORD
+        return f"{name} id={payload[3]}"
+
+    return name
+
+
+def decode_15693_response(payload: bytes) -> str | None:
+    """Decode an ISO 15693 response frame into annotation."""
+    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, "")
+            return f"ERROR 0x{code:02X} {desc}" if desc else f"ERROR 0x{code:02X}"
+        return "ERROR"
+
+    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()}"
+
+    # Try NDEF decode on response data
+    ndef_ann = decode_ndef_annotation(data)
+    if ndef_ann:
+        return f"OK {ndef_ann}"
+
+    return f"OK [{len(data)}B]"
+
+
+def decode_15693(direction: int, payload: bytes) -> str | None:
+    """Decode an ISO 15693 frame for sniff trace."""
+    if direction == 0:
+        return decode_15693_request(payload)
+    else:
+        return decode_15693_response(payload)
+
+
+# ---- ANSI colors ----
+_C_CYAN = "\033[36m"
+_C_YELLOW = "\033[33m"
+_C_DIM = "\033[2m"
+_C_CRC = "\033[1;37m"
+_C_RED = "\033[31m"
+_C_RESET = "\033[0m"
+
+
+def _color(code: str, text: str, is_tty: bool = True) -> str:
+    if not is_tty or not code:
+        return text
+    return f"{code}{text}{_C_RESET}"
+
+
+def _wrap_annotation(annotation: str, avail: int) -> list[str]:
+    if len(annotation) <= avail:
+        return [annotation]
+    if " | " in annotation:
+        parts = annotation.split(" | ")
+        lines = []
+        current = parts[0]
+        for part in parts[1:]:
+            candidate = f"{current} | {part}"
+            if len(candidate) <= avail:
+                current = candidate
+            else:
+                lines.append(current)
+                current = part
+        lines.append(current)
+        return lines
+    return [annotation[:avail - 3] + "..."]
+
+
+def format_sniff_line(entry: dict, width: int = 120, is_tty: bool = True) -> str:
+    """Format a single tracelog entry as a colored sniff line.
+
+    Uses cyan for reader→tag, yellow for tag→reader, both hex and annotations.
+    """
+    direction = 1 if entry["is_response"] else 0
+    payload = entry["data"]
+
+    if direction == 0:
+        arrow = "Reader → Tag:"
+        dir_color = _C_CYAN
+    else:
+        arrow = "Tag → Reader:"
+        dir_color = _C_YELLOW
+
+    prefix_plain = f"[Snf] {arrow}  "
+    prefix_len = len(prefix_plain)
+    prefix_colored = _color(_C_DIM, "[Snf]", is_tty) + " " + _color(dir_color, arrow, is_tty) + "  "
+
+    # Split payload into data + CRC (last 2 bytes)
+    if len(payload) > 2:
+        data = payload[:-2]
+        crc = payload[-2:]
+    else:
+        data = payload
+        crc = b""
+
+    data_hex = " ".join(f"{b:02X}" for b in data)
+    crc_hex = " ".join(f"{b:02X}" for b in crc)
+    full_hex = f"{data_hex} {crc_hex}" if crc_hex else data_hex
+
+    # Decode annotation
+    annotation = decode_15693(direction, data)
+    ann_color = dir_color
+
+    avail = width - prefix_len
+    if annotation:
+        one_line = f"{full_hex}  {annotation}"
+    else:
+        one_line = full_hex
+
+    if len(one_line) <= avail:
+        hex_colored = _color(_C_DIM, data_hex, is_tty)
+        if crc_hex:
+            hex_colored += " " + _color(_C_CRC, crc_hex, is_tty)
+        if annotation:
+            ann_colored = _color(ann_color, annotation, is_tty)
+            gap = max(avail - len(full_hex) - len(annotation), 2)
+            return f"{prefix_colored}{hex_colored}{' ' * gap}{ann_colored}"
+        return f"{prefix_colored}{hex_colored}"
+
+    # Multi-line: wrap hex
+    lines = []
+    tokens = full_hex.split(" ")
+    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)
+
+    pad = " " * prefix_len
+    parts = []
+    for i, hl in enumerate(lines):
+        colored_hl = _color(_C_DIM, hl, is_tty)
+        if i == 0:
+            parts.append(f"{prefix_colored}{colored_hl}")
+        else:
+            parts.append(f"{pad}{colored_hl}")
+    if annotation:
+        ann_lines = _wrap_annotation(annotation, avail)
+        for al in ann_lines:
+            ann_pad = max(avail - len(al), 0)
+            parts.append(f"{pad}{' ' * ann_pad}{_color(ann_color, al, is_tty)}")
+
+    return "\n".join(parts)
+
 
 class HF15Commands:
     """ISO 15693 commands."""
@@ -87,7 +543,73 @@ class HF15Commands:
         success = resp.status == 0 and len(resp.data) > 0 and resp.data[0] == 0
         return {"success": success, "block": block}
 
-    async def sniff(self) -> dict:
+    async def sniff(self, timeout: float = 60.0) -> dict:
-        """Sniff ISO15693 traffic."""
+        """Start ISO15693 sniff. Blocks until button press or timeout.
-        resp = await self._t.send_ng(Cmd.HF_ISO15693_SNIFF, timeout=30.0)
+
-        return {"status": resp.status}
+        Returns {"status": int}.
+        """
+        from .transport import encode_ng_frame
+        async with self._t._lock:
+            raw = encode_ng_frame(Cmd.HF_ISO15693_SNIFF)
+            await self._t.send_frame(raw)
+
+            # Firmware sends debug prints before the sniff completion response.
+            # Keep reading until we get the actual sniff response.
+            while True:
+                resp = await self._t.read_response(timeout=timeout)
+                if resp.cmd == Cmd.HF_ISO15693_SNIFF:
+                    return {"status": resp.status}
+                # Debug print or other message — discard and keep waiting
+
+    async def download_trace(self) -> list[dict]:
+        """Download and parse the trace buffer from the device.
+
+        Returns list of trace entries, each with:
+            timestamp, duration, is_response, data, parity
+        """
+        # First download to discover trace length
+        raw, trace_len = await self._t.download_bigbuf(
+            offset=0, length=PM3_CMD_DATA_SIZE, timeout=4.0)
+
+        if trace_len == 0:
+            return []
+
+        # If trace is larger than first chunk, download the full thing
+        if trace_len > PM3_CMD_DATA_SIZE:
+            raw, trace_len = await self._t.download_bigbuf(
+                offset=0, length=trace_len, timeout=10.0)
+
+        return parse_tracelog(raw[:trace_len])
+
+    async def sniff_decoded(self, timeout: float = 60.0) -> list[dict]:
+        """Sniff ISO15693 traffic, download trace, decode and print.
+
+        Blocks until button press or timeout. Then downloads the trace,
+        decodes each frame, and prints formatted output.
+
+        Returns the parsed trace entries.
+        """
+        import os
+        try:
+            width = os.get_terminal_size().columns
+        except (OSError, ValueError):
+            width = 120
+        is_tty = sys.stdout.isatty()
+
+        print("[Snf] Sniffing ISO 15693... press PM3 button to stop.")
+        result = await self.sniff(timeout=timeout)
+        print(f"[Snf] Sniff ended (status={result['status']})")
+
+        print("[Snf] Downloading trace...")
+        entries = await self.download_trace()
+
+        if not entries:
+            print("[Snf] No trace data captured.")
+            return []
+
+        print(f"[Snf] {len(entries)} frames captured:\n")
+        for entry in entries:
+            line = format_sniff_line(entry, width=width, is_tty=is_tty)
+            print(line)
+
+        return entries

pm3py/transport.py

@@ -233,3 +233,34 @@ class PM3Transport:
         async with self._lock:
             frame = encode_ng_frame(cmd, payload)
             await self.send_frame(frame)
+
+    async def download_bigbuf(self, offset: int = 0, length: int = PM3_CMD_DATA_SIZE,
+                              timeout: float = 4.0) -> tuple[bytes, int]:
+        """Download data from device BigBuf.
+
+        Firmware responds with reply_mix (NG frame with MIX args) for both
+        data chunks and the final ACK.
+
+        Returns (data, trace_len).
+        """
+        async with self._lock:
+            frame = encode_mix_frame(Cmd.DOWNLOAD_BIGBUF, arg0=offset, arg1=length)
+            await self.send_frame(frame)
+
+            buf = bytearray(length)
+            trace_len = 0
+
+            while True:
+                resp = await self.read_response(timeout=timeout)
+
+                if resp.cmd == Cmd.ACK:
+                    if resp.oldarg:
+                        trace_len = resp.oldarg[2]
+                    return bytes(buf[:trace_len or length]), trace_len
+
+                if resp.cmd == Cmd.DOWNLOADED_BIGBUF and resp.oldarg:
+                    chunk_offset = resp.oldarg[0]
+                    chunk_len = min(resp.oldarg[1], len(resp.data))
+                    trace_len = resp.oldarg[2]
+                    end = min(chunk_offset + chunk_len, length)
+                    buf[chunk_offset:end] = resp.data[:end - chunk_offset]

tests/test_hf_15.py

@@ -1,8 +1,14 @@
 import asyncio
+import struct
 from unittest.mock import AsyncMock
 from pm3py.protocol import Cmd
 from pm3py.transport import PM3Response
-from pm3py.hf_15 import HF15Commands
+from pm3py.hf_15 import (
+    HF15Commands, parse_tracelog, TRACELOG_HDR_SIZE,
+    decode_15693_request, decode_15693_response, decode_15693,
+    format_sniff_line, decode_ndef_annotation,
+)
+
 
 def test_15_scan():
     t = AsyncMock()
@@ -13,3 +19,271 @@ def test_15_scan():
                                           reason=0, ng=True, data=resp_data)
     result = asyncio.get_event_loop().run_until_complete(iso15.scan())
     assert "uid" in result
+
+
+# ---- Tracelog parsing ----
+
+def _make_tracelog_entry(timestamp, duration, is_response, data):
+    """Build a raw tracelog entry matching firmware's tracelog_hdr_t."""
+    data_len_flags = len(data) | (0x8000 if is_response else 0)
+    hdr = struct.pack("<IHH", timestamp, duration, data_len_flags)
+    parity_len = (len(data) + 7) // 8 if data else 0
+    parity = bytes(parity_len)
+    return hdr + data + parity
+
+
+def test_parse_tracelog_single_entry():
+    raw = _make_tracelog_entry(1000, 50, False, b"\x26\x01\x00\xAA\xBB")
+    entries = parse_tracelog(raw)
+    assert len(entries) == 1
+    assert entries[0]["timestamp"] == 1000
+    assert entries[0]["duration"] == 50
+    assert entries[0]["is_response"] is False
+    assert entries[0]["data"] == b"\x26\x01\x00\xAA\xBB"
+
+
+def test_parse_tracelog_two_entries():
+    raw = (
+        _make_tracelog_entry(1000, 50, False, b"\x26\x01\x00\xAA\xBB")
+        + _make_tracelog_entry(2000, 100, True, b"\x00\x00\x01\x02\x03\x04\x05\x06\x07\xE0\xCC\xDD")
+    )
+    entries = parse_tracelog(raw)
+    assert len(entries) == 2
+    assert entries[0]["is_response"] is False
+    assert entries[1]["is_response"] is True
+
+
+def test_parse_tracelog_empty():
+    entries = parse_tracelog(b"")
+    assert entries == []
+
+
+def test_parse_tracelog_zero_sentinel():
+    raw = _make_tracelog_entry(1000, 50, False, b"\x26\x01\x00\xAA\xBB")
+    raw += b"\x00" * TRACELOG_HDR_SIZE  # zero sentinel
+    raw += _make_tracelog_entry(9999, 50, False, b"\xFF")  # should not parse
+    entries = parse_tracelog(raw)
+    assert len(entries) == 1
+
+
+# ---- 15693 decode ----
+
+def test_decode_inventory_request():
+    ann = decode_15693_request(b"\x26\x01\x00")
+    assert ann == "INVENTORY"
+
+
+def test_decode_inventory_with_mask():
+    ann = decode_15693_request(b"\x26\x01\x08\xFF")
+    assert "mask=8" in ann
+
+
+def test_decode_read_single_unaddressed():
+    ann = decode_15693_request(b"\x02\x20\x05")
+    assert ann == "READ SINGLE BLOCK #5"
+
+
+def test_decode_read_single_addressed():
+    # flags=0x22 (address + high data rate), cmd=0x20, uid(8), block=3
+    payload = bytes([0x22, 0x20]) + bytes(8) + bytes([3])
+    ann = decode_15693_request(payload)
+    assert ann == "READ SINGLE BLOCK #3"
+
+
+def test_decode_read_multiple():
+    ann = decode_15693_request(b"\x02\x23\x00\x0F")
+    assert "READ MULTIPLE BLOCKS #0+15" in ann
+
+
+def test_decode_get_system_info():
+    ann = decode_15693_request(b"\x02\x2B")
+    assert ann == "GET SYSTEM INFO"
+
+
+def test_decode_nxp_set_password():
+    ann = decode_15693_request(bytes([0x02, 0xB3, 0x04, 0x04]))
+    assert "NXP SET PASSWORD" in ann
+    assert "id=4" in ann
+
+
+def test_decode_unknown_command():
+    ann = decode_15693_request(bytes([0x02, 0xFE]))
+    assert "UNKNOWN CMD 0xFE" in ann
+
+
+def test_decode_response_ok():
+    ann = decode_15693_response(b"\x00")
+    assert ann == "OK"
+
+
+def test_decode_response_ok_data():
+    ann = decode_15693_response(b"\x00\xDE\xAD\xBE\xEF")
+    assert "OK [4B]" in ann
+
+
+def test_decode_response_inventory():
+    # flags(1) + dsfid(1) + uid(8) = 10 bytes total, 9 after flags
+    payload = bytes([0x00, 0x00]) + bytes([0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0xE0])
+    ann = decode_15693_response(payload)
+    assert "INVENTORY" in ann
+    assert "UID=" in ann
+
+
+def test_decode_response_error():
+    ann = decode_15693_response(bytes([0x01, 0x10]))
+    assert "ERROR" in ann
+    assert "block not available" in ann
+
+
+def test_decode_15693_dispatches():
+    assert decode_15693(0, b"\x02\x2B") == "GET SYSTEM INFO"
+    assert "OK" in decode_15693(1, b"\x00")
+
+
+# ---- format_sniff_line ----
+
+def test_format_sniff_line_reader():
+    entry = {"is_response": False, "data": b"\x26\x01\x00\xAA\xBB",
+             "timestamp": 0, "duration": 0, "parity": b""}
+    line = format_sniff_line(entry, width=120, is_tty=False)
+    assert "Reader" in line
+    assert "Tag" in line
+    assert "INVENTORY" in line
+
+
+def test_format_sniff_line_tag():
+    payload = bytes([0x00, 0x00]) + bytes([0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0xE0]) + bytes([0xCC, 0xDD])
+    entry = {"is_response": True, "data": payload,
+             "timestamp": 0, "duration": 0, "parity": b""}
+    line = format_sniff_line(entry, width=120, is_tty=False)
+    assert "Tag" in line
+    assert "INVENTORY" in line
+
+
+# ---- NDEF decode ----
+
+class TestDecodeNdefAnnotation:
+    def test_cc_block(self):
+        data = bytes([0xE1, 0x40, 0x0D, 0x01])
+        ann = decode_ndef_annotation(data)
+        assert ann is not None
+        assert "CC" in ann
+        assert "v1.0" in ann
+        assert "MBREAD" in ann
+
+    def test_cc_no_mbread(self):
+        data = bytes([0xE1, 0x40, 0x10, 0x00])
+        ann = decode_ndef_annotation(data)
+        assert "CC" in ann
+        assert "MBREAD" not in ann
+
+    def test_ndef_text_record(self):
+        msg = bytes([0xD1, 0x01, 0x07, 0x54, 0x02, 0x65, 0x6E, 0x74, 0x65, 0x73, 0x74])
+        tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
+        ann = decode_ndef_annotation(tlv)
+        assert ann is not None
+        assert '"test"' in ann
+
+    def test_ndef_uri_record(self):
+        uri_payload = bytes([0x04]) + b"example.com"
+        rec = bytes([0xD1, 0x01, len(uri_payload), 0x55]) + uri_payload
+        tlv = bytes([0x03, len(rec)]) + rec + bytes([0xFE])
+        ann = decode_ndef_annotation(tlv)
+        assert "https://example.com" in ann
+
+    def test_ndef_mime_record(self):
+        mime_type = b"text/plain"
+        payload = b"hello"
+        rec = bytes([0xD2, len(mime_type), len(payload)]) + mime_type + payload
+        tlv = bytes([0x03, len(rec)]) + rec + bytes([0xFE])
+        ann = decode_ndef_annotation(tlv)
+        assert "text/plain" in ann
+
+    def test_no_ndef_magic(self):
+        ann = decode_ndef_annotation(bytes([0x00, 0x00, 0x00, 0x00]))
+        assert ann is None
+
+    def test_cc_plus_ndef(self):
+        cc = bytes([0xE1, 0x40, 0x0D, 0x01])
+        msg = bytes([0xD1, 0x01, 0x07, 0x54, 0x02, 0x65, 0x6E, 0x74, 0x65, 0x73, 0x74])
+        tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
+        data = cc + tlv + bytes(50)
+        ann = decode_ndef_annotation(data)
+        assert "CC" in ann
+        assert '"test"' in ann
+
+    def test_empty_ndef(self):
+        tlv = bytes([0x03, 0x00, 0xFE])
+        ann = decode_ndef_annotation(tlv)
+        assert ann is not None
+        assert "empty" in ann.lower() or "NDEF" in ann
+
+    def test_long_text_truncated(self):
+        # Build a long NDEF text record manually (no sim.type5 dependency)
+        long_text = "A" * 200
+        lang = b"en"
+        payload = bytes([len(lang)]) + lang + long_text.encode("utf-8")
+        rec = bytes([0xD1, 0x01, len(payload), 0x54]) + payload
+        tlv = bytes([0x03, 0xFF, (len(rec) >> 8) & 0xFF, len(rec) & 0xFF]) + rec + bytes([0xFE])
+        ann = decode_ndef_annotation(tlv)
+        assert ann is not None
+        assert "..." in ann
+
+
+class TestResponseNdefDecode:
+    def test_read_single_block_0_cc(self):
+        data = bytes([0x00, 0xE1, 0x40, 0x0D, 0x01])
+        ann = decode_15693_response(data)
+        assert "CC" in ann
+        assert "MBREAD" in ann
+
+    def test_read_multiple_blocks_ndef(self):
+        msg = bytes([0xD1, 0x01, 0x07, 0x54, 0x02, 0x65, 0x6E, 0x74, 0x65, 0x73, 0x74])
+        tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
+        data = bytes([0x00]) + tlv + bytes(50)
+        ann = decode_15693_response(data)
+        assert '"test"' in ann
+
+    def test_read_multiple_with_cc_and_ndef(self):
+        cc = bytes([0xE1, 0x40, 0x0D, 0x01])
+        msg = bytes([0xD1, 0x01, 0x07, 0x54, 0x02, 0x65, 0x6E, 0x74, 0x65, 0x73, 0x74])
+        tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
+        data = bytes([0x00]) + cc + tlv + bytes(40)
+        ann = decode_15693_response(data)
+        assert "CC" in ann
+        assert '"test"' in ann
+
+    def test_generic_data_no_ndef(self):
+        data = bytes([0x00, 0xDE, 0xAD, 0xBE, 0xEF])
+        ann = decode_15693_response(data)
+        assert "OK [4B]" in ann
+
+    def test_inventory_not_affected(self):
+        data = bytes([0x00, 0x00]) + bytes(8)
+        ann = decode_15693_response(data)
+        assert "INVENTORY" in ann
+
+
+class TestSniffLineAnnotationOverflow:
+    def test_long_annotation_wraps(self):
+        cc = bytes([0xE1, 0x40, 0x0D, 0x01])
+        # Build NDEF text record manually (no sim.type5 dependency)
+        text = "A" * 30
+        lang = b"en"
+        payload = bytes([len(lang)]) + lang + text.encode("utf-8")
+        msg = bytes([0xD1, 0x01, len(payload), 0x54]) + payload
+        tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
+        payload = bytes([0x00]) + cc + tlv + bytes([0xAA, 0xBB])
+        entry = {"is_response": True, "data": payload,
+                 "timestamp": 0, "duration": 0, "parity": b""}
+        line = format_sniff_line(entry, width=80, is_tty=False)
+        lines = line.split("\n")
+        ann_lines = [l for l in lines if "NDEF" in l]
+        assert len(ann_lines) >= 1
+
+    def test_short_annotation_no_wrap(self):
+        data = bytes([0x00, 0xE1, 0x40, 0x0D, 0x01, 0xAA, 0xBB])
+        entry = {"is_response": True, "data": data,
+                 "timestamp": 0, "duration": 0, "parity": b""}
+        line = format_sniff_line(entry, width=120, is_tty=False)
+        assert "CC" in line