pm3py/docs/plans/2026-03-19-live-sniff-plan.md

Live Sniff Streaming Implementation Plan

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: Rewrite sniff from batch download to live-streaming with button-toggle pause/resume, auto-loaded in the REPL.

Architecture: Firmware sends trace entries as fire-and-forget reply_ng() packets on RF idle (reusing sim's ring buffer pattern). Python SniffSession receives in a background thread (mirroring SimSession._trace_reader). .pythonstartup.py auto-creates session in sniff mode.

Tech Stack: C (firmware ARM), Python 3.12 (pyserial, threading), pytest with mocking

Design doc: docs/plans/2026-03-19-live-sniff-design.md

---

Task 1: Firmware — Add command IDs and structs

Files:

• Modify: firmware/include/pm3_cmd.h:734 (after CMD_HF_ISO15693_SIM_TRACE)
• Modify: pm3py/core/protocol.py:203 (after HF_ISO15693_EML_SETMEM)

Step 1: Add firmware command definitions

In firmware/include/pm3_cmd.h, after line 734 (CMD_HF_ISO15693_SIM_TRACE), add:

#define CMD_HF_SNIFF_STREAM             0x0337  // FW->Host: live sniff trace entry
#define CMD_HF_SNIFF_STATUS             0x0338  // FW->Host: sniff state change

// Sniff streaming flags (sent in sniff command payload)
#define SNIFF_FLAG_STREAMING            0x01
#define SNIFF_FLAG_BUTTON_TOGGLE        0x02

// Sniff state values (in CMD_HF_SNIFF_STATUS payload)
#define SNIFF_STATE_STARTED             0
#define SNIFF_STATE_PAUSED              1
#define SNIFF_STATE_RESUMED             2
#define SNIFF_STATE_STOPPED             3

Step 2: Add Python Cmd enum entries

In pm3py/core/protocol.py, after HF_ISO15693_EML_SETMEM = 0x0331 (line 203), add:

    HF_SNIFF_STREAM = 0x0337
    HF_SNIFF_STATUS = 0x0338

Also add constants after the Cmd class:

# Sniff streaming flags
SNIFF_FLAG_STREAMING = 0x01
SNIFF_FLAG_BUTTON_TOGGLE = 0x02

# Sniff states
SNIFF_STATE_STARTED = 0
SNIFF_STATE_PAUSED = 1
SNIFF_STATE_RESUMED = 2
SNIFF_STATE_STOPPED = 3

Step 3: Commit

git add firmware/include/pm3_cmd.h pm3py/core/protocol.py
git commit --no-gpg-sign -m "feat: add CMD_HF_SNIFF_STREAM/STATUS command IDs and constants"

---

Task 2: Firmware — Streaming sniff loop

Files:

• Modify: firmware/armsrc/iso15693.c:1654-1875 (SniffIso15693)
• Modify: firmware/armsrc/appmain.c:1500-1503 (CMD_HF_ISO15693_SNIFF handler)

Step 1: Modify appmain.c to pass payload to SniffIso15693

Currently (line 1500-1503):

case CMD_HF_ISO15693_SNIFF: {
    SniffIso15693(0, NULL, false);
    reply_ng(CMD_HF_ISO15693_SNIFF, PM3_SUCCESS, NULL, 0);
    break;
}

Change to:

case CMD_HF_ISO15693_SNIFF: {
    uint16_t idle_timeout_ms = 0;
    uint8_t sniff_flags = 0;
    if (packet->length >= 3) {
        idle_timeout_ms = packet->data.asBytes[0] | (packet->data.asBytes[1] << 8);
        sniff_flags = packet->data.asBytes[2];
    }
    SniffIso15693(0, NULL, false, idle_timeout_ms, sniff_flags);
    if (!(sniff_flags & SNIFF_FLAG_STREAMING)) {
        // Legacy mode: send completion response (streaming mode sends STATUS)
        reply_ng(CMD_HF_ISO15693_SNIFF, PM3_SUCCESS, NULL, 0);
    }
    break;
}

Step 2: Update SniffIso15693 signature and add streaming mode

In firmware/armsrc/iso15693.c, change the function signature (line 1654):

void SniffIso15693(uint8_t jam_search_len, uint8_t *jam_search_string, bool iclass,
                   uint16_t idle_timeout_ms, uint8_t sniff_flags) {

After the existing init code (after line 1698, the DMA setup), add the streaming mode branch. The key change: when sniff_flags & SNIFF_FLAG_STREAMING, use a ring buffer + idle flush loop instead of LogTrace to BigBuf.

Insert after the DMA setup succeeds (after line 1697), before the state variables (line 1699):

    // --- Streaming mode ---
    if (sniff_flags & SNIFF_FLAG_STREAMING) {

        // Ring buffer (same pattern as SimTagIso15693)
        #define SNIFF_TRACE_ENTRIES 8
        #define SNIFF_TRACE_ENTRY_MAX 72  // 8-byte header + 64 data
        static uint8_t sniff_ring[SNIFF_TRACE_ENTRIES][SNIFF_TRACE_ENTRY_MAX];
        static uint16_t sniff_ring_len[SNIFF_TRACE_ENTRIES];
        uint8_t sniff_wr = 0, sniff_rd = 0;
        uint16_t frame_count = 0;
        uint32_t last_activity = 0;
        bool sniff_paused = false;

        // Macro: push trace entry with full metadata
        // Layout: [protocol(1) | direction(1) | duration(2) | timestamp(4) | data(*)]
        #define SNIFF_PUSH(dir, data_ptr, data_len, ts, dur) do { \
            if (((sniff_wr + 1) % SNIFF_TRACE_ENTRIES) != sniff_rd) { \
                uint8_t *_e = sniff_ring[sniff_wr]; \
                _e[0] = 0;  /* protocol: 0=15693 */ \
                _e[1] = (dir); \
                _e[2] = (dur) & 0xFF; _e[3] = ((dur) >> 8) & 0xFF; \
                _e[4] = (ts) & 0xFF; _e[5] = ((ts) >> 8) & 0xFF; \
                _e[6] = ((ts) >> 16) & 0xFF; _e[7] = ((ts) >> 24) & 0xFF; \
                uint16_t _dl = (data_len); \
                if (_dl > SNIFF_TRACE_ENTRY_MAX - 8) _dl = SNIFF_TRACE_ENTRY_MAX - 8; \
                memcpy(_e + 8, (data_ptr), _dl); \
                sniff_ring_len[sniff_wr] = 8 + _dl; \
                sniff_wr = (sniff_wr + 1) % SNIFF_TRACE_ENTRIES; \
            } \
        } while(0)

        #define SNIFF_FLUSH() do { \
            while (sniff_rd != sniff_wr) { \
                reply_ng(CMD_HF_SNIFF_STREAM, PM3_SUCCESS, \
                         sniff_ring[sniff_rd], sniff_ring_len[sniff_rd]); \
                sniff_rd = (sniff_rd + 1) % SNIFF_TRACE_ENTRIES; \
            } \
        } while(0)

        // Send started status
        uint8_t status_buf[3] = { SNIFF_STATE_STARTED, 0, 0 };
        reply_ng(CMD_HF_SNIFF_STATUS, PM3_SUCCESS, status_buf, 3);

        LED_A_ON();  // green = sniffing active

        // --- Streaming sniff main loop ---
        // (This duplicates the DMA decode logic from the legacy path below,
        //  but replaces LogTrace calls with SNIFF_PUSH and adds idle flush +
        //  button toggle + BREAK_LOOP check.)

        bool tag_is_active = false;
        bool reader_is_active = false;
        bool expect_tag_answer = false;
        bool expect_fsk_answer = false;
        bool expect_fast_answer = true;
        int dma_start_time = 0;
        int samples = 0;
        const uint16_t *upTo = dma->buf;
        bool exit_sniff = false;

        while (!exit_sniff) {

            volatile int behind_by = ((uint16_t *)AT91C_BASE_PDC_SSC->PDC_RPR - upTo) & (DMA_BUFFER_SIZE - 1);
            if (behind_by < 1) {
                // No new DMA data — check idle flush, button, break
                if (!sniff_paused && sniff_wr != sniff_rd && last_activity > 0) {
                    uint32_t now = GetTickCount();
                    if ((now - last_activity) > idle_timeout_ms) {
                        SNIFF_FLUSH();
                    }
                }

                // Check for Python BREAK_LOOP
                if (data_available()) {
                    PacketCommandNG rx;
                    if (receive_ng(&rx) == PM3_SUCCESS && rx.cmd == CMD_BREAK_LOOP) {
                        SNIFF_FLUSH();
                        exit_sniff = true;
                    }
                }
                continue;
            }

            samples++;
            if (samples == 1) {
                dma_start_time = GetCountSspClk() & 0xfffffff0;
            }

            volatile uint16_t sniffdata = 0;
            volatile uint16_t sniffdata_prev = sniffdata;
            sniffdata = *upTo++;

            if (upTo >= dma->buf + DMA_BUFFER_SIZE) {
                upTo = dma->buf;

                if (AT91C_BASE_SSC->SSC_SR & (AT91C_SSC_ENDRX)) {
                    if (AT91C_BASE_PDC_SSC->PDC_RCR == 0) {
                        AT91C_BASE_PDC_SSC->PDC_RPR = (uint32_t) dma->buf;
                        AT91C_BASE_PDC_SSC->PDC_RCR = DMA_BUFFER_SIZE;
                    }
                    if (AT91C_BASE_PDC_SSC->PDC_RNCR == 0) {
                        AT91C_BASE_PDC_SSC->PDC_RNPR = (uint32_t) dma->buf;
                        AT91C_BASE_PDC_SSC->PDC_RNCR = DMA_BUFFER_SIZE;
                    }

                    WDT_HIT();

                    // Button toggle
                    if (BUTTON_PRESS()) {
                        SpinDelay(200);  // debounce
                        while (BUTTON_PRESS()) { WDT_HIT(); }  // wait for release
                        SpinDelay(200);  // debounce release

                        if (sniff_paused) {
                            sniff_paused = false;
                            LED_A_ON();
                            status_buf[0] = SNIFF_STATE_RESUMED;
                            status_buf[1] = frame_count & 0xFF;
                            status_buf[2] = (frame_count >> 8) & 0xFF;
                            reply_ng(CMD_HF_SNIFF_STATUS, PM3_SUCCESS, status_buf, 3);
                        } else {
                            SNIFF_FLUSH();
                            sniff_paused = true;
                            LED_A_OFF();
                            status_buf[0] = SNIFF_STATE_PAUSED;
                            status_buf[1] = frame_count & 0xFF;
                            status_buf[2] = (frame_count >> 8) & 0xFF;
                            reply_ng(CMD_HF_SNIFF_STATUS, PM3_SUCCESS, status_buf, 3);
                        }
                    }

                    // Check for Python BREAK_LOOP
                    if (data_available()) {
                        PacketCommandNG rx;
                        if (receive_ng(&rx) == PM3_SUCCESS && rx.cmd == CMD_BREAK_LOOP) {
                            SNIFF_FLUSH();
                            exit_sniff = true;
                        }
                    }
                }
            }

            if (sniff_paused) continue;

            // --- Reader decode (same as legacy) ---
            if (tag_is_active == false) {
                int extra_8s = 1;
                if (Handle15693SampleFromReader((sniffdata & 0x02) >> 1, &dreader) ||
                        (++extra_8s && Handle15693SampleFromReader(sniffdata & 0x01, &dreader))) {

                    if (dreader.byteCount > 0) {
                        uint32_t eof_time = dma_start_time + (samples * 16) + (extra_8s * 8) - DELAY_READER_TO_ARM_SNIFF;
                        uint32_t sof_time = eof_time
                                            - dreader.byteCount * (dreader.Coding == CODING_1_OUT_OF_4 ? 1024 : 16384)
                                            - 256 - 128;

                        uint32_t ts_us = (sof_time * 4) / 14;  // ssp_clk*4 -> ~us
                        uint16_t dur = (uint16_t)(((eof_time - sof_time) * 4) / 14);
                        SNIFF_PUSH(0, dreader.output, dreader.byteCount, ts_us, dur);
                        frame_count++;
                        last_activity = GetTickCount();

                        if (iclass == false) {
                            expect_fsk_answer = dreader.output[0] & ISO15_REQ_SUBCARRIER_TWO;
                            expect_fast_answer = dreader.output[0] & ISO15_REQ_DATARATE_HIGH;
                        }
                    }

                    DecodeTagReset(&dtag);
                    DecodeTagFSKReset(&dtagfsk);
                    reader_is_active = false;
                    expect_tag_answer = true;
                } else {
                    reader_is_active = (dreader.state >= STATE_READER_RECEIVE_DATA_1_OUT_OF_4);
                }
            }

            // --- Tag decode: single subcarrier (same as legacy) ---
            if ((reader_is_active == false) && expect_tag_answer) {
                if (expect_fsk_answer == false) {
                    if (Handle15693SamplesFromTag((sniffdata >> 4) << 2, &dtag, expect_fast_answer)) {
                        uint32_t eof_time = dma_start_time + (samples * 16) - DELAY_TAG_TO_ARM_SNIFF;
                        if (dtag.lastBit == SOF_PART2) {
                            eof_time -= (8 * 16);
                        }
                        uint32_t sof_time = eof_time
                                            - dtag.len * 1024 - 512
                                            - (dtag.lastBit != SOF_PART2 ? 512 : 0);

                        uint32_t ts_us = (sof_time * 4) / 14;
                        uint16_t dur = (uint16_t)(((eof_time - sof_time) * 4) / 14);
                        SNIFF_PUSH(1, dtag.output, dtag.len, ts_us, dur);
                        frame_count++;
                        last_activity = GetTickCount();

                        DecodeTagReset(&dtag);
                        DecodeTagFSKReset(&dtagfsk);
                        DecodeReaderReset(&dreader);
                        expect_tag_answer = false;
                        tag_is_active = false;
                    } else {
                        tag_is_active = (dtag.state >= STATE_TAG_RECEIVING_DATA);
                    }

                } else {
                    // --- Tag decode: dual subcarrier FSK (same as legacy) ---
                    if (FREQ_IS_0((sniffdata >> 2) & 0x3)) {
                        sniffdata = sniffdata_prev;
                    }
                    if (Handle15693FSKSamplesFromTag((sniffdata >> 2) & 0x3, &dtagfsk, expect_fast_answer)) {
                        if (dtagfsk.len > 0) {
                            uint32_t eof_time = dma_start_time + (samples * 16) - DELAY_TAG_TO_ARM_SNIFF;
                            if (dtagfsk.lastBit == SOF) {
                                eof_time -= (8 * 16);
                            }
                            uint32_t sof_time = eof_time
                                                - dtagfsk.len * 1016 - 512
                                                - (dtagfsk.lastBit != SOF ? 512 : 0);

                            uint32_t ts_us = (sof_time * 4) / 14;
                            uint16_t dur = (uint16_t)(((eof_time - sof_time) * 4) / 14);
                            SNIFF_PUSH(1, dtagfsk.output, dtagfsk.len, ts_us, dur);
                            frame_count++;
                            last_activity = GetTickCount();
                        }

                        DecodeTagFSKReset(&dtagfsk);
                        DecodeReaderReset(&dreader);
                        expect_tag_answer = false;
                        tag_is_active = false;
                        expect_fsk_answer = false;
                    } else {
                        tag_is_active = (dtagfsk.state >= STATE_FSK_RECEIVING_DATA_484);
                    }
                }
            }
        }

        // Cleanup and send stopped status
        FpgaDisableTracing();
        switch_off();
        status_buf[0] = SNIFF_STATE_STOPPED;
        status_buf[1] = frame_count & 0xFF;
        status_buf[2] = (frame_count >> 8) & 0xFF;
        reply_ng(CMD_HF_SNIFF_STATUS, PM3_SUCCESS, status_buf, 3);
        return;
    }

    // --- Legacy mode (existing code, unchanged from here) ---

