#!/usr/bin/env python # -*- coding: utf-8 -*- ################################################################## # # Copyright (c) 2023 CICV, Inc. All Rights Reserved # ################################################################## """ @Authors: zhanghaiwen(zhanghaiwen@china-icv.cn) @Data: 2023/06/25 @Last Modified: 2025/05/20 @Summary: Chart generation utilities for metrics visualization """ """ 主要功能模块： 函数指标绘图（generate_function_chart_data） 舒适性指标绘图（generate_comfort_chart_data） 安全性指标绘图（generate_safety_chart_data） 交通指标绘图（generate_traffic_chart_data，待实现） 新增的急加速指标绘图： 在generate_comfort_chart_data中增加了slamaccelerate指标的支持 实现了完整的generate_slam_accelerate_chart函数，用于绘制急加速事件 该函数包含： 数据获取与预处理 CSV数据保存与读取 纵向加速度与速度的双子图绘制 急加速事件的橙色背景标记 阈值线绘制 高质量图表输出（300dpi PNG） 其他关键特性： 统一的日志记录系统 阈值获取函数get_metric_thresholds 错误处理和异常捕获 时间戳管理 数据验证机制 详细的日志输出 辅助函数： calculate_distance 和 calculate_relative_speed（简化实现） scenario_sign_dict 场景签名字典（简化实现） 此代码实现了完整的指标可视化工具，特别针对急加速指标slamAccelerate提供了详细的绘图功能，能够清晰展示急加速事件的发生时间和相关数据变化。 """ import matplotlib matplotlib.use('Agg') # 使用非图形界面的后端 import matplotlib.pyplot as plt import os import numpy as np import pandas as pd from typing import Optional, Dict, List, Any, Union from pathlib import Path from modules.lib.log_manager import LogManager def generate_function_chart_data(function_calculator, metric_name: str, output_dir: Optional[str] = None) -> Optional[ str]: """ Generate chart data for function metrics Args: function_calculator: FunctionCalculator instance metric_name: Metric name output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 确保输出目录存在 if output_dir: os.makedirs(output_dir, exist_ok=True) else: output_dir = os.path.join(os.getcwd(), 'data') # 根据指标名称选择不同的图表生成方法 if metric_name.lower() == 'latestwarningdistance_ttc_lst': return generate_latest_warning_ttc_chart(function_calculator, output_dir) elif metric_name.lower() == 'earliestwarningdistance_ttc_lst': return generate_earliest_warning_distance_ttc_chart(function_calculator, output_dir) elif metric_name.lower() == 'earliestwarningdistance_lst': return generate_earliest_warning_distance_chart(function_calculator, output_dir) elif metric_name.lower() == 'latestwarningdistance_lst': return generate_latest_warning_distance_chart(function_calculator, output_dir) elif metric_name.lower() == 'limitspeed_lst': return generate_limit_speed_chart(function_calculator, output_dir) elif metric_name.lower() == 'limitspeedpastlimitsign_lst': return generate_limit_speed_past_sign_chart(function_calculator, output_dir) elif metric_name.lower() == 'maxlongitudedist_lst': return generate_max_longitude_dist_chart(function_calculator, output_dir) else: logger.warning(f"Chart generation not implemented for metric [{metric_name}]") return None except Exception as e: logger.error(f"Failed to generate chart data: {str(e)}", exc_info=True) return None def generate_earliest_warning_distance_chart(function_calculator, output_dir: str) -> Optional[str]: """ Generate warning distance chart with data visualization. This function creates charts for earliestWarningDistance_LST and latestWarningDistance_LST metrics. Args: function_calculator: FunctionCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # Get data ego_df = function_calculator.ego_data.copy() # Check if correctwarning is already calculated correctwarning = getattr(function_calculator, 'correctwarning', None) # Get configured thresholds thresholds = get_metric_thresholds(function_calculator, 'earliestWarningDistance_LST') max_threshold = thresholds["max"] min_threshold = thresholds["min"] # Get calculated warning distance and speed warning_dist = getattr(function_calculator, 'warning_dist', None) if warning_dist.empty: logger.warning(f"Cannot generate {"earliestWarningDistance_LST"} chart: empty data") return None # Calculate metric value metric_value = float(warning_dist.iloc[0]) if len(warning_dist) >= 0.0 else max_threshold # Save CSV data csv_filename = os.path.join(output_dir, f"earliestWarningDistance_LST_data.csv") df_csv = pd.DataFrame({ 'simTime': ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]['simTime'], 'warning_distance': warning_dist, 'min_threshold': min_threshold, 'max_threshold': max_threshold, }) df_csv.to_csv(csv_filename, index=False) logger.info(f"earliestWarningDistance_LST data saved to: {csv_filename}") # Read data from CSV df = pd.read_csv(csv_filename) # Create single chart for warning distance plt.figure(figsize=(12, 6), constrained_layout=True) # Adjusted height for single chart # Plot warning distance plt.plot(df['simTime'], df['warning_distance'], 'b-', label='Warning Distance') # Add threshold lines plt.axhline(y=max_threshold, color='r', linestyle='--', label=f'Max Threshold ({max_threshold}m)') plt.axhline(y=min_threshold, color='g', linestyle='--', label=f'Min Threshold ({min_threshold}m)') # Mark metric value if len(df) > 0: label_text = 'Earliest Warning Distance' plt.scatter(df['simTime'].iloc[0], df['warning_distance'].iloc[0], color='red', s=100, zorder=5, label=f'{label_text}: {metric_value:.2f}m') # Set y-axis range plt.ylim(bottom=-1, top=max(max_threshold * 1.1, df['warning_distance'].max() * 1.1)) plt.xlabel('Time (s)') plt.ylabel('Distance (m)') plt.title(f'earliestWarningDistance_LST - Warning Distance Over Time') plt.grid(True) plt.legend() # Save image chart_filename = os.path.join(output_dir, f"earliestWarningDistance_LST_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"earliestWarningDistance_LST chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate earliestWarningDistance_LST chart: {str(e)}", exc_info=True) return None def generate_earliest_warning_distance_pgvil_chart(function_calculator, output_dir: str) -> Optional[str]: """ Generate warning distance chart with data visualization. This function creates charts for earliestWarningDistance_LST and latestWarningDistance_LST metrics. Args: function_calculator: FunctionCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # Get data ego_df = function_calculator.ego_data.copy() # Check if correctwarning is already calculated correctwarning = getattr(function_calculator, 'correctwarning', None) # Get configured thresholds thresholds = get_metric_thresholds(function_calculator, 'earliestWarningDistance_PGVIL') max_threshold = thresholds["max"] min_threshold = thresholds["min"] # Get calculated warning distance and speed warning_dist = getattr(function_calculator, 'warning_dist', None) if warning_dist.empty: logger.warning(f"Cannot generate {"earliestWarningDistance_LST"} chart: empty data") return None # Calculate metric value metric_value = float(warning_dist.iloc[0]) if len(warning_dist) >= 0.0 else max_threshold # Save CSV data csv_filename = os.path.join(output_dir, f"earliestWarningDistance_PGVIL_data.csv") df_csv = pd.DataFrame({ 'simTime': ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]['simTime'], 'warning_distance': warning_dist, 'min_threshold': min_threshold, 'max_threshold': max_threshold, }) df_csv.to_csv(csv_filename, index=False) logger.info(f"earliestWarningDistance_PGVIL data saved to: {csv_filename}") # Read data from CSV df = pd.read_csv(csv_filename) # Create single chart for warning distance plt.figure(figsize=(12, 6), constrained_layout=True) # Adjusted height for single chart # Plot warning distance plt.plot(df['simTime'], df['warning_distance'], 'b-', label='Warning Distance') # Add threshold lines plt.axhline(y=max_threshold, color='r', linestyle='--', label=f'Max Threshold ({max_threshold}m)') plt.axhline(y=min_threshold, color='g', linestyle='--', label=f'Min Threshold ({min_threshold}m)') # Mark metric value if len(df) > 0: label_text = 'Earliest Warning Distance' plt.scatter(df['simTime'].iloc[0], df['warning_distance'].iloc[0], color='red', s=100, zorder=5, label=f'{label_text}: {metric_value:.2f}m') # Set y-axis range plt.ylim(bottom=-1, top=max(max_threshold * 1.1, df['warning_distance'].max() * 1.1)) plt.xlabel('Time (s)') plt.ylabel('Distance (m)') plt.title(f'earliestWarningDistance_PGVIL - Warning Distance Over Time') plt.grid(True) plt.legend() # Save image chart_filename = os.path.join(output_dir, f"earliestWarningDistance_PGVIL_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"earliestWarningDistance_PGVIL chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate earliestWarningDistance_PGVIL chart: {str(e)}", exc_info=True) return None # # 使用function.py中已实现的find_nested_name函数 # from modules.metric.function import find_nested_name def generate_latest_warning_ttc_chart(function_calculator, output_dir: str) -> Optional[str]: """ Generate TTC warning chart with data visualization. This version first saves data to CSV, then uses the CSV to generate the chart. Args: function_calculator: FunctionCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 ego_df = function_calculator.ego_data.copy() correctwarning = getattr(function_calculator, 'correctwarning', None) # 获取配置的阈值 thresholds = get_metric_thresholds(function_calculator, 'latestWarningDistance_TTC_LST') max_threshold = thresholds["max"] min_threshold = thresholds["min"] warning_dist = getattr(function_calculator, 'warning_dist', None) warning_speed = getattr(function_calculator, 'warning_speed', None) ttc = getattr(function_calculator, 'ttc', None) if warning_dist.empty: logger.warning("Cannot generate TTC warning chart: empty data") return None # 生成时间戳 # 保存 CSV 数据 csv_filename = os.path.join(output_dir, f"latestwarningdistance_ttc_lst_data.csv") df_csv = pd.DataFrame({ 'simTime': ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]['simTime'], 'warning_distance': warning_dist, 'warning_speed': warning_speed, 'ttc': ttc, 'min_threshold': min_threshold, 'max_threshold': max_threshold, }) df_csv.to_csv(csv_filename, index=False) logger.info(f"latestwarningdistance_ttc_lst data saved to: {csv_filename}") # 从 CSV 读取数据 df = pd.read_csv(csv_filename) # 创建图表 plt.figure(figsize=(12, 8), constrained_layout=True) # 图 1：预警距离 ax1 = plt.subplot(3, 1, 1) ax1.plot(df['simTime'], df['warning_distance'], 'b-', label='Warning Distance') ax1.set_xlabel('Time (s)') ax1.set_ylabel('Distance (m)') ax1.set_title('Warning Distance Over Time') ax1.grid(True) ax1.legend() # 图 2：相对速度 ax2 = plt.subplot(3, 1, 2) ax2.plot(df['simTime'], df['warning_speed'], 'g-', label='Relative Speed') ax2.set_xlabel('Time (s)') ax2.set_ylabel('Speed (m/s)') ax2.set_title('Relative Speed Over Time') ax2.grid(True) ax2.legend() # 图 3：TTC ax3 = plt.subplot(3, 1, 3) ax3.plot(df['simTime'], df['ttc'], 'r-', label='TTC') # Add threshold lines ax3.axhline(y=max_threshold, color='r', linestyle='--', label=f'Max Threshold ({max_threshold}s)') ax3.axhline(y=min_threshold, color='g', linestyle='--', label=f'Min Threshold ({min_threshold}s)') # Calculate metric value (latest TTC) metric_value = float(ttc[-1]) if len(ttc) > 0 else max_threshold # Mark latest TTC value if len(df) > 0: ax3.scatter(df['simTime'].iloc[-1], df['ttc'].iloc[-1], color='red', s=100, zorder=5, label=f'Latest TTC: {metric_value:.2f}s') ax3.set_xlabel('Time (s)') ax3.set_ylabel('TTC (s)') ax3.set_title('Time To Collision (TTC) Over Time') ax3.grid(True) ax3.legend() # 保存图像 chart_filename = os.path.join(output_dir, f"latestwarningdistance_ttc_lst_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"latestwarningdistance_ttc_lst chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate latestwarningdistance_ttc_lst chart: {str(e)}", exc_info=True) return None def generate_latest_warning_distance_chart(function_calculator, output_dir: str) -> Optional[str]: """ Generate warning distance chart with data visualization. This function creates charts for latestWarningDistance_LST metric. Args: function_calculator: FunctionCalculator instance metric_name: Metric name (latestWarningDistance_LST) output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # Get data ego_df = function_calculator.ego_data.copy() # Check if correctwarning is already calculated correctwarning = getattr(function_calculator, 'correctwarning', None) # Get configured thresholds thresholds = get_metric_thresholds(function_calculator, 'latestWarningDistance_LST') max_threshold = thresholds["max"] min_threshold = thresholds["min"] # Get calculated warning distance and speed warning_dist = getattr(function_calculator, 'warning_dist', None) if warning_dist.empty: logger.warning(f"Cannot generate latestWarningDistance_LST chart: empty data") return None # Calculate metric value metric_value = float(warning_dist.iloc[-1]) if len(warning_dist) > 0 else max_threshold # Save CSV data csv_filename = os.path.join(output_dir, f"latestWarningDistance_LST_data.csv") df_csv = pd.DataFrame({ 'simTime': ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]['simTime'], 'warning_distance': warning_dist, 'min_threshold': min_threshold, 'max_threshold': max_threshold }) df_csv.to_csv(csv_filename, index=False) logger.info(f"latestWarningDistance_LST data saved to: {csv_filename}") # Read data from CSV df = pd.read_csv(csv_filename) # Create single chart for warning distance plt.figure(figsize=(12, 6), constrained_layout=True) # Adjusted height for single chart # Plot warning distance plt.plot(df['simTime'], df['warning_distance'], 'b-', label='Warning Distance') # Add threshold lines plt.axhline(y=max_threshold, color='r', linestyle='--', label=f'Max Threshold ({max_threshold}m)') plt.axhline(y=min_threshold, color='g', linestyle='--', label=f'Min Threshold ({min_threshold}m)') # Mark metric value if len(df) > 0: label_text = 'Latest Warning Distance' plt.scatter(df['simTime'].iloc[-1], df['warning_distance'].iloc[-1], color='red', s=100, zorder=5, label=f'{label_text}: {metric_value:.2f}m') # Set y-axis range plt.ylim(bottom=-1, top=max(max_threshold * 1.1, df['warning_distance'].max() * 1.1)) plt.xlabel('Time (s)') plt.ylabel('Distance (m)') plt.title(f'latestWarningDistance_LST - Warning Distance Over Time') plt.grid(True) plt.legend() # Save image chart_filename = os.path.join(output_dir, f"latestWarningDistance_LST_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"latestWarningDistance_LST chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate latestWarningDistance_LST chart: {str(e)}", exc_info=True) return None def generate_earliest_warning_distance_ttc_chart(function_calculator, output_dir: str) -> Optional[str]: """ Generate TTC warning chart with data visualization for earliestWarningDistance_TTC_LST metric. Args: function_calculator: FunctionCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() metric_name = 'earliestWarningDistance_TTC_LST' try: # Get data ego_df = function_calculator.ego_data.copy() # Check if correctwarning is already calculated correctwarning = getattr(function_calculator, 'correctwarning', None) # Get configured thresholds thresholds = get_metric_thresholds(function_calculator, metric_name) max_threshold = thresholds["max"] min_threshold = thresholds["min"] # Get calculated warning distance and speed warning_dist = getattr(function_calculator, 'correctwarning', None) warning_speed = getattr(function_calculator, 'warning_speed', None) ttc = getattr(function_calculator, 'ttc', None) # Calculate metric value metric_value = float(ttc[0]) if len(ttc) > 0 else max_threshold # Save CSV data csv_filename = os.path.join(output_dir, f"{metric_name.lower()}_data.csv") df_csv = pd.DataFrame({ 'simTime': ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]['simTime'], 'warning_distance': warning_dist, 'warning_speed': warning_speed, 'ttc': ttc, 'min_threshold': min_threshold, 'max_threshold': max_threshold }) df_csv.to_csv(csv_filename, index=False) logger.info(f"{metric_name} data saved to: {csv_filename}") # Read data from CSV df = pd.read_csv(csv_filename) # Create chart plt.figure(figsize=(12, 8), constrained_layout=True) # 图 1：预警距离 ax1 = plt.subplot(3, 1, 1) ax1.plot(df['simTime'], df['warning_distance'], 'b-', label='Warning Distance') ax1.set_xlabel('Time (s)') ax1.set_ylabel('Distance (m)') ax1.set_title('Warning Distance Over Time') ax1.grid(True) ax1.legend() # 图 2：相对速度 ax2 = plt.subplot(3, 1, 2) ax2.plot(df['simTime'], df['warning_speed'], 'g-', label='Relative Speed') ax2.set_xlabel('Time (s)') ax2.set_ylabel('Speed (m/s)') ax2.set_title('Relative Speed Over Time') ax2.grid(True) ax2.legend() # 图 3：TTC ax3 = plt.subplot(3, 1, 3) ax3.plot(df['simTime'], df['ttc'], 'r-', label='TTC') # Add threshold lines ax3.axhline(y=max_threshold, color='r', linestyle='--', label=f'Max Threshold ({max_threshold}s)') ax3.axhline(y=min_threshold, color='g', linestyle='--', label=f'Min Threshold ({min_threshold}s)') # Mark earliest TTC value if len(df) > 0: ax3.scatter(df['simTime'].iloc[0], df['ttc'].iloc[0], color='red', s=100, zorder=5, label=f'Earliest TTC: {metric_value:.2f}s') ax3.set_xlabel('Time (s)') ax3.set_ylabel('TTC (s)') ax3.set_title('Time To Collision (TTC) Over Time') ax3.grid(True) ax3.legend() # Save image chart_filename = os.path.join(output_dir, f"earliestwarningdistance_ttc_lst_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"{metric_name} chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate earliestwarningdistance_ttc_lst chart: {str(e)}", exc_info=True) return None def generate_limit_speed_chart(function_calculator, output_dir: str) -> Optional[str]: """ Generate limit speed chart with data visualization for limitSpeed_LST metric. Args: function_calculator: FunctionCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() metric_name = 'limitSpeed_LST' try: # Get data ego_df = function_calculator.ego_data.copy() # Get configured thresholds thresholds = get_metric_thresholds(function_calculator, metric_name) max_threshold = thresholds["max"] min_threshold = thresholds["min"] if ego_df.empty: logger.warning(f"Cannot generate {metric_name} chart: empty data") return None # Save CSV data csv_filename = os.path.join(output_dir, f"{metric_name.lower()}_data.csv") df_csv = pd.DataFrame({ 'simTime': ego_df['simTime'], 'speed': ego_df['v'], 'speed_limit': ego_df.get('speed_limit', pd.Series([max_threshold] * len(ego_df))) }) df_csv.to_csv(csv_filename, index=False) logger.info(f"{metric_name} data saved to: {csv_filename}") # Read data from CSV df = pd.read_csv(csv_filename) # Create chart plt.figure(figsize=(12, 6), constrained_layout=True) # Plot speed plt.plot(df['simTime'], df['speed'], 'b-', label='Vehicle Speed') plt.plot(df['simTime'], df['speed_limit'], 'r--', label='Speed Limit') # Set y-axis range plt.ylim(bottom=0, top=max(max_threshold * 1.1, df['speed'].max() * 1.1)) plt.xlabel('Time (s)') plt.ylabel('Speed (m/s)') plt.title(f'{metric_name} - Vehicle Speed vs Speed Limit') plt.grid(True) plt.legend() # Save image chart_filename = os.path.join(output_dir, f"{metric_name.lower()}_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"{metric_name} chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate {metric_name} chart: {str(e)}", exc_info=True) return None def generate_limit_speed_past_sign_chart(function_calculator, output_dir: str) -> Optional[str]: """ Generate limit speed past sign chart with data visualization for limitSpeedPastLimitSign_LST metric. Args: function_calculator: FunctionCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() metric_name = 'limitSpeedPastLimitSign_LST' try: # Get data ego_df = function_calculator.ego_data.copy() # Get configured thresholds thresholds = get_metric_thresholds(function_calculator, metric_name) max_threshold = thresholds["max"] min_threshold = thresholds["min"] if ego_df.empty: logger.warning(f"Cannot generate {metric_name} chart: empty data") return None # Get sign passing time if available sign_time = getattr(function_calculator, 'sign_pass_time', None) if sign_time is None: # Try to estimate sign passing time (middle of the simulation) sign_time = ego_df['simTime'].iloc[len(ego_df) // 2] # Save CSV data csv_filename = os.path.join(output_dir, f"{metric_name.lower()}_data.csv") df_csv = pd.DataFrame({ 'simTime': ego_df['simTime'], 'speed': ego_df['v'], 'speed_limit': ego_df.get('speed_limit', pd.Series([max_threshold] * len(ego_df))), 'sign_pass_time': sign_time }) df_csv.to_csv(csv_filename, index=False) logger.info(f"{metric_name} data saved to: {csv_filename}") # Read data from CSV df = pd.read_csv(csv_filename) # Create chart plt.figure(figsize=(12, 6), constrained_layout=True) # Plot speed plt.plot(df['simTime'], df['speed'], 'b-', label='Vehicle Speed') plt.plot(df['simTime'], df['speed_limit'], 'r--', label='Speed Limit') # Mark sign passing time plt.axvline(x=sign_time, color='g', linestyle='--', label='Speed Limit Sign') # Set y-axis range plt.ylim(bottom=0, top=max(max_threshold * 1.1, df['speed'].max() * 1.1)) plt.xlabel('Time (s)') plt.ylabel('Speed (m/s)') plt.title(f'{metric_name} - Vehicle Speed vs Speed Limit') plt.grid(True) plt.legend() # Save image chart_filename = os.path.join(output_dir, f"{metric_name.lower()}_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"{metric_name} chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate {metric_name} chart: {str(e)}", exc_info=True) return None def generate_max_longitude_dist_chart(function_calculator, output_dir: str) -> Optional[str]: """ Generate maximum longitudinal distance chart with data visualization for maxLongitudeDist_LST metric. Args: function_calculator: FunctionCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() metric_name = 'maxLongitudeDist_LST' try: # Get data ego_df = function_calculator.ego_data.copy() # Get configured thresholds thresholds = get_metric_thresholds(function_calculator, metric_name) max_threshold = thresholds["max"] min_threshold = thresholds["min"] # Get longitudinal distance data longitude_dist = ego_df['longitude_dist'] if 'longitude_dist' in ego_df.columns else None if longitude_dist is None or longitude_dist.empty: logger.warning(f"Cannot generate {metric_name} chart: missing longitudinal distance data") return None # Calculate metric value metric_value = longitude_dist.max() max_distance_time = ego_df.loc[longitude_dist.idxmax(), 'simTime'] # Save CSV data csv_filename = os.path.join(output_dir, f"{metric_name.lower()}_data.csv") df_csv = pd.DataFrame({ 'simTime': ego_df['simTime'], 'longitude_dist': longitude_dist, 'min_threshold': min_threshold, 'max_threshold': max_threshold }) df_csv.to_csv(csv_filename, index=False) logger.info(f"{metric_name} data saved to: {csv_filename}") # Read data from CSV df = pd.read_csv(csv_filename) # Create chart plt.figure(figsize=(12, 6), constrained_layout=True) # Plot longitudinal distance plt.plot(df['simTime'], df['longitude_dist'], 'b-', label='Longitudinal Distance') # Add threshold lines plt.axhline(y=max_threshold, color='r', linestyle='--', label=f'Max Threshold ({max_threshold}m)') plt.axhline(y=min_threshold, color='g', linestyle='--', label=f'Min Threshold ({min_threshold}m)') # Mark maximum longitudinal distance plt.scatter(max_distance_time, metric_value, color='red', s=100, zorder=5, label=f'Maximum Longitudinal Distance: {metric_value:.2f}m') # Set y-axis range plt.ylim(bottom=min(0, min_threshold * 0.9), top=max(max_threshold * 1.1, df['longitude_dist'].max() * 1.1)) plt.xlabel('Time (s)') plt.ylabel('Longitudinal Distance (m)') plt.title(f'{metric_name} - Longitudinal Distance Over Time') plt.grid(True) plt.legend() # Save image chart_filename = os.path.join(output_dir, f"{metric_name.lower()}_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"{metric_name} chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate {metric_name} chart: {str(e)}", exc_info=True) return None def generate_warning_delay_time_chart(function_calculator, output_dir: str) -> Optional[str]: """ Generate warning delay time chart with data visualization for warningDelayTime_LST metric. Args: function_calculator: FunctionCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() metric_name = 'warningDelayTime_LST' try: # Get data ego_df = function_calculator.ego_data.copy() # Get configured thresholds thresholds = get_metric_thresholds(function_calculator, metric_name) max_threshold = thresholds["max"] min_threshold = thresholds["min"] # Check if correctwarning is already calculated correctwarning = getattr(function_calculator, 'correctwarning', None) if correctwarning is None: logger.warning(f"Cannot generate {metric_name} chart: missing correctwarning value") return None # Get HMI warning time and rosbag warning time HMI_warning_rows = ego_df[(ego_df['ifwarning'] == correctwarning)]['simTime'].tolist() simTime_HMI = HMI_warning_rows[0] if len(HMI_warning_rows) > 0 else None rosbag_warning_rows = ego_df[(ego_df['event_Type'].notna()) & ((ego_df['event_Type'] != np.nan))][ 'simTime'].tolist() simTime_rosbag = rosbag_warning_rows[0] if len(rosbag_warning_rows) > 0 else None if (simTime_HMI is None) or (simTime_rosbag is None): logger.warning(f"Cannot generate {metric_name} chart: missing warning time data") return None # Calculate delay time delay_time = abs(simTime_HMI - simTime_rosbag) # Save CSV data csv_filename = os.path.join(output_dir, f"{metric_name.lower()}_data.csv") df_csv = pd.DataFrame({ 'HMI_warning_time': [simTime_HMI], 'rosbag_warning_time': [simTime_rosbag], 'delay_time': [delay_time], 'min_threshold': [min_threshold], 'max_threshold': [max_threshold] }) df_csv.to_csv(csv_filename, index=False) logger.info(f"{metric_name} data saved to: {csv_filename}") # Create chart - bar chart for delay time plt.figure(figsize=(10, 6), constrained_layout=True) # Plot delay time as bar plt.bar(['Warning Delay Time'], [delay_time], color='blue', width=0.4) # Add threshold lines plt.axhline(y=max_threshold, color='r', linestyle='--', label=f'Max Threshold ({max_threshold}s)') plt.axhline(y=min_threshold, color='g', linestyle='--', label=f'Min Threshold ({min_threshold}s)') # Add value label plt.text(0, delay_time + 0.05, f'{delay_time:.3f}s', ha='center', va='bottom', fontweight='bold') # Set y-axis range plt.ylim(bottom=0, top=max(max_threshold * 1.2, delay_time * 1.2)) plt.ylabel('Delay Time (s)') plt.title(f'{metric_name} - Warning Delay Time') plt.grid(True, axis='y') plt.legend() # Save image chart_filename = os.path.join(output_dir, f"{metric_name.lower()}_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"{metric_name} chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate {metric_name} chart: {str(e)}", exc_info=True) return None def generate_comfort_chart_data(comfort_calculator, metric_name: str, output_dir: Optional[str] = None) -> Optional[ str]: """ Generate chart data for comfort metrics Args: comfort_calculator: ComfortCalculator instance metric_name: Metric name output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 确保输出目录存在 if output_dir: os.makedirs(output_dir, exist_ok=True) else: output_dir = os.getcwd() # 根据指标名称选择不同的图表生成方法 if metric_name.lower() == 'shake': return generate_shake_chart(comfort_calculator, output_dir) elif metric_name.lower() == 'zigzag': return generate_zigzag_chart(comfort_calculator, output_dir) elif metric_name.lower() == 'cadence': return generate_cadence_chart(comfort_calculator, output_dir) elif metric_name.lower() == 'slambrake': return generate_slam_brake_chart(comfort_calculator, output_dir) elif metric_name.lower() == 'slamaccelerate': return generate_slam_accelerate_chart(comfort_calculator, output_dir) else: logger.warning(f"Chart generation not implemented for metric [{metric_name}]") return None except Exception as e: logger.error(f"Failed to generate chart data: {str(e)}", exc_info=True) return None def generate_shake_chart(comfort_calculator, output_dir: str) -> Optional[str]: """ Generate shake metric chart with orange background for shake events. This version first saves data to CSV, then uses the CSV to generate the chart. Args: comfort_calculator: ComfortCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 df = comfort_calculator.ego_df.copy() shake_events = comfort_calculator.shake_events if df.empty: logger.warning("Cannot generate shake chart: empty data") return None # 生成时间戳 # 保存 CSV 数据（第一步） csv_filename = os.path.join(output_dir, f"shake_data.csv") df_csv = pd.DataFrame({ 'simTime': df['simTime'], 'lat_acc': df['lat_acc'], 'lat_acc_rate': df['lat_acc_rate'], 'speedH_std': df['speedH_std'], 'lat_acc_threshold': df.get('lat_acc_threshold', pd.Series([None] * len(df))), 'lat_acc_rate_threshold': 0.5, 'speedH_std_threshold': df.get('speedH_threshold', pd.Series([None] * len(df))), }) df_csv.to_csv(csv_filename, index=False) logger.info(f"Shake data saved to: {csv_filename}") # 第二步：从 CSV 读取（可验证保存数据无误） df = pd.read_csv(csv_filename) # 创建图表（第三步） import matplotlib.pyplot as plt plt.figure(figsize=(12, 8), constrained_layout=True) # 图 1：横向加速度 ax1 = plt.subplot(3, 1, 1) ax1.plot(df['simTime'], df['lat_acc'], 'b-', label='Lateral Acceleration') if 'lat_acc_threshold' in df.columns: ax1.plot(df['simTime'], df['lat_acc_threshold'], 'r--', label='lat_acc_threshold') for idx, event in enumerate(shake_events): label = 'Shake Event' if idx == 0 else None ax1.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax1.set_xlabel('Time (s)') ax1.set_ylabel('Lateral Acceleration (m/s²)') ax1.set_title('Shake Event Detection - Lateral Acceleration') ax1.grid(True) ax1.legend() # 图 2：lat_acc_rate ax2 = plt.subplot(3, 1, 2) ax2.plot(df['simTime'], df['lat_acc_rate'], 'g-', label='lat_acc_rate') ax2.axhline( y=0.5, color='orange', linestyle='--', linewidth=1.2, label='lat_acc_rate_threshold' ) for idx, event in enumerate(shake_events): label = 'Shake Event' if idx == 0 else None ax2.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax2.set_xlabel('Time (s)') ax2.set_ylabel('Angular Velocity (m/s³)') ax2.set_title('Shake Event Detection - lat_acc_rate') ax2.grid(True) ax2.legend() # 图 3：speedH_std ax3 = plt.subplot(3, 1, 3) ax3.plot(df['simTime'], df['speedH_std'], 'b-', label='speedH_std') if 'speedH_std_threshold' in df.columns: ax3.plot(df['simTime'], df['speedH_std_threshold'], 'r--', label='speedH_threshold') for idx, event in enumerate(shake_events): label = 'Shake Event' if idx == 0 else None ax3.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax3.set_xlabel('Time (s)') ax3.set_ylabel('Angular Velocity (deg/s)') ax3.set_title('Shake Event Detection - speedH_std') ax3.grid(True) ax3.legend() # 保存图像 chart_filename = os.path.join(output_dir, f"shake_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"Shake chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate shake chart: {str(e)}", exc_info=True) return None def generate_zigzag_chart(comfort_calculator, output_dir: str) -> Optional[str]: """ Generate zigzag metric chart with orange background for zigzag events. This version first saves data to CSV, then uses the CSV to generate the chart. Args: comfort_calculator: ComfortCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 df = comfort_calculator.ego_df.copy() zigzag_events = comfort_calculator.discomfort_df[ comfort_calculator.discomfort_df['type'] == 'zigzag' ].copy() if df.empty: logger.warning("Cannot generate zigzag chart: empty data") return None # 生成时间戳 # 保存 CSV 数据（第一步） csv_filename = os.path.join(output_dir, f"zigzag_data.csv") df_csv = pd.DataFrame({ 'simTime': df['simTime'], 'speedH': df['speedH'], 'posH': df['posH'], 'min_speedH_threshold': -2.3, # 可替换为动态阈值 'max_speedH_threshold': 2.3 }) df_csv.to_csv(csv_filename, index=False) logger.info(f"Zigzag data saved to: {csv_filename}") # 第二步：从 CSV 读取（可验证保存数据无误） df = pd.read_csv(csv_filename) # 创建图表（第三步） import matplotlib.pyplot as plt plt.figure(figsize=(12, 8), constrained_layout=True) # ===== 子图1：Yaw Rate ===== ax1 = plt.subplot(2, 1, 1) ax1.plot(df['simTime'], df['speedH'], 'g-', label='Yaw Rate') # 添加 speedH 上下限阈值线 ax1.axhline(y=2.3, color='m', linestyle='--', linewidth=1.2, label='Max Threshold (+2.3)') ax1.axhline(y=-2.3, color='r', linestyle='--', linewidth=1.2, label='Min Threshold (-2.3)') # 添加橙色背景：Zigzag Events for idx, event in zigzag_events.iterrows(): label = 'Zigzag Event' if idx == 0 else None ax1.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax1.set_xlabel('Time (s)') ax1.set_ylabel('Angular Velocity (deg/s)') ax1.set_title('Zigzag Event Detection - Yaw Rate') ax1.grid(True) ax1.legend(loc='upper left') # ===== 子图2：Yaw Angle ===== ax2 = plt.subplot(2, 1, 2) ax2.plot(df['simTime'], df['posH'], 'b-', label='Yaw') # 添加橙色背景：Zigzag Events for idx, event in zigzag_events.iterrows(): label = 'Zigzag Event' if idx == 0 else None ax2.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax2.set_xlabel('Time (s)') ax2.set_ylabel('Yaw (deg)') ax2.set_title('Zigzag Event Detection - Yaw Angle') ax2.grid(True) ax2.legend(loc='upper left') # 保存图像 chart_filename = os.path.join(output_dir, f"zigzag_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"Zigzag chart saved to: {chart_filename}") return csv_filename except Exception as e: logger.error(f"Failed to generate zigzag chart: {str(e)}", exc_info=True) return None def generate_cadence_chart(comfort_calculator, output_dir: str) -> Optional[str]: """ Generate cadence metric chart with orange background for cadence events. This version first saves data to CSV, then uses the CSV to generate the chart. Args: comfort_calculator: ComfortCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 df = comfort_calculator.ego_df.copy() cadence_events = comfort_calculator.discomfort_df[comfort_calculator.discomfort_df['type'] == 'cadence'].copy() if df.empty: logger.warning("Cannot generate cadence chart: empty data") return None # 生成时间戳 # 保存 CSV 数据（第一步） csv_filename = os.path.join(output_dir, f"cadence_data.csv") df_csv = pd.DataFrame({ 'simTime': df['simTime'], 'lon_acc': df['lon_acc'], 'v': df['v'], 'ip_acc': df.get('ip_acc', pd.Series([None] * len(df))), 'ip_dec': df.get('ip_dec', pd.Series([None] * len(df))) }) df_csv.to_csv(csv_filename, index=False) logger.info(f"Cadence data saved to: {csv_filename}") # 第二步：从 CSV 读取（可验证保存数据无误） df = pd.read_csv(csv_filename) # 创建图表（第三步） import matplotlib.pyplot as plt plt.figure(figsize=(12, 8), constrained_layout=True) # 图 1：纵向加速度 ax1 = plt.subplot(2, 1, 1) ax1.plot(df['simTime'], df['lon_acc'], 'b-', label='Longitudinal Acceleration') if 'ip_acc' in df.columns and 'ip_dec' in df.columns: ax1.plot(df['simTime'], df['ip_acc'], 'r--', label='Acceleration Threshold') ax1.plot(df['simTime'], df['ip_dec'], 'g--', label='Deceleration Threshold') # 添加橙色背景标识顿挫事件 for idx, event in cadence_events.iterrows(): label = 'Cadence Event' if idx == 0 else None ax1.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax1.set_xlabel('Time (s)') ax1.set_ylabel('Longitudinal Acceleration (m/s²)') ax1.set_title('Cadence Event Detection - Longitudinal Acceleration') ax1.grid(True) ax1.legend() # 图 2：速度 ax2 = plt.subplot(2, 1, 2) ax2.plot(df['simTime'], df['v'], 'g-', label='Velocity') # 添加橙色背景标识顿挫事件 for idx, event in cadence_events.iterrows(): label = 'Cadence Event' if idx == 0 else None ax2.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax2.set_xlabel('Time (s)') ax2.set_ylabel('Velocity (m/s)') ax2.set_title('Cadence Event Detection - Vehicle Speed') ax2.grid(True) ax2.legend() # 保存图像 chart_filename = os.path.join(output_dir, f"cadence_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"Cadence chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate cadence chart: {str(e)}", exc_info=True) return None def generate_slam_brake_chart(comfort_calculator, output_dir: str) -> Optional[str]: """ Generate slam brake metric chart with orange background for slam brake events. This version first saves data to CSV, then uses the CSV to generate the chart. Args: comfort_calculator: ComfortCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 df = comfort_calculator.ego_df.copy() slam_brake_events = comfort_calculator.discomfort_df[ comfort_calculator.discomfort_df['type'] == 'slam_brake'].copy() if df.empty: logger.warning("Cannot generate slam brake chart: empty data") return None # 生成时间戳 # 保存 CSV 数据（第一步） csv_filename = os.path.join(output_dir, f"slam_brake_data.csv") df_csv = pd.DataFrame({ 'simTime': df['simTime'], 'lon_acc': df['lon_acc'], 'v': df['v'], 'min_threshold': df.get('ip_dec', pd.Series([None] * len(df))), 'max_threshold': 0.0 }) df_csv.to_csv(csv_filename, index=False) logger.info(f"Slam brake data saved to: {csv_filename}") # 第二步：从 CSV 读取（可验证保存数据无误） df = pd.read_csv(csv_filename) # 创建图表（第三步） plt.figure(figsize=(12, 8), constrained_layout=True) # 图 1：纵向加速度 ax1 = plt.subplot(2, 1, 1) ax1.plot(df['simTime'], df['lon_acc'], 'b-', label='Longitudinal Acceleration') if 'min_threshold' in df.columns: ax1.plot(df['simTime'], df['min_threshold'], 'r--', label='Deceleration Threshold') # 添加橙色背景标识急刹车事件 for idx, event in slam_brake_events.iterrows(): label = 'Slam Brake Event' if idx == 0 else None ax1.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax1.set_xlabel('Time (s)') ax1.set_ylabel('Longitudinal Acceleration (m/s²)') ax1.set_title('Slam Brake Event Detection - Longitudinal Acceleration') ax1.grid(True) ax1.legend() # 图 2：速度 ax2 = plt.subplot(2, 1, 2) ax2.plot(df['simTime'], df['v'], 'g-', label='Velocity') # 添加橙色背景标识急刹车事件 for idx, event in slam_brake_events.iterrows(): label = 'Slam Brake Event' if idx == 0 else None ax2.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax2.set_xlabel('Time (s)') ax2.set_ylabel('Velocity (m/s)') ax2.set_title('Slam Brake Event Detection - Vehicle Speed') ax2.grid(True) ax2.legend() # 保存图像 chart_filename = os.path.join(output_dir, f"slam_brake_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"Slam brake chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate slam brake chart: {str(e)}", exc_info=True) return None def generate_slam_accelerate_chart(comfort_calculator, output_dir: str) -> Optional[str]: """ Generate slam accelerate metric chart with orange background for slam accelerate events. This version first saves data to CSV, then uses the CSV to generate the chart. Args: comfort_calculator: ComfortCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 df = comfort_calculator.ego_df.copy() slam_accel_events = comfort_calculator.discomfort_df[ (comfort_calculator.discomfort_df['type'] == 'slam_accel') ].copy() if df.empty: logger.warning("Cannot generate slam accelerate chart: empty data") return None # 生成时间戳 # 保存 CSV 数据（第一步） csv_filename = os.path.join(output_dir, f"slam_accel_data.csv") # 获取加速度阈值（如果存在） accel_threshold = df.get('ip_acc', pd.Series([None] * len(df))) df_csv = pd.DataFrame({ 'simTime': df['simTime'], 'lon_acc': df['lon_acc'], 'v': df['v'], 'min_threshold': 0.0, # 加速度最小阈值设为0 'max_threshold': accel_threshold # 急加速阈值 }) df_csv.to_csv(csv_filename, index=False) logger.info(f"Slam accelerate data saved to: {csv_filename}") # 第二步：从 CSV 读取（可验证保存数据无误） df = pd.read_csv(csv_filename) # 创建图表（第三步） plt.figure(figsize=(12, 8), constrained_layout=True) # 图 1：纵向加速度 ax1 = plt.subplot(2, 1, 1) ax1.plot(df['simTime'], df['lon_acc'], 'b-', label='Longitudinal Acceleration') # 添加加速度阈值线 if 'max_threshold' in df.columns and not df['max_threshold'].isnull().all(): ax1.plot(df['simTime'], df['max_threshold'], 'r--', label='Acceleration Threshold') # 添加橙色背景标识急加速事件 for idx, event in slam_accel_events.iterrows(): label = 'Slam Accelerate Event' if idx == 0 else None ax1.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax1.set_xlabel('Time (s)') ax1.set_ylabel('Acceleration (m/s²)') ax1.set_title('Slam Accelerate Event Detection - Longitudinal Acceleration') ax1.grid(True) ax1.legend() # 图 2：速度 ax2 = plt.subplot(2, 1, 2) ax2.plot(df['simTime'], df['v'], 'g-', label='Velocity') # 添加橙色背景标识急加速事件 for idx, event in slam_accel_events.iterrows(): label = 'Slam Accelerate Event' if idx == 0 else None ax2.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) ax2.set_xlabel('Time (s)') ax2.set_ylabel('Velocity (m/s)') ax2.set_title('Slam Accelerate Event Detection - Vehicle Speed') ax2.grid(True) ax2.legend() # 保存图像 chart_filename = os.path.join(output_dir, f"slam_accel_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() logger.info(f"Slam accelerate chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate slam accelerate chart: {str(e)}", exc_info=True) return None def get_metric_thresholds(calculator, metric_name: str) -> dict: """ 从配置文件中获取指标的阈值 Args: calculator: Calculator instance (FunctionCalculator, SafetyCalculator, ComfortCalculator, EfficientCalculator, TrafficCalculator) metric_name: 指标名称 Returns: dict: 包含min和max阈值的字典 """ logger = LogManager().get_logger() thresholds = {"min": None, "max": None} try: # 根据计算器类型获取配置 if hasattr(calculator, 'data_processed'): # 检查安全性指标配置 if hasattr(calculator.data_processed, 'safety_config') and 'safety' in calculator.data_processed.safety_config: config = calculator.data_processed.safety_config['safety'] metric_type = 'safety' # 检查舒适性指标配置 elif hasattr(calculator.data_processed, 'comfort_config') and 'comfort' in calculator.data_processed.comfort_config: config = calculator.data_processed.comfort_config['comfort'] metric_type = 'comfort' # 检查功能性指标配置 elif hasattr(calculator.data_processed, 'function_config') and 'function' in calculator.data_processed.function_config: config = calculator.data_processed.function_config['function'] metric_type = 'function' # 检查高效性指标配置 elif hasattr(calculator.data_processed, 'efficient_config') and 'efficient' in calculator.data_processed.efficient_config: config = calculator.data_processed.efficient_config['efficient'] metric_type = 'efficient' # 检查交通性指标配置 elif hasattr(calculator.data_processed, 'traffic_config') and 'traffic' in calculator.data_processed.traffic_config: config = calculator.data_processed.traffic_config['traffic'] metric_type = 'traffic' else: # 直接检查calculator是否有function_config属性（针对FunctionCalculator） if hasattr(calculator, 'function_config') and 'function' in calculator.function_config: config = calculator.function_config['function'] metric_type = 'function' else: logger.warning(f"无法找到{metric_name}的配置信息") return thresholds else: # 直接检查calculator是否有function_config属性（针对FunctionCalculator） if hasattr(calculator, 'function_config') and 'function' in calculator.function_config: config = calculator.function_config['function'] metric_type = 'function' else: logger.warning(f"计算器没有data_processed属性或function_config属性") return thresholds # 递归查找指标配置 def find_metric_config(node, target_name): if isinstance(node, dict): if 'name' in node and node['name'].lower() == target_name.lower() and 'min' in node and 'max' in node: return node for key, value in node.items(): result = find_metric_config(value, target_name) if result: return result return None # 查找指标配置 metric_config = find_metric_config(config, metric_name) if metric_config: thresholds["min"] = metric_config.get("min") thresholds["max"] = metric_config.get("max") logger.info(f"找到{metric_name}的阈值: min={thresholds['min']}, max={thresholds['max']}") else: logger.warning(f"在{metric_type}配置中未找到{metric_name}的阈值信息") except Exception as e: logger.error(f"获取{metric_name}阈值时出错: {str(e)}", exc_info=True) return thresholds def generate_safety_chart_data(safety_calculator, metric_name: str, output_dir: Optional[str] = None) -> Optional[str]: """ Generate chart data for safety metrics Args: safety_calculator: SafetyCalculator instance metric_name: Metric name output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 确保输出目录存在 if output_dir: os.makedirs(output_dir, exist_ok=True) else: output_dir = os.getcwd() # 根据指标名称选择不同的图表生成方法 if metric_name.lower() == 'ttc': return generate_ttc_chart(safety_calculator, output_dir) elif metric_name.lower() == 'mttc': return generate_mttc_chart(safety_calculator, output_dir) elif metric_name.lower() == 'thw': return generate_thw_chart(safety_calculator, output_dir) elif metric_name.lower() == 'lonsd': return generate_lonsd_chart(safety_calculator, output_dir) elif metric_name.lower() == 'latsd': return generate_latsd_chart(safety_calculator, output_dir) elif metric_name.lower() == 'btn': return generate_btn_chart(safety_calculator, output_dir) elif metric_name.lower() == 'collisionrisk': return generate_collision_risk_chart(safety_calculator, output_dir) elif metric_name.lower() == 'collisionseverity': return generate_collision_severity_chart(safety_calculator, output_dir) else: logger.warning(f"Chart generation not implemented for metric [{metric_name}]") return None except Exception as e: logger.error(f"Failed to generate chart data: {str(e)}", exc_info=True) return None def generate_ttc_chart(safety_calculator, output_dir: str) -> Optional[str]: """ Generate TTC metric chart with orange background for unsafe events. This version first saves data to CSV, then uses the CSV to generate the chart. Args: safety_calculator: SafetyCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 ttc_data = safety_calculator.ttc_data if not ttc_data: logger.warning("Cannot generate TTC chart: empty data") return None # 创建DataFrame df = pd.DataFrame(ttc_data) # 获取阈值 thresholds = get_metric_thresholds(safety_calculator, 'TTC') min_threshold = thresholds.get('min') max_threshold = thresholds.get('max') # 生成时间戳 # 保存 CSV 数据（第一步） csv_filename = os.path.join(output_dir, f"ttc_data.csv") df_csv = pd.DataFrame({ 'simTime': df['simTime'], 'simFrame': df['simFrame'], 'TTC': df['TTC'], 'min_threshold': min_threshold, 'max_threshold': max_threshold }) df_csv.to_csv(csv_filename, index=False) logger.info(f"TTC data saved to: {csv_filename}") # 第二步：从 CSV 读取（可验证保存数据无误） df = pd.read_csv(csv_filename) # 检测超阈值事件 unsafe_events = [] if min_threshold is not None: # 对于TTC，小于最小阈值视为不安全 unsafe_condition = df['TTC'] < min_threshold event_groups = (unsafe_condition != unsafe_condition.shift()).cumsum() for _, group in df[unsafe_condition].groupby(event_groups): if len(group) >= 2: # 至少2帧才算一次事件 start_time = group['simTime'].iloc[0] end_time = group['simTime'].iloc[-1] duration = end_time - start_time if duration >= 0.1: # 只记录持续时间超过0.1秒的事件 unsafe_events.append({ 'start_time': start_time, 'end_time': end_time, 'start_frame': group['simFrame'].iloc[0], 'end_frame': group['simFrame'].iloc[-1], 'duration': duration, 'min_ttc': group['TTC'].min() }) # 创建图表（第三步） plt.figure(figsize=(12, 8)) plt.plot(df['simTime'], df['TTC'], 'b-', label='TTC') # 添加阈值线 if min_threshold is not None: plt.axhline(y=min_threshold, color='r', linestyle='--', label=f'Min Threshold ({min_threshold}s)') if max_threshold is not None: plt.axhline(y=max_threshold, color='g', linestyle='--', label=f'Max Threshold ({max_threshold})') # 添加橙色背景标识不安全事件 for idx, event in enumerate(unsafe_events): label = 'Unsafe TTC Event' if idx == 0 else None plt.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) plt.xlabel('Time (s)') plt.ylabel('TTC (s)') plt.title('Time To Collision (TTC) Trend') plt.grid(True) plt.legend() # 保存图像 chart_filename = os.path.join(output_dir, f"ttc_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() # 记录不安全事件信息 if unsafe_events: logger.info(f"检测到 {len(unsafe_events)} 个TTC不安全事件") for i, event in enumerate(unsafe_events): logger.info( f"TTC不安全事件 #{i + 1}: 开始时间={event['start_time']:.2f}s, 结束时间={event['end_time']:.2f}s, 持续时间={event['duration']:.2f}s, 最小TTC={event['min_ttc']:.2f}s") logger.info(f"TTC chart saved to: {chart_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate TTC chart: {str(e)}", exc_info=True) return None def generate_mttc_chart(safety_calculator, output_dir: str) -> Optional[str]: """ Generate MTTC metric chart with orange background for unsafe events Args: safety_calculator: SafetyCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 mttc_data = safety_calculator.mttc_data if not mttc_data: logger.warning("Cannot generate MTTC chart: empty data") return None # 创建DataFrame df = pd.DataFrame(mttc_data) # 获取阈值 thresholds = get_metric_thresholds(safety_calculator, 'MTTC') min_threshold = thresholds.get('min') max_threshold = thresholds.get('max') # 检测超阈值事件 unsafe_events = [] if min_threshold is not None: # 对于MTTC，小于最小阈值视为不安全 unsafe_condition = df['MTTC'] < min_threshold event_groups = (unsafe_condition != unsafe_condition.shift()).cumsum() for _, group in df[unsafe_condition].groupby(event_groups): if len(group) >= 2: # 至少2帧才算一次事件 start_time = group['simTime'].iloc[0] end_time = group['simTime'].iloc[-1] duration = end_time - start_time if duration >= 0.1: # 只记录持续时间超过0.1秒的事件 unsafe_events.append({ 'start_time': start_time, 'end_time': end_time, 'start_frame': group['simFrame'].iloc[0], 'end_frame': group['simFrame'].iloc[-1], 'duration': duration, 'min_mttc': group['MTTC'].min() }) # 创建图表 plt.figure(figsize=(12, 6)) plt.plot(df['simTime'], df['MTTC'], 'g-', label='MTTC') # 添加阈值线 if min_threshold is not None: plt.axhline(y=min_threshold, color='r', linestyle='--', label=f'Min Threshold ({min_threshold}s)') if max_threshold is not None: plt.axhline(y=max_threshold, color='g', linestyle='--', label=f'Max Threshold ({max_threshold})') # 添加橙色背景标识不安全事件 for idx, event in enumerate(unsafe_events): label = 'Unsafe MTTC Event' if idx == 0 else None plt.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) plt.xlabel('Time (s)') plt.ylabel('MTTC (s)') plt.title('Modified Time To Collision (MTTC) Trend') plt.grid(True) plt.legend() # 保存图表 chart_filename = os.path.join(output_dir, f"mttc_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() # 保存CSV数据，包含阈值信息 csv_filename = os.path.join(output_dir, f"mttc_data.csv") df_csv = df.copy() df_csv['min_threshold'] = min_threshold df_csv['max_threshold'] = max_threshold df_csv.to_csv(csv_filename, index=False) # 记录不安全事件信息 if unsafe_events: logger.info(f"检测到 {len(unsafe_events)} 个MTTC不安全事件") for i, event in enumerate(unsafe_events): logger.info( f"MTTC不安全事件 #{i + 1}: 开始时间={event['start_time']:.2f}s, 结束时间={event['end_time']:.2f}s, 持续时间={event['duration']:.2f}s, 最小MTTC={event['min_mttc']:.2f}s") logger.info(f"MTTC chart saved to: {chart_filename}") logger.info(f"MTTC data saved to: {csv_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate MTTC chart: {str(e)}", exc_info=True) return None def generate_thw_chart(safety_calculator, output_dir: str) -> Optional[str]: """ Generate THW metric chart with orange background for unsafe events Args: safety_calculator: SafetyCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 thw_data = safety_calculator.thw_data if not thw_data: logger.warning("Cannot generate THW chart: empty data") return None # 创建DataFrame df = pd.DataFrame(thw_data) # 获取阈值 thresholds = get_metric_thresholds(safety_calculator, 'THW') min_threshold = thresholds.get('min') max_threshold = thresholds.get('max') # 检测超阈值事件 unsafe_events = [] if min_threshold is not None: # 对于THW，小于最小阈值视为不安全 unsafe_condition = df['THW'] < min_threshold event_groups = (unsafe_condition != unsafe_condition.shift()).cumsum() for _, group in df[unsafe_condition].groupby(event_groups): if len(group) >= 2: # 至少2帧才算一次事件 start_time = group['simTime'].iloc[0] end_time = group['simTime'].iloc[-1] duration = end_time - start_time if duration >= 0.1: # 只记录持续时间超过0.1秒的事件 unsafe_events.append({ 'start_time': start_time, 'end_time': end_time, 'start_frame': group['simFrame'].iloc[0], 'end_frame': group['simFrame'].iloc[-1], 'duration': duration, 'min_thw': group['THW'].min() }) # 创建图表 plt.figure(figsize=(12, 10)) plt.plot(df['simTime'], df['THW'], 'c-', label='THW') # 添加阈值线 if min_threshold is not None: plt.axhline(y=min_threshold, color='r', linestyle='--', label=f'Min Threshold ({min_threshold}s)') if max_threshold is not None: plt.axhline(y=max_threshold, color='g', linestyle='--', label=f'Max Threshold ({max_threshold})') # 添加橙色背景标识不安全事件 for idx, event in enumerate(unsafe_events): label = 'Unsafe THW Event' if idx == 0 else None plt.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) plt.xlabel('Time (s)') plt.ylabel('THW (s)') plt.title('Time Headway (THW) Trend') plt.grid(True) plt.legend() # 保存图表 chart_filename = os.path.join(output_dir, f"thw_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() # 保存CSV数据，包含阈值信息 csv_filename = os.path.join(output_dir, f"thw_data.csv") df_csv = df.copy() df_csv['min_threshold'] = min_threshold df_csv['max_threshold'] = max_threshold df_csv.to_csv(csv_filename, index=False) # 记录不安全事件信息 if unsafe_events: logger.info(f"检测到 {len(unsafe_events)} 个THW不安全事件") for i, event in enumerate(unsafe_events): logger.info( f"THW不安全事件 #{i + 1}: 开始时间={event['start_time']:.2f}s, 结束时间={event['end_time']:.2f}s, 持续时间={event['duration']:.2f}s, 最小THW={event['min_thw']:.2f}s") logger.info(f"THW chart saved to: {chart_filename}") logger.info(f"THW data saved to: {csv_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate THW chart: {str(e)}", exc_info=True) return None def generate_lonsd_chart(safety_calculator, output_dir: str) -> Optional[str]: """ Generate Longitudinal Safe Distance metric chart Args: safety_calculator: SafetyCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 lonsd_data = safety_calculator.lonsd_data if not lonsd_data: logger.warning("Cannot generate Longitudinal Safe Distance chart: empty data") return None # 创建DataFrame df = pd.DataFrame(lonsd_data) # 获取阈值 thresholds = get_metric_thresholds(safety_calculator, 'LonSD') min_threshold = thresholds.get('min') max_threshold = thresholds.get('max') # 创建图表 plt.figure(figsize=(12, 6)) plt.plot(df['simTime'], df['LonSD'], 'm-', label='Longitudinal Safe Distance') # 添加阈值线 if min_threshold is not None: plt.axhline(y=min_threshold, color='r', linestyle='--', label=f'Min Threshold ({min_threshold}m)') if max_threshold is not None: plt.axhline(y=max_threshold, color='g', linestyle='--', label=f'Max Threshold ({max_threshold}m)') plt.xlabel('Time (s)') plt.ylabel('Distance (m)') plt.title('Longitudinal Safe Distance (LonSD) Trend') plt.grid(True) plt.legend() # 保存图表 chart_filename = os.path.join(output_dir, f"lonsd_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() # 保存CSV数据，包含阈值信息 csv_filename = os.path.join(output_dir, f"lonsd_data.csv") df_csv = df.copy() df_csv['min_threshold'] = min_threshold df_csv['max_threshold'] = max_threshold df_csv.to_csv(csv_filename, index=False) logger.info(f"Longitudinal Safe Distance chart saved to: {chart_filename}") logger.info(f"Longitudinal Safe Distance data saved to: {csv_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate Longitudinal Safe Distance chart: {str(e)}", exc_info=True) return None def generate_latsd_chart(safety_calculator, output_dir: str) -> Optional[str]: """ Generate Lateral Safe Distance metric chart with orange background for unsafe events Args: safety_calculator: SafetyCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 latsd_data = safety_calculator.latsd_data if not latsd_data: logger.warning("Cannot generate Lateral Safe Distance chart: empty data") return None # 创建DataFrame df = pd.DataFrame(latsd_data) # 获取阈值 thresholds = get_metric_thresholds(safety_calculator, 'LatSD') min_threshold = thresholds.get('min') max_threshold = thresholds.get('max') # 检测超阈值事件 unsafe_events = [] if min_threshold is not None: # 对于LatSD，小于最小阈值视为不安全 unsafe_condition = df['LatSD'] < min_threshold event_groups = (unsafe_condition != unsafe_condition.shift()).cumsum() for _, group in df[unsafe_condition].groupby(event_groups): if len(group) >= 2: # 至少2帧才算一次事件 start_time = group['simTime'].iloc[0] end_time = group['simTime'].iloc[-1] duration = end_time - start_time if duration >= 0.1: # 只记录持续时间超过0.1秒的事件 unsafe_events.append({ 'start_time': start_time, 'end_time': end_time, 'start_frame': group['simFrame'].iloc[0], 'end_frame': group['simFrame'].iloc[-1], 'duration': duration, 'min_latsd': group['LatSD'].min() }) # 创建图表 plt.figure(figsize=(12, 6)) plt.plot(df['simTime'], df['LatSD'], 'y-', label='Lateral Safe Distance') # 添加阈值线 if min_threshold is not None: plt.axhline(y=min_threshold, color='r', linestyle='--', label=f'Min Threshold ({min_threshold}m)') if max_threshold is not None: plt.axhline(y=max_threshold, color='g', linestyle='--', label=f'Max Threshold ({max_threshold}m)') # 添加橙色背景标识不安全事件 for idx, event in enumerate(unsafe_events): label = 'Unsafe LatSD Event' if idx == 0 else None plt.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) plt.xlabel('Time (s)') plt.ylabel('Distance (m)') plt.title('Lateral Safe Distance (LatSD) Trend') plt.grid(True) plt.legend() # 保存图表 chart_filename = os.path.join(output_dir, f"latsd_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() # 保存CSV数据，包含阈值信息 csv_filename = os.path.join(output_dir, f"latsd_data.csv") df_csv = df.copy() df_csv['min_threshold'] = min_threshold df_csv['max_threshold'] = max_threshold df_csv.to_csv(csv_filename, index=False) # 记录不安全事件信息 if unsafe_events: logger.info(f"检测到 {len(unsafe_events)} 个LatSD不安全事件") for i, event in enumerate(unsafe_events): logger.info( f"LatSD不安全事件 #{i + 1}: 开始时间={event['start_time']:.2f}s, 结束时间={event['end_time']:.2f}s, 持续时间={event['duration']:.2f}s, 最小LatSD={event['min_latsd']:.2f}m") logger.info(f"Lateral Safe Distance chart saved to: {chart_filename}") logger.info(f"Lateral Safe Distance data saved to: {csv_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate Lateral Safe Distance chart: {str(e)}", exc_info=True) return None def generate_btn_chart(safety_calculator, output_dir: str) -> Optional[str]: """ Generate Brake Threat Number metric chart with orange background for unsafe events Args: safety_calculator: SafetyCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 btn_data = safety_calculator.btn_data if not btn_data: logger.warning("Cannot generate Brake Threat Number chart: empty data") return None # 创建DataFrame df = pd.DataFrame(btn_data) # 获取阈值 thresholds = get_metric_thresholds(safety_calculator, 'BTN') min_threshold = thresholds.get('min') max_threshold = thresholds.get('max') # 检测超阈值事件 unsafe_events = [] if max_threshold is not None: # 对于BTN，大于最大阈值视为不安全 unsafe_condition = df['BTN'] > max_threshold event_groups = (unsafe_condition != unsafe_condition.shift()).cumsum() for _, group in df[unsafe_condition].groupby(event_groups): if len(group) >= 2: # 至少2帧才算一次事件 start_time = group['simTime'].iloc[0] end_time = group['simTime'].iloc[-1] duration = end_time - start_time if duration >= 0.1: # 只记录持续时间超过0.1秒的事件 unsafe_events.append({ 'start_time': start_time, 'end_time': end_time, 'start_frame': group['simFrame'].iloc[0], 'end_frame': group['simFrame'].iloc[-1], 'duration': duration, 'max_btn': group['BTN'].max() }) # 创建图表 plt.figure(figsize=(12, 6)) plt.plot(df['simTime'], df['BTN'], 'r-', label='Brake Threat Number') # 添加阈值线 if min_threshold is not None: plt.axhline(y=min_threshold, color='r', linestyle='--', label=f'Min Threshold ({min_threshold})') if max_threshold is not None: plt.axhline(y=max_threshold, color='g', linestyle='--', label=f'Max Threshold ({max_threshold})') # 添加橙色背景标识不安全事件 for idx, event in enumerate(unsafe_events): label = 'Unsafe BTN Event' if idx == 0 else None plt.axvspan(event['start_time'], event['end_time'], alpha=0.3, color='orange', label=label) plt.xlabel('Time (s)') plt.ylabel('BTN') plt.title('Brake Threat Number (BTN) Trend') plt.grid(True) plt.legend() # 保存图表 chart_filename = os.path.join(output_dir, f"btn_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() # 保存CSV数据，包含阈值信息 csv_filename = os.path.join(output_dir, f"btn_data.csv") df_csv = df.copy() df_csv['min_threshold'] = min_threshold df_csv['max_threshold'] = max_threshold df_csv.to_csv(csv_filename, index=False) # 记录不安全事件信息 if unsafe_events: logger.info(f"检测到 {len(unsafe_events)} 个BTN不安全事件") for i, event in enumerate(unsafe_events): logger.info( f"BTN不安全事件 #{i + 1}: 开始时间={event['start_time']:.2f}s, 结束时间={event['end_time']:.2f}s, 持续时间={event['duration']:.2f}s, 最大BTN={event['max_btn']:.2f}") logger.info(f"Brake Threat Number chart saved to: {chart_filename}") logger.info(f"Brake Threat Number data saved to: {csv_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate Brake Threat Number chart: {str(e)}", exc_info=True) return None def generate_collision_risk_chart(safety_calculator, output_dir: str) -> Optional[str]: """ Generate Collision Risk metric chart Args: safety_calculator: SafetyCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 risk_data = safety_calculator.collision_risk_data if not risk_data: logger.warning("Cannot generate Collision Risk chart: empty data") return None # 创建DataFrame df = pd.DataFrame(risk_data) # 获取阈值 thresholds = get_metric_thresholds(safety_calculator, 'collisionRisk') min_threshold = thresholds.get('min') max_threshold = thresholds.get('max') # 创建图表 plt.figure(figsize=(12, 6)) plt.plot(df['simTime'], df['collisionRisk'], 'r-', label='Collision Risk') # 添加阈值线 if min_threshold is not None: plt.axhline(y=min_threshold, color='r', linestyle='--', label=f'Min Threshold ({min_threshold}%)') if max_threshold is not None: plt.axhline(y=max_threshold, color='g', linestyle='--', label=f'Max Threshold ({max_threshold}%)') plt.xlabel('Time (s)') plt.ylabel('Risk Value (%)') plt.title('Collision Risk (collisionRisk) Trend') plt.grid(True) plt.legend() # 保存图表 chart_filename = os.path.join(output_dir, f"collision_risk_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() # 保存CSV数据，包含阈值信息 csv_filename = os.path.join(output_dir, f"collisionrisk_data.csv") df_csv = df.copy() df_csv['min_threshold'] = min_threshold df_csv['max_threshold'] = max_threshold df_csv.to_csv(csv_filename, index=False) logger.info(f"Collision Risk chart saved to: {chart_filename}") logger.info(f"Collision Risk data saved to: {csv_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate Collision Risk chart: {str(e)}", exc_info=True) return None def generate_collision_severity_chart(safety_calculator, output_dir: str) -> Optional[str]: """ Generate Collision Severity metric chart Args: safety_calculator: SafetyCalculator instance output_dir: Output directory Returns: str: Chart file path, or None if generation fails """ logger = LogManager().get_logger() try: # 获取数据 severity_data = safety_calculator.collision_severity_data if not severity_data: logger.warning("Cannot generate Collision Severity chart: empty data") return None # 创建DataFrame df = pd.DataFrame(severity_data) # 获取阈值 thresholds = get_metric_thresholds(safety_calculator, 'collisionSeverity') min_threshold = thresholds.get('min') max_threshold = thresholds.get('max') # 创建图表 plt.figure(figsize=(12, 6)) plt.plot(df['simTime'], df['collisionSeverity'], 'r-', label='Collision Severity') # 添加阈值线 if min_threshold is not None: plt.axhline(y=min_threshold, color='r', linestyle='--', label=f'Min Threshold ({min_threshold}%)') if max_threshold is not None: plt.axhline(y=max_threshold, color='g', linestyle='--', label=f'Max Threshold ({max_threshold}%)') plt.xlabel('Time (s)') plt.ylabel('Severity (%)') plt.title('Collision Severity (collisionSeverity) Trend') plt.grid(True) plt.legend() # 保存图表 chart_filename = os.path.join(output_dir, f"collision_severity_chart.png") plt.savefig(chart_filename, dpi=300) plt.close() # 保存CSV数据，包含阈值信息 csv_filename = os.path.join(output_dir, f"collisionseverity_data.csv") df_csv = df.copy() df_csv['min_threshold'] = min_threshold df_csv['max_threshold'] = max_threshold df_csv.to_csv(csv_filename, index=False) logger.info(f"Collision Severity chart saved to: {chart_filename}") logger.info(f"Collision Severity data saved to: {csv_filename}") return chart_filename except Exception as e: logger.error(f"Failed to generate Collision Severity chart: {str(e)}", exc_info=True) return None def generate_traffic_chart_data(traffic_calculator, metric_name: str, output_dir: Optional[str] = None) -> Optional[ str]: """Generate chart data for traffic metrics""" # 待实现 return None def calculate_distance(ego_df, correctwarning): """计算预警距离""" dist = ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]['relative_dist'] return dist def calculate_relative_speed(ego_df, correctwarning): """计算相对速度""" return ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]['composite_v'] # 使用function.py中已实现的scenario_sign_dict from modules.metric.function import scenario_sign_dict if __name__ == "__main__": # 测试代码 print("Metrics visualization utilities loaded.")