AI-Driven Driver Monitoring System: Fatigue and Attention Detection

AI-Driven Driver Monitoring System: Fatigue and Attention Detection

According to WHO, 20% of serious highway crashes are linked to driver drowsiness. Off-the-shelf DMS (Driver Monitoring System) solutions are expensive and not adapted to specific fleets. We develop custom AI-based monitoring for driver fatigue and behavior—turnkey, from scratch, or on your existing hardware. Over the past years, we have delivered over 80 computer vision projects; DMS is one of our core areas.

The system places a camera in the cabin, pointed at the driver's face, and tracks in real time signs of fatigue, distraction, and phone usage. Below we break down the architecture using a real deployment in a bus fleet of 80 vehicles.

Why PERCLOS is the Gold Standard

Fatigue manifests through several measurable facial parameters. The most reliable is PERCLOS (Percentage of Eye Closure): the proportion of time the eyes are closed more than 80% over the last 60 seconds. We use it as the base metric.

• PERCLOS > 15% = warning, > 25% = critical
• Blink rate: normal 12–20 blinks/min, fatigue < 8 or > 30
• Blink duration: normal 150–200 ms, fatigue > 350 ms
• Head pitch: nodding down > 15° indicates falling asleep
• Gaze direction: distraction if > 3 seconds away

How AI Detects Eye Closure and Distraction

We use PERCLOS as a continuous metric combined with head pose estimation. Implementation uses MediaPipe FaceMesh and solvePnP:

import cv2
import numpy as np
import mediapipe as mp
from collections import deque
import time

class DriverMonitoringSystem:
    def __init__(self, config: dict):
        # MediaPipe Face Mesh: 478 landmarks, fast, good on embedded
        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
        )
        # Key point indices (MediaPipe Face Mesh)
        self.LEFT_EYE = [362, 385, 387, 263, 373, 380]
        self.RIGHT_EYE = [33, 160, 158, 133, 153, 144]
        self.LEFT_IRIS = [474, 475, 476, 477]
        self.RIGHT_IRIS = [469, 470, 471, 472]
        # Buffers for temporal analysis
        window = config.get('window_sec', 60) * config.get('fps', 30)
        self.ear_buffer = deque(maxlen=window)  # Eye Aspect Ratio
        self.blink_buffer = deque(maxlen=window)  # 1 if blink
        self.head_pose_buffer = deque(maxlen=300)  # 10 seconds
        # Current blink state
        self.in_blink = False
        self.blink_start = None
        self.alert_callbacks = config.get('alert_callbacks', [])

    def _eye_aspect_ratio(self, landmarks: np.ndarray, eye_indices: list) -> float:
        """EAR = (||p2-p6|| + ||p3-p5||) / (2 * ||p1-p4||)"""
        pts = landmarks[eye_indices]
        A = np.linalg.norm(pts[1] - pts[5])
        B = np.linalg.norm(pts[2] - pts[4])
        C = np.linalg.norm(pts[0] - pts[3])
        return (A + B) / (2.0 * C + 1e-6)

    def _estimate_head_pose(self, landmarks: np.ndarray, frame_size: tuple) -> dict:
        """Solvepnp for pitch/yaw/roll estimation"""
        model_points = np.float32([
            [0.0, 0.0, 0.0],  # nose tip
            [0.0, -330.0, -65.0],  # chin
            [-225.0, 170.0, -135.0],  # left eye corner
            [225.0, 170.0, -135.0],  # right eye corner
            [-150.0, -150.0, -125.0],  # left mouth corner
            [150.0, -150.0, -125.0],  # right mouth corner
        ])
        key_indices = [1, 152, 263, 33, 287, 57]
        image_points = np.float32([landmarks[i] for i in key_indices])
        h, w = frame_size
        cam_matrix = np.float32([[w, 0, w/2], [0, w, h/2], [0, 0, 1]])
        dist_coeffs = np.zeros((4, 1))
        success, rvec, tvec = cv2.solvePnP(
            model_points,
            image_points,
            cam_matrix,
            dist_coeffs
        )
        if not success:
            return {'pitch': 0, 'yaw': 0, 'roll': 0}
        rmat, _ = cv2.Rodrigues(rvec)
        angles = cv2.RQDecomp3x3(rmat)[0]
        return {'pitch': angles[0], 'yaw': angles[1], 'roll': angles[2]}

    def process_frame(self, frame: np.ndarray) -> dict:
        rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        results = self.face_mesh.process(rgb)
        if not results.multi_face_landmarks:
            return {'driver_detected': False, 'alerts': []}
        h, w = frame.shape[:2]
        lm = results.multi_face_landmarks[0].landmark
        landmarks = np.array([[l.x * w, l.y * h] for l in lm])
        # EAR for both eyes
        ear_left = self._eye_aspect_ratio(landmarks, self.LEFT_EYE)
        ear_right = self._eye_aspect_ratio(landmarks, self.RIGHT_EYE)
        ear = (ear_left + ear_right) / 2.0
        self.ear_buffer.append(ear)
        # Blink detection
        ear_threshold = 0.22
        if ear < ear_threshold:
            if not self.in_blink:
                self.in_blink = True
                self.blink_start = time.time()
        else:
            if self.in_blink:
                blink_duration = time.time() - self.blink_start
                self.blink_buffer.append(blink_duration)
                self.in_blink = False
        # PERCLOS: fraction of frames with EAR < threshold in last 60 sec
        perclos = sum(1 for e in self.ear_buffer if e < ear_threshold) / max(len(self.ear_buffer), 1)
        # Head pose
        head_pose = self._estimate_head_pose(landmarks, (h, w))
        self.head_pose_buffer.append(head_pose)
        alerts = self._generate_alerts(perclos, head_pose)
        return {
            'driver_detected': True,
            'ear': ear,
            'perclos': perclos,
            'head_pose': head_pose,
            'recent_blink_durations': list(self.blink_buffer)[-5:],
            'alerts': alerts
        }

    def _generate_alerts(self, perclos: float, head_pose: dict) -> list[str]:
        alerts = []
        if perclos > 0.25:
            alerts.append('DROWSINESS_CRITICAL')
        elif perclos > 0.15:
            alerts.append('DROWSINESS_WARNING')
        if head_pose['pitch'] < -20:
            alerts.append('HEAD_NODDING')
        if abs(head_pose['yaw']) > 30:
            alerts.append('DISTRACTION_YAW')
        return alerts