Also update the function declaration in iso15693.h (if it exists) or any forward declarations.

Step 3: Update any forward declarations

Search for SniffIso15693 declarations and update the signature to include the two new parameters.

Step 4: Commit

git add firmware/armsrc/iso15693.c firmware/armsrc/appmain.c
# Also add any header files modified
git commit --no-gpg-sign -m "feat(fw): streaming sniff with ring buffer, button-toggle, idle flush"

---

Task 3: Python — SniffSession rewrite with streaming

Files:

• Modify: pm3py/sniff/session.py (full rewrite)
• Reference: pm3py/sim/sim_session.py:54-83 (open pattern), pm3py/sim/sim_session.py:237-290 (trace reader pattern)

Step 1: Write tests for the new SniffSession

Create tests/test_sniff_session.py:

"""Tests for SniffSession — live streaming sniff with button-toggle."""
import struct
import threading
import time
from unittest.mock import MagicMock, patch, PropertyMock

import pytest

from pm3py.core.protocol import (
    Cmd, SNIFF_FLAG_STREAMING, SNIFF_FLAG_BUTTON_TOGGLE,
    SNIFF_STATE_STARTED, SNIFF_STATE_PAUSED,
    SNIFF_STATE_RESUMED, SNIFF_STATE_STOPPED,
)
from pm3py.core.transport import (
    encode_ng_frame, RESP_PREAMBLE_MAGIC, RESP_POSTAMBLE_NOCRC,
)
from pm3py.sniff.session import SniffSession


