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      # 平均速度
        
    def _max_speed(self):
        """计算最大速度
        
        Returns:
            float: 最大速度 (m/s)
        """
        return self.ego_df['v'].max()

    def _deviation_speed(self):
        """计算速度方差
        
        Returns:
            float: 速度方差
        """
        return self.ego_df['v'].var()

    def average_velocity(self):
        """计算平均速度
        
        Returns:
            float: 平均速度 (m/s)
        """
        self.average_v = self.ego_df['v'].mean()
        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):
            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.logger.info(f"检测到停车次数: {self.stop_count}, 平均停车时长: {self.stop_duration:.2f}秒")
        return self.stop_duration

    def report_statistic(self):
        """生成统计报告
        
        Returns:
            dict: 高效性评估结果
        """
        # 计算各项指标
        max_speed_ms = self._max_speed()
        deviation_speed_ms = self._deviation_speed()
        average_speed_ms = self.average_velocity()
        
        # 将 m/s 转换为 km/h 用于评分
        max_speed_kmh = max_speed_ms * 3.6
        deviation_speed_kmh = deviation_speed_ms * 3.6
        average_speed_kmh = average_speed_ms * 3.6
        
        efficient_result = {
            'maxSpeed': max_speed_kmh,        # 转换为 km/h
            'deviationSpeed': deviation_speed_kmh,  # 转换为 km/h
            'averagedSpeed': average_speed_kmh,     # 转换为 km/h
            'stopDuration': self.stop_duration_and_count()
        }
        
        self.logger.info(f"高效性指标计算完成，结果: {efficient_result}")
        
        return efficient_result


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

def deviationSpeed(data_processed) -> dict:
    """计算速度方差"""
    efficient = Efficient(data_processed)
    deviation = efficient._deviation_speed() * 3.6  # 转换为 km/h
    return {"deviationSpeed": float(deviation)}

def averagedSpeed(data_processed) -> dict:
    """计算平均速度"""
    efficient = Efficient(data_processed)
    avg_speed = efficient.average_velocity() * 3.6  # 转换为 km/h
    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)}


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