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