RFID reading and authorization

orchestrateur/sensors/rfid_reader.py

@@ -0,0 +1,95 @@
+import time
+import pigpio
+from typing import Optional
+from sensors.lock import serial_lock
+
+
+class RFIDReader:
+    """
+    Grove 125kHz RFID Reader using bit-banged software serial on GPIO 17 (Pin 11).
+    Parses 14-byte frame: [0x02 STX] + [10 ASCII ID] + [2 ASCII Checksum] + [0x03 ETX]
+    """
+
+    def __init__(self, rx_pin: int = 17, baudrate: int = 9600):
+        self.rx_pin = rx_pin
+        self.baudrate = baudrate
+        self.buffer = bytearray()
+        self.pi = pigpio.pi()
+
+        if not self.pi.connected:
+            raise RuntimeError("pigpio daemon is not running. Run 'sudo systemctl start pigpiod'.")
+
+        with serial_lock:
+            self.pi.set_mode(self.rx_pin, pigpio.INPUT)
+            
+            # Clean up lingering serial sessions on this pin
+            try:
+                self.pi.bb_serial_read_close(self.rx_pin)
+            except pigpio.error:
+                pass
+            
+            self.pi.bb_serial_read_open(self.rx_pin, self.baudrate, 8)
+
+    def read_tag(self) -> Optional[str]:
+        """
+        Reads and accumulates bytes, returning the 10-digit Tag ID.
+        """
+        with serial_lock:
+            if not self.pi or not self.pi.connected:
+                return None
+
+            count, data = self.pi.bb_serial_read(self.rx_pin)
+            if count > 0:
+                self.buffer.extend(data)
+
+            # Look for Start-of-Text (0x02)
+            stx_idx = self.buffer.find(b"\x02")
+            
+            if stx_idx != -1:
+                # Discard noise prior to 0x02
+                if stx_idx > 0:
+                    self.buffer = self.buffer[stx_idx:]
+
+                # Wait for full 14-byte payload
+                if len(self.buffer) >= 14:
+                    raw_frame = self.buffer[:14]
+                    self.buffer = self.buffer[14:]  # Flush parsed frame
+
+                    # Verify End-of-Text (0x03)
+                    if raw_frame[-1] == 0x03:
+                        try:
+                            # Extract 10-character ID (indices 1 through 10)
+                            return raw_frame[1:11].decode("ascii")
+                        except UnicodeDecodeError:
+                            return None
+            else:
+                # Flush buffer if filled with noise without STX marker
+                if len(self.buffer) > 64:
+                    self.buffer.clear()
+
+        return None
+
+    def close(self):
+        with serial_lock:
+            if self.pi and self.pi.connected:
+                try:
+                    self.pi.bb_serial_read_close(self.rx_pin)
+                except pigpio.error:
+                    pass
+                self.pi.stop()
+
+
+if __name__ == "__main__":
+    reader = RFIDReader(rx_pin=17)
+    print("RFID Reader active on GPIO 17 (Pin 11). Swipe a tag...")
+
+    try:
+        while True:
+            tag = reader.read_tag()
+            if tag:
+                print(f"Scanned Tag ID: {tag}")
+            time.sleep(0.05)
+    except KeyboardInterrupt:
+        print("\nStopping reader.")
+    finally:
+        reader.close()