fix(pm3py): disable CRC for USB (matching C client), fix CRC byte order for FPC UART
This commit is contained in:
@@ -14,15 +14,20 @@ from .protocol import (
|
||||
)
|
||||
|
||||
|
||||
def encode_ng_frame(cmd: int, payload: bytes = b"", use_crc: bool = True) -> bytes:
|
||||
"""Encode an NG command frame."""
|
||||
def _crc_to_wire(crc: int) -> int:
|
||||
"""Byte-swap CRC to match PM3 wire format: (low << 8) | high."""
|
||||
return ((crc & 0xFF) << 8) | ((crc >> 8) & 0xFF)
|
||||
|
||||
|
||||
def encode_ng_frame(cmd: int, payload: bytes = b"", use_crc: bool = False) -> bytes:
|
||||
"""Encode an NG command frame. use_crc=False for USB (default), True for FPC UART."""
|
||||
if len(payload) > PM3_CMD_DATA_SIZE:
|
||||
raise ValueError(f"Payload too large: {len(payload)} > {PM3_CMD_DATA_SIZE}")
|
||||
length_ng = len(payload) | 0x8000 # set ng bit
|
||||
preamble = struct.pack("<IHH", CMD_PREAMBLE_MAGIC, length_ng, cmd)
|
||||
body = preamble + payload
|
||||
if use_crc:
|
||||
crc = crc16_a(body)
|
||||
crc = _crc_to_wire(crc16_a(body))
|
||||
else:
|
||||
crc = CMD_POSTAMBLE_NOCRC
|
||||
postamble = struct.pack("<H", crc)
|
||||
@@ -30,8 +35,8 @@ def encode_ng_frame(cmd: int, payload: bytes = b"", use_crc: bool = True) -> byt
|
||||
|
||||
|
||||
def encode_mix_frame(cmd: int, arg0: int = 0, arg1: int = 0, arg2: int = 0,
|
||||
payload: bytes = b"", use_crc: bool = True) -> bytes:
|
||||
"""Encode a MIX command frame (NG wire format with OLD-style args)."""
|
||||
payload: bytes = b"", use_crc: bool = False) -> bytes:
|
||||
"""Encode a MIX command frame (NG wire format with OLD-style args). use_crc=False for USB."""
|
||||
args = struct.pack("<QQQ", arg0, arg1, arg2)
|
||||
full_payload = args + payload
|
||||
if len(full_payload) > PM3_CMD_DATA_SIZE:
|
||||
@@ -40,7 +45,7 @@ def encode_mix_frame(cmd: int, arg0: int = 0, arg1: int = 0, arg2: int = 0,
|
||||
preamble = struct.pack("<IHH", CMD_PREAMBLE_MAGIC, length_ng, cmd)
|
||||
body = preamble + full_payload
|
||||
if use_crc:
|
||||
crc = crc16_a(body)
|
||||
crc = _crc_to_wire(crc16_a(body))
|
||||
else:
|
||||
crc = CMD_POSTAMBLE_NOCRC
|
||||
postamble = struct.pack("<H", crc)
|
||||
@@ -67,7 +72,7 @@ def decode_response_frame(data: bytes) -> dict:
|
||||
|
||||
# Verify CRC unless it's the no-CRC magic
|
||||
if crc_received != RESP_POSTAMBLE_NOCRC:
|
||||
crc_expected = crc16_a(data[:RESP_PREAMBLE_SIZE + payload_len])
|
||||
crc_expected = _crc_to_wire(crc16_a(data[:RESP_PREAMBLE_SIZE + payload_len]))
|
||||
if crc_received != crc_expected:
|
||||
raise ValueError(f"CRC mismatch: got 0x{crc_received:04X}, expected 0x{crc_expected:04X}")
|
||||
|
||||
@@ -129,8 +134,20 @@ class PM3Transport:
|
||||
async def connect(self) -> None:
|
||||
import serial_asyncio
|
||||
self._reader, self._writer = await serial_asyncio.open_serial_connection(
|
||||
url=self.port, baudrate=self.baudrate,
|
||||
url=self.port,
|
||||
baudrate=self.baudrate,
|
||||
bytesize=8,
|
||||
parity="N",
|
||||
stopbits=1,
|
||||
xonxoff=False,
|
||||
rtscts=False,
|
||||
dsrdtr=False,
|
||||
)
|
||||
# Match C client: no flow control, flush stale data
|
||||
transport = self._writer.transport
|
||||
if hasattr(transport, 'serial'):
|
||||
transport.serial.reset_input_buffer()
|
||||
transport.serial.reset_output_buffer()
|
||||
|
||||
async def disconnect(self) -> None:
|
||||
if self._writer:
|
||||
|
||||
Reference in New Issue
Block a user