MG-YOLOv8多特征融合疲劳检测：轻量级实时方案

论文来源： Journal of Imaging 2025
论文标题： Lightweight and Real-Time Driver Fatigue Detection Based on MG-YOLOv8 with Facial Multi-Feature Fusion
链接： https://doi.org/10.3390/jimaging11110385

---

核心创新

MG-YOLOv8多特征融合疲劳检测网络，将眨眼频率(BF)、打哈欠频率(YF)、点头频率(NF)和PERCLOS融合到单一YOLOv8框架，实现实时、轻量级疲劳检测。

关键突破：

1. 单网络多任务检测（眼/嘴/头）
2. 模型大小仅8.2MB（量化后<3MB）
3. 速度>45fps（GPU）/ >15fps（CPU）
4. 准确率99.2%

---

技术架构

1. 多特征融合框架

graph TB
    A[面部图像] --> B[MG-YOLOv8骨干]
    
    B --> C[眼部检测头]
    B --> D[嘴部检测头]
    B --> E[头部检测头]
    
    C --> F[眨眼频率BF]
    C --> G[PERCLOS]
    
    D --> H[打哈欠频率YF]
    
    E --> I[点头频率NF]
    
    F --> J[多特征融合]
    G --> J
    H --> J
    I --> J
    
    J --> K[疲劳综合判定]

2. MG-YOLOv8网络结构

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Dict, List
import numpy as np

class MultiTaskHead(nn.Module):
    """多任务检测头
    
    同时检测：眼睛状态、嘴部状态、头部姿态
    """
    
    def __init__(self, in_channels: int = 256, num_anchors: int = 1):
        super().__init__()
        
        # 眼部检测头
        self.eye_head = nn.Sequential(
            nn.Conv2d(in_channels, 128, 3, padding=1),
            nn.ReLU(),
            nn.Conv2d(128, num_anchors * 3, 1)  # [open, closed, unknown]
        )
        
        # 嘴部检测头
        self.mouth_head = nn.Sequential(
            nn.Conv2d(in_channels, 128, 3, padding=1),
            nn.ReLU(),
            nn.Conv2d(128, num_anchors * 3, 1)  # [normal, yawn, unknown]
        )
        
        # 头部姿态头
        self.head_pose_head = nn.Sequential(
            nn.Conv2d(in_channels, 128, 3, padding=1),
            nn.ReLU(),
            nn.Conv2d(128, num_anchors * 4, 1)  # [pitch, yaw, roll, confidence]
        )
    
    def forward(self, x: torch.Tensor) -> Dict:
        """
        Args:
            x: (B, C, H, W) 特征图
        
        Returns:
            outputs: 各任务输出
        """
        eye_out = self.eye_head(x)
        mouth_out = self.mouth_head(x)
        head_out = self.head_pose_head(x)
        
        return {
            'eye': eye_out,
            'mouth': mouth_out,
            'head_pose': head_out
        }


class MGYOLOv8(nn.Module):
    """MG-YOLOv8多特征融合疲劳检测网络"""
    
    def __init__(self):
        super().__init__()
        
        # YOLOv8骨干（轻量级）
        self.backbone = self._build_backbone()
        
        # 特征金字塔
        self.fpn = self._build_fpn()
        
        # 多任务检测头
        self.heads = nn.ModuleList([
            MultiTaskHead(256) for _ in range(3)  # 3个尺度
        ])
    
    def _build_backbone(self) -> nn.Module:
        """构建骨干网络"""
        # 使用YOLOv8-nano骨干
        import torchvision.models as models
        
        # MobileNetV3作为轻量骨干
        model = models.mobilenet_v3_small(pretrained=True)
        features = list(model.features.children())
        
        return nn.Sequential(*features)
    
    def _build_fpn(self) -> nn.Module:
        """构建特征金字塔"""
        return nn.Sequential(
            nn.Conv2d(576, 256, 1),
            nn.BatchNorm2d(256),
            nn.ReLU()
        )
    
    def forward(self, x: torch.Tensor) -> Dict:
        """
        前向传播
        
        Args:
            x: (B, 3, 640, 640) 输入图像
        
        Returns:
            outputs: 检测输出
        """
        # 骨干特征提取
        feat = self.backbone(x)  # (B, 576, 20, 20)
        
        # FPN
        feat = self.fpn(feat)  # (B, 256, 20, 20)
        
        # 多任务检测
        outputs = [head(feat) for head in self.heads]
        
        # 合并输出
        combined = self._combine_outputs(outputs)
        
        return combined
    
    def _combine_outputs(self, outputs: List[Dict]) -> Dict:
        """合并多尺度输出"""
        eye_out = torch.cat([o['eye'] for o in outputs], dim=1)
        mouth_out = torch.cat([o['mouth'] for o in outputs], dim=1)
        head_out = torch.cat([o['head_pose'] for o in outputs], dim=1)
        
        return {
            'eye': eye_out,
            'mouth': mouth_out,
            'head_pose': head_out
        }


