microexpression_tracker.py

# 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")