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
| class CabinDataSynthesisPipeline:
"""
基于NVIDIA Omniverse的座舱数据合成完整管道
"""
def __init__(self, config: CabinSynthesisConfig):
self.config = config
self.stats = {"generated": 0, "by_state": {}}
def generate_batch(self, n_samples: int = 1000) -> dict:
"""
生成一批合成数据
Returns:
dataset_info: 包含图像路径、标注、元数据
"""
results = {
"images": [],
"annotations": [],
"metadata": []
}
for i in range(n_samples):
lighting = np.random.choice(self.config.lighting_conditions)
person_id = np.random.randint(0, self.config.metahuman_count)
accessory = np.random.choice(self.config.diversity["accessories"])
state = np.random.choice(
list(self.config.state_distribution.keys()),
p=list(self.config.state_distribution.values())
)
position = np.random.choice(["driver", "front_passenger", "rear_left", "rear_right"])
sensor = np.random.choice(self.config.sensors)
render_params = {
"lighting": lighting,
"person_id": person_id,
"accessory": accessory,
"state": state.value,
"position": position,
"sensor": sensor,
"resolution": (1920, 1080),
"frame_rate": 30,
"samples_per_pixel": 64,
"motion_blur": state != DriverState.NORMAL,
}
sample = {
"id": f"synth_{i:06d}",
"image_path": f"output/{render_params['sensor']}/synth_{i:06d}.png",
"annotations": {
"face_landmarks": f"output/annotations/synth_{i:06d}_face.json",
"gaze_vector": np.random.randn(3).tolist(),
"head_pose": np.random.randn(4).tolist(),
"eye_openness": self._simulate_eye_openness(state),
"behavior_label": state.value,
"occupant_bbox": [100, 50, 400, 500],
},
"metadata": render_params
}
results["images"].append(sample["image_path"])
results["annotations"].append(sample["annotations"])
results["metadata"].append(sample["metadata"])
self.stats["generated"] += 1
self.stats["by_state"][state.value] = \
self.stats["by_state"].get(state.value, 0) + 1
return results
def _simulate_eye_openness(self, state: DriverState) -> float:
"""根据状态模拟眼睑开度"""
if state == DriverState.NORMAL:
return np.random.normal(0.85, 0.05)
elif state == DriverState.FATIGUE_MILD:
return np.random.normal(0.65, 0.08)
elif state == DriverState.FATIGUE_SEVERE:
return np.random.normal(0.40, 0.10)
elif state == DriverState.DROWSY:
return np.random.normal(0.25, 0.12)
else:
return np.random.normal(0.80, 0.08)
pipeline = CabinDataSynthesisPipeline(config)
dataset = pipeline.generate_batch(n_samples=10000)
print(f"生成样本数: {pipeline.stats['generated']:,}")
print(f"\n按状态分布:")
for state, count in sorted(pipeline.stats["by_state"].items(),
key=lambda x: x[1], reverse=True):
pct = count / pipeline.stats["generated"] * 100
bar = '█' * int(pct/2)
print(f" {state:<22} {count:>5} ({pct:.1f}%) {bar}")
|
