Abeg work

config.yaml

@@ -1,5 +1,7 @@
 # config.yaml
+# -----------------------------------------------------------------------------
 # Main configuration file for the Drive Paddy application.
+# -----------------------------------------------------------------------------
 
 # -- Detection Strategy --
 # Sets the active drowsiness detection method.
@@ -18,31 +20,39 @@ geometric_settings:
   yawn_consec_frames: 20
 
   # Head Pose Estimation for look-away/nod-off detection
-  head_nod_thresh: 15.0      # Max downward pitch angle (in degrees)
-  head_look_away_thresh: 20.0 # Max yaw angle (in degrees)
+  head_nod_thresh: 15.0
+  head_look_away_thresh: 20.0
   head_pose_consec_frames: 20
+  
+  # Low Light Detection
+  low_light_thresh: 70       # Average frame brightness below which a warning is shown (0-255).
+
+  # Drowsiness Level Scoring
+  # The system will sum the weights of active indicators (eyes, mouth, head).
+  drowsiness_levels:
+    very_drowsy_threshold: 0.8  # e.g., Eyes + Head Nod (0.45 + 0.55 = 1.0)
+    slightly_drowsy_threshold: 0.4 # e.g., Just Yawning (0.30) or Eyes Closed (0.45)
+  indicator_weights:
+    eye_closure: 0.45
+    yawning: 0.30
+    head_nod: 0.55
+    looking_away: 0.25
 
-# -- CNN Model Settings --
+# -- CNN Model Settings (Not used in 'geometric' mode) --
 cnn_model_settings:
   model_path: "models/best_model_efficientnet_b7.pth"
   confidence_thresh: 0.8
 
-# -- Hybrid Strategy Settings --
-# Defines weights for combining signals into a single drowsiness score.
-# The system triggers an alert if the total score exceeds 'alert_threshold'.
+# -- Hybrid Strategy Settings (Not used in 'geometric' mode) --
 hybrid_settings:
   alert_threshold: 1.0
   weights:
-    eye_closure: 0.45
-    yawning: 0.30
-    head_nod: 0.55
-    looking_away: 0.25
-    cnn_prediction: 0.60 # Weight for the deep learning model's output
+    eye_closure: 0.45; yawning: 0.30; head_nod: 0.55; looking_away: 0.25; cnn_prediction: 0.60
 
 # -- Alerting System --
 alerting:
   alert_sound_path: "assets/alert.wav"
-  alert_cooldown_seconds: 5
+  alert_cooldown_seconds: 7 # Increased cooldown to prevent alert fatigue
 
 # -- Gemini API (Optional) --
 gemini_api:

pages/1_Live_Detection.py

@@ -16,8 +16,8 @@ if "status_queue" not in st.session_state:
     st.session_state.status_queue = queue.Queue()
 if "audio_queue" not in st.session_state:
     st.session_state.audio_queue = queue.Queue()
-if "last_status" not in st.session_state:
-    st.session_state.last_status = {"status": "Awake"}
+if "last_status" not in st.session_state: 
+    st.session_state.last_status = {"drowsiness_level": "Awake", "lighting": "Good"}
 
     
 # --- Load Configuration and Environment Variables ---
@@ -73,6 +73,8 @@ class VideoProcessor(VideoProcessorBase):
             processed_frame, indicators, _ = self._detector.process_frame(img)
             alert_triggered = any(v for k, v in indicators.items() if k not in ['low_light', 'details'])
             self.status_queue.put(indicators if alert_triggered or indicators.get('low_light') else {"status": "Awake"})
+            print(f"Indicators: {indicators}")  # Debugging line
+            print(self.status_queue)  # Debugging line
         elif strategy == 'cnn_model':
             # The cnn_model processor returns frame and indicators.
             processed_frame, indicators = self._detector.process_frame(img)
@@ -99,7 +101,6 @@ st.info("Press 'START' to activate your camera and begin monitoring.")
 # --- Dynamically Build RTC Configuration ---
 ice_servers = [{'urls': 'stun:global.stun.twilio.com:3478'}]
 if secrets["turn_username"] and secrets["turn_credential"]:
