Update app.py

app.py

@@ -12,6 +12,7 @@ from datetime import datetime, date
 from io import BytesIO
 from typing import Tuple, Optional, List, Dict
 import pickle
+from concurrent.futures import ThreadPoolExecutor
 
 # Third-Party Imports
 import cv2
@@ -26,18 +27,14 @@ from simple_salesforce import Salesforce
 
 # --- CONFIGURATION ---
 
-# Setup logging
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
 logger = logging.getLogger(__name__)
 
-# Load environment variables from .env file
 load_dotenv()
 
-# Hugging Face API configuration
 HF_API_URL = "https://api-inference.huggingface.co/models/Salesforce/blip-image-captioning-base"
 HF_API_TOKEN = os.getenv("HUGGINGFACE_API_TOKEN")
 
-# Salesforce configuration
 SF_CREDENTIALS = {
     "username": os.getenv("SF_USERNAME", "smartlabour@attendance.system"),
     "password": os.getenv("SF_PASSWORD", "#Prashanth@1234"),
@@ -49,7 +46,6 @@ SF_CREDENTIALS = {
 
 @retry(stop_max_attempt_number=3, wait_fixed=2000)
 def connect_to_salesforce() -> Optional[Salesforce]:
-    """Establish a connection to Salesforce with retry logic."""
     try:
         sf = Salesforce(**SF_CREDENTIALS)
         sf.describe()
@@ -62,39 +58,30 @@ def connect_to_salesforce() -> Optional[Salesforce]:
 # --- CORE LOGIC ---
 
 class AttendanceSystem:
-    """
-    Manages all backend logic for the face recognition attendance system.
-    """
     def __init__(self):
-        # State Management
         self.processing_thread = None
         self.is_processing = threading.Event()
         self.frame_queue = queue.Queue(maxsize=10)
-        self.error_message = None 
+        self.error_message = None
         self.last_processed_frame = None
         self.final_log = None
         self.worker_statuses = {}
-
-        # Data Storage
-        self.known_face_embeddings: List[np.ndarray] = []
-        self.known_face_names: List[str] = []
-        self.known_face_ids: List[str] = []
-        self.next_worker_id: int = 1
-        
-        # Session Tracking
-        self.last_recognition_time: Dict[str, float] = {}
+        self.known_face_embeddings = []
+        self.known_face_names = []
+        self.known_face_ids = []
+        self.next_worker_id = 1
+        self.last_recognition_time = {}
         self.recognition_cooldown = 5
-        self.session_log: List[str] = []
-        self.today_attendance: Dict[str, bool] = {}
-        self.last_detected_faces: Dict[str, float] = {}
-        self.processed_workers: set = set()
-
-        # Performance optimization
-        self.frame_skip = 4  # Process every 5th frame
+        self.session_log = []
+        self.today_attendance = {}
+        self.last_detected_faces = {}
+        self.processed_workers = set()
+        self.frame_skip = 9  # Process every 10th frame
         self.frame_counter = 0
-
-        # Initialize
+        self.total_frames = 0
+        self.processed_frames = 0
         self.sf = connect_to_salesforce()
+        self.executor = ThreadPoolExecutor(max_workers=2)  # For async Salesforce uploads
         self._create_directories()
         self.load_worker_data()
 
@@ -109,7 +96,6 @@ class AttendanceSystem:
                 if not workers:
                     self._load_local_worker_data()
                     return
-
                 temp_embeddings, temp_names, temp_ids, max_id = [], [], [], 0
                 for worker in workers:
                     if worker.get('Face_Embedding__c'):
@@ -122,7 +108,6 @@ class AttendanceSystem:
                                 max_id = worker_num
                         except (ValueError, TypeError):
                             continue
-                
                 self.known_face_embeddings = temp_embeddings
                 self.known_face_names = temp_names
                 self.known_face_ids = temp_ids
@@ -139,7 +124,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", [])
@@ -150,24 +136,31 @@ 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}")
-    
+
     def register_worker_manual(self, image: Image.Image, name: str) -> Tuple[str, str]:
         if image is None or not name.strip():
             return "❌ Please provide both image and name!", self.get_registered_workers_info()
         try:
             image_array = np.array(image)
+            # Preprocess image
+            image_array = cv2.equalizeHist(cv2.cvtColor(image_array, cv2.COLOR_RGB2GRAY))
+            image_array = cv2.cvtColor(image_array, cv2.COLOR_GRAY2RGB)
             analysis = DeepFace.analyze(img_path=image_array, actions=['emotion'], enforce_detection=True)
             if analysis[0]['face_confidence'] < 0.9:
                 return "❌ Face not clear enough for registration!", self.get_registered_workers_info()
-            
             embedding = DeepFace.represent(img_path=image_array, model_name='Facenet')[0]['embedding']
-            if self._is_duplicate_face(embedding):
-                 return f"❌ Face matches an existing worker!", self.get_registered_workers_info()
-
+            if self._is_duplicate_face(embedding, threshold=8.0):  # Adjusted threshold
+                return f"❌ Face matches an existing worker!", self.get_registered_workers_info()
             worker_id = f"W{self.next_worker_id:04d}"
             name = name.strip().title()
             self._add_worker_to_system(worker_id, name, embedding, image_array)
@@ -181,10 +174,12 @@ class AttendanceSystem:
 
     def _register_worker_auto(self, face_image: np.ndarray) -> Optional[Tuple[str, str]]:
         try:
+            # Preprocess image
+            face_image = cv2.equalizeHist(cv2.cvtColor(face_image, cv2.COLOR_BGR2GRAY))
+            face_image = cv2.cvtColor(face_image, cv2.COLOR_GRAY2RGB)
             embedding = DeepFace.represent(img_path=face_image, model_name='Facenet', enforce_detection=False)[0]['embedding']
-            if self._is_duplicate_face(embedding): 
+            if self._is_duplicate_face(embedding, threshold=8.0):
                 return None
-                
             worker_id = f"W{self.next_worker_id:04d}"
             worker_name = f"Worker {self.next_worker_id}"
             self._add_worker_to_system(worker_id, worker_name, embedding, face_image)
@@ -204,46 +199,43 @@ class AttendanceSystem:
         self.known_face_ids.append(worker_id)
         self.next_worker_id += 1
         face_pil = Image.fromarray(cv2.cvtColor(image_array, cv2.COLOR_BGR2RGB))
-        face_pil.save(f"data/faces/{worker_id}.jpg")
+        face_pil.save(f"data/faces/{worker_id}.jpg", quality=85)  # Compress image
         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})
