Edge-AI嵌入式疲劳检测：实时系统设计与部署

来源： Springer ICCIES 2026
发布时间： 2026年4月
链接： https://link.springer.com/chapter/10.1007/978-3-032-21628-1_44

---

核心洞察

嵌入式实时疲劳检测关键：

• 轻量级面部关键点模型
• EAR/MAR几何特征计算
• 双阈值滞回机制减少误报
• 15fps实时处理，<90ms延迟

---

一、系统架构

1.1 整体流程

摄像头输入
    ↓
面部关键点检测 (MediaPipe)
    ↓
EAR/MAR计算
    ↓
时序平滑
    ↓
双阈值滞回判定
    ↓
疲劳警报

1.2 关键技术

"""
Edge-AI实时疲劳检测系统
"""

import numpy as np
from collections import deque
from typing import Dict, Optional

class EdgeAIFatigueDetector:
    """
    边缘AI疲劳检测器
    
    特点：
    1. 轻量级：面部关键点模型 < 5MB
    2. 低延迟：单帧处理 < 50ms
    3. 低误报：双阈值滞回机制
    """
    
    def __init__(self, 
                 fps: int = 15,
                 window_size: int = 30):
        """
        Args:
            fps: 帧率
            window_size: 滑动窗口大小（帧数）
        """
        self.fps = fps
        self.window_size = window_size
        
        # 滑动窗口
        self.ear_history = deque(maxlen=window_size)
        self.mar_history = deque(maxlen=window_size)
        
        # 双阈值参数
        self.ear_threshold_high = 0.25  # 高阈值（唤醒）
        self.ear_threshold_low = 0.18   # 低阈值（触发）
        
        # 状态机
        self.state = 'alert'  # 'alert', 'drowsy'
        self.drowsy_frames = 0
        self.drowsy_threshold = int(0.5 * fps)  # 持续0.5秒
        
        # 疲劳评分
        self.drowsiness_score = 0.0
    
    def process_frame(self, 
                      landmarks: np.ndarray,
                      timestamp: float) -> Dict:
        """
        处理单帧
        
        Args:
            landmarks: 面部关键点 (468, 3) - MediaPipe格式
            timestamp: 时间戳
        
        Returns:
            result: 检测结果
        """
        # 1. 计算EAR
        ear_left = self.calculate_ear(landmarks, 'left')
        ear_right = self.calculate_ear(landmarks, 'right')
        ear = (ear_left + ear_right) / 2
        
        # 2. 计算MAR
        mar = self.calculate_mar(landmarks)
        
        # 3. 添加到历史
        self.ear_history.append(ear)
        self.mar_history.append(mar)
        
        # 4. 时序平滑
        ear_smoothed = self.smooth_signal(list(self.ear_history))
        mar_smoothed = self.smooth_signal(list(self.mar_history))
        
        # 5. 双阈值判定
        is_drowsy = self.hysteresis_detect(ear_smoothed)
        
        # 6. 计算疲劳评分
        self.drowsiness_score = self.calculate_score(ear_smoothed, mar_smoothed)
        
        return {
            'ear': ear,
            'ear_smoothed': ear_smoothed,
            'mar': mar,
            'mar_smoothed': mar_smoothed,
            'is_drowsy': is_drowsy,
            'drowsiness_score': self.drowsiness_score,
            'state': self.state,
            'timestamp': timestamp
        }
    
    def calculate_ear(self, landmarks: np.ndarray, side: str) -> float:
        """
        计算眼睛开度比
        
        MediaPipe关键点索引：
        - 左眼: 33, 160, 158, 133, 153, 144
        - 右眼: 362, 385, 387, 263, 373, 380
        """
        if side == 'left':
            indices = [33, 160, 158, 133, 153, 144]
        else:
            indices = [362, 385, 387, 263, 373, 380]
        
        p = landmarks[indices]
        
        # 垂直距离
        v1 = np.linalg.norm(p[1] - p[5])
        v2 = np.linalg.norm(p[2] - p[4])
        
        # 水平距离
        h = np.linalg.norm(p[0] - p[3])
        
        return (v1 + v2) / (2 * h + 1e-7)
    
    def calculate_mar(self, landmarks: np.ndarray) -> float:
        """
        计算嘴巴开度比
        
        MediaPipe关键点索引：
        - 外唇: 61, 291 (嘴角)
        - 上唇: 13
        - 下唇: 14
        """
        # 嘴角
        left = landmarks[61]
        right = landmarks[291]
        
        # 上下唇
        top = landmarks[13]
        bottom = landmarks[14]
        
        # 垂直距离
        v = np.linalg.norm(top - bottom)
        
        # 水平距离
        h = np.linalg.norm(left - right)
        
        return v / (h + 1e-7)
    
    def smooth_signal(self, signal: list) -> float:
        """
        时序平滑（移动平均）
        
        Args:
            signal: 信号历史
        
        Returns:
            smoothed: 平滑后的值
        """
        if len(signal) < 5:
            return signal[-1] if signal else 0.0
        
        # 5点移动平均
        return np.mean(signal[-5:])
    
    def hysteresis_detect(self, ear: float) -> bool:
        """
        双阈值滞回检测
        
        状态机：
        - alert → drowsy: EAR < low_threshold 持续 N 帧
        - drowsy → alert: EAR > high_threshold
        
        优势：减少边界抖动，降低误报
        """
        if self.state == 'alert':
            if ear < self.ear_threshold_low:
                self.drowsy_frames += 1
                
                if self.drowsy_frames >= self.drowsy_threshold:
                    self.state = 'drowsy'
                    return True
            else:
                self.drowsy_frames = 0
        
        elif self.state == 'drowsy':
            if ear > self.ear_threshold_high:
                self.state = 'alert'
                self.drowsy_frames = 0
            else:
                return True
        
        return False
    
    def calculate_score(self, ear: float, mar: float) -> float:
        """
        计算疲劳评分
        
        Args:
            ear: 平滑后的EAR
            mar: 平滑后的MAR
        
        Returns:
            score: 疲劳评分 [0, 1]
        """
        # EAR贡献
        ear_score = np.clip((self.ear_threshold_low - ear) / self.ear_threshold_low, 0, 1)
        
        # MAR贡献（打哈欠）
        mar_score = np.clip((mar - 0.5) / 0.5, 0, 1) if mar > 0.5 else 0
        
        # 加权融合
        score = 0.7 * ear_score + 0.3 * mar_score
        
        return np.clip(score, 0, 1)


