zhaoyuan/modules/metric/efficient.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
##################################################################
#
# Copyright (c) 2024 CICV, Inc. All Rights Reserved
#
##################################################################
"""
@Authors:           zhanghaiwen
@Data:              2024/12/23
@Last Modified:     2024/12/23
@Summary:           Efficient metrics calculation
"""

from modules.lib.score import Score
from modules.lib.log_manager import LogManager
import numpy as np
from typing import Dict, Tuple, Optional, Callable, Any
import pandas as pd


class Efficient:
    """高效性指标计算类"""
    
    def __init__(self, data_processed):
        """初始化高效性指标计算类
        
        Args:
            data_processed: 预处理后的数据对象
        """
        self.logger = LogManager().get_logger()
        self.data_processed = data_processed
        self.df = data_processed.object_df.copy()  # 浅拷贝
        self.ego_df = data_processed.ego_data.copy()  # 浅拷贝
        
        # 配置参数
        self.STOP_SPEED_THRESHOLD = 0.05  # 停车速度阈值 (m/s)
        self.STOP_TIME_THRESHOLD = 0.5    # 停车时间阈值 (秒)
        self.FRAME_RANGE = 13             # 停车帧数阈值
        
        # 初始化结果变量
        self.stop_count = 0     # 停车次数
        self.stop_duration = 0  # 平均停车时长
        self.average_v = 0      # 平均速度
        
        # 统计指标结果字典
        self.calculated_value = {
            'maxSpeed': 0,
            'deviationSpeed': 0,
            'averagedSpeed': 0,
            'stopDuration': 0,
            'speedUtilizationRatio': 0,
            'accelerationSmoothness': 0  # 添加新指标的默认值
        }
        
    def _max_speed(self):
        """计算最大速度
        
        Returns:
            float: 最大速度 (m/s)
        """
        max_speed = self.ego_df['v'].max() * 3.6  # 转换为 km/h
        self.calculated_value['maxSpeed'] = max_speed
        return max_speed

    def _deviation_speed(self):
        """计算速度方差
        
        Returns:
            float: 速度方差
        """
        deviation = self.ego_df['v'].var() * 3.6  # 转换为 km/h
        self.calculated_value['deviationSpeed'] = deviation
        return deviation

    def average_velocity(self):
        """计算平均速度
        
        Returns:
            float: 平均速度 (km/h)
        """
        self.average_v = self.ego_df['v'].mean() * 3.6  # 转换为 km/h
        self.calculated_value['averagedSpeed'] = self.average_v
        return self.average_v

    def stop_duration_and_count(self):
        """计算停车次数和平均停车时长
        
        Returns:
            float: 平均停车时长 (秒)
        """
        # 获取速度低于阈值的时间和帧号
        stop_mask = self.ego_df['v'] <= self.STOP_SPEED_THRESHOLD
        if not any(stop_mask):
            self.calculated_value['stopDuration'] = 0
            return 0  # 如果没有停车，直接返回0
            
        stop_time_list = self.ego_df.loc[stop_mask, 'simTime'].values.tolist()
        stop_frame_list = self.ego_df.loc[stop_mask, 'simFrame'].values.tolist()
        
        if not stop_frame_list:
            return 0  # 防止空列表导致的索引错误
            
        stop_frame_group = []
        stop_time_group = []
        sum_stop_time = 0
        f1, t1 = stop_frame_list[0], stop_time_list[0]
        
        # 检测停车段
        for i in range(1, len(stop_frame_list)):
            if stop_frame_list[i] - stop_frame_list[i - 1] != 1:  # 帧不连续
                f2, t2 = stop_frame_list[i - 1], stop_time_list[i - 1]
                # 如果停车有效（帧数差 >= FRAME_RANGE）
                if f2 - f1 >= self.FRAME_RANGE:
                    stop_frame_group.append((f1, f2))
                    stop_time_group.append((t1, t2))
                    sum_stop_time += (t2 - t1)
                    self.stop_count += 1
                # 更新起始点
                f1, t1 = stop_frame_list[i], stop_time_list[i]
        
        # 检查最后一段停车
        if len(stop_frame_list) > 0:
            f2, t2 = stop_frame_list[-1], stop_time_list[-1]
            last_frame = self.ego_df['simFrame'].values[-1]
            # 确保不是因为数据结束导致的停车
            if f2 - f1 >= self.FRAME_RANGE and f2 != last_frame:
                stop_frame_group.append((f1, f2))
                stop_time_group.append((t1, t2))
                sum_stop_time += (t2 - t1)
                self.stop_count += 1
        
        # 计算平均停车时长
        self.stop_duration = sum_stop_time / self.stop_count if self.stop_count > 0 else 0
        self.calculated_value['stopDuration'] = self.stop_duration
        
        self.logger.info(f"检测到停车次数: {self.stop_count}, 平均停车时长: {self.stop_duration:.2f}秒")
        return self.stop_duration

    def speed_utilization_ratio(self, default_speed_limit=60.0):
        """计算速度利用率
        
        速度利用率度量车辆实际速度与道路限速之间的比率，
        反映车辆对道路速度资源的利用程度。
        
        计算公式: R_v = v_actual / v_limit
        
        Args:
            default_speed_limit: 默认道路限速 (km/h)，当无法获取实际限速时使用
            
        Returns:
            float: 速度利用率 (0-1之间的比率)
        """
        # 获取车辆速度数据 (m/s)
        speeds = self.ego_df['v'].values
        
        # 尝试从数据中获取道路限速信息
        # 首先检查road_speed_max列，其次检查speedLimit列，最后使用默认值
        if 'road_speed_max' in self.ego_df.columns:
            speed_limits = self.ego_df['road_speed_max'].values
            self.logger.info("使用road_speed_max列作为道路限速信息")
        elif 'speedLimit' in self.ego_df.columns:
            speed_limits = self.ego_df['speedLimit'].values
            self.logger.info("使用speedLimit列作为道路限速信息")
        else:
            # 默认限速转换为 m/s
            default_limit_ms = default_speed_limit / 3.6
            speed_limits = np.full_like(speeds, default_limit_ms)
            self.logger.info(f"未找到道路限速信息，使用默认限速: {default_speed_limit} km/h")
        
        # 确保限速值为m/s单位，如果数据是km/h需要转换
        # 假设如果限速值大于30，则认为是km/h单位，需要转换为m/s
        if np.mean(speed_limits) > 30:
            speed_limits = speed_limits / 3.6
            self.logger.info("将限速单位从km/h转换为m/s")
        
        # 计算每一帧的速度利用率
        ratios = np.divide(speeds, speed_limits, 
                          out=np.zeros_like(speeds), 
                          where=speed_limits!=0)
        
        # 限制比率不超过1（超速按1计算）
        ratios = np.minimum(ratios, 1.0)
        
        # 计算平均速度利用率
        avg_ratio = np.mean(ratios)
        self.calculated_value['speedUtilizationRatio'] = avg_ratio
        
        self.logger.info(f"速度利用率(Speed Utilization Ratio): {avg_ratio:.4f}")
        return avg_ratio