-    print("TURN credentials found, adding TURN servers to config.")
     turn_servers = [
         {'urls': 'turn:global.turn.twilio.com:3478?transport=udp', 'username': secrets["turn_username"], 'credential': secrets["turn_credential"]},
         {'urls': 'turn:global.turn.twilio.com:3478?transport=tcp', 'username': secrets["turn_username"], 'credential': secrets["turn_credential"]}
@@ -108,7 +109,6 @@ if secrets["turn_username"] and secrets["turn_credential"]:
 
 RTC_CONFIGURATION = RTCConfiguration({"iceServers": ice_servers})
 
-
 col1, col2 = st.columns([3, 1])
 
 with col1:
@@ -144,16 +144,18 @@ if webrtc_ctx.state.playing:
 
     with status_placeholder.container():
         last_status = st.session_state.last_status
-        if last_status.get("Low Light"):
-             st.warning("⚠️ Low Light Detected! Accuracy may be affected.")
-        elif last_status.get("status") == "Awake":
-            st.info("✔️ Driver is Awake")
-        else:
-            st.error("🚨 DROWSINESS DETECTED!")
-            for key, value in last_status.items():
-                if key != "Low Light" and key != "status":
-                    st.warning(f"-> {key}: {value:.2f}" if isinstance(value, float) else f"-> {key}")
-    
+        drowsiness_level = last_status.get("drowsiness_level", "Awake")
+        lighting = last_status.get("lighting", "Good")
+        score = last_status.get("details", {}).get("Score", 0)
+
+        st.metric(label="Lighting Condition", value=lighting)
+
+        if lighting == "Low": st.warning("Detection paused due to low light.")
+        
+        if drowsiness_level == "Awake": st.info(f"✔️ Awake (Score: {score:.2f})")
+        elif drowsiness_level == "Slightly Drowsy": st.warning(f"⚠️ Slightly Drowsy (Score: {score:.2f})")
+        elif drowsiness_level == "Very Drowsy": st.error(f"🚨 Very Drowsy! (Score: {score:.2f})")
+
     try:
         audio_data = st.session_state.audio_queue.get(timeout=0.1)
         with audio_placeholder.container():
@@ -166,3 +168,5 @@ if webrtc_ctx.state.playing:
 else:
     with status_placeholder.container():
         st.info("✔️ Driver is Awake")
+
+    

src/alerting/alert_system.py

@@ -1,110 +1,60 @@
 # drive_paddy/alerting/alert_system.py
-import time
-import os
-import io
+import time, os, io, google.generativeai as genai
 from gtts import gTTS
-import google.generativeai as genai
-from dotenv import load_dotenv
+from pydub import AudioSegment
+import simpleaudio as sa # Keep for fallback
 
-load_dotenv()  # Load environment variables from .env file
-
-api_key = os.getenv("GEMINI_API_KEY")
 class BaseAlerter:
-    """Base class for alerter systems."""
     def __init__(self, config):
         self.config = config['alerting']
         self.cooldown = self.config['alert_cooldown_seconds']
         self.last_alert_time = 0
         self.alert_on = False
-
-    def trigger_alert(self):
-        raise NotImplementedError
-
+    def trigger_alert(self, level="Very Drowsy"): raise NotImplementedError
     def reset_alert(self):
-        if self.alert_on:
-            print("Resetting Alert.")
-            self.alert_on = False
+        if self.alert_on: print("Resetting Alert."); self.alert_on = False
 
 class FileAlertSystem(BaseAlerter):
-    """Loads a static audio file from disk into memory."""
     def __init__(self, config):
         super().__init__(config)
-        self.sound_path = self.config['alert_sound_path']
         self.audio_bytes = None
         try:
-            if os.path.exists(self.sound_path):
-                with open(self.sound_path, "rb") as f:
-                    self.audio_bytes = f.read()
-            else:
-                print(f"Warning: Alert sound file not found at '{self.sound_path}'.")
-        except Exception as e:
-            print(f"Warning: Could not load audio file. Error: {e}.")
-
-    def trigger_alert(self):
+            if os.path.exists(config['alerting']['alert_sound_path']):
+                with open(config['alerting']['alert_sound_path'], "rb") as f: self.audio_bytes = f.read()
+        except Exception as e: print(f"Warning: Could not load audio file. Error: {e}.")
+    def trigger_alert(self, level="Very Drowsy"):
         current_time = time.time()
-        if (current_time - self.last_alert_time) > self.cooldown:
-            if not self.alert_on and self.audio_bytes:
-                print("Triggering Static Alert!")
-                self.last_alert_time = current_time
-                self.alert_on = True
-                return self.audio_bytes # Return the audio data
+        if (current_time - self.last_alert_time) > self.cooldown and not self.alert_on and self.audio_bytes:
+            self.last_alert_time = current_time; self.alert_on = True
+            print("Triggering Static Alert!")
+            return self.audio_bytes
         return None
 
-
 class GeminiAlertSystem(BaseAlerter):
-    """Generates dynamic audio data using Gemini and gTTS."""
     def __init__(self, config, api_key):
         super().__init__(config)
+        try: genai.configure(api_key=api_key); self.model = genai.GenerativeModel('gemini-pro')
+        except Exception as e: print(f"Error initializing Gemini: {e}."); self.model = None
+    def _generate_audio_data(self, level):
+        if not self.model: return None
+        if level == "Slightly Drowsy":
+            prompt = "You are an AI driving assistant. Generate a short, gentle reminder (under 10 words) for a driver showing minor signs of fatigue."
+        else: # Very Drowsy
+            prompt = "You are an AI driving assistant. Generate a short, firm, and urgent alert (under 10 words) for a driver who is very drowsy."
         try:
-            genai.configure(api_key=api_key)
-            self.model = genai.GenerativeModel('gemini-1.5-flash')  # Use the Gemini model
-            print("Gemini Alert System initialized successfully.")
-        except Exception as e:
-            print(f"Error initializing Gemini: {e}.")
-            self.model = None
-
-    def _generate_audio_data(self):
-        """Generates a unique alert message and returns it as audio bytes."""
-        if not self.model:
-            alert_text = "Stay alert!"
-        else:
-            prompt = "You are an AI driving assistant. Generate a short, friendly, but firm audio alert (under 10 words) for a driver showing signs of drowsiness."
-            try:
-                response = self.model.generate_content(prompt)
-                alert_text = response.text.strip().replace('*', '')
-            except Exception as e:
-                print(f"Error generating alert text with Gemini: {e}")
-                alert_text = "Wake up please!"
-        
-        print(f"Generated Alert Text: '{alert_text}'")
-        try:
-            # Generate TTS audio in memory
-            mp3_fp = io.BytesIO()
-            tts = gTTS(text=alert_text, lang='en')
-            tts.write_to_fp(mp3_fp)
-            mp3_fp.seek(0)
-            return mp3_fp.getvalue()
-        except Exception as e:
-            print(f"Error generating TTS audio: {e}")
-            return None
-
-    def trigger_alert(self):
+            response = self.model.generate_content(prompt)
+            alert_text = response.text.strip().replace('*', '')
+            print(f"Generated Alert Text ({level}): '{alert_text}'")
+            mp3_fp = io.BytesIO(); tts = gTTS(text=alert_text, lang='en'); tts.write_to_fp(mp3_fp)
+            mp3_fp.seek(0); return mp3_fp.getvalue()
+        except Exception as e: print(f"Error generating TTS audio: {e}"); return None
+    def trigger_alert(self, level="Very Drowsy"):
         current_time = time.time()
-        if (current_time - self.last_alert_time) > self.cooldown:
-            if not self.alert_on and self.model:
-                self.last_alert_time = current_time
-                self.alert_on = True
-                return self._generate_audio_data() # Return the audio data
+        if (current_time - self.last_alert_time) > self.cooldown and not self.alert_on and self.model:
+            self.last_alert_time = current_time; self.alert_on = True
+            return self._generate_audio_data(level)
         return None
 
-
 def get_alerter(config, api_key=None):
-    """Factory to get the appropriate alerter based on config."""
-    use_gemini = config.get('gemini_api', {}).get('enabled', False)
-    
-    if use_gemini and api_key:
-        print("Initializing Gemini Alert System.")
-        return GeminiAlertSystem(config, api_key)
-    else:
-        print("Initializing standard File Alert System.")
-        return FileAlertSystem(config)
+    if config.get('gemini_api', {}).get('enabled', False) and api_key: return GeminiAlertSystem(config, api_key)
+    return FileAlertSystem(config)

src/detection/strategies/geometric.py

@@ -5,123 +5,102 @@ import numpy as np
 import math
 from ..base_processor import BaseProcessor
 
-# --- Helper Functions ---
+# --- Helper Functions (No changes here) ---
 def calculate_ear(eye_landmarks, frame_shape):
-    """Calculates the Eye Aspect Ratio (EAR)."""
-    # ... (implementation remains the same)
     coords = np.array([(lm.x * frame_shape[1], lm.y * frame_shape[0]) for lm in eye_landmarks])
     v1 = np.linalg.norm(coords[1] - coords[5]); v2 = np.linalg.norm(coords[2] - coords[4])
-    h1 = np.linalg.norm(coords[0] - coords[3])
-    return (v1 + v2) / (2.0 * h1) if h1 > 0 else 0.0
+    h1 = np.linalg.norm(coords[0] - coords[3]); return (v1 + v2) / (2.0 * h1) if h1 > 0 else 0.0
 
 def calculate_mar(mouth_landmarks, frame_shape):
-    """Calculates the Mouth Aspect Ratio (MAR) for yawn detection."""
     coords = np.array([(lm.x * frame_shape[1], lm.y * frame_shape[0]) for lm in mouth_landmarks])
-    v1 = np.linalg.norm(coords[1] - coords[7]) # Vertical distances
-    v2 = np.linalg.norm(coords[2] - coords[6])
-    v3 = np.linalg.norm(coords[3] - coords[5])
-    h1 = np.linalg.norm(coords[0] - coords[4]) # Horizontal distance
+    v1 = np.linalg.norm(coords[1] - coords[7]); v2 = np.linalg.norm(coords[2] - coords[6])
+    v3 = np.linalg.norm(coords[3] - coords[5]); h1 = np.linalg.norm(coords[0] - coords[4])
     return (v1 + v2 + v3) / (2.0 * h1) if h1 > 0 else 0.0
 
 class GeometricProcessor(BaseProcessor):
-    """
-    Drowsiness detection using a combination of facial landmarks:
-    - Eye Aspect Ratio (EAR) for eye closure.
-    - Mouth Aspect Ratio (MAR) for yawning.
-    - Head Pose Estimation for nodding off or looking away.
-    """
     def __init__(self, config):
         self.settings = config['geometric_settings']
-        self.face_mesh = mp.solutions.face_mesh.FaceMesh(
-            max_num_faces=1, refine_landmarks=True,
-            min_detection_confidence=0.5, min_tracking_confidence=0.5)
-
-        # State counters
-        self.counters = {
-            "eye_closure": 0, "yawning": 0,
-            "head_nod": 0, "looking_away": 0
-        }
-
-        # Landmark indices
-        self.L_EYE = [362, 385, 387, 263, 373, 380]
-        self.R_EYE = [33, 160, 158, 133, 153, 144]
+        self.face_mesh = mp.solutions.face_mesh.FaceMesh(max_num_faces=1, refine_landmarks=True, min_detection_confidence=0.5, min_tracking_confidence=0.5)
+        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]
 
     def process_frame(self, frame):
