Update app.py

app.py

@@ -73,8 +73,8 @@ class AttendanceSystem:
         self.is_processing = threading.Event()
         self.frame_queue = queue.Queue(maxsize=10)
         self.error_message = None 
-        self.last_processed_frame = None # Holds the final frame after processing
-        self.final_log = None # Holds the final log after processing
+        self.last_processed_frame = None
+        self.final_log = None
 
         # Data Storage
         self.known_face_embeddings: List[np.ndarray] = []
@@ -82,15 +82,15 @@ class AttendanceSystem:
         self.known_face_ids: List[str] = []
         self.next_worker_id: int = 1
         
-        # Session Tracking - Enhanced for better accuracy
+        # Session Tracking
         self.last_recognition_time = {}
-        self.recognition_cooldown = 10  # Cooldown to prevent duplicates
+        self.recognition_cooldown = 10
         self.session_log: List[str] = []
-        self.session_marked_present = set()  # Track who's already marked present in this session
-        self.session_registered = set()  # Track who's already auto-registered in this session
-        self.face_recognition_buffer = {}  # Buffer for multiple detections before confirming
-        self.buffer_threshold = 3  # Increased threshold for more reliable recognition
-        self.frame_skip_counter = 0  # Skip frames for better performance
+        self.session_marked_present = set()
+        self.session_registered = set()
+        self.face_recognition_buffer = {}
+        self.buffer_threshold = 3
+        self.frame_skip_counter = 0
 
         # Initialize
         self.sf = connect_to_salesforce()
@@ -138,7 +138,8 @@ class AttendanceSystem:
     def _load_local_worker_data(self):
         try:
             if os.path.exists("data/workers.pkl"):
-                with open("data/workers.pkl", "rb") as f: data = pickle.load(f)
+                with open("data/workers.pkl", "rb") as f: 
+                    data = pickle.load(f)
                 self.known_face_embeddings = data.get("embeddings", [])
                 self.known_face_names = data.get("names", [])
                 self.known_face_ids = data.get("ids", [])
@@ -149,8 +150,14 @@ class AttendanceSystem:
 
     def save_local_worker_data(self):
         try:
-            worker_data = {"embeddings": self.known_face_embeddings, "names": self.known_face_names, "ids": self.known_face_ids, "next_id": self.next_worker_id}
-            with open("data/workers.pkl", "wb") as f: pickle.dump(worker_data, f)
+            worker_data = {
+                "embeddings": self.known_face_embeddings,
+                "names": self.known_face_names,
+                "ids": self.known_face_ids,
+                "next_id": self.next_worker_id
+            }
+            with open("data/workers.pkl", "wb") as f: 
+                pickle.dump(worker_data, f)
         except Exception as e:
             logger.error(f"❌ Error saving local worker data: {e}")
     
@@ -178,7 +185,7 @@ class AttendanceSystem:
 
     def _register_worker_auto(self, face_image: np.ndarray, face_embedding: List[float]) -> Optional[Tuple[str, str]]:
         try:
-            # Check for duplicates with strict threshold for auto-registration
+            # Check for duplicates with strict threshold
             if self._is_duplicate_face(face_embedding, threshold=10.0):
                 return None
                 
@@ -213,9 +220,15 @@ class AttendanceSystem:
         caption = self._get_image_caption(face_pil)
         if self.sf:
             try:
-                worker_record = self.sf.Worker__c.create({'Name': name, 'Worker_ID__c': worker_id, 'Face_Embedding__c': json.dumps(embedding), 'Image_Caption__c': caption})
+                worker_record = self.sf.Worker__c.create({
+                    'Name': name,
+                    'Worker_ID__c': worker_id,
+                    'Face_Embedding__c': json.dumps(embedding),
+                    'Image_Caption__c': caption
+                })
                 image_url = self._upload_image_to_salesforce(face_pil, worker_record['id'], worker_id)