def _make_resp_frame(cmd: int, status: int, data: bytes) -> bytes:
    """Build a raw NG response frame (USB, no CRC)."""
    payload_len = len(data)
    preamble = struct.pack("<IHBBH",
        RESP_PREAMBLE_MAGIC,
        payload_len | 0x8000,  # NG bit set
        status & 0xFF,
        0,   # reason
        cmd,
    )
    postamble = struct.pack("<H", RESP_POSTAMBLE_NOCRC)
    return preamble + data + postamble


def _make_sniff_stream_frame(protocol: int, direction: int,
                              duration: int, timestamp: int,
                              data: bytes) -> bytes:
    """Build a CMD_HF_SNIFF_STREAM response frame."""
    payload = struct.pack("<BBHI", protocol, direction, duration, timestamp) + data
    return _make_resp_frame(Cmd.HF_SNIFF_STREAM.value, 0, payload)


def _make_status_frame(state: int, count: int) -> bytes:
    """Build a CMD_HF_SNIFF_STATUS response frame."""
    payload = struct.pack("<BH", state, count)
    return _make_resp_frame(Cmd.HF_SNIFF_STATUS.value, 0, payload)


class FakeSerial:
    """Mock serial port that feeds pre-built response frames."""

    def __init__(self, frames: list[bytes]):
        self._buf = bytearray()
        for f in frames:
            self._buf.extend(f)
        self._pos = 0
        self._closed = False

    @property
    def in_waiting(self):
        return len(self._buf) - self._pos

    def read(self, size=1):
        if self._closed or self._pos >= len(self._buf):
            time.sleep(0.01)  # avoid busy spin
            return b""
        end = min(self._pos + size, len(self._buf))
        data = bytes(self._buf[self._pos:end])
        self._pos = end
        return data

    def write(self, data):
        return len(data)

    def reset_input_buffer(self):
        pass

    def reset_output_buffer(self):
        pass

    def close(self):
        self._closed = True