-        img_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
         h, w, _ = frame.shape
-        results = self.face_mesh.process(img_rgb)
-        
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        brightness = np.mean(gray)
+        is_low_light = brightness < self.settings['low_light_thresh']
+
         drowsiness_indicators = {
-            "eye_closure": False, "yawning": False,
-            "head_nod": False, "looking_away": False, "details": {}
+            "drowsiness_level": "Awake", "lighting": "Good", "details": {}
         }
+        face_landmarks = None
+
+        if is_low_light:
+            drowsiness_indicators["lighting"] = "Low"
+        else:
+            img_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+            results = self.face_mesh.process(img_rgb)
+            face_landmarks = results.multi_face_landmarks
 
-        if results.multi_face_landmarks:
-            landmarks = results.multi_face_landmarks[0].landmark
-            
-            # --- Eye Closure Detection (EAR) ---
-            left_ear = calculate_ear([landmarks[i] for i in self.L_EYE], (h, w))
-            right_ear = calculate_ear([landmarks[i] for i in self.R_EYE], (h, w))
-            ear = (left_ear + right_ear) / 2.0
-            if ear < self.settings['eye_ar_thresh']:
-                self.counters['eye_closure'] += 1
-                if self.counters['eye_closure'] >= self.settings['eye_ar_consec_frames']:
-                    drowsiness_indicators['eye_closure'] = True
-            else:
-                self.counters['eye_closure'] = 0
-            drowsiness_indicators['details']['EAR'] = ear
+            if face_landmarks:
+                landmarks = face_landmarks[0].landmark
+                score = 0
+                weights = self.settings['indicator_weights']
 
