Update utils.py

utils.py

@@ -1,161 +1,168 @@
-"""
-Utility functions for the Attendance Analyzer application
-"""
-
 import cv2
 import numpy as np
-from datetime import datetime, date
-import os
+import base64
+import requests
 import logging
+from datetime import datetime
+from deepface import DeepFace
+from mtcnn import MTCNN
+import os
+from dotenv import load_dotenv
 
-# Setup logging
-logging.basicConfig(level=logging.INFO)
+# Configure logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 logger = logging.getLogger(__name__)
 
-def setup_logging():
-    """Setup logging configuration"""
-    log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-    logging.basicConfig(
-        level=logging.INFO,
-        format=log_format,
-        handlers=[
-            logging.FileHandler('attendance_analyzer.log'),
-            logging.StreamHandler()
-        ]
-    )
+# Load environment variables
+load_dotenv()
+SALESFORCE_TOKEN = os.getenv("SALESFORCE_TOKEN", "your_salesforce_oauth_token")
+SALESFORCE_BASE_URL = "https://construction-site.secure.force.com/WorkerRecognition/services/apexrest"
+SALESFORCE_ENDPOINT = f"{SALESFORCE_BASE_URL}/ProcessImage"
+
+class WorkerRecognitionSystem:
+    def __init__(self):
+        """Initialize the face recognition system."""
+        self.frame_rate = 10  # Process every 10th frame
+        self.confidence_threshold = 0.90  # Fine-tuned for high accuracy
+        self.image_size = (224, 224)
+
+        # Initialize MTCNN for face detection
+        self.detector = MTCNN()
+        logger.info("Initialized MTCNN face detector")
+
+        # Initialize worker database
+        self.worker_db = {}  # {worker_id: [embeddings]}
+        self.processed_workers = set()  # Track processed workers
 
-def validate_image(image):
-    """Validate uploaded image"""
-    if image is None:
-        return False, "No image provided"
-    
-    try:
-        if isinstance(image, np.ndarray):
-            if len(image.shape) != 3:
-                return False, "Invalid image format"
-            return True, "Valid image"
-        else:
-            return True, "Valid image"
-    except Exception as e:
-        return False, f"Image validation error: {str(e)}"
+    def load_worker_database(self):
+        """Load worker facial embeddings from Salesforce (simulated)."""
+        try:
+            response = requests.get(
+                f"{SALESFORCE_BASE_URL}/WorkerData",
+                headers={"Authorization": f"Bearer {SALESFORCE_TOKEN}"}
+            )
+            if response.status_code == 200:
+                workers = response.json()
+                for worker in workers:
+                    embeddings = worker.get("Facial_Features__c", [])
+                    if len(embeddings) >= 3:  # Require at least 3 embeddings
+                        self.worker_db[worker["Worker_ID__c"]] = [np.array(emb) for emb in embeddings]
+                logger.info(f"Loaded {len(self.worker_db)} workers from Salesforce")
+            else:
+                logger.error(f"Failed to load worker database: {response.status_code}")
+        except Exception as e:
+            logger.error(f"Error loading worker database: {str(e)}")
 
-def resize_image(image, max_size=(800, 600)):
-    """Resize image while maintaining aspect ratio"""
-    try:
-        height, width = image.shape[:2]
-        max_width, max_height = max_size
-        
-        if width <= max_width and height <= max_height:
-            return image
-        
-        # Calculate scaling factor
-        scale = min(max_width / width, max_height / height)
-        new_width = int(width * scale)
-        new_height = int(height * scale)
-        
-        resized = cv2.resize(image, (new_width, new_height), interpolation=cv2.INTER_AREA)
-        return resized
-    except Exception as e:
-        logger.error(f"Error resizing image: {e}")
-        return image
+    def preprocess_frame(self, frame):
+        """Preprocess frame to improve quality."""
+        frame = cv2.convertScaleAbs(frame, alpha=1.2, beta=10)  # Enhance contrast
+        frame = cv2.GaussianBlur(frame, (5, 5), 0)  # Reduce noise
+        frame = cv2.resize(frame, (1280, 720))  # Ensure sufficient resolution
+        return frame
 
-def format_datetime(dt=None):
-    """Format datetime for database storage"""
-    if dt is None:
-        dt = datetime.now()
-    return dt.strftime("%Y-%m-%d %H:%M:%S")
+    def detect_faces(self, frame):
+        """Detect faces using MTCNN."""
+        rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+        faces = self.detector.detect_faces(rgb_frame)
+        return [(face['box'][0], face['box'][1], face['box'][2], face['box'][3]) 
+               for face in faces if face['confidence'] > 0.9]
 
-def format_date(d=None):
-    """Format date for database storage"""
-    if d is None:
-        d = date.today()
-    return d.strftime("%Y-%m-%d")
+    def extract_features(self, face_image):
+        """Extract facial features using ensemble of Facenet512 and ArcFace."""
+        try:
+            face_image = cv2.resize(face_image, self.image_size)
+            face_image = cv2.cvtColor(face_image, cv2.COLOR_BGR2RGB)
+            embedding_facenet = DeepFace.represent(
+                img_path=face_image,
+                model_name="Facenet512",
+                enforce_detection=False
+            )
+            embedding_arcface = DeepFace.represent(
+                img_path=face_image,
+                model_name="ArcFace",
+                enforce_detection=False
+            )
+            embedding = (np.array(embedding_facenet) + np.array(embedding_arcface)) / 2
+            return embedding
+        except Exception as e:
+            logger.error(f"Error extracting features: {str(e)}")
+            return None
 
