Identity IoT: initial release — boot manager, provisioning, sensor/relay
This commit is contained in:
415
pn532.py
Normal file
415
pn532.py
Normal file
@@ -0,0 +1,415 @@
|
||||
# ============================================================+
|
||||
# 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
|
||||
Reference in New Issue
Block a user