class EmbeddedDeployer:
    """嵌入式部署工具"""
    
    @staticmethod
    def quantize_model(model_path: str, output_path: str):
        """
        模型量化（INT8）
        
        减少模型大小和推理时间
        """
        import tensorflow as tf
        
        # 加载模型
        model = tf.keras.models.load_model(model_path)
        
        # 量化
        converter = tf.lite.TFLiteConverter.from_keras_model(model)
        converter.optimizations = [tf.lite.Optimize.DEFAULT]
        
        # 保存
        tflite_model = converter.convert()
        with open(output_path, 'wb') as f:
            f.write(tflite_model)
        
        print(f"量化完成: {output_path}")
        print(f"模型大小: {len(tflite_model) / 1024:.1f} KB")
    
    @staticmethod
    def benchmark_model(model_path: str, 
                       test_input: np.ndarray,
                       iterations: int = 100):
        """
        模型性能基准测试
        """
        import tensorflow as tf
        import time
        
        # 加载TFLite模型
        interpreter = tf.lite.Interpreter(model_path=model_path)
        interpreter.allocate_tensors()
        
        input_details = interpreter.get_input_details()
        output_details = interpreter.get_output_details()
        
        # 预热
        interpreter.set_tensor(input_details[0]['index'], test_input)
        interpreter.invoke()
        
        # 基准测试
        times = []
        for _ in range(iterations):
            start = time.perf_counter()
            interpreter.invoke()
            end = time.perf_counter()
            times.append((end - start) * 1000)  # ms
        
        print(f"平均延迟: {np.mean(times):.2f} ms")
        print(f"最大延迟: {np.max(times):.2f} ms")
        print(f"最小延迟: {np.min(times):.2f} ms")
        print(f"标准差: {np.std(times):.2f} ms")


