""" FastAPI Attendance System - Production Ready Run with: uvicorn main:app --host 0.0.0.0 --port 5001 --workers 4 """ import base64 import io import os import sys from datetime import datetime from typing import Optional import cv2 import numpy as np import face_recognition import mysql.connector from fastapi import FastAPI, File, Form, UploadFile, HTTPException, status from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from PIL import Image from pydantic import BaseModel from mysql.connector import Error from werkzeug.utils import secure_filename # Import the core logic function from attendance_marking import markAttendance # --- Configuration --- DB_CONFIG = { 'user': 'root', 'password': 'redclaws', 'host': 'localhost', 'database': 'NewAttn', } PATH_KNOWN_DATA = 'known_data' PATH_TRAINING_IMAGES = 'Training_images' FACE_DISTANCE_THRESHOLD = 0.5 # Ensure directories exist os.makedirs(PATH_KNOWN_DATA, exist_ok=True) os.makedirs(PATH_TRAINING_IMAGES, exist_ok=True) # --- Data Models --- class AttendanceRequest(BaseModel): image: str # Base64 encoded image class AttendanceResponse(BaseModel): status: str message: str emp_id: Optional[str] = None distance: Optional[str] = None class RegistrationResponse(BaseModel): status: str message: str emp_id: Optional[str] = None # --- Helper Functions --- def load_known_data(): """Loads encodings and emp_ids from the known_data directory.""" try: encodings = np.load( os.path.join(PATH_KNOWN_DATA, 'known_encodings.npy'), allow_pickle=True ) with open(os.path.join(PATH_KNOWN_DATA, 'known_names.txt'), 'r') as f: emp_ids = [line.strip() for line in f.readlines()] print(f"✅ Loaded {len(encodings)} faces.") return encodings, emp_ids except FileNotFoundError: print("⚠️ No existing known data found. Starting fresh.") return np.array([]), [] def findEncodings(images): encodeList = [] for img in images: img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) face_encodings = face_recognition.face_encodings(img) if face_encodings: encodeList.append(face_encodings[0]) else: print("Warning: No face found in one of the training images!") return encodeList # Function to mark attendance using MySQL def markAttendance(emp_id): now = datetime.now() dtString = now.strftime('%H:%M:%S') dateString = now.strftime('%Y-%m-%d') cursor = None try: record_count=0 # cursor = db_conn.cursor() # # ✅ FIXED: Use COUNT(*) to get the number of matching records # check_query = """ # SELECT COUNT(*) # FROM dailylog # WHERE emp_id = %s AND date = %s # """ # cursor.execute(check_query, (emp_id, dateString)) # result = cursor.fetchone() # record_count = result[0] if result else 0 # print(f"DEBUG: emp_id={emp_id}, date={dateString}, count={record_count}") if record_count == 0: # insert_query = """ # INSERT INTO dailylog (emp_id, date, punch_in) # VALUES (%s, %s, %s) # """ # cursor.execute(insert_query, (emp_id, dateString, dtString)) # db_conn.commit() return True , "new" else: return False , "duplicate" except Error as e: # print(f"DB error: {e}") # db_conn.rollback() return False , f"error: {e}" # def get_db_connection(): # """Get a database connection.""" # try: # return mysql.connector.connect(**DB_CONFIG) # except Error as e: # print(f"❌ Database connection error: {e}") # raise HTTPException( # status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, # detail=f"Database connection failed: {str(e)}" # ) # --- Initialize FastAPI App --- app = FastAPI( title="Facial Attendance System API", description="Production-ready facial recognition attendance system", version="1.0.0" ) # Configure CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], # In production, replace with specific origins allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Global variables for encodings (loaded at startup) encodeListKnown = np.array([]) classNames = [] # --- Startup Event --- @app.on_event("startup") async def startup_event(): """Load known face encodings on startup.""" global encodeListKnown, classNames encodeListKnown, classNames = load_known_data() print("🚀 FastAPI Attendance System Started") # --- Health Check Endpoint --- @app.get("/health") async def health_check(): """Health check endpoint.""" return { "status": "healthy", "timestamp": datetime.now().isoformat(), "registered_employees": len(classNames) } # --- Attendance Marking Endpoint --- @app.post("/api/mark_attendance", response_model=AttendanceResponse) async def mark_attendance_api(request: AttendanceRequest): """ Mark attendance using facial recognition. Args: request: AttendanceRequest containing base64 encoded image Returns: AttendanceResponse with status and details """ if not request.image: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="No image data provided" ) #db_conn = None try: # 1. Decode and Convert Image image_bytes = base64.b64decode(request.image) img = Image.open(io.BytesIO(image_bytes)).convert('RGB') img_np = np.array(img) img_bgr = cv2.cvtColor(img_np, cv2.COLOR_RGB2BGR) # 2. Process Image (resize for faster detection) imgS = cv2.resize(img_bgr, (0, 0), None, 0.25, 0.25) imgS = cv2.cvtColor(imgS, cv2.COLOR_BGR2RGB) # 3. Detect faces facesCurFrame = face_recognition.face_locations(imgS) encodesCurFrame = face_recognition.face_encodings(imgS, facesCurFrame) if not facesCurFrame: return AttendanceResponse( status="failure", message="No face detected in the image." ) # 4. Match face encodeFace = encodesCurFrame[0] matches = face_recognition.compare_faces(encodeListKnown, encodeFace) faceDis = face_recognition.face_distance(encodeListKnown, encodeFace) if len(faceDis) == 0: return AttendanceResponse( status="failure", message="No registered employees in the system." ) matchIndex = np.argmin(faceDis) min_distance = faceDis[matchIndex] # 5. Check threshold and mark attendance if matches[matchIndex] and min_distance < FACE_DISTANCE_THRESHOLD: employee_id = classNames[matchIndex].upper() # Get DB connection # db_conn = get_db_connection() # Mark attendance was_marked, status_msg = markAttendance(employee_id) if was_marked: return AttendanceResponse( status="success", message=f"Attendance marked for {employee_id}.", emp_id=employee_id, distance=f"{min_distance:.2f}" ) elif status_msg == "duplicate": return AttendanceResponse( status="info", message=f"{employee_id} already logged today.", emp_id=employee_id, distance=f"{min_distance:.2f}" ) else: return AttendanceResponse( status="error", message=f"Failed to mark attendance: {status_msg}" ) else: return AttendanceResponse( status="failure", message=f"Unknown person detected. Min Distance: {min_distance:.2f}" ) except Error as err: print(f"❌ Database Error: {db_err}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Database error: {str(db_err)}" ) except Exception as e: print(f"❌ Exception: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) # finally: # if db_conn is not None and db_conn.is_connected(): # db_conn.close() # --- Employee Registration Endpoint --- @app.post("/api/register_employee", response_model=RegistrationResponse) async def register_employee( emp_id: str = Form(...), image: UploadFile = File(...) ): """ Register a new employee with facial recognition. Args: emp_id: Employee ID image: Employee face image file Returns: RegistrationResponse with status and details """ global encodeListKnown, classNames db_conn = None cursor = None image_path = None try: # Validate inputs if not emp_id or not image: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Employee ID and image are required." ) # Sanitize employee ID emp_id = emp_id.strip().upper() # Check if employee already exists if emp_id in classNames: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Employee {emp_id} already exists in the system." ) # Validate image file if not image.content_type.startswith('image/'): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid file type. Please upload an image." ) # Save the image filename = secure_filename(f"{emp_id}.jpg") image_path = os.path.join(PATH_TRAINING_IMAGES, filename) # Read and save image contents = await image.read() with open(image_path, 'wb') as f: f.write(contents) # Load and process the image img = cv2.imread(image_path) if img is None: if os.path.exists(image_path): os.remove(image_path) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Failed to read the uploaded image." ) # Convert to RGB for face_recognition img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) # Detect faces and generate encoding face_encodings = face_recognition.face_encodings(img_rgb) if not face_encodings: os.remove(image_path) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="No face detected in the image. Please upload a clear face photo." ) if len(face_encodings) > 1: os.remove(image_path) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Multiple faces detected. Please upload an image with only one face." ) # Get the encoding new_encoding = face_encodings[0] # Update global arrays if len(encodeListKnown) == 0: encodeListKnown = np.array([new_encoding]) else: encodeListKnown = np.vstack([encodeListKnown, new_encoding]) classNames.append(emp_id) # Save updated encodings and names np.save( os.path.join(PATH_KNOWN_DATA, 'known_encodings.npy'), encodeListKnown ) with open(os.path.join(PATH_KNOWN_DATA, 'known_names.txt'), 'w') as f: for name in classNames: f.write(f"{name}\n") # Insert into database # db_conn = get_db_connection() # cursor = db_conn.cursor() # query = """ # INSERT INTO employees (emp_id, name, email, department) # VALUES (%s, %s, %s, %s) # """ # cursor.execute(query, (emp_id, "DUMMY", "dummy@mail.com", "DummyDept")) # db_conn.commit() print(f"✅ Successfully registered employee: {emp_id}") return RegistrationResponse( status="success", message=f"Employee {emp_id} registered successfully!", emp_id=emp_id ) except HTTPException: # Re-raise HTTP exceptions raise except Exception as e: print(f"❌ Error during registration: {e}") if db_conn: db_conn.rollback() if image_path and os.path.exists(image_path): os.remove(image_path) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Registration failed: {str(e)}" ) # finally: # if cursor: # cursor.close() # if db_conn is not None and db_conn.is_connected(): # db_conn.close() # --- Get Registered Employees Endpoint --- @app.get("/api/employees") async def get_employees(): """Get list of all registered employees.""" return { "status": "success", "count": len(classNames), "employees": classNames } # --- Root Endpoint --- @app.get("/") async def root(): """Root endpoint.""" return { "message": "Facial Attendance System API", "version": "1.0.0", "docs": "/docs", "health": "/health" } if __name__ == "__main__": import uvicorn uvicorn.run( "app:app", host="0.0.0.0", port=5001, reload=False, # Set to False in production workers=1 # Increase for production (e.g., 4) )