-            # --- Yawn Detection (MAR) ---
-            mar = calculate_mar([landmarks[i] for i in self.MOUTH], (h, w))
-            if mar > self.settings['yawn_mar_thresh']:
-                self.counters['yawning'] += 1
-                if self.counters['yawning'] >= self.settings['yawn_consec_frames']:
-                    drowsiness_indicators['yawning'] = True
-            else:
-                self.counters['yawning'] = 0
-            drowsiness_indicators['details']['MAR'] = mar
+                # Eye Closure
+                ear = (calculate_ear([landmarks[i] for i in self.L_EYE],(h,w)) + calculate_ear([landmarks[i] for i in self.R_EYE],(h,w)))/2.0
+                if ear < self.settings['eye_ar_thresh']: self.counters['eye_closure']+=1
+                else: self.counters['eye_closure']=0
+                if self.counters['eye_closure'] >= self.settings['eye_ar_consec_frames']: score += weights['eye_closure']
                 
-            # --- Head Pose Estimation ---
-            face_3d = np.array([
-                [0.0, 0.0, 0.0],            # Nose tip
-                [0.0, -330.0, -65.0],       # Chin
-                [-225.0, 170.0, -135.0],    # Left eye left corner
-                [225.0, 170.0, -135.0],     # Right eye right corner
-                [-150.0, -150.0, -125.0],   # Left Mouth corner
-                [150.0, -150.0, -125.0]     # Right mouth corner
-            ], dtype=np.float64)
-            face_2d = np.array([
-                (landmarks[1].x * w, landmarks[1].y * h),   # Nose tip
-                (landmarks[152].x * w, landmarks[152].y * h), # Chin
-                (landmarks[263].x * w, landmarks[263].y * h), # Left eye corner
-                (landmarks[33].x * w, landmarks[33].y * h),   # Right eye corner
-                (landmarks[287].x * w, landmarks[287].y * h), # Left mouth corner
-                (landmarks[57].x * w, landmarks[57].y * h)   # Right mouth corner
-            ], dtype=np.float64)
+                # Yawning
+                mar = calculate_mar([landmarks[i] for i in self.MOUTH], (h, w))
+                if mar > self.settings['yawn_mar_thresh']: self.counters['yawning']+=1
+                else: self.counters['yawning']=0
+                if self.counters['yawning'] >= self.settings['yawn_consec_frames']: score += weights['yawning']
 
