ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

基于SUMO的实时路况感知与备选路径推荐实现解析

基于SUMO的实时路况感知与备选路径推荐实现解析

最近在折腾城市交通仿真时接触到一个很有意思的课题:用SUMO做备选路径推荐,同时把路口的实时动态信息“喂”给小车。简单说就是仿真里的一辆小车每走到一个路口,就反馈这个路口的信控状态、排队长度、通过时间,然后基于这些动态数据重新计算后续路线,并给出不止一条的备选路径。这个思路听起来像是导航软件的简化版,但放在SUMO里做,其实反而能让人把路径规划和实时路况之间的耦合关系看得更清楚。

这个方向特别适合两类人:一类是正在做车路协同、智慧路口课题的学生,另一类是刚接触SUMO想用它做动态路径规划试验的工程师。它的核心价值不在“能跑”,而在把“实时道路信息获取”和“备选路径推荐”这两件事串起来——很多教程只教你用TraCI读取速度,或者只教你算静态最短路径,但要玩出动态推荐,就得把“路口信息反馈”作为触发器和路径重算的依据。

下面我按自己做过的“模拟项目X”来拆解这套方案:从整体设计、核心细节、实操实现到踩坑排查,一步步说清楚。代码和配置都是可以直接拿去改的。

1. 项目整体设计与思路拆解

1.1 先弄明白“备选路径”到底要解决什么问题

导航里的备选路径,不是给你画三条好看的线,而是要让司机在“当前已知路况下”有得选。在SUMO里做这件事,本质是把路网抽象成带权图,边的权值是通行成本(时间或长度),然后在此基础上求K条最优路径。

但这里有个关键点:**传统静态算法里边的权值是固定不变的,而你要做的是让权值跟着实时路况变。**所以整体设计天然分成两层:一层是“信息采集层”,负责每到一个路口就获取这个路口的交通状态、边速度,并把这些信息转成可比较的数值;另一层是“决策推荐层”,在信息更新的基础上,重新用图搜索算法生成几条候选路线,并挑出在当前时刻成本最低的一条作为主推荐。

这两层必须解耦。如果你把“采集”和“计算”揉在一个Loop里,代码会越写越乱,而且当路网规模变大时,你根本分不清是数据没更新还是算法算慢了。我建议的做法是:采集层独立运行,每过一定时间间隔或者每触发一个事件(比如车辆到达路口),就向决策层发送一份“路况快照”;决策层拿到快照后再决定要不要重算、重算几条。

1.2 为什么选SUMO而不是其他仿真工具

没用过SUMO的朋友可能不理解,为什么要用交通仿真器来做路径推荐。现实理由很实在:SUMO是开源的、可编程的,而且它提供了TraCI这个“实时控制协议”。通过TraCI,你可以做一个Python或者Java的外部控制端,把自己写成“上帝视角”,随时读取仿真中任意车辆、车道、信号灯的状态,甚至改变它们的属性。

对比用真实地图API来做,SUMO的优势在于可控。你可以随意设置堵车点、事故路段、信号灯配时突变,然后反复验证自己的推荐算法在这些扰动下表现如何。而对比写一个纯图论的模拟器,SUMO又提供了接近真实的交通流模型和路口逻辑。这种“介于真实和简单之间”的定位,特别适合练手。

1.3 整体架构:一张图说清数据流转

虽然不能用图表展示,但我用文字描述一下这个结构:

  • 路网与车流数据:SUMO根据路网文件和车辆路由文件加载环境。
  • 控制端程序(Python):通过TraCI连接SUMO,启动仿真Step,推进仿真。
  • 路口信息采集模块:监听车辆的位置,判断其是否到达路口附近;到达后向SUMO查询该路口关联的边、车道、信号灯和排队长度。
  • 路况快照生成:把采集到的信息换算成边的时间成本,并维护一张动态权重表。
  • 路径重算模块:基于动态权重表,使用Dijkstra/Yen算法计算K条备选路径。
  • 推荐结果下发:将备选路径列表写回仿真车辆,或者打印输出用于评估。

这个链路听起来简单,但每一环都有细节。就拿“车辆到达路口”这个触发条件来说,如果你只是用坐标距离判断,那么车辆在路口前很远处就会被判定为“到达”,干扰路径重算的时机。比较好的做法是:订阅车辆的laneChange或者route信息,或者直接判断车辆所在路段(edge)是否是路口的入边。

2. 核心细节解析与实操要点

2.1 实时动态道路信息的“实时”到底怎么定义

很多人一看到“实时”就以为每一仿真秒都要重算一次全图。其实没必要,也是性能灾难。我的处理方式是:把“实时”定义为“关键事件触发”和“周期更新”相结合。

