zhaoyuan/modules/metric/function.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
##################################################################
#
# Copyright (c) 2025 CICV, Inc. All Rights Reserved
#
##################################################################
"""
@Authors:           zhanghaiwen(zhanghaiwen@china-icv.cn)
@Data:              2025/01/5
@Last Modified:     2025/01/5
@Summary:           Function Metrics Calculation
"""

import sys
from pathlib import Path

# 添加项目根目录到系统路径
root_path = Path(__file__).resolve().parent.parent
sys.path.append(str(root_path))
print(root_path)

from modules.lib.score import Score
from modules.lib.log_manager import LogManager
import numpy as np
from typing import Dict, Tuple, Optional, Callable, Any
import pandas as pd
import yaml
from modules.lib.chart_generator import generate_function_chart_data
from shapely.geometry import Point, Polygon
from modules.lib.common import get_interpolation
import pandas as pd
import yaml
import math

# ----------------------
# 基础工具函数 (Pure functions)
# ----------------------
scenario_sign_dict = {
    "LeftTurnAssist": 206,
    "HazardousLocationW": 207,
    "RedLightViolationW": 208,
    "AbnormalVehicleW":209,
    "NsightVulnerableRoadUserCollisionW":210,
    "LitterW":211,
    "ForwardCollisionW":212,
    "VisibilityW":213,
    "EmergencyBrakeW":214,
    "IntersectionCollisionW":215,
    "BlindSpotW":216,
    "DoNotPassW":217,
    "ControlLossW":218,
    "FrontTrafficJamW":219,
    "EmergencyVehicleW":220,
    "CooperativeVehicleMerge":221,
    "CooperativeLaneChange":223,
    "VulnerableRoadUserCollisionW":224,
    "CooperativeIntersectionPassing":225,
    "RampMerge":226,
    "DrivingLaneRecommendation":227,
    "TrafficJamW": 228,
    "DynamicSpeedLimitingInformation": 229,
    "EmergencyVehiclesPriority": 230,
    "RemoteSupervision": 231,
    "SignalLightReminder": 232,
    "OtherVehicleRedLightViolationW": 233,
    "GreenLightOptimalSpeedAdvisory": 234,
}

def _is_pedestrian_in_crosswalk(polygon, test_point) -> bool:
    polygon = Polygon(polygon)
    point = Point(test_point)
    return polygon.contains(point)


def _is_segment_by_interval(time_list, expected_step) -> list:
    """
    根据时间戳之间的间隔进行分段。

    参数:
    time_list (list): 时间戳列表。
    expected_step (float): 预期的固定步长。

    返回:
    list: 分段后的时间戳列表，每个元素是一个子列表。
    """
    if not time_list:
        return []

    segments = []
    current_segment = [time_list[0]]

    for i in range(1, len(time_list)):
        actual_step = time_list[i] - time_list[i - 1]
        if actual_step != expected_step:
            # 如果间隔不符合预期，则开始一个新的段
            segments.append(current_segment)
            current_segment = [time_list[i]]
        else:
            # 否则，将当前时间戳添加到当前段中
            current_segment.append(time_list[i])

    # 添加最后一个段
    if current_segment:
        segments.append(current_segment)

    return segments


# 寻找二级指标的名称
def find_nested_name(data):
    """
    查找字典中嵌套的name结构。

    :param data: 要搜索的字典
    :return: 找到的第一个嵌套name结构的值，如果没有找到则返回None
    """
    if isinstance(data, dict):
        for key, value in data.items():
            if isinstance(value, dict) and 'name' in value:
                return value['name']
            # 递归查找嵌套字典
            result = find_nested_name(value)
            if result is not None:
                return result
    elif isinstance(data, list):
        for item in data:
            result = find_nested_name(item)
            if result is not None:
                return result
    return None


def calculate_distance_PGVIL(ego_pos: np.ndarray, obj_pos: np.ndarray) -> np.ndarray:
    """向量化距离计算"""
    return np.linalg.norm(ego_pos - obj_pos, axis=1)


def calculate_relative_speed_PGVIL(
        ego_speed: np.ndarray, obj_speed: np.ndarray
) -> np.ndarray:
    """向量化相对速度计算"""
    return np.linalg.norm(ego_speed - obj_speed, axis=1)


def calculate_distance(ego_df: pd.DataFrame, correctwarning: int) -> np.ndarray:
    """向量化距离计算"""
    dist = ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]['relative_dist']
    return dist


def calculate_relative_speed(ego_df: pd.DataFrame, correctwarning: int) -> np.ndarray:
    """向量化相对速度计算"""
    return ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]['composite_v']


def extract_ego_obj(data: pd.DataFrame) -> Tuple[pd.Series, pd.DataFrame]:
    """数据提取函数"""
    ego = data[data["playerId"] == 1].iloc[0]
    obj = data[data["playerId"] != 1]
    return ego, obj


def get_first_warning(data_processed) -> Optional[pd.DataFrame]:
    """带缓存的预警数据获取"""
    ego_df = data_processed.ego_data
    obj_df = data_processed.object_df

    scenario_name = find_nested_name(data_processed.function_config["function"])
    correctwarning = scenario_sign_dict.get(scenario_name)

    if correctwarning is None:
        print("无法获取正确的预警信号标志位！")
        return None
    warning_rows = ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]

    warning_times = warning_rows['simTime']
    if warning_times.empty:
        print("没有找到预警数据！")
        return None

    first_time = warning_times.iloc[0]
    return obj_df[obj_df['simTime'] == first_time]

def getAxis(heading):
    AxisL = [0, 0]
    AxisW = [0, 0]

    AxisL[0] = math.cos(heading)
    AxisL[1] = math.sin(heading)

    AxisW[0] = -math.sin(heading)
    AxisW[1] = math.cos(heading)

    return AxisL, AxisW


def getDotMultiply(firstAxis, secondAxis):
    dotMultiply = abs(firstAxis[0] * secondAxis[0] + firstAxis[1] * secondAxis[1])

    return dotMultiply


def getProjectionRadius(AxisL, AxisW, baseAxis, halfLength, halfWidth):
    projAxisL = getDotMultiply(AxisL, baseAxis)
    projAxisW = getDotMultiply(AxisW, baseAxis)

    projectionRadius = halfLength * projAxisL + halfWidth * projAxisW

    return projectionRadius


def isCollision(
    firstAxisL,
    firstAxisW,
    firstHalfLength,
    firstHalfWidth,
    secondAxisL,
    secondAxisW,
    secondHalfLength,
    secondHalfWidth,
    disVector,
):
    isCollision = True
    axes = [firstAxisL, firstAxisW, secondAxisL, secondAxisW]
    addProjectionRadius = 0
    projectionDisVecotr = 0

    for axis in axes:
        addProjectionRadius = getProjectionRadius(
            firstAxisL, firstAxisW, axis, firstHalfLength, firstHalfWidth
        ) + getProjectionRadius(
            secondAxisL, secondAxisW, axis, secondHalfLength, secondHalfWidth
        )
        projectionDisVecotr = getDotMultiply(disVector, axis)

        if projectionDisVecotr > addProjectionRadius:
            isCollision = False
            break
    return isCollision