class MultiFeatureFusion:
    """多特征融合疲劳判定"""
    
    def __init__(self):
        # 特征权重
        self.weights = {
            'BF': 0.25,      # 眨眼频率
            'YF': 0.20,      # 打哈欠频率
            'NF': 0.15,      # 点头频率
            'PERCLOS': 0.40  # PERCLOS
        }
        
        # 正常范围
        self.normal_ranges = {
            'BF': (10, 25),      # 眨眼频率（次/分钟）
            'YF': (0, 3),        # 打哈欠频率（次/分钟）
            'NF': (0, 2),        # 点头频率（次/分钟）
            'PERCLOS': (0, 15)   # PERCLOS百分比
        }
    
    def compute_fatigue_score(self, 
                               bf: float, 
                               yf: float, 
                               nf: float, 
                               perclos: float) -> float:
        """
        计算综合疲劳得分
        
        Args:
            bf: 眨眼频率
            yf: 打哈欠频率
            nf: 点头频率
            perclos: PERCLOS值
        
        Returns:
            score: 疲劳得分（0-1）
        """
        scores = {}
        
        # 1. 眨眼频率得分
        if bf < self.normal_ranges['BF'][0]:
            scores['BF'] = 0.3  # 眨眼过少
        elif bf > self.normal_ranges['BF'][1]:
            scores['BF'] = (bf - self.normal_ranges['BF'][1]) / 20.0
        else:
            scores['BF'] = 0.0
        
        # 2. 打哈欠频率得分
        if yf > self.normal_ranges['YF'][1]:
            scores['YF'] = (yf - self.normal_ranges['YF'][1]) / 5.0
        else:
            scores['YF'] = 0.0
        
        # 3. 点头频率得分
        if nf > self.normal_ranges['NF'][1]:
            scores['NF'] = (nf - self.normal_ranges['NF'][1]) / 5.0
        else:
            scores['NF'] = 0.0
        
        # 4. PERCLOS得分
        if perclos > self.normal_ranges['PERCLOS'][1]:
            scores['PERCLOS'] = (perclos - self.normal_ranges['PERCLOS'][1]) / 50.0
        else:
            scores['PERCLOS'] = 0.0
        
        # 5. 加权融合
        total_score = sum(scores[k] * self.weights[k] for k in scores)
        
        return np.clip(total_score, 0, 1)
    
    def classify_fatigue(self, score: float) -> str:
        """分类疲劳等级"""
        if score < 0.2:
            return 'normal'
        elif score < 0.5:
            return 'mild'
        elif score < 0.8:
            return 'moderate'
        else:
            return 'severe'


