diff --git a/pm3py/lf.py b/pm3py/lf.py index 122c9b7..f75de22 100644 --- a/pm3py/lf.py +++ b/pm3py/lf.py @@ -1,7 +1,7 @@ """Low frequency commands: lf.*""" import struct import asyncio -from .protocol import Cmd +from .protocol import Cmd, PM3_CMD_DATA_SIZE from .transport import PM3Transport, PM3Error, ProgressCallback @@ -158,9 +158,44 @@ class LFCommands: resp = await self._t.send_ng(Cmd.LF_SAMPLING_SET_CONFIG, payload) return {"status": resp.status, **await self.config()} - async def search(self, on_progress: ProgressCallback = None) -> dict: - """Search for LF tags. Reads samples then returns raw data for analysis. - Note: Demodulation/tag identification must be done client-side. + async def download_samples(self, start: int = 0, count: int = 30000, + on_progress: ProgressCallback = None) -> bytes: + """Download raw ADC samples from the device's BigBuf. + + Must call read() or search() first to capture samples. + Returns raw bytes (1 byte per sample, unsigned 0-255). + """ + buf = bytearray() + remaining = count + offset = start + + while remaining > 0: + chunk = min(remaining, PM3_CMD_DATA_SIZE) + resp = await self._t.send_mix( + Cmd.DOWNLOAD_BIGBUF, offset, chunk, 0, timeout=5.0, + ) + # Device responds with CMD_DOWNLOADED_BIGBUF containing the data + dl_resp = await self._t.read_response(timeout=5.0) + if dl_resp.data: + buf.extend(dl_resp.data) + + if on_progress: + ret = on_progress({ + "type": "download", "downloaded": len(buf), "total": count, + }) + if asyncio.iscoroutine(ret): + await ret + + offset += chunk + remaining -= chunk + + return bytes(buf[:count]) + + async def search(self, on_progress: ProgressCallback = None) -> "LFSearchResult": + """Search for LF tags. Captures samples and returns a result with callable next steps. + + Note: Full tag demodulation (EM410x, HID, AWID, etc) requires the C client's + DSP stack. This method captures raw samples and provides tools to work with them. """ if on_progress: ret = on_progress({"type": "lf_search", "phase": "reading"}) @@ -174,7 +209,47 @@ class LFCommands: if asyncio.iscoroutine(ret): await ret - return { - "status": read_result["status"], - "samples_captured": read_result.get("samples", 0), - } + return LFSearchResult( + samples_captured=read_result.get("samples", 0), + status=read_result["status"], + lf=self, + ) + + +class LFSearchResult: + """Result of lf.search() with callable next steps.""" + + def __init__(self, samples_captured: int, status: int, lf: "LFCommands"): + self.samples_captured = samples_captured + self.status = status + self._lf = lf + + async def download(self, on_progress: ProgressCallback = None) -> bytes: + """Download the captured samples from the device buffer.""" + return await self._lf.download_samples( + count=self.samples_captured, on_progress=on_progress, + ) + + async def tune(self, divisor: int = 95) -> dict: + """Measure antenna voltage — use to check if a tag is present.""" + return await self._lf.tune(divisor=divisor) + + async def read_t55xx(self, block: int = 0) -> dict: + """Try reading as a T55xx tag (very common LF chip).""" + return await self._lf.t55.readbl(block=block) + + async def config(self) -> dict: + """Get current LF sampling config.""" + return await self._lf.config() + + def __repr__(self): + if self.samples_captured > 0: + return ( + f"LFSearchResult(samples={self.samples_captured})\n" + f" Samples captured. Next steps:\n" + f" await result.download() — get raw ADC samples as bytes\n" + f" await result.tune() — check antenna voltage (tag present?)\n" + f" await result.read_t55xx() — try reading as T55xx\n" + f" For full tag ID (EM410x, HID, AWID, etc), use the C client: pm3 -c 'lf search'" + ) + return "LFSearchResult(no samples captured — is a tag on the antenna?)"