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

# Live Sniff Streaming — Design

## Overview

Rewrite sniff workflow from batch (sniff → button stop → download → decode) to live
streaming with button-toggle pause/resume. User types `session.start_15693()` in the
REPL, sees trace output in real-time, toggles sniffing with the PM3 button, and
accumulates entries across pause/resume cycles.

Three changes:
1. **Auto-load** — `.pythonstartup.py` creates a `session` object in sniff mode
2. **Live streaming** — firmware flushes trace entries over USB on RF idle
3. **Button-toggle** — firmware pause/resume within a persistent sniff session

## Protocol

### New command IDs (`pm3_cmd.h`)

| Command | Value | Direction | Purpose |
|---------|-------|-----------|---------|
| `CMD_HF_SNIFF_STREAM` | 0x0337 | FW→Host | Fire-and-forget trace entry (like `SIM_TRACE`) |
| `CMD_HF_SNIFF_STATUS` | 0x0338 | FW→Host | State change: started / paused / resumed / stopped |

### Sniff command payload

Sent with existing `CMD_HF_ISO15693_SNIFF` (and future `CMD_HF_ISO14443A_SNIFF`).
Legacy behavior preserved when `flags=0`.

```c
struct sniff_params_t {
    uint16_t idle_timeout_ms;  // flush threshold (default 400ms for 15693)
    uint8_t  flags;            // bit 0: live streaming (1=stream, 0=legacy BigBuf)
                               // bit 1: button-toggle (1=pause/resume, 0=exit on press)
} PACKED;
```

### Stream trace payload (`CMD_HF_SNIFF_STREAM`)

```c
struct sniff_trace_t {
    uint8_t  protocol;     // 0=15693, 1=14443A
    uint8_t  direction;    // 0=reader->tag, 1=tag->reader
    uint16_t duration;     // frame duration ticks
    uint32_t timestamp;    // us since sniff start
    uint8_t  data[];       // frame bytes (variable length)
} PACKED;
```

### Status payload (`CMD_HF_SNIFF_STATUS`)

```c
struct sniff_status_t {
    uint8_t  state;    // 0=started, 1=paused, 2=resumed, 3=stopped
    uint16_t count;    // total frames captured so far
} PACKED;
```

## Firmware changes

### `iso15693.c` — `SniffIso15693()`

Current function signature gains parameters from the new payload. When
`flags & SNIFF_FLAG_STREAMING` is unset, existing behavior is unchanged.

#### Streaming + button-toggle flow

```
SniffIso15693(idle_timeout_ms, flags):
    if (!(flags & SNIFF_FLAG_STREAMING)):
        // existing behavior, unchanged
        return

    // --- streaming mode ---
    ring_buf[TRACE_RING_SIZE]       // 8-entry ring (same pattern as sim)
    trace_rd = trace_wr = 0
    state = SNIFF_ACTIVE
    frame_count = 0
    reply_ng(CMD_HF_SNIFF_STATUS, {state=0, count=0})   // "started"

    while (state != SNIFF_STOPPED):

        // --- button toggle ---
        if (BUTTON_PRESS()):
            debounce(200ms)
            if (state == SNIFF_ACTIVE):
                TRACE_FLUSH()
                state = SNIFF_PAUSED
                reply_ng(CMD_HF_SNIFF_STATUS, {state=1, count=frame_count})
                LED_C_OFF()
            else:
                state = SNIFF_ACTIVE
                reply_ng(CMD_HF_SNIFF_STATUS, {state=2, count=frame_count})
                LED_C_ON()

        // --- paused: spin ---
        if (state == SNIFF_PAUSED):
            WDT_HIT()
            // check for BREAK_LOOP (below)
            goto check_break

        // --- existing DMA decode logic ---
        // on decoded frame:
            TRACE_PUSH(direction, data, len, timestamp, duration)
            frame_count++
            last_activity = GetTickCount()

        // --- idle flush ---
        if (trace_wr != trace_rd &&
            (GetTickCount() - last_activity) > idle_timeout_ms):
            TRACE_FLUSH()

    check_break:
        // --- Python stop ---
        if (data_available()):
            receive_ng(&rx)
            if (rx.cmd == CMD_BREAK_LOOP):
                TRACE_FLUSH()
                state = SNIFF_STOPPED

    reply_ng(CMD_HF_SNIFF_STATUS, {state=3, count=frame_count})
```

#### TRACE_PUSH / TRACE_FLUSH macros

Reuse the sim pattern. Each `TRACE_FLUSH` entry sent as:

```c
reply_ng(CMD_HF_SNIFF_STREAM, PM3_SUCCESS, &sniff_trace_t, sizeof(hdr) + data_len);
```

Payload is `sniff_trace_t` with protocol=0 (15693), direction, duration, timestamp,
and the raw frame bytes.

#### LED feedback

| State | LED |
|-------|-----|
| Sniffing active | Green ON (Easy=A, RDV4=B) |
| Frame captured | Orange blink (Easy=C, RDV4=A) |
| Paused | Green OFF |

#### Ring buffer sizing

8 entries. At 15693 speeds (~10-20 frames/sec during active reader polling),
8 entries buffer ~400-800ms of traffic. Flush on idle empties before overflow.

### `pm3_cmd.h` — New definitions

```c
#define CMD_HF_SNIFF_STREAM             0x0337
#define CMD_HF_SNIFF_STATUS             0x0338

#define SNIFF_FLAG_STREAMING            0x01
#define SNIFF_FLAG_BUTTON_TOGGLE        0x02

#define SNIFF_STATE_STARTED             0
#define SNIFF_STATE_PAUSED              1
#define SNIFF_STATE_RESUMED             2
#define SNIFF_STATE_STOPPED             3
```

