驾驶员疲劳检测算法演进：从 PERCLOS 到深度学习

驾驶员疲劳检测算法演进：从 PERCLOS 到深度学习

疲劳检测的重要性

疲劳驾驶是交通事故的主要原因之一。Euro NCAP 2026 对疲劳检测提出明确要求：

要求 描述
检测速度 ≥50km/h 时激活
疲劳阈值 KSS >7 级或等效指标
检测类型 疲劳、微睡眠、睡眠
系统要求 直接监控（眼动追踪）

PERCLOS：经典疲劳指标

定义

PERCLOS（Percentage of Eye Closure）是最广泛使用的疲劳指标：

1
2
3
4
PERCLOS = (闭眼时间 / 时间窗口) × 100%

闭眼定义：眼睑开度 < 阈值（通常 20-30%）
时间窗口：通常 60 秒或 30 秒

PERCLOS 计算实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
import numpy as np
from typing import List, Tuple
from dataclasses import dataclass

@dataclass
class EyeState:
    """眼睛状态"""
    timestamp: float
    ear_left: float       # 左眼眼睑开度比 (Eye Aspect Ratio)
    ear_right: float      # 右眼眼睑开度比
    is_closed: bool = False


class PERCLOSCalculator:
    """
    PERCLOS 计算器
    
    计算驾驶员疲劳程度的经典指标
    """
    
    def __init__(self,
                 window_size_sec: float = 60.0,
                 closed_threshold: float = 0.2,
                 fps: int = 30):
        """
        初始化
        
        Args:
            window_size_sec: 时间窗口大小（秒）
            closed_threshold: 闭眼阈值（EAR < threshold 视为闭眼）
            fps: 帧率
        """
        self.window_size_sec = window_size_sec
        self.closed_threshold = closed_threshold
        self.fps = fps
        
        # 滑动窗口
        self.window_frames = int(window_size_sec * fps)
        self.eye_states: List[EyeState] = []
        
        # PERCLOS 历史记录
        self.perclos_history: List[Tuple[float, float]] = []
        
    def update(self, ear_left: float, ear_right: float, timestamp: float) -> dict:
        """
        更新 PERCLOS
        
        Args:
            ear_left: 左眼 EAR
            ear_right: 右眼 EAR
            timestamp: 时间戳
            
        Returns:
            result: PERCLOS 结果
        """
        # 计算平均 EAR
        ear_mean = (ear_left + ear_right) / 2
        
        # 判断闭眼状态
        is_closed = ear_mean < self.closed_threshold
        
        # 添加到窗口
        self.eye_states.append(EyeState(
            timestamp=timestamp,
            ear_left=ear_left,
            ear_right=ear_right,
            is_closed=is_closed
        ))
        
        # 维护窗口大小
        cutoff_time = timestamp - self.window_size_sec
        while self.eye_states and self.eye_states[0].timestamp < cutoff_time:
            self.eye_states.pop(0)
        
        # 计算 PERCLOS
        perclos = self._calculate_perclos()
        
        # 判断疲劳等级
        fatigue_level = self._classify_fatigue(perclos)
        
        return {
            'perclos': perclos,
            'fatigue_level': fatigue_level,
            'ear_mean': ear_mean,
            'is_closed': is_closed,
            'window_samples': len(self.eye_states)
        }
    
    def _calculate_perclos(self) -> float:
        """计算 PERCLOS 百分比"""
        if len(self.eye_states) < self.window_frames * 0.5:
            return 0.0
        
        closed_count = sum(1 for state in self.eye_states if state.is_closed)
        perclos = (closed_count / len(self.eye_states)) * 100
        
        return perclos
    
    def _classify_fatigue(self, perclos: float) -> str:
        """
        分类疲劳等级
        
        基于 PERCLOS 阈值判断疲劳程度
        """
        if perclos < 15:
            return 'normal'       # 正常
        elif perclos < 30:
            return 'mild'         # 轻度疲劳
        elif perclos < 50:
            return 'moderate'     # 中度疲劳
        else:
            return 'severe'       # 重度疲劳


