import serial import time import cv2 import numpy as np from gpiozero import Buzzer, LED from ultralytics import YOLO import sys import json # Import json for local log file import os # Import os for file path checks # === CONFIGURATION === SERIAL_PORT = '/dev/ttyAMA0' BAUD_RATE = 9600 MODEL_PATH = 'best.pt' # Ensure this points to your trained YOLOv8 best.pt file LOG_FILE_PATH = 'logs.json' # Path for the local JSON log file for access events # RFID_MAP is hardcoded as per your request RFID_MAP = { "19761163": "Rafael", "49F01A63": "Salim", "D97E1B63": "Matteo", "D9CC1063": "Roxana", "59370C63": "Alex" } ARDUINO_MESSAGE_READY = "READY: Scan Card..." ARDUINO_MESSAGE_VERIFYING = "VERIFYING:" ARDUINO_MESSAGE_GRANTED = "GRANTED" ARDUINO_MESSAGE_DENIED = "DENIED" ARDUINO_MESSAGE_ERROR = "ERROR: Pi Issue" COOLDOWN_PERIOD_SECONDS = 5 DEBUG = True # === GPIO PINS === BUZZER_PIN = 12 GREEN_LED_PIN = 6 RED_LED_PIN = 13 # Global gpiozero objects buzzer = None green_led = None red_led = None # === HELPER FUNCTIONS === def send_to_arduino(ser, message): """Sends a string message over serial to the Arduino.""" if ser and ser.is_open: try: ser.write((message + '\n').encode('utf-8')) if DEBUG: print(f"Sent to Arduino: {message}") except Exception as e: print(f"Error sending to Arduino: {e}") else: if DEBUG: print("Warning: Serial port not open. Cannot send message to Arduino.") def feedback_granted(): """Activates green LED and buzzer for access granted feedback.""" if green_led: green_led.on() if red_led: red_led.off() if buzzer: buzzer.on() time.sleep(0.4) if buzzer: buzzer.off() def feedback_denied(): """Activates red LED and short buzzer pulses for access denied feedback.""" if green_led: green_led.off() if red_led: red_led.on() if buzzer: for _ in range(2): buzzer.on() time.sleep(0.1) buzzer.off() time.sleep(0.1) def load_logs_from_file(): """Loads existing access logs from the JSON file.""" if not os.path.exists(LOG_FILE_PATH) or os.stat(LOG_FILE_PATH).st_size == 0: if DEBUG: print(f"DEBUG: {LOG_FILE_PATH} does not exist or is empty. Returning empty list.") return [] try: with open(LOG_FILE_PATH, 'r') as f: logs = json.load(f) if DEBUG: print(f"DEBUG: Successfully loaded {len(logs)} logs from {LOG_FILE_PATH}.") return logs except json.JSONDecodeError as e: print(f"Warning: {LOG_FILE_PATH} is empty or malformed: {e}. Starting with empty logs.") return [] except Exception as e: print(f"ERROR: Could not load logs from {LOG_FILE_PATH}: {e}") return [] def save_logs_to_file(logs): """Saves the current list of logs to the JSON file, forcing a write to disk.""" try: with open(LOG_FILE_PATH, 'w') as f: json.dump(logs, f, indent=4) f.flush() # Flush Python's internal buffer os.fsync(f.fileno()) # Force OS to write to disk if DEBUG: print(f"DEBUG: Successfully saved {len(logs)} logs to {LOG_FILE_PATH}.") except Exception as e: print(f"ERROR: Could not save logs to {LOG_FILE_PATH}: {e}") def log_access_event(rfid_uid, name, status): """Logs an access event to a local JSON file.""" if DEBUG: print(f"DEBUG: Attempting to log event: RFID={rfid_uid}, Name={name}, Status={status}") # Load existing logs, add new entry, then save logs = load_logs_from_file() log_entry = { "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), # Current local time string "rfid_uid": rfid_uid, "name": name, "status": status } logs.append(log_entry) save_logs_to_file(logs) if DEBUG: print(f"LOGGED to {LOG_FILE_PATH}: RFID={rfid_uid}, Name={name}, Status={status}") def initialize_components(): """Initializes serial communication, webcam, YOLO model, and GPIO.""" global buzzer, green_led, red_led # 1. Serial Communication try: ser = serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=0.1) time.sleep(2) print(f"Serial port {SERIAL_PORT} opened.") send_to_arduino(ser, ARDUINO_MESSAGE_READY) except serial.SerialException as e: print(f"ERROR: Could not open serial port {SERIAL_PORT}: {e}") print("Please ensure it's correct, not in use, and Raspberry Pi serial console is disabled.") sys.exit(1) # Exit if serial communication cannot be established # 2. Webcam cap = cv2.VideoCapture(0) # 0 is usually the default USB webcam if not cap.isOpened(): send_to_arduino(ser, ARDUINO_MESSAGE_ERROR) raise Exception("Could not open webcam. Check if connected and drivers are installed.") print("Webcam opened.") # 3. YOLO Model try: model = YOLO(MODEL_PATH) # Load your best.pt model print(f"YOLO model '{MODEL_PATH}' loaded.") except Exception as e: send_to_arduino(ser, ARDUINO_MESSAGE_ERROR) raise Exception(f"Error loading YOLO model from {MODEL_PATH}: {e}. Ensure 'best.pt' is valid and in path.") # 4. GPIO Setup with gpiozero try: buzzer = Buzzer(BUZZER_PIN) green_led = LED(GREEN_LED_PIN) red_led = LED(RED_LED_PIN) # Ensure all outputs are initially off buzzer.off() green_led.off() red_led.off() print("GPIO pins initialized with gpiozero.") except Exception as e: print(f"ERROR: Could not set up GPIO pins with gpiozero: {e}") print("Possible reasons: Pin conflict, incorrect pin number, or an underlying libgpiod issue.") print("Ensure 'sudo usermod -a -G gpio student' has been run and reboot.") send_to_arduino(ser, ARDUINO_MESSAGE_ERROR) sys.exit(1) # Exit if GPIO cannot be set up return ser, cap, model def process_frame_for_face(model, frame, expected_name=None): """ Performs face recognition on the frame and returns True if expected person is detected. Does NOT draw bounding boxes as no display is used in this version. """ results = model(frame, conf=0.6, verbose=False) # Run inference, confidence threshold 0.6 person_found = False detected_faces = [] # Store names of detected faces for r in results: boxes = r.boxes # Bounding boxes for box in boxes: cls = int(box.cls[0]) # Class ID name = model.names[cls] # Class name (e.g., "Rafael", "Salim") detected_faces.append(name.lower()) if expected_name and expected_name.lower() in detected_faces: person_found = True if DEBUG: print(f"DEBUG: Detected face: {expected_name}, matches expected.") elif expected_name: # Only if an expected_name was provided, but not found if DEBUG: print(f"DEBUG: Expected face '{expected_name}' NOT detected.") else: # If no expected name, just report what's seen passively if DEBUG: print(f"DEBUG: Passive detection: {detected_faces if detected_faces else 'No faces detected in frame.'}") return person_found def recognize_face_multiple_frames(model, cap, expected_name, num_frames=3): """ Captures and processes multiple frames to improve face recognition robustness. Returns True if the expected face is found in any of the frames. """ if DEBUG: print(f"Attempting to recognize {expected_name} over {num_frames} frames...") for i in range(num_frames): ret, frame = cap.read() if not ret: if DEBUG: print(f"Warning: Failed to capture frame {i+1} for recognition.") continue match = process_frame_for_face(model, frame, expected_name) if match: return True time.sleep(0.2) # Small delay between frames return False def main_loop(ser, cap, model): """Main loop for continuously monitoring RFID and webcam for access control.""" last_verification_time = 0 # Stores timestamp of last access attempt current_system_state = ARDUINO_MESSAGE_READY # Tracks current state for sending to Arduino last_passive_log_time = 0 # Tracks last time passive face detection was logged while True: current_time = time.time() # --- RFID Read from Arduino --- rfid_uid = None if ser.in_waiting > 0: if DEBUG: print("DEBUG: Data available on serial port.") try: line = ser.readline().decode('utf-8').strip() if line: # Ensure the received line is not empty if DEBUG: print(f"DEBUG: Received raw data from Arduino: '{line}'") # Validate if the received line looks like a plausible RFID UID (hex characters and length) if len(line) >= 8 and len(line) <= 14 and all(c in '0123456789ABCDEF' for c in line.upper()): rfid_uid = line.upper() # Convert to uppercase for consistent mapping lookup if DEBUG: print(f"DEBUG: Valid RFID UID parsed from Arduino: '{rfid_uid}'") current_system_state = ARDUINO_MESSAGE_VERIFYING # Change state to reflect ongoing verification else: if DEBUG: print(f"DEBUG: Received data is not a valid UID format: '{line}'") except Exception as e: print(f"ERROR: Error reading from serial: {e}") # Log error but don't halt the system, as it's a continuous loop # --- Verification Logic (Triggered by RFID and Cooldown) --- # Only proceed if a valid RFID UID was received AND cooldown period has passed if rfid_uid and (current_time - last_verification_time > COOLDOWN_PERIOD_SECONDS): if DEBUG: print(f"DEBUG: Processing RFID UID '{rfid_uid}'. Cooldown passed.") # Get name from hardcoded RFID_MAP expected_name = RFID_MAP.get(rfid_uid, "Unknown Person") if expected_name != "Unknown Person": send_to_arduino(ser, f"{ARDUINO_MESSAGE_VERIFYING}{expected_name}") if DEBUG: print(f"LOG: RFID matched with: {expected_name}. Proceeding to face verification.") # Attempt to recognize the face over multiple frames face_match_found = recognize_face_multiple_frames(model, cap, expected_name, num_frames=3) if face_match_found: send_to_arduino(ser, ARDUINO_MESSAGE_GRANTED) feedback_granted() # Activate granted feedback (LED, buzzer) current_system_state = ARDUINO_MESSAGE_GRANTED if DEBUG: print("LOG: Access Granted!") log_access_event(rfid_uid, expected_name, "GRANTED") # Log to local JSON file else: send_to_arduino(ser, ARDUINO_MESSAGE_DENIED) feedback_denied() # Activate denied feedback (LED, buzzer) current_system_state = ARDUINO_MESSAGE_DENIED if DEBUG: print("LOG: Access Denied: Face mismatch or no face detected.") log_access_event(rfid_uid, expected_name, "DENIED - Face Mismatch") # Log to local JSON file else: # Unknown RFID card send_to_arduino(ser, ARDUINO_MESSAGE_DENIED) feedback_denied() # Activate denied feedback current_system_state = ARDUINO_MESSAGE_DENIED if DEBUG: print(f"LOG: Access Denied: Unknown RFID card ({rfid_uid}).") log_access_event(rfid_uid, "Unknown", "DENIED - Unknown Card") # Log to local JSON file last_verification_time = current_time # Reset cooldown timer time.sleep(2) # Keep the result visible/active for a moment send_to_arduino(ser, ARDUINO_MESSAGE_READY) # Revert Arduino display to ready # Turn off LEDs via gpiozero objects if green_led: green_led.off() if red_led: red_led.off() current_system_state = ARDUINO_MESSAGE_READY # Reset internal state # --- Passive Face Detection (when not in verification state) --- else: # Only log passive detections periodically to avoid spamming the console if current_time - last_passive_log_time > 2: # Log every 2 seconds ret, frame = cap.read() if ret: # Process frame for face detection, but don't expect a specific person process_frame_for_face(model, frame, expected_name=None) else: if DEBUG: print("Warning: Failed to capture frame for passive detection.") last_passive_log_time = current_time # === Main Execution Block === if __name__ == "__main__": ser = None # Initialize serial to None cap = None # Initialize camera to None model = None # Initialize model to None try: if DEBUG: print(f"DEBUG: Checking for {LOG_FILE_PATH} at {os.getcwd()}") # Initialize an empty log file if it doesn't exist or is empty/corrupted if not os.path.exists(LOG_FILE_PATH) or os.stat(LOG_FILE_PATH).st_size == 0: with open(LOG_FILE_PATH, 'w') as f: json.dump([], f) # Write an empty JSON array print(f"Created/Initialized empty log file: {LOG_FILE_PATH}") else: # Try to load existing logs to confirm it's valid JSON try: _ = load_logs_from_file() except json.JSONDecodeError: print(f"Warning: {LOG_FILE_PATH} exists but is malformed. Re-initializing as empty array.") with open(LOG_FILE_PATH, 'w') as f: json.dump([], f) except Exception as e: print(f"ERROR: Problem verifying existing {LOG_FILE_PATH}: {e}. Skipping re-initialization.") ser, cap, model = initialize_components() # Setup all hardware and models print("System initialized. Starting main loop...") main_loop(ser, cap, model) # Run the main logic except Exception as e: print(f"CRITICAL ERROR: System stopped due to an unhandled exception: {e}") except KeyboardInterrupt: print("\nProgram terminated by user (Ctrl+C).") finally: # --- Cleanup Resources --- if cap and cap.isOpened(): cap.release() print("Webcam released.") # No cv2.destroyAllWindows() needed as no windows are created if ser and ser.is_open: ser.close() print("Serial port closed.") # gpiozero objects automatically clean up on script exit if buzzer: buzzer.close() if green_led: green_led.close() if red_led: red_led.close() print("GPIO resources cleaned up.") print("Cleanup complete. Exiting program.")