-                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})
-                logger.info(f"✅ Worker {worker_id} synced to Salesforce.")
-            except Exception as e:
-                logger.error(f"❌ Salesforce sync error for {worker_id}: {e}")
+            self.executor.submit(self._async_salesforce_upload, worker_id, name, embedding, caption, face_pil)
 
-    def _is_duplicate_face(self, embedding: List[float], threshold: float = 10.0) -> bool:
-        if not self.known_face_embeddings: return False
+    def _async_salesforce_upload(self, worker_id: str, name: str, embedding: List[float], caption: str, image: Image.Image):
+        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
+            })
+            image_url = self._upload_image_to_salesforce(image, worker_record['id'], worker_id)
+            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}")
+
+    def _is_duplicate_face(self, embedding: List[float], threshold: float = 8.0) -> bool:
+        if not self.known_face_embeddings:
+            return False
         distances = [np.linalg.norm(np.array(embedding) - known_embedding) for known_embedding in self.known_face_embeddings]
         return min(distances) < threshold
 
     def mark_attendance(self, worker_id: str, worker_name: str) -> bool:
         if worker_id in self.processed_workers:
             return False
-            
         today_str = date.today().isoformat()
         if worker_id in self.today_attendance:
             return False
-            
         current_time = time.time()
         if worker_id in self.last_recognition_time and (current_time - self.last_recognition_time[worker_id] < self.recognition_cooldown):
             return False