# Eye Aspect Ratio (EAR) 计算
def calculate_ear(landmarks: np.ndarray, eye_indices: List[int]) -> float:
    """
    计算眼睑开度比 (Eye Aspect Ratio)
    
    EAR = (|p2-p6| + |p3-p5|) / (2 × |p1-p4|)
    
    Args:
        landmarks: 面部关键点，shape=(N, 2)
        eye_indices: 眼睛 6 个关键点的索引
        
    Returns:
        ear: 眼睑开度比
    """
    # 眼睛关键点
    # p1 ─── p4 (水平方向)
    #  │     │
    # p2 p3  p5 p6
    
    p1 = landmarks[eye_indices[0]]
    p2 = landmarks[eye_indices[1]]
    p3 = landmarks[eye_indices[2]]
    p4 = landmarks[eye_indices[3]]
    p5 = landmarks[eye_indices[4]]
    p6 = landmarks[eye_indices[5]]
    
    # 垂直距离
    vertical_1 = np.linalg.norm(p2 - p6)
    vertical_2 = np.linalg.norm(p3 - p5)
    
    # 水平距离
    horizontal = np.linalg.norm(p1 - p4)
    
    # EAR
    ear = (vertical_1 + vertical_2) / (2.0 * horizontal)
    
    return ear

PERCLOS 的局限

局限性 描述
单一指标 仅依赖闭眼时间，忽略其他疲劳信号
阈值敏感 不同人群、光照条件阈值不同
误检问题 眨眼、打喷嚏可能误判为疲劳
响应延迟 需要累积足够时间窗口

微睡眠检测

定义

微睡眠（Microsleep）是指持续 1-2 秒的非自愿闭眼。

检测算法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
class MicrosleepDetector:
    """
    微睡眠检测器
    
    检测 1-2 秒的非自愿闭眼事件
    """
    
    def __init__(self,
                 min_duration_sec: float = 1.0,
                 max_duration_sec: float = 2.0,
                 closed_threshold: float = 0.2):
        self.min_duration_sec = min_duration_sec
        self.max_duration_sec = max_duration_sec
        self.closed_threshold = closed_threshold
        
        # 状态跟踪
        self.eyes_closed_start: float = None
        self.currently_closed: bool = False
        
        # 微睡眠事件记录
        self.microsleep_events: List[dict] = []
        
    def update(self, ear: float, timestamp: float) -> dict:
        """
        更新微睡眠检测
        
        Args:
            ear: 当前 EAR
            timestamp: 时间戳
            
        Returns:
            result: 检测结果
        """
        is_closed = ear < self.closed_threshold
        
        result = {
            'microsleep_detected': False,
            'eyes_closed_duration': 0.0
        }
        
        if is_closed:
            if not self.currently_closed:
                # 开始闭眼
                self.eyes_closed_start = timestamp
                self.currently_closed = True
            else:
                # 持续闭眼
                duration = timestamp - self.eyes_closed_start
                result['eyes_closed_duration'] = duration
                
                # 检测微睡眠
                if self.min_duration_sec <= duration <= self.max_duration_sec:
                    result['microsleep_detected'] = True
        else:
            if self.currently_closed:
                # 闭眼结束
                duration = timestamp - self.eyes_closed_start
                
                # 记录事件
                if duration >= self.min_duration_sec:
                    event_type = 'microsleep' if duration <= self.max_duration_sec else 'sleep'
                    
                    self.microsleep_events.append({
                        'start_time': self.eyes_closed_start,
                        'end_time': timestamp,
                        'duration': duration,
                        'type': event_type
                    })
                
                # 重置状态
                self.eyes_closed_start = None
                self.currently_closed = False
        
        return result


class SleepDetector:
    """
    睡眠检测器
    
    检测持续 ≥3 秒的闭眼（睡眠）
    """
    
    def __init__(self, threshold_sec: float = 3.0):
        self.threshold_sec = threshold_sec
        self.microsleep_detector = MicrosleepDetector(
            min_duration_sec=1.0,
            max_duration_sec=10.0  # 扩大范围
        )
        
    def update(self, ear: float, timestamp: float) -> dict:
        """更新睡眠检测"""
        result = self.microsleep_detector.update(ear, timestamp)
        
        # 判断睡眠
        if result['eyes_closed_duration'] >= self.threshold_sec:
            result['sleep_detected'] = True
        else:
            result['sleep_detected'] = False
        
        return result

