Update microexpression_tracker.py

microexpression_tracker.py

@@ -1,74 +1,422 @@
-import numpy as np
-import cv2
-import mediapipe as mp
-
-
-LEFT_EYE = [33, 133]
-RIGHT_EYE = [362, 263]
-NOSE = 1
-
-mp_face_mesh = mp.solutions.face_mesh
-def get_lip_engagement(landmarks):
-    TOP_LIP = 13
-    BOTTOM_LIP = 14
-    LIP_LEFT = 78
-    LIP_RIGHT = 308
-    top_lip = landmarks[TOP_LIP]
-    bottom_lip = landmarks[BOTTOM_LIP]
-    left_corner = landmarks[LIP_LEFT]
-    right_corner = landmarks[LIP_RIGHT]
-    lip_opening = abs(top_lip[1] - bottom_lip[1])
-    lip_width = abs(right_corner[0] - left_corner[0])
-
-    # print(f"[DEBUG] lip_opening: {lip_opening:.3f}, lip_width: {lip_width:.3f}")
-
-    # Example, adjust as per your actual values!
-    # This logic: high opening OR high width = Engaged (smile/mouth open)
-    # very small both = Not Engaged, everything else = Partially Engaged
-    if lip_opening > 0.01 or lip_width > 0.18:
-        return "Engaged"
-    elif lip_opening < 0.002 or lip_width < 0.04:
-        return "Not Engaged"
-    else:
-        return "Partially Engaged"
-
-
-
-
-
-def track_microexpressions(frame, face_mesh, calibration_ref=None):
-    if calibration_ref is None:
-        calibration_ref = {}
-    h, w, _ = frame.shape
-    frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-    results = face_mesh.process(frame_rgb)
-    micro = {
-             "eye_away": False,
-             "head_turn": False,
-             }
-    face_bbox = None
-    multiple_faces = False
-
-    if results.multi_face_landmarks:
-        if len(results.multi_face_landmarks) > 1:
-            multiple_faces = True
-
-        lm = results.multi_face_landmarks[0].landmark
-        xs = [p.x for p in lm]
-        ys = [p.y for p in lm]
-        xmin, xmax = min(xs)*w, max(xs)*w
-        ymin, ymax = min(ys)*h, max(ys)*h
-        face_bbox = [int(xmin), int(ymin), int(xmax), int(ymax)]
-
-        eye_x = (lm[LEFT_EYE[0]].x + lm[RIGHT_EYE[0]].x) / 2
-        nose_x = lm[NOSE].x
-
-        margin = 0.07
-        eye_left_th = calibration_ref.get('eye_left', 0.30)
-        eye_right_th = calibration_ref.get('eye_right', 0.70)
-        if eye_x < (eye_left_th - margin) or eye_x > (eye_right_th + margin):
-            micro["eye_away"] = True
-        if nose_x < (eye_left_th - margin) or nose_x > (eye_right_th + margin):
-            micro["head_turn"] = True
-
-    return micro, face_bbox, multiple_faces
+# import numpy as np
+# import cv2
+# import mediapipe as mp
+
+
+# LEFT_EYE = [33, 133]
+# RIGHT_EYE = [362, 263]
+# NOSE = 1
+
+# mp_face_mesh = mp.solutions.face_mesh
+# def get_lip_engagement(landmarks):
+#     TOP_LIP = 13
+#     BOTTOM_LIP = 14
+#     LIP_LEFT = 78
+#     LIP_RIGHT = 308
+#     top_lip = landmarks[TOP_LIP]
+#     bottom_lip = landmarks[BOTTOM_LIP]
+#     left_corner = landmarks[LIP_LEFT]
+#     right_corner = landmarks[LIP_RIGHT]
+#     lip_opening = abs(top_lip[1] - bottom_lip[1])
+#     lip_width = abs(right_corner[0] - left_corner[0])
+
+#     # print(f"[DEBUG] lip_opening: {lip_opening:.3f}, lip_width: {lip_width:.3f}")
+
+#     # Example, adjust as per your actual values!
+#     # This logic: high opening OR high width = Engaged (smile/mouth open)
+#     # very small both = Not Engaged, everything else = Partially Engaged
+#     if lip_opening > 0.01 or lip_width > 0.18:
+#         return "Engaged"
+#     elif lip_opening < 0.002 or lip_width < 0.04:
+#         return "Not Engaged"
+#     else:
+#         return "Partially Engaged"
+
+
+
+
+
+# def track_microexpressions(frame, face_mesh, calibration_ref=None):
+#     if calibration_ref is None:
+#         calibration_ref = {}
+#     h, w, _ = frame.shape
+#     frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+#     results = face_mesh.process(frame_rgb)
+#     micro = {
+#              "eye_away": False,
+#              "head_turn": False,
+#              }
+#     face_bbox = None
+#     multiple_faces = False
+
+#     if results.multi_face_landmarks:
+#         if len(results.multi_face_landmarks) > 1:
+#             multiple_faces = True
+
+#         lm = results.multi_face_landmarks[0].landmark
+#         xs = [p.x for p in lm]
+#         ys = [p.y for p in lm]
+#         xmin, xmax = min(xs)*w, max(xs)*w
+#         ymin, ymax = min(ys)*h, max(ys)*h
+#         face_bbox = [int(xmin), int(ymin), int(xmax), int(ymax)]
+
+#         eye_x = (lm[LEFT_EYE[0]].x + lm[RIGHT_EYE[0]].x) / 2
+#         nose_x = lm[NOSE].x
+
+#         margin = 0.07
+#         eye_left_th = calibration_ref.get('eye_left', 0.30)
+#         eye_right_th = calibration_ref.get('eye_right', 0.70)
+#         if eye_x < (eye_left_th - margin) or eye_x > (eye_right_th + margin):
+#             micro["eye_away"] = True
+#         if nose_x < (eye_left_th - margin) or nose_x > (eye_right_th + margin):
+#             micro["head_turn"] = True
+
+#     return micro, face_bbox, multiple_faces
+import numpy as np
+import cv2
+import mediapipe as mp
+from typing import Dict, List, Tuple, Optional, NamedTuple
+from dataclasses import dataclass
+from functools import lru_cache
+
+# Pre-computed landmark indices for efficiency
+class LandmarkIndices:
+    """Pre-defined landmark indices for face analysis."""
+    LEFT_EYE = [33, 133]
+    RIGHT_EYE = [362, 263]
+    NOSE = 1
+    TOP_LIP = 13
+    BOTTOM_LIP = 14
+    LIP_LEFT = 78
+    LIP_RIGHT = 308
+
+@dataclass
+class MicroExpressionResult:
+    """Structured result for microexpression analysis."""
+    eye_away: bool
+    head_turn: bool
+    face_bbox: Optional[List[int]]
+    multiple_faces: bool
+    confidence: float = 1.0
+
+class LipEngagementThresholds:
+    """Optimized thresholds for lip engagement detection."""
+    ENGAGED_OPENING = 0.01
+    ENGAGED_WIDTH = 0.18
+    NOT_ENGAGED_OPENING = 0.002
+    NOT_ENGAGED_WIDTH = 0.04
+
+class FaceAnalyzer:
+    """
+    Optimized face analyzer for microexpressions and lip engagement.
+    """
+    
+    def __init__(self, calibration_ref: Optional[Dict] = None):
+        """
+        Initialize the face analyzer.
+        
+        Args:
+            calibration_ref: Optional calibration reference dictionary
+        """
+        self.calibration_ref = calibration_ref or {}
+        self.landmarks = LandmarkIndices()
+        self.lip_thresholds = LipEngagementThresholds()
+        
+        # Cache for commonly used values
+        self._eye_left_th = self.calibration_ref.get('eye_left', 0.30)
+        self._eye_right_th = self.calibration_ref.get('eye_right', 0.70)
+        self._margin = 0.07
+        
+        # Pre-compute boundary values for efficiency
+        self._left_boundary = self._eye_left_th - self._margin
+        self._right_boundary = self._eye_right_th + self._margin
+    
+    @lru_cache(maxsize=32)
+    def _get_engagement_label(self, lip_opening: float, lip_width: float) -> str:
+        """
+        Cached lip engagement classification.
+        
+        Args:
+            lip_opening: Normalized lip opening distance
+            lip_width: Normalized lip width
+            
+        Returns:
+            Engagement label string
+        """
+        if (lip_opening > self.lip_thresholds.ENGAGED_OPENING or 
+            lip_width > self.lip_thresholds.ENGAGED_WIDTH):
+            return "Engaged"
+        elif (lip_opening < self.lip_thresholds.NOT_ENGAGED_OPENING or 
+              lip_width < self.lip_thresholds.NOT_ENGAGED_WIDTH):
+            return "Not Engaged"
+        else:
+            return "Partially Engaged"
+    
+    def get_lip_engagement(self, landmarks: List[Tuple[float, float]]) -> str:
+        """
+        Optimized lip engagement detection.
+        
+        Args:
+            landmarks: List of normalized landmark coordinates
+            
+        Returns:
+            Engagement level string
+        """
+        try:
+            # Direct indexing for better performance
+            top_lip_y = landmarks[self.landmarks.TOP_LIP][1]
+            bottom_lip_y = landmarks[self.landmarks.BOTTOM_LIP][1]
+            left_corner_x = landmarks[self.landmarks.LIP_LEFT][0]
+            right_corner_x = landmarks[self.landmarks.LIP_RIGHT][0]
+            
+            # Calculate distances using abs for efficiency
+            lip_opening = abs(top_lip_y - bottom_lip_y)
+            lip_width = abs(right_corner_x - left_corner_x)
+            
+            # Use cached classification
+            return self._get_engagement_label(lip_opening, lip_width)
+            
+        except (IndexError, TypeError):
+            return "No Face"
+    
+    def _extract_landmarks_vectorized(self, face_landmarks) -> Tuple[np.ndarray, np.ndarray]:
+        """
+        Vectorized landmark extraction for better performance.
+        
+        Args:
+            face_landmarks: MediaPipe face landmarks
+            
+        Returns:
+            Tuple of (x_coords, y_coords) as numpy arrays
+        """
+        # Convert to numpy arrays in one go
+        coords = np.array([(lm.x, lm.y) for lm in face_landmarks.landmark])
+        return coords[:, 0], coords[:, 1]
+    
+    def _calculate_bbox_vectorized(self, x_coords: np.ndarray, y_coords: np.ndarray, 
+                                 frame_width: int, frame_height: int) -> List[int]:
+        """
+        Vectorized bounding box calculation.
+        
+        Args:
+            x_coords: X coordinates array
+            y_coords: Y coordinates array
+            frame_width: Frame width
+            frame_height: Frame height
+            
+        Returns:
+            Bounding box coordinates [xmin, ymin, xmax, ymax]
+        """
+        # Use numpy min/max for vectorized operations
+        xmin = int(np.min(x_coords) * frame_width)
+        xmax = int(np.max(x_coords) * frame_width)
+        ymin = int(np.min(y_coords) * frame_height)
+        ymax = int(np.max(y_coords) * frame_height)
+        
+        return [xmin, ymin, xmax, ymax]
+    
+    def _analyze_eye_movement(self, x_coords: np.ndarray) -> bool:
+        """
+        Optimized eye movement analysis.
+        
+        Args:
+            x_coords: X coordinates array
+            
+        Returns:
+            True if eye is looking away
+        """
+        # Calculate eye center using vectorized operations
+        left_eye_x = x_coords[self.landmarks.LEFT_EYE[0]]
+        right_eye_x = x_coords[self.landmarks.RIGHT_EYE[0]]
+        eye_center_x = (left_eye_x + right_eye_x) * 0.5
+        
+        # Use pre-computed boundaries
+        return eye_center_x < self._left_boundary or eye_center_x > self._right_boundary
+    
+    def _analyze_head_turn(self, x_coords: np.ndarray) -> bool:
+        """
+        Optimized head turn analysis.
+        
+        Args:
+            x_coords: X coordinates array
+            
+        Returns:
+            True if head is turned
+        """
+        nose_x = x_coords[self.landmarks.NOSE]
+        return nose_x < self._left_boundary or nose_x > self._right_boundary
+    
+    def track_microexpressions(self, frame: np.ndarray, face_mesh) -> MicroExpressionResult:
+        """
+        Optimized microexpression tracking.
+        
+        Args:
+            frame: Input video frame
+            face_mesh: MediaPipe face mesh instance
+            
+        Returns:
+            MicroExpressionResult object
+        """
+        h, w = frame.shape[:2]
+        
+        # Convert to RGB once
+        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+        results = face_mesh.process(frame_rgb)
+        
+        # Initialize result with defaults
+        result = MicroExpressionResult(
+            eye_away=False,
+            head_turn=False,
+            face_bbox=None,
+            multiple_faces=False
+        )
+        
+        if not results.multi_face_landmarks:
+            return result
+        
+        # Check for multiple faces
+        if len(results.multi_face_landmarks) > 1:
+            result.multiple_faces = True
+        
+        # Process first face with vectorized operations
+        face_landmarks = results.multi_face_landmarks[0]
+        x_coords, y_coords = self._extract_landmarks_vectorized(face_landmarks)
+        
+        # Calculate bounding box
+        result.face_bbox = self._calculate_bbox_vectorized(x_coords, y_coords, w, h)
+        
+        # Analyze eye movement and head turn
+        result.eye_away = self._analyze_eye_movement(x_coords)
+        result.head_turn = self._analyze_head_turn(x_coords)
+        
+        # Calculate confidence based on face size
+        bbox_area = ((result.face_bbox[2] - result.face_bbox[0]) * 
+                    (result.face_bbox[3] - result.face_bbox[1]))
+        frame_area = w * h
+        result.confidence = min(1.0, bbox_area / (frame_area * 0.1))
+        
+        return result
+
+# Global analyzer instance for backward compatibility
+_global_analyzer = None
+
+def get_analyzer(calibration_ref: Optional[Dict] = None) -> FaceAnalyzer:
+    """
+    Get or create a global analyzer instance.
+    
+    Args:
+        calibration_ref: Optional calibration reference
+        
+    Returns:
+        FaceAnalyzer instance
+    """
+    global _global_analyzer
+    if _global_analyzer is None or calibration_ref is not None:
+        _global_analyzer = FaceAnalyzer(calibration_ref)
+    return _global_analyzer
+
+# Backward compatibility functions
+def get_lip_engagement(landmarks: List[Tuple[float, float]]) -> str:
+    """
+    Backward compatible lip engagement function.
+    
+    Args:
+        landmarks: List of normalized landmark coordinates
+        
+    Returns:
+        Engagement level string
+    """
+    analyzer = get_analyzer()
+    return analyzer.get_lip_engagement(landmarks)
+
+def track_microexpressions(frame: np.ndarray, face_mesh, calibration_ref: Optional[Dict] = None) -> Tuple[Dict, Optional[List[int]], bool]:
+    """
+    Backward compatible microexpression tracking function.
+    
+    Args:
+        frame: Input video frame
+        face_mesh: MediaPipe face mesh instance
+        calibration_ref: Optional calibration reference
+        
+    Returns:
+        Tuple of (micro_dict, face_bbox, multiple_faces)
+    """
+    analyzer = get_analyzer(calibration_ref)
+    result = analyzer.track_microexpressions(frame, face_mesh)
+    
+    # Convert to old format for backward compatibility
+    micro_dict = {
+        "eye_away": result.eye_away,
+        "head_turn": result.head_turn,
+        "confidence": result.confidence
+    }
+    
+    return micro_dict, result.face_bbox, result.multiple_faces
+
+# Performance monitoring utilities
+class PerformanceMonitor:
+    """Simple performance monitoring for optimization."""
+    
+    def __init__(self):
+        self.timings = {}
+        self.call_counts = {}
+    
+    def time_function(self, func_name: str):
+        """Decorator for timing functions."""
+        def decorator(func):
+            def wrapper(*args, **kwargs):
+                import time
+                start = time.time()
+                result = func(*args, **kwargs)
+                end = time.time()
+                
+                if func_name not in self.timings:
+                    self.timings[func_name] = []
+                    self.call_counts[func_name] = 0
+                
+                self.timings[func_name].append(end - start)
+                self.call_counts[func_name] += 1
+                
+                return result
+            return wrapper
+        return decorator
+    
+    def get_stats(self) -> Dict:
+        """Get performance statistics."""
+        stats = {}
+        for func_name, times in self.timings.items():
+            stats[func_name] = {
+                'avg_time': np.mean(times),
+                'total_time': np.sum(times),
+                'call_count': self.call_counts[func_name],
+                'min_time': np.min(times),
+                'max_time': np.max(times)
+            }
+        return stats
+
+# Example usage with performance monitoring
+if __name__ == "__main__":
+    # Initialize performance monitor
+    monitor = PerformanceMonitor()
+    
+    # Example usage
+    mp_face_mesh = mp.solutions.face_mesh
+    face_mesh = mp_face_mesh.FaceMesh(
+        static_image_mode=False,
+        max_num_faces=1,
+        refine_landmarks=True,
+        min_detection_confidence=0.5,
+        min_tracking_confidence=0.5
+    )
+    
+    # Initialize analyzer
+    analyzer = FaceAnalyzer()
+    
+    print("Optimized microexpression module loaded successfully!")
+    print("Key improvements:")
+    print("- Vectorized operations using NumPy")
+    print("- LRU caching for repeated calculations")
+    print("- Structured data types for better memory usage")
+    print("- Pre-computed values for boundary checks")
+    print("- Performance monitoring capabilities")
+    print("- Backward compatibility maintained")