fix(lf): correct T55xx write/wakeup structs, 2-byte LF tune, fire-and-forget sampling config
lf.tune sends the 2-byte {mode,divisor} the firmware requires (was 3 -> EINVARG/0V). t55.writebl packs the 10-byte {data,pwd,blockno,flags} struct with a single bit-packed flags byte (was 12). t55.wakeup packs 5 bytes {pwd,flags}. t55.config is maintained client-side (firmware has no GET). LF_SAMPLING_SET_CONFIG uses send_ng_no_response (firmware never replies). Pre-existing, independent of the rebase.
This commit is contained in:
michael
@@ -28,32 +28,49 @@ class T55xxCommands:
|
||||
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)
|
||||
# Firmware t55xx_write_block_t: data(4)+pwd(4)+blockno(1)+flags(1).
|
||||
# flags: 0x01 PwdMode, 0x02 Page, 0x04 test, 0x18 downlink_mode<<3.
|
||||
flags = 0
|
||||
if password is not None:
|
||||
flags |= 0x01
|
||||
if page:
|
||||
flags |= 0x02
|
||||
flags |= (downlink_mode & 0x03) << 3
|
||||
payload = struct.pack("<IIBB", data, pwd, block, flags)
|
||||
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:
|
||||
async def wakeup(self, password: int = 0, downlink_mode: int = 0) -> dict:
|
||||
"""Wake up T55xx with password."""
|
||||
payload = struct.pack("<I", password)
|
||||
# Firmware struct {uint32 password; uint8 flags}; flags = downlink_mode<<3.
|
||||
flags = (downlink_mode & 0x03) << 3
|
||||
payload = struct.pack("<IB", password, flags)
|
||||
resp = await self._t.send_ng(Cmd.LF_T55XX_WAKEUP, payload)
|
||||
return {"status": resp.status}
|
||||
|
||||
# Default T55xx downlink-mode timing table (generic/PM3 Easy build), stored
|
||||
# as raw struct values (field-clocks x 8). Mirrors T55xx_Timing in
|
||||
# armsrc/lfops.c. Firmware exposes no GET, and CMD_LF_T55XX_SET_CONFIG is
|
||||
# write-only (never replies), so this config is maintained client-side.
|
||||
_DEFAULT_MODES = (
|
||||
{"start_gap": 248, "write_gap": 160, "write_0": 144, "write_1": 400,
|
||||
"read_gap": 120, "write_2": 0, "write_3": 0}, # Fixed
|
||||
{"start_gap": 248, "write_gap": 160, "write_0": 144, "write_1": 400,
|
||||
"read_gap": 120, "write_2": 0, "write_3": 0}, # Long Leading Ref.
|
||||
{"start_gap": 248, "write_gap": 160, "write_0": 144, "write_1": 320,
|
||||
"read_gap": 120, "write_2": 0, "write_3": 0}, # Leading 0
|
||||
{"start_gap": 248, "write_gap": 160, "write_0": 144, "write_1": 272,
|
||||
"read_gap": 120, "write_2": 400, "write_3": 528}, # 1 of 4
|
||||
)
|
||||
|
||||
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}
|
||||
"""Return T55xx downlink timing configuration.
|
||||
|
||||
The firmware exposes no GET for the T55xx timing table, and
|
||||
CMD_LF_T55XX_SET_CONFIG is write-only (never replies), so the default
|
||||
table is maintained client-side.
|
||||
"""
|
||||
return {"modes": [dict(m) for m in self._DEFAULT_MODES]}
|
||||
|
||||
|
||||
class LFCommands:
|
||||
@@ -69,12 +86,12 @@ class LFCommands:
|
||||
divisor: LF frequency divisor (19-255). 95=125kHz, 88=134kHz.
|
||||
Freq = 12000/(divisor+1) kHz.
|
||||
"""
|
||||
# Send init
|
||||
payload = struct.pack("<BBB", 1, divisor, 0)
|
||||
# Send init — firmware requires length == 2 ({mode, divisor})
|
||||
payload = struct.pack("<BB", 1, divisor)
|
||||
await self._t.send_ng(Cmd.MEASURE_ANTENNA_TUNING_LF, payload, timeout=5.0)
|
||||
|
||||
# Poll for result
|
||||
payload = struct.pack("<BBB", 2, divisor, 0)
|
||||
# Measure — reply carries the antenna voltage as a uint32 (mV)
|
||||
payload = struct.pack("<BB", 2, divisor)
|
||||
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
|
||||
@@ -155,8 +172,9 @@ class LFCommands:
|
||||
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()}
|
||||
# Firmware never replies to LF_SAMPLING_SET_CONFIG — fire-and-forget
|
||||
await self._t.send_ng_no_response(Cmd.LF_SAMPLING_SET_CONFIG, payload)
|
||||
return {"status": 0, **await self.config()}
|
||||
|
||||
async def download_samples(self, start: int = 0, count: int = 30000,
|
||||
on_progress: ProgressCallback = None) -> bytes:
|
||||
|
||||
@@ -11,11 +11,15 @@ def make_response(cmd, status, data):
|
||||
def test_lf_tune():
|
||||
t = AsyncMock()
|
||||
lf = LFCommands(t)
|
||||
# First call: init, second call: poll returns data
|
||||
# First call: init, second call: measure returns uint32 voltage
|
||||
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
|
||||
assert result["voltage_mV"] == 42000
|
||||
# Firmware requires exactly 2-byte payloads ({1,div} init, {2,div} measure)
|
||||
calls = t.send_ng.call_args_list
|
||||
assert calls[0].args[1] == struct.pack("<BB", 1, 95)
|
||||
assert calls[1].args[1] == struct.pack("<BB", 2, 95)
|
||||
|
||||
def test_lf_config_get():
|
||||
t = AsyncMock()
|
||||
@@ -25,3 +29,50 @@ def test_lf_config_get():
|
||||
result = asyncio.get_event_loop().run_until_complete(lf.config())
|
||||
assert "decimation" in result
|
||||
assert "bits_per_sample" in result
|
||||
|
||||
def test_lf_config_set_fire_and_forget():
|
||||
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)
|
||||
asyncio.get_event_loop().run_until_complete(lf.config(divisor=88))
|
||||
# Firmware never replies to SET — must be fire-and-forget
|
||||
t.send_ng_no_response.assert_awaited_once()
|
||||
assert t.send_ng_no_response.call_args.args[0] == Cmd.LF_SAMPLING_SET_CONFIG
|
||||
|
||||
def test_lf_t55_writebl():
|
||||
t = AsyncMock()
|
||||
lf = LFCommands(t)
|
||||
t.send_ng.return_value = make_response(Cmd.LF_T55XX_WRITEBL, 0, b"")
|
||||
asyncio.get_event_loop().run_until_complete(
|
||||
lf.t55.writebl(block=4, data=0x12345678, page=1,
|
||||
password=0xAABBCCDD, downlink_mode=2))
|
||||
payload = t.send_ng.call_args.args[1]
|
||||
# data(4)+pwd(4)+blockno(1)+flags(1) = 10 bytes
|
||||
assert len(payload) == 10
|
||||
data, pwd, blockno, flags = struct.unpack("<IIBB", payload)
|
||||
assert data == 0x12345678
|
||||
assert pwd == 0xAABBCCDD
|
||||
assert blockno == 4
|
||||
# PwdMode(0x01) | Page(0x02) | downlink 2<<3 (0x10) = 0x13
|
||||
assert flags == 0x13
|
||||
|
||||
def test_lf_t55_wakeup():
|
||||
t = AsyncMock()
|
||||
lf = LFCommands(t)
|
||||
t.send_ng.return_value = make_response(Cmd.LF_T55XX_WAKEUP, 0, b"")
|
||||
asyncio.get_event_loop().run_until_complete(
|
||||
lf.t55.wakeup(password=0x11223344, downlink_mode=3))
|
||||
payload = t.send_ng.call_args.args[1]
|
||||
assert len(payload) == 5
|
||||
pwd, flags = struct.unpack("<IB", payload)
|
||||
assert pwd == 0x11223344
|
||||
assert flags == (3 << 3)
|
||||
|
||||
def test_lf_t55_config_client_side():
|
||||
t = AsyncMock()
|
||||
lf = LFCommands(t)
|
||||
result = asyncio.get_event_loop().run_until_complete(lf.t55.config())
|
||||
assert len(result["modes"]) == 4
|
||||
# No firmware round-trip: SET_CONFIG is write-only and never replies
|
||||
t.send_ng.assert_not_called()
|
||||
|
||||
Reference in New Issue
Block a user