lantu_vehstate/cicv_ica_longitudinal_control03_peak_time_cruise_new.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
##################################################################
#
# Copyright (c) 2023 CICV, Inc. All Rights Reserved
#
##################################################################
"""
@Authors:           zhanghaiwen, yangzihao
@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, get_status_active_data
from log import logger

"""import functions"""


# custom metric codes
class CustomMetric(object):
    def __init__(self, all_data, case_name):
        self.data = all_data
        self.optimal_dict = self.data.config
        self.status_trigger_dict = self.data.status_trigger_dict
        self.case_name = case_name
        self.markline_df = pd.DataFrame(columns=['start_time', 'end_time', 'start_frame', 'end_frame', 'type'])
        self.graph_list = []

        self.df = pd.DataFrame()
        self.ego_df = pd.DataFrame()
        self.df_ica = pd.DataFrame()

        self.peak_time_cruise = None

        self.result = {
            "name": "定速巡航峰值时间",
            "value": [],
            # "weight": [],
            "tableData": {
                "avg": "",  # 平均值，或指标值
                "max": "",
                "min": ""
            },
            "reportData": {
                "name": "定速巡航峰值时间（s）",
                # "legend": [], # 如果有多个data，则需要增加data对应的说明，如：["横向加速度", "纵向加速度"]
                "data": [],
                "markLine": [],
                "range": [],
            },
            "statusFlag": {}
        }
        self.run()
        print(f"定速巡航峰值时间: {self.result['value']}")

    def data_extract(self):
        self.df = self.data.object_df
        self.ego_df = self.data.ego_data

        active_time_ranges = self.status_trigger_dict['ICA']['ICA_cruise_time']
        self.df_ica = get_status_active_data(active_time_ranges, self.ego_df)

        # self.df_ica = self.ego_df[self.ego_df['ICA_status'] == "LLC_Follow_Line"].copy()  # 数字3对应ICA的Active
        # self.df_ica = self.df[self.df['ACC_status'] == "Shut_off"].copy()  # 数字3对应ICA的Active
        # self.df_ica = self.df[self.df['ACC_status'] == "Active"].copy()  # 数字3对应ICA的Active

        if self.df_ica.empty:
            self.result['statusFlag']['function_ICA'] = False
        else:
            self.result['statusFlag']['function_ICA'] = True

    def _get_first_change_index_cruise(self):
        """
        获取数据集中第一次巡航速度发生变化的索引。

        Args:
            无参数。

        Returns:
            Union[int, None]: 如果存在巡航速度发生变化的索引，则返回第一个发生变化的索引（int类型）；
            如果不存在，则返回None。

        """
        change_indices = self.df_ica[self.df_ica['set_cruise_speed'] != self.df_ica['set_cruise_speed'].shift()].index
        if not change_indices.empty:
            first_change_index = change_indices.min()
        else:
            first_change_index = None
        return first_change_index

    def data_analyze(self):
        if not self.df_ica.empty:
            first_change_index = self._get_first_change_index_cruise()

            if not first_change_index:
                self.peak_time_cruise = 0
                self.result['value'] = [round(self.peak_time_cruise, 3)]
                print(f"peak_time_cruise: {self.peak_time_cruise}")
            else:
                start_time = self.df_ica.loc[first_change_index, 'simTime']
                peak_time = self.df_ica.loc[self.df_ica['speedX'].idxmax(), 'simTime']
                self.peak_time_cruise = peak_time - start_time
                self.result['value'] = [round(self.peak_time_cruise, 3)]
                print(f"peak_time_cruise: {self.peak_time_cruise}")
        else:
            self.peak_time_cruise = 1000
            self.result['value'] = [1000]
            print(f"peak_time_cruise: {self.peak_time_cruise}")

    def markline_statistic(self):
        pass

    def report_data_statistic(self):
        # time_list = self.ego_df['simTime'].values.tolist()
        # graph_list = [x for x in self.graph_list if not np.isnan(x)]
        self.result['tableData']['avg'] = self.result['value'][0] if not self.df_ica.empty else '-'
        self.result['tableData']['max'] = '-'
        self.result['tableData']['min'] = '-'

        # zip_vs_time = zip_time_pairs(time_list, self.graph_list)
        self.result['reportData']['data'] = []

        # self.markline_statistic()
        # markline_slices = self.markline_df.to_dict('records')
        self.result['reportData']['markLine'] = []

        self.result['reportData']['range'] = [0, 1.2]

    def run(self):
        # logger.info(f"Custom metric run:[{self.result['name']}].")
        logger.info(f"[case:{self.case_name}] Custom metric:[peak_time_cruise:{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