# Suppress TensorFlow oneDNN warnings import os os.environ['TF_ENABLE_ONEDNN_OPTS'] = '0' import gradio as gr import cv2 import numpy as np import pandas as pd from datetime import datetime, date from typing import Tuple, Optional import logging from deepface import DeepFace import pickle from io import BytesIO import base64 from PIL import Image import json import threading import time import queue import requests from simple_salesforce import Salesforce from dotenv import load_dotenv from retrying import retry # Setup logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) # Load environment variables 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": "smartlabour@attendance.system", "password": "#Prashanth@1234", "security_token": "7xPmtDFoWlZUGK0V2QSwFZJ6c", "domain": "login" } @retry(stop_max_attempt_number=3, wait_fixed=2000) def connect_to_salesforce(): try: sf = Salesforce(**SF_CREDENTIALS) logger.info("Connected to Salesforce") sf.describe() return sf except Exception as e: logger.error(f"Salesforce connection failed: {e}") raise class AttendanceSystem: def __init__(self): self.known_face_embeddings = [] self.known_face_names = [] self.known_face_ids = [] self.attendance_records = [] self.next_worker_id = 1 self.video_capture = None self.is_streaming = False self.frame_queue = queue.Queue(maxsize=2) self.recognition_thread = None self.last_recognition_time = {} self.recognition_cooldown = 5 # seconds self.video_file_path = None self.video_processing = False # Initialize Salesforce try: self.sf = connect_to_salesforce() except Exception as e: logger.error(f"Error connecting to Salesforce: {e}") self.sf = None # Create directories os.makedirs("data", exist_ok=True) os.makedirs("data/faces", exist_ok=True) self.load_data() def load_data(self): try: if os.path.exists("data/workers.pkl"): 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", []) self.next_worker_id = data.get("next_id", 1) if os.path.exists("data/attendance.json"): with open("data/attendance.json", "r") as f: self.attendance_records = json.load(f) # Load embeddings from Salesforce for duplicate checks if self.sf: try: workers = self.sf.query_all("SELECT Worker_ID__c, Name, Face_Embedding__c FROM Worker__c")['records'] for worker in workers: if worker['Face_Embedding__c']: embedding = json.loads(worker['Face_Embedding__c']) if worker['Worker_ID__c'] not in self.known_face_ids: self.known_face_embeddings.append(embedding) self.known_face_names.append(worker['Name']) self.known_face_ids.append(worker['Worker_ID__c']) self.next_worker_id = max(self.next_worker_id, int(worker['Worker_ID__c'][1:]) + 1) except Exception as e: logger.error(f"Error loading embeddings from Salesforce: {e}") except Exception as e: logger.error(f"Error loading data: {e}") self.known_face_embeddings = [] self.known_face_names = [] self.known_face_ids = [] self.attendance_records = [] self.next_worker_id = 1 def save_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) with open("data/attendance.json", "w") as f: json.dump(self.attendance_records, f, indent=2) except Exception as e: logger.error(f"Error saving data: {e}") def get_image_caption(self, image: Image.Image) -> str: """Generate image caption using Hugging Face API""" try: img_byte_arr = BytesIO() image.save(img_byte_arr, format='JPEG', quality=85) img_data = img_byte_arr.getvalue() headers = {"Authorization": f"Bearer {HF_API_TOKEN}"} response = requests.post(HF_API_URL, headers=headers, data=img_data) if response.status_code == 200: result = response.json() if isinstance(result, list) and len(result) > 0: return result[0].get("generated_text", "No caption generated") return "No caption generated" else: logger.error(f"Hugging Face API error: {response.status_code} - {response.text}") return "Error generating caption" except Exception as e: logger.error(f"Error in Hugging Face API call: {e}") return "Error generating caption" def upload_image_to_salesforce(self, image: Image.Image, worker_salesforce_id: str, worker_id: str, worker_name: str) -> Optional[str]: """Upload worker image to Salesforce as ContentVersion""" try: if not image: logger.error("No image provided for upload") return None img_byte_arr = BytesIO() image.save(img_byte_arr, format='JPEG', quality=85) img_data = img_byte_arr.getvalue() encoded_image = base64.b64encode(img_data).decode('utf-8') content_version_data = { "Title": f"Worker_Image_{worker_id}_{worker_name.replace(' ', '_')}", "PathOnClient": f"worker_{worker_id}.jpg", "VersionData": encoded_image, "FirstPublishLocationId": worker_salesforce_id } content_version = self.sf.ContentVersion.create(content_version_data) file_url = f"https://{self.sf.sf_instance}/sfc/servlet.shepherd/version/download/{content_version['id']}" logger.info(f"Image uploaded to Salesforce for worker {worker_id}: {file_url}") return file_url except Exception as e: logger.error(f"Error uploading image to Salesforce: {e}") return None def register_worker_manual(self, image: Image.Image, name: str) -> Tuple[str, str]: """Manual worker registration with Hugging Face and Salesforce""" if image is None or not name.strip(): return "❌ Please provide both image and name!", self.get_registered_workers_info() image_array = np.array(image) try: face_analysis = DeepFace.analyze(img_path=image_array, actions=['emotion'], enforce_detection=True, detector_backend='opencv') embedding = DeepFace.represent(img_path=image_array, model_name='Facenet')[0]['embedding'] name = name.strip().title() if name in self.known_face_names: return f"❌ {name} is already registered!", self.get_registered_workers_info() # Check for duplicate face if len(self.known_face_embeddings) > 0: distances = [np.linalg.norm(np.array(embedding) - np.array(known_embedding)) for known_embedding in self.known_face_embeddings] min_distance = min(distances) if min_distance < 10: best_match_index = distances.index(min_distance) matched_name = self.known_face_names[best_match_index] matched_id = self.known_face_ids[best_match_index] return f"❌ Face matches existing worker: {matched_name} ({matched_id})!", self.get_registered_workers_info() worker_id = f"W{self.next_worker_id:04d}" caption = self.get_image_caption(image) self.known_face_embeddings.append(embedding) self.known_face_names.append(name) self.known_face_ids.append(worker_id) self.next_worker_id += 1 face_image = Image.fromarray(image_array) local_path = f"data/faces/{worker_id}_{name.replace(' ', '_')}.jpg" face_image.save(local_path) image_url = None 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_image, worker_record['id'], worker_id, name) if image_url: self.sf.Worker__c.update(worker_record['id'], {'Image_URL__c': image_url}) else: logger.warning("Image URL not set due to upload failure") except Exception as e: logger.error(f"Error saving to Salesforce: {e}") self.save_data() return f"✅ {name} has been successfully registered with ID: {worker_id}! Caption: {caption}\nImage URL: {image_url or 'Not uploaded'}", self.get_registered_workers_info() except ValueError as e: if "Face could not be detected" in str(e): return "❌ No face detected in the image!", self.get_registered_workers_info() return f"❌ Error processing image: {str(e)}", self.get_registered_workers_info() except Exception as e: return f"❌ Error during registration: {str(e)}", self.get_registered_workers_info() def register_worker_auto(self, face_image: np.ndarray) -> Tuple[Optional[str], Optional[str]]: """Automatic worker registration for unrecognized faces""" try: embedding = DeepFace.represent(img_path=face_image, model_name='Facenet')[0]['embedding'] # Check for duplicate face if len(self.known_face_embeddings) > 0: distances = [np.linalg.norm(np.array(embedding) - np.array(known_embedding)) for known_embedding in self.known_face_embeddings] min_distance = min(distances) if min_distance < 10: best_match_index = distances.index(min_distance) return self.known_face_ids[best_match_index], self.known_face_names[best_match_index] worker_id = f"W{self.next_worker_id:04d}" worker_name = f"Unknown_Worker_{self.next_worker_id}" face_pil = Image.fromarray(cv2.cvtColor(face_image, cv2.COLOR_BGR2RGB)) caption = self.get_image_caption(face_pil) self.known_face_embeddings.append(embedding) self.known_face_names.append(worker_name) self.known_face_ids.append(worker_id) self.next_worker_id += 1 local_path = f"data/faces/{worker_id}_{worker_name}.jpg" face_pil.save(local_path) image_url = None if self.sf: try: worker_record = self.sf.Worker__c.create({ 'Name': worker_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, worker_name) if image_url: self.sf.Worker__c.update(worker_record['id'], {'Image_URL__c': image_url}) except Exception as e: logger.error(f"Error saving to Salesforce: {e}") self.save_data() return worker_id, worker_name except Exception as e: logger.error(f"Error in auto registration: {e}") return None, None def mark_attendance(self, worker_id: str, worker_name: str) -> bool: try: today = date.today().isoformat() current_time = datetime.now() already_marked = any( record["worker_id"] == worker_id and record["date"] == today for record in self.attendance_records ) if not already_marked: attendance_record = { "worker_id": worker_id, "name": worker_name, "date": today, "time": current_time.strftime("%H:%M:%S"), "timestamp": current_time.isoformat(), "status": "Present", "method": "Auto" } self.attendance_records.append(attendance_record) if self.sf: try: self.sf.Attendance__c.create({ 'Worker_ID__c': worker_id, 'Name__c': worker_name, 'Date__c': today, 'Time__c': current_time.strftime("%H:%M:%S"), 'Timestamp__c': current_time.isoformat(), 'Status__c': "Present", 'Method__c': "Auto" }) logger.info(f"Attendance for {worker_name} ({worker_id}) saved to Salesforce") except Exception as e: logger.error(f"Error saving attendance to Salesforce: {e}") self.save_data() return True return False except Exception as e: logger.error(f"Error marking attendance: {e}") return False def process_video_frame(self, frame: np.ndarray) -> np.ndarray: try: rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) face_objs = DeepFace.extract_faces(img_path=rgb_frame, target_size=(160, 160), enforce_detection=False, detector_backend='opencv') current_time = time.time() for face_obj in face_objs: if face_obj['confidence'] > 0.9: face_area = face_obj['facial_area'] x, y, w, h = face_area['x'], face_area['y'], face_area['w'], face_area['h'] face_image = frame[y:y+h, x:x+w] try: embedding = DeepFace.represent(img_path=face_image, model_name='Facenet')[0]['embedding'] worker_id = None worker_name = "Unknown" color = (0, 0, 255) if len(self.known_face_embeddings) > 0: distances = [np.linalg.norm(np.array(embedding) - np.array(known_embedding)) for known_embedding in self.known_face_embeddings] min_distance = min(distances) best_match_index = distances.index(min_distance) if min_distance < 10: worker_id = self.known_face_ids[best_match_index] worker_name = self.known_face_names[best_match_index] color = (0, 255, 0) if worker_id not in self.last_recognition_time or \ current_time - self.last_recognition_time[worker_id] > self.recognition_cooldown: if self.mark_attendance(worker_id, worker_name): logger.info(f"Attendance marked for {worker_name} ({worker_id})") self.last_recognition_time[worker_id] = current_time else: if face_image.size > 0: new_id, new_name = self.register_worker_auto(face_image) if new_id: worker_id = new_id worker_name = new_name color = (255, 165, 0) logger.info(f"New worker registered: {new_name} ({new_id})") if self.mark_attendance(worker_id, worker_name): logger.info(f"Attendance marked for new worker {worker_name} ({worker_id})") cv2.rectangle(frame, (x, y), (x+w, y+h), color, 2) cv2.rectangle(frame, (x, y+h - 35), (x+w, y+h), color, cv2.FILLED) label = f"{worker_name} ({worker_id})" if worker_id else worker_name cv2.putText(frame, label, (x + 6, y+h - 6), cv2.FONT_HERSHEY_DUPLEX, 0.6, (255, 255, 255), 1) except Exception as e: logger.error(f"Error processing face: {e}") continue return frame except Exception as e: logger.error(f"Error processing frame: {e}") return frame def start_video_stream(self, camera_source: int = 0) -> str: try: if self.is_streaming: return "⚠️ Video stream is already running!" self.video_file_path = None self.video_capture = cv2.VideoCapture(camera_source) if not self.video_capture.isOpened(): return "❌ Could not open camera/video source!" self.is_streaming = True def video_loop(): while self.is_streaming: ret, frame = self.video_capture.read() if not ret: break processed_frame = self.process_video_frame(frame) if not self.frame_queue.full(): try: self.frame_queue.put_nowait(processed_frame) except queue.Full: pass time.sleep(0.1) self.recognition_thread = threading.Thread(target=video_loop) self.recognition_thread.daemon = True self.recognition_thread.start() return "✅ Live camera stream started successfully!" except Exception as e: return f"❌ Error starting video stream: {e}" def process_uploaded_video(self, video_path: str) -> str: try: if self.is_streaming: return "⚠️ Please stop current stream before processing a video file!" if not os.path.exists(video_path): return "❌ Video file not found!" self.video_file_path = video_path self.video_processing = True def video_processing_loop(): cap = cv2.VideoCapture(video_path) fps = cap.get(cv2.CAP_PROP_FPS) frame_delay = 1.0 / fps if fps > 0 else 0.03 while self.video_processing and cap.isOpened(): ret, frame = cap.read() if not ret: break processed_frame = self.process_video_frame(frame) if not self.frame_queue.full(): try: self.frame_queue.put_nowait(processed_frame) except queue.Full: pass time.sleep(frame_delay) cap.release() self.video_processing = False self.recognition_thread = threading.Thread(target=video_processing_loop) self.recognition_thread.daemon = True self.recognition_thread.start() return f"✅ Video processing started successfully! ({os.path.basename(video_path)})" except Exception as e: return f"❌ Error processing video: {e}" def stop_video_stream(self) -> str: try: self.is_streaming = False self.video_processing = False if self.video_capture: self.video_capture.release() self.video_capture = None if self.recognition_thread: self.recognition_thread.join(timeout=2) while not self.frame_queue.empty(): try: self.frame_queue.get_nowait() except queue.Empty: break return "✅ Video stream/processing stopped successfully!" except Exception as e: return f"❌ Error stopping video: {e}" def get_current_frame(self) -> Optional[np.ndarray]: try: if not self.frame_queue.empty(): return self.frame_queue.get_nowait() return None except queue.Empty: return None def get_registered_workers_info(self) -> str: if not self.sf: return "❌ Salesforce connection not established." try: workers = self.sf.query_all("SELECT Name, Worker_ID__c, Image_Caption__c, Image_URL__c FROM Worker__c")['records'] if not workers: return "No workers registered yet." info = f"**Registered Workers ({len(workers)}):**\n\n" for i, worker in enumerate(workers, 1): info += f"{i}. **{worker['Name']}** (ID: {worker['Worker_ID__c']}) - Caption: {worker['Image_Caption__c'] or 'N/A'}\n" if worker['Image_URL__c']: info += f" Image: [View]({worker['Image_URL__c']})\n" return info except Exception as e: logger.error(f"Error fetching workers from Salesforce: {e}") return self._get_local_workers_info() def _get_local_workers_info(self) -> str: if not self.known_face_names: return "No workers registered yet." info = f"**Registered Workers ({len(self.known_face_names)}):**\n\n" for i, (worker_id, name) in enumerate(zip(self.known_face_ids, self.known_face_names), 1): info += f"{i}. **{name}** (ID: {worker_id})\n" return info def get_today_attendance(self) -> str: if not self.sf: return "❌ Salesforce connection not established." today = date.today().isoformat() try: records = self.sf.query_all( f"SELECT Name__c, Worker_ID__c, Time__c, Method__c FROM Attendance__c WHERE Date__c = '{today}'" )['records'] if not records: return f"**Today's Attendance ({today}):**\n\nNo attendance marked yet." info = f"**Today's Attendance ({today}):**\n\n" for record in records: method_icon = "🤖" if record['Method__c'] == "Auto" else "👤" info += f"{method_icon} **{record['Name__c']}** (ID: {record['Worker_ID__c']}) - {record['Time__c']}\n" return info except Exception as e: logger.error(f"Error fetching attendance from Salesforce: {e}") return self._get_local_today_attendance() def _get_local_today_attendance(self) -> str: today = date.today().isoformat() today_records = [r for r in self.attendance_records if r["date"] == today] if not today_records: return f"**Today's Attendance ({today}):**\n\nNo attendance marked yet." info = f"**Today's Attendance ({today}):**\n\n" for record in today_records: method_icon = "🤖" if record.get("method") == "Auto" else "👤" info += f"{method_icon} **{record['name']}** (ID: {record['worker_id']}) - {record['time']}\n" return info def get_attendance_report(self, start_date: str, end_date: str) -> str: if not start_date or not end_date: return "Please select both start and end dates." try: datetime.strptime(start_date, '%Y-%m-%d') datetime.strptime(end_date, '%Y-%m-%d') except ValueError: return "Invalid date format. Please use YYYY-MM-DD." if not self.sf: return "❌ Salesforce connection not established." try: records = self.sf.query_all( f"SELECT Worker_ID__c, Name__c, Date__c, Time__c, Method__c FROM Attendance__c " f"WHERE Date__c >= '{start_date}' AND Date__c <= '{end_date}'" )['records'] if not records: return f"No attendance records found between {start_date} and {end_date}." df = pd.DataFrame(records) total_days = (pd.to_datetime(end_date) - pd.to_datetime(start_date)).days + 1 unique_workers = df['Worker_ID__c'].nunique() total_attendances = len(df) auto_registrations = len(df[df['Method__c'] == 'Auto']) report = f"**📊 Attendance Report ({start_date} to {end_date})**\n\n" report += f"**Summary:**\n" report += f"• Total Days: {total_days}\n" report += f"• Unique Workers: {unique_workers}\n" report += f"• Total Attendances: {total_attendances}\n" report += f"• Auto Detections: {auto_registrations}\n\n" if not df.empty: attendance_counts = df.groupby(['Worker_ID__c', 'Name__c']).size().reset_index(name='count') report += f"**👥 Individual Attendance:**\n" for _, row in attendance_counts.iterrows(): percentage = (row['count'] / total_days) * 100 report += f"• **{row['Name__c']}** ({row['Worker_ID__c']}): {row['count']} days ({percentage:.1f}%)\n" return report except Exception as e: logger.error(f"Error generating report from Salesforce: {e}") return self._get_local_attendance_report(start_date, end_date) def _get_local_attendance_report(self, start_date: str, end_date: str) -> str: filtered_records = [ r for r in self.attendance_records if start_date <= r["date"] <= end_date ] if not filtered_records: return f"No attendance records found between {start_date} and {end_date}." df = pd.DataFrame(filtered_records) total_days = (pd.to_datetime(end_date) - pd.to_datetime(start_date)).days + 1 unique_workers = df['worker_id'].nunique() total_attendances = len(df) auto_registrations = len(df[df['method'] == 'Auto']) report = f"**📊 Attendance Report ({start_date} to {end_date})**\n\n" report += f"**Summary:**\n" report += f"• Total Days: {total_days}\n" report += f"• Unique Workers: {unique_workers}\n" report += f"• Total Attendances: {total_attendances}\n" report += f"• Auto Detections: {auto_registrations}\n\n" if not df.empty: attendance_counts = df.groupby(['worker_id', 'name']).size().reset_index(name='count') report += f"**👥 Individual Attendance:**\n" for _, row in attendance_counts.iterrows(): percentage = (row['count'] / total_days) * 100 report += f"• **{row['name']}** ({row['worker_id']}): {row['count']} days ({percentage:.1f}%)\n" return report def export_attendance_csv(self) -> Tuple[Optional[str], str]: try: if not self.attendance_records: return None, "No attendance records to export." df = pd.DataFrame(self.attendance_records) timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') csv_file = f"attendance_report_{timestamp}.csv" df.to_csv(csv_file, index=False) return csv_file, f"✅ Attendance exported to {csv_file}" except Exception as e: return None, f"❌ Error exporting data: {e}" # Initialize system attendance_system = AttendanceSystem() def create_interface(): with gr.Blocks( title="🎯 Advanced Attendance System with Video Recognition", theme=gr.themes.Soft(), css=""" .gradio-container { max-width: 1400px !important; } .tab-nav { font-weight: bold; } .status-box { padding: 10px; border-radius: 5px; margin: 5px 0; } .video-option-tabs { margin-bottom: 15px; } """ ) as demo: gr.Markdown( """ # 🎯 Advanced Attendance System with Face Recognition **Comprehensive facial recognition system with live camera and video file processing, integrated with Hugging Face and Salesforce** ## 🚀 **Key Features:** - **🎥 Live Camera Recognition** - Real-time face detection from camera/CCTV - **📹 Video File Processing** - Process pre-recorded videos for attendance - **🤖 Automatic Worker Registration** - Auto-register unknown faces with unique IDs - **👤 Manual Registration** - Register workers manually with photos and AI-generated captions - **📅 24-Hour Attendance Rule** - One attendance mark per worker per day - **📊 Advanced Analytics** - Detailed reports and data export - **🤗 Hugging Face Integration** - AI-powered image captioning - **☁️ Salesforce Integration** - Store worker and attendance data in Salesforce """ ) with gr.Tabs(): with gr.Tab("🎥 Video Recognition", elem_classes="tab-nav"): gr.Markdown("### Face Recognition from Live Camera or Video File") with gr.Row(): with gr.Column(scale=1): with gr.Tabs(selected="live", elem_classes="video-option-tabs") as video_tabs: with gr.Tab("Live Camera", id="live"): camera_source = gr.Number( label="Camera Source (0 for default camera, or RTSP URL)", value=0, precision=0 ) with gr.Row(): start_stream_btn = gr.Button( "🎥 Start Live Recognition", variant="primary", size="lg" ) with gr.Tab("Upload Video", id="upload"): video_file = gr.Video( label="Upload Video File", sources=["upload"], format="mp4" ) with gr.Row(): process_video_btn = gr.Button( "📹 Process Video File", variant="primary", size="lg" ) stop_stream_btn = gr.Button( "⏹️ Stop Processing", variant="stop", size="lg" ) stream_status = gr.Textbox( label="Processing Status", value="Ready to start...", interactive=False, lines=2 ) gr.Markdown( """ **📋 Instructions:** - **Live Camera:** Select camera source and click "Start Live Recognition" - **Video File:** Upload a video file and click "Process Video File" - Click "Stop Processing" to stop current session **🎨 Color Coding:** - 🟢 **Green:** Known worker (attendance marked) - 🟠 **Orange:** New worker (auto-registered) - 🔴 **Red:** Face detected but processing """ ) with gr.Column(scale=1): video_output = gr.Image( label="Recognition Output", streaming=True, interactive=False ) live_attendance_display = gr.Markdown( value=attendance_system.get_today_attendance(), label="Live Attendance Updates" ) refresh_attendance_btn = gr.Button( "🔄 Refresh Attendance", variant="secondary" ) with gr.Tab("👤 Manual Registration", elem_classes="tab-nav"): gr.Markdown("### Register Workers Manually") with gr.Row(): with gr.Column(scale=1): register_image = gr.Image( label="Upload Worker's Photo", type="pil", height=300 ) register_name = gr.Textbox( label="Worker's Full Name", placeholder="Enter full name...", lines=1 ) register_btn = gr.Button( "👤 Register Worker", variant="primary", size="lg" ) with gr.Column(scale=1): register_output = gr.Textbox( label="Registration Status", lines=3, interactive=False ) registered_workers_info = gr.Markdown( value=attendance_system.get_registered_workers_info(), label="Registered Workers Database" ) with gr.Tab("📊 Reports & Analytics", elem_classes="tab-nav"): gr.Markdown("### Attendance Reports and Data Export") with gr.Row(): with gr.Column(): gr.Markdown("#### 📅 Generate Report") start_date = gr.Textbox( label="Start Date (YYYY-MM-DD)", value=date.today().replace(day=1).strftime('%Y-%m-%d') ) end_date = gr.Textbox( label="End Date (YYYY-MM-DD)", value=date.today().strftime('%Y-%m-%d') ) generate_report_btn = gr.Button( "📊 Generate Report", variant="primary" ) gr.Markdown("#### 💾 Export Data") export_btn = gr.Button( "📥 Export to CSV", variant="secondary" ) export_status = gr.Textbox( label="Export Status", lines=2, interactive=False ) export_file = gr.File( label="Download File", visible=False ) with gr.Column(): report_output = gr.Markdown( value="Select date range and click 'Generate Report' to view attendance analytics.", label="Attendance Report" ) start_stream_btn.click( fn=attendance_system.start_video_stream, inputs=[camera_source], outputs=[stream_status] ) process_video_btn.click( fn=attendance_system.process_uploaded_video, inputs=[video_file], outputs=[stream_status] ) stop_stream_btn.click( fn=attendance_system.stop_video_stream, outputs=[stream_status] ) refresh_attendance_btn.click( fn=attendance_system.get_today_attendance, outputs=[live_attendance_display] ) register_btn.click( fn=attendance_system.register_worker_manual, inputs=[register_image, register_name], outputs=[register_output, registered_workers_info] ) generate_report_btn.click( fn=attendance_system.get_attendance_report, inputs=[start_date, end_date], outputs=[report_output] ) def export_and_show(): file_path, status = attendance_system.export_attendance_csv() if file_path: return status, gr.update(visible=True, value=file_path) else: return status, gr.update(visible=False) export_btn.click( fn=export_and_show, outputs=[export_status, export_file] ) def update_video_frame(): start_time = time.time() while True: current_time = time.time() if current_time - start_time >= 0.03: frame = attendance_system.get_current_frame() if frame is not None: frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) return frame start_time = current_time time.sleep(0.01) video_thread = threading.Thread(target=lambda: demo.queue()(update_video_frame)()) video_thread.daemon = True video_thread.start() return demo if __name__ == "__main__": demo = create_interface() demo.launch( server_name="0.0.0.0", server_port=7860, share=False, show_error=True, debug=True )