pm3py/pm3py/lf/dsp.py

"""Modulation-agnostic LF DSP primitives.

Input everywhere is a Proxmark LF *envelope*: one unsigned byte (0-255) per carrier period
(one "field clock"), as returned by ``lf.download_samples``. So a tag clocked at RF/64 spans
64 samples per data bit, RF/32 spans 32, and so on — the RF/n divider *is* the samples-per-bit.

The functions here recover a bit stream from that envelope without knowing the protocol:
threshold to binary, detect the bit clock, then slice/decode (Manchester or biphase). Framing
(headers, parity, credential fields) lives in :mod:`pm3py.lf.protocols`.
"""
from __future__ import annotations

# T55xx standard data-bit-rate set (RF/n) — samples-per-bit candidates when the clock is
# unknown. x-mode allows any even n; the reader still recovers those via the generic detector.
RF_RATES = (8, 16, 32, 40, 50, 64, 100, 128)


def _percentile(sorted_vals: list[int], frac: float) -> float:
    """Value at ``frac`` (0..1) through an already-sorted list; robust to spikes."""
    if not sorted_vals:
        return 0.0
    idx = min(len(sorted_vals) - 1, max(0, round(frac * (len(sorted_vals) - 1))))
    return float(sorted_vals[idx])


def threshold(samples) -> float:
    """A robust high/low decision level for an ASK/OOK envelope.

    The midpoint between the 5th and 95th percentiles, so a few saturated or dropped samples
    don't drag it (a plain min/max would, and the duty cycle isn't reliably 50% so the mean
    can bias). Returns 0.0 for an empty/flat capture."""
    if not samples:
        return 0.0
    s = sorted(samples)
    lo, hi = _percentile(s, 0.05), _percentile(s, 0.95)
    return (lo + hi) / 2.0


def is_flat(samples, min_swing: int = 16) -> bool:
    """True if the capture has too little dynamic range to carry data (no tag / dead field)."""
    if len(samples) < 8:
        return True
    s = sorted(samples)
    return (_percentile(s, 0.95) - _percentile(s, 0.05)) < min_swing


def binarize(samples, level: float | None = None) -> list[int]:
    """Envelope -> list of 0/1 by thresholding (ASK/OOK)."""
    if level is None:
        level = threshold(samples)
    return [1 if v > level else 0 for v in samples]


def _edges(binary: list[int]) -> list[int]:
    return [i for i in range(1, len(binary)) if binary[i] != binary[i - 1]]


def detect_clock(binary: list[int], min_run: int = 3) -> float | None:
    """Estimate the shortest symbol width (samples) from the edge-interval distribution.

    For Manchester this is the *half-bit* (there is always a mid-bit transition, so runs are
    one or two half-bits); for plain OOK it is the bit itself. Runs shorter than ``min_run``
    are treated as noise/jitter and ignored. Returns the mean of the shortest interval cluster,
    or ``None`` when there aren't enough clean edges to decide."""
    edges = _edges(binary)
    if len(edges) < 4:
        return None
    intervals = [edges[i] - edges[i - 1] for i in range(1, len(edges)) if edges[i] - edges[i - 1] >= min_run]
    if len(intervals) < 3:
        return None
    intervals.sort()
    base = intervals[0]
    # average the cluster within 1.5x of the shortest interval -> the fundamental symbol width
    cluster = [d for d in intervals if d <= base * 1.5]
    return sum(cluster) / len(cluster)


def resample(binary: list[int], symbol: float, offset: float = 0.5) -> list[int]:
    """Sample ``binary`` once per ``symbol``-wide window (majority vote), starting at
    ``offset`` symbols in. Turns a per-sample stream into a per-symbol (e.g. per-half-bit)
    stream at the recovered clock."""
    if symbol <= 0:
        return []
    out: list[int] = []
    n = len(binary)
    pos = offset * symbol
    while pos < n:
        lo = max(0, int(pos - symbol / 4))
        hi = min(n, int(pos + symbol / 4) + 1)
        window = binary[lo:hi] or [binary[min(int(pos), n - 1)]]
        out.append(1 if sum(window) * 2 >= len(window) else 0)
        pos += symbol
    return out


def manchester_decode(halfbits: list[int], phase: int = 0, invert: bool = False):
    """Decode a half-bit stream as Manchester: each data bit is a pair of opposite half-bits.

    ``phase`` (0/1) picks the pairing alignment; ``invert`` swaps the polarity convention
    (``10`` -> 1 vs ``01`` -> 1). Pairs that are ``00``/``11`` are clock slips — returned as
    ``None`` markers so callers can resync. Returns the decoded bit list (with ``None`` gaps).
    Callers typically try both phases/polarities and keep the alignment that frames cleanly."""
    bits: list[int | None] = []
    for i in range(phase, len(halfbits) - 1, 2):
        a, b = halfbits[i], halfbits[i + 1]
        if a == b:
            bits.append(None)          # not a valid Manchester symbol -> clock slip
        else:
            bit = 1 if a == 1 else 0   # (1,0) -> 1, (0,1) -> 0
            bits.append(bit ^ 1 if invert else bit)
    return bits


def biphase_decode(halfbits: list[int], phase: int = 0, invert: bool = False):
    """Decode a half-bit stream as biphase (differential Manchester / FDX-B style): a data
    ``0`` has a mid-bit transition (half-bits differ), a data ``1`` does not (half-bits equal).
    ``phase`` aligns the pairing; ``invert`` swaps the 0/1 convention."""
    bits: list[int] = []
    for i in range(phase, len(halfbits) - 1, 2):
        differ = halfbits[i] != halfbits[i + 1]
        bit = 0 if differ else 1
        bits.append(bit ^ 1 if invert else bit)
    return bits


def bits_to_int(bits: list[int]) -> int:
    """MSB-first bit list -> integer."""
    v = 0
    for b in bits:
        v = (v << 1) | (b & 1)
    return v