st-ten-1/src/components/rfid_pn532.py

import copy
import ctypes
import sys
import platform
from PyQt5.QtCore import QMutex, Qt, QTimer, pyqtSlot, pyqtSignal
from .component import Component
import ndef
import src.lib.nfc
from src.lib.nfc.clf import RemoteTarget


class RFID_PN532(Component):
    new_id_signal = pyqtSignal(str)

    def __init__(self, config=None, name=None, period=1, lazy=True, paused=False, threaded=True):
        super().__init__(config=config, name=name, period=period, lazy=lazy, paused=paused, threaded=threaded)
        self.data_to_write = None
        self.mutex = QMutex()
        self.simulate="--sim-rfid" in sys.argv
        self.clf = None
        self.connected=False
        self.tag_present=False
        self.current_data=None
        self.is_win = platform.system().lower() == "windows"
        self.dev_list = [f"{'com'if self.is_win else 'tty'}:{self.config['fixture_rfid']['port']}:pn532"]
        self._period = 1

    def open_device(self):
        self.clf = src.lib.nfc.ContactlessFrontend()
        for dev in self.dev_list:
            self.connected = self.clf.open(dev)
            if self.connected:
                self.log.info(f"CONNECTED TO {dev}")
                break
            else:
                self.log.info(f"UNABLE TO CONNECT TO {dev}")

    def close_device(self):
        self.clf.close()

    @pyqtSlot()
    def start(self):
        super().start()

    @pyqtSlot()
    def _get(self):
        if not self.simulate:
            if self.mutex.tryLock():
                try:
                    if not self.connected:
                        self.open_device()
                    else:
                        target = self.clf.sense(RemoteTarget('106A'), RemoteTarget('106B'), RemoteTarget('212F'))
                        if target is not None:
                            tag = src.lib.nfc.tag.activate(self.clf, target)
                            if tag is not None:
                                self.log.debug("tag present")
                                if tag.ndef is not None:
                                    tag_content=tag.ndef.records[0].text
                                    if tag_content!=self.current_data:
                                        self.log.info(f"new tag detected:{tag_content}")
                                        self.current_data=tag_content
                                        self.new_id_signal.emit(self.current_data)
                                else:
                                    self.log.error("tag is not NDEF")
                        else:
                            if self.current_data:
                                self.log.info(f"tag removed:{self.current_data}")
                                self.current_data = None
                                self.new_id_signal.emit(None)
                            self.log.debug("no target present")

                except Exception as e:
                    self.log.info(f"{e}")
                    self.connected = False
                finally:
                    self.mutex.unlock()

    def write_tag(self,data):
        self.data_to_write=copy.deepcopy(data)