cicv_evaluate_master-single_eval/compliance.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/21
@Last Modified:     2023/08/21
@Summary:           Compliance 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 common import score_grade, string_concatenate, replace_key_with_value, score_over_100
from scipy.spatial.distance import euclidean


class Compliance(object):
    """
        Class for achieving compliance metrics for autonomous driving.

    Attributes:
        droadMark_df: Roadmark data, stored in dataframe format.

    """

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

        self.roadMark_df = data_processed.road_mark_df
        self.trafficLight_df = data_processed.traffic_light_df
        self.trafficSignal_df = data_processed.traffic_signal_df
        self.objState_df = data_processed.object_df

        self.violation_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'violation'])

        self.illegal_count = 0
        self.penalty_points = 0
        self.penalty_money = 0
        self.warning_count = 0

        self.config = data_processed.config
        compliance_config = self.config.config['compliance']
        self.compliance_config = compliance_config

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

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

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

        # score data
        self.weight = compliance_config['weightDimension']
        self.weight_dict = compliance_config['weight']
        self.weight_list = compliance_config['weightList']
        self.weight_type_dict = compliance_config['typeWeight']
        self.weight_type_list = compliance_config['typeWeightList']

        # type dicts
        self.type_illegal_count_dict = {}
        self.type_penalty_points_dict = {}
        self.type_penalty_money_dict = {}
        self.type_warning_count_dict = {}
        self.type_penalty_law_dict = {}

        # metric dicts
        self.metric_illegal_count_dict = {}
        self.metric_penalty_points_dict = {}
        self.metric_penalty_money_dict = {}
        self.metric_warning_count_dict = {}
        self.metric_penalty_law_dict = {}

        # self.metric_illegal_list = []
        self.metric_illegal_count_list = []
        self.metric_penalty_points_list = []
        self.metric_penalty_money_list = []
        self.metric_warning_count_list = []
        self.metric_penalty_law_list = []

    def press_solid_line(self):
        """
        #define RDB_ROADMARK_TYPE_NONE           0
        #define RDB_ROADMARK_TYPE_SOLID          1
        #define RDB_ROADMARK_TYPE_BROKEN         2
        #define RDB_ROADMARK_TYPE_CURB           3
        #define RDB_ROADMARK_TYPE_GRASS          4
        #define RDB_ROADMARK_TYPE_BOTDOT         5
        #define RDB_ROADMARK_TYPE_OTHER          6
        #define RDB_ROADMARK_TYPE_SOLID_SOLID    7
        #define RDB_ROADMARK_TYPE_BROKEN_SOLID   8
        #define RDB_ROADMARK_TYPE_SOLID_BROKEN   9
        #define RDB_ROADMARK_TYPE_LANE_CENTER   10
        #define RDB_ROADMARK_COLOR_NONE          0
        #define RDB_ROADMARK_COLOR_WHITE         1
        #define RDB_ROADMARK_COLOR_RED           2
        #define RDB_ROADMARK_COLOR_YELLOW        3
        #define RDB_ROADMARK_COLOR_OTHER         4
        #define RDB_ROADMARK_COLOR_BLUE          5
        #define RDB_ROADMARK_COLOR_GREEN         6
        """
        # Dimy = self.objState_df[self.objState_df["id"] == 1]["dimY"][0] / 2
        Dimy = self.objState_df[self.objState_df["playerId"] == 1]["dimY"][0] / 2
        dist_line = self.roadMark_df[self.roadMark_df["type"] == 1]
        dist_line = dist_line.reset_index()
        dist_press = dist_line[abs(dist_line["lateralDist"].values) <= Dimy]

        # 违规行为详情表统计
        t_list = dist_press['simTime'].values.tolist()
        f_list = dist_press['simFrame'].values.tolist()
        group_time = []
        group_frame = []
        sub_group_time = []
        sub_group_frame = []
        for i in range(len(f_list)):
            if not sub_group_time or t_list[i] - t_list[i - 1] <= 1:
                sub_group_time.append(t_list[i])
                sub_group_frame.append(f_list[i])
            else:
                group_time.append(sub_group_time)
                group_frame.append(sub_group_frame)
                sub_group_time = [t_list[i]]
                sub_group_frame = [f_list[i]]

        CONTINUOUS_FRAME_PERIOD = 13
        group_time.append(sub_group_time)
        group_frame.append(sub_group_frame)
        group_time = [g for g in group_time if len(g) >= CONTINUOUS_FRAME_PERIOD]
        group_frame = [g for g in group_frame if len(g) >= CONTINUOUS_FRAME_PERIOD]
        # group_time = [g for g in group_time if g[-1]-g[0] >= 1]
        # group_frame = [g for g in group_frame if g[-1]-g[0] >= 13]

        press_line_count = len(group_time)

        # 输出图表值
        press_line_time = [[g[0], g[-1]] for g in group_time]
        press_line_frame = [[g[0], g[-1]] for g in group_frame]

        if press_line_time:
            time_df = pd.DataFrame(press_line_time, columns=['start_time', 'end_time'])
            # frame_df = pd.DataFrame(press_line_frame, columns=['start_frame', 'end_frame'])
            time_df['violation'] = '压实线'
            # frame_df['violation'] = '压实线'
            # self.violation_df = pd.concat([self.violation_df, time_df, frame_df], ignore_index=True)
            self.violation_df = pd.concat([self.violation_df, time_df], ignore_index=True)
            # self.violation_df = pd.concat([self.violation_df, frame_df], ignore_index=True)

        warning_count = 0

        press_line_dict = {
            'metric': 'pressSolidLine',
            'weight': 3,
            'illegal_count': press_line_count,
            'penalty_points': press_line_count * 3,
            'penalty_money': press_line_count * 200,
            'warning_count': warning_count,
            'penalty_law': '《中华人民共和国道路交通安全法》第八十二条:机动车在高速公路上行驶，不得有下列行为:(三)骑、轧车行道分界线或者在路肩上行驶。'
        }

        return press_line_dict

    def normalization_processing(self, x):
        difference = x["traffic_light_h_diff"]
        while (difference >= 360):
            difference -= 360
        return difference

    def flag_red_traffic_light(self, x, cycleTime, duration_start, duration_end):
        divisor = x['simTime'] / cycleTime
        decimal_part = divisor - int(divisor)
        # decimal_part = divisor % 1
        if duration_start <= decimal_part < duration_end:
            return 1
        else:
            return x["flag_red_traffic_light"]

    def run_red_light(self):
        """

        """
        ego_df = self.objState_df[(self.objState_df.playerId == 1) & (self.objState_df.type == 1)]
        trafficLight_id_list = set(self.trafficLight_df["id"].tolist())
        l_stipulate = 10
        run_red_light_count = 0
        for trafficLight_id in trafficLight_id_list:
            trafficLight_position = self.trafficSignal_df[self.trafficSignal_df["playerId"] == trafficLight_id]
            trafficLight_character = self.trafficLight_df[self.trafficLight_df.id == trafficLight_id]
            if trafficLight_position.empty:  # trafficSign中没有记录
                continue
            trafficLight_position = trafficLight_position.iloc[:1, :]
            trafficLight_position_x = trafficLight_position['posX'].values[0]
            trafficLight_position_y = trafficLight_position['posY'].values[0]
            trafficLight_position_heading = trafficLight_position['posH'].values[0]
            cycleTime = trafficLight_character["cycleTime"].values[0]
            noPhases = trafficLight_character["noPhases"].values[0]

            ego_df["traffic_light_distance_absolute"] = ego_df[['posX', 'posY']].apply( \
                lambda x: euclidean((trafficLight_position_x, trafficLight_position_y), (x['posX'], x['posY'])), axis=1)

            # ego_df["traffic_light_distance_absolute"] = ego_df.apply(lambda x: euclidean((trafficLight_position_x, trafficLight_position_y), (x['posX'], x['posY'])), axis=1)

            ego_df["traffic_light_h_diff"] = ego_df.apply(
                lambda x: abs(x['posH'] - trafficLight_position_heading) * 57.3, axis=1)
            ego_df["traffic_light_h_diff"] = ego_df.apply(
                lambda x: self.normalization_processing(x), axis=1).copy()  # 归一化到[0,360)之间
            mask_trafftic_light = ((ego_df['traffic_light_h_diff'] <= 210) & (
                    ego_df['traffic_light_h_diff'] >= 150)) | (ego_df['traffic_light_h_diff'] <= 30) | (
                                          ego_df['traffic_light_h_diff'] >= 330)
            ego_near_light = ego_df[(ego_df.traffic_light_distance_absolute <= l_stipulate) & mask_trafftic_light]
            if ego_near_light.empty:
                continue
            """ 当前是否为红灯 """
            ego_near_light["flag_red_traffic_light"] = 0  # 不是红灯
            type_list = trafficLight_character['violation'][:noPhases]
            duration = trafficLight_character['duration'][:noPhases]
            duration_correct = [0] * noPhases
            for number in range(noPhases):
                duration_correct[number] = sum(duration[:number + 1])
                type_current = type_list.values[number]
                if type_current == 1:  # 当前duration是红灯
                    if number == 0:
                        duration_start = 0
                    else:
                        duration_start = duration_correct[number - 1]
                    duration_end = duration_correct[number]
                    ego_near_light["flag_red_traffic_light"] = ego_near_light[
                        ['simTime', 'flag_red_traffic_light']].apply(
                        lambda x: self.flag_red_traffic_light(x, cycleTime, duration_start, duration_end), axis=1)

            # 挑出闯红灯数据
            run_red_light_df = ego_near_light[ego_near_light['flag_red_traffic_light'] == 1]

            # 违规行为详情表统计
            t_list = run_red_light_df['simTime'].values.tolist()
            f_list = run_red_light_df['simFrame'].values.tolist()

            group_time = []
            group_frame = []
            sub_group_time = []
            sub_group_frame = []
            for i in range(len(f_list)):
                if not sub_group_time or t_list[i] - t_list[i - 1] <= 1:
                    sub_group_time.append(t_list[i])
                    sub_group_frame.append(f_list[i])
                else:
                    group_time.append(sub_group_time)
                    group_frame.append(sub_group_frame)
                    sub_group_time = [t_list[i]]
                    sub_group_frame = [f_list[i]]

            CONTINUOUS_FRAME_PERIOD = 13
            group_time.append(sub_group_time)
            group_frame.append(sub_group_frame)
            group_time = [g for g in group_time if len(g) >= CONTINUOUS_FRAME_PERIOD]
            group_frame = [g for g in group_frame if len(g) >= CONTINUOUS_FRAME_PERIOD]

            run_red_light_time = [[g[0], g[-1]] for g in group_time]
            run_red_light_frame = [[g[0], g[-1]] for g in group_frame]

            if run_red_light_time:
                time_df = pd.DataFrame(run_red_light_time, columns=['start_time', 'end_time'])
                # frame_df = pd.DataFrame(run_red_light_frame, columns=['start_frame', 'end_frame'])
                time_df['violation'] = '闯红灯'
                # frame_df['violation'] = '闯红灯'
                self.violation_df = pd.concat([self.violation_df, time_df], ignore_index=True)
                # self.violation_df = pd.concat([self.violation_df, frame_df], ignore_index=True)

            # 闯红灯次数统计
            if ego_near_light["flag_red_traffic_light"].any() == 1:
                run_red_light_count = run_red_light_count + 1

        run_red_light_dict = {
            'metric': 'runRedLight',
            'weight': 6,
            'illegal_count': run_red_light_count,
            'penalty_points': run_red_light_count * 6,
            'penalty_money': run_red_light_count * 200,
            'warning_count': 0,
            'penalty_law': '《中华人民共和国道路交通安全法实施条例》第四十条:(二)红色叉形灯或者箭头灯亮时，禁止本车道车辆通行。'
        }

        return run_red_light_dict

    def overspeed(self):
        """
        《中华人民共和国道路交通安全法实施条例》第四十五条:机动车在道路上行驶不得超过限速标志、标线标明的速度；
        1、高速、城市快速路超速超过规定时速10%以内，处以警告，不扣分；
        2、普通私家车高速超过规定时速10%以上未达20%的，处以200元罚款，记3分，普通私家车在除高速公路、城市快速路外的道路，超速20%以上未达到50%，会一次被记3分；
        3、超过规定时速20%以上未达50%的，处以200元罚款，记6分，或在普通道路上超速50%以上，都会一次被记6分。；
        4、超过规定时速50%以上的，可吊销驾驶证并罚款2000元，计12分。

        [0, 10) 0，0，1
        [10, 20) 0，200，1
        高速、城市快速路:[20, 50) 6，200
        高速、城市快速路:[50，）12，2000

        高速、城市以外:[20, 50）3，200
        高速、城市以外:[50，）6，1000-2000
        """
        Dimx = self.objState_df[self.objState_df["playerId"] == 1]["dimX"][0] / 2
        data_ego = self.objState_df[self.objState_df["playerId"] == 1]
        speed_limit_sign = self.trafficSignal_df[self.trafficSignal_df["type"] == 274]
        same_df_rate = pd.merge(speed_limit_sign, data_ego, on=['simTime', 'simFrame'], how='inner')
        same_df_rate = same_df_rate.reset_index()

        speed_df = same_df_rate[(abs(same_df_rate["posX_x"] - same_df_rate["posX_y"]) <= 7) & (
                abs(same_df_rate["posY_x"] - same_df_rate["posY_y"]) <= Dimx)]
        speed_df["speed"] = np.sqrt(speed_df["speedX"] ** 2 + speed_df["speedY"] ** 2) * 3.6

        # speed_df["value"] = 10
        # same_df_rate["value"] = 10

        list_sign = speed_df[speed_df["speed"] > speed_df["value"]]

        # 违规行为详情表统计
        t_list = list_sign['simTime'].values.tolist()
        f_list = list_sign['simFrame'].values.tolist()
        group_time = []
        group_frame = []
        sub_group_time = []
        sub_group_frame = []
        for i in range(len(f_list)):
            if not sub_group_time or t_list[i] - t_list[i - 1] <= 2:
                sub_group_time.append(t_list[i])
                sub_group_frame.append(f_list[i])
            else:
                group_time.append(sub_group_time)
                group_frame.append(sub_group_frame)
                sub_group_time = [t_list[i]]
                sub_group_frame = [f_list[i]]

        CONTINUOUS_FRAME_PERIOD = 13
        group_time.append(sub_group_time)
        group_frame.append(sub_group_frame)
        group_time = [g for g in group_time if len(g) >= CONTINUOUS_FRAME_PERIOD]
        group_frame = [g for g in group_frame if len(g) >= CONTINUOUS_FRAME_PERIOD]

        # 输出图表值
        overspeed_time = [[g[0], g[-1]] for g in group_time]
        overspeed_frame = [[g[0], g[-1]] for g in group_frame]

        if overspeed_time:
            time_df = pd.DataFrame(overspeed_time, columns=['start_time', 'end_time'])
            # frame_df = pd.DataFrame(overspeed_frame, columns=['start_frame', 'end_frame'])
            time_df['violation'] = '超速'
            # frame_df['violation'] = '超速'
            self.violation_df = pd.concat([self.violation_df, time_df], ignore_index=True)
            # self.violation_df = pd.concat([self.violation_df, frame_df], ignore_index=True)

        # list_sign = list_sign.reset_index()
        index_sign = list_sign.index.to_list()
        speed_df["flag_press"] = speed_df["simFrame"].apply(lambda x: 1 if x in list_sign["simFrame"] else 0)
        speed_df["diff_press"] = speed_df["flag_press"].diff()
        # overrate_count = speed_df[speed_df["diff_press"] == -1]["diff_press"].count()

        index_list = []
        subindex_list = []
        for i in range(len(index_sign)):
            if not subindex_list or index_sign[i] - index_sign[i - 1] == 1:
                subindex_list.append(index_sign[i])
            else:
                index_list.append(subindex_list)
                subindex_list.append(index_sign[i])
        index_list.append(subindex_list)

        overspeed_count_0_to_10 = 0
        overspeed_count_10_to_20 = 0
        overspeed_count_20_to_50 = 0
        overspeed_count_50_to_ = 0

        if index_list[0]:
            for i in range(len(index_list)):
                left = index_list[i][0]
                right = index_list[i][-1]
                df_tmp = speed_df.loc[left:right + 1]
                max_ratio = ((df_tmp["speed"] - df_tmp["value"]) / df_tmp["value"]).max()
                if max_ratio >= 0 and max_ratio < 0.1:
                    overspeed_count_0_to_10 += 1
                elif max_ratio >= 0.1 and max_ratio < 0.2:
                    overspeed_count_10_to_20 += 1
                elif max_ratio >= 0.2 and max_ratio < 0.5:
                    overspeed_count_20_to_50 += 1
                elif max_ratio >= 0.5:
                    overspeed_count_50_to_ += 1

        """
        [0, 10) 0，0，1
        [10, 20) 0，200，1
        高速、城市快速路:[20, 50) 6，200
        高速、城市快速路:[50，）12，2000
        """

        overspeed_0_to_10 = {
            # 'metric': 'overspeed_0_to_10',
            'metric': 'overspeed10',
            'weight': None,
            'illegal_count': overspeed_count_0_to_10,
            'penalty_points': 0,
            'penalty_money': 0,
            'warning_count': overspeed_count_0_to_10,
            'penalty_law': '《中华人民共和国道路交通安全法》第四十二条:机动车上道路行驶，不得超过限速标志标明的最高时速。'
        }

        overspeed_10_to_20 = {
            # 'metric': 'overspeed_10_to_20',
            'metric': 'overspeed10_20',
            'weight': None,
            'illegal_count': overspeed_count_10_to_20,
            'penalty_points': 0,
            'penalty_money': overspeed_count_10_to_20 * 200,
            'warning_count': overspeed_count_10_to_20,
            'penalty_law': '《中华人民共和国道路交通安全法》第四十二条:机动车上道路行驶，不得超过限速标志标明的最高时速。'
        }

        overspeed_20_to_50 = {
            # 'metric': 'overspeed_20_to_50',
            'metric': 'overspeed20_50',
            'weight': 6,
            'illegal_count': overspeed_count_20_to_50,
            'penalty_points': overspeed_count_20_to_50 * 6,
            'penalty_money': overspeed_count_20_to_50 * 200,
            'warning_count': 0,
            'penalty_law': '《中华人民共和国道路交通安全法》第四十二条:机动车上道路行驶，不得超过限速标志标明的最高时速。'
        }

        overspeed_50_to_ = {
            # 'metric': 'overspeed_50_to_',
            'metric': 'overspeed50',
            'weight': 12,
            'illegal_count': overspeed_count_50_to_,
            'penalty_points': overspeed_count_50_to_ * 12,
            'penalty_money': overspeed_count_50_to_ * 2000,
            'warning_count': 0,
            'penalty_law': '《中华人民共和国道路交通安全法》第四十二条:机动车上道路行驶，不得超过限速标志标明的最高时速。'
        }

        return overspeed_0_to_10, overspeed_10_to_20, overspeed_20_to_50, overspeed_50_to_

    def score_cal_penalty_points(self, penalty_points):
        """
        # 1_results: 0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55, 100
        # 3_results: 0, 15, 30, 45, 100
        # 6_results: 0, 30, 100
        # 9_results: 0, 15, 100
        """
        if penalty_points == 0:
            score = 100
        elif penalty_points >= 12:
            score = 0
        else:
            score = (12 - penalty_points) / 12 * 60
        return score

    def time_splice(self, start_time, end_time):
        str_time = f"[{start_time}s, {end_time}s]"
        return str_time

    def time_frame_splice(self, start_time, end_time, start_frame, end_frame):
        str_time = f"[{start_time}s, {end_time}s], {start_frame}-{end_frame}"
        return str_time

    def weight_type_cal(self):
        # penalty_list = [1, 3, 6, 9, 12]
        penalty_list = [1, 3, 6, 12]
        sum_penalty = sum(penalty_list)
        weight_type_list = [round(x / sum_penalty, 2) for x in penalty_list]
        return weight_type_list

    def compliance_statistic(self):

        # metric analysis
        press_line_dict = self.press_solid_line()
        run_red_light_dict = self.run_red_light()
        overspeed_0_to_10_dict, overspeed_10_to_20_dict, overspeed_20_to_50_dict, overspeed_50_dict = self.overspeed()

        df_list = []
        if "overspeed10" in self.metric_list:
            df_list.append(overspeed_0_to_10_dict)

        if "overspeed10_20" in self.metric_list:
            df_list.append(overspeed_10_to_20_dict)

        if "pressSolidLine" in self.metric_list:
            df_list.append(press_line_dict)

        if "runRedLight" in self.metric_list:
            df_list.append(run_red_light_dict)

        if "overspeed20_50" in self.metric_list:
            df_list.append(overspeed_20_to_50_dict)

        if "overspeed50" in self.metric_list:
            df_list.append(overspeed_50_dict)

        # generate dataframe and dicts
        compliance_df = pd.DataFrame(df_list)

        return compliance_df

    def compliance_score(self):
        """
        待优化
        可以重新构建数据结构，每个判断函数的返回类型为dict
        在多指标统计扣分、罚款时，可以将所有字典合并为一个dataframe，然后求和
        """
        # initialization
        score_type_dict = {}
        compliance_df = self.compliance_statistic()
        self.illegal_count = int(compliance_df['illegal_count'].sum())
        # self.penalty_points = compliance_df['penalty_points'].sum()
        # self.penalty_money = compliance_df['penalty_money'].sum()
        # self.warning_count = compliance_df['warning_count'].sum()

        # self.metric_illegal_list = compliance_df[compliance_df['illegal_count'] > 0]['metric'].tolist()
        # self.metric_illegal_count_list = compliance_df['illegal_count'].values.tolist()
        #
        # self.penalty_points_list = compliance_df['penalty_points'].values.tolist()
        # self.penalty_money_list = compliance_df['penalty_money'].values.tolist()
        # self.warning_count_list = compliance_df['warning_count'].values.tolist()

        self.metric_penalty_points_dict = compliance_df.set_index('metric').to_dict()['penalty_points']
        self.metric_illegal_count_dict = compliance_df.set_index('metric').to_dict()['illegal_count']
        self.metric_penalty_money_dict = compliance_df.set_index('metric').to_dict()['penalty_money']
        self.metric_warning_count_dict = compliance_df.set_index('metric').to_dict()['warning_count']
        self.metric_penalty_law_dict = compliance_df.set_index('metric').to_dict()['penalty_law']

        # deduct1
        if "deduct1" in self.type_list:
            deduct1_df = compliance_df[(compliance_df['weight'].isna()) | (compliance_df['weight'] == 1)]
            penalty_points_1 = deduct1_df['penalty_points'].sum()
            illegal_count_1 = deduct1_df['illegal_count'].sum()
            score_type_dict["deduct1"] = self.score_cal_penalty_points(penalty_points_1)
            self.type_illegal_count_dict["deduct1"] = illegal_count_1

        # deduct3
        if "deduct3" in self.type_list:
            deduct3_df = compliance_df[(compliance_df['weight'].isna()) | (compliance_df['weight'] == 3)]
            penalty_points_3 = deduct3_df['penalty_points'].sum()
            illegal_count_3 = deduct3_df['illegal_count'].sum()
            score_type_dict["deduct3"] = self.score_cal_penalty_points(penalty_points_3)
            self.type_illegal_count_dict["deduct3"] = illegal_count_3

        # deduct6
        if "deduct6" in self.type_list:
            deduct6_df = compliance_df[(compliance_df['weight'].isna()) | (compliance_df['weight'] == 6)]
            penalty_points_6 = deduct6_df['penalty_points'].sum()
            illegal_count_6 = deduct6_df['illegal_count'].sum()
            score_type_dict["deduct6"] = self.score_cal_penalty_points(penalty_points_6)
            self.type_illegal_count_dict["deduct6"] = illegal_count_6

        # deduct9
        if "deduct9" in self.type_list:
            deduct9_df = compliance_df[(compliance_df['weight'].isna()) | (compliance_df['weight'] == 9)]
            penalty_points_9 = deduct9_df['penalty_points'].sum()
            illegal_count_9 = deduct9_df['illegal_count'].sum()
            score_type_dict["deduct9"] = self.score_cal_penalty_points(penalty_points_9)
            self.type_illegal_count_dict["deduct9"] = illegal_count_9

        # deduct12
        if "deduct12" in self.type_list:
            deduct12_df = compliance_df[(compliance_df['weight'].isna()) | (compliance_df['weight'] == 12)]
            penalty_points_12 = deduct12_df['penalty_points'].sum()
            illegal_count_12 = deduct12_df['illegal_count'].sum()
            score_type_dict["deduct12"] = self.score_cal_penalty_points(penalty_points_12)
            self.type_illegal_count_dict["deduct12"] = illegal_count_12

        weight_type_list = self.weight_type_cal()
        weight_dict = {
            "overspeed10": 0.5,
            "overspeed10_20": 0.5,
            "pressSolidLine": 1.0,
            "runRedLight": 0.5,
            "overspeed20_50": 0.5,
            "overspeed50": 1.0
        }

        type_list = ["deduct1", "deduct3", "deduct6", "deduct12"]
        if not self.weight_custom:  # 客观赋权
            self.weight_type_list = weight_type_list
            self.weight_type_dict = {key: value for key, value in zip(type_list, weight_type_list)}
            self.weight_dict = weight_dict

        if self.penalty_points >= 12:
            score_compliance = 0
        elif sum(score_type_dict.values()) / len(score_type_dict) == 100:
            score_compliance = 100
        else:
            score_type_tmp = [80 if x == 100 else x for key, x in score_type_dict.items()]
            score_compliance = np.dot(self.weight_type_list, score_type_tmp)

        score_compliance = round(score_compliance, 2)

        print("\n[合规性表现及得分情况]")
        print(f"合规性得分为:{score_compliance:.2f}分。")
        print(f"合规性各分组得分为:{score_type_dict}。")
        print(f"合规性各分组权重为:{self.weight_type_list}。")
        # print(f"合规性各指标违规次数为:\n{count_metric}。")
        # print(f"违法总次数为:{self.illegal_count}。")
        # print(f"违法扣分总数为:{self.penalty_points}。")
        # print(f"罚款总金额为:{self.penalty_money}。")
        # print(f"警告总次数为:{self.warning_count}。")
        # print(compliance_df)
        return score_compliance, score_type_dict

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

        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 compliance_weight_distribution(self):
        # get weight distribution
        weight_distribution = {}
        weight_distribution["name"] = "合规性"

        if "deduct1" in self.type_list:
            deduct1_weight_indexes_dict = {}
            if "overspeed10" in self.metric_list:
                deduct1_weight_indexes_dict[
                    "overspeed10Weight"] = f"超速，但未超过10%({self.weight_dict['overspeed10'] * 100:.2f}%)"

            if "overspeed10_20" in self.metric_list:
                deduct1_weight_indexes_dict[
                    "overspeed1020Weight"] = f"超速10%-20%({self.weight_dict['overspeed10_20'] * 100:.2f}%)"

            weight_distribution['deduct1'] = {
                "deduct1Weight": f"轻微违规（1分）({self.weight_type_dict['deduct1'] * 100:.2f}%)",
                "indexes": deduct1_weight_indexes_dict
            }

        if "deduct3" in self.type_list:
            deduct3_weight_indexes_dict = {}

            if "pressSolidLine" in self.metric_list:
                deduct3_weight_indexes_dict[
                    "pressSolidLineWeight"] = f"压实线({self.weight_dict['pressSolidLine'] * 100:.2f}%)"

            weight_distribution['deduct3'] = {
                "deduct3Weight": f"中等违规（3分）({self.weight_type_dict['deduct3'] * 100:.2f}%)",
                "indexes": deduct3_weight_indexes_dict
            }

        if "deduct6" in self.type_list:
            deduct6_weight_indexes_dict = {}

            if "runRedLight" in self.metric_list:
                deduct6_weight_indexes_dict["runRedLightWeight"] = f"闯红灯({self.weight_dict['runRedLight'] * 100:.2f}%)"

            if "overspeed20_50" in self.metric_list:
                deduct6_weight_indexes_dict[
                    "overspeed2050Weight"] = f"超速20%-50%({self.weight_dict['overspeed20_50'] * 100:.2f}%)"

            weight_distribution['deduct6'] = {
                "deduct6Weight": f"危险违规（6分）({self.weight_type_dict['deduct6'] * 100:.2f}%)",
                "indexes": deduct6_weight_indexes_dict
            }

        if "deduct9" in self.type_list:
            deduct9_weight_indexes_dict = {}

            weight_distribution['deduct9'] = {
                "deduct9Weight": f"严重违规（9分）({self.weight_type_dict['deduct6'] * 100:.2f}%)",
                "indexes": deduct9_weight_indexes_dict
            }

        if "deduct12" in self.type_list:
            deduct12_weight_indexes_dict = {}

            if "overspeed50" in self.metric_list:
                deduct12_weight_indexes_dict[
                    "overspeed50Weight"] = f"超速50%以上({self.weight_dict['overspeed50'] * 100:.2f}%)"

            weight_distribution['deduct12'] = {
                "deduct12Weight": f"重大违规（12分）({self.weight_type_dict['deduct12'] * 100:.2f}%)",
                "indexes": deduct12_weight_indexes_dict
            }

        return weight_distribution

    def report_statistic(self):
        score_compliance, score_type_dict = self.compliance_score()
        grade_compliance = score_grade(score_compliance)

        score_compliance = int(score_compliance) if int(score_compliance) == score_compliance else score_compliance
        score_type = [int(n) if int(n) == n else n for key, n in score_type_dict.items()]

        # get weight distribution
        # weight_distribution = self.compliance_weight_distribution()
        weight_distribution = self._get_weight_distribution()

        # if self.metric_illegal_list:
        #     str_illegel_metric = string_concatenate(self.metric_illegal_list)

        cnt1 = self.illegal_count
        if grade_compliance == '优秀':
            comp_description1 = '车辆在本轮测试中无违反交通法规行为；'
        elif grade_compliance == '良好':
            comp_description1 = f'车辆在本轮测试中共发生{cnt1}次违反交通法规⾏为；'
        elif grade_compliance == '一般':
            comp_description1 = f'车辆在本轮测试中共有{cnt1}次违反交通法规⾏为，需要提高算法在合规性上的表现；'
        elif grade_compliance == '较差':
            comp_description1 = f'车辆在本轮测试中共有{cnt1}次违反交通法规⾏为，需要提高算法在合规性上的表现；'

        # xx指标得分超过基准线且无违规行为，算法表现良好
        # 违规次数 self.illegal_count
        # 1分xx次，3分xx次

        if cnt1 == 0:
            comp_description2 = "车辆在该用例中无违反交通法规行为，算法表现良好。"
        else:
            str_illegel_type = ""

            for type in self.type_list:
                if self.type_illegal_count_dict[type] > 0:
                    str_illegel_type += f'{self.type_name_dict[type]}行为{self.type_illegal_count_dict[type]}次，'

            # if "deduct1" in self.type_list:
            #     if self.type_illegal_count_dict["deduct1"] > 0:
            #         str_illegel_type += f'轻微违规（1分）行为{self.type_illegal_count_dict["deduct1"]}次，'
            #
            # if "deduct3" in self.type_list:
            #     if self.type_illegal_count_dict["deduct3"] > 0:
            #         str_illegel_type += f'中等违规（3分）行为{self.type_illegal_count_dict["deduct3"]}次，'
            #
            # if "deduct6" in self.type_list:
            #     if self.type_illegal_count_dict["deduct6"] > 0:
            #         str_illegel_type += f'危险违规（6分）行为{self.type_illegal_count_dict["deduct6"]}次，'
            #
            # if "deduct9" in self.type_list:
            #     if self.type_illegal_count_dict["deduct9"] > 0:
            #         str_illegel_type += f'严重违规（9分）行为{self.type_illegal_count_dict["deduct9"]}次，'
            #
            # if "deduct12" in self.type_list:
            #     if self.type_illegal_count_dict["deduct12"] > 0:
            #         str_illegel_type += f'重大违规（12分）行为{self.type_illegal_count_dict["deduct12"]}次，'

            str_illegel_type = str_illegel_type[:-1]
            comp_description2 = f"车辆在该用例共违反交通法规{cnt1}次。其中{str_illegel_type}。违规行为详情见附录C。"

        # data for violations table
        if not self.violation_df.empty:
            # self.violation_df['time'] = self.violation_df.apply(
            #     lambda row: self.time_frame_splice(row['start_time'], row['end_time'], row['start_frame'],
            #                                        row['end_frame']), axis=1)
            self.violation_df['time'] = self.violation_df.apply(
                lambda row: self.time_splice(row['start_time'], row['end_time']), axis=1)
            df_violations = self.violation_df[['time', 'violation']]
            violations_slices = df_violations.to_dict('records')
        else:
            violations_slices = []

        # violations_slices = [{'time': "[1s, 2s]", 'violation': "压实线"},
        #                      {'time': "[10s, 20s]", 'violation': "闯红灯"},
        #                      {'time': "[100s, 200s]", 'violation': "超速"}]

        deductPoint_dict = {}

        for type in self.type_list:
            type_dict = {
                "name": self.type_name_dict[type],
                "score": score_type_dict[type],
            }

            type_indexes = {}
            for metric in self.metric_list:
                type_indexes[metric] = {
                    # "name": self.name_dict[metric],
                    # "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
                    "name": f"{self.name_dict[metric]}",
                    "times": self.metric_illegal_count_dict[metric],
                    "deductPoints": self.metric_penalty_points_dict[metric],
                    "fine": self.metric_penalty_money_dict[metric],
                    "basis": self.metric_penalty_law_dict[metric]
                }
            type_dict["indexes"] = type_indexes
            deductPoint_dict[type] = type_dict

        # deduct1
        if "deduct1" in self.type_list:
            deduct1_indexes = {}
            if "overspeed10" in self.metric_list:
                tmp_dict = {
                    "name": "超速，但未超过10%",
                    "times": self.metric_illegal_count_dict["overspeed10"],
                    "deductPoints": self.metric_penalty_points_dict["overspeed10"],
                    "fine": self.metric_penalty_money_dict["overspeed10"],
                    "basis": self.metric_penalty_law_dict["overspeed10"]
                    # "basis": "《中华人民共和国道路交通安全法》第四十二条:机动车上道路行驶，不得超过限速标志标明的最高时速。"
                }
                deduct1_indexes['overspeed10'] = tmp_dict

            if "overspeed10_20" in self.metric_list:
                tmp_dict = {
                    "name": "超速10%-20%",
                    "times": self.metric_illegal_count_dict["overspeed10_20"],
                    "deductPoints": self.metric_penalty_points_dict["overspeed10_20"],
                    "fine": self.metric_penalty_money_dict["overspeed10_20"],
                    "basis": self.metric_penalty_law_dict["overspeed10_20"]
                    # "basis": "《中华人民共和国道路交通安全法》第四十二条:机动车上道路行驶，不得超过限速标志标明的最高时速。"
                }

                deduct1_indexes['overspeed10_20'] = tmp_dict

            deduct1_dict = {
                "name": "轻微违规（1分）",
                "score": score_type_dict["deduct1"],
                "indexes": deduct1_indexes
            }
            deductPoint_dict["deduct1"] = deduct1_dict

        # deduct3
        if "deduct3" in self.type_list:
            deduct3_indexes = {}
            if "pressSolidLine" in self.metric_list:
                tmp_dict = {
                    "name": "压实线",
                    "times": self.metric_illegal_count_dict["pressSolidLine"],
                    "deductPoints": self.metric_penalty_points_dict["pressSolidLine"],
                    "fine": self.metric_penalty_money_dict["pressSolidLine"],
                    "basis": self.metric_penalty_law_dict["pressSolidLine"]
                    # "basis": "《中华人民共和国道路交通安全法》第八十二条:机动车在高速公路上行驶，不得有下列行为:(三)骑、轧车行道分界线或者在路肩上行驶。"
                }
                deduct3_indexes['pressSolidLine'] = tmp_dict

            deduct3_dict = {
                "name": "中等违规（3分）",
                "score": score_type_dict["deduct3"],
                "indexes": deduct3_indexes
            }
            deductPoint_dict["deduct3"] = deduct3_dict

        # deduct6
        if "deduct6" in self.type_list:
            deduct6_indexes = {}
            if "runRedLight" in self.metric_list:
                tmp_dict = {
                    "name": "闯红灯",
                    "times": self.metric_illegal_count_dict["runRedLight"],
                    "deductPoints": self.metric_penalty_points_dict["runRedLight"],
                    "fine": self.metric_penalty_money_dict["runRedLight"],
                    "basis": self.metric_penalty_law_dict["runRedLight"]
                    # "basis": "《中华人民共和国道路交通安全法实施条例》第四十条:(二)红色叉形灯或者箭头灯亮时，禁止本车道车辆通行。"
                }
                deduct6_indexes['runRedLight'] = tmp_dict

            if "overspeed20_50" in self.metric_list:
                tmp_dict = {
                    "name": "超速20%-50%",
                    "times": self.metric_illegal_count_dict["overspeed20_50"],
                    "deductPoints": self.metric_penalty_points_dict["overspeed20_50"],
                    "fine": self.metric_penalty_money_dict["overspeed20_50"],
                    "basis": self.metric_penalty_law_dict["overspeed20_50"]
                    # "basis": "《中华人民共和国道路交通安全法》第四十二条:机动车上道路行驶，不得超过限速标志标明的最高时速。"
                }
                deduct6_indexes['overspeed20_50'] = tmp_dict

            deduct6_dict = {
                "name": "危险违规（6分）",
                "score": score_type_dict["deduct6"],
                "indexes": deduct6_indexes
            }
            deductPoint_dict["deduct6"] = deduct6_dict

        # deduct9
        if "deduct9" in self.type_list:
            deduct9_indexes = {}
            if "xx" in self.metric_list:
                tmp_dict = {
                    "name": "-",
                    "times": "0",
                    "deductPoints": "0",
                    "fine": "0",
                    "basis": "-"
                    # "basis": "-"
                }
                deduct9_indexes['xx'] = tmp_dict

            deduct9_dict = {
                "name": "严重违规（9分）",
                "score": 100,  # score_type_dict["deduct9"],
                "indexes": deduct9_indexes
            }
            deductPoint_dict["deduct9"] = deduct9_dict

        # deduct12
        if "deduct12" in self.type_list:
            deduct12_indexes = {}
            if "overspeed50" in self.metric_list:
                tmp_dict = {
                    "name": "超速50%以上",
                    "times": self.metric_illegal_count_dict["overspeed50"],
                    "deductPoints": self.metric_penalty_points_dict["overspeed50"],
                    "fine": self.metric_penalty_money_dict["overspeed50"],
                    "basis": self.metric_penalty_law_dict["overspeed50"]
                    # "basis": "《中华人民共和国道路交通安全法》第四十二条:机动车上道路行驶，不得超过限速标志标明的最高时速。"
                }
                deduct12_indexes['overspeed50'] = tmp_dict

            deduct12_dict = {
                "name": "重大违规（12分）",
                "score": score_type_dict["deduct12"],
                "indexes": deduct12_indexes
            }
            deductPoint_dict["deduct12"] = deduct12_dict

        report_dict = {
            "name": "合规性",
            "weight": f"{self.weight * 100:.2f}%",
            "weightDistribution": weight_distribution,
            "score": score_compliance,
            "level": grade_compliance,
            'score_type': score_type,
            # 'score_metric': score_metric,
            'illegalCount': self.illegal_count,

            "description1": comp_description1,
            "description2": comp_description2,
            "details": deductPoint_dict,
            "violations": violations_slices
        }

        return report_dict

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