Upload face_service.py

services/face_service.py

@@ -0,0 +1,100 @@
+# services/face_service.py
+import cv2
+import numpy as np
+import insightface
+from insightface.app import FaceAnalysis
+import logging
+
+from config import Config
+
+logger = logging.getLogger(__name__)
+
+class FaceService:
+    def __init__(self):
+        self.app = None
+        self._initialize()
+    
+    def _initialize(self):
+        """Initialize the face recognition model"""
+        try:
+            self.app = FaceAnalysis(name=Config.MODEL_NAME)
+            # ctx_id: -1 = CPU, 0 = first GPU. Defaults to CPU to match
+            # the plain `onnxruntime` package in requirements.txt.
+            self.app.prepare(ctx_id=Config.CTX_ID, det_size=Config.DETECTION_SIZE)
+            logger.info(
+                f"Face recognition model '{Config.MODEL_NAME}' initialized "
+                f"(ctx_id={Config.CTX_ID}, det_size={Config.DETECTION_SIZE})"
+            )
+        except Exception as e:
+            logger.error(f"Error initializing face recognition: {e}")
+            raise
+    
+    def extract_face(self, image_bytes):
+        """Extract face embedding from image bytes"""
+        try:
+            nparr = np.frombuffer(image_bytes, np.uint8)
+            img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+            
+            if img is None:
+                return None, "Invalid image format"
+            
+            faces = self.app.get(img)
+            
+            if len(faces) == 0:
+                return None, "No face detected"
+            
+            if len(faces) > 1:
+                return None, "Multiple faces detected. Please ensure only one person is in frame."
+            
+            face = faces[0]
+            embedding = face.embedding
+            embedding = embedding / np.linalg.norm(embedding)
+            
+            return {
+                'embedding': embedding.tolist(),
+                'bbox': face.bbox.tolist()
+            }, None
+            
+        except Exception as e:
+            return None, f"Error extracting face: {str(e)}"
+    
+    def compare_faces(self, face_embedding, stored_embeddings):
+        """Compare face embedding against stored embeddings"""
+        best_match = None
+        best_similarity = 0
+        
+        face_embedding = np.array(face_embedding)
+        face_embedding = face_embedding / np.linalg.norm(face_embedding)
+        
+        for item in stored_embeddings:
+            stored_embedding = np.array(item['embedding'])
+            stored_embedding = stored_embedding / np.linalg.norm(stored_embedding)
+            
+            similarity = np.dot(face_embedding, stored_embedding)
+            
+            if similarity > best_similarity:
+                best_similarity = similarity
+                best_match = item
+        
+        return best_match, best_similarity
+    
+    def health_check(self):
+        try:
+            if self.app is None:
+                return False, "Model not initialized"
+            return True, "Healthy"
+        except Exception as e:
+            return False, str(e)
+
+
+# --- Singleton accessor -----------------------------------------------
+# The model is large and slow to load (~seconds, significant RAM/VRAM).
+# Every router used to instantiate its own FaceService(), which loaded
+# the model 4 separate times. Use one shared instance instead.
+_face_service_instance = None
+
+def get_face_service() -> "FaceService":
+    global _face_service_instance
+    if _face_service_instance is None:
+        _face_service_instance = FaceService()
+    return _face_service_instance