#!/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""" Max = 100 # 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_acc = pd.DataFrame() self.stable_average_speed = None self.steady_error_cruise = None self.result = { "name": "定速巡航速度稳态误差", "value": [], # "weight": [], "tableData": { "avg": "", # 平均值，或指标值 "max": "", "min": "" }, "reportData": { "name": "定速巡航速度稳态误差（%）", # "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['ACC']['ACC_active_time'] self.df_acc = get_status_active_data(active_time_ranges, self.ego_df) # self.df_acc = self.ego_df[self.ego_df['ICA_status'] == "LLC_Follow_Line"].copy() # 数字3对应ICA的Active # self.df_acc = self.df[self.df['ACC_status'] == "Shut_off"].copy() # 数字3对应ICA的Active # self.df_acc = self.df[self.df['ACC_status'] == "Active"].copy() # 数字3对应ICA的Active if self.df_acc.empty: self.result['statusFlag']['function_ACC'] = False else: self.result['statusFlag']['function_ACC'] = True def _get_first_change_index_cruise(self): """ 获取数据集中第一次巡航速度发生变化的索引。 Args: 无参数。 Returns: Union[int, None]: 如果存在巡航速度发生变化的索引，则返回第一个发生变化的索引（int类型）； 如果不存在，则返回None。 """ change_indices = self.df_acc[self.df_acc['set_cruise_speed'] != self.df_acc['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 _find_stable_speed_cruise(self, window_size, percent_deviation, set_value): """ 在给定的速度数据中查找稳定的巡航速度段，并计算该段的平均速度。 Args: window_size (int): 滑动窗口的大小，表示用于计算平均速度的速度数据点数量。 percent_deviation (float): 设定值的允许偏差百分比。 set_value (float): 期望的稳定速度设定值。 Returns: None """ # speed_data = self.df['speedX'].values speed_data = self.df_acc['speedX'].values # .tolist() deviation = set_value * (percent_deviation / 100) stable_start = None stable_average_speed = None for i in range(len(speed_data) - window_size + 1): window_data = speed_data[i:i + window_size] if all(set_value - deviation <= s <= set_value + deviation for s in window_data): if stable_start is None: stable_start = i stable_end = i + window_size - 1 stable_average_speed = np.mean(window_data) j = i + window_size while j < len(speed_data) - window_size + 1: next_window_data = speed_data[j:j + window_size] if all(set_value - deviation <= s <= set_value + deviation for s in next_window_data): stable_end = j + window_size - 1 stable_average_speed = (stable_average_speed * (j - stable_start) + sum(next_window_data)) / ( j - stable_start + window_size) j += window_size else: stable_start = j + window_size - 1 stable_end = i + window_size - 1 stable_average_speed = np.mean(window_data) break # self.stable_start_time_cruise = self.df['simTime'].iloc[stable_start] self.stable_average_speed = stable_average_speed def data_analyze(self): first_change_index = self._get_first_change_index_cruise() if not first_change_index: self.steady_error_cruise = 0 self.result['value'] = [abs(round(self.steady_error_cruise, 3))] print(f"steady_error_cruise: {abs(self.steady_error_cruise)}") else: set_cruise_speed = self.df_acc.loc[first_change_index, 'set_cruise_speed'] self._find_stable_speed_cruise(window_size=4, percent_deviation=5, set_value=set_cruise_speed) if self.stable_average_speed: self.steady_error_cruise = (self.stable_average_speed - set_cruise_speed) * 100 / self.stable_average_speed else: self.steady_error_cruise = Max self.result['value'] = [abs(round(self.steady_error_cruise, 3))] print(f"steady_error_cruise: {abs(self.steady_error_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_acc.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:[steady_error_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