pm3py/pm3py/lf/ask.py

"""ASK demodulation — a faithful port of the Proxmark ASK stack (firmware/common/lfdemod.c).

Real LF envelopes are asymmetric: the tag load-modulation gives a sharp fall but the antenna
recovers on a slow RC ramp, and heavily-coupled tags saturate the rails. A naive threshold +
fixed-step resample drifts and bit-slips on that. The C client instead:

1. computeSignalProperties -> high/low/mean (robust middle mean).
2. getHiLo(75) -> clip thresholds 75% of the way mean->rail (25% clip), giving hysteresis.
3. cleanAskRawDemod -> walk rail-to-rail waves, quantising each to a full- or half-clock symbol
   (so the ramp between rails never triggers a false edge, and every wave re-syncs the clock).
4. manrawdecode -> Manchester the raw stream to data bits.

This module ports 1-4. Framing (EM4100 header/parity, T55xx config) stays in protocols.py.
"""
from __future__ import annotations

from . import dsp

_CLOCKS = (8, 16, 32, 40, 50, 64, 100, 128)


def trim_to_signal(samples, win: int = 256, thresh: int = 40):
    """Drop the leading field-settle transient (low-swing) and trailing unfilled zeros — the
    parts of a live capture that carry no modulation and would poison thresholding/clocking."""
    n = len(samples)
    start = 0
    for i in range(0, max(1, n - win), 32):
        w = samples[i:i + win]
        if w and max(w) - min(w) > thresh:
            start = i
            break
    end = n
    while end > start and samples[end - 1] == 0:
        end -= 1
    return samples[start:end]


def signal_props(samples):
    """Port of computeSignalProperties: (high, low, mean) where high/low are the rails and mean
    is the robust mean of the middle 80% (ignoring the top/bottom 10% so rails don't skew it)."""
    if not samples:
        return 0, 0, 0.0
    s = sorted(samples)
    n = len(s)
    lo10 = s[int(n * 0.1)]
    hi90 = s[int(n * 0.9)]
    mid = [v for v in samples if lo10 <= v <= hi90]
    mean = (sum(mid) / len(mid)) if mid else (sum(samples) / n)
    return max(samples), min(samples), mean


def get_hilo(high, low, mean, pct: int = 75):
    """Port of getHiLo: clip the thresholds ``pct``% of the way from the mean to each rail (the
    default 25% clip), so partially-unclipped waves still register as high/low."""
    clip_hi = mean + (high - mean) * pct / 100.0
    clip_lo = mean - (mean - low) * pct / 100.0
    return clip_hi, clip_lo


def detect_clk(samples) -> int:
    """Full-bit clock (samples per data bit), snapped to the T55xx rate set. ``dsp.detect_clock``
    finds the shortest symbol (a Manchester half-bit); the data-bit clock is twice that."""
    half = dsp.detect_clock(dsp.binarize(samples))
    if not half:
        return 0
    return min(_CLOCKS, key=lambda c: abs(c - 2 * half))


def clean_ask_raw_demod(samples, clk: int, high: float, low: float, invert: int = 0):
    """Port of ``cleanAskRawDemod``: envelope -> raw half-bit stream by measuring rail-to-rail
    wave widths (full clock -> 2 like symbols, half clock -> 1), ignoring the ramp between rails.
    Phase errors (over-long waves) become ``7`` markers."""
    out: list[int] = []
    n = len(samples)
    cl_4, cl_2 = clk // 4, clk // 2
    wave_high = True
    pos = 0
    while pos < n and samples[pos] < high:            # getNextHigh
        pos += 1
    smpl = 1
    if cl_2 - cl_4 - 1 < pos <= clk + cl_4 + 1:        # do not skip the first transition
        out.append(invert ^ 1)
    for i in range(pos, n):
        v = samples[i]
        if v >= high and wave_high:
            smpl += 1
        elif v <= low and not wave_high:
            smpl += 1
        elif (v >= high and not wave_high) or (v <= low and wave_high):
            if smpl > clk - cl_4 - 1:                  # full-clock wave -> two like symbols
                if smpl > clk + cl_4 + 1:
                    out.append(7)
                elif wave_high:
                    out += [invert, invert]
                else:
                    out += [invert ^ 1, invert ^ 1]
                wave_high = not wave_high
                smpl = 0
            elif smpl > cl_2 - cl_4 - 1:               # half-clock wave -> one symbol
                if smpl > cl_2 + cl_4 + 1:
                    out.append(7)
                out.append(invert if wave_high else invert ^ 1)
                wave_high = not wave_high
                smpl = 0
            else:
                smpl += 1
        else:
            smpl += 1                                  # in the ramp — haven't reached a rail yet
    return out


def manchester_bits(samples):
    """Yield candidate data-bit streams for an ASK/Manchester envelope, over every raw-invert and
    Manchester phase/polarity alignment — the caller keeps whichever frames + validates cleanly.
    Empty if the capture is noise or won't clock."""
    core = trim_to_signal(samples)
    if len(core) < 64 or dsp.is_flat(core):
        return
    high, low, mean = signal_props(core)
    clip_hi, clip_lo = get_hilo(high, low, mean)
    clk = detect_clk(core)
    if clk < 8:
        return
    for cinv in (0, 1):
        raw = clean_ask_raw_demod(core, clk, clip_hi, clip_lo, cinv)
        raw = [b for b in raw if b != 7]               # drop phase-error markers before pairing
        for phase in (0, 1):
            for minv in (False, True):
                yield dsp.manchester_decode(raw, phase, minv)