-            
         if self.sf:
-            try:
-                self.sf.Attendance__c.create({
-                    'Worker_ID__c': worker_id, 
-                    'Name__c': worker_name, 
-                    'Date__c': today_str, 
-                    'Timestamp__c': datetime.now().isoformat(), 
-                    'Status__c': "Present"
-                })
-            except Exception as e:
-                logger.error(f"❌ Error saving attendance to Salesforce: {e}")
-        
+            self.executor.submit(self._async_attendance_upload, worker_id, worker_name, today_str)
         self.today_attendance[worker_id] = True
         self.last_recognition_time[worker_id] = current_time
         self.processed_workers.add(worker_id)
@@ -252,63 +244,67 @@ class AttendanceSystem:
         self.session_log.append(log_msg)
         return True
 
+    def _async_attendance_upload(self, worker_id: str, worker_name: str, today_str: str):
+        try:
+            self.sf.Attendance__c.create({
+                'Worker_ID__c': worker_id,
+                'Name__c': worker_name,
+                'Date__c': today_str,
+                'Timestamp__c': datetime.now().isoformat(),
+                'Status__c': "Present"
+            })
+        except Exception as e:
+            logger.error(f"❌ Error saving attendance to Salesforce: {e}")
+
     def process_frame(self, frame: np.ndarray) -> np.ndarray:
         try:
             self.frame_counter += 1
+            self.processed_frames += 1
             if self.frame_counter % (self.frame_skip + 1) != 0:
                 return frame
-
             height, width = frame.shape[:2]
-            new_width = 480
+            new_width = 320  # Reduced resolution
             new_height = int((new_width / width) * height)
             small_frame = cv2.resize(frame, (new_width, new_height))
-
+            # Preprocess frame
+            small_frame = cv2.equalizeHist(cv2.cvtColor(small_frame, cv2.COLOR_BGR2GRAY))
+            small_frame = cv2.cvtColor(small_frame, cv2.COLOR_GRAY2RGB)
             face_objs = DeepFace.extract_faces(
-                img_path=small_frame, 
-                detector_backend='fastmtcnn',
+                img_path=small_frame,
+                detector_backend='ssd',  # Faster detector
                 enforce_detection=False,
                 align=True
             )
-
             for face_obj in face_objs:
                 confidence = face_obj['confidence']
                 if confidence < 0.9:
                     continue
-
                 facial_area = face_obj['facial_area']
                 x, y, w, h = facial_area['x'], facial_area['y'], facial_area['w'], facial_area['h']
-                
                 x = int(x * width / new_width)
                 y = int(y * height / new_height)
                 w = int(w * width / new_width)
                 h = int(h * height / new_height)
-                
                 face_image = frame[y:y+h, x:x+w]
-                if face_image.size == 0: 
+                if face_image.size == 0:
                     continue
-
                 current_time = time.time()
                 face_key = f"{x}_{y}_{w}_{h}"
                 if face_key in self.last_detected_faces and (current_time - self.last_detected_faces[face_key] < 2.0):
                     continue
                 self.last_detected_faces[face_key] = current_time
-
                 embedding = DeepFace.represent(
-                    img_path=face_image, 
-                    model_name='Facenet', 
+                    img_path=face_image,
+                    model_name='Facenet',
                     enforce_detection=False,
                     align=True
                 )[0]['embedding']
-                
                 if not self.known_face_embeddings:
                     continue
-
                 distances = [np.linalg.norm(np.array(embedding) - known) for known in self.known_face_embeddings]
                 min_dist = min(distances) if distances else float('inf')
-                match_index = distances.index(min_dist) if min_dist < 10.0 else -1
-
+                match_index = distances.index(min_dist) if min_dist < 8.0 else -1
                 color, worker_id, worker_name = (0, 0, 255), None, "Unknown"
-                
                 if match_index != -1:
                     worker_id = self.known_face_ids[match_index]
                     worker_name = self.known_face_names[match_index]