def funcIsCollision(
    firstDimX,
    firstDimY,
    firstOffX,
    firstOffY,
    firstX,
    firstY,
    firstHeading,
    secondDimX,
    secondDimY,
    secondOffX,
    secondOffY,
    secondX,
    secondY,
    secondHeading,
):
    firstAxisL = getAxis(firstHeading)[0]
    firstAxisW = getAxis(firstHeading)[1]
    secondAxisL = getAxis(secondHeading)[0]
    secondAxisW = getAxis(secondHeading)[1]

    firstHalfLength = 0.5 * firstDimX
    firstHalfWidth = 0.5 * firstDimY
    secondHalfLength = 0.5 * secondDimX
    secondHalfWidth = 0.5 * secondDimY

    firstCenter = [0, 0]
    secondCenter = [0, 0]
    disVector = [0, 0]

    firstCenter[0] = firstX + firstOffX * math.cos(firstHeading)
    firstCenter[1] = firstY + firstOffX * math.sin(firstHeading)
    secondCenter[0] = secondX + secondOffX * math.cos(secondHeading)
    secondCenter[1] = secondY + secondOffX * math.sin(secondHeading)

    disVector[0] = secondCenter[0] - firstCenter[0]
    disVector[1] = secondCenter[1] - firstCenter[1]

    varIsCollision = isCollision(
        firstAxisL,
        firstAxisW,
        firstHalfLength,
        firstHalfWidth,
        secondAxisL,
        secondAxisW,
        secondHalfLength,
        secondHalfWidth,
        disVector,
    )

    return varIsCollision

# ----------------------
# 核心计算功能函数
# ----------------------
def latestWarningDistance_LST(data) -> dict:
    """预警距离计算流水线"""
    scenario_name = find_nested_name(data.function_config["function"])
    value = data.function_config["function"][scenario_name]["latestWarningDistance_LST"]["max"]
    correctwarning = scenario_sign_dict[scenario_name]
    ego_df = data.ego_data
    warning_dist = calculate_distance(ego_df, correctwarning)
    warning_speed = calculate_relative_speed(ego_df, correctwarning)

    # 将计算结果保存到data对象中，供图表生成使用
    data.warning_dist = warning_dist
    data.warning_speed = warning_speed
    data.correctwarning = correctwarning

    if warning_dist.empty:
        return {"latestWarningDistance_LST": 0.0}

    # 生成图表数据
    generate_function_chart_data(data, 'latestWarningDistance_LST')

    return {"latestWarningDistance_LST": float(warning_dist.iloc[-1]) if len(warning_dist) > 0 else value}


def earliestWarningDistance_LST(data) -> dict:
    """预警距离计算流水线"""
    scenario_name = find_nested_name(data.function_config["function"])
    value = data.function_config["function"][scenario_name]["earliestWarningDistance_LST"]["max"]
    correctwarning = scenario_sign_dict[scenario_name]
    ego_df = data.ego_data
    warning_dist = calculate_distance(ego_df, correctwarning)
    warning_speed = calculate_relative_speed(ego_df, correctwarning)

    # 将计算结果保存到data对象中，供图表生成使用
    data.warning_dist = warning_dist
    data.warning_speed = warning_speed
    data.correctwarning = correctwarning

    if len(warning_dist) == 0:
        return {"earliestWarningDistance_LST": 0.0}

    # 生成图表数据

    generate_function_chart_data(data, 'earliestWarningDistance_LST')

    return {"earliestWarningDistance_LST": float(warning_dist.iloc[0]) if len(warning_dist) > 0 else value}


def latestWarningDistance_TTC_LST(data) -> dict:
    """TTC计算流水线"""
    scenario_name = find_nested_name(data.function_config["function"])
    value = data.function_config["function"][scenario_name]["latestWarningDistance_TTC_LST"]["max"]
    correctwarning = scenario_sign_dict[scenario_name]
    ego_df = data.ego_data
    warning_dist = calculate_distance(ego_df, correctwarning)
    if warning_dist.empty:
        return {"latestWarningDistance_TTC_LST": 0.0}

    # 将correctwarning保存到data对象中，供图表生成使用
    data.correctwarning = correctwarning

    warning_speed = calculate_relative_speed(ego_df, correctwarning)

    with np.errstate(divide='ignore', invalid='ignore'):
        ttc = np.where(warning_speed != 0, warning_dist / warning_speed, np.inf)

    # 处理无效的TTC值
    for i in range(len(ttc)):
        ttc[i] = float(value) if (not ttc[i] or ttc[i] < 0) else ttc[i]

    data.warning_dist = warning_dist
    data.warning_speed = warning_speed
    data.ttc = ttc
    # 生成图表数据
    # from modules.lib.chart_generator import generate_function_chart_data
    generate_function_chart_data(data, 'latestWarningDistance_TTC_LST')

    return {"latestWarningDistance_TTC_LST": float(ttc[-1]) if len(ttc) > 0 else value}


def earliestWarningDistance_TTC_LST(data) -> dict:
    """TTC计算流水线"""
    scenario_name = find_nested_name(data.function_config["function"])
    value = data.function_config["function"][scenario_name]["earliestWarningDistance_TTC_LST"]["max"]
    correctwarning = scenario_sign_dict[scenario_name]
    ego_df = data.ego_data
    warning_dist = calculate_distance(ego_df, correctwarning)
    if warning_dist.empty:
        return {"earliestWarningDistance_TTC_LST": 0.0}

    # 将correctwarning保存到data对象中，供图表生成使用
    data.correctwarning = correctwarning

    warning_speed = calculate_relative_speed(ego_df, correctwarning)

    with np.errstate(divide='ignore', invalid='ignore'):
        ttc = np.where(warning_speed != 0, warning_dist / warning_speed, np.inf)

    # 处理无效的TTC值
    for i in range(len(ttc)):
        ttc[i] = float(value) if (not ttc[i] or ttc[i] < 0) else ttc[i]

    # 将计算结果保存到data对象中，供图表生成使用
    data.warning_dist = warning_dist
    data.warning_speed = warning_speed
    data.ttc = ttc
    data.correctwarning = correctwarning

    # 生成图表数据
    generate_function_chart_data(data, 'earliestWarningDistance_TTC_LST')

    return {"earliestWarningDistance_TTC_LST": float(ttc[0]) if len(ttc) > 0 else value}


def warningDelayTime_LST(data):
    scenario_name = find_nested_name(data.function_config["function"])
    correctwarning = scenario_sign_dict[scenario_name]
    # 将correctwarning保存到data对象中，供图表生成使用
    data.correctwarning = correctwarning
    ego_df = data.ego_data
    HMI_warning_rows = ego_df[(ego_df['ifwarning'] == correctwarning)]['simTime'].tolist()
    simTime_HMI = HMI_warning_rows[0] if len(HMI_warning_rows) > 0 else None
    rosbag_warning_rows = ego_df[(ego_df['event_Type'].notna()) & ((ego_df['event_Type'] != np.nan))][
        'simTime'].tolist()
    simTime_rosbag = rosbag_warning_rows[0] if len(rosbag_warning_rows) > 0 else None
    if (simTime_HMI is None) or (simTime_rosbag is None):
        print("预警出错！")
        delay_time = 100.0
    else:
        delay_time = abs(simTime_HMI - simTime_rosbag)
    return {"warningDelayTime_LST": delay_time}