周期更新很简单:每5个仿真秒(相当于这里的5秒),从SUMO里统一拉取所有车道的最新速度,然后计算每条边的通行时间。事件触发的场景就是标题里说的“小车每到一个路口”:当车辆越过路口中心点、进入下一条边的那一刻,立即采集这个路口当前的信号灯状态、各进口车道的排队长度,以及最近一个周期的平均通过时间。

这两种更新的频率可以不一样。事件触发是“局部”的,周期更新是“全局”的。组合起来就能既保证新路口的突发拥堵能立刻反映到路径里,又不会因为频繁的全图扫描把CPU耗尽。实测下来,200多辆车的路网,5秒全量更新一次TraCI读取大概只占用百分之几的CPU,但如果你每0.1秒扫一次全图,CPU会暴涨。

2.2 路口信息到底采集哪些字段

标题里说“每到一个路口、就返回这个路口的信...”,信息不止是“红绿灯是红的还是绿的”。在我的项目里,我采集了以下字段:

  • 路口ID,以及当前信号的相位索引和剩余时间。
  • 这个路口所有入边(incoming edge)和出边(outgoing edge)的列表。
  • 每条入边当前的平均速度。
  • 每条入边的车道排队长度,也就是车辆几乎停止(速度低于0.1m/s)的车辆数量。
  • 当前通过这个路口(左转、直行、右转)的历史平均耗时,这个可以从仿真数据里统计。

为什么要这些?因为路径推荐最终要换算到边成本。从一个路口到下一个路口,不仅要知道这条路本身有多长,还要知道在路口等待红绿灯要多久。很多博客只把速度当成权重,忽略了路口排队,这是不完整的。

我采集这些数据的代码骨架是:

def get_intersection_status(traci, vehicle_id): # 根据车辆当前或者上一个控制点获取路口ID current_edge = traci.vehicle.getRoadID(vehicle_id) junction_id = traci.edge.getJunction(current_edge) # 获取信号灯状态 if junction_id.startswith('J') and traci.trafficlight.getIDList(): try: phase = traci.trafficlight.getPhase(junction_id) state = traci.trafficlight.getRedYellowGreenState(junction_id) except: phase = -1 state = '' # 获取入边的排队长度 incoming_edges = [edge_id for edge_id in traci.edge.getIDList() if traci.edge.getJunction(edge_id) == junction_id] queues = [] for edge_id in incoming_edges: lane_ids = traci.lane.getIDList() # 真实实现中需要按edge筛选lane queue_length = traci.edge.getLastStepVehicleNumber(edge_id) * 7.5 # 平均车长 speed = traci.edge.getLastStepMeanSpeed(edge_id) queues.append((edge_id, speed, queue_length)) return {'junction': junction_id, 'phase': phase, 'state': state, 'incoming': queues}

注意,traci.edge.getJunction并不是每个版本都有,这里只是示意。实际判断路口对应的入边,需要解析路网文件或者预先建好映射。我后来直接把路网里的节点类型为priority的节点提前存成一个列表,离线做好连接关系。

2.3 备选路径推荐的核心:不是最短路径,而是K条低代价路径

备选路径推荐不能简单地找一条最短路径再加一条绕远路,这两条路如果重叠度太高,一旦前方拥堵,备选路线也跟着堵。所以标准做法是使用K最短路径算法,其中比较经典的是Yen算法。Yen算法在Dijkstra基础上反复“偏离”,能生成多条除起点终点外不重复的路线,非常适合导航场景。

在SUMO场景里,我更推荐对Yen算法的输出做一步“去重和剪枝”:只保留重合度低于50%的候选路线,同时丢弃内部包含明显绕行(比最优路线长1.5倍以上)的候选。这样最终的推荐列表才有实际意义。

实现上,我直接用NetworkX替代自己写图数据结构。把SUMO的路网边对应到NetworkX的有向图,边的权重先从静态长度初始化,之后每个仿真周期用动态时间成本替换。完整代码可以简化成:

import networkx as nx def build_graph_from_sumo(edges_info): G = nx.DiGraph() for from_node, to_node, edge_id, travel_time in edges_info: G.add_edge(from_node, to_node, id=edge_id, weight=travel_time) return G def k_shortest_paths(G, source, target, K, weight='weight'): from networkx.algorithms.simple_paths import shortest_simple_paths generator = shortest_simple_paths(G, source, target, weight=weight) return [path for path in islice(generator, K)]

这里的shortest_simple_paths底层用的就是Yen算法的优化版。如果你不想引入重型库,也可以自己写一个,但没必要,除非你要在同一份代码里做大量算法性能实验。

