cicv_evaluate_master-single_eval/efficient.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 sys

sys.path.append('../common')
sys.path.append('../modules')
sys.path.append('../results')

import numpy as np
import pandas as pd

from data_info import DataInfoList
from score_weight import cal_score_with_priority, cal_weight_from_80
from common import score_grade, string_concatenate, replace_key_with_value, score_over_100


class Efficient(object):
    """
    Class for achieving efficient metrics for autonomous driving.

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

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

        self.data = data_processed.obj_data[1]
        self.mileage = data_processed.report_info['mileage']
        self.df = pd.DataFrame()

        self.config = data_processed.config
        efficient_config = self.config.config['efficient']
        self.efficient_config = efficient_config

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

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

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

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

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

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

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

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

        # metric data
        self.metric_dict = efficient_config['typeMetricDict']
        self.drive_metric_list = self.metric_dict['efficientDrive']
        self.stop_metric_list = self.metric_dict['efficientStop']
        # self.drive_metric_list = ["averageSpeed"]
        # self.stop_metric_list = ["stopDuration", "stopCount"]

        self._get_data()
        self._effi_param_cal()
        self.time_list = self.data['simTime'].values.tolist()
        self.value_dict = {}
        self.average_v = 0
        self.stop_count = 0
        self.stop_duration = 0
        self.pass_junction_time = 0

        self.score = 0

    def _get_data(self):
        """
        Get the data required for efficient evaluation according to efficient_INFO in DataInfoList.

        Parameters:
            df: Dataframe containing the vehicle running data.

        Returns:
            peak_valley: List of indices representing peaks and valleys.
        """
        efficient_info_list = DataInfoList.EFFICIENT_INFO
        self.df = self.data[efficient_info_list].copy()

    def _effi_param_cal(self):
        """

        """
        self.df = self.df[self.df['playerId'] == 1]
        # self.df = self.df[self.df['id'] == 1]
        # self.df['v'] = self.df.apply(lambda row: self.velocity(row['speed_x'], row['speed_y']), axis=1)
        # self.df['v'] = self.df.apply(lambda row: self.velocity(row['speedX'], row['speedY']), axis=1)

    def average_velocity(self):
        """
        1.速度的平均值
        2.平均速度 = 里程/时间
        速度不为0为开始时间，速度为0为结束时间
        """
        self.average_v = self.df['v'].mean()

    def stop_duration_and_count(self):
        """
        百公里停车次数
        平均每次停车时长
        驾照科目二考试判定为停车的标准：在考试项目区域内，汽车停顿2秒以上都算中途停车。
        """
        
        STOP_SPEED_THRESHOLD = 0.05
        STOP_TIME_THRESHOLD = 0.5  # 12.5帧
        stop_time_list = self.df[self.df['v'] <= STOP_SPEED_THRESHOLD]['simTime'].values.tolist()
        # stop_time_list = self.df[self.df['v'] == 0]['simTime'].values.tolist()
        stop_frame_list = self.df[self.df['v'] <= STOP_SPEED_THRESHOLD]['simFrame'].values.tolist()
        # stop_frame_list = self.df[self.df['v'] == 0]['simFrame'].values.tolist()

        FRAME_RANGE = 13
        stop_frame_group = []
        stop_time_group = []
        f1 = stop_frame_list[0] if stop_frame_list else 0
        t1 = stop_time_list[0] if stop_time_list else 0
        sum_stop_time = 0
        for i in range(1, len(stop_frame_list)):
            if stop_frame_list[i] - stop_frame_list[i - 1] != 1:
                f2 = stop_frame_list[i - 1]
                # 当帧数大于13时才确定为停车
                # 当时长大于0.5s才确定为停车
                if f2 - f1 >= FRAME_RANGE:
                    t2 = stop_time_list[i - 1]
                    stop_frame_group.append((f1, f2))
                    stop_time_group.append((t1, t2))
                    sum_stop_time += (t2 - t1)
                    self.stop_count += 1
                # update f1, t1
                f1 = stop_frame_list[i]
                t1 = stop_time_list[i]

        f2 = stop_frame_list[-1] if stop_frame_list else 0
        # 如果最后一段的帧数差大于13，且停车统计列表中的最后一帧不为用例的最后一帧
        if f2 - f1 >= FRAME_RANGE and f2 != self.df['simFrame'].values.tolist()[-1]:
            t2 = stop_time_list[-1]
            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.stop_count = self.stop_count / self.mileage * 1000000

    def pass_junction_time(self):
        pass

    def efficient_statistic(self):
        arr_effi = []
        self.average_velocity()
        self.stop_duration_and_count()

        if "averageSpeed" in self.metric_list:
            average_v = self.average_v
            arr_effi.append(average_v)
            self.value_dict["averageSpeed"] = average_v

        if "stopDuration" in self.metric_list:
            stop_duration = self.stop_duration
            arr_effi.append(stop_duration)
            self.value_dict["stopDuration"] = stop_duration

        if "stopCount" in self.metric_list:
            stop_count = self.stop_count
            arr_effi.append(stop_count)
            self.value_dict["stopCount"] = stop_count

        # arr_effi = [average_v, stop_count, stop_time]
        return arr_effi

    def custom_metric_param_parser(self, param_list):
        """
        param_dict = {
            "paramA" [
                {
                    "kind": "-1",
                    "optimal": "1",
                    "multiple": ["0.5","5"],
                    "spare1": null,
                    "spare2": null
                }
            ]
        }
        """
        kind_list = []
        optimal_list = []
        multiple_list = []
        spare_list = []
        # spare1_list = []
        # spare2_list = []

        for i in range(len(param_list)):
            kind_list.append(int(param_list[i]['kind']))
            optimal_list.append(float(param_list[i]['optimal']))
            multiple_list.append([float(x) for x in param_list[i]['multiple']])
            spare_list.append([item["param"] for item in param_list[i]["spare"]])
            # spare1_list.append(param_list[i]['spare1'])
            # spare2_list.append(param_list[i]['spare2'])

        result = {
            "kind": kind_list,
            "optimal": optimal_list,
            "multiple": multiple_list,
            "spare": spare_list,
            # "spare1": spare1_list,
            # "spare2": spare2_list
        }
        return result

    def custom_metric_score(self, metric, value, param_list):
        """

        """
        param = self.custom_metric_param_parser(param_list)
        self.custom_param_dict[metric] = param

        score_model = self.scoreModel(param['kind'], param['optimal'], param['multiple'], np.array([value]))
        score_sub = score_model.cal_score()
        score = sum(score_sub) / len(score_sub)
        return score

    def effi_score(self):
        score_metric_dict = {}
        score_type_dict = {}

        arr_effi = self.efficient_statistic()
        print("\n[高效性表现及得分情况]")
        print("高效性各指标值：", [round(num, 2) for num in arr_effi])

        if arr_effi:
            arr_effi = np.array([arr_effi])

            score_model = self.scoreModel(self.kind_list, self.optimal_list, self.multiple_list, arr_effi)
            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]
            score_metric_dict = {key: value for key, value in zip(metric_list, score_metric)}

        for metric in self.custom_metric_list:
            value = self.custom_data[metric]['value']
            param_list = self.customMetricParam[metric]
            score = self.custom_metric_score(metric, value, param_list)
            score_metric_dict[metric] = round(score, 2)

        score_metric_dict = {key: score_metric_dict[key] for key in self.metric_list}
        score_metric = list(score_metric_dict.values())

        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) if type_score < 100 else 100

            score_type_with_weight_dict = {key: score_type_dict[key] * self.weight_type_dict[key] for key in
                                           score_type_dict}
            score_efficient = 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_efficient = 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])
                for key, value in self.weight_dict.items():
                    if key in self.metric_dict[type]:
                        # self.weight_dict[key] = round(value / type_weight, 4)
                        self.weight_dict[key] = value / type_weight

                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) if type_score < 100 else 100

                for key in self.weight_dict:
                    self.weight_dict[key] = round(self.weight_dict[key], 4)

            score_type = list(score_type_dict.values())
            self.weight_type_list = cal_weight_from_80(score_type)
            self.weight_type_dict = {key: value for key, value in zip(self.type_list, self.weight_type_list)}

        score_efficient = round(score_efficient, 2)

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

    def effi_weight_distribution(self):
        # get weight distribution
        weight_distribution = {}
        weight_distribution["name"] = "高效性"

        if "efficientDrive" in self.type_list:
            drive_weight_indexes_dict = {key: f"{key}({value * 100:.2f}%)" for key, value in self.weight_dict.items() if
                                         key in self.drive_metric_list}

            weight_distribution_drive = {
                "driveWeight": f"行驶({self.weight_type_dict['efficientDrive'] * 100:.2f}%)",
                "indexes": drive_weight_indexes_dict
            }
            weight_distribution['efficientDrive'] = weight_distribution_drive

        if "efficientStop" in self.type_list:
            stop_weight_indexes_dict = {key: f"{key}({value * 100:.2f}%)" for key, value in self.weight_dict.items() if
                                        key in self.stop_metric_list}

            weight_distribution_stop = {
                "stopWeight": f"停车({self.weight_type_dict['efficientStop'] * 100:.2f}%)",
                "indexes": stop_weight_indexes_dict
            }
            weight_distribution['efficientStop'] = weight_distribution_stop

        return weight_distribution

    def _get_weight_distribution(self, dimension):
        # get weight distribution
        weight_distribution = {}
        weight_distribution["name"] = self.config.dimension_name[dimension]

        for type in self.type_list:
            type_weight_indexes_dict = {key: f"{self.name_dict[key]}({value * 100:.2f}%)" for key, value in
                                        self.weight_dict.items() if
                                        key in self.metric_dict[type]}

            weight_distribution_type = {
                "weight": f"{self.type_name_dict[type]}({self.weight_type_dict[type] * 100:.2f}%)",
                "indexes": type_weight_indexes_dict
            }
            weight_distribution[type] = weight_distribution_type

        return weight_distribution

    def report_statistic(self):
        """

        Returns:

        """

        # report_dict = {
        #     "name": "高效性",
        #     "weight": f"{self.weight * 100:.2f}%",
        #     "weightDistribution": weight_distribution,
        #     "score": score_efficient,
        #     "level": grade_efficient,
        #     'score_type': score_type,
        #     'score_metric': score_metric,
        #
        #     "description1": effi_description1,
        #     "description2": effi_description2,
        #     "description3": effi_description3,
        #
        #     "efficientDrive": drive_dict,
        #     "efficientStop": stop_dict,
        # }

        # brakePedal_list = self.data_processed.driver_ctrl_data['brakePedal_list']
        # throttlePedal_list = self.data_processed.driver_ctrl_data['throttlePedal_list']
        # steeringWheel_list = self.data_processed.driver_ctrl_data['steeringWheel_list']
        #
        # # common parameter calculate
        # brake_vs_time = self.zip_time_pairs(brakePedal_list, 100)
        # throttle_vs_time = self.zip_time_pairs(throttlePedal_list, 100)
        # steering_vs_time = self.zip_time_pairs(steeringWheel_list)

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

        # len_time = len(self.time_list)
        duration = self.time_list[-1]
        # duration_minute = round(duration / 60, 2)
        score_efficient, score_type_dict, score_metric_dict = self.effi_score()

        # get weight distribution
        report_dict["weightDistribution"] = self._get_weight_distribution("efficient")

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

        # for description
        good_type_list = []
        bad_type_list = []

        good_metric_list = []
        bad_metric_list = []

        # str for description
        str_over_optimal = ""
        type_details_dict = {}

        for type in self.type_list:
            bad_type_list.append(type) if score_type_dict[type] < 80 else good_type_list.append(type)

            type_dict = {
                "name": f"{self.type_name_dict[type]}",
            }

            builtin_graph_dict = {}
            custom_graph_dict = {}

            score_type = score_type_dict[type]
            grade_type = score_grade(score_type)
            type_dict["score"] = score_type
            type_dict["level"] = grade_type

            type_dict_indexes = {}
            for metric in self.metric_dict[type]:
                bad_metric_list.append(metric) if score_metric_dict[metric] < 80 else good_metric_list.append(metric)

                if metric in self.bulitin_metric_list:
                    # for indexes
                    type_dict_indexes[metric] = {
                        # "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
                        "name": f"{self.name_dict[metric]}",
                        "score": score_metric_dict[metric],
                        "value": f'{self.value_dict[metric]:.2f}',
                        # "range": f"[0, {self.optimal_dict['averageSpeed']})",
                        # "deviation": f"+{avv_deviation:.2f}%" if avv_deviation > 0 else f"{avv_deviation:.2f}%"
                    }

                    if self.kind_dict[metric] == -1:
                        type_dict_indexes[metric]["range"] = f"[0, {self.optimal_dict[metric]}]"
                        metric_over_optimal = ((self.value_dict[metric] - self.optimal_dict[metric]) /
                                               self.optimal_dict[
                                                   metric]) * 100
                        type_dict_indexes[metric][
                            "deviation"] = f"+{metric_over_optimal:.2f}%" if metric_over_optimal > 0 else f"0%"

                    elif self.kind_dict[metric] == 1:
                        type_dict_indexes[metric]["range"] = f"[{self.optimal_dict[metric]}, inf)"
                        metric_over_optimal = ((self.value_dict[metric] - self.optimal_dict[metric]) /
                                               self.optimal_dict[
                                                   metric]) * 100
                        type_dict_indexes[metric][
                            "deviation"] = f"0%" if metric_over_optimal > 0 else f"{metric_over_optimal:.2f}%"

                    elif self.kind_dict[metric] == 0:
                        value = self.value_dict[metric]
                        minn = self.optimal_dict[metric] * self.multiple_dict[metric][0]
                        maxx = self.optimal_dict[metric] * self.multiple_dict[metric][1]

                        type_dict_indexes[metric]["range"] = f"[{minn}, {maxx}]"

                        if value < minn:
                            metric_over_optimal = (value - minn) / minn * 100
                        elif value > maxx:
                            metric_over_optimal = (value - maxx) / maxx * 100
                        else:
                            metric_over_optimal = 0

                        type_dict_indexes[metric][
                            "deviation"] = f"+{metric_over_optimal:.2f}%" if metric_over_optimal > 0 else f"{metric_over_optimal:.2f}%"

                    # metric_over_optimal = ((self.value_dict[metric] - self.optimal_dict[metric]) / self.optimal_dict[
                    #     metric]) * 100
                    # type_dict_indexes[metric][
                    #     "deviation"] = f"+{metric_over_optimal:.2f}%" if metric_over_optimal > 0 else f"{metric_over_optimal:.2f}%"

                    if score_metric_dict[metric] < 80:
                        str_over_optimal += f'{self.name_dict[metric]}为{round(self.value_dict[metric], 2)}{self.unit_dict[metric]}，超过合理范围{metric_over_optimal:.2f}%；'

                else:
                    # for indexes
                    type_dict_indexes[metric] = {
                        # "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
                        "name": f"{self.name_dict[metric]}",
                        "score": score_metric_dict[metric],
                        "value": f'{self.custom_data[metric]["value"][0]:.2f}',
                        # "range": f"[0, {self.optimal_dict['averageSpeed']})",
                        # "deviation": f"+{avv_deviation}%" if avv_deviation > 0 else f"{avv_deviation}%"
                    }

                    value = self.custom_data[metric]["value"][0]
                    optimal = self.custom_param_dict[metric]['optimal'][0]

                    if self.custom_param_dict[metric]['kind'][0] == -1:
                        type_dict_indexes[metric][
                            "range"] = f"[0, {self.custom_param_dict[metric]['optimal'][0]}]"

                        metric_over_optimal = (value - optimal) / optimal * 100
                        type_dict_indexes[metric][
                            "deviation"] = f"+{metric_over_optimal:.2f}%" if metric_over_optimal > 0 else f"0%"

                    elif self.custom_param_dict[metric]['kind'][0] == 1:
                        type_dict_indexes[metric][
                            "range"] = f"[{self.custom_param_dict[metric]['optimal'][0]}, inf)"

                        metric_over_optimal = (value - optimal) / optimal * 100
                        type_dict_indexes[metric][
                            "deviation"] = f"0%" if metric_over_optimal > 0 else f"{metric_over_optimal:.2f}%"

                    elif self.custom_param_dict[metric]['kind'][0] == 0:
                        minn = value * self.custom_param_dict[metric]['multiple'][0][0]
                        maxx = value * self.custom_param_dict[metric]['multiple'][0][1]

                        type_dict_indexes[metric]["range"] = f"[{minn}, {maxx}]"

                        if value < minn:
                            metric_over_optimal = (value - minn) / minn * 100
                        elif value > maxx:
                            metric_over_optimal = (value - maxx) / maxx * 100
                        else:
                            metric_over_optimal = 0

                        type_dict_indexes[metric][
                            "deviation"] = f"+{metric_over_optimal:.2f}%" if metric_over_optimal > 0 else f"{metric_over_optimal:.2f}%"

                    # metric_over_optimal = ((self.custom_param_dict[metric]['optimal'][0] -
                    #                         self.custom_data[metric]["value"][0]) /
                    #                        self.custom_param_dict[metric]['optimal'][0]) * 100
                    # type_dict_indexes[metric][
                    #     "deviation"] = f"+{metric_over_optimal:.2f}%" if metric_over_optimal > 0 else f"{metric_over_optimal:.2f}%"

                    if score_metric_dict[metric] < 80:
                        str_over_optimal += f'{self.name_dict[metric]}为{self.custom_data[metric]["value"][0]:.2f}{self.unit_dict[metric]}，超过合理范围{metric_over_optimal:.2f}%；'

                    custom_graph_dict[metric] = self.custom_data[metric]['reportData']

                # str_uncomf_over_optimal = str_uncomf_over_optimal[:-1] + "；"
                type_dict["indexes"] = type_dict_indexes
                type_dict["builtin"] = builtin_graph_dict
                type_dict["custom"] = custom_graph_dict

            type_details_dict[type] = type_dict

        report_dict["details"] = type_details_dict
        # efficient description
        str_over_optimal = str_over_optimal[:-1]

        if grade_efficient == '优秀':
            effi_description1 = '车辆行驶效率高；'
        elif grade_efficient == '良好':
            effi_description1 = '算法在本轮测试中的表现满足设计指标要求；'
        elif grade_efficient == '一般':
            str_ineffi_type = string_concatenate(bad_metric_list)
            effi_description1 = f'算法需要在{str_ineffi_type}指标上进一步优化。其中，{str_over_optimal}；'
        elif grade_efficient == '较差':
            str_ineffi_type = string_concatenate(bad_metric_list)
            effi_description1 = f'需要提高算法在{str_ineffi_type}指标上的表现。其中，{str_over_optimal}；'

        if not bad_metric_list:
            effi_description2 = '高效性在各个指标上的表现俱佳。'
            effi_description3 = "算法的规划控制能力良好，通行效率高。"
        else:
            str_effi_type = string_concatenate(good_metric_list)
            str_ineffi_type = string_concatenate(bad_metric_list)

            effi_description2 = f"{str_effi_type}指标表现良好，{str_ineffi_type}指标表现不佳。其中，{str_over_optimal}。"
            effi_description3 = "算法应该优化车辆的规划控制逻辑，提高算法的通行效率。"

        report_dict["description1"] = replace_key_with_value(effi_description1, self.name_dict)
        report_dict["description2"] = replace_key_with_value(effi_description2, self.name_dict)
        report_dict["description3"] = effi_description3
        # report_dict['commonData'] = {
        #     "perData": [brake_vs_time, throttle_vs_time],
        #     "angData": steering_vs_time,
        # }

        # report_dict = {
        #     "name": "高效性",
        #     "weight": f"{self.weight * 100:.2f}%",
        #     "weightDistribution": weight_distribution,
        #     "score": score_efficient,
        #     "level": grade_efficient,
        #     'score_type': score_type,
        #     'score_metric': score_metric,
        #
        #     "description1": effi_description1,
        #     "description2": effi_description2,
        #     "description3": effi_description3,
        #
        #     "efficientDrive": drive_dict,
        #     "efficientStop": stop_dict,
        # }

        return report_dict

    def get_eval_data(self):
        df = self.eval_data
        return df