def warningDelayTimeofReachDecel_LST(data):
    scenario_name = find_nested_name(data.function_config["function"])
    correctwarning = scenario_sign_dict[scenario_name]
    # 将correctwarning保存到data对象中，供图表生成使用
    data.correctwarning = correctwarning
    ego_df = data.ego_data
    obj_df = data.obj_data
    # 加速度在速度方向上的投影
    speed_mag = np.sqrt(obj_df["speedX"] ** 2 + obj_df["speedY"] ** 2) + 1e-6
    obj_df["ux"] = obj_df["speedX"] / speed_mag
    obj_df["uy"] = obj_df["speedY"] / speed_mag
    obj_df["merged_accel"] = obj_df["accelX"] * obj_df["ux"] + obj_df["accelY"] * obj_df["uy"]
    obj_speed_simtime = obj_df[obj_df['merged_accel'] <= -4]['simTime'].tolist()  # 单位m/s^2
    warning_simTime = ego_df[ego_df['ifwarning'] == correctwarning]['simTime'].tolist()
    if (len(warning_simTime) == 0) and (len(obj_speed_simtime) == 0):
        return {"warningDelayTimeofReachDecel_LST": 0}
    elif (len(warning_simTime) == 0) and (len(obj_speed_simtime) > 0):
        return {"warningDelayTimeofReachDecel_LST": obj_speed_simtime[0]}
    elif (len(warning_simTime) > 0) and (len(obj_speed_simtime) == 0):
        return {"warningDelayTimeofReachDecel_LST": None}
    else:
        return {"warningDelayTimeofReachDecel_LST": warning_simTime[0] - obj_speed_simtime[0]}


def rightWarningSignal_LST(data):
    scenario_name = find_nested_name(data.function_config["function"])
    correctwarning = scenario_sign_dict[scenario_name]
    # 将correctwarning保存到data对象中，供图表生成使用
    data.correctwarning = correctwarning
    ego_df = data.ego_data
    if ego_df['ifwarning'].empty:
        print("无法获取正确预警信号标志位！")
        return
    warning_rows = ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['ifwarning'].notna())]
    if warning_rows.empty:
        return {"rightWarningSignal_LST": -1}
    else:
        return {"rightWarningSignal_LST": 1}


def ifCrossingRedLight_LST(data):
    scenario_name = find_nested_name(data.function_config["function"])
    correctwarning = scenario_sign_dict[scenario_name]
    # 将correctwarning保存到data对象中，供图表生成使用
    data.correctwarning = correctwarning
    ego_df = data.ego_data
    redlight_simtime = ego_df[
        (ego_df['ifwarning'] == correctwarning) & (ego_df['stateMask'] == 1) & (ego_df['relative_dist'] == 0) & (
                ego_df['v'] != 0)]['simTime']
    if redlight_simtime.empty:
        return {"ifCrossingRedLight_LST": -1}
    else:
        return {"ifCrossingRedLight_LST": 1}


def ifStopgreenWaveSpeedGuidance_LST(data):
    scenario_name = find_nested_name(data.function_config["function"])
    correctwarning = scenario_sign_dict[scenario_name]
    # 将correctwarning保存到data对象中，供图表生成使用
    data.correctwarning = correctwarning
    ego_df = data.ego_data
    greenlight_simtime = \
        ego_df[(ego_df['ifwarning'] == correctwarning) & (ego_df['stateMask'] == 0) & (ego_df['v'] == 0)]['simTime']
    if greenlight_simtime.empty:
        return {"ifStopgreenWaveSpeedGuidance_LST": -1}
    else:
        return {"ifStopgreenWaveSpeedGuidance_LST": 1}


# ------ 单车智能指标 ------
def limitSpeed_LST(data):
    ego_df = data.ego_data
    scenario_name = find_nested_name(data.function_config["function"])
    limit_speed = data.function_config["function"][scenario_name]["limitSpeed_LST"]["max"]
    speed_limit = ego_df[abs(ego_df['x_relative_dist']) <= 0.1]['v'].tolist()
    if len(speed_limit) == 0:
        return {"speedLimit_LST": -1}
    max_speed = max(speed_limit)
    data.speedLimit = limit_speed
    generate_function_chart_data(data, 'limitspeed_LST')
    return {"speedLimit_LST": max_speed}


def limitSpeedPastLimitSign_LST(data):
    ego_df = data.ego_data
    scenario_name = find_nested_name(data.function_config["function"])
    limit_speed = data.function_config["function"][scenario_name]["limitSpeed_LST"]["max"]
    car_length = data.function_config["vehicle"]['CAR_LENGTH']
    ego_speed = ego_df[ego_df['x_relative_dist'] <= -100 - car_length]['v'].tolist()
    ego_time = ego_df[ego_df['x_relative_dist'] <= -100 - car_length]['simTime'].tolist()
    data.speedLimit = limit_speed
    data.speedPastLimitSign_LST = ego_time[0] if len(ego_time) > 0 else None
    generate_function_chart_data(data, 'limitSpeedPastLimitSign_LST')
    if len(ego_speed) == 0:
        return {"speedPastLimitSign_LST": -1}
    return {"speedPastLimitSign_LST": ego_speed[0]}


def leastDistance_LST(data):
    ego_df = data.ego_data
    dist_row = ego_df[ego_df['v'] == 0]['relative_dist'].tolist()
    if len(dist_row) == 0:
        return {"leastDistance_LST": -1}
    else:
        min_dist = min(dist_row)
        return {"leastDistance_LST": min_dist}


def launchTimeinStopLine_LST(data):
    ego_df = data.ego_data
    simtime_row = ego_df[ego_df['v'] == 0]['simTime'].tolist()
    if len(simtime_row) == 0:
        return {"launchTimeinStopLine_LST": -1}
    else:
        delta_t = simtime_row[-1] - simtime_row[0]
        return {"launchTimeinStopLine_LST": delta_t}


def launchTimewhenFollowingCar_LST(data):
    ego_df = data.ego_data
    t_interval = ego_df['simTime'].tolist()[1] - ego_df['simTime'].tolist()[0]
    simtime_row = ego_df[ego_df['v'] == 0]['simTime'].tolist()
    if len(simtime_row) == 0:
        return {"launchTimewhenFollowingCar_LST": 0}
    else:
        time_interval = _is_segment_by_interval(simtime_row, t_interval)
        delta_t = []
        for t in time_interval:
            delta_t.append(t[-1] - t[0])
        return {"launchTimewhenFollowingCar_LST": max(delta_t)}


def noStop_LST(data):
    ego_df_ini = data.ego_data
    min_time = ego_df_ini['simTime'].min() + 5
    max_time = ego_df_ini['simTime'].max() - 5
    ego_df = ego_df_ini[(ego_df_ini['simTime'] >= min_time) & (ego_df_ini['simTime'] <= max_time)]
    speed = ego_df['v'].tolist()
    if (any(speed) == 0):
        return {"noStop_LST": -1}
    else:
        return {"noStop_LST": 1}


