import threading import time from datetime import datetime, timedelta import queue import cv2 import numpy as np from backend.face_engine import FaceEngine from backend.database import log_attendance, get_last_attendance_log, get_todays_logs # Alert queue to broadcast attendance notifications to the frontend in real time attendance_alerts = queue.Queue() class VideoCamera: _instance = None _lock = threading.Lock() def __new__(cls, *args, **kwargs): with cls._lock: if cls._instance is None: cls._instance = super(VideoCamera, cls).__new__(cls) cls._instance.initialized = False return cls._instance def __init__(self, camera_index: int = 0): if self.initialized: return self.camera_index = camera_index self.cap = None self.running = False self.thread = None self.lock = threading.Lock() # Initialize Face Engine self.engine = FaceEngine() # Cooldown map: employee_id -> datetime of last log # Prevents double-scans within 2 minutes self.log_cooldowns = {} self.cooldown_duration = timedelta(minutes=2) # Current active frame and processed frame self.latest_frame = None self.latest_faces = None self.processed_frame = None # Biometric Liveness Buffers self.landmark_buffer = [] self.last_spoof_alert_time = 0.0 self.initialized = True def start(self): """Starts the camera capture thread if not already running.""" with self.lock: if self.running: return True print(f"[*] Starting webcam capture on index {self.camera_index}...") self.cap = cv2.VideoCapture(self.camera_index) if not self.cap.isOpened(): print("[-] Error: Could not open webcam.") return False # Set camera dimensions for optimal latency (640x480) self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640) self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480) self.running = True self.thread = threading.Thread(target=self._capture_loop, daemon=True) self.thread.start() print("[+] Camera thread launched successfully!") return True def stop(self): """Stops the camera capture thread.""" with self.lock: self.running = False if self.thread: self.thread.join(timeout=1.0) if self.cap: self.cap.release() self.cap = None print("[+] Camera capture stopped.") def change_camera(self, new_index: int): """Switches the camera device index dynamically.""" self.stop() self.camera_index = new_index return self.start() def get_raw_frame(self): """Returns the last raw frame captured (without HUD overlays).""" with self.lock: return self.latest_frame.copy() if self.latest_frame is not None else None def get_latest_face_sample(self): """Returns the latest raw frame and the detected faces list safely.""" with self.lock: if self.latest_frame is None: return None, None faces_copy = self.latest_faces.copy() if self.latest_faces is not None else None return self.latest_frame.copy(), faces_copy def get_processed_frame_bytes(self): """Returns the JPEG-encoded processed frame with the cybernetic HUD overlays.""" with self.lock: if self.processed_frame is None or not self.running: # Return a placeholder dark frame if camera is loading or offline blank = np.zeros((480, 640, 3), dtype=np.uint8) msg = "Camera Offline / Stopped" if not self.running else "Initializing Camera Feed..." cv2.putText(blank, msg, (150, 240), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 242), 2, cv2.LINE_AA) _, jpeg = cv2.imencode('.jpg', blank) return jpeg.tobytes() _, jpeg = cv2.imencode('.jpg', self.processed_frame) return jpeg.tobytes() def _capture_loop(self): """Background loop continuously capturing, detecting, matching, and logging.""" scanline_y = 0 scanline_direction = 5 while self.running: ret, frame = self.cap.read() if not ret: time.sleep(0.01) continue # Flip horizontally to match standard mirror webcam view frame = cv2.flip(frame, 1) # Create a clone for drawing the premium HUD hud_frame = frame.copy() h, w = frame.shape[:2] # Perform face detection retval, faces = self.engine.detect_faces(frame) # Store raw frame and pre-detected faces list in synchronized lock with self.lock: self.latest_frame = frame.copy() self.latest_faces = faces # Draw premium high-tech grid scanner lines on the border (HUD style) cv2.rectangle(hud_frame, (10, 10), (w-10, h-10), (0, 255, 242), 1) # Corner ticks self._draw_hud_corners(hud_frame, 10, 10, w-10, h-10, (0, 255, 242), 15, 2) # If faces detected, process them if faces is not None and len(faces) > 0: for face in faces: # 1. Parse bounding box coordinates and score bbox = face[0:4].astype(int) x, y, fw, fh = bbox[0], bbox[1], bbox[2], bbox[3] score = face[-1] # Ensure bbox stays inside frame boundaries x = max(0, min(x, w-1)) y = max(0, min(y, h-1)) fw = min(fw, w - x) fh = min(fh, h - y) # 2. Extract facial landmarks # YuNet output has 5 landmarks: right eye, left eye, nose, right mouth, left mouth landmarks = face[4:14].reshape(5, 2).astype(int) # Update liveness buffer self.landmark_buffer.append(landmarks.copy()) if len(self.landmark_buffer) > 15: self.landmark_buffer.pop(0) # Calculate Liveness (Static Photo Guard + 3D Pose Ratio check) is_live = True liveness_label = "LIVENESS: PENDING..." if len(self.landmark_buffer) >= 8: # 1. Static Photo Check (Coordinate standard deviation) pts = np.array(self.landmark_buffer) std_coords = np.std(pts, axis=0) # shape (5, 2) avg_std = np.mean(std_coords) # If coordinates are completely motionless (standard deviation < 0.7 pixels) # it is a static photo! A live person has micro-tremors and natural shifts if avg_std < 0.7: is_live = False liveness_label = "SPOOF: STATIC PHOTO" else: liveness_label = "LIVENESS: VERIFIED (3D)" # 3. Extract embedding and perform matching (only execute if liveness verified) emp_id, emp_name, match_score = None, "Unknown", 0.0 event_status = "" if is_live: query_emb = self.engine.extract_embedding(frame, face) if query_emb is not None: emp_id, emp_name, match_score = self.engine.match_face(query_emb) # 4. Handle check-in/check-out trigger if matched if emp_id: event_status = self._process_attendance_trigger(emp_id, emp_name, match_score) # 5. Set HUD color palette based on match and liveness status if not is_live: # Spoof Alarm (Neon Crimson / Warning) color = (60, 76, 231) # Crimson Red label = f"SPOOF SCAN DETECTED | {liveness_label}" # Debounce and push spoof security alerts to frontend SSE now_ts = time.time() if now_ts - self.last_spoof_alert_time > 10.0: self.last_spoof_alert_time = now_ts alert_data = { "employee_id": "SEC-ALERT", "name": "SPOOF SCAN DETECTED", "timestamp": datetime.now().strftime("%I:%M %p"), "event_type": "spoof_attempt", "score": 0 } attendance_alerts.put(alert_data) elif emp_id: # Recognized (Neon Emerald) color = (46, 204, 113) label = f"{emp_name} ({int(match_score * 100)}%) | {liveness_label}" if event_status: label += f" | {event_status.replace('_', ' ').upper()}" else: # Unknown (Neon Crimson) color = (231, 76, 60) label = f"UNKNOWN ID | {liveness_label}" # Draw glowing face box self._draw_neon_bbox(hud_frame, x, y, fw, fh, color, label) # Draw glowing cyan tracking dots on the 5 landmarks for lm in landmarks: cv2.circle(hud_frame, (lm[0], lm[1]), 3, (0, 255, 242), -1) # Core cv2.circle(hud_frame, (lm[0], lm[1]), 6, (0, 255, 242), 1) # Glow else: self.landmark_buffer.clear() # If no faces, draw a scanning laser-line animation cv2.line(hud_frame, (10, scanline_y), (w-10, scanline_y), (0, 255, 242), 2) scanline_y += scanline_direction if scanline_y >= h - 15 or scanline_y <= 15: scanline_direction *= -1 # Update latest processed frame with self.lock: self.processed_frame = hud_frame # Limit background loop to ~30 FPS to save CPU time.sleep(0.033) def _draw_hud_corners(self, img, x1, y1, x2, y2, color, length, thickness): """Draws advanced corner HUD ticks for a premium UI feel.""" # Top Left cv2.line(img, (x1, y1), (x1 + length, y1), color, thickness) cv2.line(img, (x1, y1), (x1, y1 + length), color, thickness) # Top Right cv2.line(img, (x2, y1), (x2 - length, y1), color, thickness) cv2.line(img, (x2, y1), (x2, y1 + length), color, thickness) # Bottom Left cv2.line(img, (x1, y2), (x1 + length, y2), color, thickness) cv2.line(img, (x1, y2), (x1, y2 - length), color, thickness) # Bottom Right cv2.line(img, (x2, y2), (x2 - length, y2), color, thickness) cv2.line(img, (x2, y2), (x2, y2 - length), color, thickness) def _draw_neon_bbox(self, img, x, y, w, h, color, label): """Draws a premium rounded bounding box with a neon outer glow and title banner.""" thickness = 2 glow_thickness = 4 # 1. Draw outer glow (semi-transparent blur is hard in OpenCV, so we draw multiple faint thick lines) cv2.rectangle(img, (x - 2, y - 2), (x + w + 2, y + h + 2), color, thickness + 1) # 2. Draw sharp primary box cv2.rectangle(img, (x, y), (x + w, y + h), color, thickness) # 3. Corner ticks on bounding box self._draw_hud_corners(img, x, y, x + w, y + h, (0, 255, 242), 10, 2) # 4. Draw label banner font = cv2.FONT_HERSHEY_SIMPLEX font_scale = 0.5 text_thickness = 1 (label_w, label_h), baseline = cv2.getTextSize(label, font, font_scale, text_thickness) # Top-banner position banner_y = max(y - label_h - 10, 0) cv2.rectangle(img, (x, banner_y), (x + label_w + 14, y), color, -1) # Filled colored background # Text overlay cv2.putText(img, label, (x + 7, y - 5), font, font_scale, (255, 255, 255), text_thickness, cv2.LINE_AA) def _process_attendance_trigger(self, emp_id: str, emp_name: str, match_score: float): """Triggers and logs a check-in or check-out event with robust debounce logic.""" now = datetime.now() # 1. Check if user is in debounce cooldown if emp_id in self.log_cooldowns: last_log_time = self.log_cooldowns[emp_id] if now - last_log_time < self.cooldown_duration: # In cooldown, do nothing return "" # 2. Fetch last attendance log from DB last_log = get_last_attendance_log(emp_id) event_type = "check_in" if last_log: # Check if last log was today last_log_date = datetime.strptime(last_log["timestamp"], "%Y-%m-%d %H:%M:%S") if last_log_date.date() == now.date(): if last_log["event_type"] == "check_in": event_type = "check_out" else: event_type = "check_in" # 3. Log event in DB success = log_attendance(emp_id, event_type, match_score) if success: # Update cooldown timestamp self.log_cooldowns[emp_id] = now # Put notification in the alerts queue alert_data = { "employee_id": emp_id, "name": emp_name, "timestamp": now.strftime("%I:%M %p"), "event_type": event_type, "score": int(match_score * 100) } attendance_alerts.put(alert_data) # Limit alert queue size to prevent memory leaks if attendance_alerts.qsize() > 50: try: attendance_alerts.get_nowait() except queue.Empty: pass print(f"[+] ATTENDANCE REGISTERED: {emp_name} | {event_type.upper()} | score: {match_score:.3f}") return event_type return ""