2.4 动态权重如何跟静态路网融合

SUMO路网文件里的travelTime字段通常是基于不堵车时的理想速度。如果直接用动态更新后的时间覆盖掉,会导致路网里的数学属性完全失去静态基准。更好的方式是设置一个“实时时间”属性,并保留一个“自由流时间”属性。最终的路段成本 = 实时时间驱动的延迟 + 自由流时间,这样即使实时数据暂时缺失,也能退化为静态路径。

融合公式我用的是:

cost(edge) = free_flow_time(edge) * alpha + real_time_edge_cost(edge) * (1 - alpha)

其中alpha是置信系数,当实时数据越新、样本越多,alpha越小。默认取0.3,即实时数据权重占70%。这个系数可以根据实验调,我测下来在拥堵不算太严重的路网上,alpha=0.2到0.4都能得到合理结果,alpha过大就看不到实时变化,过小则会频繁变路。

3. 实操过程与核心环节实现

3.1 准备SUMO路网和仿真配置

如果是从零开始,先用netedit画一个简单路网。为了能验证“备选路径推荐”,路网至少要有两个以上并联通路,不要只有一个环线。我画的路网大概是:一个5x4的方格街区,东西方向三条主干道,南北方向四条次干道,其中有一条道路人为设为易拥堵。

路网文件是my_net.net.xml,车辆路由文件my_routes.rou.xml里定义了若干辆车的出发时间和路线。注意这里的“route”在仿真开始前就给了,我们后面要做的是通过TraCI修改它,所以初始路线随便给一条就行。

SUMO配置文件my_sumo.sumcfg里配置:

<configuration> <input> <net-file value="my_net.net.xml"/> <route-files value="my_routes.rou.xml"/> </input> <time> <begin value="0"/> <end value="1000"/> <step-length value="0.1"/> </time> <routing> <device.rerouting.probability value="1.0"/> </routing> </configuration>

step-length设为0.1秒是为了更细腻地捕捉车辆到达路口的瞬间。不过步长越小,TraCI通信越频繁,CPU压力也越大。如果不需要那么精细,可以设0.5秒,不影响整体效果。

启动SUMO时建议加上--remote-port 8813参数,开启TraCI通道。用Python连接:

import traci import sumolib sumoBinary = "sumo-gui" # 或者用sumo,无界面更快 traci.start([sumoBinary, "-c", "my_sumo.sumcfg", "--remote-port", "8813"])

3.2 建立路口与边的关系索引

这一步最繁琐,但做完之后所有采集逻辑都会变得非常顺。SUMO路网文件里,一个junction节点会被定义为<junction id="J1" type="priority" x="..." y="..." incLanes="..." intLanes="...">。我们可以用sumolib解析:

import sumolib net = sumolib.net.readNet("my_net.net.xml") junction_to_incoming = {} for junction in net.getNodes(): if junction.getType() == "priority": incoming_edges = [e.getID() for e in junction.getIncoming()] outgoing_edges = [e.getID() for e in junction.getOutgoing()] junction_to_incoming[junction.getID()] = (incoming_edges, outgoing_edges)

这样每当车辆到达一个路口,我们就可以从车辆当前所在edge推测它下一步会进入哪个junction。这里有一个细节:车辆所在edge可能很长,它可能已经在这条edge上行驶了好几秒,但还没到路口。所以我判断“到达”的标准不是“进入路口所在lane”,而是“距离路口中心点小于某个阈值”。用TraCI可以拿到车辆位置,再与路口的坐标做距离判断:

VICINITY_THRESHOLD = 15.0 def is_near_junction(vehicle_id): x, y = traci.vehicle.getPosition(vehicle_id) current_edge = traci.vehicle.getRoadID(vehicle_id) # 根据current_edge找到它前进方向的下一个junction edge = net.getEdge(current_edge) to_node = edge.getToNode() nx_, ny_ = to_node.getCoord() distance = ((x - nx_) ** 2 + (y - ny_) ** 2) ** 0.5 return distance < VICINITY_THRESHOLD, to_node.getID()

注意这里用的是toNode,因为车辆行驶方向是edge的正方向。如果路网里有双向道路,getEdge会根据道路id判断方向,所以没问题。

3.3 核心控制循环:从SUMO推进到路径推荐

我设计的主循环逻辑如下,每0.1秒推进一次仿真,但只有满足条件时才会做“路口信息采集”和“路径重算”。

