#!/usr/bin/env python # -*- coding: utf-8 -*- ################################################################## # # Copyright (c) 2023 CICV, Inc. All Rights Reserved # ################################################################## """ @Authors: yangzihao(yangzihao@china-icv.cn) @Data: 2024/02/21 @Last Modified: 2024/02/21 @Summary: The template of custom indicator. """ import math import pandas as pd import numpy as np from common import zip_time_pairs, continuous_group from log import logger """import functions""" # def zip_time_pairs(time_list, zip_list): # zip_time_pairs = zip(time_list, zip_list) # zip_vs_time = [[x, y] for x, y in zip_time_pairs if not math.isnan(y)] # return zip_vs_time # def continuous_group(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] # # time_df = pd.DataFrame(time, columns=['start_time', 'end_time']) # frame_df = pd.DataFrame(frame, columns=['start_frame', 'end_frame']) # # result_df = pd.concat([time_df, frame_df], axis=1) # # return result_df # def continous_judge(frame_list): # if not frame_list: # return 0 # # cnt = 1 # for i in range(1, len(frame_list)): # if frame_list[i] - frame_list[i - 1] <= 3: # continue # cnt += 1 # return cnt # custom metric codes class CustomMetric(object): def __init__(self, all_data, case_name): self.data = all_data self.optimal_dict = self.data.config self.case_name = case_name self.markline_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type']) self.df = pd.DataFrame() self.df_stop_and_go = pd.DataFrame() self.time_list_follow = list() self.frame_list_follow = list() self.follow_stop_time_start_list = list() self.result = { "name": "跟车启动响应时间", "value": [], # "weight": [], "tableData": { "avg": "", # 平均值，或指标值 "max": "", "min": "" }, "reportData": { "name": "跟车启动响应时间（s）", # "legend": [], # 如果有多个data，则需要增加data对应的说明，如：["横向加速度", "纵向加速度"] "data": [], "markLine": [], "range": [], }, "statusFlag": {} } self.run() def data_extract(self): self.df = self.data.object_df self.df_stop_and_go = self.df[self.df['ACC_status'] == "Shut_off"].copy() # 数字3对应ICA的Active # self.df_stop_and_go = self.df[self.df['ACC_status'] == "Active"].copy() # 数字3对应ICA的Active if self.df_stop_and_go.empty: self.result['statusFlag']['functionICA'] = False else: self.result['statusFlag']['functionICA'] = True def data_analyze(self): df = self.df_stop_and_go.copy() stop_v_threshold = 0.05 df['v'] = df['v'].apply(lambda x: 0 if x <= stop_v_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] followResponseTime = max(time_start_list) if time_start_list else 0 self.follow_stop_time_start_list = time_start_list self.result['value'] = [round(followResponseTime, 2)] if not np.isnan(followResponseTime) else [0] def markline_statistic(self): pass def report_data_statistic(self): # time_list = self.df['simTime'].values.tolist() graph_list = [x for x in self.follow_stop_time_start_list if not np.isnan(x)] self.result['tableData']['avg'] = f'{np.mean(graph_list):.2f}' if graph_list else '-' self.result['tableData']['max'] = f'{max(graph_list):.2f}' if graph_list else '-' self.result['tableData']['min'] = f'{min(graph_list):.2f}' if graph_list else '-' self.result['reportData']['data'] = [] self.markline_statistic() self.result['reportData']['markLine'] = [] self.result['reportData']['range'] = [] def run(self): # logger.info(f"Custom metric run:[{self.result['name']}].") logger.info(f"[case:{self.case_name}] Custom metric:[ica_distance_deviation:{self.result['name']}] evaluate.") try: self.data_extract() except Exception as e: logger.error(f"[case:{self.case_name}] Custom metric:{self.result['name']} data extract ERROR!", e) try: self.data_analyze() except Exception as e: logger.error(f"[case:{self.case_name}] Custom metric:{self.result['name']} data analyze ERROR!", e) try: self.report_data_statistic() except Exception as e: logger.error(f"[case:{self.case_name}] Custom metric:{self.result['name']} report data statistic ERROR!", e) # if __name__ == "__main__": # pass