-                if image_url: self.sf.Worker__c.update(worker_record['id'], {'Image_URL__c': image_url})
+                if image_url: 
+                    self.sf.Worker__c.update(worker_record['id'], {'Image_URL__c': image_url})
                 logger.info(f"✅ Worker {worker_id} synced to Salesforce.")
             except Exception as e:
                 logger.error(f"❌ Salesforce sync error for {worker_id}: {e}")
@@ -227,18 +240,20 @@ class AttendanceSystem:
         
         embedding_array = np.array(embedding)
         for known_embedding in self.known_face_embeddings:
-            # Use cosine similarity for more accurate face matching
-            cosine_sim = np.dot(embedding_array, known_embedding) / (np.linalg.norm(embedding_array) * np.linalg.norm(known_embedding))
+            # Normalize vectors
+            embedding_array_norm = embedding_array / np.linalg.norm(embedding_array)
+            known_embedding_norm = known_embedding / np.linalg.norm(known_embedding)
+            
+            # Calculate cosine similarity
+            cosine_sim = np.dot(embedding_array_norm, known_embedding_norm)
             
-            # Consider it duplicate only if similarity is very high
-            if cosine_sim > 0.85:  # Increased from 0.80 to be more strict
+            if cosine_sim > 0.85:  # Strict threshold
                 return True
         
         return False
 
     def mark_attendance(self, worker_id: str, worker_name: str) -> bool:
-        """Strict attendance marking with session tracking"""
-        # Check if already marked present in this session
+        """Attendance marking with strict checks"""
         if worker_id in self.session_marked_present:
             return False
             
@@ -259,7 +274,6 @@ class AttendanceSystem:
             except Exception as e:
                 logger.error(f"❌ Error saving attendance to Salesforce: {e}")
         
-        # Mark as present in this session
         self.session_marked_present.add(worker_id)
         
         log_msg = f"✅ [{current_time.strftime('%H:%M:%S')}] Marked Present: {worker_name} ({worker_id})"
@@ -272,25 +286,26 @@ class AttendanceSystem:
             return True
         if self.sf:
             try:
-                if self.sf.query(f"SELECT Id FROM Attendance__c WHERE Worker_ID__c = '{worker_id}' AND Date__c = '{today_str}'")['totalSize'] > 0:
+                query = f"SELECT Id FROM Attendance__c WHERE Worker_ID__c = '{worker_id}' AND Date__c = '{today_str}'"
+                if self.sf.query(query)['totalSize'] > 0:
                     return True
-            except Exception: 
-                pass
+            except Exception as e:
+                logger.error(f"Attendance check error: {e}")
         return False
 
     def _find_best_match(self, target_embedding: np.ndarray) -> Tuple[int, float]:
-        """Find best match using cosine similarity with strict threshold"""
+        """Find best match using cosine similarity"""
         if not self.known_face_embeddings:
-            return -1, float('inf')
+            return -1, 0.0
             
         best_match_idx = -1
-        best_score = 0.0  # Using similarity score (higher is better)
+        best_score = 0.0
+        
+        target_norm = target_embedding / np.linalg.norm(target_embedding)
         
         for i, known_embedding in enumerate(self.known_face_embeddings):
-            # Cosine similarity
-            cosine_sim = np.dot(target_embedding, known_embedding) / (
-                np.linalg.norm(target_embedding) * np.linalg.norm(known_embedding)
-            )
+            known_norm = known_embedding / np.linalg.norm(known_embedding)
+            cosine_sim = np.dot(target_norm, known_norm)
             
             if cosine_sim > best_score:
                 best_score = cosine_sim
@@ -300,215 +315,237 @@ class AttendanceSystem:
 
     # --- Video Processing ---
     def process_frame(self, frame: np.ndarray) -> np.ndarray:
-        """
-        Strict frame processing with high confidence requirements
-        """
+        """Frame processing with strict recognition rules"""
         try:
-            # Skip frames for better performance
+            # Skip frames for performance
             self.frame_skip_counter += 1
-            if self.frame_skip_counter % 3 != 0:  # Process every 3rd frame
+            if self.frame_skip_counter % 3 != 0:
                 return frame
                 
-            # Use multiple detection backends for better results
+            # Detect faces with multiple backends
             face_objs = []
             try:
-                face_objs = DeepFace.extract_faces(img_path=frame, detector_backend='opencv', enforce_detection=False)
-            except:
+                face_objs = DeepFace.extract_faces(
+                    img_path=frame,
+                    detector_backend='opencv',
+                    enforce_detection=False
+                )
+            except Exception as e:
+                logger.warning(f"OpenCV detector failed: {e}")
                 try:
-                    face_objs = DeepFace.extract_faces(img_path=frame, detector_backend='mtcnn', enforce_detection=False)
-                except:
-                    pass
+                    face_objs = DeepFace.extract_faces(
+                        img_path=frame,
+                        detector_backend='mtcnn',
+                        enforce_detection=False
+                    )
+                except Exception as e:
+                    logger.warning(f"MTCNN detector failed: {e}")
 
             if face_objs:
-                print(f"\n--- Frame Processed: Found {len(face_objs)} faces. ---")
+                logger.debug(f"Found {len(face_objs)} faces in frame")
 
-            for i, face_obj in enumerate(face_objs):
+            for face_obj in face_objs:
                 confidence = face_obj.get('confidence', 0.0)
-                print(f"  Face #{i+1}: Confidence Score = {confidence:.2f}")
-
+                
                 # Strict confidence threshold
-                if confidence < 0.90:  # Increased from 0.85 to be more strict
-                    print("    -> Confidence too low, skipping.")
+                if confidence < 0.90:
                     continue
 
-                # Extract facial area
                 facial_area = face_obj['facial_area']
                 x, y, w, h = facial_area['x'], facial_area['y'], facial_area['w'], facial_area['h']
-                
                 face_image = frame[y:y+h, x:x+w]
                 
-                if face_image.size == 0: 
-                    continue
-
-                # Minimum face size check
-                if w < 50 or h < 50:
-                    print("    -> Face too small, skipping.")
+                if face_image.size == 0 or w < 50 or h < 50:
                     continue
 
-                # Get embedding
                 try:
-                    embedding = DeepFace.represent(img_path=face_image, model_name='Facenet', enforce_detection=False)[0]['embedding']
+                    embedding_obj = DeepFace.represent(
+                        img_path=face_image,
+                        model_name='Facenet',
+                        enforce_detection=False
+                    )
+                    embedding = embedding_obj[0]['embedding']
                     embedding_array = np.array(embedding)
                 except Exception as e:
-                    print(f"    -> Could not generate embedding: {e}, skipping.")
+                    logger.warning(f"Embedding generation failed: {e}")
                     continue
                 
-                color, worker_id, worker_name = (0, 0, 255), None, "Unknown"
+                color = (0, 0, 255)  # Default red for unknown
+                worker_id = None
+                worker_name = "Unknown"
                 
                 if self.known_face_embeddings:
-                    # Strict matching with cosine similarity
                     match_index, match_score = self._find_best_match(embedding_array)
                     
-                    print(f"    -> Comparing to DB... Cosine Similarity: {match_score:.4f}")
-
-                    # Strict threshold for recognition
-                    if match_index != -1 and match_score > 0.85:  # Increased from 0.80
+                    # Strict matching threshold
+                    if match_index != -1 and match_score > 0.85:
                         worker_id = self.known_face_ids[match_index]
                         worker_name = self.known_face_names[match_index]
-                        color = (0, 255, 0)  # Green
-                        print(f"    ✓ MATCH! Recognized as {worker_name}")
+                        color = (0, 255, 0)  # Green for known
                         
-                        # Use buffering for consistent recognition
+                        # Buffer recognition
                         buffer_key = f"{worker_id}"
                         if buffer_key not in self.face_recognition_buffer:
-                            self.face_recognition_buffer[buffer_key] = {'count': 1, 'last_time': time.time()}
+                            self.face_recognition_buffer[buffer_key] = {
+                                'count': 1,
+                                'last_time': time.time()
+                            }
                         else:
                             self.face_recognition_buffer[buffer_key]['count'] += 1
                             self.face_recognition_buffer[buffer_key]['last_time'] = time.time()
                         