import traci # 车辆名单、目标ID vehicle_id = "veh0" target_junction = "J12" # 终点路口ID dijkstra_interval = 50 # 每50步全量重算一次 candidate_paths = [] for step in range(10000): traci.simulationStep() # 判断车辆是否还在仿真里 if vehicle_id not in traci.vehicle.getIDList(): continue # 1) 事件触发:检测是否接近路口 near, junction_id = is_near_junction(vehicle_id) if near and junction_id != last_event_junction: last_event_junction = junction_id # 采集该路口的实时信息 status = collect_junction_status(junction_id) # 更新路网权重 update_edge_weights(status) # 基于新权重重新计算备选路径 candidate_paths = calculate_k_shortest_paths(vehicle_id, target_junction, K=3) # 2) 周期触发:每50步重新计算一次,应对行驶过程中前方路况变化 if step % dijkstra_interval == 0: update_all_edge_weights() candidate_paths = calculate_k_shortest_paths(vehicle_id, target_junction, K=3) # 3) 选择最佳路径并下发 if candidate_paths: best_path = choose_best_path(candidate_paths, vehicle_id) # 注意,不能直接在节点列表上设置路线,需要转换成连续边ID edge_sequence = [net.getEdge(n1).getID() for n1, n2 in zip(best_path[:-1], best_path[1:])] edge_sequence = extract_edge_ids_from_node_path(net, best_path) traci.vehicle.setRoute(vehicle_id, edge_sequence)

这里有一个关键坑:SUMO的setRoute接收的是edge ID列表,不是node ID列表。所以用NetworkX算出来的是节点路径,必须通过路网关系转成edge路径。如果你用的是net.getEdge,需要知道每条edge的起始节点。转换函数可能是:

def extract_edge_ids_from_node_path(net, node_path): edge_ids = [] for i in range(len(node_path) - 1): from_node = net.getNode(node_path[i]) to_node = net.getNode(node_path[i+1]) for edge in from_node.getOutgoing(): if edge.getToNode().getID() == to_node.getID(): edge_ids.append(edge.getID()) break return edge_ids

这段代码我在实际运行中踩了挺多次坑,因为SUMO里有些节点间可能存在多条parallel edge(比如一条机动车道加一条公交专用道),需要选择车辆可用的那一条。更稳的办法是用traci.simulation.convertStageToTrip或者利用sumolib.net的连接信息来精确匹配。如果只是小实验,可以用traci.vehicle.setRoute直接设置连续边,确保每一条边都能到达终点即可。

3.4 信号灯状态对路径推荐的影响

刚才提到要采集信号灯状态。但很多人会问:路径推荐里,信号灯状态到底怎么用?你要想清楚:短期看,红灯会导致等待;长期看,信号灯配时方案比瞬时颜色更有参考价值。所以路径推荐不应该因为“当前刚好是红灯”就马上绕路,而是应该根据这个路口的“平均等待时间”来评估成本。

因此我采集时除了获取getPhase,还会用过去N秒内通过该路口的平均速度,计算一个动态排队系数。具体做法:在到达路口的触发点,记录从进入排队到越过停止线的平均耗时,然后把这个耗时加到这条边的额外成本中。

论文里常说的“排队延误”就是这个意思。我用一个指数滑动平均来更新每个路口的延误估计:

avg_delay[junction] = avg_delay[junction] * 0.7 + observed_delay * 0.3

这里0.7和0.3是我调的,代表对旧观测的惯性。你也可以用更严谨的α滤波,但对小规模仿真效果差不多。

3.5 如何把备选路径的可读性和可信度做出来

备选路径推荐出来后,光有节点列表没有多大意义。为了方便观察,我会在每个仿真Step末尾打印当前推荐的三条路径以及它们预估的时间成本。同时用traci.vehicle.setColor把当前选定路线的车辆标成红色,备选路线相关的边标成蓝色,这样在SUMO GUI里一眼就能看到发生了什么。

打印效果类似:

[Step 523] veh0 到达路口 J6 - 主推路线: J0 -> J3 -> J6 -> J9 -> J12,预计通行时间 87.2s - 备选1: J0 -> J1 -> J4 -> J7 -> J10 -> J11 -> J12,预计 102.5s - 备选2: J0 -> J2 -> J5 -> J8 -> J11 -> J12,预计 118.3s

这个输出对调试特别有用。当算法表现得“抽风”时,你可以直接看到是哪条边的权重异常,导致推荐路径来回跳。

4. 常见问题与排查技巧实录

4.1 TraCI连接后仿真不前进