def launchTimeinTrafficLight_LST(data):
    '''
    待修改：
    红灯的状态值：1
    绿灯的状态值：0
    '''
    ego_df = data.ego_data
    simtime_of_redlight = ego_df[ego_df['stateMask'] == 0]['simTime']
    simtime_of_stop = ego_df[ego_df['v'] == 0]['simTime']
    if len(simtime_of_stop) or len(simtime_of_redlight):
        return {"timeInterval_LST": -1}
    simtime_of_launch = simtime_of_redlight[simtime_of_redlight.isin(simtime_of_stop)].tolist()
    if len(simtime_of_launch) == 0:
        return {"timeInterval_LST": -1}
    return {"timeInterval_LST": simtime_of_launch[-1] - simtime_of_launch[0]}


def crossJunctionToTargetLane_LST(data):
    ego_df = data.ego_data
    lane_in_leftturn = set(ego_df['lane_id'].tolist())
    scenario_name = find_nested_name(data.function_config["function"])
    target_lane_id = data.function_config["function"][scenario_name]["crossJunctionToTargetLane_LST"]['max']
    if target_lane_id not in lane_in_leftturn:
        return {"crossJunctionToTargetLane_LST": -1}
    else:
        return {"crossJunctionToTargetLane_LST": target_lane_id}


def keepInLane_LST(data):
    ego_df = data.ego_data
    notkeepinlane = ego_df[ego_df['laneOffset'] > ego_df['lane_width']/2].tolist()
    if len(notkeepinlane):
        return {"keepInLane_LST": -1}
    return {"keepInLane_LST": 1}


def leastLateralDistance_LST(data):
    ego_df = data.ego_data
    lane_width = ego_df[ego_df['x_relative_dist'] == 0]['lane_width']
    if lane_width.empty():
        return {"leastLateralDistance_LST": -1}
    else:
        y_relative_dist = ego_df[ego_df['x_relative_dist'] == 0]['y_relative_dist']
        if (y_relative_dist <= lane_width / 2).all():
            return {"leastLateralDistance_LST": 1}
        else:
            return {"leastLateralDistance_LST": -1}


def waitTimeAtCrosswalkwithPedestrian_LST(data):
    ego_df = data.ego_data
    object_df = data.object_data
    data['in_crosswalk'] = []
    position_data = data.drop_duplicates(subset=['cross_id', 'cross_coords'], inplace=True)
    for cross_id in position_data['cross_id'].tolist():
        for posX, posY in object_df['posX', 'posY']:
            pedestrian_pos = (posX, posY)
            plogan_pos = position_data[position_data['cross_id'] == cross_id]['cross_coords'].tolist()[0]
            if _is_pedestrian_in_crosswalk(plogan_pos, pedestrian_pos):
                data[data['playerId'] == 2]['in_crosswalk'] = 1
            else:
                data[data['playerId'] == 2]['in_crosswalk'] = 0
    car_stop_time = ego_df[ego_df['v'] == 0]['simTime']
    pedestrian_in_crosswalk_time = data[(data['in_crosswalk'] == 1)]['simTime']
    car_wait_pedestrian = car_stop_time.isin(pedestrian_in_crosswalk_time).tolist()
    return {'waitTimeAtCrosswalkwithPedestrian_LST': car_wait_pedestrian[-1] - car_wait_pedestrian[0] if len(
        car_wait_pedestrian) > 0 else 0}


def launchTimewhenPedestrianLeave_LST(data):
    ego_df = data.ego_data
    car_stop_time = ego_df[ego_df['v'] == 0]["simTime"]
    if car_stop_time.empty():
        return {"launchTimewhenPedestrianLeave_LST": -1}
    else:
        lane_width = ego_df[ego_df['v'] == 0]['lane_width'].tolist()[0]
        car_to_pedestrain = ego_df[ego_df['y_relative_dist'] <= lane_width / 2]["simTime"]
        legal_stop_time = car_stop_time.isin(car_to_pedestrain).tolist()
        return {"launchTimewhenPedestrianLeave_LST": legal_stop_time[-1] - legal_stop_time[0]}


def noCollision_LST(data):
    ego_df = data.ego_data
    if ego_df['relative_dist'].any() == 0:
        return {"noCollision_LST": -1}
    else:
        return {"noCollision_LST": 1}


def noReverse_LST(data):
    ego_df = data.ego_data
    if (ego_df["lon_v_vehicle"] * ego_df["posH"]).any() < 0:
        return {"noReverse_LST": -1}
    else:
        return {"noReverse_LST": 1}


def turnAround_LST(data):
    ego_df = data.ego_data
    if (ego_df["lon_v_vehicle"].tolist()[0] * ego_df["lon_v_vehicle"].tolist()[-1] < 0) and (
            ego_df["lon_v_vehicle"] * ego_df["posH"].all() > 0):
        return {"turnAround_LST": 1}
    else:
        return {"turnAround_LST": -1}


def laneOffset_LST(data):
    car_width = data.function_config['vehicle']['CAR_WIDTH']
    ego_df_ini = data.ego_data
    min_time = ego_df_ini['simTime'].min() + 5
    max_time = ego_df_ini['simTime'].max() - 5
    ego_df = ego_df_ini[(ego_df_ini['simTime'] >= min_time) & (ego_df_ini['simTime'] <= max_time)]
    laneoffset = ego_df['y_relative_dist'] - car_width / 2
    return {"laneOffset_LST": max(laneoffset)}


def maxLongitudeDist_LST(data):
    ego_df = data.ego_data
    longitude_dist = abs(ego_df[ego_df['v'] == 0]['x_relative_dist'].tolist())
    data.longitude_dist = min(abs(ego_df[ego_df['v'] == 0]['x_relative_dist'].tolist()))
    stop_time = ego_df[abs(ego_df['x_relative_dist']) == min(longitude_dist)]['simTime'].tolist()
    data.stop_time = min(stop_time)
    if len(longitude_dist) == 0:
        return {"maxLongitudeDist_LST": -1}
    generate_function_chart_data(data, 'maxLongitudeDist_LST')
    return {"maxLongDist_LST": min(longitude_dist)}


def noEmergencyBraking_LST(data):
    ego_df = data.ego_data
    ego_df['ip_dec'] = ego_df['v'].apply(
        get_interpolation, point1=[18, -5], point2=[72, -3.5])
    ego_df['slam_brake'] = (ego_df['accleX'] - ego_df['ip_dec']).apply(
        lambda x: 1 if x < 0 else 0)
    if sum(ego_df['slam_brake']) == 0:
        return {"noEmergencyBraking_LST": 1}
    else:
        return {"noEmergencyBraking_LST": -1}


