feat(pm3py): LF commands (tune, read, sniff, sim, config, search, t55xx)

pm3py/client.py

@@ -1,6 +1,7 @@
 """Main Proxmark3 client with nested API."""
 from .transport import PM3Transport, PM3Response, ProgressCallback
 from .hw import HWCommands
+from .lf import LFCommands
 
 
 class Proxmark3:
@@ -18,6 +19,7 @@ class Proxmark3:
     def __init__(self, port: str, baudrate: int = 115200):
         self._transport = PM3Transport(port, baudrate)
         self.hw = HWCommands(self._transport)
+        self.lf = LFCommands(self._transport)
 
     async def connect(self) -> None:
         await self._transport.connect()

pm3py/lf.py

@@ -0,0 +1,180 @@
+"""Low frequency commands: lf.*"""
+import struct
+import asyncio
+from .protocol import Cmd
+from .transport import PM3Transport, PM3Error, ProgressCallback
+
+
+class T55xxCommands:
+    """T55xx tag commands: lf.t55."""
+
+    def __init__(self, transport: PM3Transport):
+        self._t = transport
+
+    async def readbl(self, block: int, page: int = 0, password: int | None = None,
+                     downlink_mode: int = 0) -> dict:
+        """Read a T55xx block."""
+        pwd = password if password is not None else 0
+        pwd_mode = 1 if password is not None else 0
+        payload = struct.pack("<IBBBB", pwd, block, page, pwd_mode, downlink_mode)
+        resp = await self._t.send_ng(Cmd.LF_T55XX_READBL, payload, timeout=5.0)
+        return {
+            "status": resp.status,
+            "data": resp.data.hex() if resp.data else None,
+            "raw": resp.data,
+        }
+
+    async def writebl(self, block: int, data: int, page: int = 0,
+                      password: int | None = None, downlink_mode: int = 0) -> dict:
+        """Write a T55xx block."""
+        pwd = password if password is not None else 0
+        pwd_mode = 1 if password is not None else 0
+        payload = struct.pack("<IIBBBB", data, pwd, block, page, pwd_mode, downlink_mode)
+        resp = await self._t.send_ng(Cmd.LF_T55XX_WRITEBL, payload, timeout=5.0)
+        return {"status": resp.status}
+
+    async def wakeup(self, password: int = 0) -> dict:
+        """Wake up T55xx with password."""
+        payload = struct.pack("<I", password)
+        resp = await self._t.send_ng(Cmd.LF_T55XX_WAKEUP, payload)
+        return {"status": resp.status}
+
+    async def config(self) -> dict:
+        """Get T55xx timing configuration."""
+        resp = await self._t.send_ng(Cmd.LF_T55XX_SET_CONFIG)
+        # Response is t55xx_configurations_t: 4 modes x 7 uint16_t each
+        modes = []
+        for i in range(4):
+            offset = i * 14
+            if offset + 14 <= len(resp.data):
+                fields = struct.unpack_from("<HHHHHHH", resp.data, offset)
+                modes.append({
+                    "start_gap": fields[0], "write_gap": fields[1],
+                    "write_0": fields[2], "write_1": fields[3],
+                    "read_gap": fields[4], "write_2": fields[5], "write_3": fields[6],
+                })
+        return {"modes": modes}
+
+
+class LFCommands:
+    """Low frequency commands."""
+
+    def __init__(self, transport: PM3Transport):
+        self._t = transport
+        self.t55 = T55xxCommands(transport)
+
+    async def tune(self, divisor: int = 95,
+                   on_progress: ProgressCallback = None) -> dict:
+        """Measure LF antenna voltage at a specific frequency.
+        divisor: LF frequency divisor (19-255). 95=125kHz, 88=134kHz.
+        Freq = 12000/(divisor+1) kHz.
+        """
+        # Send init
+        payload = struct.pack("<BBB", 1, divisor, 0)
+        await self._t.send_ng(Cmd.MEASURE_ANTENNA_TUNING_LF, payload, timeout=5.0)
+
+        # Poll for result
+        payload = struct.pack("<BBB", 2, divisor, 0)
+        resp = await self._t.send_ng(Cmd.MEASURE_ANTENNA_TUNING_LF, payload, timeout=5.0)
+
+        voltage = struct.unpack_from("<I", resp.data, 0)[0] if len(resp.data) >= 4 else 0
+        freq_khz = round(12000.0 / (divisor + 1), 2)
+
+        result = {
+            "voltage_mV": voltage,
+            "voltage_V": round(voltage / 1000.0, 3),
+            "frequency_kHz": freq_khz,
+            "divisor": divisor,
+        }
+
+        if on_progress:
+            ret = on_progress({"type": "lf_tune", "result": result})
+            if asyncio.iscoroutine(ret):
+                await ret
+
+        return result
+
+    async def read(self, samples: int = 12288, verbose: bool = False,
+                   on_progress: ProgressCallback = None) -> dict:
+        """Read LF ADC samples into device buffer."""
+        val = (samples & 0x3FFFFFFF)
+        if verbose:
+            val |= (1 << 31)
+        payload = struct.pack("<I", val)
+        resp = await self._t.send_ng(Cmd.LF_SNIFF_RAW_ADC, payload, timeout=10.0)
+        return {
+            "status": resp.status,
+            "samples": struct.unpack_from("<I", resp.data, 0)[0] if len(resp.data) >= 4 else 0,
+        }
+
+    async def sniff(self, samples: int = 12288,
+                    on_progress: ProgressCallback = None) -> dict:
+        """Sniff LF signal."""
+        return await self.read(samples=samples, on_progress=on_progress)
+
+    async def sim(self, gap: int = 0, data: bytes = b"",
+                  on_progress: ProgressCallback = None) -> dict:
+        """Simulate LF tag from sample buffer."""
+        payload = struct.pack("<HH", len(data), gap)
+        resp = await self._t.send_ng(Cmd.LF_SIMULATE, payload, timeout=30.0)
+        return {"status": resp.status}
+
+    async def config(self, decimation: int | None = None,
+                     bits_per_sample: int | None = None,
+                     averaging: bool | None = None,
+                     divisor: int | None = None,
+                     trigger_threshold: int | None = None,
+                     samples_to_skip: int | None = None) -> dict:
+        """Get or set LF sampling configuration.
+        Call with no args to get current config.
+        """
+        if all(v is None for v in [decimation, bits_per_sample, averaging,
+                                    divisor, trigger_threshold, samples_to_skip]):
+            # GET config
+            resp = await self._t.send_ng(Cmd.LF_SAMPLING_GET_CONFIG)
+            if len(resp.data) >= 12:
+                dec, bps, avg, div, thresh, skip, verbose = struct.unpack_from("<bbbhhib", resp.data, 0)
+                return {
+                    "decimation": dec,
+                    "bits_per_sample": bps,
+                    "averaging": bool(avg),
+                    "divisor": div,
+                    "frequency_kHz": round(12000.0 / (div + 1), 2) if div > 0 else 0,
+                    "trigger_threshold": thresh,
+                    "samples_to_skip": skip,
+                }
+            return {"raw": resp.data.hex()}
+        else:
+            # SET config -- first get current, then override
+            current = await self.config()
+            payload = struct.pack("<bbbhhib",
+                                  decimation if decimation is not None else current.get("decimation", 1),
+                                  bits_per_sample if bits_per_sample is not None else current.get("bits_per_sample", 8),
+                                  int(averaging) if averaging is not None else int(current.get("averaging", True)),
+                                  divisor if divisor is not None else current.get("divisor", 95),
+                                  trigger_threshold if trigger_threshold is not None else current.get("trigger_threshold", 0),
+                                  samples_to_skip if samples_to_skip is not None else current.get("samples_to_skip", 0),
+                                  0)
+            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.
+        """
+        if on_progress:
+            ret = on_progress({"type": "lf_search", "phase": "reading"})
+            if asyncio.iscoroutine(ret):
+                await ret
+
+        read_result = await self.read(samples=30000, on_progress=on_progress)
+
+        if on_progress:
+            ret = on_progress({"type": "lf_search", "phase": "complete"})
+            if asyncio.iscoroutine(ret):
+                await ret
+
+        return {
+            "status": read_result["status"],
+            "samples_captured": read_result.get("samples", 0),
+        }

tests/test_lf.py

@@ -0,0 +1,27 @@
+import struct
+import asyncio
+from unittest.mock import AsyncMock
+from pm3py.protocol import Cmd
+from pm3py.transport import PM3Response
+from pm3py.lf import LFCommands
+
+def make_response(cmd, status, data):
+    return PM3Response(cmd=cmd, status=status, reason=0, ng=True, data=data)
+
+def test_lf_tune():
+    t = AsyncMock()
+    lf = LFCommands(t)
+    # First call: init, second call: poll returns data
+    t.send_ng.return_value = make_response(Cmd.MEASURE_ANTENNA_TUNING_LF, 0,
+                                           struct.pack("<I", 42000))
+    result = asyncio.get_event_loop().run_until_complete(lf.tune(divisor=95))
+    assert "voltage_mV" in result
+
+def test_lf_config_get():
+    t = AsyncMock()
+    lf = LFCommands(t)
+    payload = struct.pack("<bbbhhib", 1, 8, 1, 95, 128, 0, 0)
+    t.send_ng.return_value = make_response(Cmd.LF_SAMPLING_GET_CONFIG, 0, payload)
+    result = asyncio.get_event_loop().run_until_complete(lf.config())
+    assert "decimation" in result
+    assert "bits_per_sample" in result