utils.nfc

207class NfcReader:
208    """PN532 NFC reader using a duck-typed transport.
209
210    The transport must implement:
211        - is_open: bool property
212        - write_command(data: bytes) -> bool
213        - read_response() -> Optional[bytes]
214    """
215
216    def __init__(self, transport):
217        self._transport = transport
218        self._initialized = False
219
220    @property
221    def is_ready(self) -> bool:
222        return self._transport.is_open and self._initialized
223
224    def begin(self) -> bool:
225        """Initialize the PN532 with SAMConfig (normal mode).
226
227        Returns True if the PN532 responded correctly.
228        """
229        if not self._transport.is_open:
230            return False
231
232        # SAMConfig: normal mode (0x01), timeout 0, no IRQ
233        if not self._transport.write_command(bytes([_CMD_SAM_CONFIG, 0x01, 0x00, 0x00])):
234            logger.warning("NFC SAMConfig: no ACK")
235            return False
236
237        resp = self._transport.read_response()
238        if resp is None or len(resp) < 1 or resp[0] != _RESP_SAM_CONFIG:
239            logger.warning("NFC SAMConfig: bad response")
240            return False
241
242        self._initialized = True
243        logger.info("NFC reader initialized")
244        return True
245
246    def read_passive_target(self) -> Optional[tuple]:
247        """Poll for an ISO14443A tag and return (target_num, uid), or None.
248
249        Returns a tuple of (target_number, uid_bytes) where target_number
250        is needed for subsequent InDataExchange commands.
251        """
252        if not self.is_ready:
253            return None
254
255        # InListPassiveTarget: 1 target, ISO14443A
256        cmd = bytes([_CMD_IN_LIST_PASSIVE_TARGET, 0x01, _MIFARE_ISO14443A])
257        if not self._transport.write_command(cmd):
258            return None
259
260        resp = self._transport.read_response()
261        if resp is None or len(resp) < 2:
262            return None
263
264        if resp[0] != _RESP_IN_LIST_PASSIVE_TARGET:
265            return None
266
267        num_targets = resp[1]
268        if num_targets == 0:
269            return None
270
271        # Parse target data: [tg, sens_res(2), sel_res, nfcid_len, nfcid...]
272        if len(resp) < 7:
273            return None
274
275        tg = resp[2]  # target number assigned by PN532
276        nfcid_len = resp[6]
277        if len(resp) < 7 + nfcid_len:
278            return None
279
280        uid = bytes(resp[7:7 + nfcid_len])
281        logger.debug("NFC target %d detected, UID: %s", tg, uid.hex().upper())
282        return (tg, uid)
283
284    def read_ntag_pages(self, tg: int, start_page: int,
285                        num_pages: int) -> Optional[bytes]:
286        """Read NTAG user data pages.
287
288        Each NTAG READ command returns 4 pages (16 bytes). This method
289        issues multiple reads to cover the requested page range.
290
291        Args:
292            tg: Target number from read_passive_target().
293            start_page: First page number to read (NTAG pages are 4 bytes each).
294            num_pages: Number of pages to read.
295
296        Returns:
297            Concatenated page data, or None on failure.
298        """
299        if not self.is_ready:
300            return None
301
302        result = bytearray()
303        page = start_page
304        remaining = num_pages
305
306        while remaining > 0:
307            data = self._read_4_pages(tg, page)
308            if data is None:
309                return None
310
311            # Each read returns 16 bytes (4 pages). Take only what we need.
312            pages_in_read = min(4, remaining)
313            result.extend(data[:pages_in_read * 4])
314            page += 4
315            remaining -= pages_in_read
316
317        return bytes(result)
318
319    def read_full_tag(self, user_data_start: int = 4,
320                      user_data_pages: int = 36) -> Optional[NfcTagData]:
321        """Read a tag's UID and user data in one operation.
322
323        The UID detection and page reads use the same target listing so
324        the tag must remain on the reader for the entire operation.
325
326        Args:
327            user_data_start: First user data page (default 4 for NTAG21x).
328            user_data_pages: Number of user data pages to read (default 36
329                for NTAG213; use 128 for NTAG215, 226 for NTAG216).
330
331        Returns:
332            NfcTagData with UID and user data, or None on failure.
333        """
334        result = self.read_passive_target()
335        if result is None:
336            return None
337
338        tg, uid = result
339        user_data = self.read_ntag_pages(tg, user_data_start, user_data_pages)
340        if user_data is None:
341            # Tag detected but read failed — caller may want to retry
342            logger.debug("NFC tag %s listed but page read failed", uid.hex().upper())
343            return NfcTagData(uid=uid, user_data=b"")
344
345        return NfcTagData(uid=uid, user_data=user_data)
346
347    def write_ntag_page(self, tg: int, page: int,
348                        data: bytes) -> bool:
349        """Write a single NTAG page (4 bytes).
350
351        Args:
352            tg: Target number from read_passive_target().
353            page: Page number to write.
354            data: Exactly 4 bytes of page data.
355
356        Returns:
357            True on success, False on failure.
358        """
359        if not self.is_ready:
360            return False
361        if len(data) != 4:
362            logger.error("NTAG write: data must be exactly 4 bytes, "
363                         "got %d", len(data))
364            return False
365
366        cmd = bytes([_CMD_IN_DATA_EXCHANGE, tg & 0xFF,
367                     _NTAG_CMD_WRITE, page & 0xFF]) + data
368        if not self._transport.write_command(cmd):
369            logger.debug("NFC page write: no ACK for page %d", page)
370            return False
371
372        resp = self._transport.read_response()
373        if resp is None or len(resp) < 2:
374            logger.debug("NFC page write: no/short response for page %d",
375                         page)
376            return False
377
378        if resp[0] != _RESP_IN_DATA_EXCHANGE:
379            logger.debug("NFC page write: unexpected response code 0x%02X",
380                         resp[0])
381            return False
382
383        if resp[1] != 0x00:
384            logger.debug("NFC page write error at page %d: status 0x%02X",
385                         page, resp[1])
386            return False
387
388        return True
389
390    def write_ntag_pages(self, tg: int, start_page: int,
391                         data: bytes) -> bool:
392        """Write data to consecutive NTAG pages.
393
394        Data is padded to a 4-byte boundary with zeros. Each page (4 bytes)
395        is written individually.
396
397        Args:
398            tg: Target number from read_passive_target().
399            start_page: First page to write.
400            data: Data to write (will be zero-padded to 4-byte boundary).
401
402        Returns:
403            True if all pages written successfully, False on first failure.
404        """
405        if not self.is_ready:
406            return False
407
408        # Pad to 4-byte boundary
409        padded = bytearray(data)
410        while len(padded) % 4 != 0:
411            padded.append(0x00)
412
413        for i in range(0, len(padded), 4):
414            page = start_page + (i // 4)
415            page_data = bytes(padded[i:i + 4])
416            if not self.write_ntag_page(tg, page, page_data):
417                logger.error("NFC write failed at page %d", page)
418                return False
419            logger.debug("NFC wrote page %d: %s", page,
420                         page_data.hex(' '))
421            # NTAG needs ~5ms per write; add brief delay for reliability
422            time.sleep(0.01)
423
424        return True
425
426    def _read_4_pages(self, tg: int, page: int) -> Optional[bytes]:
427        """Issue a single NTAG READ for 4 pages (16 bytes) starting at page."""
428        # InDataExchange: target number, NTAG/Ultralight READ cmd, page address
429        cmd = bytes([_CMD_IN_DATA_EXCHANGE, tg & 0xFF, _NTAG_CMD_READ, page & 0xFF])
430        if not self._transport.write_command(cmd):
431            logger.debug("NFC page read: no ACK for page %d", page)
432            return None
433
434        resp = self._transport.read_response()
435        if resp is None or len(resp) < 2:
436            logger.debug("NFC page read: no/short response for page %d (got %s)",
437                         page, resp.hex(' ') if resp else 'None')
438            return None
439
440        logger.debug("NFC page %d raw response (%d bytes): %s",
441                     page, len(resp), resp.hex(' '))
442
443        if resp[0] != _RESP_IN_DATA_EXCHANGE:
444            logger.debug("NFC page read: unexpected response code 0x%02X", resp[0])
445            return None
446
447        # First byte after command code is status (0x00 = success)
448        status = resp[1]
449        if status != 0x00:
450            logger.debug("NFC page read error at page %d: status 0x%02X", page, status)
451            return None
452
453        # Remaining bytes are the 16 bytes of page data
454        if len(resp) < 18:  # 1 cmd + 1 status + 16 data
455            logger.debug("NFC page read: short response (%d bytes) for page %d",
456                         len(resp), page)
457            return None
458
459        return bytes(resp[2:18])

161class NfcTagData:
162    """Data read from an NFC tag."""
163
164    def __init__(self, uid: bytes, user_data: bytes = b""):
165        self.uid = uid
166        self.user_data = user_data
167
168    @property
169    def uid_hex(self) -> str:
170        return self.uid.hex().upper()
171
172    def get_text_records(self) -> list:
173        """Parse NDEF text records from user data.
174
175        Returns a list of text strings found in NDEF Text records.
176        """
177        return parse_ndef_text_records(self.user_data)
178
179    def get_text(self) -> str:
180        """Get all NDEF text records joined, or raw decode as fallback."""
181        records = self.get_text_records()
182        if records:
183            return '\n'.join(records)
184        # Fallback: raw decode
185        try:
186            return self.user_data.rstrip(b'\x00').decode('utf-8',
187                                                         errors='replace')
188        except Exception:
189            return ""
190
191    def get_uri_records(self) -> list:
192        """Parse NDEF URI records from user data.
193
194        Returns a list of URI strings.
195        """
196        return parse_ndef_uri_records(self.user_data)
197
198    def get_uri(self) -> str:
199        """Get the first NDEF URI record, or empty string."""
200        uris = self.get_uri_records()
201        return uris[0] if uris else ""
202
203    def __repr__(self):
204        return f"NfcTagData(uid={self.uid_hex}, data_len={len(self.user_data)})"

 24class NfcSerialTransport:
 25    """Low-level serial transport for PN532 using HSU (UART) protocol."""
 26
 27    def __init__(self, port: str, baudrate: int = 115200, timeout: float = 0.05):
 28        self._port_name = port
 29        self._baudrate = baudrate
 30        self._timeout = timeout
 31        self._serial: serial.Serial | None = None
 32
 33    def open(self) -> bool:
 34        """Open the serial port and wake the PN532.
 35
 36        Returns True on success, False on failure.
 37        """
 38        try:
 39            self._serial = serial.Serial(
 40                self._port_name,
 41                baudrate=self._baudrate,
 42                timeout=self._timeout,
 43            )
 44            self._wake()
 45            logger.info("NFC transport opened on %s", self._port_name)
 46            return True
 47        except serial.SerialException as e:
 48            logger.warning("Failed to open NFC serial port %s: %s", self._port_name, e)
 49            self._serial = None
 50            return False
 51
 52    def close(self):
 53        """Close the serial port."""
 54        if self._serial and self._serial.is_open:
 55            self._serial.close()
 56            self._serial = None
 57
 58    @property
 59    def is_open(self) -> bool:
 60        return self._serial is not None and self._serial.is_open
 61
 62    def write_command(self, command_data: bytes) -> bool:
 63        """Send a command frame to the PN532.
 64
 65        Args:
 66            command_data: Command bytes (TFI 0xD4 is prepended automatically).
 67
 68        Returns True if the PN532 ACKed the frame.
 69        """
 70        if not self.is_open:
 71            return False
 72
 73        frame = self._build_frame(command_data)
 74        try:
 75            self._serial.reset_input_buffer()
 76            self._serial.write(frame)
 77            return self._read_ack()
 78        except serial.SerialException as e:
 79            logger.warning("NFC write error: %s", e)
 80            return False
 81
 82    def read_response(self) -> bytes | None:
 83        """Read a response frame from the PN532.
 84
 85        Returns the response data (after TFI byte) or None on error/timeout.
 86        """
 87        if not self.is_open:
 88            return None
 89
 90        try:
 91            # Find preamble
 92            if not self._wait_for_preamble():
 93                return None
 94
 95            # Read length and length checksum
 96            header = self._serial.read(2)
 97            if len(header) < 2:
 98                return None
 99
100            length = header[0]
101            lcs = header[1]
102            if (length + lcs) & 0xFF != 0:
103                logger.debug("NFC bad length checksum")
104                return None
105
106            # Read TFI + data + DCS + postamble
107            payload = self._serial.read(length + 2)
108            if len(payload) < length + 2:
109                return None
110
111            tfi_and_data = payload[:length]
112            dcs = payload[length]
113
114            # Verify data checksum
115            if (sum(tfi_and_data) + dcs) & 0xFF != 0:
116                logger.debug("NFC bad data checksum")
117                return None
118
119            # Verify TFI is PN532-to-host
120            if tfi_and_data[0] != _TFI_PN532_TO_HOST:
121                logger.debug("NFC unexpected TFI: 0x%02X", tfi_and_data[0])
122                return None
123
124            # Return data after TFI
125            return bytes(tfi_and_data[1:])
126        except serial.SerialException as e:
127            logger.warning("NFC read error: %s", e)
128            return None
129
130    def _build_frame(self, command_data: bytes) -> bytes:
131        """Build a complete PN532 HSU frame with TFI=0xD4."""
132        data_with_tfi = bytes([_TFI_HOST_TO_PN532]) + command_data
133        length = len(data_with_tfi)
134        lcs = (~length + 1) & 0xFF
135        dcs = (~sum(data_with_tfi) + 1) & 0xFF
136        return _PREAMBLE + bytes([length, lcs]) + data_with_tfi + bytes([dcs]) + _POSTAMBLE
137
138    def _wake(self):
139        """Send wake-up sequence to PN532."""
140        if self._serial:
141            self._serial.write(_WAKEUP)
142            self._serial.flush()
143            # Discard any garbage after wake
144            self._serial.reset_input_buffer()
145
146    def _read_ack(self) -> bool:
147        """Read and verify ACK frame."""
148        ack = self._serial.read(len(_ACK))
149        return ack == _ACK
150
151    def _wait_for_preamble(self) -> bool:
152        """Scan incoming bytes until preamble (00 00 FF) is found."""
153        # Read up to 100 bytes looking for preamble
154        consecutive_zeros = 0
155        for _ in range(100):
156            b = self._serial.read(1)
157            if len(b) == 0:
158                return False
159            if b[0] == 0x00:
160                consecutive_zeros += 1
161            elif b[0] == 0xFF and consecutive_zeros >= 2:
162                return True
163            else:
164                consecutive_zeros = 0
165        return False

43def parse_ndef_text_records(data: bytes) -> list:
44    """Parse NDEF text records from NTAG user data pages.
45
46    Returns a list of decoded text strings.
47    """
48    texts, _ = _parse_ndef_message_tlv(data)
49    return texts

52def parse_ndef_uri_records(data: bytes) -> list:
53    """Parse NDEF URI records from NTAG user data pages.
54
55    Returns a list of decoded URI strings.
56    """
57    _, uris = _parse_ndef_message_tlv(data)
58    return uris

118def build_battery_ndef(sn, year, note='',
119                       uri='http://www.raptacon.org',
120                       max_bytes=NTAG215_MAX_BYTES):
121    """Build a complete NDEF message for a battery tag.
122
123    Creates two records:
124      1. URI record (for phone scanning / lost item recovery)
125      2. Text record with bat: section and key=value data
126
127    Args:
128        sn: Battery serial number.
129        year: Battery year.
130        note: Optional note.
131        uri: URI for the first record (default team website).
132        max_bytes: Maximum tag capacity in bytes.
133
134    Returns:
135        bytes ready to write starting at NTAG page 4.
136
137    Raises:
138        ValueError: If the NDEF message exceeds tag capacity.
139    """
140    records = []
141
142    has_uri = bool(uri)
143
144    # Record 1: URI (MB=1, ME=0 if text follows)
145    if has_uri:
146        records.append(
147            encode_ndef_uri_record(uri, first=True, last=False)
148        )
149
150    # Record 2: Text with bat: section (MB depends on URI, ME=1)
151    text = f"bat:\nsn={sn}\nyear={year}"
152    if note:
153        text += f"\nnote={note}"
154    records.append(
155        encode_ndef_text_record(text, first=not has_uri, last=True)
156    )
157
158    data = wrap_ndef_message(records)
159
160    if len(data) > max_bytes:
161        raise ValueError(
162            f"NDEF message ({len(data)} bytes) exceeds tag capacity "
163            f"({max_bytes} bytes)")
164
165    return data

59def encode_ndef_text_record(text, lang='en', first=False, last=False):
60    """Encode a single NDEF Text record (Well-Known type 'T').
61
62    Args:
63        text: The text content.
64        lang: Language code (default 'en').
65        first: Set MB (Message Begin) flag.
66        last: Set ME (Message End) flag.
67
68    Returns:
69        bytes of the encoded NDEF record (without TLV wrapper).
70    """
71    lang_bytes = lang.encode('ascii')
72    text_bytes = text.encode('utf-8')
73    # Status byte: bit 7 = 0 (UTF-8), bits 5-0 = lang code length
74    status_byte = len(lang_bytes) & 0x3F
75    payload = bytes([status_byte]) + lang_bytes + text_bytes
76
77    # Header: TNF=0x01 (Well-Known), SR=1 (short record)
78    header = 0x11  # SR | TNF=0x01
79    if first:
80        header |= 0x80  # MB
81    if last:
82        header |= 0x40  # ME
83
84    if len(payload) > 255:
85        # Long record (SR=0, 4-byte payload length)
86        header &= ~0x10  # clear SR bit
87        hdr = bytes([header, 0x01]) + len(payload).to_bytes(4, 'big')
88        return hdr + b'T' + payload
89
90    # type = 'T', type_len = 1
91    return bytes([header, 0x01, len(payload), ord('T')]) + payload

26def encode_ndef_uri_record(uri, first=False, last=False):
27    """Encode a single NDEF URI record (Well-Known type 'U').
28
29    Args:
30        uri: Full URI string (e.g. 'http://www.raptacon.org').
31        first: Set MB (Message Begin) flag.
32        last: Set ME (Message End) flag.
33
34    Returns:
35        bytes of the encoded NDEF record (without TLV wrapper).
36    """
37    # Find longest matching prefix
38    prefix_code = 0x00
39    uri_body = uri
40    for prefix, code in _URI_PREFIX_MAP.items():
41        if uri.startswith(prefix):
42            prefix_code = code
43            uri_body = uri[len(prefix):]
44            break
45
46    payload = bytes([prefix_code]) + uri_body.encode('utf-8')
47
48    # Header: TNF=0x01 (Well-Known), SR=1 (short record)
49    header = 0x11  # SR | TNF=0x01
50    if first:
51        header |= 0x80  # MB
52    if last:
53        header |= 0x40  # ME
54
55    # type = 'U', type_len = 1
56    return bytes([header, 0x01, len(payload), ord('U')]) + payload

 94def wrap_ndef_message(records):
 95    """Wrap concatenated NDEF records in an NDEF Message TLV.
 96
 97    Args:
 98        records: list of bytes, each an encoded NDEF record.
 99
100    Returns:
101        bytes with TLV header (0x03, length) + records + terminator (0xFE).
102    """
103    msg = b''.join(records)
104    msg_len = len(msg)
105
106    if msg_len < 0xFF:
107        tlv = bytes([0x03, msg_len]) + msg
108    else:
109        # 3-byte length format
110        tlv = bytes([0x03, 0xFF,
111                     (msg_len >> 8) & 0xFF,
112                     msg_len & 0xFF]) + msg
113
114    # Append terminator TLV
115    return tlv + bytes([0xFE])