def rightWarningSignal_PGVIL(data_processed) -> dict:
    """判断是否发出正确预警信号"""

    ego_df = data_processed.ego_data
    scenario_name = find_nested_name(data_processed.function_config["function"])
    correctwarning = scenario_sign_dict[scenario_name]

    if correctwarning is None:
        print("无法获取正确的预警信号标志位！")
        return None
    # 找出本行 correctwarning 和 ifwarning 相等，且 correctwarning 不是 NaN 的行
    warning_rows = ego_df[
        (ego_df["ifwarning"] == correctwarning) & (ego_df["ifwarning"].notna())
        ]

    if warning_rows.empty:
        return {"rightWarningSignal_PGVIL": -1}
    else:
        return {"rightWarningSignal_PGVIL": 1}


def latestWarningDistance_PGVIL(data_processed) -> dict:
    """预警距离计算流水线"""
    ego_df = data_processed.ego_data
    obj_df = data_processed.object_df
    warning_data = get_first_warning(data_processed)
    if warning_data is None:
        return {"latestWarningDistance_PGVIL": 0.0}

    ego, obj = extract_ego_obj(warning_data)
    distances = calculate_distance_PGVIL(
        np.array([[ego["posX"], ego["posY"]]]), obj[["posX", "posY"]].values
    )

    # 将计算结果保存到data对象中，供图表生成使用
    data_processed.warning_dist = distances
    data_processed.warning_time = ego['simTime'].tolist()
    if distances.size == 0:
        print("没有找到数据！")
        return {"latestWarningDistance_PGVIL": 15}  # 或返回其他默认值，如0.0
    generate_function_chart_data(data_processed, 'latestWarningDistance_PGVIL')
    return {"latestWarningDistance_PGVIL": float(np.min(distances))}


def latestWarningDistance_TTC_PGVIL(data_processed) -> dict:
    """TTC计算流水线"""
    ego_df = data_processed.ego_data
    obj_df = data_processed.object_df

    warning_data = get_first_warning(data_processed)
    if warning_data is None:
        return {"latestWarningDistance_TTC_PGVIL": 0.0}

    ego, obj = extract_ego_obj(warning_data)
    # 向量化计算
    ego_pos = np.array([[ego["posX"], ego["posY"]]])
    ego_speed = np.array([[ego["speedX"], ego["speedY"]]])
    obj_pos = obj[["posX", "posY"]].values
    obj_speed = obj[["speedX", "speedY"]].values

    distances = calculate_distance_PGVIL(ego_pos, obj_pos)
    rel_speeds = calculate_relative_speed_PGVIL(ego_speed, obj_speed)

    data_processed.warning_dist = distances
    data_processed.warning_speed = rel_speeds
    data_processed.warning_time = ego['simTime'].tolist()

    with np.errstate(divide="ignore", invalid="ignore"):
        ttc = np.where(rel_speeds != 0, distances / rel_speeds, np.inf)
    if ttc.size == 0:
        print("没有找到数据！")
        return {"latestWarningDistance_TTC_PGVIL": 2}  # 或返回其他默认值，如0.0
    data_processed.ttc = ttc
    generate_function_chart_data(data_processed, 'latestWarningDistance_TTC_PGVIL')
    return {"latestWarningDistance_TTC_PGVIL": float(np.nanmin(ttc))}


def earliestWarningDistance_PGVIL(data_processed) -> dict:
    """预警距离计算流水线"""
    ego_df = data_processed.ego_data
    obj_df = data_processed.object_df

    warning_data = get_first_warning(data_processed)
    if warning_data is None:
        return {"earliestWarningDistance_PGVIL": 0}

    ego, obj = extract_ego_obj(warning_data)
    distances = calculate_distance_PGVIL(
        np.array([[ego["posX"], ego["posY"]]]), obj[["posX", "posY"]].values
    )
    # 将计算结果保存到data对象中，供图表生成使用
    data_processed.warning_dist = distances
    data_processed.warning_time = ego['simTime'].tolist()
    if distances.size == 0:
        print("没有找到数据！")
        return {"earliestWarningDistance_PGVIL": 15}  # 或返回其他默认值，如0.0
    generate_function_chart_data(data_processed, 'earliestWarningDistance_PGVIL')
    return {"earliestWarningDistance": float(np.min(distances))}


def earliestWarningDistance_TTC_PGVIL(data_processed) -> dict:
    """TTC计算流水线"""
    ego_df = data_processed.ego_data
    obj_df = data_processed.object_df

    warning_data = get_first_warning(data_processed)
    if warning_data is None:
        return {"earliestWarningDistance_TTC_PGVIL": 0.0}

    ego, obj = extract_ego_obj(warning_data)

    # 向量化计算
    ego_pos = np.array([[ego["posX"], ego["posY"]]])
    ego_speed = np.array([[ego["speedX"], ego["speedY"]]])
    obj_pos = obj[["posX", "posY"]].values
    obj_speed = obj[["speedX", "speedY"]].values

    distances = calculate_distance_PGVIL(ego_pos, obj_pos)
    rel_speeds = calculate_relative_speed_PGVIL(ego_speed, obj_speed)

    data_processed.warning_dist = distances
    data_processed.warning_speed = rel_speeds
    data_processed.warning_time = ego['simTime'].tolist()

    with np.errstate(divide="ignore", invalid="ignore"):
        ttc = np.where(rel_speeds != 0, distances / rel_speeds, np.inf)
    if ttc.size == 0:
        print("没有找到数据！")
        return {"earliestWarningDistance_TTC_PGVIL": 2}  # 或返回其他默认值，如0.0
    data_processed.ttc = ttc
    generate_function_chart_data(data_processed, 'earliestWarningDistance_TTC_PGVIL')
    return {"earliestWarningDistance_TTC_PGVIL": float(np.nanmin(ttc))}


# def delayOfEmergencyBrakeWarning(data_processed) -> dict:
#     #预警时机相对背景车辆减速度达到-4m/s2后的时延
#     ego_df = data_processed.ego_data
#     obj_df = data_processed.object_df
#     warning_data = get_first_warning(data_processed)
#     if warning_data is None:
#         return {"delayOfEmergencyBrakeWarning": -1}
#     try:
#         ego, obj = extract_ego_obj(warning_data)
#         # 向量化计算
#         obj_speed = np.array([[obj_df["speedX"], obj_df["speedY"]]])
#         # 计算背景车辆减速度
#         simtime_gap = obj["simTime"].iloc[1] - obj["simTime"].iloc[0]
#         simtime_freq = 1 / simtime_gap#每秒采样频率
#         # simtime_freq为一个时间窗，找出时间窗内的最大减速度
#         obj_speed_magnitude = np.linalg.norm(obj_speed, axis=1)#速度向量的模长
#         obj_speed_change = np.diff(speed_magnitude)#速度模长的变化量
#         obj_deceleration = np.diff(obj_speed_magnitude) / simtime_gap
#         #找到最大减速度，若最大减速度小于-4m/s2，则计算最大减速度对应的时间，和warning_data的差值进行对比
#         max_deceleration = np.max(obj_deceleration)
#         if max_deceleration < -4:
#             max_deceleration_times = obj["simTime"].iloc[np.argmax(obj_deceleration)]
#             max_deceleration_time = max_deceleration_times.iloc[0]
#             delay_time = ego["simTime"] - max_deceleration_time
#             return {"delayOfEmergencyBrakeWarning": float(delay_time)}

#         else:
#             print("没有达到预警减速度阈值:-4m/s^2")
#             return {"delayOfEmergencyBrakeWarning": -1}


