cicv_evaluate_master-single_eval/demoICA/ICA_0702(1).py

import pandas as pd
import numpy as np


class ICA:
    def __init__(self, df):
        self.df = df    #object_df
        self.segmented_data = []

        """
        注释：zhangyu 0626 新增
        """
        self.roadPos_df = None
        self.roadMark_df = None

        """
        纵向指标
        """
        self.stable_start_time_cruise = None
        self.stable_average_speed = None
        self.percentage_accel = None
        self.percentage_lon_acc_roc = None
        self.delay_time_cruise = None
        self.delay_time_THW = None
        self.rise_time_cruise = None
        self.rise_time_THW = None
        self.peak_time_cruise = None
        self.peak_time_THW = None
        self.overshoot_cruise = None
        self.overshoot_THW = None
        self.settling_time = None
        self.steady_error_cruise = None
        self.steady_error_THW = None
        self.stable_start_time_THW = None
        self.stable_average_THW = None

        """
        横向指标
        """
        self.lateral_control01_minimum_distance_nearby_lane = None
        self.lateral_control02_maximum_lateral_offset = None
        self.lateral_control03_relative_lateral_offset_distribution_expectation = None
        self.lateral_control04_relative_lateral_offset_distribution_standard_deviation = None
        self.lateral_control05_absolute_lateral_offset_distribution_expectation = None
        self.lateral_control06_absolute_lateral_offset_distribution_standard_deviation = None
        self.lateral_control07_maximum_lateral_deviation = None
        self.lateral_control08_minimum_lateral_deviation = None
        self.lateral_control09_absolute_position_oscillation_frequency = None
        self.lateral_control10_absolute_position_oscillation_difference = None
        self.lateral_control11_maximum_heading_deviation = None
        self.lateral_control12_relative_position_oscillation_frequency = None
        self.lateral_control13_relative_position_oscillation_difference = None

    def _segment_data_by_status(self, status_name, status_value):
        """
        根据状态名称和状态值对DataFrame中的数据进行分段处理。
        
        Args:
            status_name (str): DataFrame中用于标识状态的列名。
            status_value (Any): 需要进行分段处理的状态值。
        
        Returns:
            None
        
        Raises:
            KeyError: 如果DataFrame中不存在'ACC_status'或'simTime'列时，抛出KeyError异常。
        
        """
        if 'ACC_status' not in self.df.columns:
            raise KeyError("'ACC_status' column not found in the dataframe.")
        if 'simTime' not in self.df.columns:
            raise KeyError("'simTime' column not found in the dataframe.")

        filtered_data = self.df[self.df[status_name] == status_value]

        current_segment = []

        for index, row in filtered_data.iterrows():
            if not current_segment or abs(float(row['simTime']) - float(current_segment[-1]['simTime'])) <= 0.01:
                current_segment.append(row)
            else:
                self.segmented_data.append(pd.DataFrame(current_segment))
                current_segment = [row]

        if current_segment:
            self.segmented_data.append(pd.DataFrame(current_segment))

    def _get_headway_time(self, df):
        pass

    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
        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 _find_stable_THW(self, window_size, percent_deviation, set_value):
        """
        在给定的数据窗口中查找稳定跟车时距离THW，并计算该段内THW的平均值。
        
        Args:
            window_size (int): 窗口大小，表示在数据中寻找稳定段时考虑的连续数据点数量。
            percent_deviation (float): THW值相对于设定值的允许偏差百分比。
            set_value (float): THW的设定值。
        
        Returns:
            None
        
        """
        THW = self.df['THW'].values
        deviation = set_value * (percent_deviation / 100)
        stable_start = None
        stable_average_THW = None

        for i in range(len(THW) - window_size + 1):
            window_data = THW[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_THW = np.mean(window_data)

                j = i + window_size
                while j < len(THW) - window_size + 1:
                    next_window_data = THW[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_THW = (stable_average_THW * (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_THW = np.mean(window_data)
                        break

        self.stable_start_time_THW = self.df['simTime'].iloc[stable_start]
        self.stable_average_THW = stable_average_THW

    def _find_closest_time_stamp_cruise(self, df, target_speed, start_index):
        """
        在给定的数据帧df中，从start_index索引位置开始，查找与目标速度target_speed最接近的时间戳。
        
        Args:
            df (pd.DataFrame): 包含速度和时间戳等信息的数据帧，需要至少包含'speedX'和'simTime'两列。
            target_speed (float): 目标速度值，用于在数据帧中查找最接近此值的时间戳。
            start_index (int): 开始查找的索引位置，即在数据帧df中从该索引位置开始向后查找。
        
        Returns:
            pd.Timestamp: 与目标速度最接近的时间戳。
        
        """
        subset = df.loc[start_index + 1:]
        speed_diff = np.abs(subset['speedX'] - target_speed)
        closest_index = speed_diff.idxmin()
        closest_timestamp = subset.loc[closest_index, 'simTime']
        return closest_timestamp

    def _find_closest_time_stamp_THW(self, df, target_THW, start_index):
        """
        在DataFrame中找到与目标THW值最接近的时间戳。
        
        Args:
            df (pandas.DataFrame): 包含'THW'和'simTime'列的DataFrame，其中'THW'表示目标变量，'simTime'表示时间戳。
            target_THW (float): 目标THW值。
            start_index (int): 开始搜索的索引位置（不包含）。
        
        Returns:
            pandas.Timestamp: 与目标THW值最接近的时间戳。
        
        """
        subset = df.loc[start_index + 1:]
        THW_diff = np.abs(subset['THW'] - target_THW)
        closest_index = THW_diff.idxmin()
        closest_timestamp = subset.loc[closest_index, 'simTime']
        return closest_timestamp

    def _get_first_change_index_cruise(self):
        """
        获取数据集中第一次巡航速度发生变化的索引。
        
        Args:
            无参数。
        
        Returns:
            Union[int, None]: 如果存在巡航速度发生变化的索引，则返回第一个发生变化的索引（int类型）；
            如果不存在，则返回None。
        
        """
        change_indices = self.df[self.df['set_cruise_speed'] != self.df['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 _get_first_change_index_THW(self):
        """
        获取DataFrame中'set_headway_time'列首次发生变化的索引值。
        
        Args:
            无参数。
        
        Returns:
            Union[int, None]: 如果存在变化，则返回首次发生变化的索引值（int类型），否则返回None。
        
        """
        change_indices = self.df[self.df['set_headway_time'] != self.df['set_headway_time'].shift()].index
        if not change_indices.empty:
            first_change_index = change_indices.min()
        else:
            first_change_index = None
        return first_change_index

    def calculate_reasonable_acceleration_percentage(self):
        """
        计算合理的加速度百分比。
        
        Args:
            无。


        Returns:
            无返回值，但会更新对象的 percentage_accel 属性，表示合理的加速度百分比（单位：%）。
        
        """
        df = self.df_follow.copy()
        col_list = ['simTime', 'simFrame', 'playerId', 'v', 'accel', 'accel', 'lon_acc_roc']  # target_id
        df = df[col_list].copy()
        count_accel_in_range = 0
        total_accel_frames = 0

        ego_df = df[df['playerId'] == 1][['simTime', 'simFrame', 'v', 'accel', 'lon_acc_roc']]
        for row in self.segmented_data:
            df = row[col_list].copy()
            filtered_data = df[(df['accel'] >= -5) & (df['accel'] <= 4)]
            count_accel_in_range = count_accel_in_range + len(filtered_data)
            total_accel_frames = total_accel_frames + len(df)
        self.percentage_accel = (count_accel_in_range / total_accel_frames) * 100

        print(f"percentage_accel: {self.percentage_accel}")

    def calculate_reasonable_acceleration_change_rate_percentage(self):
        """
        计算纵向加速度变化率（lon_acc_roc）的合理变化率百分比。
        
        Args:
            无参数。
        
        Returns:
            无返回值，但会更新对象的 percentage_lon_acc_roc 属性。
        
        """
        df = self.df_follow.copy()
        col_list = ['simTime', 'simFrame', 'playerId', 'v', 'accel', 'accel', 'lon_acc_roc']  # target_id
        df = df[col_list].copy()
        count_lon_acc_roc_in_range = 0
        total_lon_acc_roc_frames = 0
        for row in self.segmented_data:
            df = row[col_list].copy()
            filtered_data = df[(df['lon_acc_roc'] >= -5) & (df['lon_acc_roc'] <= 6)]
            count_lon_acc_roc_in_range = count_lon_acc_roc_in_range + len(filtered_data)
            total_lon_acc_roc_frames = total_lon_acc_roc_frames + len(df)
        self.percentage_lon_acc_roc = (count_lon_acc_roc_in_range / total_lon_acc_roc_frames) * 100

    def get_delay_time_cruise(self):
        """
        计算巡航速度改变后的延迟时间。
        
        Args:
            无。
        
        Returns:
            无返回值，但会设置self.delay_time为计算得到的延迟时间（秒）。
        
        Raises:
            无特定异常，但会捕获并打印任何发生的异常信息。
        
        """
        try:
            change_indices = self.df[self.df['set_cruise_speed'] != self.df['set_cruise_speed'].shift()].index
            print(f"Change indices of set speed: {change_indices}")

            if not change_indices.empty:
                first_change_index = change_indices[change_indices != 0].min()
                set_cruise_speed_at_change = self.df.loc[first_change_index, 'set_cruise_speed']
                timestamp_at_change = self.df.loc[first_change_index, 'simTime']
                print(f"Set speed at first change: {set_cruise_speed_at_change}, Timestamp: {timestamp_at_change}")

                target_speed = (self.stable_average_speed + self.df.loc[first_change_index, 'speedX']) / 2
                closest_index = (self.df['speedX'] - target_speed).abs().idxmin()
                closest_current_speed = self.df.loc[closest_index, 'speedX']
                closest_timestamp = self.df.loc[closest_index, 'simTime']
                print(f"Closest speed: {closest_current_speed} at time: {closest_timestamp}")

                self.delay_time_cruise = closest_timestamp - timestamp_at_change
                print(f"Delay time: {self.delay_time_cruise}")

            else:
                print("No valid change point for further calculation.")

        except Exception as e:
            print(f"An error occurred: {e}")

    def get_delay_time_THW(self):
        """
        计算THW改变后的延迟时间。
        
        Args:
            无。
        
        Returns:
            无返回值，但会设置self.delay_time为计算得到的延迟时间（秒）。
        
        Raises:
            无特定异常，但会捕获并打印任何发生的异常信息。
        
        """
        try:
            change_indices = self.df[self.df['set_headway_time'] != self.df['set_headway_time'].shift()].index
            print(f"Change indices of set speed: {change_indices}")

            if not change_indices.empty:
                first_change_index = change_indices[change_indices != 0].min()
                set_THW_at_change = self.df.loc[first_change_index, 'set_headway_time']
                timestamp_at_change = self.df.loc[first_change_index, 'simTime']
                print(f"Set THW at first change: {set_THW_at_change}, Timestamp: {timestamp_at_change}")

                target_THW = (self.stable_average_THW + self.df.loc[first_change_index, 'set_headway_time']) / 2
                closest_index = (self.df['set_headway_time'] - target_THW).abs().idxmin()
                closest_current_THW = self.df.loc[closest_index, 'set_headway_time']
                closest_timestamp = self.df.loc[closest_index, 'simTime']
                print(f"Closest speed: {closest_current_THW} at time: {closest_timestamp}")

                self.delay_time_THW = closest_timestamp - timestamp_at_change
                print(f"Delay time: {self.delay_time_THW}")

            else:
                print("No valid change point for further calculation.")

        except Exception as e:
            print(f"An error occurred: {e}")

    def get_rise_time_cruise(self):
        """
        计算巡航时从初速度到达稳定平均速度的90%的所需时间（rise time）。
        
        Args:
            无。
        
        Returns:
            无返回值，但会设置实例属性self.rise_time为计算得到的rise time。
        
        """
        first_change_index = self._get_first_change_index_cruise()

        initial_speed = self.df.loc[first_change_index, 'speedX']

        target_speed_90 = initial_speed + (self.stable_average_speed - initial_speed) * 0.9
        target_speed_10 = initial_speed + (self.stable_average_speed - initial_speed) * 0.1

        timestamp_at_10 = self._find_closest_time_stamp_cruise(self.df, target_speed_10, first_change_index)
        timestamp_at_90 = self._find_closest_time_stamp_cruise(self.df, target_speed_90, first_change_index)

        print(f"Closest speed at 10% range from set speed: {timestamp_at_10}")
        print(f"Closest speed at 90% range from set speed: {timestamp_at_90}")

        self.rise_time_cruise = timestamp_at_90 - timestamp_at_10
        print(f"Rise time: {self.rise_time_cruise}")

    def get_rise_time_THW(self):
        """
        计算巡航时从初THW到达稳定THW的90%的所需时间（rise time）。
        
        Args:
            无。
        
        Returns:
            无返回值，但会设置实例属性self.rise_time_THW为计算得到的rise time。
        
        """
        first_change_index = self._get_first_change_index_THW()

        initial_THW = self.df.loc[first_change_index, 'THW']

        target_THW_90 = initial_THW + (self.stable_average_THW - initial_THW) * 0.9
        target_THW_10 = initial_THW + (self.stable_average_THW - initial_THW) * 0.1

        timestamp_at_10 = self._find_closest_time_stamp_THW(self.df, target_THW_10, first_change_index)
        timestamp_at_90 = self._find_closest_time_stamp_THW(self.df, target_THW_90, first_change_index)

        print(f"Closest speed at 10% range from set speed: {timestamp_at_10}")
        print(f"Closest speed at 90% range from set speed: {timestamp_at_90}")

        self.rise_time_THW = timestamp_at_90 - timestamp_at_10
        print(f"Rise time: {self.rise_time_THW}")

    def get_peak_time_cruise(self):
        """
        计算巡航阶段速度峰值时间。
        
        Args:
            无。
        
        Returns:
            无返回值，但将计算得到的巡航阶段速度峰值时间（从首次变化时间开始计时）存储在实例变量 self.peak_time 中。
        
        """
        first_change_index = self._get_first_change_index_cruise()
        start_time = self.df.loc[first_change_index, 'simTime']
        peak_time = self.df.loc[self.df['speedX'].idxmax(), 'simTime']
        self.peak_time_cruise = peak_time - start_time

    def get_peak_time_THW(self):
        """
        计算THW阶段速度峰值时间。
        
        Args:
            无。
        
        Returns:
            无返回值，但将计算得到的巡航阶段速度峰值时间（从首次变化时间开始计时）存储在实例变量 self.peak_time 中。
        
        """
        first_change_index = self._get_first_change_index_THW()
        start_time = self.df.loc[first_change_index, 'simTime']
        peak_time = self.df.loc[self.df['THW'].idxmax(), 'simTime']
        self.peak_time_THW = peak_time - start_time

    def get_overshoot_cruise(self):
        """
        计算超速巡航超调量。
        
        Args:
            无。
        
        Returns:
            无返回值，但会更新类的成员变量 `overshoot`。
        
        """
        first_change_index = self._get_first_change_index_cruise()

        initial_speed = self.df.loc[first_change_index, 'speedX']

        if initial_speed > self.stable_average_speed:
            self.overshoot_cruise = (self.stable_average_speed - self.df[
                'speedX'].min()) * 100 / self.stable_average_speed
        elif initial_speed < self.stable_average_speed:
            self.overshoot_cruise = (self.df[
                                         'speedX'].max() - self.stable_average_speed) * 100 / self.stable_average_speed
        else:
            self.overshoot_cruise = 0

    def get_overshoot_THW(self):
        """
        计算超速THW超调量。
        
        Args:
            无。
        
        Returns:
            无返回值，但会更新类的成员变量 `overshoot`。
        
        """
        first_change_index = self._get_first_change_index_THW()

        initial_THW = self.df.loc[first_change_index, 'THW']

        if initial_THW > self.stable_average_THW:
            self.overshoot_THW = (self.stable_average_THW - self.df['THW'].min()) * 100 / self.stable_average_THW
        elif initial_THW < self.stable_average_THW:
            self.overshoot_THW = (self.df['THW'].max() - self.stable_average_THW) * 100 / self.stable_average_THW
        else:
            self.overshoot_THW = 0

    def get_steady_error_cruise(self):
        """
        计算巡航稳态误差。
        
        Args:
            无参数。
        
        Returns:
            无返回值，但会更新对象的steady_error属性。
        
        Raises:
            ValueError: 如果stable_average_speed为None时，会抛出此异常。
        
        """

        first_change_index = self._get_first_change_index_cruise()
        set_cruise_speed = self.df.loc[first_change_index, '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:
            raise ValueError("Stable average speed is None.")

    def get_steady_error_THW(self):
        """
        计算THW稳态误差。
        
        Args:
            无参数。
        
        Returns:
            无返回值，但会更新对象的steady_error属性。
        
        Raises:
            ValueError: 如果stable_average_THW为None时 ，会抛出此异常。
        
        """

        first_change_index = self._get_first_change_index_THW()
        set_THW = self.df.loc[first_change_index, 'set_headway_time']
        if self.stable_average_THW:
            self.steady_error_THW = (self.stable_average_THW - set_THW) * 100 / self.stable_average_THW
        else:
            raise ValueError("Stable average THW is None.")

    def get_target_recongnition_cut_in_time_mid(self):
        """
        获取目标识别切入时间中间值。
        
        Args:
            无参数。
        
        Returns:
            float: 目标识别切入时间中点（单位：秒），如果未获取到则返回None。
        
        """
        pass

    def get_target_recongnition_cut_in_time_SD(self):
        """
        获取目标识别切入时间（SD）标准差
        
        Args:
            无
        
        Returns:
            目标识别切入时间（SD）的值，若函数无具体返回值则无返回（在此情况下，该函数被标记为 pass）
        
        """
        pass

    def get_target_recongnition_cut_off_time_mid(self):
        """
        获取目标识别切出时间中间值。
        
        Args:
            无参数。
        
        Returns:
            float: 目标识别切出时间中点（单位：秒），如果未获取到则返回None。
        
        """
        pass

    def get_target_recongnition_cut_off_time_SD(self):
        """
        获取目标识别切出时间（SD）标准差
        
        Args:
            无
        
        Returns:
            目标识别切出时间（SD）的值，若函数无具体返回值则无返回（在此情况下，该函数被标记为 pass）
        
        """
        pass

    def get_stop_time(self):
        pass

    def Compute_nearby_distance_to_lane_boundary(self, x, width_ego):
        """
        计算自车到车道边界线的距离，需要减去自车宽度

        Args:
            x.lateralDist ：到边界线距离
            width_ego：车宽

        Returns:
            自车到车道边界线的距离

        """
        if x.lateralDist < abs(x.right_lateral_distance):
            return x.lateralDist - width_ego / 2
        else:
            return abs(x.right_lateral_distance) - width_ego / 2

    def func_laneOffset_abs(self, x):
        """
        计算自车横向偏移量的绝对值

        Args:
            x.laneOffset ：横向偏移量

        Returns:
            横向偏移量的绝对值

        """
        return abs(x.laneOffset)

    def func_center_line_cycle_optical(self, l):
        """
        计算偏离车道中心线的振荡周期和极差,极大值要大于0，极小值要小于0

        Args:
            l ：波的数值列表，如 l = [1, 2, 1, 2, 1, 0, -1, 0, 1, 2, 1, 2, 0, -1, 2]

        Returns:
            cycle：振荡周期数
            maximum_range：最大极差

        """
        # print("\n穿过y=0的周期数: ")
        length = len(l)
        number_peak = 0
        number_trough = 0
        # 生成新的、只含有波峰波谷的列表
        new_list = []
        plus = []
        minus = []
        for number, value in enumerate(l):
            if number == 0:
                if value > l[number + 1] and value > 0:
                    number_peak += 1
                    plus.append(value)
                if value < l[number + 1] and value < 0:
                    number_trough += 1
                    minus.append(value)
                continue

            if number == length - 1:
                if value > l[number - 1] and value > 0:
                    number_peak += 1
                    plus.append(value)
                    if len(minus) != 0:
                        new_list.append(min(minus))
                        minus = []
                if value < l[number - 1] and value < 0:
                    number_trough += 1
                    minus.append(value)
                    if len(plus) != 0:
                        new_list.append(max(plus))
                        plus = []
                if len(minus) != 0:
                    new_list.append(min(minus))
                    minus = []
                if len(plus) != 0:
                    new_list.append(max(plus))
                    plus = []
                continue

            if value >= l[number - 1] and value > l[number + 1] and value > 0:
                number_peak += 1
                plus.append(value)
                if len(minus) != 0:
                    new_list.append(min(minus))
                    minus = []
            if value < l[number - 1] and value <= l[number + 1] and value < 0:
                number_trough += 1
                minus.append(value)
                if len(plus) != 0:
                    new_list.append(max(plus))
                    plus = []
        cycle = min(number_peak, number_trough) - 1
        difference_list = []
        for i in range(len(new_list) - 1):
            difference = abs(new_list[i] - new_list[i + 1])
            difference_list.append(difference)
        if len(difference_list) == 0:
            maximum_range = -1
            # print(f"极差： {maximum_range}")
        else:
            maximum_range = max(difference_list)
            # print(f"极差： {maximum_range}")
        return cycle, maximum_range

    def func_normal_cycle_optical(self, l):
        """
        计算振荡周期和极差,两个相邻的极大值（或极小值）算一次振荡

        Args:
            l ：波的数值列表，如 l = [1, 2, 1, 2, 1, 0, -1, 0, 1, 2, 1, 2, 0, -1, 2]

        Returns:
            cycle：振荡周期数
            maximum_range：最大极差

        """
        # print("正常的周期数: ")
        length = len(l)
        number_peak = 0
        number_trough = 0
        # 生成新的、只含有波峰波谷的列表
        new_list = []
        for number, value in enumerate(l):
            if number == 0:
                if value > l[number + 1]:
                    number_peak += 1
                    new_list.append(value)
                if value < l[number + 1]:
                    number_trough += 1
                    new_list.append(value)
                continue
            if number == length - 1:
                if value > l[number - 1]:
                    number_peak += 1
                    new_list.append(value)
                if value < l[number - 1]:
                    number_trough += 1
                    new_list.append(value)
                continue
            if value >= l[number - 1] and value > l[number + 1]:
                number_peak += 1
                new_list.append(value)
            if value < l[number - 1] and value <= l[number + 1]:
                number_trough += 1
                new_list.append(value)
        if abs(number_peak - number_trough) > 1:
            pass
        else:
            cycle = max(number_peak, number_trough) - 1
        difference_list = []
        for i in range(len(new_list) - 1):
            difference = abs(new_list[i] - new_list[i + 1])
            difference_list.append(difference)
        if len(difference_list) == 0:
            maximum_range = -1
            # print(f"极差： {maximum_range}")
        else:
            maximum_range = max(difference_list)
            # print(f"极差： {maximum_range}")
        return cycle, maximum_range

    def _lateral_control01_minimum_distance_nearby_lane(self):
        """
        横向控制01指标：计算离近侧车道线最小距离（m）

        Args:

        Returns:
            None

        """
        roadMark_df = self.roadMark_df
        player_df = self.df
        ego_df = player_df[player_df.playerId == 1]
        width_ego = ego_df['dimY'].values.tolist()[0]
        # 提取距离左车道线和右车道线距离
        roadMark_left_df = roadMark_df[roadMark_df.id == 0].reset_index(drop=True)
        roadMark_right_df = roadMark_df[roadMark_df.id == 2].reset_index(drop=True)
        roadMark_left_df['right_lateral_distance'] = roadMark_right_df['lateralDist']
        # 计算到车道边界线距离
        roadMark_left_df['nearby_distance_to_lane_boundary'] = roadMark_left_df.apply(
            lambda x: self.Compute_nearby_distance_to_lane_boundary(x, width_ego), axis=1)
        nearby_distance_to_lane_boundary = min(roadMark_left_df['nearby_distance_to_lane_boundary'])
        self.lateral_control01_minimum_distance_nearby_lane = round(nearby_distance_to_lane_boundary, 3)

    def _lateral_control02_maximum_lateral_offset(self):
        """
        横向控制02指标：计算最大横向偏移量（m）

        Args:

        Returns:
            None

        """
        roadPos_df = self.roadPos_df
        # 提取自车的roadPos数据
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        # 计算横向偏移量的绝对值
        roadPos_ego_df['laneOffset_abs'] = roadPos_ego_df.apply(lambda x: self.func_laneOffset_abs(x), axis=1)
        max_laneOffset_abs_index = roadPos_ego_df['laneOffset_abs'].idxmax()
        row_with_max_value = roadPos_ego_df.iloc[max_laneOffset_abs_index].laneOffset
        self.lateral_control02_maximum_lateral_offset = round(row_with_max_value, 3)

    def _lateral_control03_relative_lateral_offset_distribution_expectation(self):
        """
        横向控制03指标：计算相对横向偏移量分布期望（m）

        Args:

        Returns:
            None

        """
        # 提取自车宽度
        roadPos_df = self.roadPos_df
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        mean_laneOffset_index = roadPos_ego_df['laneOffset'].mean()
        self.lateral_control03_relative_lateral_offset_distribution_expectation = round(mean_laneOffset_index, 3)

    def _lateral_control04_relative_lateral_offset_distribution_standard_deviation(self):
        """
        横向控制04指标：计算相对横向偏移量分布标准差（m）

        Args:

        Returns:
            None

        """
        # 提取自车宽度
        roadPos_df = self.roadPos_df
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        mean_laneOffset_index = roadPos_ego_df['laneOffset'].std()
        self.lateral_control04_relative_lateral_offset_distribution_standard_deviation = round(mean_laneOffset_index, 3)

    def _lateral_control05_absolute_lateral_offset_distribution_expectation(self):
        """
        横向控制05指标：计算绝对横向偏移量分布期望（m）

        Args:

        Returns:
            None

        """
        roadPos_df = self.roadPos_df
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        roadPos_ego_df['laneOffset_abs'] = roadPos_ego_df.apply( \
            lambda x: self.func_laneOffset_abs(x), axis=1)
        mean_laneOffset_index = roadPos_ego_df['laneOffset_abs'].mean()
        self.lateral_control05_absolute_lateral_offset_distribution_expectation = round(mean_laneOffset_index, 3)

    def _lateral_control06_absolute_lateral_offset_distribution_standard_deviation(self):
        """
        横向控制06指标：计算绝对横向偏移量分布标准差（m）

        Args:

        Returns:
            None

        """
        roadPos_df = self.roadPos_df
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        roadPos_ego_df['laneOffset_abs'] = roadPos_ego_df.apply( \
            lambda x: self.func_laneOffset_abs(x), axis=1)
        mean_laneOffset_index = roadPos_ego_df['laneOffset_abs'].std()
        self.lateral_control06_absolute_lateral_offset_distribution_standard_deviation = round(mean_laneOffset_index, 3)

    def _lateral_control07_maximum_lateral_deviation(self):
        """
        横向控制07指标：横向距离极大值（m）

        Args:

        Returns:
            None

        """
        roadPos_df = self.roadPos_df
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        roadPos_ego_df['laneOffset_abs'] = roadPos_ego_df.apply(lambda x: self.func_laneOffset_abs(x), axis=1)
        max_laneOffset_abs_index = roadPos_ego_df['laneOffset_abs'].idxmax()
        row_with_max_value = roadPos_ego_df.iloc[max_laneOffset_abs_index].laneOffset
        self.lateral_control07_maximum_lateral_deviation = row_with_max_value

    def _lateral_control08_minimum_lateral_deviation(self):
        """
        横向控制08指标：横向距离极小值（m）

        Args:

        Returns:
            None

        """
        roadPos_df = self.roadPos_df
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        roadPos_ego_df['laneOffset_abs'] = roadPos_ego_df.apply(lambda x: self.func_laneOffset_abs(x), axis=1)
        max_laneOffset_abs_index = roadPos_ego_df['laneOffset_abs'].idxmin()
        row_with_min_value = roadPos_ego_df.iloc[max_laneOffset_abs_index].laneOffset
        self.lateral_control08_minimum_lateral_deviation = row_with_min_value

    def _lateral_control09_absolute_position_oscillation_frequency(self):
        """
        横向控制09指标：横向绝对位置振荡频率（Hz）

        Args:

        Returns:
            None

        """
        roadPos_df = self.roadPos_df
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        roadPos_ego_list = roadPos_ego_df['laneOffset'].values.tolist()
        cycle_number, _ = self.func_center_line_cycle_optical(roadPos_ego_list)
        self.lateral_control09_absolute_position_oscillation_frequency = round(cycle_number, 3)

    def _lateral_control10_absolute_position_oscillation_difference(self):
        """
        横向控制10指标：横向绝对位置振荡极差(m)

        Args:

        Returns:
            None

        """
        roadPos_df = self.roadPos_df
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        roadPos_ego_list = roadPos_ego_df['laneOffset'].values.tolist()
        _, maximum_range = self.func_center_line_cycle_optical(roadPos_ego_list)
        self.lateral_control10_absolute_position_oscillation_difference = round(maximum_range, 3)

    def _lateral_control11_maximum_heading_deviation(self):
        """
        横向控制11指标：最大偏航角

        Args:

        Returns:
            None

        """
        player_df = self.df
        road_mark_df = self.roadMark_df
        ego_df = player_df[player_df.playerId == 1].reset_index(drop=True)

        # 左车道线曲率，右车道线曲率，求二者平均值，计算车道线曲率，再与自车朝向相减
        road_mark_left_df = road_mark_df[road_mark_df.id == 0].reset_index(drop=True)
        road_mark_right_df = road_mark_df[road_mark_df.id == 2].reset_index(drop=True)
        road_mark_left_df['curvHor_left'] = road_mark_left_df['curvHor']
        road_mark_left_df['curvHor_right'] = road_mark_right_df['curvHor']
        road_mark_left_df['curvHor_middle'] = road_mark_left_df[['curvHor_left', 'curvHor_right']].apply( \
            lambda x: (x['curvHor_left'] + x['curvHor_right']) / 2, axis=1)
        ego_df['curvHor_middle'] = road_mark_left_df['curvHor_middle']
        ego_df['heading_deviation_abs'] = ego_df[['curvHor_middle', 'posH']].apply( \
            lambda x: abs(x['posH'] - x['curvHor_middle']), axis=1)
        row_with_max_value = max(ego_df['heading_deviation_abs'])
        self.lateral_control11_maximum_heading_deviation = row_with_max_value

    def _lateral_control12_relative_position_oscillation_frequency(self):
        """
        横向控制12指标：横向相对位置振荡频率（Hz）

        Args:

        Returns:
            None

        """
        roadPos_df = self.roadPos_df
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        roadPos_ego_list = roadPos_ego_df['laneOffset'].values.tolist()
        cycle_number, _ = self.func_normal_cycle_optical(roadPos_ego_list)
        self.lateral_control12_relative_position_oscillation_frequency = round(cycle_number, 3)

    def _lateral_control13_relative_position_oscillation_difference(self):
        """
        横向控制13指标：横向相对位置振荡极差(m)

        Args:

        Returns:
            None

        """
        roadPos_df = self.roadPos_df
        roadPos_ego_df = roadPos_df[roadPos_df.playerId == 1].reset_index(drop=True)
        roadPos_ego_list = roadPos_ego_df['laneOffset'].values.tolist()
        _, maximum_range = self.func_normal_cycle_optical(roadPos_ego_list)
        self.lateral_control13_relative_position_oscillation_difference = round(maximum_range, 3)