class TestSniffSessionStreaming:
    """Tests for live streaming mode."""

    def test_start_15693_sends_streaming_flags(self):
        """start_15693(live=True) sends correct payload."""
        ser = FakeSerial([
            _make_status_frame(SNIFF_STATE_STARTED, 0),
            _make_status_frame(SNIFF_STATE_STOPPED, 0),
        ])
        sent = bytearray()
        orig_write = ser.write
        def capture_write(data):
            sent.extend(data)
            return orig_write(data)
        ser.write = capture_write

        session = SniffSession(ser)
        session.start_15693(live=True, idle_timeout_ms=400)
        time.sleep(0.1)
        session.stop()

        # Verify payload contains flags
        assert len(sent) > 0
        # The payload should contain idle_timeout_ms (400=0x0190) and flags (0x03)
        assert b"\x90\x01\x03" in bytes(sent)

    def test_start_15693_legacy_mode(self):
        """start_15693(live=False) sends zero flags."""
        ser = FakeSerial([
            _make_resp_frame(Cmd.HF_ISO15693_SNIFF.value, 0, b""),
        ])
        sent = bytearray()
        orig_write = ser.write
        def capture_write(data):
            sent.extend(data)
            return orig_write(data)
        ser.write = capture_write

        session = SniffSession(ser)
        session.start_15693(live=False)
        # Legacy should block and return

    def test_receives_trace_entries(self):
        """Background thread decodes CMD_HF_SNIFF_STREAM frames."""
        reader_cmd = bytes([0x26, 0x01, 0x00])  # INVENTORY
        tag_resp = bytes([0x00, 0xE0, 0x04])

        ser = FakeSerial([
            _make_status_frame(SNIFF_STATE_STARTED, 0),
            _make_sniff_stream_frame(0, 0, 100, 1000, reader_cmd),
            _make_sniff_stream_frame(0, 1, 200, 2000, tag_resp),
            _make_status_frame(SNIFF_STATE_STOPPED, 2),
        ])

        session = SniffSession(ser)
        session.start_15693(live=True)
        time.sleep(0.3)  # let background thread process

        assert len(session.entries) == 2
        assert session.entries[0]["direction"] == 0
        assert session.entries[0]["data"] == reader_cmd
        assert session.entries[0]["timestamp"] == 1000
        assert session.entries[1]["direction"] == 1
        assert session.entries[1]["data"] == tag_resp

    def test_status_updates(self):
        """Status frames update session state."""
        ser = FakeSerial([
            _make_status_frame(SNIFF_STATE_STARTED, 0),
            _make_sniff_stream_frame(0, 0, 100, 1000, b"\x26\x01\x00"),
            _make_status_frame(SNIFF_STATE_PAUSED, 1),
            _make_status_frame(SNIFF_STATE_RESUMED, 1),
            _make_status_frame(SNIFF_STATE_STOPPED, 1),
        ])

        session = SniffSession(ser)
        session.start_15693(live=True)
        time.sleep(0.3)

        assert session.state == "stopped"

    def test_stop_sends_break_loop(self):
        """stop() sends CMD_BREAK_LOOP to firmware."""
        ser = FakeSerial([
            _make_status_frame(SNIFF_STATE_STARTED, 0),
            _make_status_frame(SNIFF_STATE_STOPPED, 0),
        ])
        sent = bytearray()
        orig_write = ser.write
        def capture_write(data):
            sent.extend(data)
            return orig_write(data)
        ser.write = capture_write

        session = SniffSession(ser)
        session.start_15693(live=True)
        time.sleep(0.1)
        session.stop()

        # BREAK_LOOP = 0x0118
        assert b"\x18\x01" in bytes(sent)

    def test_entries_accumulate_across_cycles(self):
        """Entries persist across pause/resume cycles."""
        ser = FakeSerial([
            _make_status_frame(SNIFF_STATE_STARTED, 0),
            _make_sniff_stream_frame(0, 0, 100, 1000, b"\x26\x01\x00"),
            _make_status_frame(SNIFF_STATE_PAUSED, 1),
            _make_status_frame(SNIFF_STATE_RESUMED, 1),
            _make_sniff_stream_frame(0, 0, 100, 3000, b"\x26\x01\x00"),
            _make_status_frame(SNIFF_STATE_STOPPED, 2),
        ])

        session = SniffSession(ser)
        session.start_15693(live=True)
        time.sleep(0.3)

        assert len(session.entries) == 2

    def test_clear_entries(self):
        """clear() resets accumulated entries."""
        ser = FakeSerial([
            _make_status_frame(SNIFF_STATE_STARTED, 0),
            _make_sniff_stream_frame(0, 0, 100, 1000, b"\x26\x01\x00"),
            _make_status_frame(SNIFF_STATE_STOPPED, 1),
        ])

        session = SniffSession(ser)
        session.start_15693(live=True)
        time.sleep(0.3)

        assert len(session.entries) == 1
        session.clear()
        assert len(session.entries) == 0

