diff --git a/src/enhance_pedestrian_safety/config_manager.py b/src/enhance_pedestrian_safety/config_manager.py index c608de6674..aa07724783 100644 --- a/src/enhance_pedestrian_safety/config_manager.py +++ b/src/enhance_pedestrian_safety/config_manager.py @@ -133,7 +133,13 @@ def optimize_for_quality(config: Dict[str, Any]) -> Dict[str, Any]: 'enabled': True, 'enable_random': True, 'quality_check': True, - 'methods': ['normalize', 'contrast', 'sharpness', 'noise'] + 'save_original': True, + 'save_enhanced': True, + 'calibration_generation': True, + 'enhanced_dir_name': 'enhanced', + 'methods': ['normalize', 'contrast', 'brightness'], + 'weather_effects': True, + 'augmentation_level': 'medium' }) return optimized @@ -173,6 +179,94 @@ def optimize_for_speed(config: Dict[str, Any]) -> Dict[str, Any]: return optimized + @staticmethod + def optimize_for_safety(config: Dict[str, Any]) -> Dict[str, Any]: + """优化配置以增强行人安全""" + optimized = copy.deepcopy(config) + + # 增加行人密度 + traffic = optimized['traffic'] + traffic.update({ + 'pedestrians': 12, # 增加行人数量 + 'pedestrian_types': [ + 'walker.pedestrian.0001', + 'walker.pedestrian.0002', + 'walker.pedestrian.0003', + 'walker.pedestrian.0004' + ] + }) + + # 优化传感器配置以更好地检测行人 + sensors = optimized['sensors'] + sensors.update({ + 'image_size': [1280, 720], + 'capture_interval': 1.5, # 更频繁地捕获 + 'vehicle_cameras': 4, + 'camera_config': { + 'fov': 100.0, # 更宽的视野 + 'post_processing': 'default', + 'exposure_mode': 'auto', + 'motion_blur': 0.0 + } + }) + + # 启用LiDAR以检测行人 + sensors['lidar_sensors'] = 1 + sensors['lidar_config'].update({ + 'channels': 64, # 更多通道以检测行人 + 'range': 120.0, + 'points_per_second': 100000, + 'max_points_per_frame': 80000, + 'downsample_ratio': 0.2 + }) + + # 启用V2X和协同感知 + v2x = optimized.setdefault('v2x', {}) + v2x.update({ + 'enabled': True, + 'communication_range': 300.0, + 'update_interval': 1.0 # 更频繁地更新 + }) + + coop = optimized.setdefault('cooperative', {}) + coop.update({ + 'num_coop_vehicles': 2, + 'enable_shared_perception': True, + 'enable_traffic_warnings': True, + 'enable_maneuver_coordination': False, + 'data_fusion_interval': 0.5, # 更频繁地融合 + 'max_shared_objects': 100, + 'object_matching_threshold': 3.0 # 更严格的对象匹配 + }) + + # 性能优化 + perf = optimized.setdefault('performance', {}) + perf.update({ + 'batch_size': 5, + 'enable_compression': True, + 'compression_level': 3, + 'enable_memory_cache': True, + 'max_cache_size': 40, + 'frame_rate_limit': 8.0 + }) + + # 输出配置 + output = optimized['output'] + output.update({ + 'save_raw': True, + 'save_stitched': True, + 'save_annotations': True, + 'save_lidar': True, + 'save_fusion': True, + 'save_cooperative': True, + 'save_enhanced': True, + 'validate_data': True, + 'run_analysis': True, + 'run_quality_check': True + }) + + return optimized + class ConfigManager: @@ -189,6 +283,10 @@ class ConfigManager: 'description': '快速采集配置 - 优先处理速度', 'optimization': 'speed' }, + 'pedestrian_safety': { + 'description': '行人安全配置 - 优化行人检测和安全评估', + 'optimization': 'safety' + }, 'v2x_focused': { 'description': 'V2X重点配置 - 优化协同数据采集', 'optimization': 'custom', @@ -438,6 +536,8 @@ def _apply_preset(config: Dict[str, Any], preset_name: str) -> Dict[str, Any]: config = ConfigOptimizer.optimize_for_quality(config) elif optimization == 'speed': config = ConfigOptimizer.optimize_for_speed(config) + elif optimization == 'safety': + config = ConfigOptimizer.optimize_for_safety(config) elif optimization == 'custom' and 'settings' in preset: config = ConfigManager._deep_update(config, preset['settings']) @@ -536,6 +636,9 @@ def merge_args(config: Dict[str, Any], args: argparse.Namespace) -> Dict[str, An if hasattr(args, 'output_format') and args.output_format: config['output']['output_format'] = args.output_format + if hasattr(args, 'enable_safety_monitor'): + config['monitoring']['enable_safety_monitor'] = args.enable_safety_monitor + return config @staticmethod diff --git a/src/enhance_pedestrian_safety/data_analyzer.py b/src/enhance_pedestrian_safety/data_analyzer.py index 14690041fb..38ee4dd6f4 100644 --- a/src/enhance_pedestrian_safety/data_analyzer.py +++ b/src/enhance_pedestrian_safety/data_analyzer.py @@ -85,7 +85,8 @@ def analyze_dataset(data_dir, force_refresh=False): 'object_statistics': DataAnalyzer._analyze_objects(data_dir), 'temporal_analysis': DataAnalyzer._analyze_temporal(data_dir), 'cooperative_data': DataAnalyzer._analyze_cooperative_data(data_dir), - 'quality_metrics': DataAnalyzer._calculate_quality_metrics(data_dir) + 'quality_metrics': DataAnalyzer._calculate_quality_metrics(data_dir), + 'safety_analysis': DataAnalyzer._analyze_safety_data(data_dir) } # 生成评分 @@ -231,6 +232,10 @@ def _analyze_file_distribution(data_dir): fusion_dir = os.path.join(data_dir, "fusion") if os.path.exists(fusion_dir): distribution['fusion'] = DataAnalyzer._analyze_fusion_data(fusion_dir) + elif dir_name == "safety_reports": + safety_dir = os.path.join(data_dir, "safety_reports") + if os.path.exists(safety_dir): + distribution['safety_reports'] = DataAnalyzer._analyze_safety_reports(safety_dir) return distribution @@ -335,6 +340,41 @@ def _analyze_fusion_data(fusion_dir): fusion_stats['total_size_mb'] = round(total_size / (1024 * 1024), 2) return fusion_stats + @staticmethod + def _analyze_safety_reports(safety_dir): + """分析安全报告数据""" + safety_stats = { + 'reports': 0, + 'high_risk': 0, + 'medium_risk': 0, + 'low_risk': 0, + 'total_interactions': 0 + } + + json_files = [f for f in os.listdir(safety_dir) if f.lower().endswith('.json')] + safety_stats['reports'] = len(json_files) + + if json_files: + # 采样分析几个文件 + sample_files = json_files[:min(5, len(json_files))] + for json_file in sample_files: + try: + with open(os.path.join(safety_dir, json_file), 'r', encoding='utf-8') as f: + data = json.load(f) + + if 'high_risk_cases' in data: + safety_stats['high_risk'] += data['high_risk_cases'] + if 'medium_risk_cases' in data: + safety_stats['medium_risk'] += data['medium_risk_cases'] + if 'low_risk_cases' in data: + safety_stats['low_risk_cases'] += data['low_risk_cases'] + if 'total_interactions' in data: + safety_stats['total_interactions'] += data['total_interactions'] + except: + pass + + return safety_stats + @staticmethod def _analyze_objects(data_dir): """分析物体统计（优化版）""" @@ -647,6 +687,70 @@ def _analyze_cooperative_data(data_dir): return analysis + @staticmethod + def _analyze_safety_data(data_dir): + """分析安全数据""" + safety_dir = os.path.join(data_dir, "safety_reports") + + if not os.path.exists(safety_dir): + return { + 'total_reports': 0, + 'risk_levels': {'high': 0, 'medium': 0, 'low': 0}, + 'safety_score': 0, + 'pedestrian_interactions': 0, + 'average_distance': 0 + } + + json_files = [f for f in os.listdir(safety_dir) if f.lower().endswith('.json')] + + safety_data = { + 'total_reports': len(json_files), + 'risk_levels': {'high': 0, 'medium': 0, 'low': 0}, + 'safety_score': 0, + 'pedestrian_interactions': 0, + 'average_distance': 0, + 'near_misses': 0, + 'safety_warnings': 0 + } + + if json_files: + distances = [] + for json_file in json_files[:min(10, len(json_files))]: + try: + with open(os.path.join(safety_dir, json_file), 'r', encoding='utf-8') as f: + data = json.load(f) + + if 'high_risk_cases' in data: + safety_data['risk_levels']['high'] += data['high_risk_cases'] + if 'medium_risk_cases' in data: + safety_data['risk_levels']['medium'] += data['medium_risk_cases'] + if 'low_risk_cases' in data: + safety_data['risk_levels']['low'] += data['low_risk_cases'] + if 'total_interactions' in data: + safety_data['pedestrian_interactions'] += data['total_interactions'] + if 'average_distance' in data: + distances.append(data['average_distance']) + if 'near_misses' in data: + safety_data['near_misses'] += data['near_misses'] + if 'safety_warnings' in data: + safety_data['safety_warnings'] += data['safety_warnings'] + + except Exception as e: + print(f"分析安全报告 {json_file} 失败: {e}") + + if distances: + safety_data['average_distance'] = round(np.mean(distances), 2) + + # 计算安全评分 + total_risks = sum(safety_data['risk_levels'].values()) + if total_risks > 0: + high_risk_ratio = safety_data['risk_levels']['high'] / total_risks + safety_data['safety_score'] = max(0, 100 - high_risk_ratio * 100) + else: + safety_data['safety_score'] = 100 + + return safety_data + @staticmethod def _calculate_quality_metrics(data_dir): """计算质量指标（增强版）""" @@ -657,6 +761,7 @@ def _calculate_quality_metrics(data_dir): 'cooperative_score': 0, 'temporal_score': 0, 'structural_score': 0, + 'safety_score': 0, 'issues_found': [], 'recommendations': [] } @@ -674,7 +779,8 @@ def _calculate_quality_metrics(data_dir): "lidar", "fusion", "annotations", - "calibration" + "calibration", + "safety_reports" ] missing_required = [] @@ -781,9 +887,17 @@ def _calculate_quality_metrics(data_dir): quality_metrics['temporal_score'] *= 0.8 # 时长不足，降低分数 quality_metrics['recommendations'].append("建议增加数据收集时长以获得更完整的时间序列") + # 7. 安全评分 + safety_data = DataAnalyzer._analyze_safety_data(data_dir) + quality_metrics['safety_score'] = safety_data.get('safety_score', 0) + + if quality_metrics['safety_score'] < 80: + quality_metrics['issues_found'].append(f"安全评分较低: {quality_metrics['safety_score']}") + quality_metrics['recommendations'].append("建议增加行人安全相关的场景和数据收集") + # 限制分数在0-100之间 for key in ['completeness_score', 'consistency_score', 'diversity_score', - 'cooperative_score', 'temporal_score', 'structural_score']: + 'cooperative_score', 'temporal_score', 'structural_score', 'safety_score']: quality_metrics[key] = max(0, min(100, quality_metrics[key])) return quality_metrics @@ -792,12 +906,13 @@ def _calculate_quality_metrics(data_dir): def _calculate_overall_score(analysis): """计算总体评分（增强版）""" weights = { - 'completeness': 0.20, # 完整性 - 'consistency': 0.15, # 一致性 - 'demporal': 0.15, # 时间性 - 'structural': 0.10, # 结构性 - 'diversity': 0.15, # 多样性 - 'cooperative': 0.15, # 协同性 + 'completeness': 0.15, # 完整性 + 'consistency': 0.12, # 一致性 + 'temporal': 0.12, # 时间性 + 'structural': 0.08, # 结构性 + 'diversity': 0.12, # 多样性 + 'cooperative': 0.12, # 协同性 + 'safety': 0.19, # 安全性 'quality_bonus': 0.10 # 质量加成 } @@ -810,7 +925,8 @@ def _calculate_overall_score(analysis): quality['temporal_score'] * weights['temporal'] + quality['structural_score'] * weights['structural'] + quality['diversity_score'] * weights['diversity'] + - quality['cooperative_score'] * weights['cooperative'] + quality['cooperative_score'] * weights['cooperative'] + + quality['safety_score'] * weights['safety'] ) # 质量加成（基于问题数量） @@ -822,7 +938,8 @@ def _calculate_overall_score(analysis): # 额外加成（如果数据集特别优秀） if (quality['completeness_score'] >= 95 and quality['consistency_score'] >= 90 and - quality['diversity_score'] >= 85): + quality['diversity_score'] >= 85 and + quality['safety_score'] >= 90): total_score += 5 return round(min(total_score, 100), 1) @@ -852,6 +969,7 @@ def _save_analysis_report(data_dir, analysis): 'total_objects': analysis['object_statistics']['total_objects'], 'num_classes': len(analysis['object_statistics']['by_class']) }, + 'safety_data': analysis.get('safety_analysis', {}), 'analysis_metadata': analysis.get('metadata', {}) } @@ -919,6 +1037,20 @@ def _print_analysis_summary(analysis): print(f" 每帧物体数统计:") print(f" 最小: {stats['min']}, 最大: {stats['max']}, 平均: {stats['mean']}, 中位数: {stats['median']}") + # 安全数据分析 + if 'safety_analysis' in analysis: + safety = analysis['safety_analysis'] + print(f"\n🚸 安全数据分析:") + print(f" 安全评分: {safety.get('safety_score', 0)}/100") + print(f" 风险等级分布:") + print(f" 高风险: {safety.get('risk_levels', {}).get('high', 0)}") + print(f" 中风险: {safety.get('risk_levels', {}).get('medium', 0)}") + print(f" 低风险: {safety.get('risk_levels', {}).get('low', 0)}") + print(f" 行人交互次数: {safety.get('pedestrian_interactions', 0)}") + print(f" 平均距离: {safety.get('average_distance', 0):.2f}米") + print(f" 近距离事件: {safety.get('near_misses', 0)}") + print(f" 安全警告: {safety.get('safety_warnings', 0)}") + # 协同数据分析 cooperative = analysis['cooperative_data'] print(f"\n🤝 协同数据分析:") @@ -958,7 +1090,8 @@ def _print_analysis_summary(analysis): ('结构性', quality['structural_score']), ('时间性', quality['temporal_score']), ('多样性', quality['diversity_score']), - ('协同性', quality['cooperative_score']) + ('协同性', quality['cooperative_score']), + ('安全性', quality['safety_score']) ] for name, score in metrics: @@ -1015,7 +1148,8 @@ def generate_comparison_report(data_dirs, output_file=None): 'object_statistics': { 'total_objects': analysis['object_statistics']['total_objects'], 'num_classes': len(analysis['object_statistics']['by_class']) - } + }, + 'safety_analysis': analysis.get('safety_analysis', {}) } if output_file: diff --git a/src/enhance_pedestrian_safety/data_validator.py b/src/enhance_pedestrian_safety/data_validator.py index fa67f6425b..b472b40eb8 100644 --- a/src/enhance_pedestrian_safety/data_validator.py +++ b/src/enhance_pedestrian_safety/data_validator.py @@ -1,5 +1,5 @@ -import os import json +import os class DataValidator: @@ -16,7 +16,8 @@ def validate_dataset(data_dir): 'metadata': DataValidator._validate_metadata(data_dir), 'lidar_data': DataValidator._validate_lidar_data(data_dir), 'cooperative_data': DataValidator._validate_cooperative_data(data_dir), - 'fusion_data': DataValidator._validate_fusion_data(data_dir) + 'fusion_data': DataValidator._validate_fusion_data(data_dir), + 'safety_data': DataValidator._validate_safety_data(data_dir) } validation_results['overall_score'] = DataValidator._calculate_score(validation_results) @@ -39,18 +40,33 @@ def _check_directory_structure(data_dir): "fusion" ] + optional_dirs = [ + "lidar", + "calibration", + "annotations", + "safety_reports" + ] + missing_dirs = [] for dir_path in required_dirs: full_path = os.path.join(data_dir, dir_path) if not os.path.exists(full_path): missing_dirs.append(dir_path) + missing_optional = [] + for dir_path in optional_dirs: + full_path = os.path.join(data_dir, dir_path) + if not os.path.exists(full_path): + missing_optional.append(dir_path) + status = 'PASS' if len(missing_dirs) == 0 else 'FAIL' result = { 'status': status, 'missing_directories': missing_dirs, - 'required_directories': required_dirs + 'missing_optional_directories': missing_optional, + 'required_directories': required_dirs, + 'optional_directories': optional_dirs } return result @@ -376,17 +392,59 @@ def _validate_fusion_data(data_dir): 'errors': errors } + @staticmethod + def _validate_safety_data(data_dir): + """验证安全数据""" + safety_dir = os.path.join(data_dir, "safety_reports") + + if not os.path.exists(safety_dir): + return {'status': 'MISSING', 'count': 0, 'errors': []} + + json_files = [f for f in os.listdir(safety_dir) if f.endswith('.json')] + errors = [] + valid_files = 0 + + for json_file in json_files[:min(5, len(json_files))]: + json_path = os.path.join(safety_dir, json_file) + try: + with open(json_path, 'r') as f: + data = json.load(f) + + # 检查必要字段 + required_keys = ['timestamp', 'total_interactions'] + for key in required_keys: + if key not in data: + errors.append(f"安全报告缺失字段 {key}: {json_file}") + + valid_files += 1 + except Exception as e: + errors.append(f"安全报告无效: {json_file} - {str(e)}") + + if len(errors) == 0 and valid_files > 0: + status = 'PASS' + elif len(errors) < 3 and valid_files > 0: + status = 'WARNING' + else: + status = 'FAIL' + + return { + 'status': status, + 'count': len(json_files), + 'errors': errors + } + @staticmethod def _calculate_score(results): weights = { 'directory_structure': 0.10, - 'raw_images': 0.20, - 'stitched_images': 0.10, - 'annotations': 0.10, - 'metadata': 0.10, - 'lidar_data': 0.15, - 'cooperative_data': 0.15, - 'fusion_data': 0.10 + 'raw_images': 0.15, + 'stitched_images': 0.05, + 'annotations': 0.08, + 'metadata': 0.08, + 'lidar_data': 0.12, + 'cooperative_data': 0.12, + 'fusion_data': 0.10, + 'safety_data': 0.20 } score = 0 @@ -459,6 +517,10 @@ def _print_validation_report(results): if 'json_files' in result: print(f" JSON文件: {result['json_files']}") + if key == 'safety_data' and isinstance(result, dict): + if 'count' in result: + print(f" 安全报告: {result['count']} 个") + if 'errors' in result and result['errors']: print(f" 错误 ({len(result['errors'])}):") for error in result['errors'][:3]: diff --git a/src/enhance_pedestrian_safety/main.py b/src/enhance_pedestrian_safety/main.py index 6861758c1d..2626fe0a82 100644 --- a/src/enhance_pedestrian_safety/main.py +++ b/src/enhance_pedestrian_safety/main.py @@ -20,6 +20,7 @@ from data_analyzer import DataAnalyzer from lidar_processor import LidarProcessor, MultiSensorFusion from multi_vehicle_manager import MultiVehicleManager +from pedestrian_safety_monitor import PedestrianSafetyMonitor carla_egg_path, remaining_argv = setup_carla_path() carla = import_carla_module() @@ -143,6 +144,11 @@ def performance(msg): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[PERF][{timestamp}] {msg}") + @staticmethod + def safety(msg): + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"[SAFETY][{timestamp}] {msg}") + class WeatherSystem: WEATHER_PRESETS = { @@ -964,6 +970,7 @@ def __init__(self, config): self.sensor_managers = {} self.multi_vehicle_manager = None self.v2x_communication = None + self.safety_monitor = None self.start_time = None self.is_running = False @@ -994,7 +1001,8 @@ def setup_directories(self): "v2x_messages", "v2xformer_format", "kitti_format", - "metadata" + "metadata", + "safety_reports" ] for subdir in directories: @@ -1082,6 +1090,9 @@ def setup_scene(self): {'type': 'vehicle', 'capabilities': ['bsm', 'rsm']} ) + # 初始化行人安全监控器 + self.safety_monitor = PedestrianSafetyMonitor(self.world, self.output_dir) + time.sleep(3.0) return True @@ -1114,6 +1125,7 @@ def collect_data(self): last_performance_sample = time.time() last_detailed_log = time.time() last_memory_check = time.time() + last_safety_check = time.time() memory_warning_issued = False early_stop_triggered = False @@ -1167,6 +1179,13 @@ def collect_data(self): self._share_perception_data() last_perception_share = current_time + # 行人安全检查 + if current_time - last_safety_check >= 1.0 and self.safety_monitor: + safety_report = self.safety_monitor.check_pedestrian_safety() + if safety_report['risk_distribution']['high'] > 0: + Log.safety(f"行人安全警告: {safety_report['high_risk_cases']}个高风险情况") + last_safety_check = current_time + if current_time - last_performance_sample >= 10.0: memory_info = self.performance_monitor.sample_memory() cpu_info = self.performance_monitor.sample_cpu() @@ -1228,6 +1247,12 @@ def collect_data(self): Log.info(f"平均帧率: {fps:.2f} FPS") Log.info(f"最大内存使用: {performance_summary['max_memory_mb']:.1f} MB") Log.info(f"平均CPU使用: {performance_summary['average_cpu_percent']:.1f}%") + + # 生成行人安全报告 + if self.safety_monitor: + final_report = self.safety_monitor.generate_final_report() + Log.safety( + f"行人安全报告: {final_report['risk_distribution']['high']}高风险, {final_report['risk_distribution']['medium']}中风险") else: Log.warning("未收集到任何数据帧") @@ -1279,6 +1304,13 @@ def cleanup(self): except: pass + if self.safety_monitor: + try: + Log.info("保存行人安全数据...") + self.safety_monitor.save_data() + except: + pass + Log.info(f"清理 {len(self.sensor_managers)} 个传感器管理器...") for vehicle_id, sensor_manager in self.sensor_managers.items(): try: @@ -1328,6 +1360,7 @@ def cleanup(self): self.traffic_manager = None self.multi_vehicle_manager = None self.v2x_communication = None + self.safety_monitor = None self.scene_center = None gc.collect() @@ -1342,6 +1375,7 @@ def cleanup(self): self.traffic_manager = None self.multi_vehicle_manager = None self.v2x_communication = None + self.safety_monitor = None gc.collect() except: pass @@ -1527,6 +1561,9 @@ def _save_metadata(self): if self.multi_vehicle_manager: metadata['cooperative_summary'] = self.multi_vehicle_manager.generate_summary() + if self.safety_monitor: + metadata['safety_report'] = self.safety_monitor.generate_final_report() + meta_path = os.path.join(self.output_dir, "metadata", "collection_info.json") with open(meta_path, 'w', encoding='utf-8') as f: json.dump(metadata, f, indent=2, ensure_ascii=False) @@ -1565,6 +1602,11 @@ def _print_summary(self): [f for f in os.listdir(os.path.join(coop_dir, "shared_perception")) if f.endswith('.json')]) print(f"协同数据: {v2x_files} V2X消息, {perception_files} 共享感知文件") + safety_dir = os.path.join(self.output_dir, "safety_reports") + if os.path.exists(safety_dir): + safety_files = len([f for f in os.listdir(safety_dir) if f.endswith('.json')]) + print(f"安全报告: {safety_files} 个") + if self.output_format == 'v2xformer': v2x_dir = os.path.join(self.output_dir, "v2xformer_format") if os.path.exists(v2x_dir): @@ -1643,6 +1685,7 @@ def main(): parser.add_argument('--enable-cooperative', action='store_true', help='启用协同感知') parser.add_argument('--enable-enhancement', action='store_true', help='启用数据增强') parser.add_argument('--enable-annotations', action='store_true', help='启用自动标注') + parser.add_argument('--enable-safety-monitor', action='store_true', default=True, help='启用行人安全监控') parser.add_argument('--run-analysis', action='store_true', help='运行数据集分析') parser.add_argument('--skip-validation', action='store_true', help='跳过数据验证') @@ -1659,7 +1702,7 @@ def main(): config['output']['output_format'] = args.output_format print("\n" + "=" * 60) - print("CVIPS 性能优化数据收集系统") + print("CVIPS 性能优化数据收集系统 - 行人安全增强版") print("=" * 60) print(f"场景: {config['scenario']['name']}") @@ -1678,6 +1721,7 @@ def main(): print(f" V2X: {'启用' if config['v2x']['enabled'] else '禁用'}") print(f" 协同: {'启用' if config['output']['save_cooperative'] else '禁用'}") print(f" 增强: {'启用' if config['enhancement']['enabled'] else '禁用'}") + print(f" 安全监控: {'启用' if args.enable_safety_monitor else '禁用'}") print(f"性能:") print(f" 批处理大小: {config['performance']['batch_size']}") @@ -1717,4 +1761,4 @@ def main(): if __name__ == "__main__": - main() + main() \ No newline at end of file diff --git a/src/enhance_pedestrian_safety/multi_vehicle_manager.py b/src/enhance_pedestrian_safety/multi_vehicle_manager.py index 4ed8890d91..62fcd37cf8 100644 --- a/src/enhance_pedestrian_safety/multi_vehicle_manager.py +++ b/src/enhance_pedestrian_safety/multi_vehicle_manager.py @@ -297,6 +297,34 @@ def share_traffic_warning(self, vehicle_id: int, warning_type: str, print(f"共享交通警告失败: {e}") return None + def share_pedestrian_warning(self, vehicle_id: int, pedestrian_location: Tuple[float, float, float], + distance: float, speed: float): + """共享行人警告""" + try: + warning_data = { + 'warning_type': 'pedestrian', + 'pedestrian_location': pedestrian_location, + 'distance': distance, + 'vehicle_speed': speed, + 'timestamp': time.time(), + 'source_vehicle': vehicle_id, + 'severity': 'high' if distance < 10.0 else 'medium' if distance < 20.0 else 'low' + } + + message = self.create_v2x_message( + vehicle_id, + 'warning', + warning_data, + priority=4 # 行人警告优先级最高 + ) + + recipients = self.broadcast_message(message) + + return message, recipients + except Exception as e: + print(f"共享行人警告失败: {e}") + return None, [] + def _fuse_shared_perception(self, source_id: int, objects: List[Dict], recipients: List[int]): """融合共享的感知数据""" fused_objects = [] diff --git a/src/enhance_pedestrian_safety/pedestrian_safety_monitor.py b/src/enhance_pedestrian_safety/pedestrian_safety_monitor.py new file mode 100644 index 0000000000..5ec545f100 --- /dev/null +++ b/src/enhance_pedestrian_safety/pedestrian_safety_monitor.py @@ -0,0 +1,368 @@ +import json +import os +import time +import math +import numpy as np +from datetime import datetime +from typing import List, Dict, Tuple, Optional +import carla + + +class PedestrianSafetyMonitor: + """行人安全监控器""" + + def __init__(self, world, output_dir): + self.world = world + self.output_dir = output_dir + self.safety_dir = os.path.join(output_dir, "safety_reports") + os.makedirs(self.safety_dir, exist_ok=True) + + # 安全参数 + self.safety_thresholds = { + 'high_risk_distance': 5.0, # 高风险距离 (米) + 'medium_risk_distance': 10.0, # 中风险距离 (米) + 'low_risk_distance': 20.0, # 低风险距离 (米) + 'safe_speed_limit': 30.0, # 安全速度限制 (km/h) + 'reaction_time': 1.5, # 反应时间 (秒) + 'braking_deceleration': 6.0 # 制动减速度 (m/s²) + } + + # 统计数据 + self.stats = { + 'total_interactions': 0, + 'high_risk_cases': 0, + 'medium_risk_cases': 0, + 'low_risk_cases': 0, + 'safe_cases': 0, + 'near_misses': 0, + 'safety_warnings': 0, + 'average_distance': 0, + 'min_distance': float('inf'), + 'max_distance': 0, + 'interaction_times': [] + } + + # 详细记录 + self.interaction_records = [] + self.warning_logs = [] + + def check_pedestrian_safety(self) -> Dict: + """检查行人安全""" + vehicles = self._get_vehicles() + pedestrians = self._get_pedestrians() + + current_interactions = [] + + for vehicle in vehicles: + vehicle_location = vehicle.get_location() + vehicle_velocity = vehicle.get_velocity() + vehicle_speed = math.sqrt(vehicle_velocity.x ** 2 + vehicle_velocity.y ** 2 + vehicle_velocity.z ** 2) + + for pedestrian in pedestrians: + pedestrian_location = pedestrian.get_location() + + # 计算距离 + distance = vehicle_location.distance(pedestrian_location) + + # 计算相对速度 + pedestrian_velocity = pedestrian.get_velocity() + relative_speed = self._calculate_relative_speed(vehicle_velocity, pedestrian_velocity) + + # 计算碰撞时间 + time_to_collision = self._calculate_ttc(distance, relative_speed) + + # 评估风险 + risk_level = self._assess_risk(distance, vehicle_speed, time_to_collision) + + interaction = { + 'timestamp': time.time(), + 'vehicle_id': vehicle.id, + 'pedestrian_id': pedestrian.id, + 'distance': distance, + 'vehicle_speed': vehicle_speed * 3.6, # 转换为km/h + 'relative_speed': relative_speed * 3.6, + 'time_to_collision': time_to_collision if time_to_collision < 100 else None, + 'risk_level': risk_level, + 'vehicle_location': { + 'x': vehicle_location.x, + 'y': vehicle_location.y, + 'z': vehicle_location.z + }, + 'pedestrian_location': { + 'x': pedestrian_location.x, + 'y': pedestrian_location.y, + 'z': pedestrian_location.z + } + } + + current_interactions.append(interaction) + + # 更新统计 + self._update_stats(interaction) + + # 记录高风险情况 + if risk_level == 'high': + self._log_high_risk(interaction) + + # 保存当前检查结果 + if current_interactions: + self._save_interaction_report(current_interactions) + + return self._generate_safety_report() + + def _get_vehicles(self) -> List[carla.Actor]: + """获取所有车辆""" + return [actor for actor in self.world.get_actors() if 'vehicle' in actor.type_id] + + def _get_pedestrians(self) -> List[carla.Actor]: + """获取所有行人""" + return [actor for actor in self.world.get_actors() if 'walker' in actor.type_id] + + def _calculate_relative_speed(self, v1: carla.Vector3D, v2: carla.Vector3D) -> float: + """计算相对速度""" + return math.sqrt((v1.x - v2.x) ** 2 + (v1.y - v2.y) ** 2 + (v1.z - v2.z) ** 2) + + def _calculate_ttc(self, distance: float, relative_speed: float) -> float: + """计算碰撞时间 (Time to Collision)""" + if relative_speed > 0.1: # 避免除以零 + return distance / relative_speed + return float('inf') + + def _assess_risk(self, distance: float, speed: float, ttc: Optional[float]) -> str: + """评估风险等级""" + speed_kmh = speed * 3.6 + + # 基于距离的风险评估 + if distance < self.safety_thresholds['high_risk_distance']: + if ttc is not None and ttc < 2.0: + return 'high' + else: + return 'medium' + elif distance < self.safety_thresholds['medium_risk_distance']: + if speed_kmh > self.safety_thresholds['safe_speed_limit']: + return 'medium' + else: + return 'low' + elif distance < self.safety_thresholds['low_risk_distance']: + return 'low' + else: + return 'safe' + + def _update_stats(self, interaction: Dict): + """更新统计数据""" + self.stats['total_interactions'] += 1 + distance = interaction['distance'] + + # 更新距离统计 + self.stats['average_distance'] = ( + (self.stats['average_distance'] * (self.stats['total_interactions'] - 1) + distance) / + self.stats['total_interactions'] + ) + self.stats['min_distance'] = min(self.stats['min_distance'], distance) + self.stats['max_distance'] = max(self.stats['max_distance'], distance) + + # 更新风险统计 + risk_level = interaction['risk_level'] + if risk_level == 'high': + self.stats['high_risk_cases'] += 1 + self.stats['near_misses'] += 1 + self.stats['safety_warnings'] += 1 + elif risk_level == 'medium': + self.stats['medium_risk_cases'] += 1 + self.stats['safety_warnings'] += 1 + elif risk_level == 'low': + self.stats['low_risk_cases'] += 1 + else: + self.stats['safe_cases'] += 1 + + # 记录交互时间 + self.stats['interaction_times'].append(interaction['timestamp']) + + # 添加到详细记录 + self.interaction_records.append(interaction) + + # 限制记录数量 + if len(self.interaction_records) > 1000: + self.interaction_records = self.interaction_records[-1000:] + + def _log_high_risk(self, interaction: Dict): + """记录高风险情况""" + warning = { + 'timestamp': datetime.now().isoformat(), + 'interaction': interaction, + 'safety_measures': self._suggest_safety_measures(interaction) + } + self.warning_logs.append(warning) + + # 保存高风险警告 + if len(self.warning_logs) % 10 == 0: + self._save_warning_logs() + + def _suggest_safety_measures(self, interaction: Dict) -> List[str]: + """建议安全措施""" + measures = [] + + if interaction['risk_level'] == 'high': + measures.extend([ + "立即制动", + "鸣喇叭警告", + "准备紧急避让", + "向其他车辆发送警告" + ]) + elif interaction['risk_level'] == 'medium': + measures.extend([ + "减速行驶", + "保持警惕", + "准备制动", + "观察行人动向" + ]) + elif interaction['risk_level'] == 'low': + measures.extend([ + "保持安全距离", + "观察周围环境", + "准备应对突发情况" + ]) + + return measures + + def _save_interaction_report(self, interactions: List[Dict]): + """保存交互报告""" + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + report_file = os.path.join(self.safety_dir, f"interactions_{timestamp}.json") + + report = { + 'timestamp': datetime.now().isoformat(), + 'total_interactions': len(interactions), + 'interactions': interactions, + 'summary': { + 'high_risk': len([i for i in interactions if i['risk_level'] == 'high']), + 'medium_risk': len([i for i in interactions if i['risk_level'] == 'medium']), + 'low_risk': len([i for i in interactions if i['risk_level'] == 'low']), + 'safe': len([i for i in interactions if i['risk_level'] == 'safe']) + } + } + + with open(report_file, 'w', encoding='utf-8') as f: + json.dump(report, f, indent=2, ensure_ascii=False) + + def _save_warning_logs(self): + """保存警告日志""" + if not self.warning_logs: + return + + warning_file = os.path.join(self.safety_dir, "warning_logs.json") + with open(warning_file, 'w', encoding='utf-8') as f: + json.dump(self.warning_logs, f, indent=2, ensure_ascii=False) + + def _generate_safety_report(self) -> Dict: + """生成安全报告""" + report = { + 'timestamp': datetime.now().isoformat(), + 'statistics': self.stats.copy(), + 'safety_thresholds': self.safety_thresholds, + 'risk_distribution': { + 'high': self.stats['high_risk_cases'], + 'medium': self.stats['medium_risk_cases'], + 'low': self.stats['low_risk_cases'], + 'safe': self.stats['safe_cases'] + }, + 'safety_score': self._calculate_safety_score(), + 'recommendations': self._generate_recommendations() + } + + return report + + def _calculate_safety_score(self) -> float: + """计算安全评分""" + if self.stats['total_interactions'] == 0: + return 100.0 + + high_risk_ratio = self.stats['high_risk_cases'] / self.stats['total_interactions'] + medium_risk_ratio = self.stats['medium_risk_cases'] / self.stats['total_interactions'] + + # 评分公式：基础分减去风险比例 + score = 100 - (high_risk_ratio * 60 + medium_risk_ratio * 30) * 100 + + # 考虑平均距离 + if self.stats['average_distance'] > 15.0: + score += 10 + elif self.stats['average_distance'] < 5.0: + score -= 20 + + return max(0, min(100, score)) + + def _generate_recommendations(self) -> List[str]: + """生成改进建议""" + recommendations = [] + + if self.stats['high_risk_cases'] > 0: + recommendations.extend([ + "增加行人安全距离阈值", + "加强车辆行人检测系统", + "实施更严格的限速措施", + "增加行人警告系统" + ]) + + if self.stats['average_distance'] < 10.0: + recommendations.append("增加车辆与行人的平均距离") + + if self.stats['near_misses'] > 5: + recommendations.append("实施紧急制动系统") + + return recommendations + + def generate_final_report(self) -> Dict: + """生成最终报告""" + final_report = self._generate_safety_report() + + # 添加历史数据 + final_report['historical_data'] = { + 'total_interaction_records': len(self.interaction_records), + 'total_warning_logs': len(self.warning_logs), + 'analysis_period': self._get_analysis_period() + } + + # 保存最终报告 + final_file = os.path.join(self.safety_dir, "final_safety_report.json") + with open(final_file, 'w', encoding='utf-8') as f: + json.dump(final_report, f, indent=2, ensure_ascii=False) + + return final_report + + def _get_analysis_period(self) -> Dict: + """获取分析时间段""" + if not self.stats['interaction_times']: + return {'start': None, 'end': None, 'duration': 0} + + start_time = min(self.stats['interaction_times']) + end_time = max(self.stats['interaction_times']) + duration = end_time - start_time + + return { + 'start': datetime.fromtimestamp(start_time).isoformat(), + 'end': datetime.fromtimestamp(end_time).isoformat(), + 'duration_seconds': duration, + 'duration_minutes': duration / 60, + 'duration_hours': duration / 3600 + } + + def save_data(self): + """保存所有数据""" + # 保存统计数据 + stats_file = os.path.join(self.safety_dir, "safety_statistics.json") + with open(stats_file, 'w', encoding='utf-8') as f: + json.dump(self.stats, f, indent=2, ensure_ascii=False) + + # 保存详细记录 + if self.interaction_records: + records_file = os.path.join(self.safety_dir, "interaction_records.json") + with open(records_file, 'w', encoding='utf-8') as f: + json.dump(self.interaction_records, f, indent=2, ensure_ascii=False) + + # 保存警告日志 + self._save_warning_logs() + + # 生成并保存最终报告 + self.generate_final_report() + + print(f"行人安全数据已保存到: {self.safety_dir}") \ No newline at end of file diff --git a/src/enhance_pedestrian_safety/scene_manager.py b/src/enhance_pedestrian_safety/scene_manager.py index dbe424d7d4..f7e081f9ed 100644 --- a/src/enhance_pedestrian_safety/scene_manager.py +++ b/src/enhance_pedestrian_safety/scene_manager.py @@ -45,6 +45,23 @@ class SceneManager: 'vehicle_density': 0.6, 'pedestrian_density': 0.5, 'weather_variations': ['rainy'] + }, + 'school_zone': { + 'description': '学校区域场景 - 行人安全重点', + 'pedestrian_behavior': ['crossing', 'walking', 'playing', 'running'], + 'vehicle_density': 0.4, + 'pedestrian_density': 0.9, + 'weather_variations': ['clear', 'cloudy'], + 'speed_limit': 20.0, + 'safety_zone_radius': 30.0 + }, + 'pedestrian_crossing': { + 'description': '人行横道场景', + 'pedestrian_behavior': ['crossing', 'waiting', 'walking'], + 'vehicle_density': 0.5, + 'pedestrian_density': 0.8, + 'weather_variations': ['clear', 'rainy'], + 'crossing_intensity': 'high' } } @@ -74,6 +91,10 @@ def setup_scene(world, config, scene_type='intersection_4way'): # 设置行人行为 config['traffic']['pedestrian_behaviors'] = scene_config['pedestrian_behavior'] + # 如果是学校区域，设置车速限制 + if scene_type == 'school_zone' and 'speed_limit' in scene_config: + config['traffic']['speed_limit'] = scene_config['speed_limit'] + # 应用场景特定设置 SceneManager._apply_scene_specifics(world, scene_type) @@ -102,9 +123,73 @@ def _apply_scene_specifics(world, scene_type): except: pass + elif scene_type == 'school_zone': + # 学校区域：设置车辆限速和行人保护区域 + for actor in world.get_actors(): + if 'vehicle' in actor.type_id: + try: + # 限制车速 + actor.enable_constant_velocity(carla.Vector3D(15, 0, 0)) + # 开启行人检测警告 + actor.set_light_state(carla.VehicleLightState.LowBeam) + except: + pass + + elif scene_type == 'pedestrian_crossing': + # 人行横道：增加可见性 + for actor in world.get_actors(): + if 'vehicle' in actor.type_id: + try: + actor.set_light_state(carla.VehicleLightState.LowBeam) + except: + pass + except Exception as e: print(f"场景特定设置失败: {e}") + @staticmethod + def spawn_pedestrian_safety_features(world, location, feature_type='crosswalk'): + """生成行人安全设施""" + blueprint_lib = world.get_blueprint_library() + features = [] + + try: + if feature_type == 'crosswalk': + # 生成人行横道标记 + for i in range(-4, 5): + line_location = carla.Location( + x=location.x + i * 0.8, + y=location.y, + z=location.z + 0.02 + ) + rotation = carla.Rotation(0, 0, 0) + + line_bp = blueprint_lib.find('static.prop.linepainting') + if line_bp: + line = world.spawn_actor(line_bp, carla.Transform(line_location, rotation)) + features.append(line) + + elif feature_type == 'speed_bump': + # 生成减速带 + bump_location = carla.Location(location.x, location.y, location.z + 0.1) + bump_bp = blueprint_lib.find('static.prop.speedbump') + if bump_bp: + bump = world.spawn_actor(bump_bp, carla.Transform(bump_location, carla.Rotation(0, 0, 0))) + features.append(bump) + + elif feature_type == 'warning_sign': + # 生成警告标志 + sign_location = carla.Location(location.x, location.y, location.z + 2.0) + sign_bp = blueprint_lib.find('static.prop.trafficsign') + if sign_bp: + sign = world.spawn_actor(sign_bp, carla.Transform(sign_location, carla.Rotation(0, 90, 0))) + features.append(sign) + + except Exception as e: + print(f"生成行人安全设施失败: {e}") + + return features + @staticmethod def spawn_traffic_cones(world, center_location, num_cones=10): """生成交通锥桶""" diff --git a/src/enhance_pedestrian_safety/sensor_enhancer.py b/src/enhance_pedestrian_safety/sensor_enhancer.py index b446c71bbe..3dba1d767b 100644 --- a/src/enhance_pedestrian_safety/sensor_enhancer.py +++ b/src/enhance_pedestrian_safety/sensor_enhancer.py @@ -778,6 +778,60 @@ def generate_enhancement_report(self, output_dir: str) -> Dict: return report + def _add_pedestrian_detection_markers(self, image: np.ndarray, pedestrians: List[Dict]) -> np.ndarray: + """添加行人检测标记""" + if not pedestrians: + return image + + # 创建副本 + marked_image = image.copy() + + for pedestrian in pedestrians: + # 模拟行人检测框 + x = random.randint(100, image.shape[1] - 100) + y = random.randint(100, image.shape[0] - 100) + width = random.randint(30, 80) + height = random.randint(80, 180) + + # 绘制检测框 + color = (0, 255, 0) # 绿色表示安全 + thickness = 2 + + cv2.rectangle(marked_image, (x, y), (x + width, y + height), color, thickness) + + # 添加标签 + label = f"Pedestrian {pedestrian.get('distance', 0):.1f}m" + cv2.putText(marked_image, label, (x, y - 10), + cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, thickness) + + return marked_image + + def _simulate_safety_warnings(self, image: np.ndarray, warnings: List[Dict]) -> np.ndarray: + """模拟安全警告""" + if not warnings: + return image + + warning_image = image.copy() + + for warning in warnings[:2]: # 最多显示2个警告 + # 添加警告文本 + text = f"WARNING: {warning.get('type', 'Pedestrian')} {warning.get('distance', 0):.1f}m" + color = (0, 0, 255) # 红色表示警告 + thickness = 2 + + position = (50, 50 + warnings.index(warning) * 40) + cv2.putText(warning_image, text, position, + cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, thickness) + + # 添加警告图标 + icon_size = 30 + icon_position = (position[0] - icon_size - 10, position[1] - icon_size // 2) + cv2.rectangle(warning_image, icon_position, + (icon_position[0] + icon_size, icon_position[1] + icon_size), + color, -1) # 填充红色矩形 + + return warning_image + # 保持向后兼容的类 class SensorCalibrator: