Try something different

.env.example

@@ -1,2 +0,0 @@
-GEMINI_API_KEY=your_gemini_api_key_here
-HUGGINGFACE_API_KEY=your_huggingface_api_key_here

app.py

@@ -1,140 +1,457 @@
-# app_webrtc.py
-import gradio as gr
-import numpy as np
-import os
-import yaml
-from dotenv import load_dotenv
-import io
-from scipy.io.wavfile import read as read_wav
-from pydub import AudioSegment
 import cv2
+import numpy as np
+import dlib
+import gradio as gr
+import threading
 import time
-from gradio_webrtc import WebRTC
-
-# Correctly import from the drive_paddy package structure
-from src.detection.factory import get_detector
-from src.alerting.alert_system import get_alerter
-
-# --- Load Configuration and Environment Variables ---
-load_dotenv()
-config_path = 'config.yaml'
-with open(config_path, 'r') as f:
-    config = yaml.safe_load(f)
-secrets = {
-    "gemini_api_key": os.getenv("GEMINI_API_KEY"),
-}
-
-# --- Initialize Backend Components ---
-detector = get_detector(config)
-alerter = get_alerter(config, secrets["gemini_api_key"])
-geo_settings = config.get('geometric_settings', {})
-drowsiness_levels = geo_settings.get('drowsiness_levels', {})
-SLIGHTLY_DROWSY_DEFAULT = drowsiness_levels.get('slightly_drowsy_threshold', 0.3)
-VERY_DROWSY_DEFAULT = drowsiness_levels.get('very_drowsy_threshold', 0.8)
-
-# --- Global state for audio (simpler than queues for this component) ---
-# We use a global variable to hold the audio data, which the UI will poll.
-# This is a common pattern in simple Gradio streaming apps.
-latest_audio_alert = None
+import queue
+import pygame
+import io
+import google.generativeai as genai
+from scipy.spatial import distance as dist
+from collections import deque
+import tempfile
+import os
+from datetime import datetime
 
-# --- Main Processing Function ---
-def process_stream(frame: np.ndarray, sensitivity_threshold: float) -> np.ndarray:
-    """
-    This is the core function. It takes a frame and returns the processed frame.
-    All logic, including status drawing and alert triggering, happens here.
-    """
-    global latest_audio_alert
+class DrowsinessDetector:
+    def __init__(self, gemini_api_key=None):
+        # Initialize face detector and landmark predictor
+        self.detector = dlib.get_frontal_face_detector()
+        self.predictor = dlib.shape_predictor('shape_predictor_68_face_landmarks.dat')
+        
+        # EAR (Eye Aspect Ratio) parameters
+        self.EAR_THRESHOLD = 0.25
+        self.EAR_CONSECUTIVE_FRAMES = 20
+        self.ear_counter = 0
+        self.ear_history = deque(maxlen=30)
+        
+        # Yawn detection parameters
+        self.YAWN_THRESHOLD = 0.6
+        self.YAWN_CONSECUTIVE_FRAMES = 15
+        self.yawn_counter = 0
+        self.yawn_history = deque(maxlen=30)
+        
+        # Head pose estimation parameters
+        self.NOD_THRESHOLD = 15
+        self.nod_counter = 0
+        self.head_pose_history = deque(maxlen=30)
+        
+        # Alert system
+        self.drowsy_alert = False
+        self.last_alert_time = 0
+        self.alert_cooldown = 10  # seconds
+        
+        # Initialize pygame for audio
+        pygame.mixer.init()
+        
+        # Initialize Gemini AI
+        if gemini_api_key:
+            genai.configure(api_key=gemini_api_key)
+            self.model = genai.GenerativeModel('gemini-pro')
+        else:
+            self.model = None
+        
+        # Eye and mouth landmark indices
+        self.LEFT_EYE = list(range(36, 42))
+        self.RIGHT_EYE = list(range(42, 48))
+        self.MOUTH = list(range(48, 68))
+        
+        # 3D model points for head pose estimation
+        self.model_points = 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
+        ])
+        
+        self.status_log = deque(maxlen=100)
+        
+    def calculate_ear(self, eye_landmarks):
+        """Calculate Eye Aspect Ratio"""
+        # Vertical eye landmarks
+        A = dist.euclidean(eye_landmarks[1], eye_landmarks[5])
+        B = dist.euclidean(eye_landmarks[2], eye_landmarks[4])
+        # Horizontal eye landmark
+        C = dist.euclidean(eye_landmarks[0], eye_landmarks[3])
+        # EAR calculation
+        ear = (A + B) / (2.0 * C)
+        return ear
     