新手最容易遇到的现象:Python脚本里调用traci.simulationStep()后,SUMO界面卡住或者不停闪退。排查方法:

  • 确认SUMO版本和Python库匹配。常见的坑是SUMO新版对TraCI的step-length做了更严格的时间单位校验,如果你传入的是整数500,会被当作500秒而不是0.5秒。
  • 连接方式上,如果用sumo-gui启动,要确保Python里traci.start的启动命令里也用了--remote-port 8813,否则两个进程各发各的。
  • 如果同时开了多个SUMO实例,端口会冲突,改成8814或动态端口。

4.2 路径重算后车辆卡在路口不走

这个问题十有八九是因为你设置的新路线不连续,或者路线包含了车辆当前位置不可到达的边。SUMO在setRoute失败时不会报错,但车辆会停在原地很多仿真秒,然后才重新用默认行为。

我排查时用的方法:

  • 验证路线列表中的每一对相邻边是否在路网中有实际连接。
  • 验证路线的第一条边是否等于车辆当前所在边。如果重算发生在车辆到达路口后,车辆可能已经进入路口前一段,此时第一条边不是当前road id,setRoute会拒绝。
  • 最好的做法是让车辆先停在原地:traci.vehicle.setSpeed(vehicle_id, 0),等设置完新路线后再恢复。但更优雅的是用traci.vehicle.rerouteTravelTime,让SUMO自己基于当前路网做内部重算。

4.3 实时权重更新过快导致路线“贪吃蛇”

如果你每个Step都全量更新权重,车辆每到一个路口都重算,很可能出现一种现象:主推路线不停变化,一会儿去东边,一会儿西边,显示为车辆在路网中来回绕圈。原因是实时速度在较短时间尺度上波动非常大,比如一辆车临时停了,边平均速度从50降到5,然后就变成了一条“避堵”绕远路。

解决方法是给权重变化加平滑,不要直接用瞬时值。

我曾经把路段时间成本做成一阶低通滤波:

smoothed_time = alpha * current_sample + (1 - alpha) * smoothed_time alpha取0.15~0.3比较好。

另外,对重算频率做最小间隔限制,比如至少3个仿真秒后才能再一次重算同一辆车的路径,避免“还没来得及走完就又被重算”的抖动。

4.4 获取路口信息时有些路口是信号灯,有些不是

SUMO里不是所有路口都有红绿灯。可能有些路口是priority类型,用让行规则。你的采集函数里需要用traci.trafficlight.getIDList()检查该junction是否对应信号灯,如果没有信号灯,排队等待时间就可以设为0,只需统计速度。

这样也符合真实场景:交叉口有的是有信号控制,有的是无信号控制。备选路径推荐时要综合考虑两种路口的通行成本,不能对无信号路口生硬套用相位等待。

5. 避坑经验与心得总结

5.1 从“能跑通”到“跑得好”的两处优化

第一个优化是数据缓存。每次周期性调用traci.edge.getLastStepMeanSpeed会遍历所有边,在500条边的路网上还好,但一旦到城区级网络会卡住。我用一个edge_cache来存储最近一次扫描的结果,只在触发更新时才查询部分边,效果好了很多。

第二个优化是提前计算路网静态结构。将“节点-边-邻接路口”预先做成字典,而不是每次用sumolib现查。这样可以避免在TraCI事件循环里反复调用网络对象,拖累仿真速度。

5.2 我个人实际操作的体会

做这个项目时,我发现最难的部分不是算法本身,而是理解SUMO底层的时间离散化。TraCI控制的是离散的仿真Step,而非连续时间,所以你在判断“车辆到达路口”时,如果用位置距离检测,可能在两个Step之间跳过了路口,导致触发丢失。解决方式是把阈值设为略大于一个Step内最大行驶距离,或者订阅车辆的lastStepLaneChangeState事件。

我还试过使用traci.vehicle.addSubscription去订阅VAR_ROAD_ID和VAR_LANEPOSITION,用回调方式检测到达,比轮询更省CPU,但是配置订阅的代码稍微复杂一些,适合进阶使用。

5.3 未来可以怎么扩展

做完这套之后,你可以在三方面继续深挖:一是把单车场景扩展到多车协同,让多辆车共享路况信息,形成群体推荐;二是引入交通事故模拟,在特定时间点人为阻塞某一车道,观察备选路径推荐能否及时改道;三是把“实时动态道路信息获取”从内部变量扩展到可视化看板,比如通过WebSocket把路口排队长度、信号相位推送出来,这就是车路协同的原型了。

我个人认为,这个项目最值得玩味的地方,在于它把“数据采集—状态估计—决策计算—执行反馈”这个闭环完整地跑通了一遍。无论你之后是做自动驾驶决策还是交通信号优化,这套思路都是能迁移的。希望这篇文章能帮你少踩几个我自己踩过的坑,后面有更深的问题也欢迎交流。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进