pujin_outdoor/accurate.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
##################################################################
#
# Copyright (c) 2023 CICV, Inc. All Rights Reserved
#
##################################################################
"""
@Authors:           xieguijin(xieguijin@china-icv.cn), yangzihao(yangzihao@china-icv.cn)
@Data:              2023/08/03
@Last Modified:     2023/08/03
@Summary:           Functionality metrics
"""

import math
import numpy as np
import pandas as pd
from scipy.spatial import KDTree

from score_weight import cal_score_with_priority, cal_weight_from_80
from common import score_grade, string_concatenate, replace_key_with_value, _cal_max_min_avg


class Accurate(object):
    """
    Class for achieving accurate metrics for autonomous driving.

    Attributes:
        df: Vehicle driving data, stored in dataframe format.
    """

    def __init__(self, data_processed, scoreModel, resultPath):
        # self.eval_data = pd.DataFrame()
        # self.data_processed = data_processed
        self.scoreModel = scoreModel
        self.resultPath = resultPath

        self.df = data_processed.ego_df
        self.df_trajectory = data_processed.trajectory_df  # 读取轨迹的数据

        self.config = data_processed.config
        accurate_config = data_processed.accurate_config
        self.accurate_config = accurate_config

        # common data
        self.builtin_metric_list = self.config.builtinMetricList

        # dimension data
        self.weight_custom = accurate_config['weightCustom']
        self.metric_list = accurate_config['metric']
        self.type_list = accurate_config['type']
        self.type_name_dict = accurate_config['typeName']
        self.name_dict = accurate_config['name']
        self.unit_dict = accurate_config['unit']

        # custom metric data
        self.customMetricParam = accurate_config['customMetricParam']
        self.custom_metric_list = list(self.customMetricParam.keys())
        self.custom_data = {}
        self.custom_param_dict = {}

        # score data
        self.weight = accurate_config['weightDimension']

        self.weight_type_dict = accurate_config['typeWeight']
        self.weight_type_list = accurate_config['typeWeightList']

        self.weight_dict = accurate_config['weight']
        self.weight_list = accurate_config['weightList']

        self.priority_dict = accurate_config['priority']
        self.priority_list = accurate_config['priorityList']

        self.kind_dict = accurate_config['kind']
        self.optimal_dict = accurate_config['optimal']
        self.multiple_dict = accurate_config['multiple']
        self.kind_list = accurate_config['kindList']
        self.optimal_list = accurate_config['optimalList']
        self.multiple_list = accurate_config['multipleList']

        # metric data
        self.metric_dict = accurate_config['typeMetricDict']
        # self.drive_metric_list = self.metric_dict['accurateDrive']
        # self.stop_metric_list = self.metric_dict['accurateStop']
        # self.drive_metric_list = ["averageSpeed"]
        # self.stop_metric_list = ["stopDuration", "stopCount"]

        # metric value
        self.positionError_list = []
        self.executeAccurateError_list = []
        self.positionError_time_max = []
        self.positionError_time_min = []

        self.positionError_dict = {}
        self.positionError_sum = 0
        # self.executeAccurateError_count = 0

        self.positionError = 0
        self.executeAccurateError = 0
        self.error_segment_time_list = 0

    def _accurate_metric_cal(self):
        self._positionError_cal()
        self._executeAccurateError_cal()

    def detect_discontinuities(self, target_error_codes):
        error_periods = {error_code: [] for error_code in target_error_codes}

        # 遍历DataFrame，检测目标错误码并统计时间段
        current_error_code = None
        start_time = None

        for index, row in self.df.iterrows():
            if row['task_error_code'] in target_error_codes:
                if current_error_code is None or current_error_code != row['task_error_code']:
                    # 如果当前错误码与前一个不同，或者这是第一个错误码，则记录新的时间段开始
                    current_error_code = row['task_error_code']
                    start_time = row['simTime']
            else:
                # 如果当前行不是目标错误码，并且我们之前记录了一个错误码，则记录时间段结束
                if current_error_code is not None:
                    # 注意：这里我们假设下一个非匹配行的时间不是当前错误码的时间段的一部分
                    # 但为了包含整个错误码持续的时间，我们不减去任何时间
                    error_periods[current_error_code].append((start_time, row['simTime']))
                    current_error_code = None

                # 处理最后一个错误码的时间段（如果DataFrame以错误码结束）
        if current_error_code is not None:
            # 由于我们想要包含最后一个错误码的时间点，所以我们需要检查是否有下一个时间点
            # 如果没有（即这是DataFrame的最后一行），我们就直接使用最后一个时间点
            end_time = self.df['simTime'].iloc[-1] if current_error_code == self.df['task_error_code'].iloc[-1] else self.df['simTime'].iloc[-2]
            error_periods[current_error_code].append((start_time, end_time))

        return error_periods

    # 对每个错误码进行分组处理
    def _executeAccurateError_cal(self):
        # 用于记录段数的变量
        error_segment_count = 0
        error_segment_time_list = []
        # 标记当前是否在目标数字的段中
        in_segment = False
        targets = [21201000300, 21201000001, 21201000002, 21202000000, 21203000200, 21203000300]
        error_periods = self.detect_discontinuities(targets)
        print("error_periods is", error_periods)
        for key, periods in error_periods.items():
            if periods:
                error_segment_count += 1
                for start, end in periods:
                    error_segment_time_list.append(start)
        self.executeAccurateError = error_segment_count
        self.error_segment_time_list = [str("{:.02f}".format(e)) + "s" for e in error_segment_time_list]
        print("self.error_segment_time_list is", self.error_segment_time_list)

    def _positionError_cal(self):
        # self.df_trajectory['ego_pos'] = self.df_trajectory.apply(lambda row: (row['ego_posX'], row['ego_posY']),
        self.df_trajectory['ego_pos'] = self.df_trajectory.apply(lambda row: (row['posX'], row['posY']),
                                                                 axis=1).tolist()
        self.df_trajectory['target_pos'] = self.df_trajectory.apply(lambda row: (row['TargetX'], row['TargetY']),
                                                                    axis=1).tolist()
        print("self.df_trajectory['target_pos'] is", self.df_trajectory['target_pos'])
        ego_pos = np.array(self.df_trajectory['ego_pos'].tolist())
        target_pos = KDTree(np.array(self.df_trajectory['target_pos'].tolist()))
        min_distances = target_pos.query(ego_pos, k=1)[0].ravel()
        # 初始化存储最小距离的列表
        min_distance = min_distances

        self.positionError_list = min_distance
        # indice_max = [idx for idx, val in enumerate(self.positionError_list) if val == max(self.positionError_list)]
        # indice_min = [idx for idx, val in enumerate(self.positionError_list) if val == min(self.positionError_list)]
        indice_max = np.where(self.positionError_list == np.max(self.positionError_list))
        indice_min = np.where(self.positionError_list == np.min(self.positionError_list))

        self.positionError_time_max = self.df_trajectory["simTime"].loc[indice_max].tolist()
        if len(self.positionError_time_max) > 2:
            positionError_time1 = self.positionError_time_max[0:2]
            self.positionError_time_max = set([str("{:.02f}".format(m)) + "s" for m in positionError_time1])

        self.positionError_time_min = self.df_trajectory["simTime"].loc[indice_min].tolist()
        if len(self.positionError_time_min) > 2:
            positionError_time2 = self.positionError_time_min[0:2]
            self.positionError_time_min = set([str("{:.02f}".format(m)) + "s" for m in positionError_time2])
        print("self.positionError_time_max is", self.positionError_time_max)
        self.positionError = np.std(np.array(min_distance))

    def _accurate_statistic(self):
        """

        """
        self._accurate_metric_cal()
        self.positionError_dict = _cal_max_min_avg(self.positionError_list) if len(self.positionError_list)>0 else {}
        arr_accurate = [[self.positionError, self.executeAccurateError]]
        return arr_accurate

    def _score_cal(self):
        """
        """
        arr_accurate = self._accurate_statistic()
        print("\n[准确性表现及得分情况]")
        print("准确性各指标值：", [[round(num, 2) for num in row] for row in arr_accurate])

        arr_accurate = np.array(arr_accurate)
        score_model = self.scoreModel(self.kind_list, self.optimal_list, self.multiple_list, arr_accurate)
        score_sub = score_model.cal_score()
        score_sub = list(map(lambda x: 80 if np.isnan(x) else x, score_sub))
        score_metric = [round(num, 2) for num in score_sub]
        metric_list = [x for x in self.metric_list if x in self.config.builtinMetricList]
        print("score_sub is", metric_list)
        score_metric_dict = {key: value for key, value in zip(metric_list, score_metric)}
        if self.executeAccurateError > 0:
            score_metric_dict['executeAccurateError'] = 0
        else:
            score_metric_dict['executeAccurateError'] = 100
        score_metric_dict = {key: score_metric_dict[key] for key in self.metric_list}
        score_metric = list(score_metric_dict.values())

        score_type_dict = {}
        if self.weight_custom:  # 自定义权重
            score_metric_with_weight_dict = {key: score_metric_dict[key] * self.weight_dict[key] for key in
                                             self.weight_dict}

            for type in self.type_list:
                type_score = sum(
                    value for key, value in score_metric_with_weight_dict.items() if key in self.metric_dict[type])
                score_type_dict[type] = round(type_score, 2)

            score_type_with_weight_dict = {key: score_type_dict[key] * self.weight_type_dict[key] for key in
                                           score_type_dict}
            score_accurate = sum(score_type_with_weight_dict.values())

        else:  # 客观赋权
            self.weight_list = cal_weight_from_80(score_metric)
            self.weight_dict = {key: value for key, value in zip(self.metric_list, self.weight_list)}
            score_accurate = cal_score_with_priority(score_metric, self.weight_list, self.priority_list)

            for type in self.type_list:
                type_weight = sum(value for key, value in self.weight_dict.items() if key in self.metric_dict[type])
                self.weight_dict = {key: round(value / type_weight, 4) for key, value in self.weight_dict.items() if
                                    key in self.metric_dict[type]}
                type_score_metric = [value for key, value in score_metric_dict.items() if key in self.metric_dict[type]]
                type_weight_list = [value for key, value in self.weight_dict.items() if key in self.metric_dict[type]]
                type_priority_list = [value for key, value in self.priority_dict.items() if
                                      key in self.metric_dict[type]]

                type_score = cal_score_with_priority(type_score_metric, type_weight_list, type_priority_list)
                score_type_dict[type] = round(type_score, 2)

        score_accurate = round(score_accurate, 2)

        print("准确性各指标基准值：", self.optimal_list)
        print(f"准确性得分为：{score_accurate:.2f}分。")
        print(f"准确性各类型得分为：{score_type_dict}。")
        print(f"准确性各指标得分为：{score_metric_dict}。")
        return score_accurate, score_type_dict, score_metric_dict

    def report_statistic(self):
        """

        Returns:

        """

        report_dict = {
            "name": "准确性",
            "weight": f"{self.weight * 100:.2f}%",
        }

        score_accurate, score_type_dict, score_metric_dict = self._score_cal()

        # score_accurate, score_metric = self.effi_score()
        score_accurate = int(score_accurate) if int(score_accurate) == score_accurate else round(score_accurate, 2)
        grade_accurate = score_grade(score_accurate)
        report_dict["score"] = score_accurate
        report_dict["level"] = grade_accurate

        description = f"· 在准确性方面，得分{score_accurate:.2f}分，表现{grade_accurate}。"
        is_good = True
        is_semicolon = False

        if self.positionError_sum > 1:
            is_good = False
            is_semicolon = True
            description += f"行驶过程中，位置偏移总误差为{self.positionError_sum:.2f}米，需重点优化"

        if self.executeAccurateError > 0:
            is_good = False
            if is_semicolon:
                description += "；"
            is_semicolon = False
            description += f"出现{self.executeAccurateError}次任务执行状态错误，需重点优化。"
        else:
            if is_semicolon:
                description += f"。"

        if is_good:
            description += f"行驶准确且任务执行准确，算法表现优秀。"

        report_dict["description"] = description
        str_tmp_positionError_max = "、".join(self.positionError_time_max)
        str_tmp_positionError_min = "、".join(self.positionError_time_min)
        description1 = f"在{str_tmp_positionError_max}时刻达到最大值{self.positionError_dict['max']:.2f}m\n"
        if self.positionError_dict['min'] > 0.3:
            description1 += f"在{str_tmp_positionError_min}时刻达到最小值{self.positionError_dict['min']:.2f}m\n"
        description1 += f"平均值：{self.positionError_dict['avg']:.2f}m" if self.positionError_dict else "位置偏移无误差"

        str_temp_error_segment = "、".join(self.error_segment_time_list)
        description2 = f"次数：{self.executeAccurateError}次"
        if len(str_temp_error_segment) > 0:
            description2 += f"\n在{str_temp_error_segment}时刻任务执行错误"
        else:
            description2 += f"\n任务执行过程中没有发生错误"

        positionError_index = {
            "weight": self.weight_dict['positionError'],
            "score": score_metric_dict['positionError'],
            "description": description1
        }

        executeAccurateError_index = {
            "weight": self.weight_dict['executeAccurateError'],
            "score": score_metric_dict['executeAccurateError'],
            "description": description2
        }

        indexes_dict = {
            "positionError": positionError_index,
            "executeAccurateError": executeAccurateError_index
        }

        report_dict["indexes"] = indexes_dict

        print(report_dict)

        return report_dict