zhaoyuan/modules/lib/chart_generator.py

#!/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() == 'latestwarningdistance_ttc_pgvil':
            return generate_latest_warning_ttc_pgvil_chart(function_calculator, output_dir)
        elif metric_name.lower() == 'earliestwarningdistance_ttc_pgvil':
            return generate_earliest_warning_distance_ttc_pgvil_chart(function_calculator, output_dir)

        elif metric_name.lower() == 'earliestwarningdistance_pgvil':
            return generate_earliest_warning_distance_pgvil_chart(function_calculator, output_dir)
        elif metric_name.lower() == 'latestwarningdistance_pgvil':
            return generate_latest_warning_distance_pgvil_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_PGVIL and latestWarningDistance_PGVIL 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)
        warning_time = getattr(function_calculator, 'warning_time', 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': warning_time,
            '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_pgvil_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_PGVIL')
        max_threshold = thresholds["max"]
        min_threshold = thresholds["min"]

        warning_dist = getattr(function_calculator, 'warning_dist', None)
        warning_speed = getattr(function_calculator, 'warning_speed', None)
        warning_time = getattr(function_calculator, 'warning_time', 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_pgvil_data.csv")
        df_csv = pd.DataFrame({
            'simTime': ego_df['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_pgvil 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_pgvil_chart.png")
        plt.savefig(chart_filename, dpi=300)
        plt.close()

        logger.info(f"latestwarningdistance_ttc_pgvil chart saved to: {chart_filename}")
        return chart_filename

    except Exception as e:
        logger.error(f"Failed to generate latestwarningdistance_ttc_pgvil chart: {str(e)}", exc_info=True)
        return None

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_latest_warning_distance_pgvil_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_PGVIL')
        max_threshold = thresholds["max"]
        min_threshold = thresholds["min"]

        # Get calculated warning distance and speed
        warning_dist = getattr(function_calculator, 'warning_dist', None)
        warning_time = getattr(function_calculator, 'warning_time', None)

        if warning_dist.empty:
            logger.warning(f"Cannot generate latestWarningDistance_PGVIL 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_PGVIL_data.csv")
        df_csv = pd.DataFrame({
            'simTime': ego_df['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_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 = '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_PGVIL - Warning Distance Over Time')
        plt.grid(True)
        plt.legend()

        # Save image
        chart_filename = os.path.join(output_dir, f"latestWarningDistance_PGVIL_chart.png")
        plt.savefig(chart_filename, dpi=300)
        plt.close()

        logger.info(f"latestWarningDistance_PGVIL chart saved to: {chart_filename}")
        return chart_filename

    except Exception as e:
        logger.error(f"Failed to generate latestWarningDistance_PGVIL 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_earliest_warning_distance_ttc_pgvil_chart(function_calculator, output_dir: str) -> Optional[str]:
    """
    Generate TTC warning chart with data visualization for earliestWarningDistance_TTC_PGVIL 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_PGVIL'

    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_pgvil_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_pgvil 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.")