Step 2: Run tests to verify they fail

cd /home/work/pm3py && python -m pytest tests/test_sniff_session.py -v

Expected: ImportError / AttributeError — new constants and SniffSession API don't exist yet.

Step 3: Implement SniffSession

Rewrite pm3py/sniff/session.py:

"""Sniff sessions — live streaming with button-toggle, or legacy batch."""
import struct
import sys
import threading
import time

import serial

from ..core.protocol import (
    Cmd, PM3_CMD_DATA_SIZE,
    SNIFF_FLAG_STREAMING, SNIFF_FLAG_BUTTON_TOGGLE,
    SNIFF_STATE_STARTED, SNIFF_STATE_PAUSED,
    SNIFF_STATE_RESUMED, SNIFF_STATE_STOPPED,
)
from ..core.transport import (
    encode_ng_frame, decode_response_frame,
    RESP_PREAMBLE_MAGIC, RESP_PREAMBLE_SIZE, RESP_POSTAMBLE_SIZE,
)
from ..trace import parse_tracelog
from ..trace.decode_iso15 import decode_15693
from ..trace.format import TraceFormatter, format_sniff_line

# Default serial port (same as SimSession)
DEFAULT_PORT = "/dev/ttyACM0"

_STATE_NAMES = {
    SNIFF_STATE_STARTED: "sniffing",
    SNIFF_STATE_PAUSED: "paused",
    SNIFF_STATE_RESUMED: "sniffing",
    SNIFF_STATE_STOPPED: "stopped",
}