-    if frame is None:
-        return np.zeros((480, 640, 3), dtype=np.uint8)
-
-    # Process the frame using our existing detector.
-    # The detector already draws landmarks and status overlays.
-    processed_frame, indicators, _ = detector.process_frame(frame)
-    drowsiness_level = indicators.get("drowsiness_level", "Awake")
-
-    # Handle audio alerts
-    if drowsiness_level != "Awake":
-        audio_data = alerter.trigger_alert(level=drowsiness_level)
-        if audio_data:
-            # Convert audio for Gradio and store it in the global variable
-            try:
-                byte_io = io.BytesIO(audio_data)
-                audio = AudioSegment.from_mp3(byte_io)
-                wav_byte_io = io.BytesIO()
-                audio.export(wav_byte_io, format="wav")
-                wav_byte_io.seek(0)
-                sample_rate, data = read_wav(wav_byte_io)
-                latest_audio_alert = (sample_rate, data)
-            except Exception as e:
-                print(f"Audio processing error: {e}")
-                latest_audio_alert = None
-    else:
-        alerter.reset_alert()
-
-    return processed_frame
-
-# --- Function to check for and return audio alerts ---
-def get_audio_update():
-    """
-    This function is polled by the UI to check for new audio alerts.
-    """
-    global latest_audio_alert
-    if latest_audio_alert:
-        audio_to_play = latest_audio_alert
-        latest_audio_alert = None  # Clear the alert after sending it
-        return audio_to_play
-    return None
-
-# --- Gradio UI Definition ---
-with gr.Blocks(theme=gr.themes.Default(primary_hue="blue", secondary_hue="blue")) as app:
-    gr.HTML(
-        """
-        <div align="center">
-            <img src="https://em-content.zobj.net/source/samsung/380/automobile_1f697.png" alt="Car Emoji" width="100"/>
-            <h1>Drive Paddyn</h1>
-        </div>
-        """
-    )
+    def calculate_mar(self, mouth_landmarks):
+        """Calculate Mouth Aspect Ratio for yawn detection"""
+        # Vertical mouth landmarks
+        A = dist.euclidean(mouth_landmarks[2], mouth_landmarks[10])  # 50, 58
+        B = dist.euclidean(mouth_landmarks[4], mouth_landmarks[8])   # 52, 56
+        # Horizontal mouth landmark
+        C = dist.euclidean(mouth_landmarks[0], mouth_landmarks[6])   # 48, 54
+        # MAR calculation
+        mar = (A + B) / (2.0 * C)
+        return mar
+    
+    def get_head_pose(self, landmarks, img_size):
+        """Estimate head pose using facial landmarks"""
+        image_points = np.array([
+            (landmarks[30][0], landmarks[30][1]),     # Nose tip
+            (landmarks[8][0], landmarks[8][1]),       # Chin
+            (landmarks[36][0], landmarks[36][1]),     # Left eye left corner
+            (landmarks[45][0], landmarks[45][1]),     # Right eye right corner
+            (landmarks[48][0], landmarks[48][1]),     # Left Mouth corner
+            (landmarks[54][0], landmarks[54][1])      # Right mouth corner
+        ], dtype="double")
+        
+        # Camera internals
+        focal_length = img_size[1]
+        center = (img_size[1]/2, img_size[0]/2)
+        camera_matrix = np.array([
+            [focal_length, 0, center[0]],
+            [0, focal_length, center[1]],
+            [0, 0, 1]], dtype="double")
+        
+        dist_coeffs = np.zeros((4,1)) # Assuming no lens distortion
+        
+        # Solve PnP
+        (success, rotation_vector, translation_vector) = cv2.solvePnP(
+            self.model_points, image_points, camera_matrix, dist_coeffs, 
+            flags=cv2.SOLVEPNP_ITERATIVE)
+        
+        # Convert rotation vector to rotation matrix
+        (rotation_matrix, jacobian) = cv2.Rodrigues(rotation_vector)
+        
+        # Calculate Euler angles
+        sy = np.sqrt(rotation_matrix[0,0] * rotation_matrix[0,0] +  rotation_matrix[1,0] * rotation_matrix[1,0])
+        singular = sy < 1e-6
+        if not singular:
+            x = np.arctan2(rotation_matrix[2,1], rotation_matrix[2,2])
+            y = np.arctan2(-rotation_matrix[2,0], sy)
+            z = np.arctan2(rotation_matrix[1,0], rotation_matrix[0,0])
+        else:
+            x = np.arctan2(-rotation_matrix[1,2], rotation_matrix[1,1])
+            y = np.arctan2(-rotation_matrix[2,0], sy)
+            z = 0
+        
+        # Convert to degrees
+        angles = np.array([x, y, z]) * 180.0 / np.pi
+        return angles
+    
+    def generate_voice_alert(self, alert_type, severity="medium"):
+        """Generate voice alert using Gemini AI"""
+        if not self.model:
+            return self.play_default_alert()
+        
+        try:
+            prompts = {
+                "drowsy": f"Generate a brief, urgent but caring voice alert (max 15 words) to wake up a drowsy driver. Severity: {severity}. Make it sound natural and concerned.",
+                "yawn": f"Generate a brief, gentle voice alert (max 12 words) for a driver who is yawning frequently. Severity: {severity}. Sound caring but alert.",
+                "nod": f"Generate a brief, firm voice alert (max 12 words) for a driver whose head is nodding. Severity: {severity}. Sound urgent but supportive."
+            }
+            
+            response = self.model.generate_content(prompts.get(alert_type, prompts["drowsy"]))
+            alert_text = response.text.strip().replace('"', '').replace("'", "")
+            
+            # Use text-to-speech (you would need to install pyttsx3 or use cloud TTS)
+            # For this example, we'll use a placeholder
+            self.log_status(f"🔊 ALERT: {alert_text}")
+            return alert_text
+            
+        except Exception as e:
+            self.log_status(f"Error generating alert: {str(e)}")
+            return self.play_default_alert()
+    
+    def play_default_alert(self):
+        """Play default beep alert"""
+        try:
+            # Generate a simple beep sound
+            duration = 0.5  # seconds
+            freq = 800  # Hz
+            sample_rate = 22050
+            frames = int(duration * sample_rate)
+            arr = np.zeros(frames)
+            
+            for i in range(frames):
+                arr[i] = np.sin(2 * np.pi * freq * i / sample_rate)
+            
+            arr = (arr * 32767).astype(np.int16)
+            sound = pygame.sndarray.make_sound(arr)
+            sound.play()
+            
+            alert_text = "⚠️ WAKE UP! Please stay alert while driving!"
+            self.log_status(f"🔊 {alert_text}")
+            return alert_text
+            
+        except Exception as e:
+            self.log_status(f"Error playing alert: {str(e)}")
+            return "Alert system activated"
+    
+    def log_status(self, message):
+        """Log status messages with timestamp"""
+        timestamp = datetime.now().strftime("%H:%M:%S")
+        self.status_log.append(f"[{timestamp}] {message}")
+    
+    def detect_drowsiness(self, frame):
+        """Main drowsiness detection function"""
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        faces = self.detector(gray)
+        
+        status_text = []
+        alert_message = ""
+        
+        if len(faces) == 0:
+            status_text.append("👤 No face detected")
+            self.log_status("No face detected in frame")
+            return frame, status_text, alert_message
+        
+        for face in faces:
+            landmarks = self.predictor(gray, face)
+            landmarks = np.array([[p.x, p.y] for p in landmarks.parts()])
+            
+            # Draw face rectangle
+            x, y, w, h = face.left(), face.top(), face.width(), face.height()
+            cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0), 2)
+            
+            # Eye Aspect Ratio calculation
+            left_eye = landmarks[self.LEFT_EYE]
+            right_eye = landmarks[self.RIGHT_EYE]
+            
+            left_ear = self.calculate_ear(left_eye)
+            right_ear = self.calculate_ear(right_eye)
+            ear = (left_ear + right_ear) / 2.0
+            self.ear_history.append(ear)
+            
+            # Draw eye contours
+            cv2.drawContours(frame, [cv2.convexHull(left_eye)], -1, (0, 255, 0), 1)
+            cv2.drawContours(frame, [cv2.convexHull(right_eye)], -1, (0, 255, 0), 1)
+            
+            # Yawn detection
+            mouth = landmarks[self.MOUTH]
+            mar = self.calculate_mar(mouth)
+            self.yawn_history.append(mar)
+            
+            # Draw mouth contour
+            cv2.drawContours(frame, [cv2.convexHull(mouth)], -1, (0, 255, 255), 1)
+            
+            # Head pose estimation
+            head_angles = self.get_head_pose(landmarks, frame.shape)
+            self.head_pose_history.append(head_angles[0])  # Pitch angle
+            
+            # Drowsiness detection logic
+            drowsy_indicators = []
+            
+            # Check EAR
+            if ear < self.EAR_THRESHOLD:
+                self.ear_counter += 1
+                if self.ear_counter >= self.EAR_CONSECUTIVE_FRAMES:
+                    drowsy_indicators.append("EYES_CLOSED")
+                    status_text.append(f"👁️ Eyes closed! EAR: {ear:.3f}")
+            else:
+                self.ear_counter = 0
+                status_text.append(f"👁️ Eyes open - EAR: {ear:.3f}")
+            
+            # Check for yawning
+            if mar > self.YAWN_THRESHOLD:
+                self.yawn_counter += 1
+                if self.yawn_counter >= self.YAWN_CONSECUTIVE_FRAMES:
+                    drowsy_indicators.append("YAWNING")
+                    status_text.append(f"🥱 Yawning detected! MAR: {mar:.3f}")
+            else:
+                self.yawn_counter = 0
+                status_text.append(f"👄 Normal mouth - MAR: {mar:.3f}")
+            
+            # Check head nodding
+            if abs(head_angles[0]) > self.NOD_THRESHOLD:
+                self.nod_counter += 1
+                if self.nod_counter >= 10:
+                    drowsy_indicators.append("HEAD_NOD")
+                    status_text.append(f"📉 Head nodding! Angle: {head_angles[0]:.1f}°")
+            else:
+                self.nod_counter = 0
+                status_text.append(f"📏 Head pose - Pitch: {head_angles[0]:.1f}°")
+            
+            # Generate alerts
+            current_time = time.time()
+            if drowsy_indicators and (current_time - self.last_alert_time) > self.alert_cooldown:
+                self.drowsy_alert = True
+                self.last_alert_time = current_time
+                
+                # Determine alert type and severity
+                if "EYES_CLOSED" in drowsy_indicators:
+                    severity = "high" if len(drowsy_indicators) > 1 else "medium"
+                    alert_message = self.generate_voice_alert("drowsy", severity)
+                elif "YAWNING" in drowsy_indicators:
+                    alert_message = self.generate_voice_alert("yawn", "medium")
+                elif "HEAD_NOD" in drowsy_indicators:
+                    alert_message = self.generate_voice_alert("nod", "medium")
+                
+                # Visual alert on frame
+                cv2.putText(frame, "⚠️ DROWSINESS ALERT! ⚠️", (50, 50), 
+                           cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 0, 255), 3)
+            
+            # Draw landmark points
+            for (x, y) in landmarks:
+                cv2.circle(frame, (x, y), 1, (255, 255, 255), -1)
+            
+            # Add metrics overlay
+            cv2.putText(frame, f"EAR: {ear:.3f}", (10, frame.shape[0] - 80), 
+                       cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
+            cv2.putText(frame, f"MAR: {mar:.3f}", (10, frame.shape[0] - 60), 
+                       cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
+            cv2.putText(frame, f"Head: {head_angles[0]:.1f}°", (10, frame.shape[0] - 40), 
+                       cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
+            
+        return frame, status_text, alert_message
 
-    with gr.Row():
-        # The WebRTC component now directly shows the processed output
-        webrtc_output = WebRTC(
-            label="Live Detection Feed",
-            video_source="webcam",
+def create_gradio_interface():
+    """Create Gradio interface for the drowsiness detection system"""
+    
+    # Global detector instance
+    detector = None
+    
+    def initialize_system(gemini_key):
+        """Initialize the drowsiness detection system"""
+        global detector
+        try:
+            detector = DrowsinessDetector(gemini_key if gemini_key.strip() else None)
+            return "✅ System initialized successfully!", "System ready for detection."
+        except Exception as e:
+            return f"❌ Error initializing system: {str(e)}", "System initialization failed."
+    
+    def process_video_frame(frame, gemini_key):
+        """Process a single video frame"""
+        global detector
+        
+        if detector is None:
+            detector = DrowsinessDetector(gemini_key if gemini_key.strip() else None)
+        
+        try:
+            processed_frame, status_list, alert_msg = detector.detect_drowsiness(frame)
+            
+            # Format status text
+            status_text = "\n".join(status_list) if status_list else "Processing..."
+            
+            # Get recent logs
+            log_text = "\n".join(list(detector.status_log)[-10:]) if detector.status_log else "No logs yet."
+            
+            return processed_frame, status_text, alert_msg, log_text
+            
+        except Exception as e:
+            error_msg = f"Error processing frame: {str(e)}"
+            return frame, error_msg, "", error_msg
+    
+    # Create the Gradio interface
+    with gr.Blocks(title="Driver Drowsiness Detection System", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("""
+        # 🚗 Real-time Driver Drowsiness Detection System
+        
+        This system uses computer vision and AI to detect driver drowsiness through:
+        - **Eye Aspect Ratio (EAR)** - Detects closed/droopy eyes
+        - **Mouth Aspect Ratio (MAR)** - Detects yawning
+        - **Head Pose Estimation** - Detects head nodding
+        - **AI Voice Alerts** - Uses Gemini AI for personalized wake-up messages
+        
+        ### 📋 Setup Instructions:
+        1. Download `shape_predictor_68_face_landmarks.dat` from [dlib models](http://dlib.net/files/shape_predictor_68_face_landmarks.dat.bz2)
+        2. Place it in the same directory as this script
+        3. (Optional) Enter your Gemini API key for AI-powered voice alerts
+        4. Start your webcam and begin monitoring!
+        """)
+        
+        with gr.Row():
+            with gr.Column(scale=2):
+                # Gemini API key input
+                gemini_key_input = os.getenv('GEMINI_API_KEY')
+                
+                
+                # Initialize button
+                init_btn = gr.Button("🚀 Initialize System", variant="primary")
+                init_status = gr.Textbox(label="Initialization Status", interactive=False)
+                
+            with gr.Column(scale=1):
+                # System info
+                gr.Markdown("""
+                ### 📊 Detection Thresholds:
+                - **EAR Threshold**: 0.25
+                - **Yawn Threshold**: 0.6  
+                - **Head Nod**: 15° deviation
+                - **Alert Cooldown**: 10 seconds
+                """)
+        
+        with gr.Row():
+            with gr.Column(scale=2):
+                # Video input/output
+                video_input = gr.Video(
+                    sources=["webcam"], 
+                    label="📹 Camera Feed",
+                    streaming=True
+                )
+                
+            with gr.Column(scale=1):
+                # Status displays
+                current_status = gr.Textbox(
+                    label="📈 Current Status", 
+                    lines=6,
+                    interactive=False
+                )
+                
+                alert_display = gr.Textbox(
+                    label="🔊 Latest Alert", 
+                    interactive=False,
+                    placeholder="No alerts yet..."
+                )
+                
+                system_logs = gr.Textbox(
+                    label="📝 System Logs", 
+                    lines=8,
+                    interactive=False,
+                    placeholder="System logs will appear here..."
+                )
+        
+        # Event handlers
+        init_btn.click(
+            fn=initialize_system,
+            inputs=[gemini_key_input],
+            outputs=[init_status, alert_display]
         )
-
-    with gr.Row():
-        sensitivity_slider = gr.Slider(
-            minimum=0.1, 
-            maximum=1.0, 
-            value=SLIGHTLY_DROWSY_DEFAULT, 
-            step=0.05, 
-            label="Alert Sensitivity Threshold",
-            info="Lower value = more sensitive to drowsiness signs."
+        
+        video_input.stream(
+            fn=process_video_frame,
+            inputs=[video_input, gemini_key_input],
+            outputs=[video_input, current_status, alert_display, system_logs],
+            stream_every=0.1,  # Process every 100ms
+            show_progress=False
         )
-
-    # Hidden audio component for playing alerts
-    audio_player = gr.Audio(autoplay=True, visible=False)
-
-    # Connect the WebRTC stream to the processing function
-    webrtc_output.stream(
-        fn=process_stream,
-        inputs=[webrtc_output, sensitivity_slider],
-        outputs=[webrtc_output],
-        # The 'every' parameter is not needed for this component; it streams as fast as possible.
-    )
+        
+        # Instructions
+        gr.Markdown("""
+        ### 🔧 Troubleshooting:
+        - **No face detected**: Ensure good lighting and face is visible to camera
+        - **Poor detection**: Adjust camera angle and distance (arm's length recommended)
+        - **No alerts**: Check if Gemini API key is valid (optional feature)
+        - **High CPU usage**: Reduce video resolution or increase stream interval
+        
+        ### ⚠️ Safety Notice:
+        This system is for demonstration purposes. Always prioritize real-world driving safety measures.
+        Pull over safely if you feel drowsy while driving.
+        """)
     
-    # Use a separate loop to poll for audio updates.
-    # This is more stable than returning multiple values in a high-frequency stream.
-    app.load(
-        fn=get_audio_update,
-        inputs=None,
-        outputs=[audio_player],
-        every=1 # Check for a new audio alert every 1 second
-    )
+    return demo
 
-
-# --- Launch the App ---
+# Main execution
 if __name__ == "__main__":
-    app.launch(debug=True)
+    # Check for required files
+    if not os.path.exists('shape_predictor_68_face_landmarks.dat'):
+        print("❌ Missing required file: shape_predictor_68_face_landmarks.dat")
+        print("📥 Please download from: http://dlib.net/files/shape_predictor_68_face_landmarks.dat.bz2")
+        print("📁 Extract and place in the same directory as this script")
+    else:
+        print("✅ All required files found!")
+        
+        # Create and launch the interface
+        demo = create_gradio_interface()
+        demo.launch(
+            share=True,
+            server_name="0.0.0.0",
+            server_port=7860,
+            show_error=True
+        )

config.yaml

@@ -1,64 +0,0 @@
-# config.yaml
-# -----------------------------------------------------------------------------
-# Main configuration file for the Drive Paddy application.
-# -----------------------------------------------------------------------------
-
-# -- Detection Strategy --
-# Sets the active drowsiness detection method.
-# Options: "geometric", "cnn_model", "hybrid"
-detection_strategy: "geometric"
-
-# -- Geometric Strategy Settings --
-# Parameters for the facial landmark-based detection methods.
-geometric_settings:
-  # Eye Aspect Ratio (EAR) for blink/closure detection
-  eye_ar_thresh: 0.23
-  eye_ar_consec_frames: 15
-
-  # Mouth Aspect Ratio (MAR) for yawn detection
-  yawn_mar_thresh: 0.70
-  yawn_consec_frames: 20
-
-  # Head Pose Estimation for look-away/nod-off detection
-  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.2
-    yawning: 0.20
-    head_nod: 0.45
-    looking_away: 0.15
-
-# -- 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 (Not used in 'geometric' mode) --
-hybrid_settings:
-  alert_threshold: 1.0
-  # CORRECTED: Each weight is now on its own line.
-  weights:
-    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: 7 # Increased cooldown to prevent alert fatigue
-
-# -- Gemini API (Optional) --
-gemini_api:
-  enabled: true

download_model.py

@@ -1,42 +0,0 @@
-# download_model.py
-import os
-from huggingface_hub import hf_hub_download
-
-# --- Configuration ---
-# Details from your Hugging Face repository screenshot.
-REPO_ID = "Testys/drowsiness-detection-model"
-FILENAME = "best_model_efficientnet_b7.pth"
-LOCAL_DIR = "models"
-
-def download_model():
-    """
-    Downloads the specified model file from Hugging Face Hub
-    and saves it to the local models/ directory.
-    """
-    print(f"Downloading model '{FILENAME}' from repository '{REPO_ID}'...")
-
-    # Ensure the local directory exists.
-    if not os.path.exists(LOCAL_DIR):
-        os.makedirs(LOCAL_DIR)
-        print(f"Created directory: {LOCAL_DIR}")
-
-    try:
-        # Download the file.
-        # local_dir_use_symlinks=False ensures the file is copied to your directory
-        # instead of just pointing to the cache.
-        model_path = hf_hub_download(
-            repo_id=REPO_ID,
-            filename=FILENAME,
-            local_dir=LOCAL_DIR,
-            local_dir_use_symlinks=False,
-            # token=True # Use token for private repos, can be omitted for public ones
-        )
-        print(f"\nModel downloaded successfully!")
-        print(f"Saved to: {model_path}")
-
-    except Exception as e:
-        print(f"\nAn error occurred during download: {e}")
-        print("Please check the repository ID, filename, and your network connection.")
-
-if __name__ == "__main__":
-    download_model()

main.py

@@ -1,80 +0,0 @@
-# drive_paddy/main.py
-import streamlit as st
-import yaml
-import os
-from dotenv import load_dotenv
-import queue
-
-# --- Main Application UI ---
-st.set_page_config(
-    page_title="Drive Paddy | Home",
-    page_icon="🚗",
-    layout="wide"
-)
-
-# Load config to display current settings on the home page
-@st.cache_resource
-def load_app_config():
-    load_dotenv()
-    gemini_api_key = os.getenv("GEMINI_API_KEY")
-    with open('config.yaml', 'r') as f:
-        config = yaml.safe_load(f)
-    return config, gemini_api_key
-
-config, gemini_api_key = load_app_config()
-
-# --- Initialize Session State ---
-# This ensures they are set when the app first loads.
-if "play_audio" not in st.session_state:
-    st.session_state.play_audio = None
-if "active_alerts" not in st.session_state:
-    st.session_state.active_alerts = {"status": "Awake"}
-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"}
-
-
-
-# --- Page Content ---
-st.title("🚗 Welcome to Drive Paddy!")
-st.subheader("Your AI-Powered Drowsiness Detection Assistant")
-
-st.markdown("""
-Drive Paddy is a real-time system designed to enhance driver safety by detecting signs of drowsiness. 
-It uses your computer's webcam to analyze facial features and head movements, providing timely alerts 
-to help prevent fatigue-related accidents.
-""")
-
-st.info("Navigate to the **Live Detection** page from the sidebar on the left to start the system.")
-
-st.markdown("---")
-
-col1, col2 = st.columns(2)
-
-with col1:
-    st.header("How It Works")
-    st.markdown("""
-    The system employs a sophisticated hybrid strategy to monitor for signs of fatigue:
-    - **👀 Eye Closure Detection**: Measures Eye Aspect Ratio (EAR) to detect prolonged blinks or closed eyes.
-    - **🥱 Yawn Detection**: Measures Mouth Aspect Ratio (MAR) to identify yawns.
-    - **😴 Head Pose Analysis**: Tracks head pitch and yaw to detect nodding off or looking away from the road.
-    - **🧠 CNN Model Inference**: A deep learning model provides an additional layer of analysis.
-    
-    These signals are combined into a single drowsiness score to trigger alerts accurately.
-    """)
-
-with col2:
-    st.header("Current Configuration")
-    alert_method = "Gemini API" if config.get('gemini_api', {}).get('enabled') and gemini_api_key else "Static Audio File"
-    st.markdown(f"""
-    - **Detection Strategy**: `{config['detection_strategy']}`
-    - **Alert Method**: `{alert_method}`
-    """)
-    st.warning("Ensure good lighting and that your face is clearly visible for best results.")
-
-st.markdown("---")
-st.markdown("Created with ❤️ using Streamlit, OpenCV, and MediaPipe.")
-

models/best_model_efficientnet_b7.pth

@@ -1,3 +0,0 @@
-version https://git-lfs.github.com/spec/v1
-oid sha256:b13c1e5e4f1a03e0e559ad8f7988c14b63d2b028c55f380814f241dd788a99df
-size 256870774

pages/1_Live_Detection.py

@@ -1,164 +0,0 @@
-# drive_paddy/pages/1_Live_Detection.py
-import streamlit as st
-from streamlit_webrtc import webrtc_streamer, RTCConfiguration, VideoProcessorBase
-import yaml
-import av
-import os
-from dotenv import load_dotenv
-import base64
-import queue
-import time
-from typing import List, Dict, Union
-
-# Correctly import from the drive_paddy package structure
-from src.detection.factory import get_detector
-from src.alerting.alert_system import get_alerter
-
-# --- Initialize Session State at the TOP of the script ---
-# This is the single source of truth for our queues and must run on every page load.
-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 = {"drowsiness_level": "Awake", "lighting": "Good"}
-
-
-# --- Load Configuration and Environment Variables ---
-@st.cache_resource
-def load_app_config():
-    """Loads config from yaml and .env files."""
-    load_dotenv()
-    # Navigate up to the root to find the config file
-    config_path = "./config.yaml"
-    with open(config_path, 'r') as f:
-        config = yaml.safe_load(f)
-    # Load secrets from environment
-    secrets = {
-        "gemini_api_key": os.getenv("GEMINI_API_KEY"),
-        "turn_username": os.getenv("TURN_USERNAME"),
-        "turn_credential": os.getenv("TURN_CREDENTIAL")
-    }
-    return config, secrets
-
-config, secrets = load_app_config()
-
-# --- Client-Side Audio Playback Function ---
-def autoplay_audio(audio_bytes: bytes):
-    """Injects HTML to autoplay audio in the user's browser."""
-    b64 = base64.b64encode(audio_bytes).decode()
-    md = f"""
-        <audio controls autoplay="true" style="display:none;">
-        <source src="data:audio/mp3;base64,{b64}" type="audio/mp3">
-        </audio>
-        """
-    st.markdown(md, unsafe_allow_html=True)
-
-# --- WebRTC Video Processor ---
-class VideoProcessor(VideoProcessorBase):
-    # The __init__ method now accepts the queues as arguments
-    def __init__(self):
-        # It uses the queues passed in from session_state, not new ones.
-        self.status_queue = queue.Queue
-        self.audio_queue = queue.Queue
-        self._detector = get_detector(config)
-        self._alerter = get_alerter(config, secrets["gemini_api_key"])
-
-    def recv(self, frame: av.VideoFrame) -> av.VideoFrame:
-        img = frame.to_ndarray(format="bgr24")
-        
-        strategy = config.get('detection_strategy')
-        
-        # The return signature of process_frame varies by strategy.
-        processed_frame, indicators, _ = self._detector.process_frame(img)
-        drowsiness_level = indicators.get("drowsiness_level", "Awake")
-        
-        # This now correctly puts data into the shared session_state queue.
-        self.status_queue.put(indicators)
-
-        if drowsiness_level != "Awake":
-            audio_data = self._alerter.trigger_alert(level=drowsiness_level)
-            if audio_data:
-                # This now correctly puts audio data into the shared queue.
-                self.audio_queue.put(audio_data)
-        else:
-            self._alerter.reset_alert()
-            
-        return av.VideoFrame.from_ndarray(processed_frame, format="bgr24")
-
-# --- Page UI ---
-st.title("📹 Live Drowsiness Detection")
-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"]:
-    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"]}
-    ]
-    ice_servers.extend(turn_servers)
-
-RTC_CONFIGURATION = RTCConfiguration({"iceServers": ice_servers})
-
-
-col1, col2 = st.columns([3, 1])
-
-with col1:
-    webrtc_ctx = webrtc_streamer(
-        key="drowsiness-detection",
-        # The factory now correctly passes the queues from session_state
-        video_processor_factory=VideoProcessor,
-        rtc_configuration=RTC_CONFIGURATION,
-        media_stream_constraints={"video": True, "audio": False},
-        async_processing=True,
-    )
-
-with col2:
-    st.header("System Status")
-    audio_placeholder = st.empty()
-    if not webrtc_ctx.state.playing:
-        st.warning("System Inactive.")
-    else:
-        st.success("✅ System Active & Monitoring")
-
-    st.subheader("Live Status:")
-    status_placeholder = st.empty()
-
-if webrtc_ctx.state.playing:
-    try:
-        # This now reads from the correct queue that the processor is writing to.
-        status_result = st.session_state.status_queue.get(timeout=0.1)
-        st.session_state.last_status = status_result
-    except queue.Empty:
-        pass
-
-    with status_placeholder.container():
-        last_status = st.session_state.last_status
-        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():
-            autoplay_audio(audio_data)
-    except queue.Empty:
-        pass
-
-    time.sleep(0.1)
-    st.rerun()
-else:
-    with status_placeholder.container():
-        st.info("✔️ Driver is Awake")

src/alerting/alert_system.py

@@ -1,57 +0,0 @@
-# drive_paddy/alerting/alert_system.py
-import time, os, io, google.generativeai as genai
-from gtts import gTTS
-class BaseAlerter:
-    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, level="Very Drowsy"): raise NotImplementedError
-    def reset_alert(self):
-        if self.alert_on: print("Resetting Alert."); self.alert_on = False
-
-class FileAlertSystem(BaseAlerter):
-    def __init__(self, config):
-        super().__init__(config)
-        self.audio_bytes = None
-        try:
-            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 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):
-    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:
-            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 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):
-    if config.get('gemini_api', {}).get('enabled', False) and api_key: return GeminiAlertSystem(config, api_key)
-    return FileAlertSystem(config)

src/detection/base_processor.py

@@ -1,26 +0,0 @@
-# drive_paddy/detection/base_processor.py
-from abc import ABC, abstractmethod
-
-class BaseProcessor(ABC):
-    """
-    Abstract Base Class for a drowsiness detection processor.
-    
-    This defines the common interface that all detection strategies
-    (e.g., Geometric, CNN Model) must follow.
-    """
-    
-    @abstractmethod
-    def process_frame(self, frame):
-        """
-        Processes a single video frame to detect drowsiness.
-
-        Args:
-            frame: The video frame (as a NumPy array) to process.
-
-        Returns:
-            A tuple containing:
-            - The processed frame (NumPy array) with visualizations.
-            - A boolean indicating if an alert should be triggered.
-        """
-        pass
-

src/detection/factory.py

@@ -1,22 +0,0 @@
-# drive_paddy/detection/factory.py
-from src.detection.strategies.geometric import GeometricProcessor
-from src.detection.strategies.cnn_model import CnnProcessor
-from src.detection.strategies.hybrid import HybridProcessor
-
-def get_detector(config):
-    """
-    Factory function to get the appropriate drowsiness detector.
-    """
-    strategy = config.get('detection_strategy', 'geometric')
-    
-    if strategy == 'geometric':
-        print("Initializing Geometric drowsiness detector...")
-        return GeometricProcessor(config)
-    elif strategy == 'cnn_model':
-        print("Initializing CNN Model drowsiness detector...")
-        return CnnProcessor(config)
-    elif strategy == 'hybrid':
-        print("Initializing Hybrid (Geometric + CNN) drowsiness detector...")
-        return HybridProcessor(config)
-    else:
-        raise ValueError(f"Unknown detection strategy: {strategy}")

src/detection/strategies/cnn_model.py

@@ -1,93 +0,0 @@
-# drive_paddy/detection/strategies/cnn_model.py
-from src.detection.base_processor import BaseProcessor
-import numpy as np
-import torch
-import torchvision.transforms as transforms
-from torchvision.models import efficientnet_b7
-import cv2
-from PIL import Image
-import os
-
-class CnnProcessor(BaseProcessor):
-    """
-    Drowsiness detection using a pre-trained EfficientNet-B7 model.
-    This version receives face landmarks from another processor instead of using dlib.
-    """
-    def __init__(self, config):
-        self.settings = config['cnn_model_settings']
-        self.model_path = self.settings['model_path']
-        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-        
-        # dlib is no longer needed.
-        # self.face_detector = dlib.get_frontal_face_detector()
-        
-        self.model = self._load_model()
-        
-        self.transform = transforms.Compose([
-            transforms.Resize((224, 224)),
-            transforms.ToTensor(),
-            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
-        ])
-
-    def _load_model(self):
-        """Loads the EfficientNet-B7 model and custom weights."""
-        if not os.path.exists(self.model_path):
-            print(f"Error: Model file not found at {self.model_path}")
-            return None
-            
-        try:
-            model = efficientnet_b7()
-            num_ftrs = model.classifier[1].in_features
-            model.classifier[1] = torch.nn.Linear(num_ftrs, 2)
-            model.load_state_dict(torch.load(self.model_path, map_location=self.device))
-            model.to(self.device)
-            model.eval()
-            print(f"CNN Model '{self.model_path}' loaded successfully on {self.device}.")
-            return model
-        except Exception as e:
-            print(f"Error loading CNN model: {e}")
-            return None
-
-    def process_frame(self, frame, face_landmarks=None):
-        """
-        Processes a frame using the CNN model with pre-supplied landmarks.
-        """
-        if self.model is None or face_landmarks is None:
-            return frame, {"cnn_prediction": False}
-
-        is_drowsy_prediction = False
-        h, w, _ = frame.shape
-        
-        landmarks = face_landmarks[0].landmark
-        
-        # Calculate bounding box from landmarks
-        x_coords = [lm.x * w for lm in landmarks]
-        y_coords = [lm.y * h for lm in landmarks]
-        x1, y1 = int(min(x_coords)), int(min(y_coords))
-        x2, y2 = int(max(x_coords)), int(max(y_coords))
-
-        # Add some padding to the bounding box
-        padding = 10
-        x1 = max(0, x1 - padding)
-        y1 = max(0, y1 - padding)
-        x2 = min(w, x2 + padding)
-        y2 = min(h, y2 + padding)
-
-        # Crop the face
-        face_crop = frame[y1:y2, x1:x2]
-        
-        if face_crop.size > 0:
-            pil_image = Image.fromarray(cv2.cvtColor(face_crop, cv2.COLOR_BGR2RGB))
-            image_tensor = self.transform(pil_image).unsqueeze(0).to(self.device)
-            
-            with torch.no_grad():
-                outputs = self.model(image_tensor)
-                _, preds = torch.max(outputs, 1)
-                if preds.item() == 1: # Assuming class 1 is 'drowsy'
-                    is_drowsy_prediction = True
-
-            cv2.rectangle(frame, (x1, y1), (x2, y2), (255, 255, 0), 2)
-            label = "Drowsy" if is_drowsy_prediction else "Awake"
-            cv2.putText(frame, f"CNN: {label}", (x1, y1 - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 0), 2)
-            
-        return frame, {"cnn_prediction": is_drowsy_prediction}

src/detection/strategies/geometric.py

@@ -1,111 +0,0 @@
-# drive_paddy/detection/strategies/geometric.py
-import cv2
-import mediapipe as mp
-import numpy as np
-import math
-from ..base_processor import BaseProcessor
-
-# --- Helper Functions (No changes here) ---
-def calculate_ear(eye_landmarks, frame_shape):
-    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
-
-def calculate_mar(mouth_landmarks, frame_shape):
-    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]); 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):
-    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)
-        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):
-        # Create a writable copy to prevent read-only errors from Gradio/OpenCV
-        frame = frame.copy()
-        
-        h, w, _ = frame.shape
-        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
-        brightness = np.mean(gray)
-        is_low_light = brightness < self.settings['low_light_thresh']
-
-        drowsiness_indicators = {
-            "drowsiness_level": "Awake", "lighting": "Good", "details": {}
-        }
-        face_landmarks = None
-
-        if not is_low_light:
-            img_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-            results = self.face_mesh.process(img_rgb)
-            face_landmarks = results.multi_face_landmarks
-
-            if face_landmarks:
-                landmarks = face_landmarks[0].landmark
-                score = 0
-                weights = self.settings['indicator_weights']
-
-                # --- Draw Facial Landmarks (Logic Added Back) ---
-                # This will draw the green dots for eyes and mouth to show what is being tracked.
-                eye_mouth_landmarks_indices = self.L_EYE + self.R_EYE + self.MOUTH
-                for idx in eye_mouth_landmarks_indices:
-                    lm = landmarks[idx]
-                    x, y = int(lm.x * w), int(lm.y * h)
-                    cv2.circle(frame, (x, y), 1, (0, 255, 0), -1)
-
-                # --- Drowsiness Calculations ---
-                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']
-                
-                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']
-
-                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']
-                
-                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
-        else: # is_low_light is True
-            drowsiness_indicators["lighting"] = "Low"
-        
-        # --- 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
-
-        cv2.rectangle(frame, (0, 0), (w, h), color, 10)
-        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)
-
-        return frame, drowsiness_indicators, face_landmarks