-                        # Mark attendance only after consistent detections and if confidence is high
-                        if (self.face_recognition_buffer[buffer_key]['count'] >= self.buffer_threshold and 
+                        # Mark attendance after consistent detections
+                        if (self.face_recognition_buffer[buffer_key]['count'] >= self.buffer_threshold and
                             confidence >= 0.90):
                             if self.mark_attendance(worker_id, worker_name):
                                 self.last_recognition_time[worker_id] = time.time()
-                            # Reset buffer after marking
                             del self.face_recognition_buffer[buffer_key]
-                            
                     else:
-                        # Only register as new if similarity is very low
-                        if match_score < 0.70:  # Only register if very different from existing faces
-                            color = (0, 165, 255)  # Orange for potential new worker
-                            print(f"    ✗ NO MATCH. Attempting to register as new worker...")
+                        # Only register new if very different from existing faces
+                        if match_score < 0.70:  # Low similarity threshold
+                            color = (0, 165, 255)  # Orange for new
                             new_worker = self._register_worker_auto(face_image, embedding)
                             if new_worker:
-                                worker_id, worker_name = new_worker[0], new_worker[1]
-                                if confidence >= 0.90 and self.mark_attendance(worker_id, worker_name):
-                                    self.last_recognition_time[worker_id] = time.time()
-                        else:
-                            print("    ✗ Uncertain match, not registering as new worker.")
+                                worker_id, worker_name = new_worker
+                                if confidence >= 0.90:
+                                    self.mark_attendance(worker_id, worker_name)
                 else:
-                    # No known faces, auto-register only if confidence is high
+                    # No known faces, auto-register with high confidence
                     if confidence >= 0.90:
-                        color = (0, 165, 255)  # Orange for new worker
-                        print("    -> No known faces in database. Auto-registering...")
+                        color = (0, 165, 255)  # Orange for new
                         new_worker = self._register_worker_auto(face_image, embedding)
                         if new_worker:
-                            worker_id, worker_name = new_worker[0], new_worker[1]
-                            if self.mark_attendance(worker_id, worker_name):
-                                self.last_recognition_time[worker_id] = time.time()
+                            worker_id, worker_name = new_worker
+                            self.mark_attendance(worker_id, worker_name)
                 
                 # Clean old buffer entries
                 current_time = time.time()
-                keys_to_remove = [k for k, v in self.face_recognition_buffer.items() 
-                                if current_time - v['last_time'] > 5.0]
-                for key in keys_to_remove:
-                    del self.face_recognition_buffer[key]
+                for key in list(self.face_recognition_buffer.keys()):
+                    if current_time - self.face_recognition_buffer[key]['last_time'] > 5.0:
+                        del self.face_recognition_buffer[key]
                 
+                # Draw bounding box and label
                 label = f"{worker_name}" + (f" ({worker_id})" if worker_id else "")
                 cv2.rectangle(frame, (x, y), (x+w, y+h), color, 2)
                 cv2.putText(frame, label, (x, y-10), cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2)
 
             return frame
         except Exception as e:
-            print(f"ERROR in process_frame: {e}")
+            logger.error(f"Frame processing error: {e}")
             return frame
 
     def _processing_loop(self, source):
         video_capture = cv2.VideoCapture(source)
         if not video_capture.isOpened():
-            err_msg = f"❌ **Error:** Could not open video source. The file may be corrupt or in an unsupported format. Please try converting it to a standard MP4."
+            err_msg = "❌ Could not open video source"
             self.error_message = err_msg
             self.is_processing.clear()
             return
+            
         while self.is_processing.is_set():
             ret, frame = video_capture.read()
-            if not ret: break
+            if not ret:
+                break
+                
             processed_frame = self.process_frame(frame)
-            if not self.frame_queue.full(): self.frame_queue.put(processed_frame)
-            self.last_processed_frame = processed_frame # Continuously update last frame
+            if not self.frame_queue.full():
+                self.frame_queue.put(processed_frame)
+            self.last_processed_frame = processed_frame
             time.sleep(0.05)
-        self.final_log = self.session_log.copy() # Save the final log
+            
+        self.final_log = self.session_log.copy()
         video_capture.release()
         self.is_processing.clear()
 
     def start_processing(self, source) -> str:
-        if self.is_processing.is_set(): return "⚠️ Processing is already active."
-        # Reset states for the new session
+        if self.is_processing.is_set():
+            return "⚠️ Processing is already active."
+            
+        # Reset session state
         self.session_log.clear()
         self.last_recognition_time.clear()
-        self.session_marked_present.clear()  # Reset session attendance tracking
-        self.session_registered.clear()  # Reset session registration tracking
-        self.face_recognition_buffer.clear()  # Reset recognition buffer
+        self.session_marked_present.clear()
+        self.session_registered.clear()
+        self.face_recognition_buffer.clear()
         self.error_message = None
         self.last_processed_frame = None
         self.final_log = None
         self.frame_skip_counter = 0
+        
         self.is_processing.set()
-        self.processing_thread = threading.Thread(target=self._processing_loop, args=(source,))
-        self.processing_thread.daemon = True
+        self.processing_thread = threading.Thread(
+            target=self._processing_loop,
+            args=(source,),
+            daemon=True
+        )
         self.processing_thread.start()
-        return f"✅ Started processing..."
+        return "✅ Started processing..."
 
     def stop_processing(self) -> str:
-        # Reset states when stopping manually
         self.is_processing.clear()
         self.error_message = None
         self.last_processed_frame = None
         self.final_log = None
         self.face_recognition_buffer.clear()
-        return "✅ Processing stopped by user."
+        return "✅ Processing stopped."
 
-    # --- Helper & Reporting ---
+    # --- Helper Methods ---
     def _get_image_caption(self, image: Image.Image) -> str:
-        if not HF_API_TOKEN: return "Hugging Face API token not configured."
+        if not HF_API_TOKEN:
+            return "Hugging Face API token not configured."
         try:
             buffered = BytesIO()
             image.save(buffered, format="JPEG")
-            img_data = buffered.getvalue()
             headers = {"Authorization": f"Bearer {HF_API_TOKEN}"}
-            response = requests.post(HF_API_URL, headers=headers, data=img_data)
+            response = requests.post(HF_API_URL, headers=headers, data=buffered.getvalue())
             response.raise_for_status()
-            result = response.json()
-            return result[0].get("generated_text", "No caption found.")
+            return response.json()[0].get("generated_text", "No caption found.")
         except Exception as e:
             logger.error(f"Hugging Face API error: {e}")
             return "Caption generation failed."
 
     def _upload_image_to_salesforce(self, image: Image.Image, record_id: str, worker_id: str) -> Optional[str]:
-        if not self.sf: return None
+        if not self.sf:
+            return None
         try:
             buffered = BytesIO()
             image.save(buffered, format="JPEG")
             encoded_image = base64.b64encode(buffered.getvalue()).decode('utf-8')
-            cv = self.sf.ContentVersion.create({'Title': f'Image_{worker_id}', 'PathOnClient': f'{worker_id}.jpg', 'VersionData': encoded_image, 'FirstPublishLocationId': record_id})
-            return f"/{cv['id']}" # Relative URL
+            cv = self.sf.ContentVersion.create({
+                'Title': f'Image_{worker_id}',
+                'PathOnClient': f'{worker_id}.jpg',
+                'VersionData': encoded_image,
+                'FirstPublishLocationId': record_id
+            })
+            return f"/{cv['id']}"
         except Exception as e:
             logger.error(f"Salesforce image upload error: {e}")
             return None
 
     def get_registered_workers_info(self) -> str:
-        if not self.sf: return "❌ Salesforce not connected."
+        if not self.sf:
+            return "❌ Salesforce not connected."
         try:
-            records = self.sf.query_all("SELECT Name, Worker_ID__c FROM Worker__c ORDER BY Name")['records']
-            if not records: return "No workers registered."
-            return f"**👥 Registered Workers ({len(records)})**\n" + "\n".join([f"- **{w['Name']}** (ID: {w['Worker_ID__c']})" for w in records])
-        except Exception as e: return f"Error: {e}"
-        
+            records = self.sf.query_all(
+                "SELECT Name, Worker_ID__c FROM Worker__c ORDER BY Name"
+            )['records']
+            if not records:
+                return "No workers registered."
+            worker_list = "\n".join(
+                f"- **{w['Name']}** (ID: {w['Worker_ID__c']})"
+                for w in records
+            )
+            return f"**👥 Registered Workers ({len(records)})**\n{worker_list}"
+        except Exception as e:
+            return f"Error: {e}"
+
 # --- GRADIO UI ---
 attendance_system = AttendanceSystem()
+
 def create_interface():
     with gr.Blocks(theme=gr.themes.Soft(), title="Attendance System") as demo:
         gr.Markdown("# 🎯 Advanced Face Recognition Attendance System")
+        
         with gr.Tabs():
             with gr.Tab("⚙️ Controls & Status"):
                 gr.Markdown("### 1. Choose Input Source & Start Processing")
@@ -546,29 +583,57 @@ def create_interface():
                         refresh_workers_btn = gr.Button("🔄 Refresh List")
 
         # --- Event Handlers ---
-        def on_tab_select(evt: gr.SelectData): return evt.index
+        def on_tab_select(evt: gr.SelectData):
+            return evt.index
+            
         video_tabs.select(fn=on_tab_select, inputs=None, outputs=[selected_tab_index])
+        
         def start_wrapper(tab_index, cam_src, vid_path):
             source = cam_src if tab_index == 0 else vid_path
-            return "Please provide an input source." if source is None else attendance_system.start_processing(source)
-        start_btn.click(fn=start_wrapper, inputs=[selected_tab_index, camera_source, video_file], outputs=[status_box])
-        stop_btn.click(fn=attendance_system.stop_processing, inputs=None, outputs=[status_box])
-        register_btn.click(fn=attendance_system.register_worker_manual, inputs=[register_image, register_name], outputs=[register_output, registered_workers_info])
-        refresh_workers_btn.click(fn=attendance_system.get_registered_workers_info, outputs=[registered_workers_info])
+            if source is None:
+                return "Please provide an input source."
+            return attendance_system.start_processing(source)
+            
+        start_btn.click(
+            fn=start_wrapper,
+            inputs=[selected_tab_index, camera_source, video_file],
+            outputs=[status_box]
+        )
+        
+        stop_btn.click(
+            fn=attendance_system.stop_processing,
+            inputs=None,
+            outputs=[status_box]
+        )
+        
+        register_btn.click(
+            fn=attendance_system.register_worker_manual,
+            inputs=[register_image, register_name],
+            outputs=[register_output, registered_workers_info]
+        )
+        
+        refresh_workers_btn.click(
+            fn=attendance_system.get_registered_workers_info,
+            outputs=[registered_workers_info]
+        )
         
         def update_ui_generator():
             while True:
                 if attendance_system.error_message:
                     yield None, attendance_system.error_message
-                    time.sleep(2); attendance_system.error_message = None
+                    time.sleep(2)
+                    attendance_system.error_message = None
                     continue
+                    
                 if attendance_system.is_processing.is_set():
                     frame, log_md = None, "\n".join(reversed(attendance_system.session_log)) or "Processing..."
                     try:
                         if not attendance_system.frame_queue.empty():
                             frame = attendance_system.frame_queue.get_nowait()
-                            if frame is not None: frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-                    except queue.Empty: pass
+                            if frame is not None:
+                                frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+                    except queue.Empty:
+                        pass
                     yield frame, log_md
                 else:
                     if attendance_system.last_processed_frame is not None:
@@ -579,7 +644,10 @@ def create_interface():
                         yield None, "System stopped. Go to 'Controls & Status' to start."
                 time.sleep(0.1)
 
-        demo.load(fn=update_ui_generator, outputs=[video_output, session_log_display])
+        demo.load(
+            fn=update_ui_generator,
+            outputs=[video_output, session_log_display]
+        )
     return demo
 
 if __name__ == "__main__":