_STATE_ICONS = {
    SNIFF_STATE_STARTED: ">>",
    SNIFF_STATE_PAUSED: "||",
    SNIFF_STATE_RESUMED: ">>",
    SNIFF_STATE_STOPPED: "[]",
}

_STATE_LABELS = {
    SNIFF_STATE_STARTED: "Started",
    SNIFF_STATE_PAUSED: "Paused",
    SNIFF_STATE_RESUMED: "Resumed",
    SNIFF_STATE_STOPPED: "Stopped",
}


class SniffSession:
    """Persistent sniff session with live streaming and button-toggle.

    For live mode (default):
        session = SniffSession.open()
        session.start_15693()    # streams trace live, button toggles pause
        session.stop()           # or let button cycles accumulate
        session.entries          # all captured frames

    For legacy mode:
        session.start_15693(live=False)  # blocks until button, batch download
    """

    def __init__(self, ser):
        self._ser = ser
        self._active = False
        self._entries: list[dict] = []
        self._reader_thread: threading.Thread | None = None
        self._formatter: TraceFormatter | None = None
        self._state = "idle"

    @classmethod
    def open(cls, port: str = DEFAULT_PORT, baudrate: int = 115200) -> "SniffSession":
        """Open serial connection to PM3. Auto-detects port if default."""
        ser = serial.Serial()
        ser.port = port
        ser.baudrate = baudrate
        ser.bytesize = 8
        ser.parity = "N"
        ser.stopbits = 1
        ser.timeout = 0.1
        ser.xonxoff = False
        ser.rtscts = False
        ser.dsrdtr = False
        ser.open()

        ser.reset_input_buffer()
        ser.reset_output_buffer()
        time.sleep(0.5)
        ser.reset_input_buffer()

        return cls(ser)

    def start_15693(self, live: bool = True, idle_timeout_ms: int = 400) -> None:
        """Start 15693 sniff session.

        live=True (default): streaming with button-toggle pause/resume.
        live=False: legacy BigBuf batch mode, blocks until button exits.
        """
        self._protocol = 0  # 15693
        flags = 0
        if live:
            flags = SNIFF_FLAG_STREAMING | SNIFF_FLAG_BUTTON_TOGGLE

        payload = struct.pack("<HB", idle_timeout_ms, flags)
        frame = encode_ng_frame(Cmd.HF_ISO15693_SNIFF, payload)
        self._ser.write(frame)

        if live:
            self._formatter = TraceFormatter(
                mode="sniff", decoder=decode_15693, crc_len=2)
            self._active = True
            self._state = "starting"
            self._reader_thread = threading.Thread(
                target=self._stream_reader, daemon=True)
            self._reader_thread.start()
        else:
            self._legacy_sniff()

    def _stream_reader(self) -> None:
        """Background thread: read and print trace frames live."""
        buf = bytearray()
        while self._active:
            try:
                chunk = self._ser.read(self._ser.in_waiting or 1)
                if not chunk:
                    continue
                buf.extend(chunk)

                while self._try_decode_frame(buf):
                    pass

            except serial.SerialException:
                break
            except Exception:
                continue

    def _try_decode_frame(self, buf: bytearray) -> bool:
        """Decode one response frame from buf. Returns True if consumed."""
        if len(buf) < RESP_PREAMBLE_SIZE + RESP_POSTAMBLE_SIZE:
            return False

        idx = buf.find(struct.pack("<I", RESP_PREAMBLE_MAGIC))
        if idx < 0:
            buf[:] = buf[-3:]
            return False
        if idx > 0:
            del buf[:idx]

        if len(buf) < RESP_PREAMBLE_SIZE:
            return False

        length_ng = struct.unpack_from("<H", buf, 4)[0]
        payload_len = length_ng & 0x7FFF
        frame_len = RESP_PREAMBLE_SIZE + payload_len + RESP_POSTAMBLE_SIZE

        if len(buf) < frame_len:
            return False

        try:
            resp = decode_response_frame(bytes(buf[:frame_len]))
            del buf[:frame_len]
        except Exception:
            del buf[:4]
            return True

        cmd = resp["cmd"]
        data = resp.get("data", b"")

        if cmd == Cmd.HF_SNIFF_STREAM.value:
            self._handle_trace(data)
        elif cmd == Cmd.HF_SNIFF_STATUS.value:
            self._handle_status(data)
        elif cmd == Cmd.HF_ISO15693_SNIFF.value:
            # Final response (legacy path shouldn't hit here, but safety)
            self._active = False

        return True

    def _handle_trace(self, data: bytes) -> None:
        """Parse and print a trace entry from CMD_HF_SNIFF_STREAM."""
        if len(data) < 8:
            return

        protocol, direction, duration, timestamp = struct.unpack_from("<BBHI", data)
        payload = data[8:]

        entry = {
            "protocol": protocol,
            "direction": direction,
            "duration": duration,
            "timestamp": timestamp,
            "data": payload,
        }
        self._entries.append(entry)

        if self._formatter:
            self._formatter.print(direction, payload)

    def _handle_status(self, data: bytes) -> None:
        """Handle a CMD_HF_SNIFF_STATUS state change."""
        if len(data) < 3:
            return

        state, count = struct.unpack_from("<BH", data)
        self._state = _STATE_NAMES.get(state, "unknown")

        icon = _STATE_ICONS.get(state, "??")
        label = _STATE_LABELS.get(state, "Unknown")
        count_str = f" ({count} frames)" if count > 0 else ""

        msg = f"\n[Snf] {icon} {label}{count_str}\n"
        sys.stdout.write(msg)
        sys.stdout.flush()

        if state == SNIFF_STATE_STOPPED:
            self._active = False

    def _legacy_sniff(self) -> None:
        """Legacy batch mode: block until button, download BigBuf, decode."""
        from ..core.transport import PM3Transport

        print("[Snf] Sniffing ISO 15693... press PM3 button to stop.")

        # Read frames until we get the sniff completion response
        buf = bytearray()
        while True:
            chunk = self._ser.read(self._ser.in_waiting or 1)
            if chunk:
                buf.extend(chunk)

            idx = buf.find(struct.pack("<I", RESP_PREAMBLE_MAGIC))
            if idx < 0:
                continue
            if idx > 0:
                buf = buf[idx:]

            if len(buf) < RESP_PREAMBLE_SIZE:
                continue

            length_ng = struct.unpack_from("<H", buf, 4)[0]
            payload_len = length_ng & 0x7FFF
            frame_len = RESP_PREAMBLE_SIZE + payload_len + RESP_POSTAMBLE_SIZE

            if len(buf) < frame_len:
                continue

            try:
                resp = decode_response_frame(bytes(buf[:frame_len]))
                buf = buf[frame_len:]
                if resp["cmd"] == Cmd.HF_ISO15693_SNIFF.value:
                    break
            except Exception:
                buf = buf[4:]

        print("[Snf] Sniff ended, downloading trace...")

        # Download BigBuf (synchronous, blocking reads)
        # Send DOWNLOAD_BIGBUF command
        from ..core.protocol import PM3_CMD_DATA_SIZE as CHUNK
        dl_frame = encode_ng_frame(Cmd.DOWNLOAD_BIGBUF,
                                   struct.pack("<QQQ", 0, CHUNK, 0)[:24])
        self._ser.write(dl_frame)

        # This is simplified — for legacy mode, we recommend using
        # the async PM3Transport path via pm3.hf.iso15.sniff_decoded()
        print("[Snf] Legacy download not fully implemented in SniffSession.")
        print("[Snf] Use pm3.hf.iso15.sniff_decoded() for legacy batch mode.")

    def stop(self) -> None:
        """End sniff session from Python (sends BREAK_LOOP)."""
        if self._active:
            frame = encode_ng_frame(Cmd.BREAK_LOOP, b"")
            self._ser.write(frame)
            if self._reader_thread:
                self._reader_thread.join(timeout=2.0)
        self._active = False
        self._state = "stopped"

    def close(self) -> None:
        """Stop and close serial port."""
        if self._active:
            self.stop()
        self._ser.close()

    @property
    def state(self) -> str:
        """Current session state: idle, sniffing, paused, stopped."""
        return self._state

    @property
    def entries(self) -> list[dict]:
        """All captured trace entries across pause/resume cycles."""
        return list(self._entries)

    def clear(self) -> None:
        """Clear accumulated entries."""
        self._entries.clear()

