|
|
@@ -0,0 +1,1899 @@
|
|
|
+#!/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
|
|
|
+"""
|
|
|
+
|
|
|
+import os
|
|
|
+import numpy as np
|
|
|
+import pandas as pd
|
|
|
+import matplotlib
|
|
|
+matplotlib.use('Agg') # 使用非图形界面的后端
|
|
|
+import matplotlib.pyplot as plt
|
|
|
+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.getcwd()
|
|
|
+
|
|
|
+ # 根据指标名称选择不同的图表生成方法
|
|
|
+ if metric_name.lower() == 'latestwarningdistance_ttc_lst':
|
|
|
+ return generate_warning_ttc_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_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()
|
|
|
+ scenario_name = function_calculator.data.function_config["function"]["scenario"]["name"]
|
|
|
+ correctwarning = scenario_sign_dict[scenario_name]
|
|
|
+
|
|
|
+ warning_dist = calculate_distance(ego_df, correctwarning)
|
|
|
+ warning_speed = calculate_relative_speed(ego_df, correctwarning)
|
|
|
+
|
|
|
+ if warning_dist.empty:
|
|
|
+ logger.warning("Cannot generate TTC warning chart: empty data")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 生成时间戳
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+
|
|
|
+ # 保存 CSV 数据
|
|
|
+ csv_filename = os.path.join(output_dir, f"warning_ttc_data_{timestamp}.csv")
|
|
|
+ df_csv = pd.DataFrame({
|
|
|
+ 'simTime': ego_df['simTime'],
|
|
|
+ 'warning_distance': warning_dist,
|
|
|
+ 'warning_speed': warning_speed,
|
|
|
+ 'ttc': warning_dist / warning_speed
|
|
|
+ })
|
|
|
+ df_csv.to_csv(csv_filename, index=False)
|
|
|
+ logger.info(f"Warning TTC 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')
|
|
|
+ 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"warning_ttc_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ logger.info(f"Warning TTC chart saved to: {chart_filename}")
|
|
|
+ return chart_filename
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f"Failed to generate warning TTC 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)
|
|
|
+ elif metric_name.lower() == 'vdv':
|
|
|
+ return generate_vdv_chart(comfort_calculator, output_dir)
|
|
|
+ elif metric_name.lower() == 'ava_vav':
|
|
|
+ return generate_ava_vav_chart(comfort_calculator, output_dir)
|
|
|
+ elif metric_name.lower() == 'msdv':
|
|
|
+ return generate_msdv_chart(comfort_calculator, output_dir)
|
|
|
+ elif metric_name.lower() == 'motionsickness':
|
|
|
+ return generate_motion_sickness_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
|
|
|
+
|
|
|
+ # 生成时间戳
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+
|
|
|
+ # 保存 CSV 数据（第一步）
|
|
|
+ csv_filename = os.path.join(output_dir, f"shake_data_{timestamp}.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_{timestamp}.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
|
|
|
+
|
|
|
+ # 生成时间戳
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+
|
|
|
+ # 保存 CSV 数据（第一步）
|
|
|
+ csv_filename = os.path.join(output_dir, f"zigzag_data_{timestamp}.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_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ logger.info(f"Zigzag chart saved to: {chart_filename}")
|
|
|
+ return chart_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
|
|
|
+
|
|
|
+ # 生成时间戳
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+
|
|
|
+ # 保存 CSV 数据（第一步）
|
|
|
+ csv_filename = os.path.join(output_dir, f"cadence_data_{timestamp}.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_{timestamp}.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
|
|
|
+
|
|
|
+ # 生成时间戳
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+
|
|
|
+ # 保存 CSV 数据（第一步）
|
|
|
+ csv_filename = os.path.join(output_dir, f"slam_brake_data_{timestamp}.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_{timestamp}.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
|
|
|
+
|
|
|
+ # 生成时间戳
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+
|
|
|
+ # 保存 CSV 数据（第一步）
|
|
|
+ csv_filename = os.path.join(output_dir, f"slam_accel_data_{timestamp}.csv")
|
|
|
+ df_csv = pd.DataFrame({
|
|
|
+ 'simTime': df['simTime'],
|
|
|
+ 'lon_acc': df['lon_acc'],
|
|
|
+ 'v': df['v'],
|
|
|
+ 'min_threshold': 0.0,
|
|
|
+ 'max_threshold': df.get('ip_acc', pd.Series([None]*len(df)))
|
|
|
+ })
|
|
|
+ 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:
|
|
|
+ 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('Longitudinal 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_{timestamp}.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 (SafetyCalculator or ComfortCalculator)
|
|
|
+ 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'
|
|
|
+ else:
|
|
|
+ logger.warning(f"无法找到{metric_name}的配置信息")
|
|
|
+ return thresholds
|
|
|
+ else:
|
|
|
+ logger.warning(f"计算器没有data_processed属性")
|
|
|
+ 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')
|
|
|
+
|
|
|
+ # 生成时间戳
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+
|
|
|
+ # 保存 CSV 数据（第一步）
|
|
|
+ csv_filename = os.path.join(output_dir, f"ttc_data_{timestamp}.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_{timestamp}.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()
|
|
|
+
|
|
|
+ # 保存图表
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+ chart_filename = os.path.join(output_dir, f"mttc_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ # 保存CSV数据，包含阈值信息
|
|
|
+ csv_filename = os.path.join(output_dir, f"mttc_data_{timestamp}.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, 6))
|
|
|
+ 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()
|
|
|
+
|
|
|
+ # 保存图表
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+ chart_filename = os.path.join(output_dir, f"thw_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ # 保存CSV数据，包含阈值信息
|
|
|
+ csv_filename = os.path.join(output_dir, f"thw_data_{timestamp}.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()
|
|
|
+
|
|
|
+ # 保存图表
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+ chart_filename = os.path.join(output_dir, f"lonsd_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ # 保存CSV数据，包含阈值信息
|
|
|
+ csv_filename = os.path.join(output_dir, f"lonsd_data_{timestamp}.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()
|
|
|
+
|
|
|
+ # 保存图表
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+ chart_filename = os.path.join(output_dir, f"latsd_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ # 保存CSV数据，包含阈值信息
|
|
|
+ csv_filename = os.path.join(output_dir, f"latsd_data_{timestamp}.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()
|
|
|
+
|
|
|
+ # 保存图表
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+ chart_filename = os.path.join(output_dir, f"btn_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ # 保存CSV数据，包含阈值信息
|
|
|
+ csv_filename = os.path.join(output_dir, f"btn_data_{timestamp}.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()
|
|
|
+
|
|
|
+ # 保存图表
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+ chart_filename = os.path.join(output_dir, f"collision_risk_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ # 保存CSV数据，包含阈值信息
|
|
|
+ csv_filename = os.path.join(output_dir, f"collisionrisk_data_{timestamp}.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()
|
|
|
+
|
|
|
+ # 保存图表
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+ chart_filename = os.path.join(output_dir, f"collision_severity_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ # 保存CSV数据，包含阈值信息
|
|
|
+ csv_filename = os.path.join(output_dir, f"collisionseverity_data_{timestamp}.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_vdv_chart(comfort_calculator, output_dir: str) -> Optional[str]:
|
|
|
+ """
|
|
|
+ Generate VDV (Vibration Dose Value) metric chart with data saved to CSV first.
|
|
|
+ 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()
|
|
|
+ vdv_value = comfort_calculator.calculated_value.get('vdv', 0)
|
|
|
+
|
|
|
+ if df.empty:
|
|
|
+ logger.warning("Cannot generate VDV chart: empty data")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 确保有必要的列
|
|
|
+ if 'accelX' not in df.columns or 'accelY' not in df.columns:
|
|
|
+ logger.warning("Missing required columns for VDV chart")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 获取阈值
|
|
|
+ thresholds = get_metric_thresholds(comfort_calculator, 'vdv')
|
|
|
+ min_threshold = thresholds.get('min')
|
|
|
+ max_threshold = thresholds.get('max')
|
|
|
+
|
|
|
+ # 将东北天坐标系下的加速度转换为车身坐标系下的加速度
|
|
|
+ if 'posH' not in df.columns:
|
|
|
+ logger.warning("Missing heading angle data for coordinate transformation")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 车身坐标系：X轴指向车头，Y轴指向车辆左侧，Z轴指向车顶
|
|
|
+ df['posH_rad'] = np.radians(df['posH'])
|
|
|
+
|
|
|
+ # 转换加速度到车身坐标系
|
|
|
+ df['a_x_body'] = df['accelX'] * np.sin(df['posH_rad']) + df['accelY'] * np.cos(df['posH_rad'])
|
|
|
+ df['a_y_body'] = df['accelX'] * np.cos(df['posH_rad']) - df['accelY'] * np.sin(df['posH_rad'])
|
|
|
+ df['a_z_body'] = df['accelZ'] if 'accelZ' in df.columns else pd.Series(np.zeros(len(df)))
|
|
|
+
|
|
|
+ # 生成时间戳
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+
|
|
|
+ # 保存 CSV 数据（第一步）
|
|
|
+ csv_filename = os.path.join(output_dir, f"vdv_data_{timestamp}.csv")
|
|
|
+ df_csv = pd.DataFrame({
|
|
|
+ 'simTime': df['simTime'],
|
|
|
+ 'a_x_body': df['a_x_body'],
|
|
|
+ 'a_y_body': df['a_y_body'],
|
|
|
+ 'a_z_body': df['a_z_body'],
|
|
|
+ 'v': df['v'],
|
|
|
+ 'min_threshold': min_threshold,
|
|
|
+ 'max_threshold': max_threshold,
|
|
|
+ 'vdv_value': vdv_value
|
|
|
+ })
|
|
|
+ df_csv.to_csv(csv_filename, index=False)
|
|
|
+ logger.info(f"VDV data saved to: {csv_filename}")
|
|
|
+
|
|
|
+ # 第二步：从 CSV 读取（可验证保存数据无误）
|
|
|
+ df = pd.read_csv(csv_filename)
|
|
|
+
|
|
|
+ # 创建图表（第三步）
|
|
|
+ plt.figure(figsize=(12, 8))
|
|
|
+
|
|
|
+ # 绘制三轴加速度
|
|
|
+ plt.subplot(3, 1, 1)
|
|
|
+ plt.plot(df['simTime'], df['a_x_body'], 'r-', label='X-axis Acceleration')
|
|
|
+
|
|
|
+ # 添加阈值线
|
|
|
+ if 'min_threshold' in df.columns and df['min_threshold'].iloc[0] is not None:
|
|
|
+ min_threshold = df['min_threshold'].iloc[0]
|
|
|
+ plt.axhline(y=min_threshold, color='r', linestyle=':', label=f'Min Threshold ({min_threshold})')
|
|
|
+ if 'max_threshold' in df.columns and df['max_threshold'].iloc[0] is not None:
|
|
|
+ max_threshold = df['max_threshold'].iloc[0]
|
|
|
+ plt.axhline(y=max_threshold, color='g', linestyle=':', label=f'Max Threshold ({max_threshold})')
|
|
|
+
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Acceleration (m/s²)')
|
|
|
+ plt.title('Body X-axis Acceleration (Longitudinal)')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.subplot(3, 1, 2)
|
|
|
+ plt.plot(df['simTime'], df['a_y_body'], 'g-', label='Y-axis Acceleration')
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Acceleration (m/s²)')
|
|
|
+ plt.title('Body Y-axis Acceleration (Lateral)')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.subplot(3, 1, 3)
|
|
|
+ plt.plot(df['simTime'], df['a_z_body'], 'b-', label='Z-axis Acceleration')
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Acceleration (m/s²)')
|
|
|
+ vdv_value = df['vdv_value'].iloc[0] if 'vdv_value' in df.columns else 0
|
|
|
+ plt.title(f'Body Z-axis Acceleration (Vertical) - VDV value: {vdv_value:.4f}')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.tight_layout()
|
|
|
+
|
|
|
+ # 保存图像
|
|
|
+ chart_filename = os.path.join(output_dir, f"vdv_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ logger.info(f"VDV chart saved to: {chart_filename}")
|
|
|
+ return chart_filename
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f"Failed to generate VDV chart: {str(e)}", exc_info=True)
|
|
|
+ return None
|
|
|
+
|
|
|
+def generate_ava_vav_chart(comfort_calculator, output_dir: str) -> Optional[str]:
|
|
|
+ """
|
|
|
+ Generate AVA_VAV (Average Vibration Acceleration Value) metric chart with data saved to CSV first.
|
|
|
+ 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()
|
|
|
+ ava_vav_value = comfort_calculator.calculated_value.get('ava_vav', 0)
|
|
|
+
|
|
|
+ if df.empty:
|
|
|
+ logger.warning("Cannot generate AVA_VAV chart: empty data")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 确保有必要的列
|
|
|
+ if 'accelX' not in df.columns or 'accelY' not in df.columns:
|
|
|
+ logger.warning("Missing required columns for AVA_VAV chart")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 获取阈值
|
|
|
+ thresholds = get_metric_thresholds(comfort_calculator, 'ava_vav')
|
|
|
+ min_threshold = thresholds.get('min')
|
|
|
+ max_threshold = thresholds.get('max')
|
|
|
+
|
|
|
+ # 将东北天坐标系下的加速度转换为车身坐标系下的加速度
|
|
|
+ if 'posH' not in df.columns:
|
|
|
+ logger.warning("Missing heading angle data for coordinate transformation")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 车身坐标系：X轴指向车头，Y轴指向车辆左侧，Z轴指向车顶
|
|
|
+ df['posH_rad'] = np.radians(df['posH'])
|
|
|
+
|
|
|
+ # 转换加速度到车身坐标系
|
|
|
+ df['a_x_body'] = df['accelX'] * np.sin(df['posH_rad']) + df['accelY'] * np.cos(df['posH_rad'])
|
|
|
+ df['a_y_body'] = df['accelX'] * np.cos(df['posH_rad']) - df['accelY'] * np.sin(df['posH_rad'])
|
|
|
+ df['a_z_body'] = df['accelZ'] if 'accelZ' in df.columns else pd.Series(np.zeros(len(df)))
|
|
|
+
|
|
|
+ # 角速度数据
|
|
|
+ df['omega_roll'] = df['rollRate'] if 'rollRate' in df.columns else pd.Series(np.zeros(len(df)))
|
|
|
+ df['omega_pitch'] = df['pitchRate'] if 'pitchRate' in df.columns else pd.Series(np.zeros(len(df)))
|
|
|
+ df['omega_yaw'] = df['speedH'] # 使用航向角速度作为偏航角速度
|
|
|
+
|
|
|
+ # 生成时间戳
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+
|
|
|
+ # 保存 CSV 数据（第一步）
|
|
|
+ csv_filename = os.path.join(output_dir, f"ava_vav_data_{timestamp}.csv")
|
|
|
+ df_csv = pd.DataFrame({
|
|
|
+ 'simTime': df['simTime'],
|
|
|
+ 'a_x_body': df['a_x_body'],
|
|
|
+ 'a_y_body': df['a_y_body'],
|
|
|
+ 'a_z_body': df['a_z_body'],
|
|
|
+ 'omega_roll': df['omega_roll'],
|
|
|
+ 'omega_pitch': df['omega_pitch'],
|
|
|
+ 'omega_yaw': df['omega_yaw'],
|
|
|
+ 'min_threshold': min_threshold,
|
|
|
+ 'max_threshold': max_threshold,
|
|
|
+ 'ava_vav_value': ava_vav_value
|
|
|
+ })
|
|
|
+ df_csv.to_csv(csv_filename, index=False)
|
|
|
+ logger.info(f"AVA_VAV data saved to: {csv_filename}")
|
|
|
+
|
|
|
+ # 第二步：从 CSV 读取（可验证保存数据无误）
|
|
|
+ df = pd.read_csv(csv_filename)
|
|
|
+
|
|
|
+ # 创建图表（第三步）
|
|
|
+ plt.figure(figsize=(12, 10))
|
|
|
+
|
|
|
+ # 绘制三轴加速度
|
|
|
+ plt.subplot(3, 2, 1)
|
|
|
+ plt.plot(df['simTime'], df['a_x_body'], 'r-', label='X-axis Acceleration')
|
|
|
+
|
|
|
+ # 添加阈值线
|
|
|
+ if 'min_threshold' in df.columns and df['min_threshold'].iloc[0] is not None:
|
|
|
+ min_threshold = df['min_threshold'].iloc[0]
|
|
|
+ plt.axhline(y=min_threshold, color='r', linestyle=':', label=f'Min Threshold ({min_threshold})')
|
|
|
+ if 'max_threshold' in df.columns and df['max_threshold'].iloc[0] is not None:
|
|
|
+ max_threshold = df['max_threshold'].iloc[0]
|
|
|
+ plt.axhline(y=max_threshold, color='g', linestyle=':', label=f'Max Threshold ({max_threshold})')
|
|
|
+
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Acceleration (m/s²)')
|
|
|
+ plt.title('Body X-axis Acceleration (Longitudinal)')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.subplot(3, 2, 3)
|
|
|
+ plt.plot(df['simTime'], df['a_y_body'], 'g-', label='Y-axis Acceleration')
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Acceleration (m/s²)')
|
|
|
+ plt.title('Body Y-axis Acceleration (Lateral)')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.subplot(3, 2, 5)
|
|
|
+ plt.plot(df['simTime'], df['a_z_body'], 'b-', label='Z-axis Acceleration')
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Acceleration (m/s²)')
|
|
|
+ plt.title('Body Z-axis Acceleration (Vertical)')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ # 绘制三轴角速度
|
|
|
+ plt.subplot(3, 2, 2)
|
|
|
+ plt.plot(df['simTime'], df['omega_roll'], 'r-', label='Roll Rate')
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Angular Velocity (deg/s)')
|
|
|
+ plt.title('Roll Rate')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.subplot(3, 2, 4)
|
|
|
+ plt.plot(df['simTime'], df['omega_pitch'], 'g-', label='Pitch Rate')
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Angular Velocity (deg/s)')
|
|
|
+ plt.title('Pitch Rate')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.subplot(3, 2, 6)
|
|
|
+ plt.plot(df['simTime'], df['omega_yaw'], 'b-', label='Yaw Rate')
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Angular Velocity (deg/s)')
|
|
|
+ ava_vav_value = df['ava_vav_value'].iloc[0] if 'ava_vav_value' in df.columns else 0
|
|
|
+ plt.title(f'Yaw Rate - AVA_VAV value: {ava_vav_value:.4f}')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.tight_layout()
|
|
|
+
|
|
|
+ # 保存图像
|
|
|
+ chart_filename = os.path.join(output_dir, f"ava_vav_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ logger.info(f"AVA_VAV chart saved to: {chart_filename}")
|
|
|
+ return chart_filename
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f"Failed to generate AVA_VAV chart: {str(e)}", exc_info=True)
|
|
|
+ return None
|
|
|
+
|
|
|
+def generate_msdv_chart(comfort_calculator, output_dir: str) -> Optional[str]:
|
|
|
+ """
|
|
|
+ Generate MSDV (Motion Sickness Dose Value) metric chart with data saved to CSV first.
|
|
|
+ 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()
|
|
|
+ msdv_value = comfort_calculator.calculated_value.get('msdv', 0)
|
|
|
+ motion_sickness_prob = comfort_calculator.calculated_value.get('motionSickness', 0)
|
|
|
+
|
|
|
+ if df.empty:
|
|
|
+ logger.warning("Cannot generate MSDV chart: empty data")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 确保有必要的列
|
|
|
+ if 'accelX' not in df.columns or 'accelY' not in df.columns:
|
|
|
+ logger.warning("Missing required columns for MSDV chart")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 获取阈值
|
|
|
+ thresholds = get_metric_thresholds(comfort_calculator, 'msdv')
|
|
|
+ min_threshold = thresholds.get('min')
|
|
|
+ max_threshold = thresholds.get('max')
|
|
|
+
|
|
|
+ # 将东北天坐标系下的加速度转换为车身坐标系下的加速度
|
|
|
+ if 'posH' not in df.columns:
|
|
|
+ logger.warning("Missing heading angle data for coordinate transformation")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 车身坐标系：X轴指向车头，Y轴指向车辆左侧，Z轴指向车顶
|
|
|
+ df['posH_rad'] = np.radians(df['posH'])
|
|
|
+
|
|
|
+ # 转换加速度到车身坐标系
|
|
|
+ df['a_x_body'] = df['accelX'] * np.sin(df['posH_rad']) + df['accelY'] * np.cos(df['posH_rad'])
|
|
|
+ df['a_y_body'] = df['accelX'] * np.cos(df['posH_rad']) - df['accelY'] * np.sin(df['posH_rad'])
|
|
|
+ df['a_z_body'] = df['accelZ'] if 'accelZ' in df.columns else pd.Series(np.zeros(len(df)))
|
|
|
+
|
|
|
+ # 生成时间戳
|
|
|
+ import datetime
|
|
|
+ timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
+
|
|
|
+ # 保存 CSV 数据（第一步）
|
|
|
+ csv_filename = os.path.join(output_dir, f"msdv_data_{timestamp}.csv")
|
|
|
+ df_csv = pd.DataFrame({
|
|
|
+ 'simTime': df['simTime'],
|
|
|
+ 'a_x_body': df['a_x_body'],
|
|
|
+ 'a_y_body': df['a_y_body'],
|
|
|
+ 'a_z_body': df['a_z_body'],
|
|
|
+ 'v': df['v'],
|
|
|
+ 'min_threshold': min_threshold,
|
|
|
+ 'max_threshold': max_threshold,
|
|
|
+ 'msdv_value': msdv_value,
|
|
|
+ 'motion_sickness_prob': motion_sickness_prob
|
|
|
+ })
|
|
|
+ df_csv.to_csv(csv_filename, index=False)
|
|
|
+ logger.info(f"MSDV data saved to: {csv_filename}")
|
|
|
+
|
|
|
+ # 第二步：从 CSV 读取（可验证保存数据无误）
|
|
|
+ df = pd.read_csv(csv_filename)
|
|
|
+
|
|
|
+ # 创建图表（第三步）
|
|
|
+ plt.figure(figsize=(12, 8))
|
|
|
+
|
|
|
+ # 绘制三轴加速度
|
|
|
+ plt.subplot(3, 1, 1)
|
|
|
+ plt.plot(df['simTime'], df['a_x_body'], 'r-', label='X-axis Acceleration')
|
|
|
+
|
|
|
+ # 添加阈值线
|
|
|
+ if 'min_threshold' in df.columns and df['min_threshold'].iloc[0] is not None:
|
|
|
+ min_threshold = df['min_threshold'].iloc[0]
|
|
|
+ plt.axhline(y=min_threshold, color='r', linestyle=':', label=f'Min Threshold ({min_threshold})')
|
|
|
+ if 'max_threshold' in df.columns and df['max_threshold'].iloc[0] is not None:
|
|
|
+ max_threshold = df['max_threshold'].iloc[0]
|
|
|
+ plt.axhline(y=max_threshold, color='g', linestyle=':', label=f'Max Threshold ({max_threshold})')
|
|
|
+
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Acceleration (m/s²)')
|
|
|
+ plt.title('Body X-axis Acceleration (Longitudinal)')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.subplot(3, 1, 2)
|
|
|
+ plt.plot(df['simTime'], df['a_y_body'], 'g-', label='Y-axis Acceleration')
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Acceleration (m/s²)')
|
|
|
+ plt.title('Body Y-axis Acceleration (Lateral)')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.subplot(3, 1, 3)
|
|
|
+ plt.plot(df['simTime'], df['a_z_body'], 'b-', label='Z-axis Acceleration')
|
|
|
+ plt.xlabel('Time (s)')
|
|
|
+ plt.ylabel('Acceleration (m/s²)')
|
|
|
+ msdv_value = df['msdv_value'].iloc[0] if 'msdv_value' in df.columns else 0
|
|
|
+ motion_sickness_prob = df['motion_sickness_prob'].iloc[0] if 'motion_sickness_prob' in df.columns else 0
|
|
|
+ plt.title(f'Body Z-axis Acceleration (Vertical) - MSDV: {msdv_value:.4f}, Motion Sickness Probability: {motion_sickness_prob:.2f}%')
|
|
|
+ plt.grid(True)
|
|
|
+ plt.legend()
|
|
|
+
|
|
|
+ plt.tight_layout()
|
|
|
+
|
|
|
+ # 保存图像
|
|
|
+ chart_filename = os.path.join(output_dir, f"msdv_chart_{timestamp}.png")
|
|
|
+ plt.savefig(chart_filename, dpi=300)
|
|
|
+ plt.close()
|
|
|
+
|
|
|
+ logger.info(f"MSDV chart saved to: {chart_filename}")
|
|
|
+ return chart_filename
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f"Failed to generate MSDV 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 generate_function_chart_data(function_calculator, metric_name: str, output_dir: Optional[str] = None) -> Optional[str]:
|
|
|
+ """Generate chart data for function metrics"""
|
|
|
+ # 待实现
|
|
|
+ return None
|
cicv
XGJ_zhaoyuan