# 测试代码
if __name__ == "__main__":
    # 初始化检测器
    detector = EdgeAIFatigueDetector(fps=15, window_size=30)
    
    # 模拟数据
    landmarks = np.random.rand(468, 3)
    
    # 测试100帧
    for i in range(100):
        result = detector.process_frame(landmarks, timestamp=i/15)
        
        if result['is_drowsy']:
            print(f"[{i}] 检测到疲劳！评分: {result['drowsiness_score']:.2f}")
    
    print(f"\n最终状态: {result['state']}")
    print(f"EAR: {result['ear_smoothed']:.3f}")
    print(f"MAR: {result['mar_smoothed']:.3f}")

---

二、双阈值滞回机制

2.1 原理

        高阈值 (0.25)
         ↓
    ┌─────────────┐
    │   alert     │
    └─────────────┘
         ↑
        低阈值 (0.18)
         ↓
    ┌─────────────┐
    │   drowsy    │
    └─────────────┘

- alert → drowsy: EAR < 0.18 持续 0.5秒
- drowsy → alert: EAR > 0.25

2.2 优势

特性	单阈值	双阈值滞回
边界抖动	严重	无
误报率	8-12%	3-5%
响应延迟	立即	0.5秒

---

三、实验结果

3.1 测试场景

场景	描述
正常驾驶	清醒状态，正常眨眼
模拟疲劳	闭眼1-2秒，打哈欠
谈话	正常谈话，嘴巴运动
变光	光照变化测试

3.2 性能指标

指标	数值
平均帧率	15 fps
最差延迟	< 90 ms
检测准确率	> 92%
误报率	< 5%
模型大小	< 5 MB

---

四、IMS开发启示

4.1 部署架构

# 嵌入式疲劳检测部署配置
deployment:
  target: "qualcomm_qcs8255"
  
  model:
    name: "mediapipe_face_mesh"
    size: "5.2 MB"
    format: "tflite_int8"
  
  performance:
    fps: 15
    latency_ms: 50
    cpu_utilization: 30
    
  memory:
    ram_mb: 256
    flash_mb: 10

4.2 优化技巧

# 优化技巧示例

# 1. 降低输入分辨率
input_resolution = (320, 240)  # 原始640x480

# 2. 跳帧检测
detect_interval = 2  # 每2帧检测一次
frame_count = 0

if frame_count % detect_interval == 0:
    # 执行检测
    pass
frame_count += 1

# 3. ROI裁剪
roi = frame[100:400, 200:500]  # 只处理面部区域

# 4. 多线程流水线
# 线程1: 图像采集
# 线程2: 关键点检测
# 线程3: 特征计算

4.3 硬件选型

平台	CPU	NPU	内存	适用场景
QCS8255	8核	26 TOPS	8GB	高端车型
QCS6490	8核	12 TOPS	4GB	中端车型
i.MX8M Plus	4核	2.3 TOPS	2GB	入门车型

---

五、总结

维度	评估	备注
创新性	⭐⭐⭐⭐	双阈值滞回机制
实用性	⭐⭐⭐⭐⭐	直接部署
可复现性	⭐⭐⭐⭐⭐	开源实现
部署难度	⭐⭐⭐	需嵌入式优化
IMS价值	⭐⭐⭐⭐⭐	生产级方案

优先级： 🔥🔥🔥🔥🔥
建议落地： 作为疲劳检测核心算法

---

参考文献

1. Nguyen et al. “Edge-AI Smart Camera System for Early Driver Drowsiness Detection.” ICCIES 2026.
2. Soukupová & Čech. “Real-time eye blink detection using facial landmarks.” CVWW 2016.
3. MediaPipe. “Face Mesh.” Google, 2024.

---

发布时间： 2026-04-23
标签： #Edge-AI #嵌入式部署 #实时检测 #双阈值滞回 #疲劳检测 #IMS开发