212 lines
7.4 KiB
Python
212 lines
7.4 KiB
Python
from micropython import const
|
|
import machine
|
|
import utime
|
|
|
|
# from pn532uart import PN532_UART
|
|
# import PN532
|
|
|
|
# PN532 Commands
|
|
_WAKEUP = b'\x55\x55\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
|
|
_COMMAND_GETFIRMWAREVERSION = const(0x02)
|
|
_COMMAND_INLISTPASSIVETARGET = const(0x4A)
|
|
_COMMAND_INRELEASE = const(0x52)
|
|
_COMMAND_SAMCONFIGURATION = const(0x14)
|
|
|
|
# Send Frames
|
|
_PREAMBLE = const(0x00)
|
|
_STARTCODE1 = const(0x00)
|
|
_STARTCODE2 = const(0xFF)
|
|
_POSTAMBLE = const(0x00)
|
|
|
|
# Message parts
|
|
_HOSTTOPN532 = const(0xD4)
|
|
_PN532TOHOST = const(0xD5)
|
|
_ACK = b'\x00\x00\xFF\x00\xFF\x00'
|
|
_FRAME_START = b'\x00\x00\xFF'
|
|
|
|
# Codes
|
|
_MIFARE_ISO14443A = const(0x00)
|
|
|
|
class PN532Uart(object):
|
|
"""
|
|
Class for interacting with the PN532 via the uart interface.
|
|
"""
|
|
def __init__(self, uart_no, tx=None, rx=None, debug=False):
|
|
if tx and rx:
|
|
self.uart = machine.UART(uart_no, baudrate=115200, tx=tx, rx=rx)
|
|
else:
|
|
self.uart = machine.UART(uart_no, baudrate=115200)
|
|
|
|
self.debug = debug
|
|
# self.swriter = asyncio.StreamWriter(self.uart, {})
|
|
# self.sreader = asyncio.StreamReader(self.uart)
|
|
|
|
def wait_read_len(self, len):
|
|
timeout_ms = 1000
|
|
start_time = utime.ticks_ms()
|
|
end_time = start_time + timeout_ms
|
|
while self.uart.any() < len and utime.ticks_ms() < end_time:
|
|
continue
|
|
|
|
if(utime.ticks_ms() >= end_time):
|
|
raise RuntimeError('No response from PN532!')
|
|
|
|
return self.uart.read(len)
|
|
|
|
def _write_frame(self, data):
|
|
"""Write a frame to the PN532 with the specified data bytearray."""
|
|
assert data is not None and 1 < len(data) < 255, 'Data must be array of 1 to 255 bytes.'
|
|
|
|
# Build frame to send as:
|
|
# - Preamble (0x00)
|
|
# - Start code (0x00, 0xFF)
|
|
# - Command length (1 byte)
|
|
# - Command length checksum
|
|
# - Command bytes
|
|
# - Checksum
|
|
# - Postamble (0x00)
|
|
length = len(data)
|
|
frame = bytearray(length+8)
|
|
frame[0] = _PREAMBLE
|
|
frame[1] = _STARTCODE1
|
|
frame[2] = _STARTCODE2
|
|
checksum = sum(frame[0:3])
|
|
frame[3] = length & 0xFF
|
|
frame[4] = (~length + 1) & 0xFF
|
|
frame[5:-2] = data
|
|
checksum += sum(data)
|
|
frame[-2] = ~checksum & 0xFF
|
|
frame[-1] = _POSTAMBLE
|
|
|
|
# Send frame.
|
|
if self.debug:
|
|
print('_write_frame: ', [hex(i) for i in frame])
|
|
|
|
# HACK! Timeouts can cause there to be data in the read buffer that was for an old command (ie read_passive_target).
|
|
# Additonally, the device needs to be woken sometimes, and it seems safe to do that before every command.
|
|
# TODO: Does WAKEUP cancel the passive read_command?
|
|
# Before sending the real command, clear the read buffer
|
|
self.uart.write(_WAKEUP)
|
|
|
|
waiting = self.uart.any()
|
|
while waiting > 0:
|
|
if self.debug:
|
|
print("Removing %d bytes in the read buffer")
|
|
self.uart.read(waiting)
|
|
waiting = self.uart.any()
|
|
|
|
self.uart.write(bytes(frame))
|
|
#self.uart.flush()
|
|
|
|
ack = self.wait_read_len(len(_ACK))
|
|
|
|
if self.debug:
|
|
print('_write_frame: ACK: ', [hex(i) for i in ack])
|
|
if ack != _ACK:
|
|
raise RuntimeError('Did not receive expected ACK from PN532!')
|
|
|
|
def _read_frame(self):
|
|
"""
|
|
Read a response frame from the PN532 and return the data inside the frame,
|
|
otherwise raises an exception if there is an error parsing the frame.
|
|
"""
|
|
# Read the Frame start and header
|
|
response = self.wait_read_len(len(_FRAME_START)+2)
|
|
if self.debug:
|
|
print('_read_frame: frame_start + header:', [hex(i) for i in response])
|
|
|
|
if len(response) < (len(_FRAME_START) + 2) or response[:-2] != _FRAME_START:
|
|
raise RuntimeError('Response does not begin with _FRAME_START!')
|
|
|
|
# Read the header (length & length checksum) and make sure they match.
|
|
frame_len = response[-2]
|
|
frame_checksum = response[-1]
|
|
if (frame_len + frame_checksum) & 0xFF != 0:
|
|
raise RuntimeError('Response length checksum did not match length!')
|
|
|
|
# read the frame (data + data checksum + end frame) & validate
|
|
data = self.wait_read_len(frame_len+2)
|
|
|
|
if self.debug:
|
|
print('_read_frame: data: ', [hex(i) for i in data])
|
|
|
|
checksum = sum(data) & 0xFF
|
|
if checksum != 0:
|
|
raise RuntimeError('Response checksum did not match expected value: ', checksum)
|
|
|
|
if data[-1] != 0x00:
|
|
raise RuntimeError('Response does not include Frame End')
|
|
|
|
# Return frame data.
|
|
return data[0:frame_len]
|
|
|
|
def call_function(self, command, params=[]):
|
|
"""
|
|
Send specified command to the PN532 and return the response.
|
|
Note: There is no timeout option. Use async.wait_for(function(), timeout) instead
|
|
"""
|
|
data = bytearray(2 + len(params))
|
|
data[0] = _HOSTTOPN532
|
|
data[1] = command & 0xFF
|
|
for i, val in enumerate(params):
|
|
data[2+i] = val
|
|
|
|
# Send the frame and read the response
|
|
self._write_frame(data)
|
|
response = self._read_frame()
|
|
|
|
if len(response) < 2:
|
|
raise RuntimeError('Received smaller than expected frame')
|
|
|
|
if not(response[0] == _PN532TOHOST and response[1] == (command+1)):
|
|
raise RuntimeError('Received unexpected command response!')
|
|
|
|
# Return response data.
|
|
return response[2:]
|
|
|
|
def SAM_configuration(self):
|
|
if self.debug:
|
|
print("Sending SAM_CONFIGURATION")
|
|
|
|
response = self.call_function(_COMMAND_SAMCONFIGURATION, params=[0x01, 0x14, 0x01])
|
|
if self.debug:
|
|
print('SAM_configuration:', [hex(i) for i in response])
|
|
|
|
def get_firmware_version(self):
|
|
"""
|
|
Call PN532 GetFirmwareVersion function and return a tuple with the IC,
|
|
Ver, Rev, and Support values.
|
|
"""
|
|
if self.debug:
|
|
print("Sending GET_FIRMWARE_VERSION")
|
|
|
|
response = self.call_function(_COMMAND_GETFIRMWAREVERSION)
|
|
if response is None:
|
|
raise RuntimeError('Failed to detect the PN532')
|
|
return tuple(response)
|
|
|
|
def read_passive_target(self, card_baud=_MIFARE_ISO14443A):
|
|
"""
|
|
Wait for a MiFare card to be available and return its UID when found.
|
|
Will wait up to timeout seconds and return None if no card is found,
|
|
otherwise a bytearray with the UID of the found card is returned.
|
|
"""
|
|
if self.debug:
|
|
print("Sending INIT_PASSIVE_TARGET")
|
|
# Send passive read command for 1 card. Expect at most a 7 byte UUID.
|
|
response = self.call_function(_COMMAND_INLISTPASSIVETARGET, params=[0x01, card_baud])
|
|
|
|
# Check only 1 card with up to a 7 byte UID is present.
|
|
if response[0] != 0x01:
|
|
raise RuntimeError('More than one card detected!')
|
|
if response[5] > 7:
|
|
raise RuntimeError('Found card with unexpectedly long UID!')
|
|
|
|
# Return UID of card.
|
|
return response[6:6+response[5]]
|
|
|
|
def release_targets(self):
|
|
if self.debug:
|
|
print("Release Targets")
|
|
response = self.call_function(_COMMAND_INRELEASE, params=[0x00])
|