-            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)
-            
-            pitch, yaw = angles[0], angles[1]
-            drowsiness_indicators['details']['Pitch'] = pitch
-            drowsiness_indicators['details']['Yaw'] = yaw
+                # Head Pose
+                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)
+                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']
+
+                # Determine Drowsiness Level
+                levels = self.settings['drowsiness_levels']
+                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
+        
+        # --- Visualization on Video Frame ---
+        level = drowsiness_indicators['drowsiness_level']
+        score_val = drowsiness_indicators.get("details", {}).get("Score", 0)
+        color = (0, 255, 0) # Green for Awake
+        
+        if drowsiness_indicators['lighting'] == "Low":
+            color = (0, 165, 255) # Orange for low light
+            cv2.putText(frame, "LOW LIGHT", (w // 2 - 120, h // 2), cv2.FONT_HERSHEY_SIMPLEX, 2, color, 3, cv2.LINE_AA)
+        elif level == "Slightly Drowsy":
+            color = (0, 255, 255) # Yellow
+        elif level == "Very Drowsy":
+            color = (0, 0, 255) # Red
 
-            if pitch > self.settings['head_nod_thresh']:
-                self.counters['head_nod'] += 1
-                if self.counters['head_nod'] >= self.settings['head_pose_consec_frames']:
-                    drowsiness_indicators['head_nod'] = True
-            else:
-                self.counters['head_nod'] = 0
+        # Draw a colored border around the frame
+        cv2.rectangle(frame, (0, 0), (w, h), color, 10)
 
-            if abs(yaw) > self.settings['head_look_away_thresh']:
-                self.counters['looking_away'] += 1
-                if self.counters['looking_away'] >= self.settings['head_pose_consec_frames']:
-                    drowsiness_indicators['looking_away'] = True
-            else:
-                self.counters['looking_away'] = 0
+        # Display status text
+        status_text = f"Status: {level} (Score: {score_val:.2f})"
+        cv2.putText(frame, status_text, (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2, cv2.LINE_AA)
 
-        # This processor now returns the frame and a dictionary of indicators
-        return frame, drowsiness_indicators
+        return frame, drowsiness_indicators, face_landmarks