深度学习疲劳检测

多模态融合方案

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
import torch
import torch.nn as nn
from typing import Dict

class DeepFatigueDetector(nn.Module):
    """
    深度学习疲劳检测器
    
    融合多模态特征：
    - 眼动特征
    - 面部表情
    - 头部姿态
    - 行为模式
    """
    
    def __init__(self, 
                 eye_feature_dim: int = 32,
                 face_feature_dim: int = 64,
                 head_feature_dim: int = 16,
                 hidden_dim: int = 128):
        super().__init__()
        
        # 眼动特征编码器
        self.eye_encoder = nn.Sequential(
            nn.Linear(10, 32),
            nn.ReLU(),
            nn.Linear(32, eye_feature_dim),
            nn.ReLU()
        )
        
        # 面部表情编码器
        self.face_encoder = nn.Sequential(
            nn.Linear(17, 64),
            nn.ReLU(),
            nn.Linear(64, face_feature_dim),
            nn.ReLU()
        )
        
        # 头部姿态编码器
        self.head_encoder = nn.Sequential(
            nn.Linear(3, 16),
            nn.ReLU(),
            nn.Linear(16, head_feature_dim),
            nn.ReLU()
        )
        
        # 时序建模
        self.temporal_encoder = nn.LSTM(
            input_size=eye_feature_dim + face_feature_dim + head_feature_dim,
            hidden_size=hidden_dim,
            num_layers=2,
            batch_first=True,
            bidirectional=True
        )
        
        # 疲劳分类头
        self.fatigue_classifier = nn.Sequential(
            nn.Linear(hidden_dim * 2, 64),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(64, 4)  # 4 级：正常、轻度、中度、重度
        )
        
        # PERCLOS 回归头
        self.perclos_regressor = nn.Sequential(
            nn.Linear(hidden_dim * 2, 32),
            nn.ReLU(),
            nn.Linear(32, 1),
            nn.Sigmoid()
        )
        
    def forward(self, 
                eye_features: torch.Tensor,
                face_features: torch.Tensor,
                head_features: torch.Tensor) -> Dict[str, torch.Tensor]:
        """
        前向传播
        
        Args:
            eye_features: 眼动特征, shape=(batch, seq_len, 10)
            face_features: 面部特征, shape=(batch, seq_len, 17)
            head_features: 头部姿态, shape=(batch, seq_len, 3)
            
        Returns:
            outputs: 检测结果
        """
        batch_size, seq_len = eye_features.shape[:2]
        
        # 特征编码
        eye_encoded = self.eye_encoder(eye_features)
        face_encoded = self.face_encoder(face_features)
        head_encoded = self.head_encoder(head_features)
        
        # 特征融合
        fused_features = torch.cat([eye_encoded, face_encoded, head_encoded], dim=-1)
        
        # 时序建模
        temporal_features, _ = self.temporal_encoder(fused_features)
        
        # 取最后时刻特征
        last_features = temporal_features[:, -1, :]
        
        # 输出
        fatigue_logits = self.fatigue_classifier(last_features)
        perclos_pred = self.perclos_regressor(last_features) * 100
        
        return {
            'fatigue_logits': fatigue_logits,
            'fatigue_probs': torch.softmax(fatigue_logits, dim=-1),
            'perclos_prediction': perclos_pred
        }

数据集与训练

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class FatigueDataset(torch.utils.data.Dataset):
    """
    疲劳检测数据集
    
    数据格式：
    - 眼动特征：EAR、眨眼频率、瞳孔直径等
    - 面部特征：面部关键点、表情标签
    - 头部姿态：yaw、pitch、roll
    - 标签：疲劳等级、PERCLOS 值
    """
    
    def __init__(self, data_path: str, sequence_length: int = 300):
        self.sequence_length = sequence_length
        self.data = self._load_data(data_path)
        
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        sample = self.data[idx]
        
        return {
            'eye_features': torch.tensor(sample['eye_features'], dtype=torch.float32),
            'face_features': torch.tensor(sample['face_features'], dtype=torch.float32),
            'head_features': torch.tensor(sample['head_features'], dtype=torch.float32),
            'fatigue_label': torch.tensor(sample['fatigue_label'], dtype=torch.long),
            'perclos': torch.tensor(sample['perclos'], dtype=torch.float32)
        }
    
    def _load_data(self, path):
        # 加载数据
        import json
        with open(path, 'r') as f:
            return json.load(f)