Step 4: Run tests

cd /home/work/pm3py && python -m pytest tests/test_sniff_session.py -v

Expected: All pass.

Step 5: Commit

git add pm3py/sniff/session.py tests/test_sniff_session.py pm3py/core/protocol.py
git commit --no-gpg-sign -m "feat: rewrite SniffSession with live streaming and button-toggle"

---

Task 4: Python — Update .pythonstartup.py and sniff/init.py

Files:

• Modify: .pythonstartup.py:57-63 (SNIFF imports), .pythonstartup.py:98-105 (auto-connect)
• Modify: pm3py/sniff/__init__.py

Step 1: Update .pythonstartup.py

In the SNIFF import group (lines 57-63), the SniffSession import stays. The key change is in the auto-connect section.

After the existing if _MODE == "core": block (lines 98-105), add sniff mode auto-session:

    if _MODE == "sniff":
        try:
            session = SniffSession.open()
            print("Sniff session ready -- session.start_15693()")
        except Exception as e:
            print(f"  (Sniff session failed: {e})")

Step 2: Update sniff/init.py exports

No changes needed — SniffSession is already exported. The new constants are in core/protocol.py.

Step 3: Run existing tests to verify nothing broke

cd /home/work/pm3py && python -m pytest tests/ -v

Expected: All existing tests pass.

