Update src/detection/strategies/geometric.py

src/detection/strategies/geometric.py

@@ -3,7 +3,7 @@ import cv2
 import mediapipe as mp
 import numpy as np
 import math
-from ..base_processor import BaseProcessor
+from src.detection.base_processor import BaseProcessor
 
 # --- Helper Functions (Unchanged) ---
 def calculate_ear(eye_landmarks, frame_shape):
@@ -22,16 +22,21 @@ def calculate_mar(mouth_landmarks, frame_shape):
     return (v1 + v2 + v3) / (2.0 * h1) if h1 > 0 else 0.0
 
 class GeometricProcessor(BaseProcessor):
+    L_EYE = [362, 385, 387, 263, 373, 380]
+    R_EYE = [33, 160, 158, 133, 153, 144]
+    MOUTH = [61, 291, 39, 181, 0, 17, 84, 178]
+
     def __init__(self, config):
         self.settings = config['geometric_settings']
         self.face_mesh = mp.solutions.face_mesh.FaceMesh(
             max_num_faces=1,
-            refine_landmarks=True,
+            refine_landmarks=False,
             min_detection_confidence=0.5,
             min_tracking_confidence=0.5)
 
-        self.downscale_factor = self.settings.get('downscale_factor', 0.5) 
-        self.skip_frames = self.settings.get('skip_frames', 2)
+        self.downscale_factor = self.settings.get('downscale_factor', 0.35) 
+        self.default_skip = max(1, self.settings.get("skip_frames", 2))
+
         
         # State variables for frame skipping
         self.frame_counter = 0
@@ -40,27 +45,30 @@ class GeometricProcessor(BaseProcessor):
 
         # Landmark and counter definitions
         self.counters = { "eye_closure": 0, "yawning": 0, "head_nod": 0, "looking_away": 0 }
-        self.L_EYE = [362, 385, 387, 263, 373, 380]
-        self.R_EYE = [33, 160, 158, 133, 153, 144]
-        self.MOUTH = [61, 291, 39, 181, 0, 17, 84, 178]
 
+        # Pre-allocated buffers
+        self.zeros_4x1 = np.zeros((4, 1), np.float32)
+        self.landmark_ids = self.L_EYE + self.R_EYE + self.MOUTH
+
+        
     def process_frame(self, frame):
         self.frame_counter += 1
         
-        # --- FRAME SKIPPING LOGIC ---
-        # If we are skipping this frame, simply draw the last known status on the current frame and return.
-        if self.frame_counter % self.skip_frames != 0:
-            # We use the original, full-sized 'frame' for drawing to keep the video feed smooth.
-            frame_with_last_results = self.draw_visuals(frame, self.last_indicators, self.last_landmarks)
-            return frame_with_last_results, self.last_indicators, self.last_landmarks
-            
+        # adaptive skipping: fewer analyses when driver is awake
+        last_level = self.last_indicators.get("drowsiness_level", "Awake")
+        skip_n = 1 if last_level != "Awake" else self.default_skip
+
+        if self.frame_ctr % skip_n != 0:
+            drawn = self.draw_visuals(frame.copy(), self.last_indicators, self.last_landmarks)
+            return drawn, self.last_indicators, self.last_landmarks
+
         # --- FRAME PROCESSING ---
         original_frame = frame.copy()
         h_orig, w_orig, _ = original_frame.shape
 
         # --- OPTIMIZATION 1: DOWNSCALING ---
         # Create a small frame for fast processing
-        small_frame = cv2.resize(original_frame, (0, 0), fx=self.downscale_factor, fy=self.downscale_factor)
+        small_frame = cv2.resize(original_frame, (0, 0), fx=self.downscale_factor, fy=self.downscale_factor, interpolation=cv2.INTER_AREA)
         h, w, _ = small_frame.shape
         
         # --- FIX 1: PROCESS THE SMALL FRAME, NOT THE ORIGINAL ---
@@ -95,26 +103,35 @@ class GeometricProcessor(BaseProcessor):
                 if self.counters['yawning'] >= self.settings['yawn_consec_frames']: score += weights['yawning']
 
                 # --- Head Pose Estimation (also uses small frame dimensions 'h' and 'w') ---
