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
| class CosmosCabinDataPipeline: """ 基于 Cosmos 3 的座舱数据合成管道 目标: 生成多样化座舱监控训练数据 挑战: 仿真数据在真实摄像头上的性能差距 """ def __init__(self): self.cosmos_model = "Cosmos-3-Edge-4B" self.omniverse_scene = "cabin_interior.usd" def generate_cabin_scenarios(self, scenario_config: dict) -> list: """ 生成座舱监控场景 场景类型: 1. 疲劳驾驶(PERCLOS变化) 2. 分心行为(手机/交谈) 3. 儿童遗留(CPD) 4. OOP姿态(半躺/侧身) 5. 多乘员场景 """ scenarios = [] for scenario_type in scenario_config['types']: prompt = self._build_prompt(scenario_type, scenario_config) frames = self._generate_with_cosmos(prompt, scenario_config) sensor_data = self._render_sensors(frames, scenario_config) annotations = self._auto_annotate(sensor_data) scenarios.append({ 'type': scenario_type, 'frames': frames, 'sensor_data': sensor_data, 'annotations': annotations, }) return scenarios def _build_prompt(self, scenario_type: str, config: dict) -> str: """构建Cosmos场景生成提示""" prompts = { 'fatigue': "Driver gradually becoming drowsy, eyelids drooping, " "head nodding forward, in a car cabin at night with IR illumination", 'distraction_phone': "Driver looking down at phone, typing, " "eyes off road, daytime cabin lighting", 'cpd_child': "Empty car cabin, small child sleeping in rear seat " "covered with blanket, parked car, daylight", 'oop_recline': "Driver seat reclined 45 degrees, " "driver leaning back, seatbelt mispositioned", } return prompts.get(scenario_type, "") def bridge_sim2real(self, synthetic_data: list) -> list: """ Sim2Real 桥接策略 关键: 不能只靠合成数据,必须有真实数据补充 """ bridged = [] for sample in synthetic_data: sample = self._domain_randomization(sample) sample = self._style_transfer(sample) sample = self._inject_sensor_noise(sample) bridged.append(sample) return bridged def _domain_randomization(self, sample): """域随机化: 光照/纹理/相机角度变化""" return sample def _style_transfer(self, sample): """风格迁移: 仿真帧→真实摄像头外观""" return sample def _inject_sensor_noise(self, sample): """注入传感器噪声: 高斯/泊松/固定模式""" import numpy as np for frame in sample.get('frames', []): noise = np.random.normal(0, 5, frame.shape) frame['data'] = np.clip(frame['data'] + noise, 0, 255) return sample
|