fix(hf,hw): fire-and-forget dropfield, full HF_SNIFF struct, standalone payload
dropfield waited on a reply the firmware never sends (bare hf_field_off();break;) -> 2s stall + raise; use send_ng_no_response. hf.sniff under-sent the 10-byte {u32,u32,u8,u8} struct and mis-parsed the uint16 reply. hw.standalone sent an empty payload for a {arg,mlen,mode[10]} struct and blocked; pack it + fire-and-forget. Pre-existing, independent of rebase.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -70,14 +70,30 @@ class HFCommands:
|
||||
results["found"] = len(results) > 0
|
||||
return results
|
||||
|
||||
async def sniff(self, samples: int = 0, skip: int = 0,
|
||||
async def sniff(self, samples_to_skip: int = 0, triggers_to_skip: int = 0,
|
||||
skip_mode: int = 0, skip_ratio: int = 0,
|
||||
on_progress: ProgressCallback = None) -> dict:
|
||||
"""Sniff HF traffic."""
|
||||
payload = struct.pack("<II", samples, skip) if samples else b""
|
||||
"""Sniff HF traffic.
|
||||
|
||||
Firmware reads a 10-byte PACKED struct
|
||||
``{uint32 samplesToSkip; uint32 triggersToSkip; uint8 skipMode;
|
||||
uint8 skipRatio}`` and, when the capture ends, replies with a
|
||||
``{uint16 len}`` sample count. The struct must always be sent whole
|
||||
(matching the C client's ``sizeof(params)``); under-sending leaves
|
||||
the firmware reading uninitialised BigBuf bytes.
|
||||
"""
|
||||
payload = struct.pack("<IIBB", samples_to_skip, triggers_to_skip,
|
||||
skip_mode, skip_ratio)
|
||||
resp = await self._t.send_ng(Cmd.HF_SNIFF, payload, timeout=30.0)
|
||||
return {"status": resp.status, "data": resp.data.hex()}
|
||||
samples = struct.unpack_from("<H", resp.data, 0)[0] if len(resp.data) >= 2 else 0
|
||||
return {"status": resp.status, "samples": samples}
|
||||
|
||||
async def dropfield(self) -> dict:
|
||||
"""Turn off HF field."""
|
||||
resp = await self._t.send_ng(Cmd.HF_DROPFIELD)
|
||||
return {"status": resp.status}
|
||||
"""Turn off HF field.
|
||||
|
||||
The firmware handler is a bare ``hf_field_off(); break;`` with no
|
||||
``reply_ng``, so waiting for a response just blocks until timeout.
|
||||
Fire-and-forget instead.
|
||||
"""
|
||||
await self._t.send_ng_no_response(Cmd.HF_DROPFIELD)
|
||||
return {"status": 0}
|
||||
|
||||
@@ -198,10 +198,22 @@ class HWCommands:
|
||||
"""Send break to stop long-running operations."""
|
||||
await self._t.send_ng_no_response(Cmd.BREAK_LOOP)
|
||||
|
||||
async def standalone(self) -> dict:
|
||||
"""Start standalone mode."""
|
||||
resp = await self._t.send_ng(Cmd.STANDALONE, timeout=5.0)
|
||||
return {"status": resp.status}
|
||||
async def standalone(self, arg: int = 1, mode: str | bytes = b"") -> dict:
|
||||
"""Start standalone mode.
|
||||
|
||||
Firmware casts the payload to a PACKED ``{uint8 arg; uint8 mlen;
|
||||
uint8 mode[10]}``. With ``mlen == 0`` it stashes ``arg`` at
|
||||
``BigBuf_get_EM_addr()[0]``; otherwise it copies ``mode`` (e.g.
|
||||
``"14a"``, ``"15"``, ``"iclass"`` for UniSniff). It then calls
|
||||
``RunMod()`` and never replies, so this is fire-and-forget.
|
||||
"""
|
||||
if isinstance(mode, str):
|
||||
mode = mode.encode("ascii")
|
||||
if len(mode) > 10:
|
||||
raise ValueError("standalone mode string max 10 bytes")
|
||||
payload = struct.pack("<BB", arg & 0xFF, len(mode)) + mode + bytes(10 - len(mode))
|
||||
await self._t.send_ng_no_response(Cmd.STANDALONE, payload)
|
||||
return {"status": 0}
|
||||
|
||||
async def reset(self) -> None:
|
||||
"""Reset the device."""
|
||||
|
||||
@@ -16,3 +16,29 @@ def test_hf_tune():
|
||||
# Firmware requires the START(1)/MEASURE(2)/STOP(3) mode-byte sequence
|
||||
modes = [call.args[1] for call in t.send_ng.call_args_list]
|
||||
assert modes == [bytes([1]), bytes([2]), bytes([3])]
|
||||
|
||||
|
||||
def test_hf_dropfield():
|
||||
t = AsyncMock()
|
||||
hf = HFCommands(t)
|
||||
result = asyncio.get_event_loop().run_until_complete(hf.dropfield())
|
||||
assert result["status"] == 0
|
||||
# Firmware CMD_HF_DROPFIELD sends no reply -> must be fire-and-forget
|
||||
t.send_ng_no_response.assert_awaited_once()
|
||||
assert t.send_ng_no_response.call_args.args[0] == Cmd.HF_DROPFIELD
|
||||
t.send_ng.assert_not_called()
|
||||
|
||||
|
||||
def test_hf_sniff():
|
||||
t = AsyncMock()
|
||||
hf = HFCommands(t)
|
||||
# Firmware replies with a uint16 sample count (1234 = 0x04D2)
|
||||
t.send_ng.return_value = PM3Response(cmd=Cmd.HF_SNIFF, status=0,
|
||||
reason=0, ng=True, data=b"\xd2\x04")
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
hf.sniff(samples_to_skip=5, triggers_to_skip=6, skip_mode=1, skip_ratio=2))
|
||||
assert result["samples"] == 1234
|
||||
assert result["status"] == 0
|
||||
# Always sends the full 10-byte PACKED {u32 u32 u8 u8} struct
|
||||
payload = t.send_ng.call_args.args[1]
|
||||
assert payload == b"\x05\x00\x00\x00\x06\x00\x00\x00\x01\x02"
|
||||
|
||||
@@ -99,3 +99,17 @@ def test_led_pwm_non_capable_raises():
|
||||
with pytest.raises(PM3Error, match="do not support"):
|
||||
asyncio.get_event_loop().run_until_complete(
|
||||
hw.led("blue", brightness=50)) # blue=D, not PWM on Easy
|
||||
|
||||
|
||||
def test_standalone():
|
||||
transport = make_mock_transport()
|
||||
hw = HWCommands(transport)
|
||||
result = asyncio.get_event_loop().run_until_complete(hw.standalone(arg=3, mode="14a"))
|
||||
assert result["status"] == 0
|
||||
# Firmware RunMod() never replies -> fire-and-forget
|
||||
transport.send_ng_no_response.assert_awaited_once()
|
||||
cmd, payload = transport.send_ng_no_response.call_args.args
|
||||
assert cmd == Cmd.STANDALONE
|
||||
# PACKED {uint8 arg; uint8 mlen; uint8 mode[10]} = 12 bytes
|
||||
assert payload == struct.pack("<BB", 3, 3) + b"14a" + bytes(7)
|
||||
transport.send_ng.assert_not_called()
|
||||
|
||||
Reference in New Issue
Block a user