bambuddy/spoolbuddy/daemon/nfc_reader.py

"""NFC reader wrapper with state machine for tag presence detection."""

import logging
import time
from enum import Enum, auto

logger = logging.getLogger(__name__)

MISS_THRESHOLD = 3  # Consecutive misses before declaring tag removed
ERROR_RECOVERY_THRESHOLD = 10  # Consecutive errors before attempting RF reset


class NFCState(Enum):
    IDLE = auto()
    TAG_PRESENT = auto()


class NFCReader:
    def __init__(self):
        self._nfc = None
        self._state = NFCState.IDLE
        self._current_uid: str | None = None
        self._current_sak: int | None = None
        self._miss_count = 0
        self._ok = False
        self._error_count = 0
        self._poll_count = 0
        self._last_status_log = 0.0

        try:
            from .pn5180 import PN5180

            self._nfc = PN5180()
            self._init_rf()
            self._ok = True
            logger.info("NFC reader initialized")
        except Exception as e:
            logger.warning("NFC not available: %s", e)

    def _init_rf(self):
        """Full RF initialization sequence."""
        self._nfc.reset()
        self._nfc.load_rf_config(0x00, 0x80)
        time.sleep(0.010)
        self._nfc.rf_on()
        time.sleep(0.030)
        self._nfc.set_transceive_mode()

    def _full_reset(self):
        """Full hardware reset + RF init to recover from stuck state."""
        try:
            self._init_rf()
            self._error_count = 0
            logger.info("NFC reader recovered after full reset")
            return True
        except Exception as e:
            logger.warning("NFC full reset failed: %s", e)
            return False

    @property
    def reader_type(self) -> str:
        """Return NFC reader hardware type."""
        return "PN5180" if self._nfc is not None else "Unknown"

    @property
    def connection(self) -> str:
        """Return NFC reader connection type."""
        return "SPI" if self._nfc is not None else "None"

    @property
    def ok(self) -> bool:
        return self._ok

    @property
    def state(self) -> NFCState:
        return self._state

    @property
    def current_uid(self) -> str | None:
        return self._current_uid

    def close(self):
        try:
            self._nfc.rf_off()
            self._nfc.close()
        except Exception:
            pass

    @property
    def current_sak(self) -> int | None:
        return self._current_sak

    def write_ntag(self, data: bytes, data_v2: bytes | None = None) -> tuple[bool, str]:
        """Write raw NDEF bytes to currently present NTAG tag.

        *data* is OpenTag3D v1, which fits every NTAG. *data_v2* (OpenTag3D
        2.004, #3500) is written instead when the tag's Capability Container
        says it has room for it; v2 does not fit an NTAG213.

        Requires tag in TAG_PRESENT state with SAK=0x00.
        Returns (success, message).
        """
        if self._state != NFCState.TAG_PRESENT:
            return False, "No tag present"
        if self._current_sak not in (0x00, 0x04):
            return False, f"Not an NTAG (SAK=0x{self._current_sak:02X})"
        if not self._nfc:
            return False, "NFC reader not available"

        try:
            # Reactivate card before writing
            result = self._nfc.reactivate_card()
            if result is None:
                return False, "Failed to reactivate card for write"

            uid_bytes, sak = result
            if uid_bytes.hex().upper() != self._current_uid:
                return False, "Tag UID changed during write"

            version = "v1"
            if data_v2 is not None:
                # Page 3 is the Capability Container; READ returns pages 3-6.
                try:
                    cc_pages = self._nfc.ntag_read_pages(start_page=3, num_pages=1)
                except Exception as e:
                    logger.warning("Could not read the tag's Capability Container: %s", e)
                    cc_pages = None
                data, version = choose_ndef(cc_pages, data, data_v2)
                # A fresh select before writing, as the write path always had.
                if self._nfc.reactivate_card() is None:
                    return False, "Failed to reactivate card for write"

            # Write starting at page 4
            success = self._nfc.ntag_write_pages(start_page=4, data=data)
            if success:
                logger.info(
                    "NTAG write successful: %d bytes (OpenTag3D %s) to tag %s", len(data), version, self._current_uid
                )
                return True, f"Write successful (OpenTag3D {version})"
            else:
                return False, "Write or verification failed"
        except Exception as e:
            logger.error("NTAG write error: %s", e)
            return False, f"Write error: {e}"

    def poll(self) -> tuple[str, dict | None]:
        """Poll for tag. Returns (event_type, event_data).

        event_type: "none", "tag_detected", "tag_removed"
        """
        self._poll_count += 1

        # Periodic status log (every 60s)
        now = time.monotonic()
        if now - self._last_status_log >= 60.0:
            logger.info(
                "NFC status: state=%s, polls=%d, errors=%d, ok=%s",
                self._state.name,
                self._poll_count,
                self._error_count,
                self._ok,
            )
            self._last_status_log = now

        if self._state == NFCState.IDLE:
            # Full hardware reset before every idle poll. Each activate_type_a()
            # call that returns None corrupts the PN5180 state — subsequent calls
            # silently fail even when a tag is present. Only a full RST pin toggle
            # recovers the reader. ~240ms overhead per poll, giving ~1.8 Hz poll
            # rate which is fine for a spool tag reader.
            try:
                self._init_rf()
            except Exception as e:
                logger.warning("NFC pre-poll reset failed: %s", e)
        else:
            # Tag present: light RF cycle to reset card from ACTIVE back to IDLE
            # state after previous SELECT, so it responds to the next WUPA/REQA.
            try:
                self._nfc.rf_off()
                time.sleep(0.003)
                self._nfc.rf_on()
                time.sleep(0.010)
            except Exception:
                pass  # Will be caught by activate_type_a() error handling below

        try:
            result = self._nfc.activate_type_a()
        except Exception as e:
            self._error_count += 1
            self._ok = False

            if self._error_count == 1:
                logger.warning("NFC poll error: %s", e)
            elif self._error_count == ERROR_RECOVERY_THRESHOLD:
                logger.warning(
                    "NFC reader stuck (%d consecutive errors), attempting recovery...",
                    self._error_count,
                )
                if self._full_reset():
                    return "none", None
                # Reset failed — will keep trying on next threshold
                self._error_count = 0
            elif self._error_count % ERROR_RECOVERY_THRESHOLD == 0:
                logger.warning("NFC recovery attempt #%d", self._error_count // ERROR_RECOVERY_THRESHOLD)
                self._full_reset()

            return "none", None

        # Successful poll — clear error streak
        if self._error_count > 0:
            logger.info("NFC reader recovered after %d errors", self._error_count)
        self._error_count = 0
        self._ok = True

        if result is not None:
            uid_bytes, sak = result
            uid_hex = uid_bytes.hex().upper()
            self._miss_count = 0

            if self._state == NFCState.IDLE:
                self._state = NFCState.TAG_PRESENT
                self._current_uid = uid_hex
                self._current_sak = sak

                # Try reading Bambu tag data
                tray_uuid = None
                raw_blocks = None
                tag_type = "mifare_classic" if sak in (0x08, 0x18) else "ntag" if sak in (0x00, 0x04) else "unknown"

                if sak in (0x08, 0x18):
                    # A reader fault mid-read (a BUSY timeout, an SPI error) must
                    # not escape poll() and stop the loop: report the tag without
                    # its tray UUID, so matching falls back to the UID (#3377)
                    try:
                        blocks = self._nfc.read_bambu_tag(uid_bytes)
                    except Exception as e:
                        logger.warning("Bambu tag read failed for %s: %s", uid_hex, e)
                        blocks = None
                    if blocks:
                        tray_uuid = _extract_tray_uuid(blocks)
                        raw_blocks = _raw_blocks(blocks)

                logger.info("Tag detected: %s (SAK=0x%02X, type=%s)", uid_hex, sak, tag_type)
                return "tag_detected", {
                    "tag_uid": uid_hex,
                    "sak": sak,
                    "tag_type": tag_type,
                    "tray_uuid": tray_uuid,
                    "raw_blocks": raw_blocks,
                }

            # Tag still present — no event
            return "none", None

        # No tag found
        if self._state == NFCState.TAG_PRESENT:
            self._miss_count += 1
            if self._miss_count >= MISS_THRESHOLD:
                old_uid = self._current_uid
                self._state = NFCState.IDLE
                self._current_uid = None
                self._current_sak = None
                self._miss_count = 0
                logger.info("Tag removed: %s", old_uid)
                return "tag_removed", {"tag_uid": old_uid}

        return "none", None


def _extract_tray_uuid(blocks: dict[int, bytes]) -> str | None:
    """Extract tray_uuid from Bambu MIFARE Classic data blocks.

    The tray UUID is the 16 raw bytes of block 9, as 32 uppercase hex chars.
    Both tags of a spool carry the same value, and it is what the AMS reports
    as tray_uuid. Blocks 4-5 hold the filament type ("PLA Matte"), not the UUID:
    reading them made every spool of one type share an id (#984).
    """
    data = blocks.get(9)
    if not data or len(data) < 16:
        return None
    uuid_str = data[:16].hex().upper()
    if uuid_str == "0" * 32:
        return None
    return uuid_str


def choose_ndef(cc_pages: bytes | None, data_v1: bytes, data_v2: bytes) -> tuple[bytes, str]:
    """Pick the OpenTag3D encoding the tag has room for (#3500).

    *cc_pages* starts with page 3, the NDEF Capability Container
    (E1, version, data area size / 8, access). v2 is chosen only when the CC
    reads as NDEF and its data area holds the whole v2 message: an NTAG213
    (size byte 0x12, 144 bytes) gets v1, an NTAG215/216 (0x3E/0x6D) gets v2.
    Anything unreadable gets v1, which fits every NTAG.
    """
    if cc_pages and len(cc_pages) >= 4 and cc_pages[0] == 0xE1 and len(data_v2) <= cc_pages[2] * 8:
        return data_v2, "v2"
    return data_v1, "v1"


def _raw_blocks(blocks: dict[int, bytes]) -> dict[str, str]:
    """Bambu tag blocks as {"block number": "32 uppercase hex chars"}, for the backend to decode (#2685)."""
    return {str(block): data.hex().upper() for block, data in sorted(blocks.items())}