-                face_3d = np.array([[0.0,0.0,0.0],[0.0,-330.0,-65.0],[-225.0,170.0,-135.0],[225.0,170.0,-135.0],[-150.0,-150.0,-125.0],[150.0,-150.0,-125.0]],dtype=np.float64)
-                face_2d = np.array([(landmarks[1].x*w,landmarks[1].y*h),(landmarks[152].x*w,landmarks[152].y*h),(landmarks[263].x*w,landmarks[263].y*h),(landmarks[33].x*w,landmarks[33].y*h),(landmarks[287].x*w,landmarks[287].y*h),(landmarks[57].x*w,landmarks[57].y*h)],dtype=np.float64)
-                cam_matrix = np.array([[w,0,w/2],[0,w,h/2],[0,0,1]],dtype=np.float64)
-                _, rot_vec, _ = cv2.solvePnP(face_3d, face_2d, cam_matrix, np.zeros((4,1),dtype=np.float64))
-                rmat, _ = cv2.Rodrigues(rot_vec); angles, _, _, _, _, _ = cv2.RQDecomp3x3(rmat)
+                face_3d = np.array([[0.0,0.0,0.0],[0.0,-330.0,-65.0],[-225.0,170.0,-135.0],[225.0,170.0,-135.0],[-150.0,-150.0,-125.0],[150.0,-150.0,-125.0]],dtype=np.float32)
+                face_2d = np.array([(landmarks[1].x*w,landmarks[1].y*h),(landmarks[152].x*w,landmarks[152].y*h),(landmarks[263].x*w,landmarks[263].y*h),(landmarks[33].x*w,landmarks[33].y*h),(landmarks[287].x*w,landmarks[287].y*h),(landmarks[57].x*w,landmarks[57].y*h)],dtype=np.float32)
+                cam_matrix = np.array([[w,0,w/2],[0,w,h/2],[0,0,1]],dtype=np.float32)
+                 _, rvec, _ = cv2.solvePnP(face_3d, face_2d, cam_mtx, self.zeros_4x1, flags=cv2.SOLVEPNP_EPNP)
+                rmat, _ = cv2.Rodrigues(rvec)
+                angles, _, _, _, _, _ = cv2.RQDecomp3x3(rmat)
                 pitch, yaw = angles[0], angles[1]
 
-                if pitch > self.settings['head_nod_thresh']: self.counters['head_nod']+=1
-                else: self.counters['head_nod']=0
-                if self.counters['head_nod'] >= self.settings['head_pose_consec_frames']: score += weights['head_nod']
-
-                if abs(yaw) > self.settings['head_look_away_thresh']: self.counters['looking_away']+=1
-                else: self.counters['looking_away']=0
-                if self.counters['looking_away'] >= self.settings['head_pose_consec_frames']: score += weights['looking_away']
+                if pitch > self.settings['head_nod_thresh']: 
+                    self.counters['head_nod']+=1
+                else: 
+                    self.counters['head_nod']=0
+                if self.counters['head_nod'] >= self.settings['head_pose_consec_frames']: 
+                    score += weights['head_nod']
+
+                if abs(yaw) > self.settings['head_look_away_thresh']: 
+                    self.counters['looking_away']+=1
+                else: 
+                    self.counters['looking_away']=0
+                if self.counters['looking_away'] >= self.settings['head_pose_consec_frames']: 
+                    score += weights['looking_away']
                 
                 levels = self.settings['drowsiness_levels']
-                if score >= levels['very_drowsy_threshold']: drowsiness_indicators['drowsiness_level'] = "Very Drowsy"
+                if score >= levels['very_drowsy_threshold']: 
+                    drowsiness_indicators['drowsiness_level'] = "Very Drowsy"
                 elif score >= levels['slightly_drowsy_threshold']: drowsiness_indicators['drowsiness_level'] = "Slightly Drowsy"
                 
                 drowsiness_indicators['details']['Score'] = score
+
         else:
             drowsiness_indicators["lighting"] = "Low"