-def create_directory_if_not_exists(directory_path):
-    """Create directory if it doesn't exist"""
-    try:
-        if not os.path.exists(directory_path):
-            os.makedirs(directory_path)
-            logger.info(f"Created directory: {directory_path}")
-        return True
-    except Exception as e:
-        logger.error(f"Error creating directory {directory_path}: {e}")
-        return False
+    def recognize_worker(self, embedding):
+        """Compare embedding with worker database."""
+        max_confidence = 0
+        worker_id = None
+        for wid, stored_embeddings in self.worker_db.items():
+            confidences = [
+                np.dot(embedding, stored_emb) / (np.linalg.norm(embedding) * np.linalg.norm(stored_emb))
+                for stored_emb in stored_embeddings
+            ]
+            confidence = np.mean([(sim + 1) / 2 for sim in confidences])
+            if confidence > max_confidence:
+                max_confidence = confidence
+                worker_id = wid
+        return worker_id, max_confidence
 
-def cleanup_old_files(directory, days_old=30):
-    """Clean up old files from specified directory"""
-    try:
-        current_time = datetime.now()
-        for filename in os.listdir(directory):
-            file_path = os.path.join(directory, filename)
-            if os.path.isfile(file_path):
-                file_time = datetime.fromtimestamp(os.path.getctime(file_path))
-                if (current_time - file_time).days > days_old:
-                    os.remove(file_path)
-                    logger.info(f"Removed old file: {filename}")
-    except Exception as e:
-        logger.error(f"Error cleaning up old files: {e}")
+    def send_to_salesforce(self, worker_id, confidence, image_base64, entry_time):
+        """Send recognition result to Salesforce."""
+        payload = {
+            "workerId": worker_id,
+            "confidence": confidence,
+            "image": image_base64,
+            "entryTime": entry_time.isoformat(),
+            "needsReview": confidence < self.confidence_threshold
+        }
+        headers = {
+            "Authorization": f"Bearer {SALESFORCE_TOKEN}",
+            "Content-Type": "application/json"
+        }
+        try:
+            response = requests.post(SALESFORCE_ENDPOINT, json=payload, headers=headers)
+            if response.status_code == 200:
+                logger.info(f"Successfully sent data to Salesforce for worker {worker_id or 'unknown'}")
+                return response.json()
+            else:
+                logger.error(f"Failed to send to Salesforce: {response.status_code} - {response.text}")
+        except Exception as e:
+            logger.error(f"Error sending to Salesforce: {str(e)}")
+        return None
 
-def validate_name(name):
-    """Validate person name"""
-    if not name or not name.strip():
-        return False, "Name cannot be empty"
-    
-    if len(name.strip()) < 2:
-        return False, "Name must be at least 2 characters long"
-    
-    if len(name.strip()) > 50:
-        return False, "Name cannot exceed 50 characters"
-    
-    # Check for valid characters (letters, spaces, hyphens, apostrophes)
-    import re
-    if not re.match(r"^[a-zA-Z\s\-'\.]+$", name.strip()):
-        return False, "Name contains invalid characters"
-    
-    return True, "Valid name"
+    def process_frame(self, frame):
+        """Process a single frame for recognition."""
+        faces = self.detect_faces(frame)
+        results = []
+        for (x, y, w, h) in faces:
+            face_image = frame[y:y+h, x:x+w]
+            embedding = self.extract_features(face_image)
+            if embedding is None:
+                continue
 
-def validate_email(email):
-    """Validate email address"""
-    if not email:
-        return True, "Email is optional"
-    
-    import re
-    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
-    if re.match(email_pattern, email):
-        return True, "Valid email"
-    else:
-        return False, "Invalid email format"
+            worker_id, confidence = self.recognize_worker(embedding)
+            entry_time = datetime.now()
+            worker_key = f"{worker_id}_{entry_time.date().isoformat()}"
+            if worker_key in self.processed_workers:
+                continue
 
-def get_system_info():
-    """Get system information for debugging"""
-    import platform
-    import psutil
-    
-    info = {
-        "platform": platform.platform(),
-        "python_version": platform.python_version(),
-        "cpu_count": psutil.cpu_count(),
-        "memory_total": f"{psutil.virtual_memory().total / (1024**3):.2f} GB",
-        "memory_available": f"{psutil.virtual_memory().available / (1024**3):.2f} GB",
-    }
-    return info
+            _, buffer = cv2.imencode('.jpg', face_image)
+            image_base64 = base64.b64encode(buffer).decode('utf-8')
 
-def benchmark_face_detection(image, iterations=5):
-    """Benchmark face detection performance"""
-    import time
-    import face_recognition
-    
-    times = []
-    for _ in range(iterations):
-        start_time = time.time()
-        face_locations = face_recognition.face_locations(image)
-        end_time = time.time()
-        times.append(end_time - start_time)
-    
-    avg_time = sum(times) / len(times)
-    return {
-        "average_time": avg_time,
-        "min_time": min(times),
-        "max_time": max(times),
-        "iterations": iterations
-    }
+            if confidence >= self.confidence_threshold and worker_id:
+                logger.info(f"Recognized worker {worker_id} with confidence {confidence:.2f}")
+                self.processed_workers.add(worker_key)
+                self.send_to_salesforce(worker_id, confidence, image_base64, entry_time)
+                results.append({
+                    "worker_id": worker_id,
+                    "confidence": confidence,
+                    "entry_time": entry_time,
+                    "image_base64": image_base64
+                })
+            else:
+                logger.info(f"Unrecognized face, confidence {confidence:.2f}, flagging for verification")
+                self.send_to_salesforce(None, confidence, image_base64, entry_time)
+                results.append({
+                    "worker_id": "Unknown",
+                    "confidence": confidence,
+                    "entry_time": entry_time,
+                    "image_base64": image_base64
+                })
+        return results