415 lines
14 KiB
Python
415 lines
14 KiB
Python
# ============================================================+
|
|
# Identity IoT — PN532 NFC Driver for MicroPython
|
|
# (c) Copyright : ae Aeonian Engineering Limited - Hong Kong
|
|
# (c) Copyright : WIDE di D. Papa - Naples - Italy
|
|
# ============================================================+
|
|
|
|
import utime
|
|
from machine import Pin, I2C
|
|
|
|
PN532_I2C_ADDRESS = 0x24
|
|
PN532_CMD_GETFIRMWAREVERSION = 0x02
|
|
PN532_CMD_SAMCONFIGURATION = 0x14
|
|
PN532_CMD_INLISTPASSIVETARGET= 0x4A
|
|
PN532_CMD_INDATAEXCHANGE = 0x40
|
|
PN532_CMD_INRELEASE = 0x52
|
|
PN532_HOSTTOPN532 = 0xD4
|
|
PN532_PN532TOHOST = 0xD5
|
|
|
|
|
|
class PN532Error(Exception):
|
|
pass
|
|
|
|
|
|
class PN532_I2C:
|
|
|
|
def __init__(self, i2c: I2C, address: int = PN532_I2C_ADDRESS):
|
|
self._i2c = i2c
|
|
self._addr = address
|
|
|
|
# Wake
|
|
try:
|
|
self._i2c.writeto(self._addr, bytes([0x00]))
|
|
except Exception:
|
|
pass
|
|
utime.sleep_ms(500)
|
|
|
|
fw = self.get_firmware_version()
|
|
if not fw:
|
|
raise PN532Error('PN532 not found or not responding')
|
|
print('[pn532] firmware: IC=%02x Ver=%d Rev=%d' % (fw['ic'], fw['ver'], fw['rev']))
|
|
self._sam_config()
|
|
|
|
def _build_frame(self, cmd: list) -> bytes:
|
|
body = [PN532_HOSTTOPN532] + cmd
|
|
length = len(body)
|
|
lcs = (~length + 1) & 0xFF
|
|
dcs = (~(sum(body) & 0xFF) + 1) & 0xFF
|
|
return bytes([0x00, 0x00, 0xFF, length, lcs] + body + [dcs, 0x00])
|
|
|
|
def _send_command(self, cmd: list):
|
|
frame = self._build_frame(cmd)
|
|
self._i2c.writeto(self._addr, frame)
|
|
|
|
def _wait_ready(self, timeout=1000) -> bool:
|
|
t = timeout
|
|
while t > 0:
|
|
try:
|
|
b = self._i2c.readfrom(self._addr, 1)
|
|
if b[0] == 0x01:
|
|
return True
|
|
except Exception:
|
|
pass
|
|
utime.sleep_ms(10)
|
|
t -= 10
|
|
return False
|
|
|
|
def _read_response(self, length=32, timeout=1000):
|
|
"""Read after command — PN532 sends ACK then response."""
|
|
# Read ACK (6 bytes + status)
|
|
if not self._wait_ready(timeout):
|
|
return None
|
|
self._i2c.readfrom(self._addr, 7) # status + ACK frame
|
|
utime.sleep_ms(10)
|
|
|
|
# Read response
|
|
if not self._wait_ready(timeout):
|
|
return None
|
|
buf = self._i2c.readfrom(self._addr, length + 1)
|
|
|
|
# Parse: skip status byte, find data after header
|
|
# Frame: [status][0x00][0x00][0xFF][len][lcs][TFI][cmd+1][data...][dcs][0x00]
|
|
if len(buf) < 8:
|
|
return None
|
|
data_len = buf[4] # length field
|
|
if data_len < 2:
|
|
return bytes()
|
|
# data starts at index 7 (after status+preamble+start+len+lcs+TFI)
|
|
data = buf[7:7 + data_len - 1]
|
|
return bytes(data)
|
|
|
|
def _sam_config(self):
|
|
self._send_command([PN532_CMD_SAMCONFIGURATION, 0x01, 0x14, 0x01])
|
|
self._read_response(timeout=200)
|
|
|
|
def get_firmware_version(self) -> dict:
|
|
try:
|
|
self._send_command([PN532_CMD_GETFIRMWAREVERSION])
|
|
# First read: ACK frame
|
|
if not self._wait_ready(500):
|
|
return None
|
|
self._i2c.readfrom(self._addr, 7) # status + ACK
|
|
utime.sleep_ms(10)
|
|
# Second read: response frame
|
|
if not self._wait_ready(500):
|
|
return None
|
|
buf = self._i2c.readfrom(self._addr, 14)
|
|
# buf[0]=status, buf[1..3]=preamble, buf[4]=len, buf[5]=lcs
|
|
# buf[6]=TFI(0xD5), buf[7]=cmd(0x03), buf[8..11]=IC,Ver,Rev,Support
|
|
if len(buf) < 12:
|
|
return None
|
|
return {
|
|
'ic': buf[8],
|
|
'ver': buf[9],
|
|
'rev': buf[10],
|
|
'support': buf[11],
|
|
}
|
|
except Exception as e:
|
|
print('[pn532] fw error:', e)
|
|
return None
|
|
|
|
def read_passive_target(self, timeout=1000) -> bytes:
|
|
"""Wait for ISO14443A card. Returns UID bytes or None."""
|
|
self._send_command([PN532_CMD_INLISTPASSIVETARGET, 0x01, 0x00])
|
|
resp = self._read_response(length=32, timeout=timeout)
|
|
if not resp:
|
|
return None
|
|
# resp[0]=cmd_response(0x4B), resp[1]=num_targets, resp[2]=tg
|
|
# resp[3][4]=ATQA, resp[5]=SAK, resp[6]=uid_len, resp[7+]=UID
|
|
if len(resp) < 8:
|
|
return None
|
|
if resp[1] != 0x01:
|
|
return None
|
|
uid_len = resp[6]
|
|
return bytes(resp[7:7 + uid_len])
|
|
|
|
def read_uid(self, timeout=500) -> bytes:
|
|
return self.read_passive_target(timeout=timeout)
|
|
|
|
def read_page(self, page: int) -> bytes:
|
|
self._send_command([PN532_CMD_INDATAEXCHANGE, 0x01, 0x30, page & 0xFF])
|
|
resp = self._read_response(length=16, timeout=500)
|
|
if not resp or len(resp) < 4:
|
|
return None
|
|
return bytes(resp[:4])
|
|
|
|
def write_page(self, page: int, data: bytes) -> bool:
|
|
if len(data) != 4:
|
|
raise PN532Error('write_page: data must be 4 bytes')
|
|
self._send_command([PN532_CMD_INDATAEXCHANGE, 0x01, 0xA2, page & 0xFF] + list(data))
|
|
resp = self._read_response(timeout=500)
|
|
return resp is not None
|
|
|
|
def release_target(self) -> bool:
|
|
try:
|
|
self._send_command([PN532_CMD_INRELEASE, 0x01])
|
|
self._read_response(timeout=200)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def read_ndef(self, timeout=2000) -> str:
|
|
"""
|
|
Read NDEF TextRecord from NTAG.
|
|
Returns text content string or None.
|
|
NDEF TLV starts at page 4 (byte offset 16).
|
|
"""
|
|
# Read pages 4..19 (64 bytes) — enough for our payload
|
|
data = bytearray()
|
|
for page in range(4, 45):
|
|
chunk = self.read_page(page)
|
|
if chunk is None:
|
|
break
|
|
data += chunk
|
|
|
|
if not data:
|
|
return None
|
|
|
|
# Parse TLV — find NDEF TLV (type 0x03)
|
|
i = 0
|
|
while i < len(data):
|
|
t = data[i]; i += 1
|
|
if t == 0xFE: # Terminator TLV
|
|
break
|
|
if t == 0x00: # Null TLV
|
|
continue
|
|
# Length
|
|
if i >= len(data):
|
|
break
|
|
l = data[i]; i += 1
|
|
if l == 0xFF: # 3-byte length
|
|
if i + 2 >= len(data):
|
|
break
|
|
l = (data[i] << 8) | data[i+1]; i += 2
|
|
if t != 0x03: # Skip non-NDEF TLVs
|
|
i += l
|
|
continue
|
|
# NDEF message
|
|
ndef_msg = data[i:i+l]
|
|
break
|
|
else:
|
|
return None
|
|
|
|
if not ndef_msg:
|
|
return None
|
|
|
|
# Parse NDEF record
|
|
# Byte 0: flags (MB|ME|CF|SR|IL|TNF)
|
|
# Byte 1: type length
|
|
# Byte 2: payload length (SR=1) or 4 bytes (SR=0)
|
|
# Byte 3+: type, id?, payload
|
|
if len(ndef_msg) < 4:
|
|
return None
|
|
|
|
flags = ndef_msg[0]
|
|
type_len = ndef_msg[1]
|
|
sr = (flags >> 4) & 1 # Short Record
|
|
il = (flags >> 3) & 1 # ID Length present
|
|
|
|
idx = 2
|
|
if sr:
|
|
payload_len = ndef_msg[idx]; idx += 1
|
|
else:
|
|
if idx + 4 > len(ndef_msg): return None
|
|
payload_len = (ndef_msg[idx]<<24 | ndef_msg[idx+1]<<16 |
|
|
ndef_msg[idx+2]<<8 | ndef_msg[idx+3]); idx += 4
|
|
|
|
if il:
|
|
id_len = ndef_msg[idx]; idx += 1
|
|
idx += id_len
|
|
|
|
# Skip type bytes
|
|
idx += type_len
|
|
|
|
if idx + payload_len > len(ndef_msg):
|
|
return None
|
|
|
|
payload = ndef_msg[idx:idx+payload_len]
|
|
|
|
# TextRecord: first byte = status (lang length), then lang, then text
|
|
if len(payload) < 1:
|
|
return None
|
|
lang_len = payload[0] & 0x3F
|
|
text_start = 1 + lang_len
|
|
if text_start > len(payload):
|
|
return None
|
|
|
|
try:
|
|
return bytes(payload[text_start:]).decode('utf-8')
|
|
except Exception:
|
|
return None
|
|
|
|
# =========================================================================
|
|
# ISO 7816 / NTAG424 DNA NDEF read
|
|
# =========================================================================
|
|
|
|
def _apdu(self, apdu: list, timeout=500, resp_len=64):
|
|
"""Send ISO 7816 APDU via InDataExchange and return response bytes."""
|
|
self._send_command([PN532_CMD_INDATAEXCHANGE, 0x01] + apdu)
|
|
# Step 1: wait ready, read ACK
|
|
if not self._wait_ready(timeout):
|
|
return None
|
|
self._i2c.readfrom(self._addr, 7) # ACK frame
|
|
utime.sleep_ms(20)
|
|
# Step 2: wait ready, read response
|
|
if not self._wait_ready(timeout):
|
|
return None
|
|
buf = self._i2c.readfrom(self._addr, resp_len + 8)
|
|
# buf: [status][preamble x3][len][lcs][TFI=0xD5][cmd=0x41][err][data...][SW1][SW2][dcs][0x00]
|
|
# data after err byte: index 9, length = buf[4]-3 (TFI+cmd+err)
|
|
if len(buf) < 10:
|
|
return None
|
|
data_len = buf[4]
|
|
if data_len < 3:
|
|
return None
|
|
err = buf[8]
|
|
if err != 0x00:
|
|
print('[pn532] apdu err: %02x' % err)
|
|
return None
|
|
# data = everything from index 9 to 9+(data_len-3), then SW1 SW2
|
|
data = bytes(buf[9:9 + data_len - 2])
|
|
return data
|
|
|
|
def _sw(self, resp) -> tuple:
|
|
"""Extract SW1 SW2 — last 2 bytes of clean APDU response."""
|
|
if not resp or len(resp) < 2:
|
|
return (0x00, 0x00)
|
|
return (resp[-2], resp[-1])
|
|
|
|
def _sw2(self, resp) -> tuple:
|
|
"""Extract SW1 SW2 — first 2 bytes when response is SW only."""
|
|
if not resp or len(resp) < 2:
|
|
return (0x00, 0x00)
|
|
return (resp[0], resp[1])
|
|
|
|
def read_ndef_iso(self, timeout=2000) -> str:
|
|
"""
|
|
Read NDEF from NTAG424 DNA via ISO 7816 command set.
|
|
No authentication required for default NDEF file.
|
|
Returns text content or None.
|
|
Must be called while tag is in field — polls for tag first.
|
|
"""
|
|
# 0. Poll for tag — activate it for InDataExchange
|
|
uid = self.read_passive_target(timeout=timeout)
|
|
if uid is None:
|
|
return None, None, None
|
|
utime.sleep_ms(100)
|
|
|
|
# 1. Select NDEF Application (D2 76 00 00 85 01 01)
|
|
sel_app = [0x00, 0xA4, 0x04, 0x00, 0x07,
|
|
0xD2, 0x76, 0x00, 0x00, 0x85, 0x01, 0x01, 0x00]
|
|
resp = self._apdu(sel_app)
|
|
if not resp:
|
|
return None, None
|
|
sw1, sw2 = resp[0], resp[1]
|
|
if sw1 != 0x90:
|
|
print('[pn532] select app failed: %02x %02x' % (sw1, sw2))
|
|
return None, None
|
|
|
|
# 2. Select NDEF file (E1 04)
|
|
sel_file = [0x00, 0xA4, 0x00, 0x0C, 0x02, 0xE1, 0x04]
|
|
resp = self._apdu(sel_file)
|
|
if not resp:
|
|
return None, None
|
|
sw1, sw2 = resp[0], resp[1]
|
|
if sw1 != 0x90:
|
|
print('[pn532] select file failed: %02x %02x' % (sw1, sw2))
|
|
return None, None
|
|
|
|
# 3. Read first 2 bytes to get NDEF length
|
|
read_len = [0x00, 0xB0, 0x00, 0x00, 0x02]
|
|
resp = self._apdu(read_len)
|
|
if not resp or len(resp) < 4:
|
|
return None, None
|
|
# resp = [len_hi][len_lo][SW1][SW2]
|
|
if len(resp) < 4:
|
|
return None, None
|
|
sw1, sw2 = resp[2], resp[3]
|
|
if sw1 != 0x90:
|
|
print('[pn532] read len failed: %02x %02x' % (sw1, sw2))
|
|
return None, None
|
|
ndef_len = (resp[0] << 8) | resp[1]
|
|
if ndef_len == 0:
|
|
return None, None
|
|
|
|
# 4. Read NDEF data (max 200 bytes per read)
|
|
ndef_data = bytearray()
|
|
offset = 2
|
|
remaining = ndef_len
|
|
|
|
while remaining > 0:
|
|
chunk = min(remaining, 50) # PN532 I2C frame limit
|
|
read_data = [0x00, 0xB0,
|
|
(offset >> 8) & 0xFF,
|
|
offset & 0xFF,
|
|
chunk & 0xFF]
|
|
resp = self._apdu(read_data, timeout=timeout, resp_len=chunk+3)
|
|
if not resp:
|
|
break
|
|
if len(resp) < 2:
|
|
break
|
|
sw1, sw2 = resp[-2], resp[-1]
|
|
if sw1 != 0x90:
|
|
print('[pn532] read data failed: %02x %02x' % (sw1, sw2))
|
|
break
|
|
ndef_data += resp[:-2] # exclude SW1 SW2
|
|
offset += chunk
|
|
remaining -= chunk
|
|
|
|
if not ndef_data:
|
|
return None, None
|
|
|
|
# 5. Parse NDEF message
|
|
# ndef_data = clean NDEF bytes, no response code prefix
|
|
if len(ndef_data) < 3:
|
|
return None, None
|
|
|
|
flags = ndef_data[0]
|
|
type_len = ndef_data[1]
|
|
sr = (flags >> 4) & 1
|
|
il = (flags >> 3) & 1
|
|
|
|
idx = 2
|
|
if sr:
|
|
if idx >= len(ndef_data): return None
|
|
payload_len = ndef_data[idx]; idx += 1
|
|
else:
|
|
if idx + 4 > len(ndef_data): return None
|
|
payload_len = (ndef_data[idx]<<24 | ndef_data[idx+1]<<16 |
|
|
ndef_data[idx+2]<<8 | ndef_data[idx+3]); idx += 4
|
|
|
|
if il:
|
|
id_len = ndef_data[idx]; idx += 1
|
|
idx += id_len
|
|
|
|
idx += type_len # skip type bytes
|
|
|
|
if idx + payload_len > len(ndef_data):
|
|
return None, None
|
|
|
|
payload = ndef_data[idx:idx+payload_len]
|
|
|
|
# TextRecord: byte 0 = status (lang_len), then lang, then text
|
|
if len(payload) < 1:
|
|
return None, None
|
|
lang_len = payload[0] & 0x3F
|
|
text_start = 1 + lang_len
|
|
if text_start > len(payload):
|
|
return None, None
|
|
|
|
try:
|
|
import ubinascii as _ub
|
|
uid_hex = _ub.hexlify(uid).decode()
|
|
return uid_hex, bytes(payload[text_start:]).decode('utf-8')
|
|
except Exception:
|
|
return None, None |