#!/usr/bin/env python
# -*- coding: utf-8 -*-
##################################################################
#
# Copyright (c) 2023 CICV, Inc. All Rights Reserved
#
##################################################################
"""
@Authors: zhanghaiwen(zhanghaiwen@china-icv.cn), yangzihao(yangzihao@china-icv.cn)
@Data: 2023/08/03
@Last Modified: 2023/08/03
@Summary: Function metrics
"""
import sys
sys.path.append('../common')
sys.path.append('../modules')
sys.path.append('../results')
import math
import numpy as np
import pandas as pd
import scipy.signal as sg
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 Function(object):
"""
Class for achieving function 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.status_trigger_dict = data_processed.status_trigger_dict
self.df = data_processed.object_df
self.ego_df = data_processed.ego_data
self.obj_id_list = data_processed.obj_id_list
self.df_roadmark = data_processed.road_mark_df
self.df_roadpos = data_processed.road_pos_df
self.df_drivectrl = data_processed.driver_ctrl_df
self.unfunc_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
self.unfunc_follow_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
self.unfunc_lane_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
self.unfunc_custom_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
self.custom_markline_list = []
# config infos for calculating score
self.config = data_processed.config
function_config = self.config.config['function']
self.function_config = function_config
# common data
self.bulitin_metric_list = self.config.builtinMetricList
# dimension data
self.weight_custom = function_config['weightCustom']
self.metric_list = function_config['metric']
self.type_list = function_config['type']
self.type_name_dict = function_config['typeName']
self.name_dict = function_config['name']
self.unit_dict = function_config['unit']
# custom metric data
self.customMetricParam = function_config['customMetricParam']
self.custom_metric_list = list(self.customMetricParam.keys())
self.custom_data = custom_data
self.custom_param_dict = {}
# score data
self.weight = function_config['weightDimension']
self.weight_type_dict = function_config['typeWeight']
self.weight_type_list = function_config['typeWeightList']
self.weight_dict = function_config['weight']
self.weight_list = function_config['weightList']
self.priority_dict = function_config['priority']
self.priority_list = function_config['priorityList']
self.kind_dict = function_config['kind']
self.optimal_dict = function_config['optimal']
self.multiple_dict = function_config['multiple']
self.kind_list = function_config['kindList']
self.optimal_list = function_config['optimalList']
self.multiple_list = function_config['multipleList']
# self.unit_dict = function_config['unit']
self.metric_dict = function_config['typeMetricDict']
if "functionACC" in self.metric_dict.keys():
self.acc_metric_list = self.metric_dict['functionACC']
if "functionLKA" in self.metric_dict.keys():
self.lka_metric_list = self.metric_dict['functionLKA']
# self.acc_metric_list = ['followSpeedDeviation', 'followDistanceDeviation', 'followStopDistance',
# 'followResponseTime']
# self.lka_metric_list = ['laneDistance', 'centerDistanceExpectation', 'centerDistanceStandardDeviation',
# 'centerDistanceMax', 'centerDistanceMin', 'centerDistanceFrequency',
# 'centerDistanceRange']
# custom metric
self.flag_type_dict = {}
# lists of drving control info
self.time_list = data_processed.driver_ctrl_data['time_list']
self.frame_list = data_processed.driver_ctrl_data['frame_list']
self.time_list_follow = []
self.frame_list_follow = []
self.dist_list = []
self.v_deviation_list = []
self.v_relative_list = []
self.dist_deviation_list = []
self.stop_distance_list = []
self.follow_stop_time_start_list = [] # 起停跟车响应时长
self.dist_list_full_time = list()
self.dist_deviation_list_full_time = list()
self.line_dist = []
self.center_dist = []
self.center_dist_with_nan = []
self.center_time_list = []
self.center_frame_list = []
self.ICA_FLAG = False
self.follow_stop_count = 0
self.result = {}
def _following(self, df):
"""
稳定跟车行驶：
1.稳态控制速度偏差：筛选出跟车行驶数据后，
2.稳态控制距离偏差
ACC stastus
0x7: Stand-wait State
0x6: Stand-active State
0x5: Passive State
0x4: Standby State
0x3: Shut-off State
0x2: Override State
0x1: Active State
0x0: Off
# 速度变化不能以0到非0为参考，要改为从0到0.05的变化，有一个阈值滤波，避免前车车辆抖动误差影响传感器
弯道行驶： 最小跟随弯道半径
"""
# df = self.df[self.df['ACC_status'] == "Active"].copy() # 跟车状态下
# df = self.df[self.df['ACC_status'] == 1].copy() # 跟车状态下
col_list = ['simTime', 'simFrame', 'playerId', 'v', 'posX', 'posY'] # target_id
df = df[col_list].copy()
ego_df = df[df['playerId'] == 1][['simTime', 'simFrame', 'v', 'posX', 'posY']]
# 筛选目标车（同一车道内，距离最近的前车）
# obj_df = df[df['playerId'] == df['target_id']]
target_id = 2
obj_df = df[df['playerId'] == target_id][['simTime', 'simFrame', 'v', 'posX', 'posY']] # 目标车
obj_df = obj_df.rename(columns={'v': 'v_obj', 'posX': 'posX_obj', 'posY': 'posY_obj'})
df_merge = pd.merge(ego_df, obj_df, on=['simTime', 'simFrame'], how='left')
df_merge['v_relative'] = df_merge['v'] - df_merge['v_obj']
df_merge['dist'] = df_merge.apply(
lambda row: self.dist(row['posX'], row['posY'], row['posX_obj'], row['posY_obj']), axis=1)
self.dist_list = df_merge['dist'].values.tolist()
df_merge['time_gap'] = df_merge['dist'] / df_merge['v']
SAFE_TIME_GAP = 3
df_merge['dist_deviation'] = df_merge['time_gap'].apply(
lambda x: 0 if (x >= SAFE_TIME_GAP) else (SAFE_TIME_GAP - x))
df_stop = df_merge[(df_merge['v'] == 0) & (df_merge['v_obj'] == 0)]
stop_distance_list = df_stop['dist'].values.tolist()
self.stop_distance_list = stop_distance_list
t_list = df_stop['simTime'].values.tolist()
stop_group = []
sub_group = []
CONTINUOUS_TIME_PERIOD = 1
CONTINUOUS_FRAME_PERIOD = 13
for i in range(len(t_list)):
if not sub_group or t_list[i] - t_list[i - 1] <= CONTINUOUS_TIME_PERIOD:
sub_group.append(t_list[i])
else:
stop_group.append(sub_group)
sub_group = [t_list[i]]
stop_group.append(sub_group)
stop_group = [g for g in stop_group if len(g) >= CONTINUOUS_FRAME_PERIOD]
self.follow_stop_count = len(stop_group)
# solution 1: 跟车状态下，算法输出预设跟车速度
# df['velocity_deviation'] = abs(df['v'] - df['set_velocity'])
# max_velocity_deviation = df['velocity_deviation'].max()
# solution 2: 跟车状态下，跟车设定速度为目标车速度
# velocity_deviation = []
# distance_deviation = []
# stop_distance = []
#
# for f, data in df.groupby('simFrame'):
# ego_data = data.iloc[0].copy()
# # df.loc[len(df)] = ego_data
#
# if len(data) < 2:
# continue
# v1 = ego_data['v']
# x1 = ego_data['posX']
# y1 = ego_data['posY']
#
# for i in range(1, len(data)):
# obj_data = data.iloc[i].copy()
#
# v2 = obj_data['v']
# x2 = obj_data['posX']
# y2 = obj_data['posY']
#
# v_delta = abs(v1 - v2)
# velocity_deviation.append(v_delta)
#
# dist = self.dist(x1, y1, x2, y2)
# distance_deviation.append(dist)
#
# if v2 == 0 and v1 == 0:
# stop_distance.append(dist)
# df.loc[len(df)] = obj_data
# self.df = df
df_merge.replace([np.inf, -np.inf], np.nan, inplace=True) # 异常值处理
self.time_list_follow = df_merge['simTime'].values.tolist()
self.frame_list_follow = df_merge['simFrame'].values.tolist()
self.v_relative_list = df_merge['v_relative'].values.tolist()
self.v_deviation_list = abs(df_merge['v_relative']).values.tolist()
self.dist_deviation_list = df_merge['dist_deviation'].values.tolist()
tmp_df = self.ego_df[['simTime', 'simFrame']].copy()
v_rel_df = df_merge[['simTime', 'v_relative']].copy()
df_merged1 = pd.merge(tmp_df, v_rel_df, on='simTime', how='left')
self.v_relative_list_full_time = df_merged1['v_relative'].values.tolist()
dist_deviation_df = df_merge[['simTime', 'dist_deviation']].copy()
df_merged1 = pd.merge(tmp_df, dist_deviation_df, on='simTime', how='left')
self.dist_deviation_list_full_time = df_merged1['dist_deviation'].values.tolist()
dist_df = df_merge[['simTime', 'dist']].copy()
df_merged1 = pd.merge(tmp_df, dist_df, on='simTime', how='left')
self.dist_list_full_time = df_merged1['dist'].values.tolist()
max_velocity_deviation = abs(df_merge['v_relative']).max()
distance_deviation = df_merge['dist_deviation'].max()
min_stop_distance = min(self.stop_distance_list) if self.stop_distance_list else 9999
self.result['followSpeedDeviation'] = max_velocity_deviation
self.result['followDistanceDeviation'] = distance_deviation
self.result['followStopDistance'] = min_stop_distance
def _stop_and_go(self, df):
"""
停走功能：
1.跟停距离3-4m，
2.启动跟车响应时间<=1.2s
3.停车跟车响应时间<=1.2s
decisionType_target
0 无障碍物巡航
1 停车减速让行
2 跟车
3 绕行
4 到达终点？
"""
# df = self.df[self.df['ACC_status'].isin(["Active", "Stand-active", "Stand-wait"])].copy() # 跟车状态下
# df = self.df[self.df['ACC_status'] == 1].copy() # 跟车状态下
df = df.dropna(subset=['type'])
STOP_SPEED_THRESHOLD = 0.05
df['v'] = df['v'].apply(lambda x: 0 if x <= STOP_SPEED_THRESHOLD else 1) # 区分速度为0或非0
target_id = 2
df_ego = df[df['playerId'] == 1].copy()
df_obj = df[df['playerId'] == target_id].copy() # 目标车
df_obj_time = df_obj['simTime'].values.tolist()
df_ego = df_ego[df_ego['simTime'].isin(df_obj_time)].copy()
df_ego = df_ego.drop_duplicates(["simTime", "simFrame"])
df_obj = df_obj.drop_duplicates(["simTime", "simFrame"])
df_ego['v_diff'] = df_ego['v'].diff()
df_ego['v_start_flag'] = df_ego['v_diff'].apply(lambda x: 1 if x == 1 else 0) # 起步即为1
df_ego['v_stop_flag'] = df_ego['v_diff'].apply(lambda x: 1 if x == -1 else 0) # 停车即为-1
df_obj['v_diff'] = df_obj['v'].diff()
obj_v_start_flag = df_obj['v_diff'].apply(lambda x: 1 if x == 1 else 0).values # 起步即为1
obj_v_stop_flag = df_obj['v_diff'].apply(lambda x: 1 if x == -1 else 0).values # 停车即为1
df_ego['obj_v_start_flag'] = obj_v_start_flag
df_ego['obj_v_stop_flag'] = obj_v_stop_flag
df_ego['flag_start'] = df_ego['obj_v_start_flag'] - df_ego['v_start_flag'] # 目标车起步即为1，自车起步即为-1
df_ego['flag_stop'] = df_ego['obj_v_stop_flag'] - df_ego['v_stop_flag'] # 目标车停车即为1，自车停车即为-1
flag_start_list = df_ego['flag_start'].values
flag_stop_list = df_ego['flag_stop'].values
time_list = df_ego['simTime'].values
time_start_list = []
time_stop_list = []
for i, flag in enumerate(flag_start_list):
if flag:
t1 = time_list[i]
if flag == -1:
t2 = time_list[i]
time_start_list.append(t2 - t1) # t2-t1即为自车起步响应时间
for i, flag in enumerate(flag_stop_list):
if flag:
t1 = time_list[i]
if flag == -1:
t2 = time_list[i]
time_stop_list.append(t2 - t1) # t2-t1即为自车停车响应时间
time_start_list = [i for i in time_start_list if i != 0]
self.result['followResponseTime'] = max(time_start_list) if time_start_list else 0
self.follow_stop_time_start_list = time_start_list
# self.result['跟车停止响应最长时间'] = max(time_stop_list)
def dist(self, x1, y1, x2, y2):
dis = math.sqrt((x1 - x2) ** 2 + (y1 - y2) ** 2)
return dis
def _line_dist_merge(self):
# line_dist_df = self.df_roadmark[self.df_roadmark['id'] == 0 & self.df_roadmark['id'] == 2].copy()
# line_dist_group = line_dist_df.groupby("simFrame").apply(lambda t: abs(t['lateralDist']).min()).values.tolist()
# line_dist_df['line_dist'] = line_dist_group
# line_dist_df = line_dist_df[['simFrame', 'line_dist']]
# line_dist_df.columns = ['simFrame', 'line_dist']
# line_dist_df = line_dist_df.drop_duplicates()
line_dist_df = self.df_roadmark[(self.df_roadmark['id'] == 0) | (self.df_roadmark['id'] == 2)].copy()
# df = line_dist_df[line_dist_df['id'] == 2][['simTime', 'simFrame']].copy()
# df['line_dist'] = line_dist_df.groupby('simFrame').apply(lambda t: abs(t['lateralDist']).min()).values.tolist()
if line_dist_df.empty:
self.ego_df['line_dist'] = pd.Series(dtype=float)
else:
df = line_dist_df.groupby('simFrame').apply(lambda t: abs(t['lateralDist']).min()).reset_index()
# df = df.rename(columns={'lateralDist': 'line_dist'})
df.columns = ["simFrame", "line_dist"]
self.ego_df = pd.merge(self.ego_df, df, on='simFrame', how='left')
def _lane_transverse_distance(self, df):
"""
0x7:Reserved
0x6: Reserved
0x5: Reserved
0x4: Error
0x3: Active
0x2: Standby
0x1: Passive
0x0: Off
"""
# 车道内偏移行驶时，与近侧车道线的横向距离
# data_group = lane_df.groupby("simFrame").apply(lambda t: abs(t['lateralDist']).min())
# data_group = self.df_roadmark.groupby("simFrame").apply(lambda t: abs(t['lateralDist']).min())
# df = data_group.to_frame(name='line_dist')
# df['simFrame'] = data_group.index.tolist()
# df = self.df_roadmark[self.df_roadmark['id'] == 0][['simTime', 'simFrame']].copy()
# df['line_dist'] = self.df_roadmark.groupby('simFrame').apply(
# lambda t: abs(t['lateralDist']).min()).values.tolist()
# self.ego_df = pd.merge(self.ego_df, df, on=['simTime', 'simFrame'], how='left')
# self._line_dist_merge()
# # df = self.df[self.df['LKA_status'] == 3]
# df = self.ego_df[self.ego_df['LKA_status'] == "Active"]
self.line_dist = df['line_dist'].to_list()
self.line_dist = [x for x in self.line_dist if not np.isnan(x)]
self.result['laneDistance'] = min(self.line_dist) if self.line_dist else self.optimal_dict['laneDistance']
def _lateral_dist_of_center_line(self, df_laneoffset):
# df_laneoffset = self.df[(self.df['LKA_status'] == "Active") & (self.df['playerId'] == 1)]
self.center_dist_with_nan = df_laneoffset['laneOffset'].to_list()
self.center_time_list = df_laneoffset['simTime'].to_list()
self.center_frame_list = df_laneoffset['simFrame'].to_list()
# self.center_dist_df = self.df_roadpos[self.df_roadpos["playerId"] == 1]
# self.center_dist = self.center_dist_df["laneOffset"]
# self.center_time_list = self.center_dist_df['simTime'].to_list()
# self.center_frame_list = self.center_dist_df['simFrame'].to_list()
self.center_dist = [x for x in self.center_dist_with_nan if not np.isnan(x)]
if not self.center_dist:
self.result['centerDistanceExpectation'] = 0
self.result['centerDistanceStandardDeviation'] = 0
# self.result['centerDistanceMax'] = abs(extreme_max_value).max()
# self.result['centerDistanceMin'] = abs(extreme_min_value).min()
self.result['centerDistanceMax'] = 0
self.result['centerDistanceMin'] = 0
self.result['centerDistanceFrequency'] = 0
self.result['centerDistanceRange'] = 0
else:
center_dist = [abs(x) for x in self.center_dist]
# 车道中心线横向距离分布期望与标准差
E_lane_center_dist = np.mean(center_dist)
D_lane_center_dist = np.std(center_dist)
# 车道中心线横向距离分布极值
center_dist = np.array(center_dist)
extrmax_index = sg.argrelmax(center_dist)[0]
extrmin_index = sg.argrelmin(center_dist)[0]
extreme_max_value = center_dist[extrmax_index]
extreme_min_value = center_dist[extrmin_index]
# 横向相对位置震荡频率
# 极值即函数局部的最大值或最小值，周期表示两个极大值之间的时间长度或者两个极小值之间的时间长度
length1 = len(extrmax_index) # 极大值列表长度
length2 = len(extrmin_index) # 极小值列表长度
# 极值下标之间的时间间隔列表乘以每两个下标之间的时间间隔
FREQUENCY = 100
time_diff = 1.0 / FREQUENCY
period1 = time_diff * (extrmax_index[1:length1] - extrmax_index[0:length1 - 1])
period2 = time_diff * (extrmin_index[1:length2] - extrmin_index[0:length2 - 1])
try:
period = (sum(period1) + sum(period2)) / (length1 + length2) # 所有时间间隔(即周期)的均值作为最后的周期
except ZeroDivisionError:
period = 0
frequency = 1 / period if period else 0
# length = min(len(extreme_max_value), len(extreme_min_value))
# period = np.mean(extrmax_index[0:length] - extrmin_index[0:length])
# frequency = 1 / period
# 横向相对位置震荡极差
length = min(len(extreme_max_value), len(extreme_min_value))
extr_diff = extreme_max_value[0:length] - extreme_min_value[0:length]
self.result['centerDistanceExpectation'] = E_lane_center_dist
self.result['centerDistanceStandardDeviation'] = D_lane_center_dist
# self.result['centerDistanceMax'] = abs(extreme_max_value).max()
# self.result['centerDistanceMin'] = abs(extreme_min_value).min()
self.result['centerDistanceMax'] = center_dist.max()
self.result['centerDistanceMin'] = center_dist.min()
self.result['centerDistanceFrequency'] = abs(frequency)
self.result['centerDistanceRange'] = abs(extr_diff).max() if list(extr_diff) else 0
def _continous_judge(self, frame_list):
CONTINUOUS_FRAME_PERIOD = 3
if not frame_list:
return 0
cnt = 1
for i in range(1, len(frame_list)):
if frame_list[i] - frame_list[i - 1] <= CONTINUOUS_FRAME_PERIOD:
continue
cnt += 1
return cnt
def _lane_departure_warning(self, ldw_df):
"""
工作车速范围：55/60-200/250 km/h
最小可用车道宽度：2.6m
误报警：距离安全，但是预警
漏报警：距离危险，但是没报警
灵敏度差：线外40cm
灵敏度零级：压线
灵敏度一级：线内20cm（20%）
灵敏度二级：线内40cm（40%）
灵敏度三级：线内60cm（60%）
偏离车速1.0m/s：线内60cm报警
偏离车速0.5m/s：线内40cm报警
"""
# 自车, 车道线宽度>2.6m，转向灯没开
# ego_df = self.df[self.df['playerId'] == 1].copy()
# ego_df['line_dist'] = self.line_dist
# lane_dist_df = self.df_roadmark[self.df_roadmark['id'] == 0 & self.df_roadmark['id'] == 2].copy()
# lane_dist_group = lane_dist_df.groupby("simFrame").apply(
# lambda t: abs(t['lateralDist']).min()).values.tolist()
# lane_dist_df['line_dist'] = lane_dist_group
# lane_dist_df = lane_dist_df[['simFrame', 'line_dist']]
# lane_dist_df.columns = ['simFrame', 'line_dist']
# # lane_dist_df = lane_dist_df[['simTime', 'simFrame', 'line_dist']]
# # lane_dist_df.columns = ['simTime', 'simFrame', 'line_dist']
# lane_dist_df = lane_dist_df.drop_duplicates()
# ldw_status_df = self.df[self.df['playerId'] == 1].copy()[['simTime', 'simFrame', 'LDW_status']].drop_duplicates()
# ldw_status_df = self.ego_df[['simTime', 'simFrame', 'LDW_status', 'line_dist']].copy().drop_duplicates()
# ldw_df = pd.merge_asof(ldw_status_df, lane_dist_df, on='simFrame', direction='nearest')
# ldw_df = self.ego_df[['simTime', 'simFrame', 'LDW_status', 'line_dist']].copy()
# calculate max DLC
warning_dist_max = ldw_df[ldw_df['LDW_status'] == "Active"]['line_dist'].max()
# warning_dist_max = ldw_df[ldw_df['LDW_status'] == 3]['line_dist'].max()
self.result['预警时距离车道线最大距离'] = warning_dist_max
# count miss warning
miss_warning_df = ldw_df[(ldw_df['line_dist'] <= 0.4) & (ldw_df['LDW_status'] != "Active")]
miss_warning_frame_list = miss_warning_df['simFrame'].values.tolist()
miss_warning_count = self._continous_judge(miss_warning_frame_list)
self.result['车道偏离漏预警次数'] = miss_warning_count
# count false warning
false_warning_df = ldw_df[(ldw_df['line_dist'] >= 0.4) & (ldw_df['LDW_status'] == "Active")]
false_warning_frame_list = false_warning_df['simFrame'].values.tolist()
false_warning_count = self._continous_judge(false_warning_frame_list)
self.result['车道偏离误预警次数'] = false_warning_count
def _autonomous_emergency_brake(self):
"""
刹得准、刹得稳
碰撞预警 + 紧急制动
<=4s 预警>1s，制动<3s
触发目标类型
目标类型准确度
目标为车辆的速度降：
弱势目标的速度降：
触发时TTC<4s
一级制动减速度：4m/s^2
二级制动减速度：10m/s^2
刹停行驶距离：10-20m
刹停时前车距离：0.8-1.5m
误触发次数：频率 n/10万km
漏触发：
能绕开就没必要刹停：开始刹车时距离远近
Returns:
"""
# ego_df = self.df[self.df['playerId'] == 1]
df = self.ego_df[self.ego_df['Aeb_status'] == 'Active']
# df = self.ego_df[self.ego_df['Aeb_status'] == 0]
# calculate brakeDecelerate
df['lon_accel'] = df['accelX'] * np.cos(df['posH']) + df['accelY'] * np.sin(df['posH'])
decelerate_max = df['lon_accel'].min()
# calculate brake stop travelDist
df_dist = df[df['lon_accel'] < 0]
frame_list = df_dist['simFrame'].to_list()
travelDist_list = df_dist['travelDist'].to_list()
travelDist_group = []
start = 0
for i in range(1, len(frame_list)):
if frame_list[i] - frame_list[i - 1] > 5:
end = i - 1
# travelDist_group.append(travelDist_list[start:end])
travelDist_group.append(travelDist_list[end] - travelDist_list[start])
start = i
end = -1
travelDist_group.append(travelDist_list[end] - travelDist_list[start])
brake_distance_max = max(travelDist_group) if travelDist_group else np.inf
self.result['brakeDecelerate'] = abs(round(decelerate_max, 4))
self.result['brakeDistance'] = round(brake_distance_max, 4)
def _integrated_cruise_assist(self, df):
df_LCC = df[df['ICA_status'].isin(['LLC_follow_vehicle', 'LLC_follow_line'])].copy()
self.LCC_line_dist = df_LCC['line_dist'].to_list()
self.result['LCClaneDistance'] = min(self.LCC_line_dist) if self.LCC_line_dist else self.optimal_dict[
'LCClaneDistance']
self.LCC_center_dist = df_LCC['laneOffset'].to_list()
self.LCC_center_time_list = df_LCC['simTime'].to_list()
self.LCC_center_frame_list = df_LCC['simFrame'].to_list()
if not self.LCC_center_dist:
self.result['LCCcenterDistanceExpectation'] = 0
self.result['LCCcenterDistanceStandardDeviation'] = 0
# self.result['centerDistanceMax'] = abs(extreme_max_value).max()
# self.result['centerDistanceMin'] = abs(extreme_min_value).min()
self.result['LCCcenterDistanceMax'] = 0
self.result['LCCcenterDistanceMin'] = 0
self.result['LCCcenterDistanceFrequency'] = 0
self.result['LCCcenterDistanceRange'] = 0
else:
center_dist = [abs(x) for x in self.LCC_center_dist]
# 车道中心线横向距离分布期望与标准差
E_lane_center_dist = np.mean(center_dist)
D_lane_center_dist = np.std(center_dist)
# 车道中心线横向距离分布极值
center_dist = np.array(center_dist)
extrmax_index = sg.argrelmax(center_dist)[0]
extrmin_index = sg.argrelmin(center_dist)[0]
extreme_max_value = center_dist[extrmax_index]
extreme_min_value = center_dist[extrmin_index]
# 横向相对位置震荡频率
# 极值即函数局部的最大值或最小值，周期表示两个极大值之间的时间长度或者两个极小值之间的时间长度
length1 = len(extrmax_index) # 极大值列表长度
length2 = len(extrmin_index) # 极小值列表长度
# 极值下标之间的时间间隔列表乘以每两个下标之间的时间间隔
FREQUENCY = 100
time_diff = 1.0 / FREQUENCY
period1 = time_diff * (extrmax_index[1:length1] - extrmax_index[0:length1 - 1])
period2 = time_diff * (extrmin_index[1:length2] - extrmin_index[0:length2 - 1])
try:
period = (sum(period1) + sum(period2)) / (length1 + length2) # 所有时间间隔(即周期)的均值作为最后的周期
except ZeroDivisionError:
period = 0
frequency = 1 / period if period else 0
# length = min(len(extreme_max_value), len(extreme_min_value))
# period = np.mean(extrmax_index[0:length] - extrmin_index[0:length])
# frequency = 1 / period
# 横向相对位置震荡极差
length = min(len(extreme_max_value), len(extreme_min_value))
extr_diff = extreme_max_value[0:length] - extreme_min_value[0:length]
self.result['LCCcenterDistanceExpectation'] = E_lane_center_dist
self.result['LCCcenterDistanceStandardDeviation'] = D_lane_center_dist
# self.result['centerDistanceMax'] = abs(extreme_max_value).max()
# self.result['centerDistanceMin'] = abs(extreme_min_value).min()
self.result['LCCcenterDistanceMax'] = center_dist.max()
self.result['LCCcenterDistanceMin'] = center_dist.min()
self.result['LCCcenterDistanceFrequency'] = abs(frequency)
self.result['LCCcenterDistanceRange'] = abs(extr_diff).max() if list(extr_diff) else 0
df_LC = df[df['ICA_status'] == 'Only_Longitudinal_Control'].copy()
v_max = df_LC['v'].max()
v_min = df_LC['v'].min()
self.result['OLC_v_deviation'] = v_max - v_min
def _function_statistic(self):
"""
"""
ACC_optimal_list = []
if "functionACC" in self.type_list:
for function_type in self.type_list:
if "ICA" in function_type:
continue
if len(self.obj_id_list) > 1:
# 给定的双层列表
follow_time_ranges = self.status_trigger_dict['ACC']['ACC_active_time']
# 使用Pandas的between函数（注意between是闭区间）结合列表推导来过滤
# 这里需要遍历所有时间范围，然后使用`|`（逻辑或）将它们的结果合并
filtered_df = pd.DataFrame(columns=self.df.columns)
for start, end in follow_time_ranges:
mask = (self.df['simTime'] >= start) & (self.df['simTime'] <= end)
if filtered_df is None:
filtered_df = self.df[mask]
else:
filtered_df = pd.concat([filtered_df, self.df[mask]])
# 上述方式可能引入重复行，使用drop_duplicates去重
df_follow = filtered_df.drop_duplicates()
df_stop_and_go = df_follow.copy()
# 重置索引（如果需要）
# filtered_df.reset_index(drop=True, inplace=True)
# df_follow111 = self.df[self.df['ACC_status'] == "Active"].copy() # 跟车状态下
self._following(df_follow)
# df_stop_and_go = self.df[
# self.df['ACC_status'].isin(["Active", "Stand-active", "Stand-wait"])].copy() # 跟车状态下
self._stop_and_go(df_stop_and_go)
if df_follow.empty and df_stop_and_go.empty:
self.flag_type_dict['functionACC'] = False
else:
self.flag_type_dict['functionACC'] = True
else:
ACC_optimal_list = [value for key, value in self.optimal_dict.items() if key in self.acc_metric_list]
if "functionLKA" in self.type_list:
self._line_dist_merge()
# df_line_dist = self.df[self.df['LKA_status'] == 3]
# 给定的双层列表
follow_time_ranges = self.status_trigger_dict['LKA']['LKA_active_time']
# 使用Pandas的between函数（注意between是闭区间）结合列表推导来过滤
# 这里需要遍历所有时间范围，然后使用`|`（逻辑或）将它们的结果合并
filtered_df = pd.DataFrame(columns=self.ego_df.columns)
for start, end in follow_time_ranges:
mask = (self.ego_df['simTime'] >= start) & (self.ego_df['simTime'] <= end)
if filtered_df is None:
filtered_df = self.ego_df[mask]
else:
filtered_df = pd.concat([filtered_df, self.ego_df[mask]])
# 上述方式可能引入重复行，使用drop_duplicates去重
df_lka = filtered_df.drop_duplicates()
# df_lka = self.ego_df[self.ego_df['LKA_status'] == "Active"]
if df_lka.empty:
self.flag_type_dict['functionLKA'] = False
else:
self.flag_type_dict['functionLKA'] = True
self._lane_transverse_distance(df_lka)
self._lateral_dist_of_center_line(df_lka)
# if "functionLDW" in self.type_list:
# df_ldw = self.ego_df[['simTime', 'simFrame', 'LDW_status', 'line_dist']].copy()
# self._lane_departure_warning(df_ldw)
# if "functionAEB" in self.type_list:
# self._autonomous_emergency_brake()
# if self.df['ICA_status'].nunique() != 1:
# self.ICA_FLAG = True
# # if "functionICA" in self.type_list:
# # self._integrated_cruise_assist(self.ego_df)
# # ACC
# if len(self.obj_id_list) > 1:
# df_follow = self.df[self.df['ACC_status'] == "Shut_off"].copy() # 数字3，对应Active
# self._following(df_follow)
#
# df_stop_and_go = self.df[
# self.df['ACC_status'].isin(["Shut_off", "Stand-active", "Stand-wait"])].copy() # 跟车状态下
# self._stop_and_go(df_stop_and_go)
#
# else:
# ACC_optimal_list = [value for key, value in self.optimal_dict.items() if key in self.acc_metric_list]
#
# # LKA
# self._line_dist_merge()
# # df_line_dist = self.df[self.df['LKA_status'] == 3]
# df_lka = self.ego_df[self.ego_df['ICA_status'] == "LLC_Follow_Line"]
# self._lane_transverse_distance(df_lka)
# self._lateral_dist_of_center_line(df_lka)
arr_func = [value for key, value in self.result.items() if key in self.metric_list]
# arr_func = list(self.result.values())
if "functionACC" in self.type_list and len(self.obj_id_list) == 1:
arr_func = ACC_optimal_list + arr_func
return arr_func
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 func_score(self):
score_metric_dict = {}
score_type_dict = {}
arr_func = self._function_statistic()
print("\n[功能性表现及得分情况]")
print("功能性各指标值：", [round(num, 2) for num in arr_func])
if arr_func:
arr_func = np.array([arr_func])
score_model = self.scoreModel(self.kind_list, self.optimal_list, self.multiple_list, arr_func)
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)}
flag_custom_type_dict = {}
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)
if 'statusFlag' in self.custom_data[metric].keys():
custom_type, flag_custom_type = next(iter(self.custom_data[metric]['statusFlag'].items()))
if (custom_type in flag_custom_type_dict) and (flag_custom_type == True):
flag_custom_type_dict[custom_type] = True
else:
flag_custom_type_dict[custom_type] = flag_custom_type
flag_custom_type_dict1 = {k: v for k, v in flag_custom_type_dict.items() if v == True}
flag_builtin_type_dict2 = {k: v for k, v in self.flag_type_dict.items() if v == True}
flag_builtin_type_dict2.update(flag_custom_type_dict1)
self.type_list = list(flag_builtin_type_dict2.keys())
self.metric_list = []
for type in self.type_list:
self.metric_list.extend(self.metric_dict[type])
score_metric_dict = {key: score_metric_dict[key] for key in self.metric_list}
score_metric = list(score_metric_dict.values())
# no score in list
if not score_metric:
return 0, {}, {}
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_function = 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_function = 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_function = round(score_function, 2)
print(f"功能性得分为：{score_function:.2f}分。")
print(f"功能性各类型得分为：{score_type_dict}分。")
print(f"功能性各指标得分为：{score_metric_dict}。")
return score_function, score_type_dict, score_metric_dict
def continuous_group(self, df):
time_list = df['simTime'].values.tolist()
frame_list = df['simFrame'].values.tolist()
group_time = []
group_frame = []
sub_group_time = []
sub_group_frame = []
for i in range(len(frame_list)):
if not sub_group_time or frame_list[i] - frame_list[i - 1] <= 1:
sub_group_time.append(time_list[i])
sub_group_frame.append(frame_list[i])
else:
group_time.append(sub_group_time)
group_frame.append(sub_group_frame)
sub_group_time = [time_list[i]]
sub_group_frame = [frame_list[i]]
group_time.append(sub_group_time)
group_frame.append(sub_group_frame)
group_time = [g for g in group_time if len(g) >= 2]
group_frame = [g for g in group_frame if len(g) >= 2]
# 输出图表值
time = [[g[0], g[-1]] for g in group_time]
frame = [[g[0], g[-1]] for g in group_frame]
unfunc_time_df = pd.DataFrame(time, columns=['start_time', 'end_time'])
unfunc_frame_df = pd.DataFrame(frame, columns=['start_frame', 'end_frame'])
unfunc_df = pd.concat([unfunc_time_df, unfunc_frame_df], axis=1)
return unfunc_df
def unfunctional_follow_v_deviation_df_statistic(self):
unfunc_df = pd.DataFrame({'simTime': self.time_list_follow, 'simFrame': self.frame_list_follow,
'v_deviation': self.v_deviation_list})
unfunc_df = unfunc_df[unfunc_df['simFrame'] > 1]
v_df = unfunc_df[unfunc_df['v_deviation'] > self.optimal_dict['followSpeedDeviation']]
v_df = v_df[['simTime', 'simFrame', 'v_deviation']]
v_follow_df = self.continuous_group(v_df)
# v_follow_df['type'] = 'follow'
# v_follow_df['type'] = 'ACC'
v_follow_df['type'] = f"{self.type_name_dict['functionACC']}"
self.unfunc_follow_df = pd.concat([self.unfunc_follow_df, v_follow_df], ignore_index=True)
def unfunctional_follow_dist_deviation_df_statistic(self):
unfunc_df = pd.DataFrame({'simTime': self.time_list_follow, 'simFrame': self.frame_list_follow,
'dist_deviation': self.dist_deviation_list})
unfunc_df = unfunc_df[unfunc_df['simFrame'] > 1]
dist_df = unfunc_df[unfunc_df['dist_deviation'] > self.optimal_dict['followDistanceDeviation']] # 阈值由1改为了3
dist_df = dist_df[['simTime', 'simFrame', 'dist_deviation']]
dist_follow_df = self.continuous_group(dist_df)
# v_follow_df['type'] = 'follow'
# dist_follow_df['type'] = 'ACC'
dist_follow_df['type'] = f"{self.type_name_dict['functionACC']}"
self.unfunc_follow_df = pd.concat([self.unfunc_follow_df, dist_follow_df], ignore_index=True)
def unfunctional_lane_df_statistic(self):
unfunc_df = pd.DataFrame(
{'simTime': self.center_time_list, 'simFrame': self.center_frame_list,
'center_dist': self.center_dist_with_nan})
unfunc_df = unfunc_df[unfunc_df['simFrame'] > 1]
unfunc_df = unfunc_df.dropna(subset=['center_dist'])
lane_df = unfunc_df[abs(unfunc_df['center_dist']) > self.optimal_dict['centerDistanceMax']]
lane_df = lane_df[['simTime', 'simFrame', 'center_dist']]
dist_lane_df = self.continuous_group(lane_df)
# dist_lane_df['type'] = 'lane'
# dist_lane_df['type'] = 'LKA'
dist_lane_df['type'] = f"{self.type_name_dict['functionLKA']}"
self.unfunc_lane_df = pd.concat([self.unfunc_lane_df, dist_lane_df], ignore_index=True)
# def zip_time_pairs(self, zip_list, upper_limit=9999):
# zip_time_pairs = zip(self.time_list, zip_list)
# zip_vs_time = [[x, upper_limit if y > upper_limit else y] for x, y in zip_time_pairs if not math.isnan(y)]
# return zip_vs_time
def zip_time_pairs(self, zip_list):
zip_time_pairs = zip(self.time_list, zip_list)
zip_vs_time = [[x, "" if math.isnan(y) else y] for x, y in zip_time_pairs]
return zip_vs_time
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):
# report_dict = {
# "name": "功能性",
# "weight": f"{self.weight * 100:.2f}%",
# "weightDistribution": weight_distribution,
# "score": score_function,
# "level": grade_function,
# 'score_type': score_type,
# 'score_metric': score_metric,
# 'followStopCount': self.follow_stop_count,
# "description1": func_description1,
# "description2": func_description2,
#
# "functionACC": acc_dict,
# "functionLKA": lka_dict,
#
# "speData": [ego_speed_vs_time, obj_speed_vs_time, rel_speed_vs_time],
# "accData": [lat_acc_vs_time, lon_acc_vs_time],
#
# }
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)
throttle_vs_time = self.zip_time_pairs(throttlePedal_list)
steering_vs_time = self.zip_time_pairs(steeringWheel_list)
report_dict = {
"name": self.config.dimension_name["function"],
"weight": f"{self.weight * 100:.2f}%",
'followStopCount': self.follow_stop_count,
}
# upper_limit = 40
times_upper = 2
# len_time = len(self.time_list)
duration = self.time_list[-1]
# score_function, score_type, score_metric = self.func_score()
score_function, score_type_dict, score_metric_dict = self.func_score()
# no metric
if not score_metric_dict:
return {}
# get weight distribution
report_dict["weightDistribution"] = self._get_weight_distribution("function")
score_function = int(score_function) if int(score_function) == score_function else round(
score_function, 2)
# score_type = [int(n) if int(n) == n else n for key, n in score_type_dict.items()]
# score_metric = [int(n) if int(n) == n else n for key, n in score_metric_dict.items()]
grade_function = score_grade(score_function)
report_dict["score"] = score_function
report_dict["level"] = grade_function
# report_dict["score_type"] = score_type
# report_dict["score_metric"] = score_metric
# speed data
ego_speed_list = self.ego_df['v'].values.tolist()
ego_speed_vs_time = self.zip_time_pairs(ego_speed_list)
obj_speed_vs_time = []
rel_speed_vs_time = []
# report_dict["speData"] = [ego_speed_vs_time, obj_speed_vs_time, rel_speed_vs_time]
# accData
lat_acc_list = self.ego_df['lat_acc'].values.tolist()
lat_acc_vs_time = self.zip_time_pairs(lat_acc_list)
lon_acc_list = self.ego_df['lon_acc'].values.tolist()
lon_acc_vs_time = self.zip_time_pairs(lon_acc_list)
# acc data
# lat_acc_list = self.ego_df['lat_acc'].values.tolist()
# lat_acc_vs_time = self.zip_time_pairs(lat_acc_list)
# lon_acc_list = self.ego_df['lon_acc'].values.tolist()
# lon_acc_vs_time = self.zip_time_pairs(lon_acc_list)
# report_dict["accData"] = [lat_acc_vs_time, lon_acc_vs_time]
# statistic type report infos
unfunc_metric_list = []
func_over_optimal = []
# function description
func_type_list = []
unfunc_type_list = []
type_details_dict = {}
for type in self.type_list:
if type == "functionACC":
builtin_graph_dict = {}
custom_graph_dict = {}
if len(self.obj_id_list) == 1:
follow_description1 = "无目标车数据可计算；"
follow_description2 = ""
follow_description3 = "无目标车数据可计算；"
follow_description4 = "无目标车数据可计算；"
acc_dict_indexes = {}
for metric in self.metric_dict[type]:
acc_dict_indexes[metric] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": 80,
"avg": "-",
"max": "-",
"min": "-",
}
if self.kind_dict[metric] == -1:
acc_dict_indexes[metric]["range"] = f"[0, {self.optimal_dict[metric]}]"
elif self.kind_dict[metric] == 1:
acc_dict_indexes[metric]["range"] = f"[{self.optimal_dict[metric]}, inf)"
elif self.kind_dict[metric] == 0:
acc_dict_indexes[metric][
"range"] = f"[{self.optimal_dict[metric] * self.multiple_dict[metric][0]}, {self.optimal_dict[metric] * self.multiple_dict[metric][1]}]"
acc_dict = {
"name": f"{self.type_name_dict[type]}",
"score": 80,
"level": "良好",
"description1": follow_description1,
"description2": follow_description2,
"description3": follow_description3,
"description4": follow_description4,
"noObjectCar": True,
"indexes": acc_dict_indexes,
"builtin": {},
"custom": {}
}
# follow cruise description
func_follow_metric_list = []
unfunc_follow_metric_list = []
str_follow_over_optimal = ''
distance_vs_time = self.zip_time_pairs(self.dist_list_full_time)
else:
acc_dict = {
"name": f"{self.type_name_dict[type]}",
"noObjectCar": False,
}
# follow dict
score_follow = score_type_dict[type]
grade_follow = score_grade(score_follow)
acc_dict["score"] = score_follow
acc_dict["level"] = grade_follow
unfunc_type_list.append(type) if score_type_dict[type] < 80 else func_type_list.append(type)
# follow cruise description
func_follow_metric_list = []
unfunc_follow_metric_list = []
for metric in self.metric_dict[type]:
unfunc_follow_metric_list.append(metric) if score_metric_dict[
metric] < 80 else func_follow_metric_list.append(
metric)
str_follow_over_optimal = ''
if not unfunc_follow_metric_list:
str_func_follow_metric = string_concatenate(func_follow_metric_list)
follow_description1 = f"{str_func_follow_metric}指标均表现良好"
else:
for metric in unfunc_follow_metric_list:
if metric in self.bulitin_metric_list:
value = self.result[metric]
if self.kind_dict[metric] == -1:
metric_over_optimal = ((value - self.optimal_dict[metric]) / self.optimal_dict[
metric]) * 100
elif self.kind_dict[metric] == 1:
metric_over_optimal = ((self.optimal_dict[metric] - value) / self.optimal_dict[
metric]) * 100
elif self.kind_dict[metric] == 0:
metric_over_optimal = (abs(self.optimal_dict[metric] - value) / self.optimal_dict[
metric]) * 100
else:
value = self.custom_data[metric]["value"][0]
if self.custom_param_dict[metric]['kind'][0] == -1:
metric_over_optimal = ((value - self.custom_param_dict[metric]['optimal'][0]) /
self.custom_param_dict[metric]['optimal'][0]) * 100
elif self.custom_param_dict[metric]['kind'][0] == 1:
metric_over_optimal = ((self.custom_param_dict[metric]['optimal'][0] - value) /
self.custom_param_dict[metric]['optimal'][0]) * 100
elif self.custom_param_dict[metric]['kind'][0] == 0:
metric_over_optimal = (abs(self.custom_param_dict[metric]['optimal'][0] - value) /
self.custom_param_dict[metric]['optimal'][0]) * 100
# str_follow_over_optimal += f"{metric}为{value:.2f}，超过合理范围{metric_over_optimal:.2f}%；"
# str_follow_over_optimal += f"{metric}为{value:.2f}{self.unit_dict[metric]}，超过合理范围{metric_over_optimal:.2f}%；"
str_follow_over_optimal += f"{metric}为{round(value, 2)}{self.unit_dict[metric]}，超过合理范围{round(metric_over_optimal, 2)}%；"
str_follow_over_optimal = str_follow_over_optimal[:-1] if str_follow_over_optimal else ""
str_func_follow_metric = string_concatenate(func_follow_metric_list)
str_unfunc_follow_metric = string_concatenate(unfunc_follow_metric_list)
if not func_follow_metric_list:
follow_description1 = f"{str_unfunc_follow_metric}指标表现不佳。{str_follow_over_optimal}"
else:
follow_description1 = f"{str_func_follow_metric}指标表现良好，{str_unfunc_follow_metric}指标表现不佳。{str_follow_over_optimal}"
# for follow_description2
follow_duration = (len(self.time_list_follow) - 2) * 0.04 if self.time_list_follow else 0
follow_description2 = f"自车在行驶时有{round(follow_duration, 2)}s处于跟车状态，"
follow_description3 = ""
if "followResponseTime" in self.metric_list:
# for follow_description3
cnt3_1 = len(self.follow_stop_time_start_list)
tmp3_list = [x for x in self.follow_stop_time_start_list if
x > self.optimal_dict['followResponseTime']]
cnt3_2 = len(tmp3_list)
percent3 = 0 if cnt3_1 == 0 else round(cnt3_2 / cnt3_1 * 100, 2)
follow_description3 = f"起停跟车响应时间有{cnt3_2}次超出合理范围，占比为{percent3}%；"
follow_description4 = ""
if "followDistanceDeviation" in self.metric_list:
# for follow_description4
cnt4_1 = len(self.stop_distance_list)
tmp4_list = [x for x in self.stop_distance_list if
x < self.optimal_dict['followDistanceDeviation']]
cnt4_2 = len(tmp4_list)
percent4 = 0 if cnt4_1 == 0 else round(cnt4_2 / cnt4_1 * 100, 2)
follow_description4 = f"在本次测试中，跟车状态下目标车共发生了{cnt4_1}次停车，有{cnt4_2}次超出合理范围，占比为{percent4}%；"
# distance_list = obj_df['dist'].values.tolist()
# distance_deviation_vs_time = self.zip_time_pairs(self.dist_deviation_list_full_time)
#
# acc_dict["description1"] = replace_key_with_value(follow_description1, self.name_dict)
# acc_dict["description2"] = replace_key_with_value(follow_description2, self.name_dict)
# acc_dict["description3"] = replace_key_with_value(follow_description3, self.name_dict)
# acc_dict["description4"] = replace_key_with_value(follow_description4, self.name_dict)
# common parameter calculate
obj_df = self.df[self.df['playerId'] == 2]
obj_speed_list = obj_df['v'].values.tolist()
obj_speed_vs_time = self.zip_time_pairs(obj_speed_list)
rel_speed_vs_time = self.zip_time_pairs(self.v_relative_list_full_time)
distance_vs_time = self.zip_time_pairs(self.dist_list_full_time)
# acc_dict["speData"] = [ego_speed_vs_time, obj_speed_vs_time, rel_speed_vs_time]
# acc_dict["disData"] = distance_vs_time
unfunc_follow_df = self.unfunc_follow_df.copy()
unfunc_follow_df['type'] = "origin"
unfunc_follow_slices = unfunc_follow_df.to_dict('records')
# acc_dict["speMarkLine"] = unfunc_follow_slices
# acc_dict["disMarkLine"] = unfunc_follow_slices
distance_deviation_vs_time = self.zip_time_pairs(self.dist_deviation_list_full_time)
# function ACC data
acc_dict_indexes = {}
for metric in self.metric_dict[type]:
if metric == "followSpeedDeviation":
self.unfunctional_follow_v_deviation_df_statistic()
unfunc_v_df = self.unfunc_follow_df.copy()
unfunc_v_df.loc[unfunc_v_df['type'] != 'ACC', 'type'] = "origin"
# unfunc_v_df.loc[unfunc_v_df['type'] == 'time', 'type'] = "time"
unfunc_v_slices = unfunc_v_df.to_dict('records')
df1 = self.unfunc_follow_df[self.unfunc_follow_df['type'] == 'ACC']
df1['v_time'] = df1['end_time'] - df1['start_time']
devi_v_time = df1['v_time'].sum()
follow_duration = (len(self.time_list_follow) - 2) * 0.04 if self.time_list_follow else 0
percent2_1 = devi_v_time / follow_duration * 100 if follow_duration else 0
if devi_v_time != 0:
follow_description2 += f"跟车状态下自车和目标车的速度差共有{round(devi_v_time, 2)}s超出合理范围，占比为{round(percent2_1, 2)}%；"
else:
follow_description2 += "跟车状态下自车和目标车的速度差均在合理范围内；"
v_deviation_list = [x for x in self.v_deviation_list if not np.isnan(x)]
acc_dict_indexes["followSpeedDeviation"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict[
"followSpeedDeviation"],
"avg": round(np.mean(v_deviation_list), 2) if v_deviation_list else '-',
"max": round(max(v_deviation_list), 2) if v_deviation_list else '-',
"min": round(min(v_deviation_list), 2) if v_deviation_list else '-',
# "range": f"[0, {self.optimal_dict['followSpeedDeviation']}]"
"range": f"[-{self.optimal_dict['followSpeedDeviation']}, {self.optimal_dict['followSpeedDeviation']}]"
}
fsd_data = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"data": rel_speed_vs_time,
"range": f"[-{self.optimal_dict['followSpeedDeviation']}, {self.optimal_dict['followSpeedDeviation']}]",
# "range": f"[0, {self.optimal_dict['followSpeedDeviation']}]",
# "markLine": unfunc_v_slices,
# "markLine2": [-1 * self.optimal_dict['followSpeedDeviation'],
# self.optimal_dict['followSpeedDeviation']],
}
builtin_graph_dict["followSpeedDeviation"] = fsd_data
if metric == "followDistanceDeviation":
self.unfunctional_follow_dist_deviation_df_statistic()
unfunc_dist_df = self.unfunc_follow_df.copy()
unfunc_dist_df.loc[unfunc_dist_df['type'] != 'ACC', 'type'] = "origin"
# unfunc_dist_df.loc[unfunc_dist_df['type'] == 'distance', 'type'] = "distance"
unfunc_dist_slices = unfunc_dist_df.to_dict('records')
df2 = self.unfunc_follow_df[self.unfunc_follow_df['type'] == 'ACC']
df2['dist_time'] = df2['end_time'] - df2['start_time']
devi_dist_time = df2['dist_time'].sum()
follow_duration = (len(self.time_list_follow) - 2) * 0.04 if self.time_list_follow else 0
percent2_2 = devi_dist_time / follow_duration * 100 if follow_duration else 0
if devi_dist_time != 0:
follow_description2 += f"跟车状态下自车和目标车的距离差共有{round(devi_dist_time, 2)}s超出合理范围，占比为{round(percent2_2, 2)}%；"
else:
follow_description2 += "跟车状态下自车和目标车的距离差均在合理范围内；"
dist_deviation_list = [x for x in self.dist_deviation_list if not np.isnan(x)]
acc_dict_indexes["followDistanceDeviation"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict["followDistanceDeviation"],
"avg": round(np.mean(dist_deviation_list), 2) if dist_deviation_list else '-',
"max": round(max(dist_deviation_list), 2) if dist_deviation_list else '-',
"min": round(min(dist_deviation_list), 2) if dist_deviation_list else '-',
"range": f"[0, {self.optimal_dict['followDistanceDeviation']}]"
}
fdd_data = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"data": distance_deviation_vs_time,
"range": f"[0, {self.optimal_dict['followDistanceDeviation']}]",
# "markLine": unfunc_dist_slices,
# "markLine2": [self.optimal_dict['followDistanceDeviation']],
}
builtin_graph_dict["followDistanceDeviation"] = fdd_data
if metric == "followStopDistance":
acc_dict_indexes["followStopDistance"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict["followStopDistance"],
"avg": round(np.mean(self.stop_distance_list), 2) if self.stop_distance_list else '-',
"max": round(max(self.stop_distance_list), 2) if self.stop_distance_list else '-',
"min": round(min(self.stop_distance_list), 2) if self.stop_distance_list else '-',
"range": f"[{self.optimal_dict['followStopDistance']}, inf)"
}
# sdd_data = {
# # "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
# "name": f"{self.name_dict[metric]}",
# "data": self.stop_distance_list,
# "range": f"[{self.optimal_dict['followStopDistance']}, inf)",
# # "ref": [self.optimal_dict['followStopDistance'] - 1,
# # self.optimal_dict['followStopDistance']],
# }
# builtin_graph_dict["followStopDistance"] = sdd_data
if metric == "followResponseTime":
acc_dict_indexes["followResponseTime"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict["followResponseTime"],
"avg": round(np.mean(self.follow_stop_time_start_list),
2) if self.follow_stop_time_start_list else '-',
"max": round(max(self.follow_stop_time_start_list),
2) if self.follow_stop_time_start_list else '-',
"min": round(min(self.follow_stop_time_start_list),
2) if self.follow_stop_time_start_list else '-',
"range": f"[0, {self.optimal_dict['followResponseTime']}]"
}
# rt_data = {
# # "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
# "name": f"{self.name_dict[metric]}",
# "data": self.follow_stop_time_start_list,
# # "range": f"[0, {self.optimal_dict['followResponseTime']})"
# }
# builtin_graph_dict["followResponseTime"] = rt_data
if metric not in self.bulitin_metric_list:
acc_dict_indexes[metric] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict[metric],
"avg": self.custom_data[metric]['tableData']['avg'],
"max": self.custom_data[metric]['tableData']['max'],
"min": self.custom_data[metric]['tableData']['min'],
}
if self.custom_param_dict[metric]['kind'][0] == -1:
acc_dict_indexes[metric][
"range"] = f"[0, {self.custom_param_dict[metric]['optimal'][0]}]"
elif self.custom_param_dict[metric]['kind'][0] == 1:
acc_dict_indexes[metric][
"range"] = f"[{self.custom_param_dict[metric]['optimal'][0]}, inf)"
elif self.custom_param_dict[metric]['kind'][0] == 0:
acc_dict_indexes[metric][
"range"] = f"[{self.custom_param_dict[metric]['optimal'][0] * self.custom_param_dict[metric]['multiple'][0][0]}, {self.custom_param_dict[metric]['optimal'][0] * self.custom_param_dict[metric]['multiple'][0][1]}]"
custom_graph_dict[metric] = self.custom_data[metric]["reportData"]
self.custom_markline_list.extend(self.custom_data[metric]["reportData"]['markLine'])
acc_dict["indexes"] = acc_dict_indexes
acc_dict["builtin"] = builtin_graph_dict
acc_dict["custom"] = custom_graph_dict
acc_dict["description1"] = replace_key_with_value(follow_description1, self.name_dict)
acc_dict["description2"] = replace_key_with_value(follow_description2, self.name_dict)
acc_dict["description3"] = replace_key_with_value(follow_description3, self.name_dict)
acc_dict["description4"] = replace_key_with_value(follow_description4, self.name_dict)
type_details_dict[type] = acc_dict
unfunc_metric_list += unfunc_follow_metric_list
func_over_optimal.append(str_follow_over_optimal)
elif type == "functionLKA":
lka_dict = {
"name": f"{self.type_name_dict[type]}",
}
builtin_graph_dict = {}
custom_graph_dict = {}
# get score and grade
score_lane = score_type_dict["functionLKA"]
grade_lane = score_grade(score_lane)
lka_dict['score'] = score_lane
lka_dict['level'] = grade_lane
# unfunc_type_list.append('车道保持') if score_type_dict["functionLKA"] < 80 else func_type_list.append(
# '车道保持')
unfunc_type_list.append(type) if score_type_dict[type] < 80 else func_type_list.append(type)
# lane keep description
func_lane_metric_list = []
unfunc_lane_metric_list = []
for metric in self.metric_dict[type]:
unfunc_lane_metric_list.append(metric) if score_metric_dict[
metric] < 80 else func_lane_metric_list.append(
metric)
lka_dict_indexes = {}
for metric in self.metric_dict[type]:
if metric == "laneDistance":
lka_dict_indexes["laneDistance"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict["laneDistance"],
"avg": round(np.mean(self.line_dist), 2) if self.line_dist else "-",
"max": round(max(self.line_dist), 2) if self.line_dist else "-",
"min": round(min(self.line_dist), 2) if self.line_dist else "-",
"range": f"[{self.optimal_dict['laneDistance']}, 1.875]"
}
if metric == "centerDistanceExpectation":
lka_dict_indexes["centerDistanceExpectation"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict['centerDistanceExpectation'],
"avg": round(self.result["centerDistanceExpectation"], 2) if self.center_dist else "-",
"max": "-",
"min": "-",
"range": f"[0, {self.optimal_dict['centerDistanceExpectation']}]"
}
if metric == "centerDistanceStandardDeviation":
lka_dict_indexes["centerDistanceStandardDeviation"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict['centerDistanceStandardDeviation'],
"avg": round(self.result["centerDistanceStandardDeviation"],
2) if self.center_dist else "-",
"max": "-",
"min": "-",
"range": f"[0, {self.optimal_dict['centerDistanceStandardDeviation']}]"
}
if metric == "centerDistanceMax":
lka_dict_indexes["centerDistanceMax"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict['centerDistanceMax'],
"avg": "-",
"max": round(self.result["centerDistanceMax"], 2) if self.center_dist else "-",
"min": "-",
"range": f"[0, {self.optimal_dict['centerDistanceMax']}]"
}
if metric == "centerDistanceMin":
lka_dict_indexes["centerDistanceMin"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict['centerDistanceMin'],
"avg": "-",
"max": "-",
"min": round(self.result["centerDistanceMin"], 2) if self.center_dist else "-",
"range": f"[0, {self.optimal_dict['centerDistanceMin']}]"
}
if metric == "centerDistanceFrequency":
lka_dict_indexes["centerDistanceFrequency"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict['centerDistanceFrequency'],
"avg": round(self.result["centerDistanceFrequency"], 2) if self.center_dist else "-",
"max": "-",
"min": "-",
"range": f"[0, {self.optimal_dict['centerDistanceFrequency']}]"
}
if metric == "centerDistanceRange":
lka_dict_indexes["centerDistanceRange"] = {
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict['centerDistanceRange'],
"avg": round(self.result["centerDistanceRange"], 2) if self.center_dist else "-",
"max": "-",
"min": "-",
"range": f"[0, {self.optimal_dict['centerDistanceRange']}]"
}
if metric not in self.bulitin_metric_list:
lka_dict_indexes[metric] = {
# "name": self.custom_data[metric]['name'],
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict[metric],
"avg": self.custom_data[metric]['tableData']['avg'],
"max": self.custom_data[metric]['tableData']['max'],
"min": self.custom_data[metric]['tableData']['min'],
}
if self.custom_param_dict[metric]['kind'][0] == -1:
lka_dict_indexes[metric]["range"] = f"[0, {self.custom_param_dict[metric]['optimal'][0]}]"
elif self.custom_param_dict[metric]['kind'][0] == 1:
lka_dict_indexes[metric][
"range"] = f"[{self.custom_param_dict[metric]['optimal'][0]}, inf)"
elif self.custom_param_dict[metric]['kind'][0] == 0:
lka_dict_indexes[metric][
"range"] = f"[{self.custom_param_dict[metric]['optimal'][0] * self.custom_param_dict[metric]['multiple'][0][0]}, {self.custom_param_dict[metric]['optimal'][0] * self.custom_param_dict[metric]['multiple'][0][1]}]"
custom_graph_dict[metric] = self.custom_data[metric]["reportData"]
self.custom_markline_list.extend(self.custom_data[metric]["reportData"]['markLine'])
lka_dict["indexes"] = lka_dict_indexes
str_lane_over_optimal = ''
if not unfunc_lane_metric_list:
str_func_lane_metric = string_concatenate(func_lane_metric_list)
lane_description1 = f"{str_func_lane_metric}指标均表现良好"
else:
for metric in unfunc_lane_metric_list:
if metric in self.bulitin_metric_list:
value = self.result[metric]
if self.kind_dict[metric] == -1:
metric_over_optimal = ((value - self.optimal_dict[metric]) / self.optimal_dict[
metric]) * 100
elif self.kind_dict[metric] == 1:
metric_over_optimal = ((self.optimal_dict[metric] - value) / self.optimal_dict[
metric]) * 100
elif self.kind_dict[metric] == 0:
metric_over_optimal = (abs(self.optimal_dict[metric] - value) / self.optimal_dict[
metric]) * 100
else:
value = self.custom_data[metric]["value"][0]
if self.custom_param_dict[metric]['kind'][0] == -1:
metric_over_optimal = ((value - self.custom_param_dict[metric]['optimal'][0]) /
self.custom_param_dict[metric]['optimal'][0]) * 100
elif self.custom_param_dict[metric]['kind'][0] == 1:
metric_over_optimal = ((self.custom_param_dict[metric]['optimal'][0] - value) /
self.custom_param_dict[metric]['optimal'][0]) * 100
elif self.custom_param_dict[metric]['kind'][0] == 0:
metric_over_optimal = (abs(self.custom_param_dict[metric]['optimal'][0] - value) /
self.custom_param_dict[metric]['optimal'][0]) * 100
str_lane_over_optimal += f"{metric}为{value:.2f}{self.unit_dict[metric]}，超过合理范围{metric_over_optimal:.2f}%；"
str_lane_over_optimal = str_lane_over_optimal[:-1] if str_lane_over_optimal else ""
str_func_lane_metric = string_concatenate(func_lane_metric_list)
str_unfunc_lane_metric = string_concatenate(unfunc_lane_metric_list)
if not func_lane_metric_list:
lane_description1 = f"{str_unfunc_lane_metric}指标表现不佳。{str_lane_over_optimal}"
else:
lane_description1 = f"{str_func_lane_metric}指标表现良好,{str_unfunc_lane_metric}指标表现不佳。{str_lane_over_optimal}"
lane_description2 = ""
center_distance_vs_time = []
if "centerDistanceMax" in self.metric_list:
cnt2_1 = len(self.center_dist)
tmp2_list = [x for x in self.center_dist if x > self.optimal_dict['centerDistanceMax']]
cnt2_2 = len(tmp2_list)
percent2 = 0 if cnt2_1 == 0 else (cnt2_2 / cnt2_1 * 100)
dist_time = percent2 * duration / 100
if dist_time != 0:
lane_description2 = f"距离有{dist_time:.2f}秒超出合理范围，占比为{percent2:.2f}%"
else:
lane_description2 = f"距离均在合理范围内，算法车道保持表现良好"
# lane keep data for graph
# center_distance_vs_time = self.zip_time_pairs(self.center_dist)
center_distance_vs_time = self.zip_time_pairs(self.center_dist_with_nan)
self.unfunctional_lane_df_statistic()
#
# lka_dict["description1"] = replace_key_with_value(lane_description1, self.name_dict)
# lka_dict["description2"] = lane_description2
unfunc_lane_df = self.unfunc_lane_df.copy()
unfunc_lane_df['type'] = "origin"
unfunc_lane_slices = unfunc_lane_df.to_dict('records')
unfunc_lane_dist_df = self.unfunc_lane_df.copy()
unfunc_lane_dist_df.loc[unfunc_lane_dist_df['type'] != 'LKA', 'type'] = "origin"
unfunc_lane_dist_slices = unfunc_lane_dist_df.to_dict('records')
lka_data = {
"name": "车辆中心线横向距离（m）",
"data": center_distance_vs_time,
# "markLine": unfunc_lane_dist_slices
}
if 'centerDistanceMax' in self.optimal_dict:
lka_range = f"[0, {self.optimal_dict['centerDistanceMax']}]"
elif 'centerDistanceMin' in self.optimal_dict:
lka_range = f"[0, {self.optimal_dict['centerDistanceMin']}]"
else:
lka_range = f"[0, 0.5]"
lka_data["range"] = lka_range
builtin_graph_dict["centerDistance"] = lka_data
lka_dict["builtin"] = builtin_graph_dict
lka_dict["custom"] = custom_graph_dict
lka_dict["description1"] = replace_key_with_value(lane_description1, self.name_dict)
lka_dict["description2"] = lane_description2
# lka_dict["centerDisData"] = center_distance_vs_time
# lka_dict["centerDisMarkLine"] = unfunc_lane_dist_slices
type_details_dict[type] = lka_dict
unfunc_metric_list += unfunc_lane_metric_list
func_over_optimal.append(str_lane_over_optimal)
else:
type_dict = {
"name": f"{self.type_name_dict[type]}",
}
builtin_graph_dict = {}
custom_graph_dict = {}
# get score and grade
score_custom_type = score_type_dict[type]
grade_custom_type = score_grade(score_custom_type)
type_dict["score"] = score_custom_type
type_dict["level"] = grade_custom_type
# custom type description
good_custom_metric_list = []
bad_custom_metric_list = []
unfunc_type_list.append(type) if score_type_dict[type] < 80 else func_type_list.append(type)
type_dict_indexes = {}
for metric in self.metric_dict[type]:
bad_custom_metric_list.append(metric) if score_metric_dict[
metric] < 80 else good_custom_metric_list.append(
metric)
type_dict_indexes[metric] = {
# "name": self.custom_data[metric]['name'],
# "name": f"{self.name_dict[metric]}({self.unit_dict[metric]})",
"name": f"{self.name_dict[metric]}",
"score": score_metric_dict[metric],
"avg": self.custom_data[metric]['tableData']['avg'],
"max": self.custom_data[metric]['tableData']['max'],
"min": self.custom_data[metric]['tableData']['min'],
}
if self.custom_param_dict[metric]['kind'][0] == -1:
type_dict_indexes[metric]["range"] = f"[0, {self.custom_param_dict[metric]['optimal'][0]:.3f}]"
elif self.custom_param_dict[metric]['kind'][0] == 1:
type_dict_indexes[metric][
"range"] = f"[{self.custom_param_dict[metric]['optimal'][0]:.3f}, inf)"
elif self.custom_param_dict[metric]['kind'][0] == 0:
type_dict_indexes[metric][
"range"] = f"[{self.custom_param_dict[metric]['optimal'][0] * self.custom_param_dict[metric]['multiple'][0][0]:.3f}, {self.custom_param_dict[metric]['optimal'][0] * self.custom_param_dict[metric]['multiple'][0][1]:.3f}]"
custom_graph_dict[metric] = self.custom_data[metric]['reportData']
self.custom_markline_list.extend(self.custom_data[metric]["reportData"]['markLine'])
type_dict["indexes"] = type_dict_indexes
str_type_over_optimal = ""
if not bad_custom_metric_list:
str_good_custom_metric = string_concatenate(good_custom_metric_list)
type_description1 = f"{str_good_custom_metric}指标均表现良好"
type_description2 = f"指标均在合理范围内，算法表现良好"
else:
for metric in bad_custom_metric_list:
value = self.custom_data[metric]["value"][0]
if self.custom_param_dict[metric]['kind'][0] == -1:
metric_over_optimal = ((value - self.custom_param_dict[metric]['optimal'][0]) /
self.custom_param_dict[metric]['optimal'][0]) * 100
elif self.custom_param_dict[metric]['kind'][0] == 1:
metric_over_optimal = ((self.custom_param_dict[metric]['optimal'][0] - value) /
self.custom_param_dict[metric]['optimal'][0]) * 100
elif self.custom_param_dict[metric]['kind'][0] == 0:
metric_over_optimal = (abs(self.custom_param_dict[metric]['optimal'][0] - value) /
self.custom_param_dict[metric]['optimal'][0]) * 100
str_type_over_optimal += f"{metric}为{value:.2f}{self.unit_dict[metric]}，超过合理范围{metric_over_optimal:.2f}%；"
str_type_over_optimal = str_type_over_optimal[:-1]
str_good_custom_metric = string_concatenate(good_custom_metric_list)
str_bad_custom_metric = string_concatenate(bad_custom_metric_list)
if not good_custom_metric_list:
type_description1 = f"{str_bad_custom_metric}指标表现不佳"
type_description2 = f"{str_type_over_optimal}"
else:
type_description1 = f"{str_good_custom_metric}指标表现良好，{str_bad_custom_metric}指标表现不佳"
type_description2 = f"{str_type_over_optimal}"
type_dict["builtin"] = builtin_graph_dict
type_dict["custom"] = custom_graph_dict
type_dict["description1"] = replace_key_with_value(type_description1, self.name_dict)
type_dict["description2"] = replace_key_with_value(type_description2, self.name_dict)
type_details_dict[type] = type_dict
unfunc_metric_list += bad_custom_metric_list
func_over_optimal.append(str_type_over_optimal)
report_dict["details"] = type_details_dict
# report_dict["speData"] = [ego_speed_vs_time, obj_speed_vs_time, rel_speed_vs_time]
func_over_optimal = [s for s in func_over_optimal if s]
str_func_over_optimal = '；'.join(func_over_optimal)
if grade_function == '优秀':
str_func_type = string_concatenate(func_type_list)
func_description1 = f'算法在{str_func_type}功能上表现优秀；'
elif grade_function == '良好':
str_func_type = string_concatenate(func_type_list)
func_description1 = f'算法在{str_func_type}功能上总体表现良好，满足设计指标要求；'
elif grade_function == '一般':
str_unfunc_type = string_concatenate(unfunc_type_list)
str_unfunc_metric = string_concatenate(unfunc_metric_list)
str_unfunc_metric = replace_key_with_value(str_unfunc_metric, self.type_name_dict)
func_description1 = f'算法在{str_unfunc_type}功能上表现一般、需要在{str_unfunc_metric}指标上进一步优化。其中，{str_func_over_optimal}；'
elif grade_function == '较差':
str_unfunc_type = string_concatenate(unfunc_type_list)
func_description1 = f'算法在{str_unfunc_type}功能上表现较差，需要提高算法的功能性表现。其中，{str_func_over_optimal}；'
if not unfunc_type_list:
func_description2 = '功能性在各个功能上的表现俱佳。'
else:
str_unfunc_type = string_concatenate(unfunc_type_list)
func_description2 = f"算法在{str_unfunc_type}功能上需要重点优化。"
# report_dict["description1"] = replace_key_with_value(func_description1, self.name_dict)
report_dict["description1"] = replace_key_with_value(replace_key_with_value(func_description1, self.name_dict),
self.type_name_dict)
report_dict["description2"] = replace_key_with_value(func_description2, self.type_name_dict)
report_dict['commonData'] = {
"per": {
"name": "脚刹/油门踏板开度（百分比）",
"legend": ["刹车踏板开度", "油门踏板开度"],
"data": [brake_vs_time, throttle_vs_time]
},
"ang": {
"name": "方向盘转角（角度°）",
"data": steering_vs_time
},
"spe": {
"name": "速度（km/h）",
"legend": ["自车速度", "目标车速度", "自车与目标车相对速度"],
"data": [ego_speed_vs_time, obj_speed_vs_time, rel_speed_vs_time]
},
"acc": {
"name": "加速度（m/s²)",
"legend": ["横向加速度", "纵向加速度"],
"data": [lat_acc_vs_time, lon_acc_vs_time]
},
# "dis": {
# "name": "前车距离（m）",
# "data": distance_vs_time
# }
}
if "functionACC" in self.type_list:
report_dict['commonData']["dis"] = {
"name": "前车距离（m）",
"data": distance_vs_time
}
self.unfunc_df = pd.concat([self.unfunc_df, self.unfunc_follow_df, self.unfunc_lane_df], ignore_index=True)
unfunc_df = self.unfunc_df.copy().dropna()
unfunc_slices = unfunc_df.to_dict('records')
unfunc_slices.extend(self.custom_markline_list)
report_dict["commonMarkLine"] = unfunc_slices
# report_dict = {
# "name": "功能性",
# "weight": f"{self.weight * 100:.2f}%",
# "weightDistribution": weight_distribution,
# "score": score_function,
# "level": grade_function,
# 'followStopCount': self.follow_stop_count,
# "description1": func_description1,
# "description2": func_description2,
#
# "functionACC": acc_dict,
# "functionLKA": lka_dict,
#
# "speData": [ego_speed_vs_time, obj_speed_vs_time, rel_speed_vs_time],
# "accData": [lat_acc_vs_time, lon_acc_vs_time],
#
# }
self.eval_data = self.ego_df.copy()
return report_dict
def get_eval_data(self):
if 'line_dist' in self.eval_data.columns:
df = self.eval_data[['simTime', 'simFrame', 'playerId', 'lat_acc_roc', 'lon_acc_roc', 'line_dist']].copy()
else:
df = self.eval_data[['simTime', 'simFrame', 'playerId', 'lat_acc_roc', 'lon_acc_roc']].copy()
self.df['line_dist'] = pd.Series(dtype=float)
return df