Step 4: Commit

git add .pythonstartup.py
git commit --no-gpg-sign -m "feat: auto-create sniff session in REPL sniff mode"

---

Task 5: Verify end-to-end with mocked serial

Files:

• Modify: tests/test_sniff_session.py (add integration-style test)

Step 1: Add end-to-end test simulating full REPL workflow

Add to tests/test_sniff_session.py:

class TestSniffSessionEndToEnd:
    """Integration-style test: full start → stream → pause → resume → stop."""

    def test_full_workflow(self):
        """Simulate complete sniff session lifecycle."""
        inv_cmd = bytes([0x26, 0x01, 0x00])
        inv_resp = bytes([0x00, 0xE0, 0x04, 0x12, 0x34, 0x56, 0x78])
        rdbl_cmd = bytes([0x22, 0x20, 0x04])
        rdbl_resp = bytes([0x00, 0xE1, 0x10, 0x06, 0x01])

        frames = [
            _make_status_frame(SNIFF_STATE_STARTED, 0),
            # First burst
            _make_sniff_stream_frame(0, 0, 100, 1000, inv_cmd),
            _make_sniff_stream_frame(0, 1, 200, 1500, inv_resp),
            _make_sniff_stream_frame(0, 0, 100, 2000, rdbl_cmd),
            _make_sniff_stream_frame(0, 1, 200, 2500, rdbl_resp),
            # Pause
            _make_status_frame(SNIFF_STATE_PAUSED, 4),
            # Resume
            _make_status_frame(SNIFF_STATE_RESUMED, 4),
            # Second burst
            _make_sniff_stream_frame(0, 0, 100, 5000, inv_cmd),
            _make_sniff_stream_frame(0, 1, 200, 5500, inv_resp),
            # Stop
            _make_status_frame(SNIFF_STATE_STOPPED, 6),
        ]

        ser = FakeSerial(frames)
        session = SniffSession(ser)
        session.start_15693(live=True)
        time.sleep(0.5)

        assert session.state == "stopped"
        assert len(session.entries) == 6

        # Verify timestamps are monotonically increasing
        timestamps = [e["timestamp"] for e in session.entries]
        assert timestamps == sorted(timestamps)

        # Verify directions alternate (reader/tag pairs)
        directions = [e["direction"] for e in session.entries]
        assert directions == [0, 1, 0, 1, 0, 1]

    def test_close_cleans_up(self):
        """close() stops session and closes serial port."""
        ser = FakeSerial([
            _make_status_frame(SNIFF_STATE_STARTED, 0),
            _make_status_frame(SNIFF_STATE_STOPPED, 0),
        ])

        session = SniffSession(ser)
        session.start_15693(live=True)
        time.sleep(0.1)
        session.close()

        assert session.state == "stopped"
        assert ser._closed

Step 2: Run all tests

cd /home/work/pm3py && python -m pytest tests/test_sniff_session.py -v

Expected: All pass.

Step 3: Run full test suite

cd /home/work/pm3py && python -m pytest tests/ -v

Expected: All pass, no regressions.

Step 4: Commit

git add tests/test_sniff_session.py
git commit --no-gpg-sign -m "test: add end-to-end sniff session workflow tests"

---

Task 6: Firmware build verification

Files:

• No new files — verify firmware compiles

Step 1: Build firmware

cd /home/work/pm3py/firmware && make armsrc/obj/iso15693.o

Expected: Compiles without errors or warnings. If there are compile errors, fix them (likely missing includes or function signature mismatches).

Step 2: Check for any forward declaration mismatches

grep -rn "SniffIso15693" /home/work/pm3py/firmware/armsrc/*.h /home/work/pm3py/firmware/include/

Update any declarations to match the new 5-parameter signature.

Step 3: Commit any fixes

git add firmware/
git commit --no-gpg-sign -m "fix(fw): update SniffIso15693 forward declarations"