Python实战：从HighD数据集中精准提取超车变道及邻近车辆轨迹

1. 项目缘起从“一堆数据”到“一个场景”如果你正在做自动驾驶或者高级驾驶辅助系统ADAS的研究尤其是跟车、变道、超车这些决策相关的算法那你大概率听说过HighD数据集。这个由德国亚琛工业大学汽车工程研究所ika发布的自然驾驶数据集在业内名气不小。它包含了德国高速公路上超过16.5万辆车、总里程超过4.5万公里的轨迹数据精度高场景丰富是研究车辆微观行为的宝贵资源。但问题来了当你兴冲冲地下载了那几十个G的CSV文件打开一看可能会瞬间懵掉。数据字段密密麻麻从车辆ID、帧号、到长宽高、速度、加速度、车道ID再到前车后车ID一应俱全。数据是有了可怎么用呢比如你想研究“超车”这个行为。在HighD里并没有一个现成的标签叫“is_overtaking”。你需要自己从海量的轨迹点中把那些发生了超车行为的片段“挖”出来。更进一步研究超车不能只看主角车自己。超车是一个与周围环境尤其是与被超车辆邻近车辆强交互的过程。你需要准确地筛选出在超车事件发生前后与主车相关的邻近车辆数据。这个“筛选”的过程就是本次要解决的核心问题如何用Python从HighD这一大盘“原料”里精准地烹饪出“超车变道邻近车辆数据”这道“菜”。这不仅仅是写几行pandas的filter那么简单它涉及到对数据结构的理解、对驾驶场景的定义、对时空关系的判断以及如何高效处理大规模数据。接下来我就结合自己的实操经验把整个流程掰开揉碎了讲清楚。2. 庖丁解牛深入理解HighD数据结构与核心字段在动手写代码之前我们必须像熟悉自己的工具一样熟悉HighD的数据结构。盲目处理只会事倍功半。HighD数据集主要包含两个核心文件*_tracks.csv轨迹文件和*_tracksMeta.csv元数据文件。此外还有记录场景信息的*_recordingMeta.csv但对于车辆筛选前两者是关键。2.1 轨迹文件 (*_tracks.csv): 每一帧的瞬间状态这个文件记录了所有车辆在每一帧下的状态。你可以把它想象成一个超大的Excel表每一行是某个车辆在某个时刻的一张“快照”。它的核心字段包括trackId:车辆的唯一标识符。这是串联所有数据的“主键”在同一段录制Recording内是唯一的。frame: 帧编号。从0开始连续递增代表了时间序列。HighD的帧率通常是25Hz每秒25帧所以相邻帧间隔0.04秒。x,y: 车辆中心在图像坐标系下的位置单位米。特别注意HighD使用的是图像坐标系原点在左上角x轴向右y轴向下。这与我们通常理解的地图坐标系北东天或车辆坐标系前左天不同在计算相对位置和距离时需要小心。xVelocity,yVelocity: 在x和y方向上的速度分量米/秒。xAcceleration,yAcceleration: 在x和y方向上的加速度分量米/秒²。frontSightDistance,backSightDistance: 到车道前/后可见距离内最近物体的距离。这个信息有时可用于辅助判断但并非绝对可靠。dhw(Distance Headway),thw(Time Headway): 到前车Preceding Vehicle的距离和时间车头时距。这是筛选前车的直接依据。如果车辆前方没有车这些值会被设为-1。precedingId,followingId: 在同一车道内本车的前车和后车的trackId。这是筛选同车道邻近车辆的关键字段如果不存在值为0。leftPrecedingId,leftAlongsideId,leftFollowingId: 在左侧相邻车道本车的前车、并行车、后车的trackId。rightPrecedingId,rightAlongsideId,rightFollowingId: 在右侧相邻车道本车的前车、并行车、后车的trackId。laneId: 车辆所在车道的ID。通常从最左侧车道开始编号如1, 2, 3...。这是判断变道行为的核心依据之一。注意precedingId/followingId与leftPrecedingId等字段是互斥的。precedingId特指同车道的前车。当车辆变道时这些关联ID会动态变化这为我们捕捉交互对象提供了直接线索。2.2 元数据文件 (*_tracksMeta.csv): 每辆车的全局信息这个文件记录了每辆车在整个录制过程中的整体属性一行对应一辆车一个trackId。重要字段包括trackId: 与轨迹文件对应。initialFrame,finalFrame: 该车辆首次和末次出现的帧号。可以快速计算车辆存在的总时长。numFrames: 车辆出现的总帧数。width,length: 车辆的宽度和长度米用于计算安全边界或绘制车辆轮廓。class: 车辆类型如Car,Truck等。在研究超车行为时你可能需要区分小车超卡车和卡车超小车等不同场景。理解这些字段是第一步。接下来我们需要基于这些字段定义出什么是我们关心的“超车变道”事件。3. 定义事件如何从数据中识别一次“超车变道”在代码中我们不能模糊地说“找超车”必须给出精确的、可量化的定义。一个典型的超车变道事件可以分解为以下几个阶段我们可以用数据字段来刻画每个阶段跟随阶段主车Ego Vehicle在原始车道例如车道2内跟随一辆速度较慢的前车precedingId有效且不为0。此时dhw距离和thw时距可能较小主车速度可能低于期望速度。意图与准备阶段主车产生超车意图。在数据上可能表现为开始接近前车dhw减小、计算与相邻车道车辆的空间关系观察leftPrecedingId,leftFollowingId等字段是否显示有足够空隙。变道执行阶段这是最关键的阶段。主车的laneId发生变化例如从车道2变为车道1向左变道或车道3向右变道。在HighD中一次laneId的跳变如2-1通常就代表了一次变道事件。我们需要找到laneId发生变化的起止帧。超越阶段在主车变道至目标车道后其x坐标沿道路方向逐渐超过原车道前车的x坐标。注意由于HighD使用图像坐标系x轴代表道路方向录制方向所以比较x坐标可以判断纵向位置的先后。并回阶段可选在超越完成后主车可能再次变道回原车道。这定义了完整的“超车-并回”行为。基于以上阶段我们可以定义一个超车变道事件的数据化条件必要条件主车的laneId在连续帧中发生了变化例如在帧区间[100, 150]内laneId从2变成了1。核心条件在变道发生前的一段时间内例如变道前2秒主车在原车道有一个有效的precedingId即存在一辆前车。在变道发生后的一段时间内例如变道后3秒主车在目标车道的x坐标最终超过了那辆原前车的x坐标。辅助条件用于净化数据变道过程持续时间合理例如1-5秒变道前后主车速度大于被超车辆速度变道过程中与目标车道前后车保持安全距离通过leftPrecedingId等关联车辆的相对距离判断。有了清晰的定义我们就可以开始用Python来寻找这些事件了。首要任务是把数据高效地读进来。4. 实战第一步使用Pandas高效加载与预处理数据HighD的单个轨迹文件就可能包含数十万行直接全量读入内存可能效率低下。我们需要有策略地进行加载和初步过滤。4.1 分块读取与初步筛选如果你的内存充足直接使用pandas.read_csv是最简单的。但为了稳健尤其是处理多个文件时可以考虑分块chunk读取。import pandas as pd import numpy as np from pathlib import Path # 假设我们只处理一个录制文件 tracks_path Path(./data/01_tracks.csv) meta_path Path(./data/01_tracksMeta.csv) # 读取元数据我们可能先根据元数据筛选车辆类型如只选小汽车 df_meta pd.read_csv(meta_path) car_ids df_meta[df_meta[class] Car][trackId].tolist() # 读取轨迹数据可以只读取我们关心的列节省内存 usecols [trackId, frame, x, y, xVelocity, yVelocity, laneId, precedingId, followingId, leftPrecedingId, leftFollowingId, leftAlongsideId, rightPrecedingId, rightFollowingId, rightAlongsideId] df_tracks pd.read_csv(tracks_path, usecolsusecols) # 可选如果只研究小汽车提前过滤 df_tracks df_tracks[df_tracks[trackId].isin(car_ids)]4.2 关键衍生字段计算原始数据中的速度是分量我们通常更关心车辆沿道路方向的速度纵向速度。由于HighD的x轴大致是道路方向我们可以用xVelocity来近似。更严谨的做法是计算合速度在道路方向上的投影但这需要道路方向向量比较复杂。对于筛选和初步分析用xVelocity作为纵向速度是常见的简化。# 计算纵向速度近似取绝对值因为xVelocity可能为负如果录制方向与x轴相反需根据recordingMeta确认 # 通常HighD中车辆前进方向是x轴正方向 df_tracks[lonVelocity] df_tracks[xVelocity] # 计算加速度 df_tracks[lonAcceleration] df_tracks[xAcceleration]4.3 数据排序与分组为了后续按车辆、按时间查找数据正确的排序至关重要。# 确保数据按车辆ID和帧号排序 df_tracks.sort_values(by[trackId, frame], inplaceTrue) # 重置索引 df_tracks.reset_index(dropTrue, inplaceTrue)预处理完成后我们手头就是一个干净的、按车辆和时间排好序的DataFrame。接下来就是最核心的部分算法实现。5. 核心算法实现逐车扫描定位变道与筛选邻近车我们的目标是遍历每一辆可能作为主车的车辆找出它所有的变道事件并为每个事件筛选出相关的邻近车辆数据。这里采用一个清晰的、模块化的思路。5.1 第一步为单辆车找出所有变道点我们首先定义一个函数输入一辆车的轨迹数据一个DataFrame输出一个列表列表中的每个元素是一个字典代表一次变道事件包含start_frame变道开始帧、end_frame变道结束帧、from_lane起始车道、to_lane目标车道。def find_lane_changes_for_vehicle(vehicle_df): 找出单辆车的所有变道事件。 参数: vehicle_df: 单辆车的轨迹DataFrame已按frame排序。 返回: list of dict: 每个dict包含一次变道事件的信息。 lane_changes [] # 获取车道ID序列 lane_ids vehicle_df[laneId].values frames vehicle_df[frame].values i 0 n len(lane_ids) while i n: current_lane lane_ids[i] start_frame frames[i] # 寻找车道ID变化的点 j i 1 while j n and lane_ids[j] current_lane: j 1 # 如果j n说明在j处车道发生了变化 if j n: # 变道结束帧可以认为是变化前的最后一帧或者变化后的第一帧这里定义为变化前的最后一帧。 # 更精确的做法是寻找车道ID变化的过渡区间但HighD数据中变道通常发生在1-几帧内。 # 我们简单地将车道ID发生变化的帧区间[start_frame, frames[j])定义为一次变道。 # 实际上frames[j]是新车道的起始帧。 end_frame frames[j-1] # 旧车道的最后一帧 to_lane lane_ids[j] # 新车道的ID lane_changes.append({ start_frame: start_frame, # 变道开始旧车道最后一帧这里需要明确 end_frame: end_frame, # 变道结束旧车道最后一帧 change_frame: frames[j], # 新车道的起始帧这是一个关键帧 from_lane: current_lane, to_lane: to_lane }) # 从新车道的起始点开始继续搜索 i j else: break # 上述简单算法可能会把短暂的GPS抖动识别为变道。需要过滤。 # 过滤条件变道前后车道ID应该是合理的如相邻车道且变道过程持续时间不能太短。 filtered_changes [] for lc in lane_changes: duration lc[change_frame] - lc[start_frame] # 假设帧率25Hz合理变道时间通常在1-4秒即25-100帧。过滤掉持续时间过短如10帧的“抖动”。 if duration 10 and abs(lc[to_lane] - lc[from_lane]) 1: filtered_changes.append(lc) return filtered_changes5.2 第二步判断一次变道是否为“超车”变道不是所有变道都是为了超车。可能是为了下匝道、避让或单纯换道。我们需要用前面定义的核心条件来验证。def is_overtaking_lane_change(vehicle_df, lane_change_event, df_tracks_all): 判断一次变道事件是否是一次超车变道。 参数: vehicle_df: 主车的轨迹DataFrame。 lane_change_event: find_lane_changes_for_vehicle返回的一个事件字典。 df_tracks_all: 完整的轨迹DataFrame用于查找其他车辆。 返回: tuple: (bool, dict) 是否为超车变道以及包含被超车辆ID等信息的详情字典。 ego_id vehicle_df[trackId].iloc[0] change_frame lane_change_event[change_frame] from_lane lane_change_event[from_lane] to_lane lane_change_event[to_lane] # 1. 寻找变道前在主车原车道的前车 (precedingId) # 取变道前一段时间如2秒50帧的数据 look_back_start max(vehicle_df[frame].min(), change_frame - 50) pre_change_data vehicle_df[(vehicle_df[frame] look_back_start) (vehicle_df[frame] change_frame)] if pre_change_data.empty: return False, {} # 获取变道前最接近变道时刻的precedingId # 注意precedingId可能在整个阶段变化我们取最后时刻的值 last_preceding_id pre_change_data.iloc[-1][precedingId] if pd.isna(last_preceding_id) or last_preceding_id 0: # 变道前没有前车可能不是跟随超车可能是自由流变道 return False, {} # 2. 获取被超车辆leading_vehicle的轨迹 leading_df df_tracks_all[df_tracks_all[trackId] last_preceding_id] # 3. 判断在变道后一段时间内主车是否超越了前车 # 取变道后一段时间如3秒75帧的数据 look_forward_end min(vehicle_df[frame].max(), change_frame 75) post_change_data vehicle_df[(vehicle_df[frame] change_frame) (vehicle_df[frame] look_forward_end)] if post_change_data.empty or leading_df.empty: return False, {} # 需要确保在变道后时间段内被超车辆仍然存在 # 计算主车和被超车辆在变道后每一帧的纵向位置x坐标 # 我们需要一个时间对齐的对比。可以取两个数据集在时间帧上的交集。 common_frames np.intersect1d(post_change_data[frame].values, leading_df[frame].values) if len(common_frames) 10: # 共同帧太少无法可靠判断 return False, {} ego_x_post post_change_data.set_index(frame)[x].reindex(common_frames) lead_x_post leading_df.set_index(frame)[x].reindex(common_frames) # 判断在共同时间序列的后期主车x是否持续大于被超车辆x即完成了超越 # 可以看最后若干帧的平均位置差 if len(common_frames) 20: check_frames common_frames[-20:] else: check_frames common_frames[-5:] ego_x_check ego_x_post.loc[check_frames].mean() lead_x_check lead_x_post.loc[check_frames].mean() is_overtaken ego_x_check lead_x_check if is_overtaken: # 还可以添加更多验证例如变道前主车速度是否小于前车速度有超车动机 # 变道前后主车速度是否提升 return True, { ego_id: ego_id, leading_id: last_preceding_id, change_frame: change_frame, from_lane: from_lane, to_lane: to_lane, overtake_completion_frame: common_frames[-1] # 以最后共同帧作为超越完成参考 } else: return False, {}5.3 第三步为一次超车事件筛选所有相关邻近车辆一旦确认了一次超车变道事件我们就需要提取事件前后一个时间窗口内所有与主车有交互的车辆数据。这包括被超车辆Leading Vehicle上面已经识别。原车道后车Original Lane Following Vehicle变道前主车在原车道的后车followingId。目标车道前车Target Lane Preceding Vehicle变道后主车在目标车道的前车根据变道方向取leftPrecedingId或rightPrecedingId在变道时刻的值。目标车道后车Target Lane Following Vehicle变道后主车在目标车道的后车leftFollowingId或rightFollowingId。其他潜在相关车辆如目标车道的并排车辆alongsideId或者更宽时间窗口内进入交互区域的车辆。def get_surrounding_vehicles_for_event(ego_id, event_info, df_tracks_all, time_window_seconds5.0): 获取一次超车事件前后时间窗口内所有相关车辆的轨迹数据。 参数: ego_id: 主车ID。 event_info: is_overtaking_lane_change返回的详情字典。 df_tracks_all: 完整轨迹数据。 time_window_seconds: 时间窗口半长秒默认取事件前后5秒。 返回: dict: 键为角色如ego, leading, orig_following, ...值为对应车辆的轨迹DataFrame截取时间窗口内。 frame_rate 25.0 # HighD典型帧率 window_frames int(time_window_seconds * frame_rate) event_frame event_info[change_frame] start_frame max(0, event_frame - window_frames) end_frame event_frame window_frames # 1. 获取主车数据 ego_df df_tracks_all[(df_tracks_all[trackId] ego_id) (df_tracks_all[frame] start_frame) (df_tracks_all[frame] end_frame)].copy() if ego_df.empty: return {} result {ego: ego_df} leading_id event_info.get(leading_id) # 2. 获取被超车辆数据 if leading_id: leading_df df_tracks_all[(df_tracks_all[trackId] leading_id) (df_tracks_all[frame] start_frame) (df_tracks_all[frame] end_frame)].copy() if not leading_df.empty: result[leading] leading_df # 3. 获取原车道后车变道前一刻 # 找到变道前最近一帧主车的数据 ego_pre_change ego_df[ego_df[frame] event_frame] if not ego_pre_change.empty: last_frame_pre ego_pre_change[frame].max() last_row_pre ego_df[ego_df[frame] last_frame_pre].iloc[0] orig_following_id last_row_pre[followingId] if pd.notna(orig_following_id) and orig_following_id ! 0: orig_following_df df_tracks_all[(df_tracks_all[trackId] orig_following_id) (df_tracks_all[frame] start_frame) (df_tracks_all[frame] end_frame)].copy() if not orig_following_df.empty: result[original_following] orig_following_df # 4. 获取目标车道前车和后车变道后一刻 # 找到变道后最近一帧主车的数据 ego_post_change ego_df[ego_df[frame] event_frame] if not ego_post_change.empty: first_frame_post ego_post_change[frame].min() first_row_post ego_df[ego_df[frame] first_frame_post].iloc[0] # 判断变道方向 from_lane event_info[from_lane] to_lane event_info[to_lane] if to_lane from_lane: # 向左变道 target_preceding_id first_row_post[leftPrecedingId] target_following_id first_row_post[leftFollowingId] else: # 向右变道 target_preceding_id first_row_post[rightPrecedingId] target_following_id first_row_post[rightFollowingId] for role, tid in [(target_preceding, target_preceding_id), (target_following, target_following_id)]: if pd.notna(tid) and tid ! 0: vehicle_df df_tracks_all[(df_tracks_all[trackId] tid) (df_tracks_all[frame] start_frame) (df_tracks_all[frame] end_frame)].copy() if not vehicle_df.empty: result[role] vehicle_df return result5.4 第四步主循环与结果整合最后我们将上述模块组合起来遍历所有车辆收集所有超车事件及其邻近车辆数据。def extract_all_overtaking_events(df_tracks, df_meta): 从整个数据集中提取所有超车事件及邻近车辆数据。 返回一个列表每个元素是一个字典代表一个完整的事件数据包。 all_events_data [] unique_vehicle_ids df_tracks[trackId].unique() for ego_id in unique_vehicle_ids[:100]: # 示例先处理前100辆车避免耗时过长 ego_df df_tracks[df_tracks[trackId] ego_id].copy() if len(ego_df) 50: # 轨迹太短忽略 continue # 找出该车的所有变道事件 lane_changes find_lane_changes_for_vehicle(ego_df) for lc in lane_changes: is_overtake, event_info is_overtaking_lane_change(ego_df, lc, df_tracks) if is_overtake: # 获取该事件相关的所有车辆数据 surrounding_data get_surrounding_vehicles_for_event(ego_id, event_info, df_tracks, time_window_seconds4.0) if surrounding_data: # 将事件信息和车辆数据打包 event_package { event_info: event_info, surrounding_data: surrounding_data } all_events_data.append(event_package) print(f发现超车事件: 主车 {ego_id} 在帧 {event_info[change_frame]} 超越 {event_info[leading_id]}) print(f总共发现 {len(all_events_data)} 个超车事件。) return all_events_data运行extract_all_overtaking_events函数你就能得到一个列表里面包含了每一个识别出的超车事件的所有关键数据。每个事件包都包含了主车、被超车以及其他邻近车辆在事件前后时间窗口内的完整轨迹。这些数据可以直接用于后续的可视化、特征提取和模型训练。6. 避坑指南与性能优化实战经验理论很美好但实际处理数据时总会遇到各种坑。下面分享几个我踩过坑后总结的经验。6.1 坐标系的坑图像坐标系与真实距离HighD使用的是图像坐标系x向右y向下。这意味着相对位置计算车辆A在车辆B的“左边”并不意味着在真实道路上A在B的左侧车道。你需要根据laneId来判断车道关系leftPrecedingId等字段已经帮你做好了这种基于车道的关联比自己用x,y坐标算更可靠。纵向距离两车在道路方向上的距离近似等于它们x坐标的差值假设道路是水平的且与x轴平行。但严格来说应该使用dhw车头时距字段它提供了沿车道中心线的距离。在筛选“近距离”车辆时优先使用dhw而不是欧几里得距离。速度方向xVelocity在车辆前进方向与x轴正方向一致时为正。在计算相对速度时要注意符号。6.2 数据噪声与异常值的处理车道ID抖动车辆在车道内行驶时laneId偶尔会有1-2帧的跳变这通常不是真实的变道。这就是为什么在find_lane_changes_for_vehicle函数中我们需要过滤持续时间过短的“变道”。阈值可以设为10-15帧0.4-0.6秒。关联ID的瞬时消失precedingId等字段可能在连续帧中短暂变为0或NaN然后又恢复。这可能是传感器误检或遮挡。在判断“前车”时最好看一个时间窗口内的稳定状态而不是单帧数据。例如在is_overtaking_lane_change函数中我们查看变道前2秒的数据来确认前车。缺失帧虽然HighD数据很连续但极端情况下可能有帧丢失。在按帧对齐不同车辆的数据时如判断超越时刻使用np.intersect1d找共同帧是稳健的做法。6.3 大规模数据处理的性能优化当处理多个录制文件几十GB时纯Python循环会非常慢。以下是一些优化策略向量化操作尽量使用pandas和numpy的向量化函数避免在DataFrame上使用apply或逐行循环。例如计算所有车辆的纵向速度直接用df[xVelocity]赋值。使用数据索引对df_tracks按trackId和frame建立索引可以大幅加速查询。df_tracks_indexed df_tracks.set_index([trackId, frame]).sort_index() # 查询特定车辆特定帧的数据会很快 # ego_data df_tracks_indexed.loc[(ego_id, slice(start_frame, end_frame)), :]但注意多级索引在某些操作上可能不如布尔索引直观。分治与并行最耗时的部分是遍历每辆车找变道事件。这个过程是独立的可以并行。按车辆分组并行使用joblib或multiprocessing库将车辆ID列表分成若干份分配给多个进程同时处理。from joblib import Parallel, delayed def process_one_vehicle(vid, df_tracks, df_meta): # 提取该车辆数据并运行上述查找逻辑 # 返回该车辆发现的所有事件列表 pass all_vehicle_ids df_tracks[trackId].unique() results Parallel(n_jobs4)(delayed(process_one_vehicle)(vid, df_tracks, df_meta) for vid in all_vehicle_ids[:200]) # 然后合并results使用Dask如果数据量极大可以考虑使用Dask库进行分布式计算它提供了类似pandas的接口但能处理超出内存的数据。选择性加载一开始就通过usecols参数只加载必需的列能极大减少内存占用和I/O时间。结果缓存识别超车事件的计算成本很高。一旦计算完成应将结果all_events_data保存为文件如pickle或parquet格式后续分析直接加载结果避免重复计算。6.4 定义超车事件的阈值调参在is_overtaking_lane_change函数中我们使用了几个关键阈值look_back_start change_frame - 50查看变道前2秒的数据来判断前车。look_forward_end change_frame 75查看变道后3秒的数据来判断是否完成超越。duration 10过滤短于10帧0.4秒的虚假变道。这些阈值需要根据你的具体研究问题进行调整。例如研究激进超车可能用更短的时间窗口研究保守超车可能用更长的。建议在少量数据上可视化几个案例来验证你的阈值是否合理。7. 从数据到洞察可视化验证与特征提取算法跑完了怎么知道抓取得对不对可视化是必不可少的验证步骤。同时提取出来的数据需要转化为可用于机器学习模型的特征。7.1 使用Matplotlib进行事件可视化我们可以绘制单个超车事件中所有相关车辆在时间-空间x坐标上的轨迹。import matplotlib.pyplot as plt def plot_overtaking_event(event_package, df_tracks): event_info event_package[event_info] surrounding event_package[surrounding_data] fig, ax plt.subplots(figsize(12, 6)) ego_id event_info[ego_id] change_frame event_info[change_frame] # 为每种角色定义颜色和标签 roles_config { ego: {color: red, label: Ego, linewidth: 3}, leading: {color: blue, label: Leading, linewidth: 2}, original_following: {color: green, label: Orig Follow, linestyle: --}, target_preceding: {color: orange, label: Target Precede, linestyle: :}, target_following: {color: purple, label: Target Follow, linestyle: :}, } for role, df in surrounding.items(): if role in roles_config: config roles_config[role] # 绘制纵向位置 (x) 随时间 (frame) 的变化 ax.plot(df[frame], df[x], colorconfig[color], labelconfig[label], linewidthconfig.get(linewidth, 1), linestyleconfig.get(linestyle, -)) # 标记变道时刻 ax.axvline(xchange_frame, colorblack, linestyle--, alpha0.5, labelLane Change) ax.set_xlabel(Frame) ax.set_ylabel(Longitudinal Position (x) [m]) ax.set_title(fOvertaking Event: Ego {ego_id} - Leading {event_info.get(leading_id, N/A)}) ax.legend() ax.grid(True, alpha0.3) plt.tight_layout() plt.show() # 可视化第一个事件 if all_events_data: plot_overtaking_event(all_events_data[0], df_tracks)通过看图你可以直观地检查主车红色是否在变道时刻黑色虚线后其轨迹x坐标超过了被超车辆蓝色。其他车辆的轨迹也能帮助你理解整个交互场景。7.2 提取机器学习特征对于每个超车事件我们可以从surrounding_data中提取一系列特征用于后续的驾驶行为分类、风险评估等模型。特征可以分为几类主车特征变道前的平均速度、加速度、与前车的dhw、thw。相对运动特征主车与被超车的相对速度、相对距离在变道前、变道时刻、超越完成时刻。间隙特征变道时目标车道前后车的距离dhw与thw的变体即切入间隙和跟随间隙。时间特征变道持续时间、从产生意图到开始变道的时间、超越全过程时间。交互特征是否存在并排车辆、后车是否加速逼近等。例如提取变道前主车状态的特征def extract_features_from_event(event_package): event_info event_package[event_info] surrounding event_package[surrounding_data] features {} ego_df surrounding.get(ego) leading_df surrounding.get(leading) change_frame event_info[change_frame] if ego_df is not None and leading_df is not None: # 1. 变道前主车状态取变道前1秒的数据平均 pre_change ego_df[ego_df[frame] change_frame] last_second pre_change[pre_change[frame] (change_frame - 25)] # 前1秒 if not last_second.empty: features[ego_pre_speed_mean] last_second[lonVelocity].mean() features[ego_pre_acc_mean] last_second[lonAcceleration].mean() # 2. 变道时刻的间隙需要从原始数据中获取这里简化 # 假设我们能从ego_df中拿到变道时刻的leftPrecedingDistance等实际需计算 # features[gap_to_target_lead] ... # 3. 超越完成时的相对速度 # 找到超越完成帧附近的数据 completion_frame event_info.get(overtake_completion_frame, change_frame 50) ego_post ego_df[ego_df[frame] completion_frame].iloc[0] if not ego_df[ego_df[frame] completion_frame].empty else None lead_post leading_df[leading_df[frame] completion_frame].iloc[0] if not leading_df[leading_df[frame] completion_frame].empty else None if ego_post is not None and lead_post is not None: features[relative_speed_at_completion] ego_post[lonVelocity] - lead_post[lonVelocity] # ... 可以提取更多特征 return features将这些特征提取出来组织成一个新的DataFrame就可以作为机器学习模型的输入了。整个流程从数据加载、事件定义、算法实现、到优化避坑和结果应用形成了一个完整的闭环。处理HighD这样的数据集核心在于对数据字段的深刻理解和对驾驶场景的准确定义。代码本身并不复杂但其中的逻辑和细节决定了你提取的数据质量。希望这篇近万字的详细拆解能让你在处理HighD或其他类似轨迹数据时思路更清晰操作更顺畅。