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
| import cv2
import numpy as np
from pathlib import Path
class NTHUDDDLoader:
"""
NTHU-DDD数据集加载器
支持功能：
- 视频帧提取
- 人脸检测与对齐
- 数据增强
"""
def __init__(self,
data_root: str,
target_size: tuple = (224, 224)):
self.data_root = Path(data_root)
self.target_size = target_size
self.face_detector = cv2.CascadeClassifier(
cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
)
def load_video(self,
video_path: str,
max_frames: int = 1000) -> np.ndarray:
"""
加载视频
Args:
video_path: 视频文件路径
max_frames: 最大帧数
Returns:
frames: 视频帧数组, shape=(N, H, W, C)
"""
cap = cv2.VideoCapture(video_path)
frames = []
while len(frames) < max_frames:
ret, frame = cap.read()
if not ret:
break
frames.append(frame)
cap.release()
return np.array(frames)
def detect_face(self,
frame: np.ndarray) -> np.ndarray:
"""
人脸检测与裁剪
Args:
frame: 输入图像
Returns:
face: 裁剪后的人脸图像
"""
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
faces = self.face_detector.detectMultiScale(
gray, scaleFactor=1.1, minNeighbors=5
)
if len(faces) == 0:
return cv2.resize(frame, self.target_size)
x, y, w, h = max(faces, key=lambda f: f[2] * f[3])
margin = 0.2
x = max(0, int(x - w * margin))
y = max(0, int(y - h * margin))
w = min(frame.shape[1] - x, int(w * (1 + 2 * margin)))
h = min(frame.shape[0] - y, int(h * (1 + 2 * margin)))
face = frame[y:y+h, x:x+w]
face = cv2.resize(face, self.target_size)
return face
def augment_frame(self,
frame: np.ndarray,
augment_type: str = "random") -> np.ndarray:
"""
数据增强
Args:
frame: 输入帧
augment_type: 增强类型
Returns:
augmented_frame: 增强后的帧
"""
if augment_type == "random":
augment_type = np.random.choice([
"brightness", "contrast", "blur", "noise", "flip"
])
if augment_type == "brightness":
factor = np.random.uniform(0.7, 1.3)
frame = cv2.convertScaleAbs(frame, alpha=factor, beta=0)
elif augment_type == "contrast":
factor = np.random.uniform(0.7, 1.3)
mean = frame.mean()
frame = cv2.convertScaleAbs(frame, alpha=factor, beta=mean * (1 - factor))
elif augment_type == "blur":
kernel_size = np.random.choice([3, 5, 7])
frame = cv2.GaussianBlur(frame, (kernel_size, kernel_size), 0)
elif augment_type == "noise":
noise = np.random.randn(*frame.shape) * 10
frame = np.clip(frame + noise, 0, 255).astype(np.uint8)
elif augment_type == "flip":
frame = cv2.flip(frame, 1)
return frame
def create_dataloader(self,
batch_size: int = 32,
shuffle: bool = True):
"""
创建PyTorch数据加载器
Args:
batch_size: 批次大小
shuffle: 是否打乱
Returns:
dataloader: PyTorch DataLoader
"""
import torch
from torch.utils.data import Dataset, DataLoader
class NTHUDDDataset(Dataset):
def __init__(self, loader):
self.loader = loader
def __len__(self):
return 100
def __getitem__(self, idx):
frame = np.random.randint(0, 255, (224, 224, 3), dtype=np.uint8)
label = np.random.randint(0, 5)
return torch.from_numpy(frame).permute(2, 0, 1).float(), label
dataset = NTHUDDDataset(self)
return DataLoader(
dataset,
batch_size=batch_size,
shuffle=shuffle
)
if __name__ == "__main__":
loader = NTHUDDDLoader(data_root="/path/to/nthu-ddd")
dataloader = loader.create_dataloader(batch_size=16)
print(f"数据加载器创建完成")
print(f"批次数: {len(dataloader)}")
for frames, labels in dataloader:
print(f"批次帧形状: {frames.shape}")
print(f"批次标签: {labels}")
break
|