@@ -321,11 +317,9 @@ class AttendanceSystem:
                     if new_worker:
                         worker_id, worker_name = new_worker
                         self.mark_attendance(worker_id, worker_name)
-                
                 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:
             logger.error(f"ERROR in process_frame: {e}")
@@ -338,31 +332,27 @@ class AttendanceSystem:
             self.error_message = err_msg
             self.is_processing.clear()
             return
-            
+        self.total_frames = int(video_capture.get(cv2.CAP_PROP_FRAME_COUNT))
+        self.processed_frames = 0
         if isinstance(source, int):
             video_capture.set(cv2.CAP_PROP_FPS, 30)
-            
         while self.is_processing.is_set():
             ret, frame = video_capture.read()
-            if not ret: 
+            if not ret:
                 break
-                
             processed_frame = self.process_frame(frame)
-            
-            if not self.frame_queue.full(): 
+            if not self.frame_queue.full():
                 self.frame_queue.put(processed_frame)
-                
             self.last_processed_frame = processed_frame
-            time.sleep(0.01)
-            
+            time.sleep(0.005)  # Reduced sleep time
         self.final_log = self.session_log.copy()
         video_capture.release()
         self.is_processing.clear()
+        self.executor.shutdown(wait=True)  # Ensure all async tasks complete
 
     def start_processing(self, source) -> str:
-        if self.is_processing.is_set(): 
+        if self.is_processing.is_set():
             return "⚠️ Processing is already active."
-            
         self.session_log.clear()
         self.last_recognition_time.clear()
         self.today_attendance.clear()
@@ -373,10 +363,11 @@ class AttendanceSystem:
         self.last_processed_frame = None
         self.final_log = None
         self.frame_counter = 0
-        
+        self.total_frames = 0
+        self.processed_frames = 0
         self.is_processing.set()
         self.processing_thread = threading.Thread(
-            target=self._processing_loop, 
+            target=self._processing_loop,
             args=(source,),
             daemon=True
         )
@@ -387,15 +378,16 @@ class AttendanceSystem:
         self.is_processing.clear()
         self.error_message = None
         self.last_processed_frame = None
-        self.final_log = None
+        self.final_log = self.session_log.copy()
+        self.executor.shutdown(wait=True)
         return "✅ Processing stopped by user."
 
     def _get_image_caption(self, image: Image.Image) -> str:
-        if not HF_API_TOKEN: 
+        if not HF_API_TOKEN:
             return "Hugging Face API token not configured."
         try:
             buffered = BytesIO()
-            image.save(buffered, format="JPEG")
+            image.save(buffered, format="JPEG", quality=85)
             img_data = buffered.getvalue()
             headers = {"Authorization": f"Bearer {HF_API_TOKEN}"}
             response = requests.post(HF_API_URL, headers=headers, data=img_data)
@@ -407,11 +399,11 @@ class AttendanceSystem:
             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: 
+        if not self.sf:
             return None
         try:
             buffered = BytesIO()