def warningDelayTime_PGVIL(data_processed) -> dict:
    """车端接收到预警到HMI发出预警的时延"""
    ego_df = data_processed.ego_data
    # #打印ego_df的列名
    # print(ego_df.columns.tolist())

    warning_data = get_first_warning(data_processed)

    if warning_data is None:
        return {"warningDelayTime_PGVIL": -1}
    try:
        ego, obj = extract_ego_obj(warning_data)

        # 找到event_Type不为空，且playerID为1的行
        rosbag_warning_rows = ego_df[(ego_df["event_Type"].notna())]

        first_time = rosbag_warning_rows["simTime"].iloc[0]
        warning_time = warning_data[warning_data["playerId"] == 1]["simTime"].iloc[0]
        delay_time = warning_time - first_time

        return {"warningDelayTime_PGVIL": float(delay_time)}

    except Exception as e:
        print(f"计算预警时延时发生错误: {e}")
        return {"warningDelayTime_PGVIL": -1}


def get_car_to_stop_line_distance(ego, car_point, stop_line_points):
    """
    计算主车后轴中心点到停止线的距离
    :return 距离
    """
    distance_carpoint_carhead = ego["dimX"] / 2 + ego["offX"]
    # 计算停止线的方向向量
    line_vector = np.array(
        [
            stop_line_points[1][0] - stop_line_points[0][0],
            stop_line_points[1][1] - stop_line_points[0][1],
        ]
    )
    direction_vector_norm = np.linalg.norm(line_vector)
    direction_vector_unit = (
        line_vector / direction_vector_norm
        if direction_vector_norm != 0
        else np.array([0, 0])
    )
    # 计算主车后轴中心点到停止线投影的坐标（垂足）
    projection_length = np.dot(car_point - stop_line_points[0], direction_vector_unit)
    perpendicular_foot = stop_line_points[0] + projection_length * direction_vector_unit

    # 计算主车后轴中心点到垂足的距离
    distance_to_foot = np.linalg.norm(car_point - perpendicular_foot)
    carhead_distance_to_foot = distance_to_foot - distance_carpoint_carhead

    return carhead_distance_to_foot


def ifCrossingRedLight_PGVIL(data_processed) -> dict:
    # 判断车辆是否闯红灯

    stop_line_points = np.array([(276.555, -35.575), (279.751, -33.683)])
    X_OFFSET = 258109.4239876
    Y_OFFSET = 4149969.964821
    stop_line_points += np.array([[X_OFFSET, Y_OFFSET]])
    ego_df = data_processed.ego_data

    prev_distance = float("inf")  # 初始化为正无穷大
    """
    traffic_light_status
    0x100000为绿灯，1048576
    0x1000000为黄灯，16777216
    0x10000000为红灯，268435456
    """
    red_light_violation = False
    for index, ego in ego_df.iterrows():
        car_point = (ego["posX"], ego["posY"])
        stateMask = ego["stateMask"]
        simTime = ego["simTime"]
        distance_to_stopline = get_car_to_stop_line_distance(
            ego, car_point, stop_line_points
        )

        # 主车车头跨越停止线时非绿灯，返回-1，闯红灯
        if prev_distance > 0 and distance_to_stopline < 0:
            if stateMask is not None and stateMask != 1048576:
                red_light_violation = True
            break
        prev_distance = distance_to_stopline

    if red_light_violation:
        return {"ifCrossingRedLight_PGVIL": -1}  # 闯红灯
    else:
        return {"ifCrossingRedLight_PGVIL": 1}  # 没有闯红灯


# def ifStopgreenWaveSpeedGuidance(data_processed) -> dict:
#     #在绿波车速引导期间是否发生停车


# def mindisStopline(data_processed) -> dict:
#     """
#     当有停车让行标志/标线时车辆最前端与停车让行线的最小距离应在0-4m之间
#     """
#     ego_df = data_processed.ego_data
#     obj_df = data_processed.object_df
#     stop_giveway_simtime = ego_df[
#         ego_df["sign_type1"] == 32 |
#         ego_df["stopline_type"] == 3
#     ]["simTime"]
#     stop_giveway_data = ego_df[
#         ego_df["sign_type1"] == 32 |
#         ego_df["stopline_type"] == 3
#     ]["simTime"]
#     if stop_giveway_simtime.empty:
#         print("没有停车让行标志/标线")

#     ego_data = stop_giveway_data[stop_giveway_data['playerId'] == 1]
#     distance_carpoint_carhead = ego_data['dimX'].iloc[0]/2 + ego_data['offX'].iloc[0]
#     distance_to_stoplines = []
#     for _,row in ego_data.iterrows():
#         ego_pos = np.array([row["posX"], row["posY"], row["posH"]])
#         stop_line_points = [
#             [row["stopline_x1"], row["stopline_y1"]],
#             [row["stopline_x2"], row["stopline_y2"]],
#         ]
#         distance_to_stopline = get_car_to_stop_line_distance(ego_pos, stop_line_points)
#         distance_to_stoplines.append(distance_to_stopline)

#     mindisStopline = np.min(distance_to_stoplines) - distance_carpoint_carhead
#     return {"mindisStopline": mindisStopline}

def limitSpeed_PGVIL(data):
    ego_df = data.ego_data
    max_speed = max(ego_df["v"])
    if len(ego_df["v"]) == 0:
        return {"speedLimit_PGVIL": -1}
    else:
        return {"speedLimit_PGVIL": max_speed}


def leastDistance_PGVIL(data):
    exclude_seconds = 2.0
    ego_df = data.ego_data
    max_sim_time = ego_df["simTime"].max()
    threshold_time = max_sim_time - exclude_seconds
    for index, ego in ego_df.iterrows():
        if ego["simTime"] >= threshold_time:
            continue

        if "v" in ego and ego["v"] == 0:
            car_point = (ego["posX"], ego["posY"])

            p1 = (ego["stopline_x1"], ego["stopline_y1"])
            p2 = (ego["stopline_x2"], ego["stopline_y2"])
            stop_line_points = np.array([p1, p2], dtype=float)

            distance_to_stopline = get_car_to_stop_line_distance(
                ego, car_point, stop_line_points
            )

            return {"leastDistance_PGVIL": distance_to_stopline}

    return {"leastDistance_PGVIL": -1}


def launchTimeinStopLine_PGVIL(data):
    ego_df = data.ego_data
    in_stop = False
    start_time = None
    last_low_time = None
    max_duration = 0.0
    v_thresh = 0.01
    for _, row in ego_df.iterrows():
        simt = row["simTime"]
        v = row.get("v", None)
        if v is None or np.isnan(v) or abs(v) > v_thresh:
            if in_stop:
                duration = last_low_time - start_time
                if duration > max_duration:
                    max_duration = duration
                in_stop = False
        else:
            if not in_stop:
                start_time = simt
                in_stop = True
            # 更新最近一次“停车时刻”的 simTime
            last_low_time = simt
    if in_stop:
        duration = last_low_time - start_time
        if duration > max_duration:
            max_duration = duration

    if max_duration <= 0:
        return {"launchTimeinStopLine_PGVIL": -1}
    else:
        return {"launchTimeinStopLine_PGVIL": float(max_duration)}