src/detection/strategies/hybrid.py

@@ -1,84 +0,0 @@
-# drive_paddy/detection/strategies/hybrid.py
-from src.detection.base_processor import BaseProcessor
-from src.detection.strategies.geometric import GeometricProcessor
-from src.detection.strategies.cnn_model import CnnProcessor
-import cv2
-import concurrent.futures
-
-class HybridProcessor(BaseProcessor):
-    """
-    Combines outputs from multiple detection strategies (Geometric and CNN)
-    concurrently to make a more robust and efficient drowsiness decision.
-    This version includes frame skipping for the CNN model to improve performance.
-    """
-    def __init__(self, config):
-        self.geometric_processor = GeometricProcessor(config)
-        self.cnn_processor = CnnProcessor(config)
-        self.weights = config['hybrid_settings']['weights']
-        self.alert_threshold = config['hybrid_settings']['alert_threshold']
-        self.active_alerts = {}
-        
-        # --- Performance Optimization ---
-        self.frame_counter = 0
-        self.cnn_process_interval = 10  # Run CNN every 10 frames
-        self.last_cnn_indicators = {"cnn_prediction": False} # Cache the last CNN result
-
-        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)
-
-    def process_frame(self, frame):
-        self.frame_counter += 1
-
-        # --- Concurrent Execution ---
-        # The geometric processor runs on every frame.
-        geo_future = self.executor.submit(self.geometric_processor.process_frame, frame.copy())
-
-        # The CNN processor only runs on specified intervals.
-        if self.frame_counter % self.cnn_process_interval == 0:
-            cnn_future = self.executor.submit(self.cnn_processor.process_frame, frame.copy())
-        
-        # Get the result from the geometric processor.
-        geo_frame, geo_indicators = geo_future.result()
-
-        # Get the CNN result if it was run, otherwise use the cached result.
-        if self.frame_counter % self.cnn_process_interval == 0:
-            _, self.last_cnn_indicators = cnn_future.result()
-        
-        cnn_indicators = self.last_cnn_indicators
-        
-        # Calculate weighted drowsiness score from the combined results.
-        score = 0
-        self.active_alerts.clear()
-
-        if geo_indicators.get("eye_closure"):
-            score += self.weights['eye_closure']
-            self.active_alerts['Eyes Closed'] = geo_indicators['details'].get('EAR', 0)
-        if geo_indicators.get("yawning"):
-            score += self.weights['yawning']
-            self.active_alerts['Yawning'] = geo_indicators['details'].get('MAR', 0)
-        if geo_indicators.get("head_nod"):
-            score += self.weights['head_nod']
-            self.active_alerts['Head Nod'] = geo_indicators['details'].get('Pitch', 0)
-        if geo_indicators.get("looking_away"):
-            score += self.weights['looking_away']
-            self.active_alerts['Looking Away'] = geo_indicators['details'].get('Yaw', 0)
-        if cnn_indicators.get("cnn_prediction"):
-            score += self.weights['cnn_prediction']
-            self.active_alerts['CNN Alert'] = 'Active'
-
-        # --- Visualization ---
-        output_frame = geo_frame
-        y_pos = 30
-        for alert, value in self.active_alerts.items():
-            text = f"{alert}: {value:.2f}" if isinstance(value, float) else alert
-            cv2.putText(output_frame, text, (10, y_pos), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 255), 2)
-            y_pos += 25
-            
-        cv2.putText(output_frame, f"Score: {score:.2f}", (output_frame.shape[1] - 150, 30),
-                           cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
-
-        alert_triggered = score >= self.alert_threshold
-        if alert_triggered:
-            cv2.rectangle(output_frame, (0, 0), (output_frame.shape[1], output_frame.shape[0]), (0, 0, 255), 5)
-
-        # Return the processed frame, the alert trigger, and the active alert details
-        return output_frame, alert_triggered, self.active_alerts

utils.py

@@ -1,76 +0,0 @@
-# utils.py
-
-import numpy as np
-import cv2
-# Removed: import random, string, generate_gibberish
-
-# Function to calculate Eye Aspect Ratio (EAR)
-def calculate_ear(eye_landmarks, frame_shape):
-    """
-    Calculates the Eye Aspect Ratio (EAR) for a given eye.
-
-    Args:
-        eye_landmarks: A list of 6 MediaPipe landmark objects for the eye.
-                       Expected order: [p1, p2, p3, p4, p5, p6]
-                       where p1, p4 are horizontal extremes, and p2, p3, p5, p6
-                       are vertical extremes.
-        frame_shape: Tuple (height, width) of the frame.
-
-    Returns:
-        The calculated EAR value.
-    """
-    if len(eye_landmarks) != 6:
-        # print("Warning: Expected 6 eye landmarks, but received", len(eye_landmarks)) # Optional warning
-        return 0.0 # Return 0 or handle error appropriately
-
-    # Convert MediaPipe landmarks to numpy array (pixel coordinates)
-    coords = np.array([(landmark.x * frame_shape[1], landmark.y * frame_shape[0])
-                      for landmark in eye_landmarks])
-
-    # Calculate the Euclidean distances between the two sets of vertical eye landmarks
-    # p2-p6 and p3-p5
-    vertical_dist1 = np.linalg.norm(coords[1] - coords[5])
-    vertical_dist2 = np.linalg.norm(coords[2] - coords[4])
-
-    # Calculate the Euclidean distance between the horizontal eye landmark
-    # p1-p4
-    horizontal_dist = np.linalg.norm(coords[0] - coords[3])
-
-    # Calculate the EAR
-    # Avoid division by zero
-    if horizontal_dist == 0:
-        return 0.0
-
-    ear = (vertical_dist1 + vertical_dist2) / (2.0 * horizontal_dist)
-
-    return ear
-
-def draw_landmarks(image, landmarks, connections=None, point_color=(0, 255, 0), connection_color=(255, 255, 255)):
-    """
-    Draws landmarks and connections on the image.
-
-    Args:
-        image: The image (numpy array) to draw on.
-        landmarks: A list of MediaPipe landmark objects.
-        connections: A list of tuples representing landmark connections (e.g., [(0, 1), (1, 2)]).
-        point_color: Color for the landmarks (BGR tuple).
-        connection_color: Color for the connections (BGR tuple).
-    """
-    if not landmarks:
-        return image
-
-    img_h, img_w, _ = image.shape
-    landmark_points = [(int(l.x * img_w), int(l.y * img_h)) for l in landmarks]
-
-    # Draw connections
-    if connections:
-        for connection in connections:
-            p1 = landmark_points[connection[0]]
-            p2 = landmark_points[connection[1]]
-            cv2.line(image, p1, p2, connection_color, 1)
-
-    # Draw points
-    for point in landmark_points:
-         cv2.circle(image, point, 2, point_color, -1)
-
-    return image