def train_fatigue_model():
    """训练疲劳检测模型"""
    
    # 数据
    train_dataset = FatigueDataset('data/train_fatigue.json')
    train_loader = torch.utils.data.DataLoader(
        train_dataset, batch_size=32, shuffle=True
    )
    
    # 模型
    model = DeepFatigueDetector()
    model = model.cuda()
    
    # 损失函数
    cls_loss = nn.CrossEntropyLoss()
    reg_loss = nn.MSELoss()
    
    # 优化器
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
    
    # 训练
    for epoch in range(100):
        model.train()
        total_loss = 0
        
        for batch in train_loader:
            eye = batch['eye_features'].cuda()
            face = batch['face_features'].cuda()
            head = batch['head_features'].cuda()
            fatigue_label = batch['fatigue_label'].cuda()
            perclos = batch['perclos'].cuda()
            
            # 前向传播
            outputs = model(eye, face, head)
            
            # 损失计算
            loss_cls = cls_loss(outputs['fatigue_logits'], fatigue_label)
            loss_reg = reg_loss(outputs['perclos_prediction'].squeeze(), perclos)
            
            loss = loss_cls + 0.5 * loss_reg
            
            # 反向传播
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
        
        if (epoch + 1) % 10 == 0:
            print(f"Epoch {epoch+1}, Loss: {total_loss/len(train_loader):.4f}")
    
    # 保存模型
    torch.save(model.state_dict(), 'fatigue_detector.pth')

Euro NCAP 2026 疲劳检测要求

测试场景

场景 测试内容 通过条件
F-01 正常驾驶 不误报疲劳
F-02 PERCLOS ≥30% 检测并发出警告
F-03 微睡眠 1.5秒 ≤3秒发出一级警告
F-04 睡眠 ≥3秒 ≤4秒发出二级警告
F-05 持续疲劳 分级警告递增

KSS 量表对照

1
2
3
4
5
6
7
8
9
10
11
Karolinska Sleepiness Scale (KSS):

1 = 极度清醒
2 = 非常清醒
3 = 清醒
4 = 稍微清醒
5 = 既不清醒也不困倦
6 = 稍微困倦
7 = 困倦，但容易保持清醒 ← Euro NCAP 阈值
8 = 困倦，需要努力保持清醒
9 = 非常困倦，极力保持清醒

IMS 开发启示

算法选型建议

方案 适用场景 性能
PERCLOS 快速部署、资源受限 基础可用
PERCLOS + 微睡眠 Euro NCAP 合规 满足要求
深度学习多模态 高精度要求 最佳性能

硬件配置

组件 要求
红外摄像头 ≥25fps, 全局快门
处理器 ≥2 TOPS (PERCLOS), ≥8 TOPS (深度学习)
内存 ≥1GB

部署优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 模型量化
def quantize_model(model):
    """动态量化"""
    quantized_model = torch.quantization.quantize_dynamic(
        model,
        {nn.LSTM, nn.Linear},
        dtype=torch.qint8
    )
    return quantized_model

# ONNX 导出
def export_to_onnx(model, path):
    """导出为 ONNX"""
    model.eval()
    dummy_input = {
        'eye_features': torch.randn(1, 300, 10),
        'face_features': torch.randn(1, 300, 17),
        'head_features': torch.randn(1, 300, 3)
    }
    
    torch.onnx.export(
        model,
        (dummy_input['eye_features'], 
         dummy_input['face_features'], 
         dummy_input['head_features']),
        path,
        opset_version=11
    )

参考来源：

驾驶员监控 > 算法研究
#DMS #深度学习 #疲劳检测 #Euro NCAP #PERCLOS #KSS
驾驶员疲劳检测算法演进：从 PERCLOS 到深度学习
https://dapalm.com/2026/06/13/2026-06-13-Driver-Fatigue-Detection-PERCLOS-Deep-Learning/
作者
Mars
发布于
2026年6月13日
许可协议