cicv_evaluate_master-single_eval/custom/ica_response_time.py

#!/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