## Python changes

### `sniff/session.py` — `SniffSession` rewrite

Persistent session with background reader thread, mirroring `SimSession`.

```python
class SniffSession:
    def __init__(self, ser):
        self._ser = ser              # raw pyserial (like SimSession)
        self._active = False
        self._entries = []           # accumulated parsed trace entries
        self._reader_thread = None
        self._formatter = TraceFormatter()
        self._state = "idle"         # idle / sniffing / paused / stopped

    @classmethod
    def open(cls, port=None, baudrate=115200):
        """Auto-detect PM3 port and open serial connection."""
        if port is None:
            port = find_pm3_port()
        ser = serial.Serial(port, baudrate, timeout=0.1)
        return cls(ser)

    def start_15693(self, live=True, idle_timeout_ms=400):
        """Start 15693 sniff session.

        live=True:  streaming mode with button-toggle pause/resume (default)
        live=False: legacy BigBuf mode, button exits
        """
        self._protocol = 0
        flags = 0
        if live:
            flags |= SNIFF_FLAG_STREAMING | SNIFF_FLAG_BUTTON_TOGGLE
        payload = struct.pack("<HB", idle_timeout_ms, flags)
        self._send_ng(Cmd.HF_ISO15693_SNIFF, payload)
        self._active = True

        if live:
            self._reader_thread = threading.Thread(
                target=self._stream_reader, daemon=True)
            self._reader_thread.start()
        else:
            # Legacy: block, download, decode (existing behavior)
            self._legacy_sniff()

    def _stream_reader(self):
        """Background thread: read and print trace frames live."""
        buf = bytearray()
        while self._active:
            chunk = self._ser.read(self._ser.in_waiting or 1)
            buf.extend(chunk)
            while self._try_decode_frame(buf):
                pass

    def _try_decode_frame(self, buf):
        """Decode one frame from buffer. Returns True if frame consumed."""
        # Scan for response magic 0x62334d50
        # CMD_HF_SNIFF_STREAM → parse, append to self._entries, print live
        # CMD_HF_SNIFF_STATUS → update self._state, print status line
        # CMD_HF_ISO15693_SNIFF (final) → self._active = False
        ...

    def stop(self):
        """End sniff session from Python (sends BREAK_LOOP)."""
        self._send_ng(Cmd.BREAK_LOOP, b"")
        if self._reader_thread:
            self._reader_thread.join(timeout=2.0)
        self._active = False
        self._state = "stopped"

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

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

    def clear(self):
        """Clear accumulated entries."""
        self._entries.clear()
```

#### Status output

```
[Snf] >> Started
[Snf] Reader -> Tag:  26 01 00              INVENTORY
[Snf] Tag -> Reader:  00 E0 04 ...          INVENTORY RESPONSE
[Snf] || Paused (12 frames)
[Snf] >> Resumed (12 frames)
[Snf] [] Stopped (28 frames)
```

### `.pythonstartup.py` — Auto-load session

In `sniff` mode (and `all` mode), auto-create session:

```python
if _MODE in ("sniff", "all"):
    from pm3py.sniff.session import SniffSession
    # ... other sniff imports ...

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

No `pm3` client created in sniff mode — `session` owns the serial port exclusively,
same pattern as sim mode.

## Protocol-agnostic design

The `sniff_trace_t.protocol` field and `sniff_params_t` payload structure are shared
across protocols. Only 15693 is implemented now. To add 14443-A later:

1. Parse `sniff_params_t` in `SniffIso14443a()`
2. Use same `CMD_HF_SNIFF_STREAM` / `CMD_HF_SNIFF_STATUS` commands
3. Set `protocol=1` in trace entries
4. Python `_stream_reader` dispatches to `decode_14a()` based on protocol field
5. Add `session.start_14a(live=True, idle_timeout_ms=100)` — 14443-A is faster,
   shorter idle timeout appropriate

## Idle timeout rationale

15693 default: **400ms**. A typical inventory+read cycle completes in ~20-30ms. Reader
poll interval is 50-100ms. 400ms = 4 missed polls = reader is definitely gone. Short
enough that trace appears almost instantly after removing the device.

Configurable via `idle_timeout_ms` parameter for different reader behaviors.

## Legacy compatibility

`live=False` preserves existing behavior exactly:
- `flags=0` in payload → firmware takes legacy path (unchanged code)
- Python blocks, waits for button exit, downloads BigBuf, decodes batch
- No background thread, no streaming

## REPL workflow

```
$ source activate sniff
$ python
Sniff session ready -- session.start_15693()

>>> session.start_15693()
[Snf] >> Started
    # hold phone to antenna...
[Snf] Reader -> Tag:  26 01 00              INVENTORY
[Snf] Tag -> Reader:  00 E0 04 ...          INVENTORY RESPONSE
[Snf] Reader -> Tag:  22 20 04 ...          READ SINGLE BLOCK blk=0
[Snf] Tag -> Reader:  00 E1 10 06 01 ...    READ SINGLE BLOCK RESPONSE
    # pull phone away, 400ms later trace flushes
    # press button to pause
[Snf] || Paused (12 frames)
    # examine data in REPL
>>> session.entries[-1]
{'timestamp': 42318, 'direction': 1, 'data': b'\x00\xe1\x10...', ...}
    # press button to resume for more samples
[Snf] >> Resumed (12 frames)
    # done
>>> session.stop()
[Snf] [] Stopped (28 frames)
>>> len(session.entries)
28
```