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
| import numpy as np
from typing import Dict, Optional
import hashlib
class PIPLCompliantDMS:
"""符合 PIPL 的 DMS 系统"""
def __init__(self):
self.face_detector = FaceDetector()
self.gaze_estimator = GazeEstimator()
self.fatigue_detector = FatigueDetector()
self.local_processing_only = True
def process_frame(self, frame: np.ndarray) -> Dict:
"""处理帧（PIPL 合规）
Returns:
仅返回匿名化的结果，不返回原始图像或生物特征
"""
face_result = self.face_detector.detect(frame)
if not face_result['detected']:
return {
'face_detected': False,
'status': 'UNKNOWN',
'alert': None
}
gaze = self.gaze_estimator.estimate(face_result['roi'])
fatigue_score = self.fatigue_detector.detect(face_result['roi'])
return {
'face_detected': True,
'gaze_direction': gaze,
'fatigue_score': fatigue_score,
'status': self._determine_status(gaze, fatigue_score),
'alert': self._generate_alert(gaze, fatigue_score),
}
def _determine_status(self, gaze, fatigue) -> str:
"""确定状态"""
if fatigue > 0.7:
return 'FATIGUED'
elif abs(gaze[1]) > 15:
return 'DISTRACTED'
else:
return 'NORMAL'
def _generate_alert(self, gaze, fatigue) -> Optional[str]:
"""生成警告"""
if fatigue > 0.7:
return "疲劳警告：建议休息"
elif abs(gaze[1]) > 15:
return "分心警告：请注视前方"
return None
def anonymize_for_upload(self, result: Dict) -> Dict:
"""进一步匿名化用于上传"""
return {
'status': result['status'],
'timestamp': result.get('timestamp'),
'session_hash': hashlib.sha256(
str(result.get('session_id', '')).encode()
).hexdigest()[:16]
}
|