def launchTimeinTrafficLight_PGVIL(data):
    GREEN = 0x100000
    RED = 0x10000000
    ego_df = data.ego_data
    # 找到第一次红灯 to 绿灯的切换时刻
    is_transition = (ego_df["stateMask"] == GREEN) & (
        ego_df["stateMask"].shift(1) == RED
    )
    transition_times = ego_df.loc[is_transition, "simTime"]
    if transition_times.empty:
        return {"timeInterval_PGVIL": -1}
    time_red2green = transition_times.iloc[0]

    car_move_times = ego_df.loc[
        (ego_df["simTime"] >= time_red2green) & (ego_df["v"] > 0), "simTime"
    ]
    if car_move_times.empty:
        return {"timeInterval_PGVIL": -1}

    time_move = move_times.iloc[0]
    return {"timeInterval_PGVIL": time_move - time_red2green}


def crossJunctionToTargetLane_PGVIL(data):
    ego_df = data.ego_data
    lane_ids = set(ego_df["lane_id"].dropna())

    try:
        scenario_name = find_nested_name(data.function_config["function"])
        target_lane_id = data.function_config["function"][scenario_name][
            "crossJunctionToTargetLane_PGVIL"
        ]["max"]
    except KeyError:
        raise ValueError("请在配置文件中指定目标车道ID！")

    result = target_lane_id if target_lane_id in lane_ids else -1
    return {"crossJunctionToTargetLane_PGVIL": result}


def noStop_PGVIL(data):
    exclude_end_seconds = 5.0
    exclude_start_seconds = 5.0
    ego_df = data.ego_data
    max_sim_time = ego_df["simTime"].max()

    start_threshold = min_sim_time + exclude_start_seconds
    end_threshold = max_sim_time - exclude_end_seconds
    filtered_df = ego_df[
        (ego_df["simTime"] >= start_threshold) & (ego_df["simTime"] <= end_threshold)
    ]

    if (filtered_df["v"] == 0).any():
        return {"noStop_PGVIL": -1}
    else:
        return {"noStop_PGVIL": 1}


def noEmergencyBraking_PGVIL(data):
    ego_df = data.ego_data
    ego_df["ip_dec"] = ego_df["v"].apply(
        get_interpolation, point1=[18, -5], point2=[72, -3.5]
    )
    ego_df["slam_brake"] = (ego_df["accleX"] - ego_df["ip_dec"]).apply(
        lambda x: 1 if x < 0 else 0
    )
    if sum(ego_df["slam_brake"]) == 0:
        return {"noEmergencyBraking_PGVIL": 1}
    else:
        return {"noEmergencyBraking_PGVIL": -1}


def noReverse_PGVIL(data):
    ego_df = data.ego_data.copy()
    heading_x = np.cos(ego_df["posH"])
    reverse_flag = (ego_df["speedX"] * heading_x) < 0

    if reverse_flag.any():
        return {"noReverse_PGVIL": -1}
    else:
        return {"noReverse_PGVIL": 1}


def laneOffset_PGVIL(data):
    car_width = data.function_config["vehicle"]["CAR_WIDTH"]
    ego_df = data.ego_data
    is_zero = ego_df["v"] == 0
    if not is_zero.any():
        # 全程未停车
        return {"laneOffset_PGVIL": None}
    groups = (is_zero != is_zero.shift(fill_value=False)).cumsum()
    last_zero_group = groups[is_zero].iloc[-1]
    last_stop = ego_df[groups == last_zero_group]

    # 距离右侧车道线
    edge_dist = (last_stop["lane_width"] / 2 + last_stop["laneOffset"]) - (
        car_width / 2
    )
    return {"laneOffset_PGVIL": edge_dist.max()}


def maxLongitudelDistance_PGVIL(data):
    scenario_name = find_nested_name(data.function_config["function"])
    stopX_pos = data.function_config["function"][scenario_name][
        "maxLongitudelDistance_PGVIL"
    ]["max"]
    stopY_pos = data.function_config["function"][scenario_name][
        "maxLongitudelDistance_PGVIL"
    ]["min"]


def keepInLane_PGVIL(data):
    ego_df = data.ego_data.copy()
    ego_df = ego_df.dropna(subset=["laneOffset", "lane_width"])
    if ego_df.empty:
        return {"keepInLane_PGVIL": -1}
    threshold = ego_df["lane_width"] / 2
    if (ego_df["laneOffset"].abs() > threshold).any():
        return {"keepInLane_PGVIL": -1}
    else:
        return {"keepInLane_PGVIL": 1}


def launchTimewhenPedestrianLeave_PGVIL(data):
    ego_df = data.ego_data
    ped_df = data.object_data.loc[
        data.object_data["playerId"] == 2, ["simTime", "posX", "posY"]
    ]
    merged = ego_df.merge(ped_df, on="simTime", how="inner", suffixes=("", "_ped"))
    if merged.empty:
        return {"launchTimewhenPedestrianLeave_PGVIL": -1}

    yaw_rad = np.deg2rad(merged["posH"])  # 度→弧度
    dx = merged["posX_ped"] - merged["posX"]
    dy = merged["posY_ped"] - merged["posY"]
    merged["lateral_dist"] = np.abs(-np.sin(yaw_rad) * dx + np.cos(yaw_rad) * dy)
    merged["threshold"] = merged["lane_width"] / 2

    entry = merged[merged["lateral_dist"] <= merged["threshold"]]
    if entry.empty:
        return {"launchTimewhenPedestrianLeave_PGVIL": -1}
    t_entry = entry["simTime"].iloc[0]

    after_e = merged[merged["simTime"] > t_entry]
    leave = after_e[after_e["lateral_dist"] > after_e["threshold"]]
    if leave.empty:
        return {"launchTimewhenPedestrianLeave_PGVIL": -1}
    t_leave = leave["simTime"].iloc[0]
    ped_x, ped_y = leave[["posX_ped", "posY_ped"]].iloc[0]

    stops = ego_df[(ego_df["simTime"] > t_leave) & (ego_df["v"] == 0)].copy()
    if stops.empty:
        return {"launchTimewhenPedestrianLeave_PGVIL": -1}

    front_overPos = ego_df["dimX"] / 2 + ego_df["offX"]
    yaw_stop = np.deg2rad(stops["posH"])

    stops["front_x"] = stops["posX"] + np.cos(yaw_stop) * front_overPos
    stops["front_y"] = stops["posY"] + np.sin(yaw_stop) * front_overPos
    dx_s = stops["front_x"] - ped_x
    dy_s = stops["front_y"] - ped_y
    proj = dx_s * np.cos(yaw_stop) + dy_s * np.sin(yaw_stop)
    valid = stops[proj <= 0]
    t_stop = valid["simTime"].iloc[0]
    launches = ego_df[(ego_df["simTime"] > t_stop) & (ego_df["v"] > 0)]["simTime"]
    t_launch = launches.iloc[0]
    return {"launchTimewhenPedestrianLeave_PGVIL": t_launch - t_stop}


