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 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209
| import torch import torch.nn as nn import numpy as np from dataclasses import dataclass
@dataclass class WindowState: """时间窗口状态""" current_size: float = 30.0 min_size: float = 5.0 max_size: float = 60.0 confidence: float = 0.5 trend: str = 'stable'
class ConfidenceAdaptiveWindow: """ 置信度驱动自适应时间窗口 原理: - 高置信度→缩短窗口→降低延迟 - 低置信度→扩大窗口→提升精度 - 类似生物注意力机制:信号可靠时快速反应 """ def __init__(self, min_window: float = 5.0, max_window: float = 60.0, init_window: float = 30.0, confidence_high: float = 0.85, confidence_low: float = 0.50, expand_rate: float = 1.5, shrink_rate: float = 0.7): self.min_window = min_window self.max_window = max_window self.current_window = init_window self.conf_high = confidence_high self.conf_low = confidence_low self.expand_rate = expand_rate self.shrink_rate = shrink_rate self.history = [] def update(self, confidence: float) -> float: """ 根据预测置信度调整窗口大小 Args: confidence: 当前预测置信度 [0, 1] Returns: new_window: 调整后的窗口大小(秒) """ self.history.append(confidence) if len(self.history) > 5: avg_conf = np.mean(self.history[-5:]) else: avg_conf = confidence if avg_conf > self.conf_high: self.current_window *= self.shrink_rate self.current_window = max(self.current_window, self.min_window) elif avg_conf < self.conf_low: self.current_window *= self.expand_rate self.current_window = min(self.current_window, self.max_window) return self.current_window
class FatigueDetector(nn.Module): """ 疲劳检测模型(轻量CNN+MC Dropout) 输出:疲劳分类 + 置信度 """ def __init__(self, n_classes: int = 3): super().__init__() self.backbone = nn.Sequential( nn.Conv2d(3, 16, 3, stride=2, padding=1), nn.BatchNorm2d(16), nn.ReLU6(), nn.Conv2d(16, 32, 3, stride=2, padding=1), nn.BatchNorm2d(32), nn.ReLU6(), nn.Conv2d(32, 64, 3, stride=2, padding=1), nn.BatchNorm2d(64), nn.ReLU6(), nn.AdaptiveAvgPool2d(1), nn.Flatten(), ) self.classifier = nn.Sequential( nn.Linear(64, 32), nn.ReLU(), nn.Dropout(0.2), nn.Linear(32, n_classes) ) def forward(self, x, n_samples=1): feat = self.backbone(x) if n_samples > 1: logits_list = [] for _ in range(n_samples): logits_list.append(self.classifier(feat)) logits_stack = torch.stack(logits_list) mean_logits = logits_stack.mean(dim=0) probs = torch.softmax(mean_logits, dim=-1) confidence = probs.max(dim=-1)[0] uncertainty = 1 - confidence else: mean_logits = self.classifier(feat) probs = torch.softmax(mean_logits, dim=-1) confidence = probs.max(dim=-1)[0] uncertainty = 1 - confidence return mean_logits, confidence, uncertainty
class AdaptiveFatigueSystem: """ 完整自适应疲劳检测系统 架构: 1. 自适应窗口管理器 2. 疲劳检测模型 3. 置信度反馈循环 """ def __init__(self): self.window_manager = ConfidenceAdaptiveWindow() self.detector = FatigueDetector(n_classes=3) self.fatigue_history = [] def process_frame(self, frame: torch.Tensor) -> dict: """ 处理一帧:自适应窗口→检测→反馈 Args: frame: [B, 3, H, W] 单帧或序列 """ window_size = self.window_manager.current_window with torch.no_grad(): logits, confidence, uncertainty = \ self.detector(frame, n_samples=10) new_window = self.window_manager.update(confidence.item()) probs = torch.softmax(logits, dim=-1) pred = probs.argmax(dim=-1) if confidence > 0.85: alert_level = 'high_confidence' elif confidence > 0.50: alert_level = 'medium_confidence' else: alert_level = 'low_confidence_degraded' return { 'prediction': pred.item(), 'confidence': confidence.item(), 'uncertainty': uncertainty.item(), 'window_size': window_size, 'new_window': new_window, 'alert_level': alert_level, }
if __name__ == "__main__": system = AdaptiveFatigueSystem() print("=== 自适应窗口疲劳检测 ===") print(f"初始窗口: {system.window_manager.current_window:.0f}s") print("\n--- 清醒阶段 ---") for i in range(10): frame = torch.randn(1, 3, 96, 96) result = system.process_frame(frame) if i % 3 == 0: print(f" 帧{i}: pred={result['prediction']}, " f"conf={result['confidence']:.2f}, " f"window={result['new_window']:.0f}s, " f"level={result['alert_level']}") print("\n--- 疲劳阶段 ---") for i in range(10): frame = torch.randn(1, 3, 96, 96) * 0.5 result = system.process_frame(frame) if i % 2 == 0: print(f" 帧{i}: pred={result['prediction']}, " f"conf={result['confidence']:.2f}, " f"window={result['new_window']:.0f}s, " f"level={result['alert_level']}")
|