-            image.save(buffered, format="JPEG")
+            image.save(buffered, format="JPEG", quality=85)
             encoded_image = base64.b64encode(buffered.getvalue()).decode('utf-8')
             cv = self.sf.ContentVersion.create({
                 'Title': f'Image_{worker_id}',
@@ -425,16 +417,16 @@ class AttendanceSystem:
             return None
 
     def get_registered_workers_info(self) -> str:
-        if not self.sf: 
+        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: 
+            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: 
+        except Exception as e:
             return f"Error: {e}"
 
     def get_worker_statuses(self) -> str:
@@ -444,7 +436,14 @@ class AttendanceSystem:
             [status for status in self.worker_statuses.values()]
         )
 
+    def get_processing_progress(self) -> str:
+        if self.total_frames > 0:
+            progress = (self.processed_frames / self.total_frames) * 100
+            return f"Processing: {progress:.1f}% complete"
+        return "Processing..."
+
 # --- GRADIO UI ---
+
 attendance_system = AttendanceSystem()
 
 def create_interface():
@@ -465,9 +464,9 @@ def create_interface():
                         start_btn = gr.Button("▶️ Start Processing", variant="primary")
                         stop_btn = gr.Button("⏹️ Stop Processing", variant="stop")
                         status_box = gr.Textbox(label="Status", interactive=False, value="System Ready.")
+                        progress_box = gr.Textbox(label="Progress", interactive=False, value="Ready")
                 gr.Markdown("### 2. View Results in the 'Output & Log' Tab")
                 gr.Markdown("**🎨 Color Coding:** <font color='green'>Green</font> = Known, <font color='orange'>Orange</font> = New, <font color='red'>Red</font> = Unknown")
-            
             with gr.Tab("📊 Output & Log"):
                 with gr.Row():
                     with gr.Column(scale=2):
@@ -475,7 +474,6 @@ def create_interface():
                     with gr.Column(scale=1):
                         session_log_display = gr.Markdown(label="📋 Session Log", value="System is ready.")
                         worker_status_display = gr.Markdown(label="👥 Worker Statuses", value=attendance_system.get_worker_statuses())
-
             with gr.Tab("👤 Worker Management"):
                 with gr.Row():
                     with gr.Column():
@@ -486,70 +484,61 @@ def create_interface():
                     with gr.Column():
                         registered_workers_info = gr.Markdown(value=attendance_system.get_registered_workers_info())
                         refresh_workers_btn = gr.Button("🔄 Refresh List")
-
         # --- Event Handlers ---
-        def on_tab_select(evt: gr.SelectData): 
+        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], 
+            fn=start_wrapper,
+            inputs=[selected_tab_index, camera_source, video_file],
             outputs=[status_box]
         )
-        
         stop_btn.click(
-            fn=attendance_system.stop_processing, 
-            inputs=None, 
+            fn=attendance_system.stop_processing,
+            inputs=None,
             outputs=[status_box]
         )
-        
         register_btn.click(
-            fn=attendance_system.register_worker_manual, 
-            inputs=[register_image, register_name], 
+            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, 
+            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, attendance_system.get_worker_statuses()
+                    yield None, attendance_system.error_message, attendance_system.get_worker_statuses(), attendance_system.error_message
                     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[-20:])) or "Processing..."
                     status_md = attendance_system.get_worker_statuses()
+                    progress = attendance_system.get_processing_progress()
                     try:
                         if not attendance_system.frame_queue.empty():
                             frame = attendance_system.frame_queue.get_nowait()
-                            if frame is not None: 
+                            if frame is not None:
                                 frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-                    except queue.Empty: 
+                    except queue.Empty:
                         pass
-                    yield frame, log_md, status_md
+                    yield frame, log_md, status_md, progress
                 else:
-                    if attendance_system.last_processed_frame is not None:
-                        final_frame = cv2.cvtColor(attendance_system.last_processed_frame, cv2.COLOR_BGR2RGB)
-                        final_log_md = "\n".join(reversed(attendance_system.final_log[-20:])) or "Processing complete. No log entries."
-                        status_md = attendance_system.get_worker_statuses()
-                        yield final_frame, final_log_md, status_md
-                    else:
-                        yield None, "System stopped. Go to 'Controls & Status' to start.", attendance_system.get_worker_statuses()
-                time.sleep(0.1)
-
-        demo.load(fn=update_ui_generator, outputs=[video_output, session_log_display, worker_status_display])
+                    final_log_md = "\n".join(reversed(attendance_system.final_log[-20:])) if attendance_system.final_log else "Processing complete. No log entries."
+                    status_md = attendance_system.get_worker_statuses()
+                    final_frame = cv2.cvtColor(attendance_system.last_processed_frame, cv2.COLOR_BGR2RGB) if attendance_system.last_processed_frame is not None else None
+                    yield final_frame, final_log_md, status_md, "Processing complete."
+                time.sleep(0.05)  # Faster UI updates
+        demo.load(
+            fn=update_ui_generator,
+            outputs=[video_output, session_log_display, worker_status_display, progress_box]
+        )
     return demo
 
 if __name__ == "__main__":