How We Detect Phone Use

A separate YOLOv8n model fine-tuned on the Driver Phone Use Dataset. Simple:

class PhoneUseDetector:
    def __init__(self, model_path: str):
        self.model = YOLO(model_path)
        self.detection_buffer = deque(maxlen=15)  # 0.5 sec @ 30fps

    def detect(self, frame: np.ndarray) -> bool:
        dets = self.model(frame, conf=0.6, classes=['phone', 'cell phone'])
        self.detection_buffer.append(len(dets[0].boxes) > 0)
        # Alert if phone detected in 10+ of last 15 frames
        return sum(self.detection_buffer) >= 10

Performance on Embedded

On Qualcomm SA8295P (ADAS SoC): total <25 ms — real time at 30 FPS without drops. On Raspberry Pi 4 (4GB RAM): 35 ms at 720p — acceptable for commercial fleet monitoring. We optimize the model for target hardware: use INT8 quantization via ONNX Runtime, trim YOLO backbone to Nano if needed to fit within 15 ms on older SoCs.

How Temporal Smoothing Improves Accuracy

PERCLOS alone gives false alarms from glare or head turns. Combining EAR, head pose, and blink rate through a sliding window delivers >95% accuracy on our test set.

Case Study: Bus Fleet, 80 Vehicles (from Our Practice)

We installed DSM (Driver Safety Monitor) in 80 city route buses. Over several months:

• 1,240 DROWSINESS_WARNING events recorded, 87 CRITICAL
• After system deployment and driver training: critical event reduction of 64%
• 340 instances of phone use while driving recorded — forwarded to HR

Why it worked? Our DMS outperforms open-source solutions (e.g., OpenFace) by 2–3x in eye-closure detection accuracy and is 40% faster due to quantized models and careful temporal smoothing.

What's Included in the Work

• Requirements analysis and hardware selection (camera, SoC/consumables)
• Model development and calibration for specific cabin type
• Integration with CAN bus, alert system, and cloud platform
• Documentation, driver and dispatcher training
• 12-month warranty support, extendable by contract

Process

1. Analytics and prototype (2–4 weeks): select sensors, build initial pipeline, test in real cabin.
2. Production solution design (1–2 weeks): architecture, MLOps, retraining pipeline.
3. Implementation (4–8 weeks): fine-tune YOLO, adjust thresholds, integrate with onboard systems.
4. Testing (2 weeks): A/B test on 3–5 vehicles, collect metrics.
5. Deployment and monitoring (2–4 weeks): roll out to fleet, connect analytics.

Typical DMS Implementation Mistakes

• Relying solely on PERCLOS without head pose analysis: the driver may close eyes due to bright light, not fatigue.
• Ignoring temporal filtering: a single frame with closed eyes is not an alert; smoothing is needed.
• Not accounting for race and facial features: our model is trained on multi-ethnic datasets and has a non-bias certification.

Get a consultation with a computer vision engineer experienced in DMS — we will send a technical specification and preliminary implementation plan within a week.