feat(lf): NRZ + biphase raw physical-layer demod, with a fallback in identify_lf
Add the two missing ASK line codes so an unrecognised tag yields its raw bits
instead of reading as "nothing":
- ask.nrz_bits: NRZ/direct demod. Keys the bit clock off the shortest run (not
2x a half-bit as in Manchester), resamples, yields both polarities/phases.
- ask.biphase_bits: biphase/differential-Manchester, reusing dsp.biphase_raw_decode
(the FDX-B path).
- capture.demod_raw: runs both, finds the shortest clean repeating frame, returns
{period, bits, hex} per encoding + the recovered clock. Sits BELOW the credential
decoders — it names no format, just hands back the bits.
- identify_lf falls back to demod_raw when no credential decodes off the air, so a
strong-but-unknown LF signal surfaces its frame instead of None.
Hardware-hardened: on a live non-NRZ/biphase tag the first cut reported a spurious
16-bit "frame" — biphase-decoding a non-biphase signal emits a short prefix then a
constant run that trivially repeats at any period. _frame_period now rejects
near-constant streams and finds the smallest *true* period (so 0101 fills read as
period 2 and are rejected, not reported at min_p). Regression tests cover all three.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -17,6 +17,6 @@ Layers:
|
||||
Everything is testable without hardware: the LF transponder models already *encode* these
|
||||
protocols, so a synthetic envelope built from an encoder round-trips through the demod.
|
||||
"""
|
||||
from .capture import demod_samples, read_config, identify_lf
|
||||
from .capture import demod_samples, demod_raw, read_config, identify_lf
|
||||
|
||||
__all__ = ["demod_samples", "read_config", "identify_lf"]
|
||||
__all__ = ["demod_samples", "demod_raw", "read_config", "identify_lf"]
|
||||
|
||||
@@ -127,3 +127,41 @@ def manchester_bits(samples):
|
||||
for phase in (0, 1):
|
||||
for minv in (False, True):
|
||||
yield dsp.manchester_decode(raw, phase, minv)
|
||||
|
||||
|
||||
def nrz_bits(samples):
|
||||
"""Yield candidate raw NRZ / direct bit streams. In NRZ (the T5577 "direct" line code) the
|
||||
data bit is output with no coding — the envelope sampled once per bit *is* the data — so the
|
||||
bit clock is the shortest run itself, not twice a half-bit as in Manchester. Yields both
|
||||
polarities over a few phase offsets; the caller keeps whichever repeats cleanly."""
|
||||
core = trim_to_signal(samples)
|
||||
if len(core) < 64 or dsp.is_flat(core):
|
||||
return
|
||||
binary = dsp.binarize(core)
|
||||
half = dsp.detect_clock(binary)
|
||||
if not half or half < 8:
|
||||
return
|
||||
clk = min(_CLOCKS, key=lambda c: abs(c - half)) # NRZ: one bit = the shortest run
|
||||
for offset in (0.5, 0.0, 0.25, 0.75):
|
||||
raw = dsp.resample(binary, clk, offset)
|
||||
if len(raw) < 32:
|
||||
continue
|
||||
yield raw
|
||||
yield [b ^ 1 for b in raw]
|
||||
|
||||
|
||||
def biphase_bits(samples):
|
||||
"""Yield candidate raw bit streams for a biphase / differential-Manchester ASK envelope — the
|
||||
FDX-B line code (also Indala, Gallagher, …). Resample to half-bit resolution and run the
|
||||
biphase decoder for both polarities; phase-error markers are dropped."""
|
||||
core = trim_to_signal(samples)
|
||||
if len(core) < 64 or dsp.is_flat(core):
|
||||
return
|
||||
binary = dsp.binarize(core)
|
||||
half = dsp.detect_clock(binary)
|
||||
if not half or half < 4:
|
||||
return
|
||||
raw = dsp.resample(binary, half)
|
||||
for invert in (0, 1):
|
||||
bits, _err = dsp.biphase_raw_decode(raw, 0, invert)
|
||||
yield [b for b in bits if b != 7]
|
||||
|
||||
@@ -10,6 +10,7 @@ from __future__ import annotations
|
||||
import inspect
|
||||
|
||||
from . import ask
|
||||
from . import dsp
|
||||
from . import protocols
|
||||
|
||||
# how many envelope samples to pull from BigBuf after a capture (a T55xx/EM frame is a few
|
||||
@@ -53,6 +54,70 @@ def demod_samples(samples) -> dict | None:
|
||||
return None
|
||||
|
||||
|
||||
def _transition_density(bits) -> float:
|
||||
return sum(1 for i in range(1, len(bits)) if bits[i] != bits[i - 1]) / max(1, len(bits) - 1)
|
||||
|
||||
|
||||
def _frame_period(bits, min_p: int = 16) -> int | None:
|
||||
"""The smallest bit-period the stream repeats at (its frame length), or None.
|
||||
|
||||
Guards against the failure mode where the wrong line code decodes a signal into a mostly
|
||||
*constant* run (e.g. biphase-decoding a non-biphase tag emits a short prefix then all-1s):
|
||||
such a stream trivially "repeats" at any period. So we reject a near-constant stream up front,
|
||||
require room for a few real repeats, and reject a frame that is itself near-constant or
|
||||
trivially sub-periodic (an alternating 0101 fill)."""
|
||||
n = len(bits)
|
||||
if n < 3 * min_p:
|
||||
return None
|
||||
if _transition_density(bits) < 0.10: # mostly constant -> not a real frame
|
||||
return None
|
||||
# find the *smallest true* period, from 2 up — so a trivial fill (0101 -> period 2, 0011 ->
|
||||
# period 4) is caught as tiny and rejected, instead of being reported at the first p >= min_p.
|
||||
for p in range(2, n // 3 + 1):
|
||||
span = n - p
|
||||
if sum(1 for i in range(span) if bits[i] == bits[i + p]) >= span * 0.97:
|
||||
return p if p >= min_p else None # real tag frame is >= min_p bits
|
||||
return None
|
||||
|
||||
|
||||
def demod_raw(samples) -> dict | None:
|
||||
"""Physical-layer fallback for an ASK tag no named decoder recognises: recover the raw
|
||||
repeating frame under the NRZ and biphase line codes so the tag still yields its bits.
|
||||
|
||||
Returns whichever encodings produced a clean repeating frame (bit string + hex + period in
|
||||
bits) plus the recovered bit clock, or None if nothing repeats cleanly. This is deliberately
|
||||
*below* the credential decoders — it names no format, it just hands back the bits an expert
|
||||
(or a future format decoder) can read."""
|
||||
core = ask.trim_to_signal(samples)
|
||||
if len(core) < 64 or dsp.is_flat(core):
|
||||
return None
|
||||
half = dsp.detect_clock(dsp.binarize(core))
|
||||
if not half:
|
||||
return None
|
||||
encs: dict = {}
|
||||
for name, gen in (("nrz", ask.nrz_bits), ("biphase", ask.biphase_bits)):
|
||||
best = None
|
||||
for bits in gen(samples):
|
||||
p = _frame_period(bits)
|
||||
if p and (best is None or p < best[0]):
|
||||
best = (p, bits[:p])
|
||||
if best:
|
||||
p, frame = best
|
||||
s = "".join(str(b) for b in frame)
|
||||
encs[name] = {"period": p, "bits": s, "hex": f"{int(s, 2):0{-(-p // 4)}X}"}
|
||||
if not encs:
|
||||
return None
|
||||
clk = min(ask._CLOCKS, key=lambda c: abs(c - half))
|
||||
return {"protocol": "raw", "shortest_run": round(half, 1), "rf": f"RF/{clk}",
|
||||
"encodings": encs}
|
||||
|
||||
|
||||
def raw_label(raw: dict) -> str:
|
||||
"""One-line name for a raw physical-layer dump (no credential decoded)."""
|
||||
parts = [f"{k} {v['hex']} ({v['period']}b)" for k, v in raw["encodings"].items()]
|
||||
return f"unknown LF ({raw['rf']}) — " + "; ".join(parts)
|
||||
|
||||
|
||||
# download a fixed, safe span rather than trusting lf.read()'s reported count — on a flaky NG
|
||||
# link that count is sometimes garbage, and over-reading spills into BigBuf memory past the
|
||||
# samples. 12000 is under a normal acquisition, enough frames for any LF frame to repeat.
|
||||
@@ -159,7 +224,14 @@ def identify_lf(device, emit=None) -> dict | None:
|
||||
config = read_config(device) # T55xx? -> chip + emulation
|
||||
emitted = read_emitted(device) # what's on the air -> EM4100 ID (etc.)
|
||||
|
||||
if config is None and emitted is None:
|
||||
# nothing named decoded off the air -> try the physical-layer fallback (NRZ/biphase) so a tag
|
||||
# no credential decoder handles still yields its raw bits rather than reading as "nothing".
|
||||
raw = None
|
||||
if emitted is None:
|
||||
data = _capture(device, lambda: _try(lambda: device.lf.read(samples=_SAMPLE_COUNT), None))
|
||||
raw = demod_raw(data) if data else None
|
||||
|
||||
if config is None and emitted is None and raw is None:
|
||||
return None
|
||||
|
||||
chip = None
|
||||
@@ -172,15 +244,21 @@ def identify_lf(device, emit=None) -> dict | None:
|
||||
f" -> {emulating}")
|
||||
if emitted is not None:
|
||||
emit(f" emitted: {credential_label(emitted)}")
|
||||
elif raw is not None:
|
||||
emit(f" emitted (raw, no format decoded): {raw_label(raw)}")
|
||||
|
||||
# label: name the chip and what it's actually emitting (the reliable, decoded credential);
|
||||
# fall back to the config's emulation guess, or just the emitted credential.
|
||||
# fall back to the raw physical dump, then the config's emulation guess.
|
||||
if chip and emitted:
|
||||
label = f"{chip} ({credential_label(emitted)})"
|
||||
elif chip and raw:
|
||||
label = f"{chip} ({raw_label(raw)})"
|
||||
elif chip:
|
||||
label = f"{chip} ({emulating})"
|
||||
else:
|
||||
elif emitted:
|
||||
label = credential_label(emitted)
|
||||
else:
|
||||
label = raw_label(raw)
|
||||
|
||||
return {
|
||||
"found": True,
|
||||
@@ -190,5 +268,6 @@ def identify_lf(device, emit=None) -> dict | None:
|
||||
"config": config["config_hex"] if config else None,
|
||||
"emulating": emulating,
|
||||
"emitted": emitted,
|
||||
"raw": raw,
|
||||
"label": label,
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ from unittest.mock import MagicMock
|
||||
|
||||
from pm3py.lf import dsp, fsk, demod_samples, read_config, identify_lf
|
||||
from pm3py.lf import protocols
|
||||
from pm3py.lf import capture
|
||||
from pm3py.transponders.lf.em.em4100 import EM4100Tag
|
||||
from pm3py.transponders.lf.hid.hid import HIDProxTag
|
||||
from pm3py.transponders.lf.atmel.t5577 import T5577Config, CONFIG_EM4100
|
||||
@@ -272,6 +273,71 @@ def _fdxb_env(country, national, animal=0, **kw):
|
||||
return _biphase_env(_fdxb_bits(country, national, animal), **kw)
|
||||
|
||||
|
||||
# --- NRZ / raw physical-layer fallback --------------------------------------------------------
|
||||
|
||||
def _nrz_env(databits, rf=32, high=200, low=60, repeats=8):
|
||||
"""Direct/NRZ ASK envelope: each data bit is output as-is, held for one bit-width (rf
|
||||
samples). The inverse of ask.nrz_bits."""
|
||||
env = []
|
||||
for d in databits * repeats:
|
||||
env += [high if d else low] * rf
|
||||
return bytes(env)
|
||||
|
||||
|
||||
def _cyclic_match(bitstr, frame) -> bool:
|
||||
"""Is ``bitstr`` a rotation of ``frame`` or its bitwise inverse? Raw demod carries polarity
|
||||
and start-of-frame ambiguity, so cyclic-plus-invert is the right equivalence."""
|
||||
f = "".join(str(b) for b in frame)
|
||||
if len(bitstr) != len(f):
|
||||
return False
|
||||
inv = "".join("1" if c == "0" else "0" for c in f)
|
||||
return bitstr in (f + f) or bitstr in (inv + inv)
|
||||
|
||||
|
||||
class TestRawDemod:
|
||||
# a 32-bit frame with no short internal period (so the recovered frame length is 32)
|
||||
FRAME = [1, 0, 1, 1, 0, 0, 1, 0, 1, 1, 1, 0, 0, 0, 1, 0,
|
||||
0, 1, 1, 0, 1, 0, 0, 1, 1, 1, 0, 1, 0, 0, 0, 1]
|
||||
|
||||
def test_nrz_round_trips(self):
|
||||
r = capture.demod_raw(_nrz_env(self.FRAME, rf=32))
|
||||
assert r and "nrz" in r["encodings"]
|
||||
enc = r["encodings"]["nrz"]
|
||||
assert enc["period"] == 32
|
||||
assert _cyclic_match(enc["bits"], self.FRAME)
|
||||
assert r["rf"] == "RF/32"
|
||||
|
||||
def test_nrz_various_rates(self):
|
||||
for rf in (32, 64):
|
||||
r = capture.demod_raw(_nrz_env(self.FRAME, rf=rf))
|
||||
assert r and _cyclic_match(r["encodings"]["nrz"]["bits"], self.FRAME)
|
||||
|
||||
def test_biphase_round_trips(self):
|
||||
r = capture.demod_raw(_biphase_env(self.FRAME, rf=32))
|
||||
assert r and "biphase" in r["encodings"]
|
||||
assert _cyclic_match(r["encodings"]["biphase"]["bits"], self.FRAME)
|
||||
|
||||
def test_noise_returns_none(self):
|
||||
assert capture.demod_raw(bytes([128] * 4000)) is None
|
||||
assert capture.demod_raw(b"") is None
|
||||
|
||||
def test_raw_is_below_credentials(self):
|
||||
# a genuine EM4100 must decode as a credential; the raw fallback is only for the unnamed
|
||||
assert demod_samples(_em4100_env(0x0102030405))["protocol"] == "EM4100"
|
||||
|
||||
def test_generators_yield_bits(self):
|
||||
from pm3py.lf import ask
|
||||
assert any(len(b) >= 32 for b in ask.nrz_bits(_nrz_env(self.FRAME, rf=32)))
|
||||
assert any(len(b) >= 32 for b in ask.biphase_bits(_biphase_env(self.FRAME, rf=32)))
|
||||
|
||||
def test_rejects_constant_tail(self):
|
||||
# hardware regression: biphase-decoding a non-biphase tag emits a short prefix then a long
|
||||
# constant run, which trivially "repeats" at any period. That must NOT report a frame.
|
||||
assert capture._frame_period([0, 0, 0, 1, 0, 1, 0, 1] + [1] * 200) is None
|
||||
assert capture._frame_period([1] * 200) is None # fully constant
|
||||
assert capture._frame_period(([0, 1] * 100)) is None # alternating fill, sub-periodic
|
||||
|
||||
|
||||
class TestCRC16:
|
||||
def test_known_vectors(self):
|
||||
# "123456789" -> CRC-16/XMODEM 0x31C3, CRC-16/KERMIT 0x2189 (anchors the CRC impl)
|
||||
|
||||
Reference in New Issue
Block a user