124 lines
5.1 KiB
Python
124 lines
5.1 KiB
Python
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)
|
|
rfid_error_signal = pyqtSignal(bool)
|
|
|
|
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()
|
|
if self.connected:
|
|
self.rfid_error_signal.emit(True)
|
|
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)
|
|
self.rfid_error_signal.emit(self.connected)
|
|
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.rfid_error_signal.emit(False)
|
|
self.log.debug("no target present")
|
|
self.rfid_error_signal.emit(self.connected)
|
|
|
|
except Exception as e:
|
|
self.log.info(f"{e}")
|
|
self.connected = False
|
|
finally:
|
|
if not self.connected:
|
|
self.close_device()
|
|
self.rfid_error_signal.emit(False)
|
|
self.mutex.unlock()
|
|
|
|
def write_tag(self, data):
|
|
self.data_to_write = copy.deepcopy(data)
|
|
if not self.simulate:
|
|
if not self.connected:
|
|
self.open_device()
|
|
|
|
if not self.connected:
|
|
self.log.error("Failed to connect to device")
|
|
return
|
|
|
|
if self.mutex.tryLock():
|
|
try:
|
|
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 and tag.ndef is not None:
|
|
self.log.info("Detected NFC tag, writing data...")
|
|
ndef_record = ndef.TextRecord(self.data_to_write)
|
|
|
|
# IMPORTANT: DON'T CLOSE DEVICE BEFORE WRITING
|
|
tag.ndef.records = [ndef_record] # Write the record to the tag
|
|
self.log.info(f"Data '{self.data_to_write}' written successfully to tag.")
|
|
else:
|
|
self.log.error("Tag is not NDEF-compatible or invalid")
|
|
else:
|
|
self.log.debug("No tag present for writing")
|
|
except Exception as e:
|
|
self.log.error(f"Error during tag writing: {e}")
|
|
finally:
|
|
# Move closing outside the try block
|
|
self.mutex.unlock() # Unlock mutex always
|
|
|
|
# Close AFTER writing and mutex release
|
|
if self.connected:
|
|
self.close_device() # close_device can raise exception
|
|
|