XGJ_code_cangku/safe.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 os
import math
import numpy as np
import pandas as pd

from collections import defaultdict
import scipy.integrate as spi

from config import DANGEROUS_DIST, LONGITUDINAL_DISTANCE_FRONT, LONGITUDINAL_DISTANCE_REAR, LATERAL_DISTANCE_ABS, \
    OVERSPEED_THRESHOLD
from common import score_grade, string_concatenate, replace_key_with_value, score_over_100, _cal_max_min_avg
from score_weight import cal_score_with_priority, cal_weight_from_80


class Safe(object):
    """
    Class for achieving safe metrics for autonomous driving.

    Attributes:
        df: Vehicle driving data, stored in dataframe format.
        df_drivectrl: Vehcile driving control data, including brakepedal, throttle pedal and steerignwheel.

    Methods:
        _safe_param_cal_new: Calculate parameters for evaluateion.
        _cal_v_ego_projection: Calculate ego car velocity project over distance.
        _cal_v_projection: Calculate relative velocity project over distance.
        _cal_a_projection: Calculate relative acceleration project over distance.
        _calculate_derivative: Calculate derivative.
        _cal_relative_angular_v: Calculate relative angular velocity.
        _death_pr: Calculate death probability.
        _cal_collisionRisk_level: Calculate collisionRisk level.
        dist: Calculate distance of two cars.

    """

    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.object_df.copy()
        self.ego_df = data_processed.ego_df
        # print("self.ego_df is", self.ego_df)
        self.obj_id_list = data_processed.obj_id_list
        # print("self.obj_id_list is", self.obj_id_list)
        #
        # # config infos for calculating score
        self.config = data_processed.config
        # print("self.config is", self.config)
        safe_config = self.config.config['safe']
        self.safe_config = safe_config

        # # common data
        self.bulitin_metric_list = self.config.builtinMetricList
        #
        # # dimension data
        self.weight_custom = safe_config['weightCustom']
        self.metric_list = safe_config['metric']
        self.type_list = safe_config['type']
        self.type_name_dict = safe_config['typeName']
        self.name_dict = safe_config['name']
        self.unit_dict = safe_config['unit']

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

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

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

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

        self.kind1_dict = safe_config['kind']
        # self.kind1_dict = self.kind_dict[0]
        self.optimal1_dict = safe_config['optimal']
        # self.optimal1_dict = self.optimal_dict[0]
        self.multiple_dict = safe_config['multiple']
        self.kind_list = safe_config['kindList']
        self.optimal_list = safe_config['optimalList']
        self.multiple_list = safe_config['multipleList']

        self.metric_dict = safe_config['typeMetricDict']
        # # self.time_metric_list = self.metric_dict['safeTime']
        # # self.distance_metric_list = self.metric_dict['safeDistance']
        #
        # # lists of drving control info
        # self.time_list = data_processed.driver_ctrl_data['time_list']
        # self.frame_list = self.df['simFrame'].values.tolist()
        if self.df['simFrame'].values.tolist()[-1] <= self.ego_df['simFrame'].values.tolist()[-1]:
            self.frame_list = self.df['simFrame'].values.tolist()
        else:
            self.frame_list = self.ego_df['simFrame'].values.tolist()

        self.collision_dist_list = []
        self.collisionRisk_list = []
        self.collisionSeverity_list = []
        self.overSpeed_list = []

        self.collision_risk_dict = dict()
        self.collision_severity_dict = dict()
        self.over_speed_dict = dict()

        self.collisionCount = 0
        self.collisionRisk = 0
        self.collisionSeverity = 0
        self.overSpeed = 0

        self.empty_flag = True

        if len(self.obj_id_list) > 1:
            # self.most_dangerous = {}
            # self.pass_percent = {}
            self._safe_metric_cal()
            self._overspeed_cal()

        # # no car following scene
        # if len(self.obj_id_list) > 1:
        #     self.unsafe_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
        #     self.unsafe_time_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
        #     self.unsafe_dist_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
        #     # self.unsafe_acce_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
        #     self.unsafe_acce_drac_df = pd.DataFrame(
        #         columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
        #     self.unsafe_acce_xtn_df = pd.DataFrame(
        #         columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
        #     self.unsafe_prob_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
        #     self.most_dangerous = {}
        #     self.pass_percent = {}
        #     self._safe_param_cal_new()

    def _overspeed_cal(self):
        """

        :return:
        """
        self.overSpeed_list = self.ego_df[self.ego_df['v'] > OVERSPEED_THRESHOLD]['v'].values.tolist()

        if self.overSpeed_list:
            max_speed = max(self.overSpeed_list)
            overspeed = (max_speed - OVERSPEED_THRESHOLD) / OVERSPEED_THRESHOLD * 100
            self.overSpeed = round(overspeed, 2)
        else:
            self.overSpeed = 0

    def _safe_metric_cal(self):
        """

        """
        self.collisionRisk_list = []
        self.collision_dist_list = []
        self.collisionSeverity_list = []

        Tc = 0.3
        rho = 0.3  # 驾驶员制动反应时间
        ego_accel_max = 6  # 自车油门最大加速度
        obj_decel_max = 8  # 前车刹车最大减速度
        ego_decel_min = 1  # 自车刹车最小减速度 ug
        ego_decel_lon_max = 8
        ego_decel_lat_max = 1

        # 构建双层字典数据结构
        obj_dict = defaultdict(dict)
        obj_data_dict = self.df.to_dict('records')

        for item in obj_data_dict:
            obj_dict[item['simFrame']][item['playerId']] = item
            # print("obj_dict is", obj_dict)

        df_list = []
        EGO_PLAYER_ID = 1

        self.empty_flag = True
        # print("obj_dict[1] is", self.frame_list)
        for frame_num in self.frame_list:
            # ego_data = obj_dict[frame_num][EGO_PLAYER_ID]
            # ego_data = self.ego_df.iloc[frame_num]
            ego_data = self.ego_df[self.ego_df['simFrame'] == frame_num]
            # if frame_num == 1:
            #     print("ego_data is", ego_data)
            v1 = ego_data['v']  # km/h to m/s
            # print("ego_data['posX'].values is", ego_data['posX'].values[0])
            x1 = ego_data['posX'].values[0]
            # print("x1 is", x1)
            y1 = ego_data['posY'].values[0]
            # print("x1 and y1 are", x1, y1)
            h1 = ego_data['yaw']
            # len1 = 2.3  # 随手录入的数值
            # width1 = ego_data['dimY']  #
            # o_x1 = ego_data['offX']
            v_x1 = ego_data['lon_v'].values[0]  # km/h to m/s
            # v_y1 = ego_data['speedY'] / 3.6  # km/h to m/s
            v_y1 = 0  # 随手录入的数值
            a_x1 = ego_data['speedH'].values[0]
            a_y1 = 0  # 随手录入的数值
            # a_y1 = ego_data['accelY']
            # a1 = ego_data['accel']
            # print("calculate is beginning...")
            # print("self.obj_id_list is", self.obj_id_list)
            for playerId in self.obj_id_list:
                dist_list = []
                # print("playerId is", playerId)

                if playerId == EGO_PLAYER_ID:
                    continue

                try:
                    obj_data = obj_dict[frame_num][playerId]
                except KeyError:
                    continue

                x2 = obj_data['posX']
                y2 = obj_data['posY']
                # print(" x2 and y2 are", x2, y2)
                # print("x2 is", y2)
                dist = self.dist(x1, y1, x2, y2)
                # print("dist is", dist)
                # DANGEROUS_DIST = 0.10  # 与障碍物距离小于此值即为危险
                if dist <= DANGEROUS_DIST:
                    dist_list.append(0)
                else:
                    dist_list.append(1)
                obj_data['dist'] = dist
                # print("obj_data['dist'] is", obj_data['dist'])

                v_x2 = obj_data['lon_v']  # km/h to m/s
                v_y2 = obj_data['lat_v']  # km/h to m/s
                v2 = math.sqrt(v_x2 ** 2 + v_y2 ** 2)
                # v2 = obj_data['v'] / 3.6  # km/h to m/s
                # h2 = obj_data['posH']
                # len2 = obj_data['dimX']
                # width2 = obj_data['dimY']
                # o_x2 = obj_data['offX']
                # a_x2 = obj_data['accelX']
                # a_y2 = obj_data['accelY']
                a_x2 = obj_data['lon_acc']
                a_y2 = obj_data['lat_acc']
                # a2 = obj_data['accel']

                dx, dy = x2 - x1, y2 - y1

                # 定义矢量A和x轴正向向量x
                A = np.array([dx, dy])
                # print("A is", A)
                x = np.array([1, 0])

                # 计算点积和向量长度
                dot_product = np.dot(A, x)
                vector_length_A = np.linalg.norm(A)
                vector_length_x = np.linalg.norm(x)

                # 计算夹角的余弦值
                cos_theta = dot_product / (vector_length_A * vector_length_x)

                # 将余弦值转换为角度值（弧度制）
                beta = np.arccos(cos_theta)  # 如何通过theta正负确定方向

                lon_d = dist * math.cos(beta - h1)
                lat_d = abs(dist * math.sin(beta - h1))  # 需要增加左正右负的判断，但beta取值为[0,pi)

                obj_dict[frame_num][playerId]['lon_d'] = lon_d
                obj_dict[frame_num][playerId]['lat_d'] = lat_d

                if lon_d > 5 or lon_d < -2 or lat_d > 1:
                    continue

                self.empty_flag = False

                vx, vy = v_x1 - v_x2, v_y1 - v_y2
                ax, ay = a_x2 - a_x1, a_y2 - a_y1

                v_ego_p = self._cal_v_ego_projection(dx, dy, v_x1, v_y1)
                v_obj_p = self._cal_v_ego_projection(dx, dy, v_x2, v_y2)
                vrel_projection_in_dist = self._cal_v_projection(dx, dy, vx, vy)
                arel_projection_in_dist = self._cal_a_projection(dx, dy, vx, vy, ax, ay, x1, y1, x2, y2, v_x1, v_y1,
                                                                 v_x2, v_y2)

                obj_dict[frame_num][playerId]['vrel_projection_in_dist'] = vrel_projection_in_dist
                obj_dict[frame_num][playerId]['arel_projection_in_dist'] = arel_projection_in_dist
                obj_dict[frame_num][playerId]['v_ego_projection_in_dist'] = v_ego_p
                obj_dict[frame_num][playerId]['v_obj_projection_in_dist'] = v_obj_p

                # obj_type = obj_data['type']

                TTC = self._cal_TTC(dist, vrel_projection_in_dist)
                # MTTC = self._cal_MTTC(TTC, vrel_projection_in_dist, arel_projection_in_dist)
                # THW = self._cal_THW(dist, v_ego_p)

                # 单车道时可用
                # LonSD = self._cal_longitudinal_safe_dist(v_ego_p, v_obj_p, rho, ego_accel_max, ego_decel_min,
                #                                          obj_decel_max)

                # lat_dist = 0.5
                # v_right = v1
                # v_left = v2
                # a_right_lat_brake_min = 1
                # a_left_lat_brake_min = 1
                # a_lat_max = 5

                # LatSD = self._cal_lateral_safe_dist(lat_dist, v_right, v_left, rho, a_right_lat_brake_min,
                #                                     a_left_lat_brake_min,
                #                                     a_lat_max)

                # DRAC = self._cal_DRAC(dist, vrel_projection_in_dist, len1, len2, width1, width2, o_x1, o_x2)

                lon_a1 = a_x1 * math.cos(h1) + a_y1 * math.sin(h1)
                lon_a2 = a_x2 * math.cos(h1) + a_y2 * math.sin(h1)
                # lon_a = abs(lon_a1 - lon_a2)
                # lon_d = dist * abs(math.cos(beta - h1))
                # lon_v = v_x1 * math.cos(h1) + v_y1 * math.sin(h1)
                # BTN = self._cal_BTN_new(lon_a1, lon_a, lon_d, lon_v, ego_decel_lon_max)

                lat_a1 = a_x1 * math.sin(h1) * -1 + a_y1 * math.cos(h1)
                lat_a2 = a_x2 * math.sin(h1) * -1 + a_y2 * math.cos(h1)
                lat_a = abs(lat_a1 - lat_a2)
                lat_d = dist * abs(math.sin(beta - h1))
                lat_v = v_x1 * math.sin(h1) * -1 + v_y1 * math.cos(h1)
                # STN = self._cal_STN_new(TTC, lat_a1, lat_a, lat_d, lat_v, ego_decel_lat_max, width1, width2)

                obj_dict[frame_num][playerId]['lat_v_rel'] = v_x1 - v_x2
                obj_dict[frame_num][playerId]['lon_v_rel'] = v_y1 - v_y2

                # BTN = self.cal_BTN(a_y1, ay, dy, vy, max_ay)
                # STN = self.cal_STN(TTC, a_x1, ax, dx, vx, max_ax, len1, len2)

                TTC = None if (not TTC or TTC < 0) else TTC
                # MTTC = None if (not MTTC or MTTC < 0) else MTTC
                # THW = None if (not THW or THW < 0) else THW
                #
                # DRAC = 10 if DRAC >= 10 else DRAC
                # print("calculate is beginning...")
                if not TTC or TTC > 4000:  # threshold = 4258.41
                    collisionSeverity = 0
                    pr_death = 0
                    collisionRisk = 0
                else:
                    result, error = spi.quad(self._normal_distribution, 0, TTC - Tc)
                    collisionSeverity = 1 - result
                    # print("collisionSeverity is", collisionSeverity)
                    # pr_death = self._death_pr(obj_type, vrel_projection_in_dist)
                    pr_death = self._death_pr(vrel_projection_in_dist)
                    collisionRisk = 0.4 * pr_death + 0.6 * collisionSeverity
                # print("collinsionRisk is", collisionRisk)
                self.collisionRisk_list.append(collisionRisk)
                self.collisionSeverity_list.append(collisionSeverity)
                self.collision_dist_list.append(max(dist_list))
        # print("obj_data['dist'] is", collision_dist_list)

        prev_value = self.collision_dist_list[0]  # 初始化前一个元素为列表的第一个元素

        # 遍历列表（从第二个元素开始）
        for i in range(1, len(self.collision_dist_list)):
            current_value = self.collision_dist_list[i]  # 当前元素

            # 如果当前元素是0且与前一个元素不同，或者这是第一个元素且为0
            if (current_value == 0 and prev_value != 0) or (i == 0 and current_value == 0):
                self.collisionCount += 1  # 增加0的段的计数

        self.collisionRisk = max(self.collisionRisk_list)
        self.collisionSeverity = max(self.collisionSeverity_list)

        print("collision count is", self.collisionCount)
        print("collisionRisk is", self.collisionRisk)
        print("collisionSeverity is", self.collisionSeverity)

    def _cal_v_ego_projection(self, dx, dy, v_x1, v_y1):
        # 计算 AB 连线的向量 AB
        # dx = x2 - x1
        # dy = y2 - y1

        # 计算 AB 连线的模长 |AB|
        AB_mod = math.sqrt(dx ** 2 + dy ** 2)

        # 计算 AB 连线的单位向量 U_AB
        U_ABx = dx / AB_mod
        U_ABy = dy / AB_mod

        # 计算 A 在 AB 连线上的速度 V1_on_AB
        V1_on_AB = v_x1 * U_ABx + v_y1 * U_ABy

        return V1_on_AB

    def _cal_v_projection(self, dx, dy, vx, vy):
        # 计算 AB 连线的向量 AB
        # dx = x2 - x1
        # dy = y2 - y1

        # 计算 AB 连线的模长 |AB|
        AB_mod = math.sqrt(dx ** 2 + dy ** 2)

        # 计算 AB 连线的单位向量 U_AB
        U_ABx = dx / AB_mod
        U_ABy = dy / AB_mod

        # 计算 A 相对于 B 的速度 V_relative
        # vx = vx1 - vx2
        # vy = vy1 - vy2

        # 计算 A 相对于 B 在 AB 连线上的速度 V_on_AB
        V_on_AB = vx * U_ABx + vy * U_ABy

        return V_on_AB

    def _cal_a_projection(self, dx, dy, vx, vy, ax, ay, x1, y1, x2, y2, v_x1, v_y1, v_x2, v_y2):
        # 计算 AB 连线的向量 AB
        # dx = x2 - x1
        # dy = y2 - y1

        # 计算 θ
        V_mod = math.sqrt(vx ** 2 + vy ** 2)
        AB_mod = math.sqrt(dx ** 2 + dy ** 2)
        if V_mod == 0 or AB_mod == 0:
            return 0

        cos_theta = (vx * dx + vy * dy) / (V_mod * AB_mod)
        theta = math.acos(cos_theta)

        # 计算 AB 连线的模长 |AB|
        AB_mod = math.sqrt(dx ** 2 + dy ** 2)

        # 计算 AB 连线的单位向量 U_AB
        U_ABx = dx / AB_mod
        U_ABy = dy / AB_mod

        # 计算 A 相对于 B 的加速度 a_relative
        # ax = ax1 - ax2
        # ay = ay1 - ay2

        # 计算 A 相对于 B 在 AB 连线上的加速度 a_on_AB
        a_on_AB = ax * U_ABx + ay * U_ABy

        VA = np.array([v_x1, v_y1])
        VB = np.array([v_x2, v_y2])
        D_A = np.array([x1, y1])
        D_B = np.array([x2, y2])
        V_r = VA - VB
        V = np.linalg.norm(V_r)
        w = self._cal_relative_angular_v(theta, D_A, D_B, VA, VB)
        a_on_AB_back = self._calculate_derivative(a_on_AB, w, V, theta)
        return a_on_AB_back

    # 计算相对加速度
    def _calculate_derivative(self, a, w, V, theta):
        # 计算(V×cos(θ))'的值
        # derivative = a * math.cos(theta) - w * V * math.sin(theta)theta
        derivative = a - w * V * math.sin(theta)
        return derivative

    def _cal_relative_angular_v(self, theta, A, B, VA, VB):
        dx = A[0] - B[0]
        dy = A[1] - B[1]
        dvx = VA[0] - VB[0]
        dvy = VA[1] - VB[1]
        # (dx * dvy - dy * dvx)
        angular_velocity = math.sqrt(dvx ** 2 + dvy ** 2) * math.sin(theta) / math.sqrt(dx ** 2 + dy ** 2)
        return angular_velocity

    # def _death_pr(self, obj_type, v_relative):
    def _death_pr(self, v_relative):
        # if obj_type == 5:
        #     p_death = 1 / (1 + np.exp(7.723 - 0.15 * v_relative))
        # else:
        p_death = 1 / (1 + np.exp(8.192 - 0.12 * v_relative))
        return p_death

    # def _cal_collisionRisk_level(self, obj_type, v_relative, collisionSeverity):
    def _cal_collisionRisk_level(self, v_relative, collisionSeverity):
        # if obj_type == 5:
        #     p_death = 1 / (1 + np.exp(7.723 - 0.15 * v_relative))
        # else:
        p_death = 1 / (1 + np.exp(8.192 - 0.12 * v_relative))
        collisionRisk = 0.4 * p_death + 0.6 * collisionSeverity
        return collisionRisk

    # 求两车之间当前距离
    def dist(self, x1, y1, x2, y2):
        dist = np.sqrt((x1 - x2) ** 2 + (y1 - y2) ** 2)
        return dist

    def _cal_count_(self, dist):
        count = 0

    # TTC (time to collision)
    def _cal_TTC(self, dist, vrel_projection_in_dist):
        if vrel_projection_in_dist == 0:
            return math.inf
        TTC = dist / vrel_projection_in_dist
        return TTC

    def _normal_distribution(self, x):
        mean = 1.32
        std_dev = 0.26
        return (1 / (math.sqrt(std_dev * 2 * math.pi))) * math.exp(-0.5 * (x - mean) ** 2 / std_dev)

    def _safe_statistic(self):
        """

        """
        self.collision_risk_dict = _cal_max_min_avg(self.collisionRisk_list)
        self.collision_severity_dict = _cal_max_min_avg(self.collisionSeverity_list)
        self.over_speed_dict = _cal_max_min_avg(self.overSpeed_list)

        arr_safe = [[self.collisionCount, self.collisionRisk, self.collisionSeverity, self.overSpeed]]
        return arr_safe

    def safe_score_new(self):
        """

        """
        score_metric_dict = {}
        score_type_dict = {}

        arr_safe = self._safe_statistic()

        print("\n[安全性表现及得分情况]")
        print("安全性各指标值：", [[round(num, 2) for num in row] for row in arr_safe])

        if arr_safe:
            arr_safe = np.array(arr_safe)
            score_model = self.scoreModel(self.kind_list, self.optimal_list, self.multiple_list, arr_safe)
            score_sub = score_model.cal_score()
            metric_list = [x for x in self.metric_list if x in self.config.builtinMetricList]
            # metric_num = len(metric_list)

            if len(self.obj_id_list) > 1 and not self.empty_flag:
                # score_sub[-metric_num:] = [num * 1.25 for num in score_sub[-metric_num:]]
                pass
            else:
                score_sub = [80] * len(score_sub)

            score_sub = list(map(lambda x: 100 if np.isnan(x) else x, score_sub))  # 对None值做特判

            score_metric = score_sub
            score_metric_dict = {key: value for key, value in zip(metric_list, score_metric)}

        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_safe = 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_safe = 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_safe = round(score_safe, 2)
        # score_type = [round(x, 2) for key, x in score_type_dict.items()]
        # score_metric = [round(x, 2) for key, x in score_metric_dict.items()]

        print("安全性各指标基准值：", self.optimal_list)
        print(f"安全性得分为：{score_safe:.2f}分。")
        print(f"安全性各类型得分为：{score_type_dict}。")
        print(f"安全性各指标得分为：{score_metric_dict}。")
        # return score_safe, score_type, score_metric
        return score_safe, score_type_dict, score_metric_dict

    def report_statistic(self):

        report_dict = {
            "name": "安全性",
            "weight": f"{self.weight * 100:.2f}%",
            # 'collisionRisk': self.collisionRisk,
            # "noObjectCar": True,
        }

        # No objects
        if self.obj_id_list == 0:
            score_safe = 80
            grade_safe = "良好"
            score_type_dict = {
                "safeCollision": 80,
                "safeSpeed": 80
            }
            score_metric_dict = {
                "collisionCount": 80,
                "collisionRisk": 80,
                "collisionSeverity": 80,
                "overSpeed": 80
            }

            description = "安全性在无目标车情况下表现良好。"

            description1 = "次数：0次"

            description2 = f"最大值：0%；" \
                           f"最小值：0%；" \
                           f"平均值：0%"

            description3 = f"最大值：0%；" \
                           f"最小值：0%；" \
                           f"平均值：0%"

            description4 = f"最大值：{100 * self.over_speed_dict['max']}%；" \
                           f"最小值：{100 * self.over_speed_dict['min']}%；" \
                           f"平均值：{100 * self.over_speed_dict['avg']}%"

        # with objects
        else:
            score_safe, score_type_dict, score_metric_dict = self.safe_score_new()
            score_safe = int(score_safe) if int(score_safe) == score_safe else round(score_safe, 2)
            grade_safe = score_grade(score_safe)

            description = f"· 在安全性方面，得分{score_safe}分，表现{grade_safe}，"
            is_good = True

            if self.collisionCount > 0:
                is_good = False
                description += f"出现{self.collisionCount}次碰撞，需重点优化。"

            if self.over_speed_dict['max'] > 0:
                is_good = False
                description += f"超速比例最大值{100 * self.over_speed_dict['max']}%，需重点优化。"

            if is_good:
                description += "速度控制良好。"

            description1 = f"次数：{self.collisionCount}次"

            description2 = f"最大值：{self.collision_risk_dict['max']}%；" \
                           f"最小值：{self.collision_risk_dict['min']}%；" \
                           f"平均值：{self.collision_risk_dict['avg']}%"

            description3 = f"最大值：{self.collision_severity_dict['max']}%；" \
                           f"最小值：{self.collision_severity_dict['min']}%；" \
                           f"平均值：{self.collision_severity_dict['avg']}%"

            description4 = f"最大值：{self.over_speed_dict['max']}%；" \
                           f"最小值：{self.over_speed_dict['min']}%；" \
                           f"平均值：{self.over_speed_dict['avg']}%"

        report_dict["score"] = score_safe
        report_dict["level"] = grade_safe
        report_dict["description"] = replace_key_with_value(description, self.name_dict)

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

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

        collisionSeverity_index = {
            "weight": self.weight_dict['collisionSeverity'],
            "score": score_metric_dict['collisionSeverity'],
            "description": description3
        }

        overSpeed_index = {
            "weight": self.weight_dict['overSpeed'],
            "score": score_metric_dict['overSpeed'],
            "description": description4
        }

        indexes_dict = {
            "collisionCount": collisionCount_index,
            "collisionRisk": collisionRisk_index,
            "collisionSeverity": collisionSeverity_index,
            "overSpeed": overSpeed_index
        }

        report_dict["indexes"] = indexes_dict

        print(report_dict)

        return report_dict

# if __name__ == "__main__":
#     pass
#
#     print("---started---")
#     count = 0
#     dataPath = './task/case0613'  # 设置路径
#     configPath = r"./config_safe.json"
#     config = ConfigParse(configPath)
#     data_processed = DataProcess(dataPath, config)  # 数据处理，包括objectstate这些
#     print("data_processed is", data_processed.object_df.playerId)  # dir(data_processed)可以查看data_processed的数据属性
#     safe_indicator = Safe(data_processed)
#     collisionRisk_list, collisionSeverity_list, collision_dist_count = safe_indicator._safe_param_cal_new()
#
#     prev_value = collision_dist_count[0]  # 初始化前一个元素为列表的第一个元素
#
#     # 遍历列表（从第二个元素开始）
#     for i in range(1, len(collision_dist_count)):
#         current_value = collision_dist_count[i]  # 当前元素
#
#         # 如果当前元素是0且与前一个元素不同，或者这是第一个元素且为0
#         if (current_value == 0 and prev_value != 0) or (i == 0 and current_value == 0):
#             count += 1  # 增加0的段的计数
#     print("collisionRisk is", max(collisionRisk_list))
#     print("collisionSeverity is", max(collisionSeverity_list))
#     print("collision count is", count)
#     print("---end---")