def waitTimeAtCrosswalkwithPedestrian_PGVIL(data):
    ego_df = data.ego_data
    ped_df = data.object_data.loc[
        data.object_data["playerId"] == 2, ["simTime", "posX", "posY"]
    ]

    first_launch = ego_df.loc[ego_df["v"] > 0, "simTime"]
    if first_launch.empty:
        t0_launch = 0.0
    else:
        t0_launch = first_launch.iloc[0]

    merged = ego_df.merge(ped_df, on="simTime", how="inner", suffixes=("", "_ped"))
    merged["front_offset"] = merged["dimX"] / 2 + merged["offX"]
    yaw = np.deg2rad(merged["posH"])

    merged["front_x"] = merged["posX"] + np.cos(yaw) * merged["front_offset"]
    merged["front_y"] = merged["posY"] + np.sin(yaw) * merged["front_offset"]
    dx = merged["posX_ped"] - merged["front_x"]
    dy = merged["posY_ped"] - merged["front_y"]
    valid_times = (dx * np.cos(yaw) + dy * np.sin(yaw)) <= 0

    stops_df = merged.loc[
        (merged["simTime"] >= t0_launch) & (merged["v"] == 0) & valid_times
    ].sort_values("simTime")
    if stops_df.empty:
        return {"waitTimeAtCrosswalkwithPedestrian_PGVIL": -1}
    wait_time = stops_df["simTime"].iloc[-1] - stops_df["simTime"].iloc[0]
    return {"waitTimeAtCrosswalkwithPedestrian_PGVIL": wait_time}


def noCollision_PGVIL(data):
    ego = data.ego_data[["simTime", "posX", "posY", "posH", "dimX", "dimY", "offX", "offY"]]
    tar = data.object_data.loc[data.object_data["playerId"] == 2,
        ["simTime", "posX", "posY", "posH", "dimX", "dimY", "offX", "offY"]]
    df = ego.merge(tar, on="simTime", suffixes=("", "_tar"))

    df["posH_tar_rad"] = np.deg2rad(df["posH_tar"])
    df["posH_rad"] = np.deg2rad(df["posH"])
    df["collision"] = df.apply(lambda row: funcIsCollision(
            row["dimX"],row["dimY"],row["offX"],row["offY"],row["posX"],row["posY"],
            row["posH_rad"],row["dimX_tar"],row["dimY_tar"],row["offX_tar"],row["offY_tar"],
            row["posX_tar"],row["posY_tar"],row["posH_tar_rad"],),axis=1,)

    if df["collision"].any():
        return {"noCollision_PGVIL": -1}
    else:
        return {"noCollision_PGVIL": 1}


def leastLateralDistance_PGVIL(data):
    ego_df = data.ego_data
    cones = data.object_data.loc[data.object_data['playerId'] != 1,
        ['simTime','playerId','posX','posY']].copy()
    df = ego.merge(cones, on='simTime', how='inner', suffixes=('','_cone'))
    yaw = np.deg2rad(df['posH'])
    x_c = df['posX'] + df['offX'] * np.cos(yaw)
    y_c = df['posY'] + df['offX'] * np.sin(yaw)
    dx = df['posX']   - x_c
    dy = df['posY']   - y_c
    dx = df['posX_cone'] - x_c
    dy = df['posY_cone'] - y_c
    local_x =  np.cos(yaw) * dx + np.sin(yaw) * dy
    local_y = -np.sin(yaw) * dx + np.cos(yaw) * dy
    half_length = df['dimX'] / 2
    half_width  = df['dimY'] / 2

    inside = (np.abs(local_x) <= half_length) & (np.abs(local_y) <= half_width)
    collisions = df.loc[inside, ['simTime','playerId']]

    if collisions.empty:
        return {"noConeRectCollision_PGVIL": 1}
    else:
        collision_times = collisions['simTime'].unique().tolist()
        collision_ids   = collisions['playerId'].unique().tolist()
        return {"noConeRectCollision_PGVIL": -1}


class FunctionRegistry:
    """动态函数注册器（支持参数验证）"""

    def __init__(self, data_processed):
        self.logger = LogManager().get_logger()  # 获取全局日志实例
        self.data = data_processed
        # 检查function_config是否为空
        if not hasattr(data_processed, 'function_config') or not data_processed.function_config:
            self.logger.warning("功能配置为空，跳过功能指标计算")
            self.fun_config = {}
            self.level_3_merics = []
            self._registry = {}
            return
        self.fun_config = data_processed.function_config.get("function", {})
        self.level_3_merics = self._extract_level_3_metrics(self.fun_config)
        self._registry: Dict[str, Callable] = {}
        self._registry = self._build_registry()

    def _extract_level_3_metrics(self, config_node: dict) -> list:
        """DFS遍历提取第三层指标（时间复杂度O(n)）[4](@ref)"""
        metrics = []

        def _recurse(node):
            if isinstance(node, dict):
                if "name" in node and not any(
                        isinstance(v, dict) for v in node.values()
                ):
                    metrics.append(node["name"])
                for v in node.values():
                    _recurse(v)

        _recurse(config_node)
        self.logger.info(f"评比的功能指标列表:{metrics}")
        return metrics

    def _build_registry(self) -> dict:
        """自动注册指标函数（防御性编程）"""
        registry = {}
        for func_name in self.level_3_merics:
            try:
                registry[func_name] = globals()[func_name]
            except KeyError:
                print(f"未实现指标函数: {func_name}")
                self.logger.error(f"未实现指标函数: {func_name}")
        return registry

    def batch_execute(self) -> dict:
        """批量执行指标计算（带熔断机制）"""
        results = {}
        # 如果配置为空或没有注册的指标，直接返回空结果
        if not hasattr(self, 'fun_config') or not self.fun_config or not self._registry:
            self.logger.info("功能配置为空或无注册指标，返回空结果")
            return results
            
        for name, func in self._registry.items():
            try:
                result = func(self.data)  # 统一传递数据上下文
                results.update(result)
            except Exception as e:
                print(f"{name} 执行失败: {str(e)}")
                self.logger.error(f"{name} 执行失败: {str(e)}", exc_info=True)
                results[name] = None
        self.logger.info(f"功能指标计算结果:{results}")
        return results


class FunctionManager:
    """管理功能指标计算的类"""

    def __init__(self, data_processed):
        self.data = data_processed
        self.logger = LogManager().get_logger()
        # 检查function_config是否为空
        if not hasattr(data_processed, 'function_config') or not data_processed.function_config:
            self.logger.warning("功能配置为空，跳过功能指标计算初始化")
            self.function = None
        else:
            self.function = FunctionRegistry(self.data)

    def report_statistic(self):
        """
        计算并报告功能指标结果。
        :return: 评估结果
        """
        # 如果function为None，直接返回空字典
        if self.function is None:
            self.logger.info("功能指标管理器未初始化，返回空结果")
            return {}
            
        function_result = self.function.batch_execute()

        print("\n[功能性表现及评价结果]")
        return function_result
        # self.logger.info(f'Function Result: {function_result}')


# 使用示例
if __name__ == "__main__":
    pass
    # print("\n[功能类表现及得分情况]")