# ----------------------
# 基础指标计算函数
# ----------------------
def maxSpeed(data_processed) -> dict:
    """计算最大速度"""
    efficient = Efficient(data_processed)
    max_speed = efficient._max_speed()
    return {"maxSpeed": float(max_speed)}

def deviationSpeed(data_processed) -> dict:
    """计算速度方差"""
    efficient = Efficient(data_processed)
    deviation = efficient._deviation_speed()
    return {"deviationSpeed": float(deviation)}

def averagedSpeed(data_processed) -> dict:
    """计算平均速度"""
    efficient = Efficient(data_processed)
    avg_speed = efficient.average_velocity()
    return {"averagedSpeed": float(avg_speed)}

def stopDuration(data_processed) -> dict:
    """计算停车持续时间和次数"""
    efficient = Efficient(data_processed)
    stop_duration = efficient.stop_duration_and_count()
    return {"stopDuration": float(stop_duration)}

def speedUtilizationRatio(data_processed) -> dict:
    """计算速度利用率"""
    efficient = Efficient(data_processed)
    ratio = efficient.speed_utilization_ratio()
    return {"speedUtilizationRatio": float(ratio)}

def acceleration_smoothness(data_processed) -> dict:
    """计算加速度平稳度"""
    efficient = Efficient(data_processed)
    smoothness = efficient.acceleration_smoothness()
    return {"accelerationSmoothness": float(smoothness)}


