import argparse import json import os import queue import sqlite3 import threading import time from datetime import datetime from pathlib import Path from typing import List, Optional, Tuple import cv2 import customtkinter as ctk import face_recognition import numpy as np from PIL import Image, ImageTk from tkinter import filedialog, messagebox APP_TITLE = "FaceDesk Vision" DB_PATH = Path("face_registry.sqlite") REFERENCE_DIR = Path("registered_faces") CAMERA_WIDTH = 960 CAMERA_HEIGHT = 540 RECOGNITION_SCALE = 0.25 RECOGNITION_EVERY_N_FRAMES = 5 MATCH_TOLERANCE = 0.48 MAX_QUEUE_SIZE = 2 class FaceDatabase: """SQLite-backed face embedding store.""" def __init__(self, db_path: Path): self.db_path = db_path self._lock = threading.Lock() self._connect().close() self._create_schema() def _connect(self) -> sqlite3.Connection: conn = sqlite3.connect(self.db_path) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") return conn def _create_schema(self) -> None: with self._lock, self._connect() as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS people ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, encoding_json TEXT NOT NULL, image_path TEXT, created_at TEXT NOT NULL ) """ ) conn.execute("CREATE INDEX IF NOT EXISTS idx_people_name ON people(name)") def add_face(self, name: str, encoding: np.ndarray, image_path: Optional[str]) -> int: payload = json.dumps(encoding.astype(float).tolist()) created_at = datetime.utcnow().isoformat(timespec="seconds") + "Z" with self._lock, self._connect() as conn: cursor = conn.execute( "INSERT INTO people(name, encoding_json, image_path, created_at) VALUES (?, ?, ?, ?)", (name.strip(), payload, image_path, created_at), ) return int(cursor.lastrowid) def load_faces(self) -> Tuple[List[str], List[np.ndarray]]: with self._lock, self._connect() as conn: rows = conn.execute("SELECT name, encoding_json FROM people ORDER BY created_at ASC").fetchall() names: List[str] = [] encodings: List[np.ndarray] = [] for name, encoding_json in rows: try: vector = np.array(json.loads(encoding_json), dtype=np.float64) if vector.shape == (128,): names.append(name) encodings.append(vector) except (json.JSONDecodeError, TypeError, ValueError): continue return names, encodings def count(self) -> int: with self._lock, self._connect() as conn: row = conn.execute("SELECT COUNT(*) FROM people").fetchone() return int(row[0] if row else 0) class CameraReader(threading.Thread): """Continuously captures frames without blocking the GUI.""" def __init__(self, camera_index: int, frame_queue: queue.Queue): super().__init__(daemon=True) self.camera_index = camera_index self.frame_queue = frame_queue self.stop_event = threading.Event() self.capture: Optional[cv2.VideoCapture] = None self.error: Optional[str] = None def run(self) -> None: self.capture = cv2.VideoCapture(self.camera_index, cv2.CAP_DSHOW if os.name == "nt" else 0) if not self.capture or not self.capture.isOpened(): self.error = f"No webcam detected at index {self.camera_index}." return self.capture.set(cv2.CAP_PROP_FRAME_WIDTH, CAMERA_WIDTH) self.capture.set(cv2.CAP_PROP_FRAME_HEIGHT, CAMERA_HEIGHT) self.capture.set(cv2.CAP_PROP_FPS, 30) while not self.stop_event.is_set(): ok, frame = self.capture.read() if not ok or frame is None: self.error = "The webcam stopped returning frames." time.sleep(0.1) continue if self.frame_queue.full(): try: self.frame_queue.get_nowait() except queue.Empty: pass self.frame_queue.put(frame) if self.capture: self.capture.release() def stop(self) -> None: self.stop_event.set() class FaceDeskApp(ctk.CTk): def __init__(self, camera_index: int): super().__init__() ctk.set_appearance_mode("System") ctk.set_default_color_theme("blue") self.title(APP_TITLE) self.geometry("1240x760") self.minsize(980, 640) self.protocol("WM_DELETE_WINDOW", self.on_close) REFERENCE_DIR.mkdir(exist_ok=True) self.db = FaceDatabase(DB_PATH) self.known_names, self.known_encodings = self.db.load_faces() self.frame_queue: queue.Queue = queue.Queue(maxsize=MAX_QUEUE_SIZE) self.camera = CameraReader(camera_index, self.frame_queue) self.camera.start() self.current_frame: Optional[np.ndarray] = None self.latest_results: List[Tuple[int, int, int, int, str, bool]] = [] self.frame_counter = 0 self.recognition_busy = False self.recognition_lock = threading.Lock() self.last_fps_time = time.time() self.frames_seen = 0 self.display_image = None self._build_ui() self.after(150, self._check_camera_startup) self.after(10, self._ui_loop) def _build_ui(self) -> None: self.grid_columnconfigure(0, weight=1) self.grid_columnconfigure(1, weight=0) self.grid_rowconfigure(0, weight=1) self.video_panel = ctk.CTkFrame(self, corner_radius=14) self.video_panel.grid(row=0, column=0, sticky="nsew", padx=(18, 9), pady=18) self.video_panel.grid_rowconfigure(0, weight=1) self.video_panel.grid_columnconfigure(0, weight=1) self.video_label = ctk.CTkLabel(self.video_panel, text="Starting camera...", font=ctk.CTkFont(size=18, weight="bold")) self.video_label.grid(row=0, column=0, sticky="nsew", padx=12, pady=12) self.sidebar = ctk.CTkFrame(self, width=330, corner_radius=14) self.sidebar.grid(row=0, column=1, sticky="ns", padx=(9, 18), pady=18) self.sidebar.grid_propagate(False) title = ctk.CTkLabel(self.sidebar, text=APP_TITLE, font=ctk.CTkFont(size=24, weight="bold")) title.pack(anchor="w", padx=20, pady=(22, 6)) subtitle = ctk.CTkLabel( self.sidebar, text="Enroll faces locally, then recognize them in real time.", justify="left", wraplength=280, text_color=("gray35", "gray75"), ) subtitle.pack(anchor="w", padx=20, pady=(0, 18)) self.name_entry = ctk.CTkEntry(self.sidebar, placeholder_text="Person name", height=42) self.name_entry.pack(fill="x", padx=20, pady=(0, 12)) self.capture_button = ctk.CTkButton(self.sidebar, text="Register from camera", height=42, command=self.register_from_camera) self.capture_button.pack(fill="x", padx=20, pady=(0, 10)) self.upload_button = ctk.CTkButton(self.sidebar, text="Register from image", height=42, command=self.register_from_file) self.upload_button.pack(fill="x", padx=20, pady=(0, 18)) self.reload_button = ctk.CTkButton(self.sidebar, text="Reload database", height=42, fg_color="transparent", border_width=1, command=self.reload_database) self.reload_button.pack(fill="x", padx=20, pady=(0, 18)) self.stats_label = ctk.CTkLabel(self.sidebar, text="", justify="left", anchor="w") self.stats_label.pack(fill="x", padx=20, pady=(0, 18)) self.status_label = ctk.CTkLabel( self.sidebar, text="Ready.", justify="left", wraplength=280, text_color=("gray30", "gray80"), ) self.status_label.pack(fill="x", padx=20, pady=(0, 18)) help_text = ( "Tips:\n" "• Use bright, even lighting.\n" "• Register one clear frontal face per person.\n" "• Lower tolerance for stricter matching.\n" "• Keep biometric data protected." ) help_label = ctk.CTkLabel(self.sidebar, text=help_text, justify="left", wraplength=280, text_color=("gray35", "gray75")) help_label.pack(anchor="w", padx=20, pady=(10, 0)) self._update_stats(0.0) def _check_camera_startup(self) -> None: if self.camera.error: self.status_label.configure(text=self.camera.error) messagebox.showerror("Camera error", self.camera.error) def _update_stats(self, fps: float) -> None: self.stats_label.configure( text=f"Registered faces: {len(self.known_names)}\nDatabase records: {self.db.count()}\nDisplay FPS: {fps:.1f}\nTolerance: {MATCH_TOLERANCE}" ) def _ui_loop(self) -> None: try: while True: self.current_frame = self.frame_queue.get_nowait() except queue.Empty: pass if self.current_frame is not None: self.frame_counter += 1 display = self.current_frame.copy() if self.frame_counter % RECOGNITION_EVERY_N_FRAMES == 0: self._start_recognition_if_idle(self.current_frame.copy()) for top, right, bottom, left, name, known in self.latest_results: color = (42, 180, 85) if known else (40, 40, 230) cv2.rectangle(display, (left, top), (right, bottom), color, 2) label = name if known else "Unknown" cv2.rectangle(display, (left, max(0, top - 30)), (right, top), color, cv2.FILLED) cv2.putText(display, label, (left + 6, max(20, top - 8)), cv2.FONT_HERSHEY_SIMPLEX, 0.65, (255, 255, 255), 2) self._render_frame(display) self._measure_fps() elif self.camera.error: self.video_label.configure(text=self.camera.error) self.after(10, self._ui_loop) def _measure_fps(self) -> None: self.frames_seen += 1 now = time.time() elapsed = now - self.last_fps_time if elapsed >= 1.0: fps = self.frames_seen / elapsed self.frames_seen = 0 self.last_fps_time = now self._update_stats(fps) def _render_frame(self, frame_bgr: np.ndarray) -> None: frame_rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB) image = Image.fromarray(frame_rgb) panel_width = max(640, self.video_panel.winfo_width() - 24) panel_height = max(360, self.video_panel.winfo_height() - 24) image.thumbnail((panel_width, panel_height), Image.Resampling.LANCZOS) self.display_image = ImageTk.PhotoImage(image=image) self.video_label.configure(image=self.display_image, text="") def _start_recognition_if_idle(self, frame: np.ndarray) -> None: with self.recognition_lock: if self.recognition_busy: return self.recognition_busy = True thread = threading.Thread(target=self._recognize_frame, args=(frame,), daemon=True) thread.start() def _recognize_frame(self, frame: np.ndarray) -> None: try: small = cv2.resize(frame, (0, 0), fx=RECOGNITION_SCALE, fy=RECOGNITION_SCALE) rgb_small = cv2.cvtColor(small, cv2.COLOR_BGR2RGB) locations = face_recognition.face_locations(rgb_small, model="hog") encodings = face_recognition.face_encodings(rgb_small, locations) results: List[Tuple[int, int, int, int, str, bool]] = [] for location, encoding in zip(locations, encodings): name = "Unknown" known = False if self.known_encodings: distances = face_recognition.face_distance(self.known_encodings, encoding) best_index = int(np.argmin(distances)) if distances[best_index] <= MATCH_TOLERANCE: name = self.known_names[best_index] known = True top, right, bottom, left = location scale = int(1 / RECOGNITION_SCALE) results.append((top * scale, right * scale, bottom * scale, left * scale, name, known)) self.latest_results = results except Exception as exc: self.status_label.configure(text=f"Recognition error: {exc}") finally: with self.recognition_lock: self.recognition_busy = False def _validate_name(self) -> Optional[str]: name = self.name_entry.get().strip() if len(name) < 2: messagebox.showwarning("Name required", "Enter a name with at least two characters before enrolling a face.") return None return name def register_from_camera(self) -> None: name = self._validate_name() if not name: return if self.current_frame is None: messagebox.showerror("No frame", "The camera has not produced a frame yet.") return self._register_frame(name, self.current_frame.copy(), source_label="camera") def register_from_file(self) -> None: name = self._validate_name() if not name: return path = filedialog.askopenfilename( title="Choose a face image", filetypes=[("Image files", "*.jpg *.jpeg *.png *.bmp *.webp"), ("All files", "*.*")], ) if not path: return image = cv2.imread(path) if image is None: messagebox.showerror("Invalid image", "OpenCV could not read the selected image.") return self._register_frame(name, image, source_label="image") def _register_frame(self, name: str, frame_bgr: np.ndarray, source_label: str) -> None: try: rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB) locations = face_recognition.face_locations(rgb, model="hog") if not locations: messagebox.showerror("No face found", "No face was detected. Try better lighting and a frontal pose.") return if len(locations) > 1: self.status_label.configure(text="Multiple faces found; enrolling the largest detected face.") location = self._largest_location(locations) encodings = face_recognition.face_encodings(rgb, [location]) if not encodings: messagebox.showerror("Embedding failed", "A face was detected, but the embedding could not be computed.") return image_path = self._save_reference_image(name, frame_bgr, location) person_id = self.db.add_face(name, encodings[0], str(image_path)) self.reload_database(show_dialog=False) self.status_label.configure(text=f"Enrolled {name} from {source_label} as record #{person_id}.") messagebox.showinfo("Enrollment complete", f"{name} has been registered successfully.") except Exception as exc: messagebox.showerror("Enrollment error", str(exc)) @staticmethod def _largest_location(locations: List[Tuple[int, int, int, int]]) -> Tuple[int, int, int, int]: return max(locations, key=lambda box: max(0, box[2] - box[0]) * max(0, box[1] - box[3])) @staticmethod def _safe_filename(name: str) -> str: safe = "".join(ch if ch.isalnum() or ch in ("-", "_") else "_" for ch in name.strip()) return safe[:80] or "registered_face" def _save_reference_image(self, name: str, frame_bgr: np.ndarray, location: Tuple[int, int, int, int]) -> Path: top, right, bottom, left = location pad = 32 height, width = frame_bgr.shape[:2] top = max(0, top - pad) right = min(width, right + pad) bottom = min(height, bottom + pad) left = max(0, left - pad) crop = frame_bgr[top:bottom, left:right] timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S") path = REFERENCE_DIR / f"{self._safe_filename(name)}_{timestamp}.jpg" cv2.imwrite(str(path), crop if crop.size else frame_bgr) return path def reload_database(self, show_dialog: bool = True) -> None: self.known_names, self.known_encodings = self.db.load_faces() self._update_stats(0.0) if show_dialog: messagebox.showinfo("Database reloaded", f"Loaded {len(self.known_names)} registered face embeddings.") def on_close(self) -> None: self.camera.stop() self.destroy() def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Real-time webcam facial recognition desktop app.") parser.add_argument("--camera", type=int, default=0, help="OpenCV camera index. Default: 0") return parser.parse_args() def main() -> None: args = parse_args() app = FaceDeskApp(camera_index=args.camera) app.mainloop() if __name__ == "__main__": main()