# 完整疲劳检测管道
class MGYOLOv8Pipeline:
    """MG-YOLOv8疲劳检测管道"""
    
    def __init__(self, model_path: str = None):
        self.model = MGYOLOv8()
        
        if model_path:
            self.model.load_state_dict(torch.load(model_path))
        
        self.model.eval()
        
        self.fusion = MultiFeatureFusion()
        
        # 时序统计
        self.blink_history = []
        self.yawn_history = []
        self.nod_history = []
        self.perclos_history = []
    
    def process_frame(self, frame: np.ndarray) -> Dict:
        """
        处理单帧
        
        Args:
            frame: (H, W, 3) BGR图像
        
        Returns:
            result: 检测结果
        """
        # 1. 预处理
        input_tensor = self._preprocess(frame)
        
        # 2. 模型推理
        with torch.no_grad():
            outputs = self.model(input_tensor)
        
        # 3. 后处理
        detections = self._postprocess(outputs)
        
        # 4. 更新时序统计
        self._update_history(detections)
        
        # 5. 计算特征
        bf = self._compute_blink_frequency()
        yf = self._compute_yawn_frequency()
        nf = self._compute_nod_frequency()
        perclos = self._compute_perclos()
        
        # 6. 融合判定
        fatigue_score = self.fusion.compute_fatigue_score(bf, yf, nf, perclos)
        fatigue_level = self.fusion.classify_fatigue(fatigue_score)
        
        return {
            'fatigue_score': fatigue_score,
            'fatigue_level': fatigue_level,
            'features': {
                'blink_frequency': bf,
                'yawn_frequency': yf,
                'nod_frequency': nf,
                'perclos': perclos
            }
        }
    
    def _preprocess(self, frame: np.ndarray) -> torch.Tensor:
        """预处理"""
        # 缩放、归一化
        frame_resized = cv2.resize(frame, (640, 640))
        frame_rgb = cv2.cvtColor(frame_resized, cv2.COLOR_BGR2RGB)
        frame_normalized = frame_rgb.astype(np.float32) / 255.0
        
        # 转换为tensor
        tensor = torch.from_numpy(frame_normalized).permute(2, 0, 1).unsqueeze(0)
        
        return tensor
    
    def _postprocess(self, outputs: Dict) -> Dict:
        """后处理"""
        # 简化：返回模拟检测结果
        return {
            'eye_state': 'open',  # 或 'closed'
            'mouth_state': 'normal',  # 或 'yawn'
            'head_pose': {'pitch': 0, 'yaw': 0, 'roll': 0}
        }
    
    def _update_history(self, detections: Dict):
        """更新历史记录"""
        self.blink_history.append(detections['eye_state'] == 'closed')
        self.yawn_history.append(detections['mouth_state'] == 'yawn')
        # ...
    
    def _compute_blink_frequency(self) -> float:
        """计算眨眼频率"""
        if len(self.blink_history) < 30:
            return 15.0  # 默认值
        
        # 最近30秒的眨眼次数
        blinks = sum(self.blink_history[-900:])  # 30fps × 30秒
        return blinks / 30.0 * 60.0  # 次/分钟
    
    def _compute_yawn_frequency(self) -> float:
        """计算打哈欠频率"""
        if len(self.yawn_history) < 30:
            return 0.0
        
        yawns = sum(self.yawn_history[-900:])
        return yawns / 30.0 * 60.0
    
    def _compute_nod_frequency(self) -> float:
        """计算点头频率"""
        return 1.0  # 简化
    
    def _compute_perclos(self) -> float:
        """计算PERCLOS"""
        if len(self.blink_history) < 1800:
            return 10.0
        
        closed = sum(self.blink_history[-1800:])
        return closed / 1800.0 * 100.0


# 测试
if __name__ == "__main__":
    import cv2
    
    pipeline = MGYOLOv8Pipeline()
    
    # 模拟视频流
    frame = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
    
    result = pipeline.process_frame(frame)
    
    print(f"疲劳得分: {result['fatigue_score']:.2f}")
    print(f"疲劳等级: {result['fatigue_level']}")
    print(f"特征: {result['features']}")

---

性能指标

指标	MG-YOLOv8	传统方法
准确率	99.2%	92-95%
帧率(GPU)	45+ fps	30 fps
帧率(CPU)	15+ fps	5 fps
模型大小	8.2MB	50MB+
量化后大小	2.8MB	-

---

IMS开发启示

1. 模型量化部署

# ONNX导出
python export.py --model mgyolov8.pth --output mgyolov8.onnx

# TensorRT优化
trtexec --onnx=mgyolov8.onnx --saveEngine=mgyolov8.trt --fp16

# INT8量化
trtexec --onnx=mgyolov8.onnx --int8 --calib=calib_data.bin

2. 多特征融合权重调优

根据实际数据调整权重：

• PERCLOS权重最高（40%）
• 眨眼频率次之（25%）
• 打哈欠和点头辅助（20%/15%）

---

参考文献

1. Journal of Imaging, “MG-YOLOv8 with Facial Multi-Feature Fusion”, 2025
2. YOLOv8, “Real-time Object Detection”, 2023

---

本文为MG-YOLOv8多特征融合疲劳检测的详细解读与代码实现，面向IMS开发者提供轻量级实时方案。