class EfficientRegistry:
    """高效性指标注册器"""
    
    def __init__(self, data_processed):
        self.logger = LogManager().get_logger()  # 获取全局日志实例
        self.data = data_processed
        self.eff_config = data_processed.efficient_config["efficient"]
        self.metrics = self._extract_metrics(self.eff_config)
        self._registry = self._build_registry()
    
    def _extract_metrics(self, config_node: dict) -> list:
        """DFS遍历提取指标"""
        metrics = []
        def _recurse(node):
            if isinstance(node, dict):
                if 'name' in node and not any(isinstance(v, dict) for v in node.values()):
                    metrics.append(node['name'])
                for v in node.values():
                    _recurse(v)
        _recurse(config_node)
        self.logger.info(f'评比的高效性指标列表:{metrics}')
        return metrics
    
    def _build_registry(self) -> dict:
        """自动注册指标函数"""
        registry = {}
        for metric_name in self.metrics:
            try:
                registry[metric_name] = globals()[metric_name]
            except KeyError:
                self.logger.error(f"未实现指标函数: {metric_name}")
        return registry
    
    def batch_execute(self) -> dict:
        """批量执行指标计算"""
        results = {}
        for name, func in self._registry.items():
            try:
                result = func(self.data)
                results.update(result)
                # 新增：将每个指标的结果写入日志
                self.logger.info(f'高效性指标[{name}]计算结果: {result}')
            except Exception as e:
                self.logger.error(f"{name} 执行失败: {str(e)}", exc_info=True)
                results[name] = None
        self.logger.info(f'高效性指标计算结果:{results}')
        return results


class EfficientManager:
    """高效性指标管理类"""  
    def __init__(self, data_processed):
        self.data = data_processed
        self.efficient = EfficientRegistry(self.data)
    
    def report_statistic(self):
        """Generate the statistics and report the results."""
        # 使用注册表批量执行指标计算
        efficient_result = self.efficient.batch_execute()
        return efficient_result


def acceleration_smoothness(self):
        """计算加速度平稳度
        
        加速度平稳度用以衡量车辆加减速过程的平滑程度，
        通过计算加速度序列的波动程度（标准差）来评估。
        平稳度指标定义为 1-σ_a/a_max（归一化后靠近1代表加速度更稳定）。
        
        Returns:
            float: 加速度平稳度 (0-1之间的比率，越接近1表示越平稳)
        """
        # 获取加速度数据
        # 优先使用车辆坐标系下的加速度数据
        if 'lon_acc_vehicle' in self.ego_df.columns and 'lat_acc_vehicle' in self.ego_df.columns:
            # 使用车辆坐标系下的加速度计算合成加速度
            lon_acc = self.ego_df['lon_acc_vehicle'].values
            lat_acc = self.ego_df['lat_acc_vehicle'].values
            accel_magnitude = np.sqrt(lon_acc**2 + lat_acc**2)
            self.logger.info("使用车辆坐标系下的加速度计算合成加速度")
        elif 'accelX' in self.ego_df.columns and 'accelY' in self.ego_df.columns:
            # 计算合成加速度（考虑X和Y方向）
            accel_x = self.ego_df['accelX'].values
            accel_y = self.ego_df['accelY'].values
            accel_magnitude = np.sqrt(accel_x**2 + accel_y**2)
            self.logger.info("使用accelX和accelY计算合成加速度")
        else:
            # 从速度差分计算加速度
            velocity = self.ego_df['v'].values
            time_diff = self.ego_df['simTime'].diff().fillna(0).values
            # 避免除以零
            time_diff[time_diff == 0] = 1e-6
            accel_magnitude = np.abs(np.diff(velocity, prepend=velocity[0]) / time_diff)
            self.logger.info("从速度差分计算加速度")
        
        # 过滤掉异常值（可选）
        # 使用3倍标准差作为阈值
        mean_accel = np.mean(accel_magnitude)
        std_accel = np.std(accel_magnitude)
        threshold = mean_accel + 3 * std_accel
        filtered_accel = accel_magnitude[accel_magnitude <= threshold]
        
        # 如果过滤后数据太少，则使用原始数据
        if len(filtered_accel) < len(accel_magnitude) * 0.8:
            filtered_accel = accel_magnitude
            self.logger.info("过滤后数据太少，使用原始加速度数据")
        else:
            self.logger.info(f"过滤掉 {len(accel_magnitude) - len(filtered_accel)} 个异常加速度值")
        
        # 计算加速度标准差
        accel_std = np.std(filtered_accel)
        
        # 计算最大加速度（使用95百分位数以避免极端值影响）
        accel_max = np.percentile(filtered_accel, 95)
        
        # 防止除以零
        if accel_max < 0.001:
            accel_max = 0.001
        
        # 计算平稳度指标: 1 - σ_a/a_max
        smoothness = 1.0 - (accel_std / accel_max)
        
        # 限制在0-1范围内
        smoothness = np.clip(smoothness, 0.0, 1.0)
        
        self.calculated_value['accelerationSmoothness'] = smoothness
        
        self.logger.info(f"加速度标准差: {accel_std:.4f} m/s²")
        self.logger.info(f"加速度最大值(95百分位): {accel_max:.4f} m/s²")
        self.logger.info(f"加速度平稳度(Acceleration Smoothness): {smoothness:.4f}")
        
        return smoothness