feat(lf): validate + retry LF capture against a flaky NG link

On this device the NG link is intermittent — a read/download that returns clean samples
one minute comes back as desynced BigBuf memory the next, and lf.read()'s reported count
is sometimes garbage (so sizing the download off it over-reads into memory). Rather than
trust either, capture now:

- downloads a fixed safe span (_DOWNLOAD_BYTES) instead of the flaky read-count,
- validates the result is an envelope, not memory (rejects the 0xDEADBEEF stack canary,
  an embedded PM3 response magic, or a mostly-zero unwritten buffer),
- retries the read+download a few times, giving up cleanly to b'' if the device only
  returns memory (needs a power-cycle / lower debug level).

read_emitted/read_config go through this, so identify recovers from transient desyncs
instead of demodulating garbage. Tests: validator rejects memory patterns; capture retries
past a garbage download to the good one, and gives up when all attempts are garbage.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
michael
2026-07-14 17:41:03 -07:00
parent a06f4d12b8
commit 26da98ede6
2 changed files with 70 additions and 31 deletions

View File

@@ -364,10 +364,29 @@ class TestIdentifyLF:
assert identify_lf(None) is None
class TestAcquiredBytes:
def test_bit_count_to_bytes(self):
from pm3py.lf.capture import _acquired_bytes
assert _acquired_bytes({"samples": 98304}, 15000) == 12288 # 98304 bits / 8 = 12288 samples
assert _acquired_bytes({"samples": 0}, 15000) == 15000 # no count -> fallback
assert _acquired_bytes(None, 15000) == 15000
assert _acquired_bytes({"samples": 100}, 15000) == 512 # clamped up to a floor
class TestCaptureValidation:
def test_looks_like_samples_rejects_memory(self):
from pm3py.lf.capture import _looks_like_samples
env = _em4100_env(0x0102030405)
assert _looks_like_samples(env) # a real envelope passes
assert not _looks_like_samples(b"\xef\xbe\xad\xde" * 500) # DEADBEEF stack canary
assert not _looks_like_samples(b"\x50\x4d\x33\x62" + bytes(500)) # PM3 response magic
assert not _looks_like_samples(bytes(4000)) # all-zero (unwritten) buffer
assert not _looks_like_samples(b"\x80" * 100) # too short
def test_capture_retries_past_garbage(self):
from pm3py.lf.capture import _capture
good = _em4100_env(0x0102030405)
dev = MagicMock()
# first download is desynced memory, second is a real envelope -> retry returns the good one
dev.lf.download_samples.side_effect = [b"\xef\xbe\xad\xde" * 500, good]
dev.lf.read.return_value = {"status": 0}
data = _capture(dev, lambda: dev.lf.read(samples=15000))
assert data == good
def test_capture_gives_up_all_garbage(self):
from pm3py.lf.capture import _capture
dev = MagicMock()
dev.lf.download_samples.return_value = b"\xef\xbe\xad\xde" * 500
dev.lf.read.return_value = {"status": 0}
assert _capture(dev, lambda: dev.lf.read(samples=15000)) == b""