【Python】 SUMO + Python 联合仿真平台

* **边 (Edge)** : 代表道路的一段，具有方向性。通常连接两个节点。一条物理道路的双向行驶通常由两条反向的边表示。
* **节点 (Node/Junction)** : 代表道路的交叉口或道路的端点。节点定义了不同边之间的连接关系和通行规则（如信号灯、优先权）。
* **车道 (Lane)** : 边可以包含一条或多条车道。车辆在车道上行驶，并遵循车道规则（如速度限制、允许的车辆类型）。
* **连接 (Connection)** : 定义了在交叉口处，车辆从一条边的某个车道如何行驶到另一条边的某个车道。连接可以有关联的信号灯相位。
* **交通信号灯 (Traffic Light System, TLS)** : 控制交叉口交通流的设备。SUMO 允许定义复杂的信号灯逻辑和相位。
* **类型 (Type)** : 可以为边和车道定义类型，例如高速公路、城市道路等，这些类型可以关联默认的速度限制、优先级等属性。

* **车辆 (Vehicle)** : 仿真的基本动态单元。每辆车都有其类型（如轿车、卡车、公交车）、出发时间、路径以及特定的驾驶行为模型参数。
* **路径 (Route)** : 车辆从起点到终点所经过的一系列边的序列。路径可以是预定义的，也可以由 SUMO 的内置路由算法动态计算。
* **行程 (Trip)** : 定义车辆的起点边和终点边，以及出发时间。SUMO 可以基于行程自动计算路径。
* **车流 (Flow)** : 定义在特定时间段内，从某条边出发，以一定速率（如 veh/h）或数量生成的车辆。常用于模拟背景交通。
* **车辆类型 (Vehicle Type, vType)** : 定义车辆的物理属性（长度、最大速度、加速度、减速度等）和行为模型参数（如跟驰模型、换道模型参数）。

* **仿真步长 (Step Length)** : 仿真时间推进的最小单位，通常为秒级（如1秒或0.1秒）。较小的步长可以提高仿真精度，但会增加计算时间。
* **开始与结束时间 (Begin/End Time)** : 定义仿真的总时长。
* **随机种子 (Seed)** : 用于控制仿真中的随机过程（如车辆出发时间、驾驶行为的随机性），确保仿真的可复现性。

* SUMO 可以生成多种类型的输出文件，记录仿真过程中的详细数据，如车辆轨迹、交叉口排队长度、行程时间、排放等。这些数据对于分析仿真结果至关重要。

    <configuration xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://sumo.dlr.de/xsd/sumoConfiguration.xsd"> <!-- XML 声明和 schema 位置 -->
    
    <input> <!-- 输入文件配置区域 -->
        <net-file value="example.net.xml"/> <!-- 指定路网文件，value 属性指向文件名 -->
        <route-files value="example.rou.xml"/> <!-- 指定路径文件，可以有多个，用逗号分隔 -->
        <!-- <additional-files value="tls_logic.add.xml, detectors.add.xml"/> --> <!-- 指定附加文件，如信号灯逻辑、探测器等 -->
    </input>
    
    <time> <!-- 时间相关配置区域 -->
        <begin value="0"/> <!-- 仿真开始时间，单位通常是秒 -->
        <end value="3600"/> <!-- 仿真结束时间，例如3600秒（1小时） -->
        <step-length value="1"/> <!-- 仿真步长，例如1秒 -->
    </time>
    
    <processing> <!-- 处理相关配置区域 -->
        <!-- <ignore-route-errors value="true"/> --> <!-- 是否忽略路径错误，例如车辆无法到达目的地 -->
        <!-- <max-depart-delay value="60"/> --> <!-- 车辆最大允许的出发延迟时间 -->
    </processing>
    
    <routing> <!-- 路径选择相关配置 -->
        <!-- <device.rerouting.period value="300"/> --> <!-- 启用车辆动态重新路由，并设置周期为300秒 -->
    </routing>
    
    <report> <!-- 报告和日志配置区域 -->
        <!-- <verbose value="true"/> --> <!-- 是否输出详细日志信息 -->
        <!-- <duration-log.statistics value="true"/> --> <!-- 是否记录仿真执行时间的统计信息 -->
        <!-- <no-step-log value="true"/> --> <!-- 是否禁止输出每个时间步的日志 (可以减少日志量) -->
    </report>
    
    <!-- <gui_only> --> <!-- 仅用于 sumo-gui 的配置 -->
        <!-- <start value="true"/> --> <!-- sumo-gui 启动时是否自动开始仿真 -->
        <!-- <tracker-interval value="0.1"/> --> <!-- sumo-gui 中车辆位置更新的视觉间隔 -->
    <!-- </gui_only> -->
    
    </configuration> <!-- 配置结束标签 -->
    
    
    xml
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/yjQPODeUhprksaYAE4KNRJIqx9Si.png)

 * 在启动 SUMO (或 `sumo-gui`) 时，需要指定一个端口号，使其作为 TraCI 服务器监听来自客户端的连接。这通常通过命令行参数 `--remote-port <PORT_NUMBER>` 实现。
 * 例如：`sumo -c my_simulation.sumocfg --remote-port 8813`

 * Python 脚本使用 SUMO 提供的 `traci` Python 库来建立与 SUMO 服务器的连接。
 * `import traci`
 * `traci.connect(port=<PORT_NUMBER>)`

 * 一旦连接建立，Python 脚本可以通过 `traci.simulationStep()` 命令使 SUMO 仿真向前推进一个时间步。
 * 在每个时间步之前或之后，Python 脚本可以调用 `traci` 库中的各种函数来： 
   * **获取 (Get)** 仿真对象（车辆、车道、信号灯等）的状态信息。
   * **设置 (Set/Change)** 仿真对象的属性或行为。
   * **订阅 (Subscribe)** 特定对象的特定变量，以便在后续时间步高效获取其变化。

 * 仿真结束后或不再需要交互时，客户端应关闭连接。
 * `traci.close()`

    import os, sys # 导入 os 和 sys 模块

    if 'SUMO_HOME' in os.environ: # 检查 SUMO_HOME 环境变量是否存在
    tools = os.path.join(os.environ['SUMO_HOME'], 'tools') # 构建 tools 目录的路径
    sys.path.append(tools) # 将 tools 目录添加到 Python 的搜索路径中
    else:
    sys.exit("please declare environment variable 'SUMO_HOME'") # 如果未设置 SUMO_HOME，则退出程序并提示
    import traci # 导入 traci 库
    
    
    python
    
    

    # 使用 libsumo 时，启动方式可能略有不同

    # import libsumo as traci # 将 libsumo 导入并别名为 traci
    # traci.start(["sumo", "-c", "my_simulation.sumocfg"]) # 使用 libsumo 启动仿真
    # ... 后续 traci 命令与标准 traci 类似
    
    
    python
    
    

* `traci.simulationStep(time_step)`: 使仿真前进指定的逻辑时间（如果 `time_step` 未提供或为0，则前进一个仿真配置文件中定义的 `step-length`）。这是 TraCI 交互的核心循环驱动。
* `traci.simulation.getTime()`: 获取当前仿真时间。
* `traci.simulation.getDeltaT()`: 获取当前仿真步长。
* `traci.simulation.loadState(filepath)` / `traci.simulation.saveState(filepath)`: 加载/保存仿真状态。
* `traci.simulation.getMinExpectedNumber()`: 获取当前时间步期望的客户端数量（通常是1，如果使用多客户端则更多）。

* `traci.vehicle.getIDList()`: 获取当前路网中所有车辆的 ID 列表。
* `traci.vehicle.getIDCount()`: 获取当前路网中车辆的总数。
* `traci.vehicle.getPosition(vehID)`: 获取指定车辆的 (x, y) 坐标。
* `traci.vehicle.getAngle(vehID)`: 获取车辆的行驶角度 (0度为x轴正向，逆时针增加)。
* `traci.vehicle.getSpeed(vehID)`: 获取车辆的当前速度 (m/s)。
* `traci.vehicle.getLaneID(vehID)`: 获取车辆当前所在车道的 ID。
* `traci.vehicle.getRoadID(vehID)`: 获取车辆当前所在边的 ID。
* `traci.vehicle.getRoute(vehID)`: 获取车辆当前的路径（边的列表）。
* `traci.vehicle.setSpeed(vehID, speed)`: 设置车辆的目标速度。
* `traci.vehicle.changeTarget(vehID, edgeID)`: 改变车辆的当前路径的下一个目标边（会触发重新规划路径到新的最终目的地）。
* `traci.vehicle.setRoute(vehID, edgeList)`: 为车辆设置一条全新的路径。
* `traci.vehicle.moveTo(vehID, laneID, pos)`: 将车辆移动到指定车道的指定位置 (慎用，可能破坏物理真实性)。
* `traci.vehicle.add(vehID, routeID, typeID="DEFAULT_VEHTYPE", depart=simTime, departLane="first", departPos="base", departSpeed="0")`: 在仿真中动态添加一辆新车。
* `traci.vehicle.remove(vehID)`: 从仿真中移除一辆车。
* `traci.vehicle.setColor(vehID, (R, G, B, A))` : 设置车辆颜色 (A是alpha透明度, 0-255)。
* `traci.vehicle.getLeader(vehID, dist)`: 获取指定车辆前方一定距离内的前车信息。
* `traci.vehicle.getFollowSpeed(vehID, speed, gap, leaderSpeed, leaderGap)`: 计算基于跟驰模型的安全跟随速度。

* `traci.lane.getLength(laneID)`: 获取车道的长度。
* `traci.lane.getMaxSpeed(laneID)`: 获取车道的最大允许速度。
* `traci.lane.getLastStepVehicleIDs(laneID)`: 获取上一个时间步在该车道上的车辆列表。
* `traci.lane.getLastStepMeanSpeed(laneID)`: 获取上一个时间步该车道的平均车速。
* `traci.lane.getLastStepOccupancy(laneID)`: 获取上一个时间步该车道的占用率。
* `traci.lane.getWaitingTime(laneID)`: 获取该车道上车辆的总等待时间。

* `traci.edge.getLastStepVehicleIDs(edgeID)`: 获取上一个时间步在该边上的车辆列表。
* `traci.edge.getLastStepMeanSpeed(edgeID)`: 获取上一个时间步该边的平均车速。
* `traci.edge.getTraveltime(edgeID)`: 获取通过该边的（动态计算的）行程时间。
* `traci.edge.setAllowed(edgeID, allowedVehicleClasses)`: 设置允许通过该边的车辆类型。

* `traci.junction.getPosition(junctionID)`: 获取交叉口的 (x, y) 坐标。

* `traci.trafficlight.getIDList()`: 获取所有交通信号灯系统的 ID 列表。
* `traci.trafficlight.getRedYellowGreenState(tlsID)`: 获取信号灯当前各进口道信号相位的状态字符串 (如 “rrgGrrgG”)。
* `traci.trafficlight.getPhase(tlsID)`: 获取当前信号灯的相位索引。
* `traci.trafficlight.setPhase(tlsID, phaseIndex)`: 设置信号灯到指定的相位索引。
* `traci.trafficlight.setPhaseDuration(tlsID, phaseDuration)`: 设置当前相位的持续时间。
* `traci.trafficlight.setProgram(tlsID, programID)`: 切换到指定的信号灯控制程序。
* `traci.trafficlight.getCompleteRedYellowGreenDefinition(tlsID)`: 获取信号灯所有相位的完整定义。

* `traci.route.add(routeID, edgeList)`: 在仿真中添加一个新的路径定义。
* `traci.route.getEdges(routeID)`: 获取指定路径ID包含的边列表。

* 用于在 `sumo-gui` 中添加和控制可视化的点或形状。
* `traci.poi.add(poiID, x, y, color, type="", layer=0)`
* `traci.polygon.add(polygonID, shape_xy_list, color, fill=True, type="", layer=0)`

* 为了高效获取大量数据，TraCI 提供了订阅机制。客户端可以预先订阅特定对象的特定变量。之后，在每个 `traci.simulationStep()` 执行后，SUMO 会自动将这些被订阅变量的最新值发送给客户端，无需客户端为每个变量单独发送 `get` 请求。
* `traci.vehicle.subscribe(vehID, varIDs=[traci.constants.VAR_SPEED, traci.constants.VAR_POSITION])`
* `traci.vehicle.getSubscriptionResults(vehID)`
* `traci.constants` 模块定义了大量可订阅的变量常量，例如 `VAR_SPEED`, `VAR_POSITION`, `VAR_LANE_ID`, `LANE_WAITING_TIME`, `LAST_STEP_MEAN_SPEED` 等。

 * 从 SUMO 官方网站 (https://sumo.dlr.de/ 或 https://eclipse.dev/sumo/) 下载适合您操作系统的最新稳定版 SUMO。
 * Windows: 通常提供预编译的安装包。
 * Linux: 可以通过包管理器 (如 `sudo apt-get install sumo sumo-tools sumo-doc`) 或从源码编译安装。
 * macOS: 可以通过 Homebrew (`brew install sumo`) 或从源码编译。
 * 安装完成后，确保 SUMO 的 `bin` 目录（包含 `sumo`, `sumo-gui`, `netconvert` 等可执行文件）和 `tools` 目录（包含 Python 脚本和 `traci` 库）已正确添加到系统环境变量 `PATH` 和 `PYTHONPATH` (或 `SUMO_HOME` 已设置)。
 * 验证安装：在命令行输入 `sumo --version`，应能看到版本信息。

 * 如果尚未安装，请从 Python 官网 (https://www.python.org/) 下载并安装 Python (推荐 3.7 或更高版本)。
 * 确保 Python 的 `Scripts` 目录（通常包含 `pip`）已添加到系统 `PATH`。

 * 打开 Python 解释器，尝试 `import traci`。如果没有错误，说明 `traci` 库可以被 Python 找到。如果找不到，请检查 `SUMO_HOME` 和 `PYTHONPATH` 设置，或者确保 `$SUMO_HOME/tools` 在 `sys.path` 中。

    <nodes xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://sumo.dlr.de/xsd/nodes_file.xsd"> <!-- 节点文件根元素 -->
    <node id="N1" x="-500.0" y="0.0" type="priority"/> <!-- 定义节点N1，坐标(-500,0)，类型为优先通行（非信号灯控制） -->
    <node id="N2" x="0.0"  y="0.0" type="traffic_light"/> <!-- 定义节点N2（交叉口中心），坐标(0,0)，类型为信号灯控制 -->
    <node id="N3" x="500.0" y="0.0" type="priority"/> <!-- 定义节点N3，坐标(500,0)，类型为优先通行 -->
    <node id="N4" x="0.0"  y="500.0" type="priority"/> <!-- 定义节点N4，坐标(0,500)，类型为优先通行 -->
    <node id="N5" x="0.0"  y="-500.0" type="priority"/> <!-- 定义节点N5，坐标(0,-500)，类型为优先通行 -->
    </nodes> <!-- 节点文件结束 -->
    
    
    xml

    <edges xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://sumo.dlr.de/xsd/edges_file.xsd"> <!-- 边文件根元素 -->
    <edge id="E1" from="N1" to="N2" priority="78" numLanes="2" speed="13.89"/> <!-- 定义边E1，从N1到N2，2车道，限速13.89m/s (50km/h) -->
    <edge id="E2" from="N2" to="N3" priority="46" numLanes="2" speed="13.89"/> <!-- 定义边E2，从N2到N3 -->
    <edge id="E3" from="N4" to="N2" priority="78" numLanes="1" speed="11.11"/> <!-- 定义边E3，从N4到N2，1车道，限速11.11m/s (40km/h) -->
    <edge id="E4" from="N2" to="N5" priority="46" numLanes="1" speed="11.11"/> <!-- 定义边E4，从N2到N5 -->
    </edges> <!-- 边文件结束 -->
    
    
    xml

    netconvert --node-files=example.nod.xml --edge-files=example.edg.xml --output-file=example.net.xml
    
    
    bash

    <routes xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://sumo.dlr.de/xsd/routes_file.xsd"> <!-- 路径文件根元素 -->
    <vType id="car_eastbound" accel="2.6" decel="4.5" sigma="0.5" length="5" maxSpeed="13.89" color="1,0,0"/> <!-- 定义东行车辆类型 "car_eastbound"，红色 -->
    <vType id="car_northbound" accel="2.6" decel="4.5" sigma="0.5" length="5" maxSpeed="11.11" color="0,0,1"/> <!-- 定义北行车辆类型 "car_northbound"，蓝色 -->
    
    <route id="route_east" edges="E1 E2"/> <!-- 定义路径 "route_east"，从 E1 到 E2 -->
    <route id="route_north" edges="E3 E4"/> <!-- 定义路径 "route_north"，从 E3 到 E4 -->
    
    <!-- 每隔一段时间产生车辆 -->
    <flow id="flow_east" type="car_eastbound" route="route_east" begin="0" end="300" period="10" departLane="random" departSpeed="max"/> <!-- 定义东向车流，0-300秒内，每10秒一辆 -->
    <flow id="flow_north" type="car_northbound" route="route_north" begin="0" end="300" period="15" departLane="random" departSpeed="max"/> <!-- 定义北向车流，0-300秒内，每15秒一辆 -->
    
    <!-- 也可以定义单个车辆 -->
    <!-- <vehicle id="veh0" type="car_eastbound" route="route_east" depart="0" departLane="0" departSpeed="10"/> --> <!-- 定义单个车辆 veh0 -->
    </routes> <!-- 路径文件结束 -->
    
    
    xml
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/UTht98uDNz5n4cf6HYekswj0POQl.png)

    <configuration xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://sumo.dlr.de/xsd/sumoConfiguration.xsd"> <!-- XML 声明 -->
    <input> <!-- 输入文件 -->
        <net-file value="example.net.xml"/> <!-- 路网文件 -->
        <route-files value="example.rou.xml"/> <!-- 路径文件 -->
    </input>
    <time> <!-- 时间配置 -->
        <begin value="0"/> <!-- 开始时间 -->
        <end value="360"/> <!-- 结束时间，例如360秒 -->
        <step-length value="1"/> <!-- 步长1秒 -->
    </time>
    <report> <!-- 报告配置 -->
        <verbose value="true"/> <!-- 详细输出 -->
        <duration-log.statistics value="true"/> <!-- 仿真时长统计 -->
    </report>
    <!-- 如果使用 sumo-gui，可以添加以下配置 -->
    <gui_only> <!-- 仅用于 sumo-gui -->
        <delay value="100"/> <!-- sumo-gui 中每个仿真步之间的延迟 (ms)，用于减慢可视化 -->
        <start value="true"/> <!-- 启动时自动开始仿真 -->
    </gui_only>
    </configuration> <!-- 配置结束 -->
    
    
    xml
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/HxhswoI7zPbgQ1JWivBp9rmeclku.png)

    import os       # 导入os模块，用于操作系统相关功能，如环境变量
    import sys      # 导入sys模块，用于访问Python解释器相关变量和函数
    import traci    # 导入traci库，用于与SUMO进行交互
    import time     # 导入time模块，用于添加延迟等时间相关操作
    
    # SUMO 启动命令，根据你的系统和SUMO安装位置可能需要调整
    # 如果使用 sumo-gui，可以将 "sumo" 替换为 "sumo-gui"
    SUMO_BINARY = "sumo"  # 定义SUMO可执行文件的名称（可以是 "sumo" 或 "sumo-gui"）
    # SUMO_BINARY = "sumo-gui" # 如果想用图形界面，取消这行注释，并注释掉上一行
    
    # 检查 SUMO_HOME 环境变量，并将 tools 目录添加到 Python 路径
    if 'SUMO_HOME' in os.environ:                                           # 检查 'SUMO_HOME' 是否在环境变量中定义
    tools_path = os.path.join(os.environ['SUMO_HOME'], 'tools')         # 构建 SUMO tools 目录的路径
    if tools_path not in sys.path:                                      # 如果 tools 目录不在 Python 的模块搜索路径中
        sys.path.append(tools_path)                                     # 将其添加进去
    else:                                                                   # 如果 'SUMO_HOME' 未定义
    sys.exit("请设置环境变量 'SUMO_HOME' 指向您的SUMO安装目录。")     # 退出脚本并提示用户设置环境变量
    
    # SUMO 配置文件路径
    sumo_config_file = "example.sumocfg"                                    # 定义SUMO配置文件的名称
    
    # TraCI 连接参数
    traci_port = 8813                                                       # 定义TraCI服务器将监听的端口号
    use_libsumo = False                                                      # 是否使用 libsumo (默认为 False，使用 TCP 连接)
    
    def run_simulation():                                                   # 定义一个函数来仿真
    """
    启动SUMO作为服务器，并使用TraCI控制仿真。
    """
    if not use_libsumo:                                                 # 如果不使用 libsumo (即使用标准的 TCP TraCI 连接)
        # 构建 SUMO 启动命令
        sumo_cmd = [
            SUMO_BINARY,                                                # SUMO 可执行文件 (sumo 或 sumo-gui)
            "-c", sumo_config_file,                                     # "-c" 参数后跟配置文件名
            "--remote-port", str(traci_port),                           # "--remote-port" 参数后跟指定的端口号
            "--step-length", "1",                                       # 设置仿真步长为1秒 (也可以在sumocfg中定义)
            "--quit-on-end"                                             # 仿真结束后自动退出SUMO
        ]
        print(f"正在启动 SUMO: {' '.join(sumo_cmd)}")                 # 打印将要执行的SUMO启动命令
        
        # 在一个独立的进程中启动 SUMO
        # 注意：对于更健壮的实现，可以使用 subprocess 模块并更好地管理子进程
        # 这里为了简单，我们假设用户会手动先启动SUMO，或者在另一个终端启动
        # 更推荐的方式是使用 subprocess.Popen 来启动和管理 SUMO 进程
        # import subprocess
        # sumo_proc = subprocess.Popen(sumo_cmd)
        # time.sleep(2) # 等待SUMO启动
    
        # !!! 重要提示: 为了简单起见，此脚本假设SUMO已经由用户在另一个终端手动启动，
        # 或者您将在此Python脚本之前先启动SUMO服务器。
        # 例如，手动打开一个终端并:
        # sumo -c example.sumocfg --remote-port 8813
        # 或者 sumo-gui -c example.sumocfg --remote-port 8813 --start --delay 100
        #
        # 如果希望脚本自动启动SUMO，请使用 subprocess 模块。
        print(f"\n请确保 SUMO 服务器已在端口 {traci_port} 上启动并加载了 {sumo_config_file}") # 提示用户启动SUMO
        print("例如，在另一个终端: sumo -c example.sumocfg --remote-port 8813")        # 给出示例命令
        input("按 Enter键 在确认SUMO已启动后继续...")                                  # 等待用户确认
    
        try:
            # 连接到 SUMO TraCI 服务器
            traci.connect(port=traci_port)                              # 尝试连接到指定端口的TraCI服务器
            print(f"已成功连接到 SUMO (端口: {traci_port})")            # 打印连接成功信息
        except traci.exceptions.TraCIException as e:                    # 如果连接失败，捕获TraCI异常
            print(f"连接 SUMO 失败: {e}")                               # 打印错误信息
            print("请确保SUMO实例正在，并且 --remote-port 设置正确。")# 提示检查SUMO状态和端口
            return                                                      # 结束函数执行
    
    else: # 使用 libsumo
        print("正在使用 libsumo 启动仿真...")                           # 打印使用libsumo的信息
        # 注意：如果使用 libsumo，导入时通常用 import libsumo as traci
        # 但如果 SUMO_HOME/tools 在路径中，原生的 traci 模块也可以被 libsumo "劫持"
        # 为了清晰，假设我们已经 import traci (它可能是 libsumo 的接口)
        traci.start([SUMO_BINARY, "-c", sumo_config_file, "--quit-on-end"]) # 使用traci.start（libsumo方式）启动仿真
        print("libsumo 仿真已启动。")                                   # 打印启动成功信息
    
    # 仿真主循环
    step = 0                                                            # 初始化仿真步数计数器
    max_steps = 360                                                     # 定义最大仿真步数 (与sumocfg中的end时间对应)
    
    try:
        while step < max_steps:                                         # 当仿真步数小于最大步数时循环
            traci.simulationStep()                                      # 执行一个仿真步
            
            current_time = traci.simulation.getTime()                   # 获取当前仿真时间
            vehicle_ids = traci.vehicle.getIDList()                     # 获取当前路网中所有车辆的ID列表
            num_vehicles = len(vehicle_ids)                             # 计算车辆数量
            
            print(f"仿真时间: {current_time:.2f}s, 当前车辆数: {num_vehicles}") # 打印当前时间和车辆数
    
            if num_vehicles > 0 and step % 10 == 0:                     # 如果有车辆并且是每10步
                first_veh_id = vehicle_ids[0]                           # 获取第一辆车的ID
                speed = traci.vehicle.getSpeed(first_veh_id)            # 获取该车的速度
                lane = traci.vehicle.getLaneID(first_veh_id)            # 获取该车所在的车道ID
                print(f"  -> 示例车辆 '{first_veh_id}': 速度={speed:.2f} m/s, 车道='{lane}'") # 打印示例车辆信息
    
            # 在这里可以添加更多的 TraCI 命令来控制或获取信息
            # 例如，改变交通灯相位，改变车辆速度等
            if current_time > 50 and current_time < 100:                # 如果仿真时间在50到100秒之间
                tls_ids = traci.trafficlight.getIDList()                # 获取所有交通信号灯的ID
                if "N2" in tls_ids:                                     # 如果ID为 "N2" 的信号灯存在
                    # 简单示例：强制将N2信号灯的某个相位（例如索引0）持续时间延长
                    # 注意：直接控制相位可能与SUMO内置逻辑冲突，更复杂控制需禁用默认逻辑
                    # traci.trafficlight.setPhase("N2", 0) # 切换到相位0
                    # traci.trafficlight.setPhaseDuration("N2", 30) # 将当前相位持续时间设为30秒
                    pass # 暂时不执行实际的信号灯控制，避免与默认逻辑冲突
    
            step += 1                                                   # 仿真步数加1
            # time.sleep(0.05) # 如果希望减慢Python脚本的执行速度（主要用于观察，不影响SUMO仿真速度）
    
    except traci.exceptions.TraCIException as e:                        # 捕获仿真过程中的TraCI异常
        print(f"仿真过程中发生 TraCI 错误: {e}")                       # 打印错误信息
    except Exception as e:                                              # 捕获其他可能的Python异常
        print(f"仿真过程中发生未知错误: {e}")                           # 打印未知错误信息
    finally:                                                            # 无论成功或失败，最后都执行
        print("正在关闭 TraCI 连接...")                                 # 打印关闭连接信息
        traci.close()                                                   # 关闭与SUMO的TraCI连接
        print("TraCI 连接已关闭。")                                     # 打印连接已关闭信息
        # if not use_libsumo and 'sumo_proc' in locals() and sumo_proc.poll() is None:
        #     print("正在终止 SUMO 进程...")
        #     sumo_proc.terminate()
        #     sumo_proc.wait()
        #     print("SUMO 进程已终止。")
    
    if __name__ == "__main__":                                              # 如果该脚本是主程序
    # 确保 SUMO 配置文件存在
    if not os.path.exists(sumo_config_file):                            # 检查配置文件是否存在
        print(f"错误: SUMO 配置文件 '{sumo_config_file}' 未找到。")    # 打印错误信息
        print("请先创建场景文件 (example.nod.xml, example.edg.xml, example.rou.xml) 并用 netconvert 生成 example.net.xml。") # 提示创建文件
    else:
        run_simulation()                                                # 调用仿真函数
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/JTXY16nr0Dt7NxGhwBEQ5pSH3WvC.png)

   1. 打开一个终端 (我们称之为 **终端1**)。
   2. 在 **终端1** 中，导航到包含所有文件的目录。
   3. 启动 SUMO 服务器。如果你想看图形界面：

        sumo-gui -c example.sumocfg --remote-port 8813 --start --delay 100

        
        
        bash

        sumo -c example.sumocfg --remote-port 8813

        
        
        bash

   4. 打开另一个终端 (我们称之为 **终端2**)。
   5. 在 **终端2** 中，导航到相同的目录。
   6.  Python 脚本中的提示，在 `runner.py` 脚本中，当看到 “按 Enter键 在确认SUMO已启动后继续…” 时，按 Enter。
   7. 然后 Python 脚本：

        python runner.py

        
        
        bash

   1. 修改 `runner.py`，将 `use_libsumo = True`。
   2. (可选) 修改 `SUMO_BINARY`，如果想用 GUI 并且 `libsumo` 支持 (通常是支持的)，可以设为 `"sumo-gui"`。
   3. 直接 Python 脚本：

        python runner.py

        
        
        bash

 * 如果你使用了 `sumo-gui`，你会在图形界面中看到车辆在路网上行驶，信号灯在变化。
 * 在 Python 脚本的终端（终端2 或直接 `libsumo` 时的终端）中，你会看到类似以下的输出：

    正在启动 SUMO: sumo-gui -c example.sumocfg --remote-port 8813 --step-length 1 --quit-on-end (如果未使用libsumo且未手动启动)

    ...
    请确保 SUMO 服务器已在端口 8813 上启动并加载了 example.sumocfg
    例如，在另一个终端: sumo -c example.sumocfg --remote-port 8813
    按 Enter键 在确认SUMO已启动后继续...
    已成功连接到 SUMO (端口: 8813)
    仿真时间: 1.00s, 当前车辆数: 0
    仿真时间: 2.00s, 当前车辆数: 0
    ...
    仿真时间: 10.00s, 当前车辆数: 1
      -> 示例车辆 'flow_east.0': 速度=13.89 m/s, 车道='E1_0'
    仿真时间: 11.00s, 当前车辆数: 1
    ...
    仿真时间: 15.00s, 当前车辆数: 2
    仿真时间: 16.00s, 当前车辆数: 2
    ...
    仿真时间: 20.00s, 当前车辆数: 3
      -> 示例车辆 'flow_east.0': 速度=0.00 m/s, 车道='E1_0' (可能在等红灯)
    ...
    正在关闭 TraCI 连接...
    TraCI 连接已关闭。
    
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/MX3LAWiSrCkexl7OvQ9K4P8jdnVh.png)

    # (假设 traci 已经连接，并且仿真正在)
    # 在仿真循环的某个时间步内：
    # current_sim_time = traci.simulation.getTime() # 获取当前仿真时间
    
    vehicle_ids = traci.vehicle.getIDList() # 获取当前路网中所有活动车辆的ID列表 (返回一个字符串元组)
    print(f"当前时间: {traci.simulation.getTime():.2f}s, 车辆ID列表: {vehicle_ids}") # 打印当前时间和车辆ID列表
    
    if len(vehicle_ids) > 0: # 如果路网中有车辆
    for veh_id in vehicle_ids: # 遍历每一辆车的ID
        try:
            # 获取位置 (x, y 坐标)
            pos_x, pos_y = traci.vehicle.getPosition(veh_id) # 获取车辆veh_id的笛卡尔坐标(x,y)
            # SUMO 内部坐标系，通常原点在路网的某个特定位置
    
            # 获取地理位置 (经纬度，如果路网有地理参考)
            # lon, lat = traci.vehicle.getPosition3D(veh_id) # 如果路网是地理参考的，这会返回(lon, lat, z)
            # 或者使用 traci.simulation.convertGeo(pos_x, pos_y, fromGeo=False) 转换为经纬度
            # 如果路网本身就是用经纬度定义的，getPosition 可能直接反映一定程度的地理位置
            
            speed = traci.vehicle.getSpeed(veh_id) # 获取车辆veh_id的当前速度 (m/s)
            
            angle = traci.vehicle.getAngle(veh_id) # 获取车辆veh_id的行驶角度 (度)，0度沿X轴正向，逆时针增加
                                                 # 这个角度是车辆几何中心的朝向
            
            road_id = traci.vehicle.getRoadID(veh_id) # 获取车辆veh_id当前所在的边的ID
            
            lane_id = traci.vehicle.getLaneID(veh_id) # 获取车辆veh_id当前所在的车道的ID
            
            lane_index = traci.vehicle.getLaneIndex(veh_id) # 获取车辆veh_id当前所在车道在其所在边上的索引 (从0开始，最右侧为0)
            
            lane_position = traci.vehicle.getLanePosition(veh_id) # 获取车辆veh_id在其当前车道上的位置 (从车道起点开始计算的距离，m)
            
            vehicle_type = traci.vehicle.getTypeID(veh_id) # 获取车辆veh_id的类型ID (例如 "car_eastbound")
            
            route_id = traci.vehicle.getRouteID(veh_id) # 获取车辆veh_id当前使用的路径ID
            
            current_route_edges = traci.vehicle.getRoute(veh_id) # 获取车辆veh_id当前剩余路径的边列表 (一个字符串元组)
            
            # 累积等待时间 (车辆速度低于某一阈值时开始累积)
            waiting_time = traci.vehicle.getWaitingTime(veh_id) # 获取车辆veh_id的总累积等待时间 (s)
            
            # 车辆行驶的总距离
            distance_travelled = traci.vehicle.getDistance(veh_id) # 获取车辆veh_id从出发开始已行驶的总距离 (m)
    
            # 获取车辆的长度
            length = traci.vehicle.getLength(veh_id) # 获取车辆veh_id的长度 (m)
    
            # 获取车辆颜色 (RGBA格式，0-255)
            color = traci.vehicle.getColor(veh_id) # 获取车辆veh_id的颜色，返回一个 (R, G, B, Alpha) 元组
    
            print(f"  车辆ID: {veh_id}") # 打印车辆ID
            print(f"    位置: ({pos_x:.2f}, {pos_y:.2f})") # 打印车辆位置，保留两位小数
            print(f"    速度: {speed:.2f} m/s") # 打印车辆速度，保留两位小数
            print(f"    角度: {angle:.2f} 度") # 打印车辆角度，保留两位小数
            print(f"    所在边: {road_id}, 车道: {lane_id} (索引 {lane_index})") # 打印车辆所在边、车道和车道索引
            print(f"    车道内位置: {lane_position:.2f} m") # 打印车辆在车道内的位置，保留两位小数
            print(f"    类型: {vehicle_type}, 路径ID: {route_id}") # 打印车辆类型和路径ID
            print(f"    剩余路径: {current_route_edges}") # 打印车辆剩余路径
            print(f"    累积等待时间: {waiting_time:.2f} s") # 打印车辆累积等待时间，保留两位小数
            print(f"    已行驶距离: {distance_travelled:.2f} m") # 打印车辆已行驶距离，保留两位小数
            print(f"    长度: {length:.2f} m, 颜色: {color}") # 打印车辆长度和颜色
    
        except traci.exceptions.TraCIException as e: # 捕获 TraCI 异常 (例如，车辆可能在上一个时间步离开了路网)
            print(f"获取车辆 {veh_id} 信息时出错: {e}") # 打印错误信息
            # 这种情况可能发生在该车辆刚好在这个时间步之前离开了仿真区域
            # 或者ID不再有效
            pass # 继续处理下一辆车
    else:
    print("  路网中当前没有车辆。") # 如果路网中没有车辆，打印提示信息
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/iUtkH26TxPNmQ70fAh4IuG5cSnly.png)

    # (假设 traci 已经连接，并且仿真正在)
    # 在仿真循环的某个时间步内：
    vehicle_ids = traci.vehicle.getIDList() # 获取当前路网中所有活动车辆的ID列表
    
    if "flow_east.0" in vehicle_ids: # 假设我们知道一个特定的车辆ID，例如 "flow_east.0"
    veh_id_to_control = "flow_east.0" # 定义要控制的车辆ID
    
    try:
        # 1. 修改车辆速度 (影响期望速度，实际速度受跟驰模型等限制)
        target_speed = 10.0  # 设置目标速度为 10 m/s (约 36 km/h)
        traci.vehicle.setSpeed(veh_id_to_control, target_speed) # 为车辆 veh_id_to_control 设置新的期望速度
        print(f"  车辆 {veh_id_to_control}: 已设置期望速度为 {target_speed} m/s") # 打印设置速度的信息
    
        # setSpeedMode: 控制车辆如何遵守速度限制和安全距离等规则
        # mode=0: 忽略所有安全检查 (非常危险，车辆可能碰撞)
        # mode=31 (默认): 遵守所有检查 (限速、安全距离、红灯等)
        # traci.vehicle.setSpeedMode(veh_id_to_control, 0) # 例如，让车辆忽略安全检查 (不推荐)
        # traci.vehicle.setSpeedMode(veh_id_to_control, 31) # 恢复默认安全模式
    
        # 2. 修改车辆目标边 (会触发到该新目标边的路径重规划)
        # 注意：这会改变车辆当前的 "下一个主要目标边"，并可能导致其最终目的地改变
        # 如果只是想改变路径中的下一条边，而最终目的地不变，使用 rerouteTraveltime 或 setRoute
        current_route = traci.vehicle.getRoute(veh_id_to_control) # 获取车辆当前路径
        if len(current_route) > 1: # 如果路径中至少还有两条边
            # 假设我们想让它跳过下一条边，直接去下下条边 (如果合法)
            # 这是一个简化的例子，实际应用中需要更复杂的逻辑
            # new_target_edge = current_route[1] # 获取当前路径的第二条边作为新目标
            # traci.vehicle.changeTarget(veh_id_to_control, new_target_edge) # 改变车辆的目标边
            # print(f"  车辆 {veh_id_to_control}: 目标边已改为 {new_target_edge}")
            pass # 暂时不执行，因为简单改变目标可能导致不合理路径
    
        # 3. 重新规划路径到新的最终目的地
        # 假设我们想让车辆开往 "E4" 这条边 (假设 "E4" 存在且可达)
        new_destination_edge = "E4" # 定义新的目的地边
        if new_destination_edge in traci.edge.getIDList(): # 检查目的地边是否存在
            # rerouteTraveltime 会让车辆基于当前的行程时间信息重新规划到新目的地的最快路径
            traci.vehicle.changeTarget(veh_id_to_control, new_destination_edge) # 改变车辆的最终目的地边
            # 注意: changeTarget 会使车辆寻找一条到达新目标边的路径。
            # 如果想立即改变当前行驶的边，用 setRoute 更合适。
            # SUMO 会自动为车辆找到一条通往新目的地的路径
            print(f"  车辆 {veh_id_to_control}: 已将最终目的地边改为 {new_destination_edge}，将自动重规划路径。") # 打印信息
            # 为了确保路径立即更新并被车辆采用，有时需要检查 traci.vehicle.isRouteValid()
            # 并且如果需要，强制使用 setRoute 设置一条已知的有效路径。
        else:
            print(f"  警告: 目标边 {new_destination_edge} 不在路网中，无法为 {veh_id_to_control} 改变目的地。") # 打印警告
    
        # 4. 设置一条全新的路径 (边的列表)
        # 这会立即覆盖车辆的当前路径
        new_route_edges = ("E1", "E4") # 定义一条新路径 (假设 E1 -> (交叉口) -> E4 是可能的)
                                    # 注意: 这条路径可能在我们的示例路网中不是直接相连的或合法的。
                                    # 确保提供的路径是连续且有效的。
        try:
            # 检查路径有效性 (SUMO本身在setRoute时也会检查，但提前检查更好)
            # is_valid_path_check = True # 假设路径是有效的
            # for i in range(len(new_route_edges) -1 ):
            #    if new_route_edges[i+1] not in traci.edge.getSuccessors(new_route_edges[i]): # 伪代码
            #        is_valid_path_check = False; break
            
            # traci.vehicle.setRoute(veh_id_to_control, new_route_edges) # 为车辆设置一条全新的路径
            # print(f"  车辆 {veh_id_to_control}: 已设置新路径为 {new_route_edges}") # 打印信息
            pass # 实际使用 setRoute 需要非常小心路径的有效性。
    
        except traci.exceptions.TraCIException as e_route: # 捕获设置路径时可能发生的错误
            print(f"  为车辆 {veh_id_to_control} 设置新路径 {new_route_edges} 时出错: {e_route}") # 打印错误
            print(f"  请确保路径中的边是连续且车辆允许通行的。") # 提示检查路径
    
        # 5. 改变车辆类型
        available_vtypes = traci.vehicletype.getIDList() # 获取所有已定义的车辆类型ID
        if "car_northbound" in available_vtypes: # 检查 "car_northbound" 类型是否存在
            traci.vehicle.setType(veh_id_to_control, "car_northbound") # 将车辆类型改为 "car_northbound"
            print(f"  车辆 {veh_id_to_control}: 类型已改为 'car_northbound'") # 打印类型更改信息
            # 车辆的物理属性（如长度、最大速度）和行为模型会随之改变
    
        # 6. 改变车辆颜色
        new_color = (0, 255, 0, 255)  # 设置为绿色 (R, G, B, Alpha)，Alpha=255为不透明
        traci.vehicle.setColor(veh_id_to_control, new_color) # 为车辆设置新颜色
        print(f"  车辆 {veh_id_to_control}: 颜色已改为 {new_color}") # 打印颜色更改信息
    
        # 7. 让车辆执行紧急停车 (在指定车道、位置、持续时间)
        # current_lane = traci.vehicle.getLaneID(veh_id_to_control) # 获取当前车道
        # stop_duration = 30 # 停车30秒
        # traci.vehicle.setStop(vehID=veh_id_to_control, edgeID=traci.lane.getEdgeID(current_lane), 
        #                       pos=traci.vehicle.getLanePosition(veh_id_to_control) + 5, # 停在当前位置前方5米
        #                       laneIndex=traci.vehicle.getLaneIndex(veh_id_to_control), 
        #                       duration=stop_duration, flags=traci.constants.STOP_DEFAULT)
        # print(f"  车辆 {veh_id_to_control}: 已设置在当前车道前方紧急停车 {stop_duration} 秒。")
        # flags 可以是 traci.constants.STOP_PARKING (模拟路边停车), traci.constants.STOP_TRIGGERED (等待行人等)
    
        # 8. 恢复车辆正常行驶 (如果之前执行了 setStop)
        # traci.vehicle.resume(veh_id_to_control)
        # print(f"  车辆 {veh_id_to_control}: 已恢复正常行驶。")
    
        # 9. 强制车辆换道
        # target_lane_index = traci.vehicle.getLaneIndex(veh_id_to_control) - 1 # 尝试向左换道（索引减1）
        # if target_lane_index >= 0: # 确保目标车道索引有效
        #    # changeLane(vehID, laneIndex, duration)
        #    # duration 是车辆尝试完成换道的时间，如果为0，则表示“尽可能快”
        #    traci.vehicle.changeLane(veh_id_to_control, target_lane_index, 10) # 尝试在10秒内换到左侧车道
        #    print(f"  车辆 {veh_id_to_control}: 正在尝试向左换道到索引 {target_lane_index}")
        # 注意：换道是否成功取决于交通状况和换道模型的逻辑。
    
        # 10. 移动车辆到特定位置 (Move To - 慎用)
        # 这个命令会立即将车辆“传送”到新位置，可能破坏仿真物理真实性，主要用于调试或特殊场景。
        # target_lane_id = "E2_0" # 假设E2_0是一个有效的车道ID
        # target_position_on_lane = 10.0 # 距离车道起点10米
        # if target_lane_id in traci.lane.getIDList():
        #    try:
        #        traci.vehicle.moveTo(veh_id_to_control, target_lane_id, target_position_on_lane)
        #        print(f"  车辆 {veh_id_to_control}: 已被移动到车道 {target_lane_id} 的 {target_position_on_lane}m 位置。")
        #    except traci.exceptions.TraCIException as e_move:
        #        print(f"  移动车辆 {veh_id_to_control} 出错: {e_move}")
        pass # moveTo 破坏性较大，一般不用于标准仿真逻辑
    
    except traci.exceptions.TraCIException as e: # 捕获对该特定车辆操作时的TraCI错误
        print(f"  操作车辆 {veh_id_to_control} 时出错: {e}") # 打印错误信息
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/Y7Nva3jAVyFipKUfotmnb0ZsXhxM.png)

    # (假设 traci 已经连接，并且仿真正在)
    # current_sim_time = traci.simulation.getTime() # 获取当前仿真时间
    
    # 1. 动态添加车辆
    if current_sim_time == 100.0: # 假设在仿真时间100秒时添加一辆车
    new_veh_id = f"dynamic_veh_{current_sim_time}" # 创建一个唯一的车辆ID，例如基于当前时间
    route_id_for_new_veh = "route_east" # 指定新车辆要使用的路径ID (需已在.rou.xml或通过TraCI定义)
    vehicle_type_for_new_veh = "car_eastbound" # 指定新车辆的类型ID (需已定义)
    
    try:
        # 检查路径ID是否存在
        if route_id_for_new_veh not in traci.route.getIDList(): # 获取所有已定义的路径ID列表并检查
            # 如果路径不存在，可以动态添加它
            # 假设 route_east_edges 是一个边的列表，例如 ("E1", "E2")
            # traci.route.add(route_id_for_new_veh, route_east_edges)
            # print(f"  动态添加了路径ID: {route_id_for_new_veh}")
            pass # 在此示例中，我们假设 route_east 已存在
    
        # traci.vehicle.add(vehID, routeID, typeID="DEFAULT_VEHTYPE", depart=-3, departLane="first", 
        #                   departPos="base", departSpeed="0", arrivalLane="current", arrivalPos="max", 
        #                   arrivalSpeed="current", fromTaz="", toTaz="", line="", personCapacity=0, personNumber=0)
        # depart=-3 (MAGIC_DEPART_TIME_TRIGGERED): 车辆会尝试在下一个可能的时间步出发
        # departLane: "first", "last", "random", "free", "allowed", "best", 或车道索引
        # departPos: "base" (车道起点), "last" (车道末端，刚好能放下车身), "random", "random_free", 或数值
        # departSpeed: "max", "random", "desired", 或数值
        traci.vehicle.add(
            vehID=new_veh_id,
            routeID=route_id_for_new_veh,
            typeID=vehicle_type_for_new_veh,
            depart="triggered",  # 或 depart=str(current_sim_time) 如果希望精确控制
                                 # "triggered" 或 -3 表示尽快出发
            departLane="random", # 在随机车道出发
            departPos="base",    # 在车道起点出发
            departSpeed="max"    # 以最大速度出发
        )
        print(f"  在时间 {current_sim_time:.2f}s: 动态添加了车辆 {new_veh_id} (类型: {vehicle_type_for_new_veh}, 路径: {route_id_for_new_veh})") # 打印添加车辆信息
    except traci.exceptions.TraCIException as e: # 捕获添加车辆时可能发生的错误
        print(f"  动态添加车辆 {new_veh_id} 时出错: {e}") # 打印错误信息
        print(f"  请确保路径ID '{route_id_for_new_veh}' 和车辆类型ID '{vehicle_type_for_new_veh}' 已定义且有效。") # 提示检查路径和类型
    
    
    # 2. 动态移除车辆
    vehicle_ids = traci.vehicle.getIDList() # 获取当前车辆列表
    if len(vehicle_ids) > 5: # 如果车辆数量超过5辆
    veh_to_remove = vehicle_ids[0] # 选择第一辆车进行移除 (可以是任何选择逻辑)
    try:
        traci.vehicle.remove(veh_to_remove) # 从仿真中移除车辆 veh_to_remove
        print(f"  在时间 {current_sim_time:.2f}s: 动态移除了车辆 {veh_to_remove}") # 打印移除车辆信息
    except traci.exceptions.TraCIException as e: # 捕获移除车辆时可能发生的错误
        print(f"  动态移除车辆 {veh_to_remove} 时出错: {e}") # 打印错误信息
    
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/pRS9TLwsJKcWiZbM2yChmu70EdAF.png)

    # (假设 traci 已经连接，并且仿真正在)
    # current_sim_time = traci.simulation.getTime() # 获取当前仿真时间
    vehicle_ids = traci.vehicle.getIDList() # 获取车辆列表
    
    if len(vehicle_ids) > 0: # 如果有车辆
    veh_id_of_interest = vehicle_ids[0] # 选择第一辆车作为我们关注的车辆
    
    try:
        # 1. 获取前车信息 (在同一车道)
        # getLeader(vehID, dist) 返回一个元组 (leaderID, leaderDistToVehFront)
        # leaderID: 前车的ID (字符串)，如果没前车则为 None
        # leaderDistToVehFront: 前车尾部到本车头部的距离 (m)，如果没前车则为浮点数最大值或一个非常大的值
        # dist: 查找前车的最大向前搜索距离 (m)，如果为0或负数，则表示只查找紧邻的前车。
        #       通常设为0即可获取直接前车。
        leader_info = traci.vehicle.getLeader(veh_id_of_interest, 0) # 获取紧邻的前车信息
        if leader_info and leader_info[0] is not None: # 检查是否有前车 (leader_info[0] 是前车ID)
            leader_id = leader_info[0] # 前车ID
            distance_to_leader_front_bumper = leader_info[1] # 本车车头到前车车尾的间距
            leader_speed = traci.vehicle.getSpeed(leader_id) # 获取前车的速度
            print(f"  车辆 {veh_id_of_interest}: 前车ID='{leader_id}', "
                  f"与前车间距={distance_to_leader_front_bumper:.2f}m, 前车速度={leader_speed:.2f}m/s") # 打印前车信息
        else:
            print(f"  车辆 {veh_id_of_interest}: 当前车道前方无紧邻车辆。") # 打印无前车信息
    
        # 2. 获取邻近车道的前车和后车信息 (用于换道决策)
        # getNeighbors(vehID, mode)
        # mode: 一个位掩码，指定要查找的邻居类型
        #       traci.constants.NEIGHBOR_LEFT (1)
        #       traci.constants.NEIGHBOR_RIGHT (2)
        #       traci.constants.NEIGHBOR_FRONT (4) - (已废弃, 用 getLeader)
        #       traci.constants.NEIGHBOR_REAR (8) - (已废弃, 用 getFollower)
        #       NEIGHBOR_LEFT_FRONT (1 | 4)
        #       NEIGHBOR_LEFT_REAR (1 | 8)
        #       NEIGHBOR_RIGHT_FRONT (2 | 4)
        #       NEIGHBOR_RIGHT_REAR (2 | 8)
        # 返回一个元组列表，每个元组是 (neighborVehID, neighborDist, neighborState)
        #   neighborDist: 与该邻居的距离 (具体含义取决于mode)
        #   neighborState: 邻居的状态 (位掩码，例如是否正在刹车)
        
        # 查找左侧车道的前方和后方车辆 (更现代的 SUMO 版本推荐使用更具体的函数)
        # lane_change_state = traci.vehicle.getLaneChangeState(veh_id_of_interest, traci.constants.LANECHANGE_TARGET_LEFT)
        # lane_change_state_right = traci.vehicle.getLaneChangeState(veh_id_of_interest, traci.constants.LANECHANGE_TARGET_RIGHT)
        # lane_change_state 是一个元组，包含很多关于换道可行性的信息，例如：
        # (stateBits, safeLeft, safeRight, safeLeftOpposite, safeRightOpposite, ...)
        # stateBits & traci.constants.LCA_LEFT == traci.constants.LCA_LEFT 表示可以向左换道
    
        # 获取左侧相邻车道的信息
        left_leader_id, left_follower_id = None, None
        # getBestLanes(vehID) 返回一个包含元组的列表，每个元组描述一条候选换道车道的信息
        # (laneID, length, occupation, bestSpeed, offset, allowsContinuation, nextLanes)
        # 这个函数更侧重于整体的换道选择，而非直接的邻车ID
        
        # 一个更直接的方式可能是先判断本车左右是否有车道，然后获取那些车道上的车辆
        current_lane_id = traci.vehicle.getLaneID(veh_id_of_interest) # 获取当前车道ID
        current_edge_id = traci.lane.getEdgeID(current_lane_id) # 获取当前边ID
        current_lane_index = traci.vehicle.getLaneIndex(veh_id_of_interest) # 获取当前车道索引
        num_lanes_on_edge = traci.edge.getLaneNumber(current_edge_id) # 获取当前边上的车道总数
    
        # 检查左侧车道
        if current_lane_index > 0: # 如果当前不是最左侧车道 (索引从右到左增加，或从左到右，取决于路网定义，通常0是最右)
                                   # 假设索引从右到左增加 (如SUMO默认)，则左侧车道索引为 current_lane_index - 1
                                   # 但SUMO文档中 lane index 0 is the rightmost one.
            left_lane_index = current_lane_index -1 # 如果当前车道索引为1 (第二条道), 左侧是索引0 (最右) -- 这不对
                                                    # 如果索引0是最右，那么左边的车道索引是 current_lane_index + 1
            
            # 需要确定车道索引的方向。SUMO标准是：lane index 0 is the rightmost one.
            # 所以，左边的车道是 target_lane_index = current_lane_index + 1
            # 右边的车道是 target_lane_index = current_lane_index - 1
            
            # 尝试获取左侧车道 (索引 current_lane_index + 1)
            target_left_lane_index = current_lane_index + 1 # 目标左侧车道的索引
            if target_left_lane_index < num_lanes_on_edge: # 确保左侧车道索引在范围内
                left_lane_id = f"{current_edge_id}_{target_left_lane_index}" # 构建左侧车道的ID
                # 获取左侧车道上的所有车辆ID
                # vehicles_on_left_lane = traci.lane.getLastStepVehicleIDs(left_lane_id)
                # print(f"    左侧车道 ({left_lane_id}) 上的车辆: {vehicles_on_left_lane}")
                
                # 使用 getLeftFollowers 和 getLeftLeaders (SUMO 1.1.0+)
                # (vehID, dist, relSpeed)
                left_followers = traci.vehicle.getLeftFollowers(veh_id_of_interest) # 获取左后方车辆列表
                left_leaders = traci.vehicle.getLeftLeaders(veh_id_of_interest)     # 获取左前方车辆列表
    
                if left_leaders: # 如果左前方有车
                    # left_leaders 是一个列表，每个元素是 (leaderID, distance, relSpeed, currentDist (到交叉口))
                    # distance 是到本车前端的纵向距离，负数表示在本车前方，正数表示在本车后方
                    # relSpeed 是相对速度 (leaderSpeed - egoSpeed)
                    closest_left_leader = min(left_leaders, key=lambda x: abs(x[1])) if left_leaders else None
                    if closest_left_leader:
                        print(f"    左前方车辆: ID={closest_left_leader[0]}, 纵向距离={closest_left_leader[1]:.2f}m, 相对速度={closest_left_leader[2]:.2f}m/s") # 打印信息
                if left_followers: # 如果左后方有车
                    closest_left_follower = min(left_followers, key=lambda x: abs(x[1])) if left_followers else None
                    if closest_left_follower:
                        print(f"    左后方车辆: ID={closest_left_follower[0]}, 纵向距离={closest_left_follower[1]:.2f}m, 相对速度={closest_left_follower[2]:.2f}m/s") # 打印信息
            else:
                print(f"    车辆 {veh_id_of_interest} 已在最左侧车道或左侧无更多车道。") # 打印信息
    
        # 检查右侧车道 (索引 current_lane_index - 1)
        target_right_lane_index = current_lane_index - 1 # 目标右侧车道的索引
        if target_right_lane_index >= 0: # 确保右侧车道索引有效
            # right_lane_id = f"{current_edge_id}_{target_right_lane_index}" # 构建右侧车道的ID
            # vehicles_on_right_lane = traci.lane.getLastStepVehicleIDs(right_lane_id)
            # print(f"    右侧车道 ({right_lane_id}) 上的车辆: {vehicles_on_right_lane}")
    
            right_followers = traci.vehicle.getRightFollowers(veh_id_of_interest) # 获取右后方车辆列表
            right_leaders = traci.vehicle.getRightLeaders(veh_id_of_interest)     # 获取右前方车辆列表
    
            if right_leaders: # 如果右前方有车
                closest_right_leader = min(right_leaders, key=lambda x: abs(x[1])) if right_leaders else None
                if closest_right_leader:
                    print(f"    右前方车辆: ID={closest_right_leader[0]}, 纵向距离={closest_right_leader[1]:.2f}m, 相对速度={closest_right_leader[2]:.2f}m/s") # 打印信息
            if right_followers: # 如果右后方有车
                closest_right_follower = min(right_followers, key=lambda x: abs(x[1])) if right_followers else None
                if closest_right_follower:
                    print(f"    右后方车辆: ID={closest_right_follower[0]}, 纵向距离={closest_right_follower[1]:.2f}m, 相对速度={closest_right_follower[2]:.2f}m/s") # 打印信息
        else:
            print(f"    车辆 {veh_id_of_interest} 已在最右侧车道。") # 打印信息
    
        # 3. 获取车辆的跟驰模型参数 (如果需要深入分析或修改行为)
        # 这些参数通常在 vType 中定义，但也可以通过 TraCI 获取
        # 例如，获取车辆的最小安全间隙 (minGap)
        min_gap = traci.vehicle.getMinGap(veh_id_of_interest) # 获取车辆的最小安全车头时距 (s) 或距离 (m) (取决于跟驰模型)
        # print(f"    车辆 {veh_id_of_interest}: 最小安全间隙 (minGap) = {min_gap:.2f}")
    
        # 4. 获取车辆的加速度
        acceleration = traci.vehicle.getAcceleration(veh_id_of_interest) # 获取车辆当前加速度 (m/s^2)
        # print(f"    车辆 {veh_id_of_interest}: 当前加速度 = {acceleration:.2f} m/s^2")
    
    
    except traci.exceptions.TraCIException as e: # 捕获 TraCI 异常
        print(f"  在为车辆 {veh_id_of_interest} 获取环境信息时出错: {e}") # 打印错误信息
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/otU2pG5Ydz8Cgbw301iK6AaVQy7D.png)

    # (假设 traci 已经连接，并且仿真正在)
    # current_sim_time = traci.simulation.getTime() # 获取当前仿真时间
    
    tls_ids = traci.trafficlight.getIDList() # 获取路网中所有交通信号灯系统的ID列表 (字符串元组)
    print(f"当前时间: {traci.simulation.getTime():.2f}s, 信号灯系统ID列表: {tls_ids}") # 打印信息
    
    if len(tls_ids) > 0: # 如果有信号灯系统
    for tls_id in tls_ids: # 遍历每个信号灯系统
        try:
            print(f"  信号灯系统ID: {tls_id}") # 打印信号灯ID
    
            # 获取当前信号灯状态 (一个字符串，表示每个受控连接的灯色)
            # 例如 "rryyGGggrryyGGgg" (r=red, y=yellow, g=green, G=green with priority, s=blinking yellow, o=off)
            # 字符串的顺序对应于信号灯定义中连接的顺序
            current_state_string = traci.trafficlight.getRedYellowGreenState(tls_id) # 获取当前信号灯状态字符串
            print(f"    当前状态字符串: {current_state_string}") # 打印状态字符串
    
            # 获取当前活动的相位索引 (在当前程序中)
            current_phase_index = traci.trafficlight.getPhase(tls_id) # 获取当前相位索引
            print(f"    当前相位索引: {current_phase_index}") # 打印相位索引
    
            # 获取当前信号灯程序ID
            current_program_id = traci.trafficlight.getProgram(tls_id) # 获取当前使用的信号灯程序ID
            print(f"    当前程序ID: '{current_program_id}'") # 打印程序ID
    
            # 获取当前相位的剩余持续时间 (如果信号灯是固定周期或基于时间的)
            # 注意: 对于 TraCI 控制的信号灯，这个值可能不那么直接，因为它取决于外部逻辑何时改变相位
            # next_switch_time = traci.trafficlight.getNextSwitch(tls_id) # 获取下一次自动切换的仿真时间
            # time_to_next_switch = next_switch_time - current_sim_time
            # print(f"    下一次自动切换时间: {next_switch_time:.2f}s (剩余 {time_to_next_switch:.2f}s)")
    
            # 获取信号灯控制的所有车道
            controlled_lanes = traci.trafficlight.getControlledLanes(tls_id) # 获取该信号灯控制的所有车道ID列表
            print(f"    控制的车道: {controlled_lanes}") # 打印受控车道列表
    
            # 获取信号灯控制的所有连接
            controlled_links_raw = traci.trafficlight.getControlledLinks(tls_id) # 获取该信号灯控制的所有连接信息
            # controlled_links_raw 是一个元组列表，每个内部元组代表一个受控连接
            # 每个内部元组包含 (incomingLaneID, outgoingLaneID, viaLaneID_or_internalLaneID)
            # 这是一个比较底层的表示，更常用的是getRedYellowGreenState来理解灯色
            # print(f"    控制的连接 (原始): {controlled_links_raw}")
    
            # 获取完整的信号灯逻辑定义 (包括所有相位)
            # 返回一个包含所有程序定义的元组，每个程序是一个 traci.trafficlight.Logic 对象
            # 每个 Logic 对象包含 phase 列表，每个 phase 是 traci.trafficlight.Phase 对象
            # Phase 对象有 duration, minDur, maxDur, state (灯色字符串) 等属性
            all_logic = traci.trafficlight.getCompleteRedYellowGreenDefinition(tls_id) # 获取完整的信号灯逻辑定义
            if all_logic and len(all_logic) > 0: # 如果有逻辑定义
                program_logic = all_logic[0] # 获取第一个程序 (通常是默认程序，除非有多个程序)
                print(f"    程序 '{program_logic.programID}' (子ID: {program_logic.subID}) 有 {len(program_logic.phases)} 个相位:") # 打印程序信息
                for i, phase in enumerate(program_logic.phases): # 遍历每个相位
                    print(f"      相位 {i}: 持续时间={phase.duration:.1f}s, "
                          f"最小={phase.minDur:.1f}s, 最大={phase.maxDur:.1f}s, 状态='{phase.state}'") # 打印相位详细信息
            
        except traci.exceptions.TraCIException as e: # 捕获 TraCI 异常
            print(f"  获取信号灯 {tls_id} 信息时出错: {e}") # 打印错误信息
    else:
    print("  路网中当前没有交通信号灯系统。") # 打印无信号灯信息
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/KwP0qGQkUsc2954lJL6H8DjOgiVS.png)

    # (假设 traci 已经连接，并且仿真正在)
    # current_sim_time = traci.simulation.getTime() # 获取当前仿真时间
    tls_id_to_control = "N2" # 假设我们要控制ID为 "N2" 的信号灯 (在我们的示例场景中存在)
    
    if tls_id_to_control in traci.trafficlight.getIDList(): # 检查信号灯是否存在
    try:
        # 1. 切换到指定的相位索引
        # 假设我们知道 N2 的信号灯逻辑，并且想切换到它的第2个相位 (索引为1)
        target_phase_index = 1 # 定义目标相位索引
        # 在切换之前，最好获取当前程序的所有相位，确保 target_phase_index 是有效的
        all_logic_n2 = traci.trafficlight.getCompleteRedYellowGreenDefinition(tls_id_to_control) # 获取N2的完整逻辑
        if all_logic_n2 and len(all_logic_n2[0].phases) > target_phase_index: # 检查目标相位索引是否有效
            traci.trafficlight.setPhase(tls_id_to_control, target_phase_index) # 设置信号灯到指定的相位索引
            print(f"  信号灯 {tls_id_to_control}: 已切换到相位索引 {target_phase_index}") # 打印切换信息
            current_phase_state = traci.trafficlight.getRedYellowGreenState(tls_id_to_control) # 获取切换后的状态
            print(f"    切换后的状态: {current_phase_state}") # 打印新状态
        else:
            print(f"  信号灯 {tls_id_to_control}: 目标相位索引 {target_phase_index} 无效。") # 打印无效索引信息
    
        # 2. 设置当前相位的持续时间
        # 这会覆盖当前相位在预定义逻辑中的标准时长
        # 如果信号灯完全由TraCI控制，这个命令决定了当前相位会持续多久，直到TraCI再次改变它
        new_phase_duration = 25.0 # 设置当前相位持续25秒
        # traci.trafficlight.setPhaseDuration(tls_id_to_control, new_phase_duration)
        # print(f"  信号灯 {tls_id_to_control}: 当前相位的持续时间已尝试设置为 {new_phase_duration}s")
        # 注意: setPhaseDuration 的行为可能取决于信号灯是否完全由TraCI控制。
        # 如果SUMO仍在尝试执行其内置逻辑，可能会有冲突。
        # 对于完全的TraCI控制，通常是在每个仿真步检查是否需要切换相位，然后使用setPhase。
    
        # 3. 切换到不同的信号灯程序 (如果定义了多个程序)
        # target_program_id = "my_peak_hour_program" # 假设存在名为 "my_peak_hour_program" 的程序
        # if target_program_id in [prog.programID for prog in all_logic_n2]: # 检查程序ID是否存在
        #    traci.trafficlight.setProgram(tls_id_to_control, target_program_id)
        #    print(f"  信号灯 {tls_id_to_control}: 已切换到程序 '{target_program_id}'")
        # else:
        #    print(f"  信号灯 {tls_id_to_control}: 程序ID '{target_program_id}' 未找到。")
    
        # 4. 完全由 TraCI 定义和控制信号灯逻辑 (最灵活的方式)
        # 这通常涉及到在每个仿真步：
        #   a. 收集交通数据 (如排队长度、等待时间)。
        #   b. 根据自定义算法 (如自适应控制、强化学习) 决定下一个应该激活的相位。
        #   c. 使用 traci.trafficlight.setRedYellowGreenState(tlsID, stateString) 直接设置灯色。
        #      或者，如果想保持相位的概念，但由TraCI决定切换时机和下一个相位：
        #      traci.trafficlight.setPhase(tlsID, nextPhaseIndex)
    
        # 示例：强制设置一个特定的灯色状态 (假设我们知道 N2 信号灯状态字符串的结构)
        # 例如，让主干道 (E1->E2) 绿灯，其他方向红灯
        # 这个 state_string 的具体格式和长度取决于 netconvert 如何为该交叉口生成连接和信号灯。
        # 需要通过 getRedYellowGreenState() 或 getCompleteRedYellowGreenDefinition() 来确定其结构。
        # 假设对于我们的 N2 交叉口，一个东西向绿灯的状态可能是 "GGgrrrGGgrrr" (长度和具体字符需确认)
        
        # 查找东西向绿灯的相位状态字符串
        east_west_green_state = None
        if all_logic_n2: # 如果N2有逻辑定义
            for phase_def in all_logic_n2[0].phases: # 遍历第一个程序的所有相位
                # 这里需要一种方式来识别哪个相位对应东西向绿灯
                # 例如，通过观察 phase_def.state 字符串的模式
                # 假设包含 "GGg" 的是东西向绿灯，并且没有太多 'r'
                # 这是一个启发式的方法，实际中需要更精确的相位识别
                if "GGg" in phase_def.state and phase_def.state.count('r') < len(phase_def.state) / 2:
                    east_west_green_state = phase_def.state # 获取该相位的状态字符串
                    print(f"  找到一个可能是东西向绿灯的相位状态: '{east_west_green_state}'") # 打印找到的状态
                    break # 找到后退出循环
        
        if east_west_green_state and current_sim_time > 60 and current_sim_time < 90 : # 假设在60-90秒间强制东西绿灯
            current_tls_state_str = traci.trafficlight.getRedYellowGreenState(tls_id_to_control) # 获取当前状态
            if current_tls_state_str != east_west_green_state: # 如果当前状态不是我们想要的绿灯状态
                traci.trafficlight.setRedYellowGreenState(tls_id_to_control, east_west_green_state) # 直接设置信号灯状态字符串
                print(f"  信号灯 {tls_id_to_control}: 在时间 {current_sim_time:.1f}s, 强制设置为状态 '{east_west_green_state}'") # 打印强制设置信息
        
        # **重要提示关于 setRedYellowGreenState**:
        # - 当你使用 setRedYellowGreenState 时，你实际上是在覆盖SUMO的内置信号灯逻辑或当前的程序。
        # - SUMO将保持你设置的状态，直到你再次通过TraCI命令改变它。
        # - 你需要自己处理黄灯过渡和最小绿灯时间等安全要求，如果你不希望出现不安全的信号切换。
        # - 或者，更常见的方式是使用 setPhase(tlsID, phaseIndex) 并在外部逻辑中管理相位的持续时间和切换。
        #   这样可以利用SUMO预定义的相位（包括黄灯和全红清空时间），而TraCI只决定何时切换到哪个相位。
    
        # 5. 修改信号灯程序的相位定义 (例如，改变某个相位的时长)
        # 这比较高级，需要先获取完整的逻辑，修改后，再设置回去。
        # new_program_logic =deepcopy(all_logic_n2[0]) # 创建一个深拷贝以进行修改
        # new_program_logic.phases[0].duration = 45 # 例如，将第一个相位的时长改为45秒
        # new_program_logic.phases[1].minDur = 10 # 修改第二个相位的最小绿灯时间
        # traci.trafficlight.setCompleteRedYellowGreenDefinition(tls_id_to_control, new_program_logic)
        # print(f"  信号灯 {tls_id_to_control}: 已尝试修改并设置新的程序逻辑。")
        # 注意：setCompleteRedYellowGreenDefinition 需要一个 Logic 对象或列表。
        # 这个操作会完全替换掉该信号灯当前程序ID（或所有程序，如果提供列表）的逻辑。
        pass
    
    except traci.exceptions.TraCIException as e: # 捕获操作信号灯时的 TraCI 错误
        print(f"  操作信号灯 {tls_id_to_control} 时出错: {e}") # 打印错误信息
    
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/PFA8r0SqyvK2ZmfhdDiX3Op7GTxU.png)

    # (假设 traci 已经连接，并且仿真正在)
    # current_sim_time = traci.simulation.getTime() # 获取当前仿真时间
    
    edge_ids = traci.edge.getIDList() # 获取路网中所有边的ID列表 (字符串元组)
    print(f"当前时间: {traci.simulation.getTime():.2f}s, 边ID列表: {edge_ids}") # 打印信息
    
    if len(edge_ids) > 0: # 如果有边
    for edge_id in edge_ids: # 遍历每个边ID
        try:
            print(f"  边ID: {edge_id}") # 打印边ID
    
            # 获取边上的车道数量
            num_lanes = traci.edge.getLaneNumber(edge_id) # 获取边edge_id上的车道数量
            print(f"    车道数量: {num_lanes}") # 打印车道数量
    
            # 获取边上最后一步的车辆ID列表
            last_step_vehicle_ids = traci.edge.getLastStepVehicleIDs(edge_id) # 获取上一个仿真步在边edge_id上的所有车辆ID
            print(f"    上一时间步车辆列表: {last_step_vehicle_ids}") # 打印车辆列表
    
            # 获取边上最后一步的平均速度
            last_step_mean_speed = traci.edge.getLastStepMeanSpeed(edge_id) # 获取上一个仿真步在边edge_id上的平均车速 (m/s)
            print(f"    上一时间步平均速度: {last_step_mean_speed:.2f} m/s") # 打印平均速度
    
            # 获取边上最后一步的车辆总数
            last_step_veh_count = traci.edge.getLastStepVehicleNumber(edge_id) # 获取上一个仿真步在边edge_id上的车辆总数
            print(f"    上一时间步车辆总数: {last_step_veh_count}") # 打印车辆总数
            
            # 获取边的拥堵指数 (CO2, fuel, electricity等也可以获取，如果配置了排放模型)
            # traveltime: 动态计算的通过该边的行程时间 (考虑当前拥堵)
            current_travel_time = traci.edge.getTraveltime(edge_id) # 获取当前通过边edge_id的预估行程时间 (s)
            # efforts: 考虑拥堵的行程时间，通常与 traveltime 类似或用于特定路由算法
            # current_efforts = traci.edge.getEfforts(edge_id) 
    
            # 获取边的CO2排放量 (如果启用了排放模型)
            # co2_emission = traci.edge.getCO2Emission(edge_id) # 获取边edge_id上一步产生的CO2排放 (mg)
            # print(f"    CO2排放 (上一步): {co2_emission:.2f} mg")
    
            # 修改边的最大允许速度 (这会影响该边上所有车道的限速，除非车道有自己的特定限速)
            if edge_id == "E1" and current_sim_time > 120: # 假设在120秒后，降低E1边的限速
                new_max_speed_E1 = 8.0 # 设置新的最大速度为 8 m/s (约 28.8 km/h)
                original_max_speed_E1 = traci.lane.getMaxSpeed(f"{edge_id}_0") # 获取边上第一条车道的当前限速作为参考
                
                traci.edge.setMaxSpeed(edge_id, new_max_speed_E1) # 为边edge_id设置新的最大允许速度
                print(f"    边 {edge_id}: 在时间 {current_sim_time:.1f}s, 最大速度从约 {original_max_speed_E1:.2f} m/s 修改为 {new_max_speed_E1} m/s") # 打印修改信息
                # 验证修改是否生效 (通常会立即影响新进入或在该边上的车辆的行为)
                # new_lane_max_speed = traci.lane.getMaxSpeed(f"{edge_id}_0")
                # print(f"      -> 边 {edge_id} 上车道 {edge_id}_0 的新限速: {new_lane_max_speed:.2f} m/s")
    
            # 禁止/允许特定类型的车辆通过某条边
            # 假设 E3 原本允许所有车，现在我们禁止 "car_eastbound" 类型通过
            if edge_id == "E3" and current_sim_time > 150: # 在150秒后修改E3的通行权限
                # 获取当前允许的车辆类型列表
                # allowed_classes_str = traci.edge.getAllowed(edge_id) # 返回一个空格分隔的字符串
                # disallowed_classes_str = traci.edge.getDisallowed(edge_id) # 返回一个空格分隔的字符串
                
                # 要禁止 "car_eastbound"，我们需要先知道原本允许哪些，然后排除它
                # 或者，如果我们知道所有其他应该允许的类型
                # 例如，假设除了 "car_eastbound"，我们还允许 "car_northbound" 和 "truck"
                new_allowed_for_E3 = ["car_northbound", "truck"] # 定义新的允许类型列表
                # traci.edge.setAllowed(edge_id, new_allowed_for_E3) # 设置边edge_id允许通行的车辆类型列表
                # print(f"    边 {edge_id}: 在时间 {current_sim_time:.1f}s, 已设置允许类型为: {new_allowed_for_E3}")
                
                # 更简单的方式是禁止特定类型，如果只想禁止一个
                # traci.edge.setDisallowed(edge_id, ["car_eastbound"])
                # print(f"    边 {edge_id}: 在时间 {current_sim_time:.1f}s, 已禁止类型 'car_eastbound' 通行。")
                pass # 实际操作需要更仔细地管理允许/禁止列表
    
        except traci.exceptions.TraCIException as e: # 捕获TraCI异常
            print(f"  操作边 {edge_id} 时出错: {e}") # 打印错误信息
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/6lx1b0gqj8eRodMhntcmkHuJABaU.png)

    # (假设 traci 已经连接，并且仿真正在)
    # current_sim_time = traci.simulation.getTime() # 获取当前仿真时间
    
    lane_ids = traci.lane.getIDList() # 获取路网中所有车道的ID列表 (字符串元组)
    # print(f"当前时间: {traci.simulation.getTime():.2f}s, 车道ID列表 (部分): {lane_ids[:10] if len(lane_ids) > 10 else lane_ids}") # 打印部分车道ID
    
    if len(lane_ids) > 0: # 如果有车道
    # 让我们关注某条特定车道，例如 "E1_0" (边E1的第一条车道)
    lane_of_interest = "E1_0" # 定义我们感兴趣的车道ID
    if lane_of_interest in lane_ids: # 检查该车道是否存在
        try:
            print(f"  车道ID: {lane_of_interest}") # 打印车道ID
    
            # 获取车道长度
            length = traci.lane.getLength(lane_of_interest) # 获取车道lane_of_interest的长度 (m)
            print(f"    长度: {length:.2f} m") # 打印长度
    
            # 获取车道允许的最大速度
            max_speed = traci.lane.getMaxSpeed(lane_of_interest) # 获取车道lane_of_interest的最大允许速度 (m/s)
            print(f"    最大速度: {max_speed:.2f} m/s") # 打印最大速度
    
            # 获取车道上最后一步的车辆ID列表
            last_step_vehicle_ids_lane = traci.lane.getLastStepVehicleIDs(lane_of_interest) # 获取上一个仿真步在车道上的车辆ID
            print(f"    上一时间步车辆列表: {last_step_vehicle_ids_lane}") # 打印车辆列表
    
            # 获取车道上最后一步的平均速度
            last_step_mean_speed_lane = traci.lane.getLastStepMeanSpeed(lane_of_interest) # 获取上一个仿真步车道上的平均速度
            print(f"    上一时间步平均速度: {last_step_mean_speed_lane:.2f} m/s") # 打印平均速度
    
            # 获取车道上最后一步的占用率 (0到1之间)
            occupancy = traci.lane.getLastStepOccupancy(lane_of_interest) # 获取上一个仿真步车道的时间占用率
            print(f"    上一时间步占用率: {occupancy:.3f}") # 打印占用率
    
            # 获取车道上车辆的总等待时间
            waiting_time_lane = traci.lane.getWaitingTime(lane_of_interest) # 获取车道上所有车辆的累积等待时间之和 (s)
                                                                        # 这个值反映了该车道的拥堵程度
            print(f"    总等待时间: {waiting_time_lane:.2f} s") # 打印总等待时间
    
            # 获取车道所属的边ID
            edge_id_of_lane = traci.lane.getEdgeID(lane_of_interest) # 获取车道lane_of_interest所属的边的ID
            print(f"    所属边ID: {edge_id_of_lane}") # 打印所属边ID
    
            # 获取车道允许通行的车辆类型
            # allowed_vclasses_lane = traci.lane.getAllowed(lane_of_interest) # 返回一个空格分隔的字符串
            # print(f"    允许的车辆类型: '{allowed_vclasses_lane}'")
    
            # 修改车道的最大允许速度 (这会覆盖其所属边的限速)
            if current_sim_time > 180: # 假设在180秒后修改 "E1_0" 的限速
                new_max_speed_lane_E1_0 = 5.0 # 设置新限速为 5 m/s
                traci.lane.setMaxSpeed(lane_of_interest, new_max_speed_lane_E1_0) # 为车道lane_of_interest设置新的最大允许速度
                print(f"    车道 {lane_of_interest}: 在时间 {current_sim_time:.1f}s, 最大速度修改为 {new_max_speed_lane_E1_0} m/s") # 打印修改信息
            
            # 禁止/允许特定类型的车辆使用某条车道 (与边的操作类似)
            # if current_sim_time > 200:
            #    traci.lane.setDisallowed(lane_of_interest, ["truck"]) # 禁止卡车使用车道 E1_0
            #    print(f"    车道 {lane_of_interest}: 在时间 {current_sim_time:.1f}s, 已禁止 'truck' 类型通行。")
    
    
        except traci.exceptions.TraCIException as e: # 捕获TraCI异常
            print(f"  操作车道 {lane_of_interest} 时出错: {e}") # 打印错误信息
    else:
        print(f"  警告: 车道 {lane_of_interest} 不在路网中。") # 打印警告
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/2N6HxV3CoYzSe0KdqRFr4G1QcOiD.png)

    # (假设 traci 已经连接，并且仿真正在)
    junction_ids = traci.junction.getIDList() # 获取路网中所有交叉口(节点)的ID列表
    # print(f"当前时间: {traci.simulation.getTime():.2f}s, 交叉口ID列表: {junction_ids}")
    
    if len(junction_ids) > 0: # 如果有交叉口
    for junc_id in junction_ids: # 遍历每个交叉口ID
        try:
            # 获取交叉口的位置 (中心点坐标)
            pos_x, pos_y = traci.junction.getPosition(junc_id) # 获取交叉口junc_id的(x,y)坐标
            print(f"  交叉口ID: {junc_id}, 位置: ({pos_x:.2f}, {pos_y:.2f})") # 打印交叉口信息
            
            # 获取交叉口的形状 (构成边界的多边形顶点列表)
            # shape = traci.junction.getShape(junc_id) # 返回一个 ((x1,y1), (x2,y2), ...) 的元组
            # print(f"    形状顶点: {shape}")
    
        except traci.exceptions.TraCIException as e: # 捕获TraCI异常
            print(f"  获取交叉口 {junc_id} 信息时出错: {e}") # 打印错误信息
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/SHfqI2VicnLzZdrJYCEu3AaxpU05.png)

    import os, sys, traci, time # 导入所需模块
    from traci import constants as tc # 导入 traci.constants 并使用别名 tc，方便调用
    
    # (确保 SUMO_HOME/tools 在 sys.path 中)
    # (假设 SUMO 已经启动并在端口 8813 等待连接，或者使用 libsumo)
    sumo_config_file = "example.sumocfg" # SUMO配置文件
    traci_port = 8813 # TraCI端口
    use_libsumo_subscription_example = False # 是否使用libsumo
    
    def run_subscription_simulation(): # 定义订阅仿真的函数
    if not use_libsumo_subscription_example: # 如果不使用libsumo
        # (此处省略了手动启动SUMO的提示，假设已启动或使用subprocess)
        print(f"尝试连接到 SUMO (端口: {traci_port}) 进行订阅示例...") # 打印连接提示
        try:
            traci.connect(port=traci_port) # 连接到SUMO
            print("订阅示例: 连接 SUMO 成功。") # 打印连接成功信息
        except traci.exceptions.TraCIException as e: # 捕获连接异常
            print(f"订阅示例: 连接 SUMO 失败: {e}") # 打印连接失败信息
            return # 返回
    else: # 如果使用libsumo
        print("订阅示例: 使用 libsumo 启动仿真...") # 打印libsumo启动信息
        traci.start(["sumo", "-c", sumo_config_file, "--quit-on-end"]) # libsumo方式启动
        print("订阅示例: libsumo 仿真已启动。") # 打印启动成功信息
    
    # 订阅仿真时间 (虽然通常不需要订阅，因为simulationStep后可直接获取)
    # traci.simulation.subscribe(varIDs=[tc.VAR_TIME])
    
    step = 0 # 初始化步数
    max_steps_sub = 360 # 最大仿真步数
    
    # 用于存储已订阅的车辆ID，避免重复订阅
    subscribed_vehicle_ids = set() # 创建一个空集合，用于存放已订阅的车辆ID
    
    # 我们也对一个特定的信号灯 "N2" 的状态感兴趣
    tls_id_to_subscribe = "N2" # 定义要订阅的信号灯ID
    subscribed_tls = False # 标记信号灯是否已订阅
    
    try:
        while step < max_steps_sub: # 仿真主循环
            traci.simulationStep() # 执行一个仿真步
            current_sim_time_val = traci.simulation.getTime() # 获取当前仿真时间
    
            # 获取仿真订阅结果 (如果订阅了simulation域)
            # sim_results = traci.simulation.getSubscriptionResults()
            # if sim_results and tc.VAR_TIME in sim_results:
            #    current_sim_time_val = sim_results[tc.VAR_TIME]
            
            print(f"\n--- 步骤 {step}, 时间 {current_sim_time_val:.2f}s ---") # 打印当前步骤和时间
    
            # 动态订阅新出现的车辆
            current_vehicle_ids = set(traci.vehicle.getIDList()) # 获取当前所有车辆ID，并转为集合
            new_vehicles_to_subscribe = current_vehicle_ids - subscribed_vehicle_ids # 找出新出现的车辆 (集合差集)
    
            for new_veh_id in new_vehicles_to_subscribe: # 遍历所有新出现的车辆
                traci.vehicle.subscribe(new_veh_id, 
                                        varIDs=[tc.VAR_SPEED, tc.VAR_POSITION, tc.VAR_LANE_ID, tc.VAR_ANGLE]) # 订阅该车的速度、位置、车道ID和角度
                subscribed_vehicle_ids.add(new_veh_id) # 将该车ID添加到已订阅集合中
                print(f"  已为新车辆 {new_veh_id} 添加订阅 (速度, 位置, 车道, 角度)。") # 打印订阅信息
    
            # 获取已订阅车辆的数据
            if subscribed_vehicle_ids: # 如果有已订阅的车辆
                print("  获取已订阅车辆数据:") # 打印提示
            for veh_id_s in list(subscribed_vehicle_ids): # 遍历已订阅车辆ID列表 (转为list以防在迭代中修改集合)
                try:
                    # getSubscriptionResults 返回一个字典，键是常量，值是数据
                    subscription_data = traci.vehicle.getSubscriptionResults(veh_id_s) # 获取车辆veh_id_s的订阅结果
                    if subscription_data: # 如果成功获取到数据
                        speed_s = subscription_data.get(tc.VAR_SPEED, "N/A") # 从字典中获取速度，若无则为"N/A"
                        pos_s = subscription_data.get(tc.VAR_POSITION, "N/A") # 获取位置
                        lane_s = subscription_data.get(tc.VAR_LANE_ID, "N/A") # 获取车道ID
                        angle_s = subscription_data.get(tc.VAR_ANGLE, "N/A") # 获取角度
                        print(f"    车辆 {veh_id_s}: 速度={speed_s:.2f} m/s, 位置={pos_s}, 车道='{lane_s}', 角度={angle_s:.2f}°") # 打印订阅到的数据
                    else:
                        # 如果返回 None 或空字典，说明车辆可能已离开或订阅失效
                        print(f"    车辆 {veh_id_s}: 未获取到订阅数据 (可能已离开)。") # 打印未获取到数据信息
                        subscribed_vehicle_ids.discard(veh_id_s) # 从已订阅集合中移除该车ID
                except traci.exceptions.TraCIException as e_sub_veh: # 捕获获取订阅结果时的异常
                    print(f"    获取车辆 {veh_id_s} 订阅数据时出错: {e_sub_veh}。可能已离开。") # 打印错误信息
                    subscribed_vehicle_ids.discard(veh_id_s) # 移除该车ID
    
            # 订阅信号灯 "N2" (如果尚未订阅且存在)
            if not subscribed_tls and tls_id_to_subscribe in traci.trafficlight.getIDList(): # 如果未订阅且信号灯存在
                traci.trafficlight.subscribe(tls_id_to_subscribe, 
                                             varIDs=[tc.TL_RED_YELLOW_GREEN_STATE, tc.TL_PHASE_INDEX]) # 订阅信号灯的状态和相位索引
                subscribed_tls = True # 标记为已订阅
                print(f"  已为信号灯 {tls_id_to_subscribe} 添加订阅 (状态, 相位索引)。") # 打印订阅信息
    
            # 获取已订阅信号灯的数据
            if subscribed_tls: # 如果信号灯已订阅
                try:
                    tls_subscription_data = traci.trafficlight.getSubscriptionResults(tls_id_to_subscribe) # 获取信号灯的订阅结果
                    if tls_subscription_data: # 如果获取到数据
                        state_str_s = tls_subscription_data.get(tc.TL_RED_YELLOW_GREEN_STATE, "N/A") # 获取灯色状态字符串
                        phase_idx_s = tls_subscription_data.get(tc.TL_PHASE_INDEX, "N/A") # 获取相位索引
                        print(f"  信号灯 {tls_id_to_subscribe}: 状态='{state_str_s}', 相位索引={phase_idx_s}") # 打印订阅到的数据
                    else:
                        # 信号灯通常不会消失，除非路网被动态修改
                        print(f"  信号灯 {tls_id_to_subscribe}: 未获取到订阅数据。") # 打印未获取到数据信息
                except traci.exceptions.TraCIException as e_sub_tls: # 捕获获取信号灯订阅结果时的异常
                    print(f"    获取信号灯 {tls_id_to_subscribe} 订阅数据时出错: {e_sub_tls}。") # 打印错误信息
    
    
            # 移除已离开路网的车辆的订阅 (getSubscriptionResults 返回 None 时已处理)
            current_vehicle_ids_after_step = set(traci.vehicle.getIDList()) # 再次获取当前车辆ID
            departed_vehicles = subscribed_vehicle_ids - current_vehicle_ids_after_step # 找出已离开的车辆
            for departed_veh_id in departed_vehicles: # 遍历已离开的车辆
                print(f"  车辆 {departed_veh_id} 已离开路网，从订阅列表中移除。") # 打印移除信息
                subscribed_vehicle_ids.discard(departed_veh_id) # 从集合中移除
    
            step += 1 # 步数加1
            # time.sleep(0.01) # (可选) 略微减慢脚本执行速度
    
    except traci.exceptions.TraCIException as e: # 捕获仿真过程中的TraCI异常
        print(f"订阅示例: 仿真过程中发生 TraCI 错误: {e}") # 打印错误信息
    except KeyboardInterrupt: # 捕获用户中断 (Ctrl+C)
        print("订阅示例: 用户请求中断仿真。") # 打印中断信息
    finally: # 最终执行块
        print("订阅示例: 正在关闭 TraCI 连接...") # 打印关闭连接信息
        traci.close() # 关闭TraCI连接
        print("订阅示例: TraCI 连接已关闭。") # 打印连接已关闭信息
    
    if __name__ == "__main__": # 如果是主程序
    # (确保 example.sumocfg 等文件存在)
    if not os.path.exists(sumo_config_file):                            # 检查配置文件是否存在
        print(f"错误: SUMO 配置文件 '{sumo_config_file}' 未找到。")    # 打印错误信息
    else:
        run_subscription_simulation() # 带订阅的仿真函数
    
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/KJ3oHaNMQx7rD4fngij6z1pS9e0E.png)

* 返回上一个时间步在车道 `laneID` 上处于“停止”状态（速度低于某一阈值，通常是跟驰模型中的 `emergencyDecel` 或一个小的速度值）的车辆数量。这可以作为排队车辆数的一个良好近似。
* **优点** : 简单直接。
* **缺点** : 只统计完全停止的车辆，缓慢移动的拥堵队列可能被低估。

* 返回车道 `laneID` 上所有车辆的累积等待时间之和。等待时间长的车道通常意味着有较长的排队。
* **优点** : 能反映拥堵的严重程度。
* **缺点** : 不是直接的排队长度，需要转换为可比较的指标。

* 可以在路网中放置多种类型的探测器，如： 
  * **环形线圈探测器 (Induction Loops Detectors - E1)** : 可以获取通过车辆数、平均速度、占有率等。`traci.inductionloop.getLastStepVehicleNumber(detID)`，`traci.inductionloop.getTimeSinceDetection(detID)` 等。
  * **车道区域探测器 (Lane Area Detectors - E2)** : 监控一个车道区域，可以获取区域内的车辆数、平均速度、占有率等。`traci.lanearea.getLastStepVehicleNumber(detID)`，`traci.lanearea.getJamLengthVehicle(detID)` (返回拥堵队列中的车辆数)，`traci.lanearea.getJamLengthMeters(detID)` (返回拥堵队列的长度，米)。 **E2 探测器的`getJamLengthVehicle` 和 `getJamLengthMeters` 是获取排队长度的较好方式。** * **多入口/多出口探测器 (Multi-Entry/Multi-Exit Detectors - E3)** : 用于统计进入和离开一个区域的车辆，计算行程时间等。

* 探测器需要在附加文件 (`.add.xml`) 中定义，并与特定的车道和位置关联。
* **优点** : 可以提供更精确和定制化的数据。E2探测器直接提供排队长度。
* **缺点** : 需要预先在路网中配置探测器。

* 获取车道上所有车辆的位置 `traci.vehicle.getLanePosition(vehID)` 和速度 `traci.vehicle.getSpeed(vehID)`。
* 从车道末端（交叉口停车线附近）向前追溯，统计速度低于某一阈值（例如 1 m/s）的连续车辆数量或队列长度。
* **优点** : 非常灵活，不需要预设探测器。
* **缺点** : 计算相对复杂，可能比直接调用 TraCI 函数慢。

    # (在 run_simulation 函数内部，traci 连接后)
    tls_id_control = "N2" # 要控制的信号灯ID
    phase_definitions = {} # 用于存储每个相位索引对应的主要进口车道和灯色状态
    min_green_time = 10 # 秒，最小绿灯时间
    max_green_time = 60 # 秒，最大绿灯时间
    yellow_time = 3     # 秒，黄灯时间 (假设所有黄灯相位都是这个时长)
    current_phase_start_time = 0 # 当前相位开始的仿真时间
    active_phase_index = -1 # 当前活动的相位索引
    
    # 获取信号灯的完整逻辑定义
    try:
    logic = traci.trafficlight.getCompleteRedYellowGreenDefinition(tls_id_control) # 获取N2的完整逻辑
    if logic and len(logic) > 0: # 如果有逻辑定义
        program = logic[0] # 获取第一个（通常是默认的）程序
        print(f"信号灯 '{tls_id_control}' 程序 '{program.programID}' 的相位信息:") # 打印程序信息
        for i, phase_obj in enumerate(program.phases): # 遍历程序中的每个相位对象
            print(f"  相位 {i}: 状态='{phase_obj.state}', 预设时长={phase_obj.duration}s, "
                  f"minDur={phase_obj.minDur}s, maxDur={phase_obj.maxDur}s") # 打印相位详细信息
            # 这里需要根据 phase_obj.state (灯色字符串) 来判断该相位主要服务于哪些进口车道
            # 这通常需要对路网和信号灯的XML定义有所了解，或者通过观察GUI和状态字符串的对应关系
            # 例如，假设我们通过观察或分析，确定了：
            # 相位0 ("GGgrrrGGgrrr...") 主要服务于 边 E1 (进口车道 E1_0, E1_1)
            # 相位2 ("rrrGGgrrrGGg...") 主要服务于 边 E3 (进口车道 E3_0)
            # (注意：实际的相位索引和状态字符串会因 netconvert 生成的逻辑而异，需要根据实际情况调整)
            # 为了示例，我们硬编码这个映射关系。在真实应用中，可能需要更智能的解析。
            if i == 0: # 假设相位0是东西向绿 (服务于E1)
                phase_definitions[i] = {
                    "name": "East-West Green", # 相位名称
                    "state_str": phase_obj.state, # 该相位的灯色状态字符串
                    "duration": phase_obj.duration, # 预设时长
                    "min_green": phase_obj.minDur if phase_obj.minDur > 0 else min_green_time, # 最小绿灯时间
                    "max_green": phase_obj.maxDur if phase_obj.maxDur > 0 else max_green_time, # 最大绿灯时间
                    "monitored_lanes": ["E1_0", "E1_1"] # 该相位绿灯时需要监控排队的进口车道列表
                }
            elif i == 2: # 假设相位2是南北向绿 (服务于E3) - 注意这只是一个假设的索引
                phase_definitions[i] = {
                    "name": "North-South Green (E3)", # 相位名称
                    "state_str": phase_obj.state, # 该相位的灯色状态字符串
                    "duration": phase_obj.duration, # 预设时长
                    "min_green": phase_obj.minDur if phase_obj.minDur > 0 else min_green_time, # 最小绿灯时间
                    "max_green": phase_obj.maxDur if phase_obj.maxDur > 0 else max_green_time, # 最大绿灯时间
                    "monitored_lanes": ["E3_0"] # 该相位绿灯时需要监控排队的进口车道列表
                }
            # 应该为所有主要的绿灯相位都这样做，并确保包含了黄灯和全红的中间相位
            # 在一个简单的十字路口，通常有4个车辆相位（例如：东西直行，东西左转，南北直行，南北左转）
            # 加上它们之间的黄灯和全红相位。
            # 为了简化此示例，我们只关注两个冲突的主方向。
            # 实际的 netconvert 可能生成如：
            # Phase 0: EW Green
            # Phase 1: EW Yellow
            # Phase 2: NS Green
            # Phase 3: NS Yellow
            # 在这种情况下，我们的 phase_definitions 需要包含对黄灯相位的识别和处理。
    
        # 筛选出我们定义的、要进行自适应控制的绿灯相位
        # 我们只对那些有 "monitored_lanes" 的相位感兴趣，并且它们应该是绿灯相位
        adaptive_phases = {idx: p_info for idx, p_info in phase_definitions.items() if "monitored_lanes" in p_info} # 筛选出包含监控车道的相位
        if not adaptive_phases: # 如果没有找到可自适应的相位
            print(f"警告: 未能在信号灯 '{tls_id_control}' 中识别出可用于自适应控制的绿灯相位定义。请检查 phase_definitions 的硬编码。") # 打印警告
            # return # 可能需要退出或使用默认逻辑
        else:
            print(f"用于自适应控制的相位定义: {adaptive_phases}") # 打印自适应相位定义
    
        # 初始化：设置到第一个自适应相位 (如果存在)
        if adaptive_phases: # 如果存在自适应相位
            initial_adaptive_phase_idx = sorted(adaptive_phases.keys())[0] # 获取第一个自适应相位的索引
            traci.trafficlight.setPhase(tls_id_control, initial_adaptive_phase_idx) # 设置初始相位
            active_phase_index = initial_adaptive_phase_idx # 更新当前活动相位索引
            current_phase_start_time = traci.simulation.getTime() # 记录当前相位开始时间
            print(f"  控制器初始化: 信号灯 '{tls_id_control}' 设置到初始相位 {active_phase_index} ('{adaptive_phases[active_phase_index]['name']}')") # 打印初始化信息
        
    else:
        print(f"错误: 无法获取信号灯 '{tls_id_control}' 的逻辑定义。") # 打印错误信息
        return # 退出函数
    
    except traci.exceptions.TraCIException as e: # 捕获TraCI异常
    print(f"初始化信号灯控制时发生 TraCI 错误: {e}") # 打印错误信息
    return # 退出函数
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/QNRE28hwc5jA1I7YyqOxzXa0tbgV.png)

    # (在 run_simulation 函数的主循环 while step < max_steps: 内部)
    # traci.simulationStep() # 已在循环开始处执行
    
    current_time = traci.simulation.getTime() # 获取当前仿真时间
    
    # 检查是否需要进行信号灯决策 (例如，只在活动相位是自适应相位时)
    if active_phase_index in adaptive_phases: # 如果当前活动相位是我们定义的自适应相位之一
    current_phase_info = adaptive_phases[active_phase_index] # 获取当前活动相位的信息
    time_in_current_phase = current_time - current_phase_start_time # 计算当前相位已持续的时间
    
    # 1. 数据采集：获取当前相位监控车道的排队长度 (使用 getLastStepHaltingNumber)
    current_phase_total_queue = 0 # 初始化当前相位服务的总排队车辆数
    print(f"  时间 {current_time:.1f}s - 当前活动相位: {active_phase_index} ('{current_phase_info['name']}')，已持续 {time_in_current_phase:.1f}s") # 打印当前相位信息
    for lane_id in current_phase_info["monitored_lanes"]: # 遍历当前相位监控的所有车道
        if lane_id in traci.lane.getIDList(): # 确保车道ID有效 (虽然通常是有效的)
            halting_num = traci.lane.getLastStepHaltingNumber(lane_id) # 获取该车道上停止的车辆数
            current_phase_total_queue += halting_num #累加到总排队数
            print(f"    车道 {lane_id}: 排队车辆数 = {halting_num}") # 打印该车道的排队数
        else:
            print(f"    警告: 监控的车道 {lane_id} 未找到。") # 打印警告
    
    # 2. 决策逻辑：是否切换相位？
    # 切换条件：
    #   a) 当前相位已达到其最小绿灯时间。
    #   b) 并且 (当前相位服务的队列已基本消散 (例如，排队长度 < 阈值) 
    #           OR 当前相位已达到其最大绿灯时间)。
    
    switch_phase = False # 初始化切换相位标志为False
    if time_in_current_phase >= current_phase_info["min_green"]: # 如果已达到最小绿灯时间
        queue_cleared_threshold = 2 # 定义队列基本消散的阈值 (例如，少于2辆车)
        if current_phase_total_queue <= queue_cleared_threshold: # 如果当前相位队列已消散
            print(f"    决策: 当前相位 {active_phase_index} 队列已消散 (排队={current_phase_total_queue} <= {queue_cleared_threshold})。") # 打印决策信息
            switch_phase = True # 设置切换标志为True
        elif time_in_current_phase >= current_phase_info["max_green"]: # 或者已达到最大绿灯时间
            print(f"    决策: 当前相位 {active_phase_index} 已达到最大绿灯时间 ({time_in_current_phase:.1f}s >= {current_phase_info['max_green']}s)。") # 打印决策信息
            switch_phase = True # 设置切换标志为True
    
    # 3. 如果决定切换相位，选择下一个相位
    if switch_phase: # 如果决定切换相位
        print(f"    执行相位切换...") # 打印执行切换信息
        # 找到所有其他（冲突的）自适应相位，并计算它们的"需求" (例如，总排队长度)
        candidate_next_phases = {} # 用于存储候选下一相位及其排队长度的字典
        for phase_idx, phase_info_contender in adaptive_phases.items(): # 遍历所有自适应相位
            if phase_idx == active_phase_index: # 跳过当前活动相位
                continue 
    
            contender_total_queue = 0 # 初始化竞争相位的总排队数
            for lane_id_contender in phase_info_contender["monitored_lanes"]: # 遍历竞争相位监控的车道
                if lane_id_contender in traci.lane.getIDList(): # 确保车道有效
                    halting_num_contender = traci.lane.getLastStepHaltingNumber(lane_id_contender) # 获取排队数
                    contender_total_queue += halting_num_contender # 累加
            candidate_next_phases[phase_idx] = contender_total_queue # 将相位索引和其总排队数存入字典
            print(f"      候选相位 {phase_idx} ('{phase_info_contender['name']}'): 总排队 = {contender_total_queue}") # 打印候选相位及其排队
    
        if not candidate_next_phases: # 如果没有其他候选的自适应相位 (例如，只有一个自适应相位被定义)
            print("      警告: 没有其他候选的自适应相位进行切换。保持当前相位（或需要更复杂的逻辑）。") # 打印警告
            # 这种情况下，可能需要让当前相位再持续一段时间，或者切换到一个默认的休息相位(如全红)
            # 为了简单，我们这里什么都不做，它会在下一个max_green时再次尝试切换
        else:
            # 选择排队最长的候选相位
            # max() 在字典上默认作用于键，我们需要作用于值
            # best_next_phase_index = max(candidate_next_phases, key=candidate_next_phases.get) # 获取排队最长的候选相位的索引
    
            # 考虑一个更健壮的选择：如果所有其他方向都没车，可以延长当前绿灯，除非达到max_green
            # 但我们这里的逻辑是，一旦决定切换，就找下一个。
            
            # 找到排队数最大的那个相位索引
            best_next_phase_index = -1 # 初始化最佳下一相位索引
            max_queue_found = -1       # 初始化找到的最大排队数
            for p_idx, q_val in candidate_next_phases.items(): # 遍历候选相位及其排队数
                if q_val > max_queue_found: # 如果当前排队数更大
                    max_queue_found = q_val # 更新最大排队数
                    best_next_phase_index = p_idx # 更新最佳相位索引
                elif q_val == max_queue_found and best_next_phase_index != -1: # 如果排队数相同
                    # 如果排队相同，可以加入一些轮转逻辑或保持之前的选择，避免频繁在排队相同的相位间抖动
                    # 这里简单地选择第一个遇到的最大值
                    pass
    
            if best_next_phase_index != -1: # 如果找到了最佳下一相位
                print(f"    选定的下一相位: {best_next_phase_index} ('{adaptive_phases[best_next_phase_index]['name']}')，其排队为 {max_queue_found}") # 打印选定的下一相位信息
    
                # **处理黄灯和全红过渡**
                # 假设我们的 `program.phases` (从 getCompleteRedYellowGreenDefinition 获取的)
                # 已经正确地包含了从当前绿灯相位 `active_phase_index` 到 
                # `best_next_phase_index` 之间的黄灯和全红相位。
                # 例如，如果 Phase 0 是 EW绿，Phase 1 是 EW黄，Phase 2 是 NS绿。
                # 从 Phase 0 切换到 Phase 2，我们应该先切换到 Phase 1 (EW黄)。
                # 这需要我们对信号灯的完整相位序列有清晰的了解。
    
                # 简化的过渡处理：我们直接设置到下一个选定的绿灯相位。
                # SUMO的 setPhase 可能会尝试智能处理，但依赖它可能不总是可靠。
                # 一个更稳妥（但更复杂）的方法是，显式地管理黄灯和全红。
    
                # 为了本示例的简单性，我们直接切换到选定的下一个主绿灯相位。
                # 警告：这在实际中可能跳过必要的黄灯/全红，除非SUMO能正确插入。
                # 更好的做法是，如果 phase_definitions 包含了黄灯相位，
                # 我们应该先找到从 active_phase_index 到 best_next_phase_index 的
                # "过渡相位序列"（例如，当前绿 -> 黄 -> 全红 -> 下一个绿），然后依次设置它们。
                
                # 查找从 active_phase_index 到 best_next_phase_index 的过渡路径
                # 这需要知道信号灯的完整相位图。
                # 假设我们的 `program.phases` 是一个完整的周期，包含了所有黄灯和可能的清空相位。
                # 例如：P0 (EW G), P1 (EW Y), P2 (ALL R), P3 (NS G), P4 (NS Y), P5 (ALL R)
                # 如果当前是 P0，下一个是 P3，则需要先切换到 P1，等待，再到 P2，等待，再到 P3。
                # 这个逻辑比较复杂，暂时简化处理。
    
                # --- 简化切换逻辑 ---
                # 我们将直接设置到下一个主绿灯相位，并假设SUMO或相位定义能处理中间过渡。
                # 或者，我们应该在这里只设置“意图”，并在接下来的几个仿真步中手动通过黄灯相位。
    
                # 为了这个示例，我们假设直接调用 setPhase 到下一个主绿灯相位是可接受的，
                # SUMO会使用该相位定义的灯色。
                # 这通常意味着，如果下一个主绿灯相位与当前相位冲突，SUMO的 setPhase 
                # 不会自动插入黄灯。你需要自己管理黄灯阶段。
    
                # --- 更推荐的黄灯管理方式 ---
                # 1. 识别当前绿灯相位对应的黄灯相位。
                # 2. `traci.trafficlight.setPhase(tls_id_control, yellow_phase_index_of_current_green)`
                # 3. 等待黄灯时间 `yellow_time`。
                # 4. (可选) 切换到全红相位，等待全红时间。
                # 5. `traci.trafficlight.setPhase(tls_id_control, best_next_phase_index)`
                
                # --- 为本示例采用的简化（但可能不安全的）切换 ---
                # 我们将标记一个 "切换请求"，并在下一个循环开始时处理黄灯（如果需要）。
                # 或者，更简单：我们直接设置到下一个绿灯相位，并重置计时器。
                # 这依赖于 setPhase 的行为。如果 setPhase 只是改变灯色字符串，
                # 那它不会自动处理黄灯。
                
                # 假设我们有一个函数来获取从当前相位到目标相位的黄灯相位
                # yellow_phase_for_current = find_yellow_phase(active_phase_index, program.phases)
                # if yellow_phase_for_current is not None:
                #    traci.trafficlight.setPhase(tls_id_control, yellow_phase_for_current)
                #    active_phase_index = yellow_phase_for_current 
                #    current_phase_start_time = current_time
                #    # 然后在黄灯时间过后，再切换到下一个绿灯
                # else: # 如果找不到黄灯，直接切 (不推荐)
                #    traci.trafficlight.setPhase(tls_id_control, best_next_phase_index)
                #    active_phase_index = best_next_phase_index
                #    current_phase_start_time = current_time
                
                # --- 最简单的（用于教学，但忽略黄灯的）切换 ---
                print(f"    切换到下一绿灯相位 {best_next_phase_index} (为简化，本示例忽略黄灯过渡的显式管理)。") # 打印切换信息
                traci.trafficlight.setPhase(tls_id_control, best_next_phase_index) # 直接设置到选定的最佳下一相位
                active_phase_index = best_next_phase_index # 更新当前活动相位索引
                current_phase_start_time = current_time # 更新当前相位开始时间
            else:
                print(f"    决策: 所有其他方向均无排队，或无法确定最佳下一相位。保持当前相位（可能延长）。") # 打印保持当前相位信息
                # 这种情况下，如果当前相位队列也空了，可以考虑进入一个短暂的“全红”或“休息”相位，
                # 或者简单地让当前绿灯持续到其max_green。
                # 如果 current_phase_total_queue == 0，也可以延长当前绿灯时间，直到max_green。
                # 这里为了简单，如果没有更好的选择，它会在下一次max_green时再次评估。
    
    elif active_phase_index != -1 and active_phase_index not in adaptive_phases: # 如果当前是黄灯或全红等非自适应控制的相位
        # 这里处理黄灯/全红相位的持续时间
        # 假设我们知道黄灯相位应该持续 yellow_time 秒
        # 我们需要知道这个非自适应相位是哪个绿灯相位之后的黄灯/全红
        # 这个逻辑的实现依赖于 phase_definitions 是否包含了所有相位（包括黄、红）
        # 并且需要一个状态机来跟踪我们是从哪个绿灯切换过来的，以及下一个目标绿灯是什么。
        # 例如：
        # if "yellow" in phase_name_of_active_phase_index:
        #    if time_in_current_phase >= yellow_time:
        #        # 切换到预定的下一个相位 (可能是全红，或下一个绿灯)
        #        target_after_yellow = get_target_phase_after_yellow(active_phase_index) # 需要辅助函数
        #        traci.trafficlight.setPhase(tls_id_control, target_after_yellow)
        #        active_phase_index = target_after_yellow
        #        current_phase_start_time = current_time
        pass # 为简化，本示例不显式管理黄灯/全红的计时和切换，依赖于我们只在主绿灯相位做决策。
             # 这种简化意味着黄灯和全红必须由 `traci.trafficlight.setPhase` 切换到下一个主绿灯时
             # SUMO的相位定义来隐式处理，或者我们选择的下一个 `adaptive_phase` 紧邻当前绿灯。
             # 一个更完整的控制器需要明确的状态（如 "GREEN", "YELLOW", "ALL_RED"）和转换逻辑。
    
    else: # 如果 active_phase_index 是 -1 (例如刚开始或未识别自适应相位)
    # 尝试重新初始化或保持一个默认状态
    if adaptive_phases and (active_phase_index == -1 or active_phase_index not in adaptive_phases): # 如果存在自适应相位但当前不是
        # 可能在黄灯/全红阶段，或者刚启动
        # 尝试找到一个排队最多的自适应相位作为起点 (或者默认第一个)
        # (与上面 switch_phase 后的选择逻辑类似)
        # ... (省略重复的相位选择代码) ...
        # 如果找到了，则 traci.trafficlight.setPhase(...) 并更新 active_phase_index, current_phase_start_time
        pass # 初始阶段由之前的初始化代码处理，这里先跳过复杂重试逻辑
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/aEGut4TM8gkVfYizN75KOxW9PbFJ.png)

    import os
    import sys
    import traci
    import time # 仅用于可能的调试延迟
    from traci import constants as tc # 使用别名 tc
    
    # --- (SUMO_HOME 检查代码，与之前相同，此处省略以减少重复) ---
    if 'SUMO_HOME' in os.environ:
    tools_path = os.path.join(os.environ['SUMO_HOME'], 'tools')
    if tools_path not in sys.path:
        sys.path.append(tools_path)
    else:
    sys.exit("请设置环境变量 'SUMO_HOME' 指向您的SUMO安装目录。")
    # --- (结束 SUMO_HOME 检查) ---
    
    SUMO_BINARY = "sumo-gui"  # 使用GUI方便观察
    sumo_config_file = "example.sumocfg"
    traci_port = 8813
    use_libsumo_adaptive_ctrl = False # True for libsumo, False for TCP
    
    # --- 自适应信号控制参数 ---
    TLS_ID_TO_CONTROL = "N2" # 要控制的信号灯ID
    MIN_GREEN_TIME = 10      # 秒，最小绿灯时间
    MAX_GREEN_TIME = 45      # 秒，最大绿灯时间
    # YELLOW_TIME = 3        # 秒，黄灯时间 (如果手动管理黄灯，会用到)
    # ALL_RED_TIME = 2       # 秒，全红清空时间 (如果手动管理)
    QUEUE_CLEARED_THRESHOLD = 2 # 辆，队列消散的车辆数阈值
    DECISION_INTERVAL = 1    # 秒，每隔多少仿真秒做一次决策 (设为1则每步都决策)
    
    # 全局变量存储信号灯控制状态
    g_adaptive_tls_definitions = {} # {tls_id: {phase_idx: phase_info_dict}}
    g_active_phase_index = {}       # {tls_id: current_phase_index}
    g_current_phase_start_time = {} # {tls_id: sim_time}
    
    def initialize_adaptive_tls_control(): # 初始化自适应信号灯控制的函数
    """获取并解析要控制的信号灯的相位信息。"""
    global g_adaptive_tls_definitions, g_active_phase_index, g_current_phase_start_time # 声明使用全局变量
    
    if TLS_ID_TO_CONTROL not in traci.trafficlight.getIDList(): # 检查要控制的信号灯是否存在
        print(f"错误: 信号灯 '{TLS_ID_TO_CONTROL}' 在路网中未找到。") # 打印错误信息
        return False # 返回False表示初始化失败
    
    g_adaptive_tls_definitions[TLS_ID_TO_CONTROL] = {} # 为该信号灯初始化相位定义字典
    phase_definitions_for_tls = {} # 临时的相位定义存储
    
    try:
        logic = traci.trafficlight.getCompleteRedYellowGreenDefinition(TLS_ID_TO_CONTROL) # 获取信号灯的完整逻辑
        if not logic or not logic[0].phases: # 如果没有逻辑或没有相位
            print(f"错误: 信号灯 '{TLS_ID_TO_CONTROL}' 没有找到有效的相位定义。") # 打印错误信息
            return False # 返回失败
    
        program = logic[0] # 获取第一个程序
        print(f"\n信号灯 '{TLS_ID_TO_CONTROL}' 程序 '{program.programID}' 的原始相位信息:") # 打印原始相位信息
        for i, phase_obj in enumerate(program.phases): # 遍历所有相位
            print(f"  原始相位 {i}: 状态='{phase_obj.state}', 预设时长={phase_obj.duration:.1f}s, "
                  f"minDur={phase_obj.minDur:.1f}s, maxDur={phase_obj.maxDur:.1f}s") # 打印原始相位详细信息
    
            # --- 关键: 识别哪些相位是我们要控制的主绿灯相位，以及它们服务的车道 ---
            # 这部分通常需要根据具体的 .net.xml 或 netedit 中的信号灯定义来确定。
            # 我们需要将相位索引映射到它所服务的进口车道。
            # 简化示例假设 (需要根据你的 'example.net.xml' 中 N2 的实际相位调整):
            # Phase 0: EW Green (services E1_0, E1_1)
            # Phase 1: EW Yellow
            # Phase 2: NS Green (services E3_0)
            # Phase 3: NS Yellow
            # (实际 netconvert 可能生成更多相位，例如全红清空)
            
            # 为了这个示例，我们假设如下（你需要根据你的实际情况修改这里的判断条件！）
            # 提示: 查看 `phase_obj.state` 字符串，'G' 或 'g' 代表绿灯。
            # 通过 `traci.trafficlight.getControlledLanes(TLS_ID_TO_CONTROL)` 和
            # `traci.trafficlight.getControlledLinks(TLS_ID_TO_CONTROL)` 结合 `phase_obj.state`
            # 可以更精确地判断一个相位服务于哪些车道/转向。
            
            is_ew_green = "GGg" in phase_obj.state[:3] and "rrr" in phase_obj.state[3:6] # 一个简单的判断东西绿的启发式方法
            is_ns_green = "rrr" in phase_obj.state[:3] and "GGg" in phase_obj.state[3:6] # 一个简单的判断南北绿的启发式方法
            # 上述判断非常粗略，实际中需要更可靠的相位识别逻辑！
    
            monitored_lanes_for_phase = [] # 初始化该相位的监控车道列表
            phase_name = f"Phase {i}" # 默认相位名称
            is_adaptive_green_phase = False # 标记是否为我们关注的自适应绿灯相位
    
            if i == 0: # 假设相位0是我们的东西向绿灯 (基于我们之前硬编码的认知)
                phase_name = "EW_Green" # 设置相位名称
                monitored_lanes_for_phase = ["E1_0", "E1_1"] # 设置监控车道
                is_adaptive_green_phase = True # 标记为自适应绿灯相位
            elif i == 2: # 假设相位2是我们的南北向绿灯 (基于我们之前硬编码的认知)
                      # 注意: 实际的索引可能不同！你需要检查你的 N2 信号灯定义！
                phase_name = "NS_Green_E3" # 设置相位名称
                monitored_lanes_for_phase = ["E3_0"] # 设置监控车道
                is_adaptive_green_phase = True # 标记为自适应绿灯相位
            
            # 也可以添加对黄灯相位的识别，如果需要手动管理黄灯
            # elif "y" in phase_obj.state.lower():
            #    phase_name = f"Yellow_After_P{i-1}" # 假设黄灯总是在绿灯之后
    
            if is_adaptive_green_phase: # 如果是我们定义的自适应绿灯相位
                phase_definitions_for_tls[i] = { # 存储相位信息
                    "name": phase_name, # 相位名称
                    "state_str": phase_obj.state, # 灯色状态字符串
                    "min_green": MIN_GREEN_TIME, # 使用全局最小绿灯时间
                    "max_green": MAX_GREEN_TIME, # 使用全局最大绿灯时间
                    "monitored_lanes": monitored_lanes_for_phase, # 监控的车道
                    "is_green_phase": True # 标记这是个主绿灯相位
                }
            # else: # 对于非我们主动控制的相位 (如黄灯、全红)，也可能需要记录它们的预设时长
            #    phase_definitions_for_tls[i] = {
            #        "name": phase_name,
            #        "state_str": phase_obj.state,
            #        "duration": phase_obj.duration, # 记录预设时长
            #        "is_green_phase": False
            #    }
    
    
        if not any(p.get("is_green_phase", False) for p in phase_definitions_for_tls.values()): # 检查是否有被识别的绿灯相位
            print(f"警告: 未能在信号灯 '{TLS_ID_TO_CONTROL}' 中识别出任何可用于自适应控制的绿灯相位。请调整 initialize_adaptive_tls_control 中的相位识别逻辑。") # 打印警告
            return False # 返回失败
    
        g_adaptive_tls_definitions[TLS_ID_TO_CONTROL] = phase_definitions_for_tls # 将解析的相位定义存入全局变量
        print(f"\n控制器: 为信号灯 '{TLS_ID_TO_CONTROL}' 解析出的自适应相位定义:") # 打印解析结果
        for p_idx, p_info in g_adaptive_tls_definitions[TLS_ID_TO_CONTROL].items(): # 遍历解析出的相位
            if p_info.get("is_green_phase"): # 如果是绿灯相位
                print(f"  接管的相位 {p_idx} ('{p_info['name']}'): 监控车道={p_info['monitored_lanes']}, "
                      f"minG={p_info['min_green']}, maxG={p_info['max_green']}") # 打印详细信息
        
        # 初始化：设置到第一个被识别的自适应绿灯相位
        first_adaptive_phase_idx = -1 # 初始化第一个自适应相位索引
        sorted_phase_indices = sorted(g_adaptive_tls_definitions[TLS_ID_TO_CONTROL].keys()) # 获取排序后的相位索引
        for p_idx in sorted_phase_indices: # 遍历排序后的索引
            if g_adaptive_tls_definitions[TLS_ID_TO_CONTROL][p_idx].get("is_green_phase"): # 如果是绿灯相位
                first_adaptive_phase_idx = p_idx # 设置为第一个自适应相位
                break # 退出循环
        
        if first_adaptive_phase_idx != -1: # 如果找到了第一个自适应相位
            traci.trafficlight.setPhase(TLS_ID_TO_CONTROL, first_adaptive_phase_idx) # 设置初始相位
            g_active_phase_index[TLS_ID_TO_CONTROL] = first_adaptive_phase_idx # 更新全局活动相位索引
            g_current_phase_start_time[TLS_ID_TO_CONTROL] = traci.simulation.getTime() # 更新全局相位开始时间
            print(f"控制器初始化: 信号灯 '{TLS_ID_TO_CONTROL}' 已设置为初始相位 {first_adaptive_phase_idx} "
                  f"('{g_adaptive_tls_definitions[TLS_ID_TO_CONTROL][first_adaptive_phase_idx]['name']}')") # 打印初始化成功信息
            return True # 返回成功
        else:
            print(f"错误: 初始化时未能找到任何可设置的初始自适应绿灯相位。") # 打印错误信息
            return False # 返回失败
    
    except traci.exceptions.TraCIException as e: # 捕获TraCI异常
        print(f"初始化自适应信号灯 '{TLS_ID_TO_CONTROL}' 控制时发生 TraCI 错误: {e}") # 打印错误信息
        return False # 返回失败
    
    def adaptive_tls_logic_step(tls_id): # 定义自适应信号灯逻辑的单步执行函数
    """在一个仿真步中为指定的tls_id执行自适应控制逻辑。"""
    global g_active_phase_index, g_current_phase_start_time # 声明使用全局变量
    
    if tls_id not in g_adaptive_tls_definitions or not g_adaptive_tls_definitions[tls_id]: # 检查该信号灯是否有定义
        # print(f"调试: tls_id {tls_id} 没有在 g_adaptive_tls_definitions 中找到定义。")
        return # 如果没有定义，则直接返回
    
    current_time = traci.simulation.getTime() # 获取当前仿真时间
    active_phase_idx = g_active_phase_index.get(tls_id, -1) # 获取当前活动相位索引，若无则为-1
    phase_start_time = g_current_phase_start_time.get(tls_id, current_time) # 获取相位开始时间
    
    if active_phase_idx == -1 or not g_adaptive_tls_definitions[tls_id].get(active_phase_idx, {}).get("is_green_phase"): # 如果当前不是一个我们正在控制的绿灯相位
        # print(f"调试: 当前活动相位 {active_phase_idx} 不是一个受控的绿灯相位，跳过决策。")
        # 这可能意味着它正处于黄灯/全红过渡，或者初始化失败。
        # 一个完整的控制器需要在这里处理黄灯/全红的计时。
        # 为了简化，我们假设：如果不是我们定义的绿灯相位，我们暂时不干预，等待它自然结束或下次被我们控制。
        # 或者，如果知道黄灯时长，可以在这里计时并切换。
        # 简单的做法：如果不是我们控制的绿灯，就尝试找到第一个我们控制的绿灯并切换过去（如果空闲）
        # （这部分逻辑可以做的更完善）
        return # 暂时跳过
    
    current_phase_info = g_adaptive_tls_definitions[tls_id][active_phase_idx] # 获取当前相位信息
    time_in_current_phase = current_time - phase_start_time # 计算当前相位已持续时间
    
    # 1. 数据采集
    current_phase_total_queue = 0 # 初始化当前相位总排队
    # print(f"  TLS {tls_id} - 时间 {current_time:.1f}s - 当前相位: {active_phase_idx} ('{current_phase_info['name']}'), "
    #       f"已持续 {time_in_current_phase:.1f}s (minG={current_phase_info['min_green']}, maxG={current_phase_info['max_green']})") # 打印调试信息
    for lane_id in current_phase_info["monitored_lanes"]: # 遍历监控车道
        if lane_id in traci.lane.getIDList(): # 检查车道是否存在
            halting_num = traci.lane.getLastStepHaltingNumber(lane_id) # 获取停止车辆数
            current_phase_total_queue += halting_num # 累加
            # print(f"    车道 {lane_id}: 排队 = {halting_num}")
        # else:
            # print(f"    警告: TLS {tls_id} 监控的车道 {lane_id} 未找到。")
    
    # 2. 切换决策
    switch_phase_flag = False # 初始化切换标志
    if time_in_current_phase >= current_phase_info["min_green"]: # 如果达到最小绿灯时间
        if current_phase_total_queue <= QUEUE_CLEARED_THRESHOLD: # 并且队列已消散
            # print(f"    TLS {tls_id} 决策: 相位 {active_phase_idx} 队列已消散。")
            switch_phase_flag = True # 设置切换标志
        elif time_in_current_phase >= current_phase_info["max_green"]: # 或者达到最大绿灯时间
            # print(f"    TLS {tls_id} 决策: 相位 {active_phase_idx} 已达最大绿灯时间。")
            switch_phase_flag = True # 设置切换标志
    
    # 3. 选择并执行切换
    if switch_phase_flag: # 如果决定切换
        # print(f"    TLS {tls_id}: 执行相位切换...")
        candidate_queues = {} # 候选相位及其排队数字典
        for p_idx_contender, p_info_contender in g_adaptive_tls_definitions[tls_id].items(): # 遍历所有定义的相位
            if not p_info_contender.get("is_green_phase") or p_idx_contender == active_phase_idx: # 如果不是绿灯相位或就是当前相位，则跳过
                continue
            
            contender_queue = 0 # 初始化竞争者排队数
            for lane_id_c in p_info_contender["monitored_lanes"]: # 遍历其监控车道
                if lane_id_c in traci.lane.getIDList(): # 检查车道是否存在
                    contender_queue += traci.lane.getLastStepHaltingNumber(lane_id_c) # 累加排队数
            candidate_queues[p_idx_contender] = contender_queue # 存储候选相位及其排队数
            # print(f"      TLS {tls_id} 候选相位 {p_idx_contender} ('{p_info_contender['name']}'): 总排队 = {contender_queue}")
    
        if candidate_queues: # 如果有候选相位
            best_next_phase_idx = max(candidate_queues, key=candidate_queues.get) # 找到排队最长的候选相位
            
            # --- 黄灯和全红管理 (简化) ---
            # 一个真正的控制器需要在这里插入黄灯和全红相位。
            # 例如，可以定义一个 "transition_to_phase(tls_id, target_green_phase_idx)" 函数
            # 该函数会负责：
            # 1. 找到当前绿灯相位对应的黄灯相位。
            # 2. traci.trafficlight.setPhase(tls_id, yellow_phase)
            # 3. 在Python中等待黄灯时间 (通过多个simulationStep)。
            # 4. 切换到全红相位 (如果定义了)。
            # 5. 等待全红时间。
            # 6. traci.trafficlight.setPhase(tls_id, target_green_phase_idx)
            
            # 为了本示例的简单性，我们直接切换到下一个选定的绿灯相位。
            # 这假设了SUMO的相位定义中，黄灯/全红是作为独立相位存在的，并且
            # 我们在 phase_definitions 中正确识别了它们，或者 setPhase 能智能处理。
            # 如果没有正确处理黄灯，这在真实世界中是危险的！
            
            # 检查是否真的需要切换 (例如，如果下一个最佳相位也是当前相位，且队列未清空，可能只是延长)
            if best_next_phase_idx != active_phase_idx or current_phase_total_queue <= QUEUE_CLEARED_THRESHOLD: # 如果最佳下一相位不是当前相位，或者当前队列已清空
                print(f"  TLS {tls_id} 控制器: 时间 {current_time:.1f}s, 从相位 {active_phase_idx} "
                      f"切换到相位 {best_next_phase_idx} ('{g_adaptive_tls_definitions[tls_id][best_next_phase_idx]['name']}'). "
                      f"排队: 当前={current_phase_total_queue}, 目标={candidate_queues[best_next_phase_idx]}. "
                      f"(简化切换，未显式管理黄灯)") # 打印切换信息
                
                traci.trafficlight.setPhase(tls_id, best_next_phase_idx) # 切换到最佳下一相位
                g_active_phase_index[tls_id] = best_next_phase_idx # 更新活动相位索引
                g_current_phase_start_time[tls_id] = current_time # 更新相位开始时间
            # else:
                # print(f"    TLS {tls_id} 决策: 最佳下一相位仍是当前相位 {active_phase_idx}，但队列未清空且未到maxGreen。延长当前绿灯。")
    
        # else: # 没有其他候选绿灯相位
            # print(f"    TLS {tls_id} 警告: 没有其他候选绿灯相位可切换。")
    
    
    def run_adaptive_simulation(): # 定义自适应仿真的主函数
    """启动SUMO并带有自适应信号控制的仿真。"""
    global g_current_phase_start_time, g_active_phase_index # 声明使用全局变量
    
    if not use_libsumo_adaptive_ctrl: # 如果不使用libsumo
        sumo_cmd_adaptive = [
            SUMO_BINARY, "-c", sumo_config_file,
            "--remote-port", str(traci_port),
            "--step-length", "1", # 与 DECISION_INTERVAL 匹配或更小
            "--quit-on-end"
        ]
        print(f"请手动启动 SUMO: {' '.join(sumo_cmd_adaptive)}") # 提示手动启动SUMO
        input("按 Enter 确认 SUMO 已启动...") # 等待用户确认
    
        try:
            traci.connect(port=traci_port) # 连接到SUMO
            print("自适应控制: 连接 SUMO 成功。") # 打印连接成功
        except traci.exceptions.TraCIException as e: # 捕获连接异常
            print(f"自适应控制: 连接 SUMO 失败: {e}") # 打印连接失败
            return # 返回
    else: # 如果使用libsumo
        print("自适应控制: 使用 libsumo 启动仿真...") # 打印libsumo启动
        traci.start([SUMO_BINARY, "-c", sumo_config_file, "--step-length", "1", "--quit-on-end"]) # libsumo启动
        print("自适应控制: libsumo 仿真已启动。") # 打印启动成功
    
    # 初始化自适应控制器
    if not initialize_adaptive_tls_control(): # 调用初始化函数
        print("自适应信号控制初始化失败，将不执行控制逻辑。") # 打印初始化失败
        # traci.close() # 可以选择关闭连接
        # return
    else:
        print("\n自适应信号控制已初始化，准备开始主循环。\n") # 打印初始化成功
    
    step = 0 # 初始化步数
    simulation_end_time = float(traci.simulation.getParameter("", "end")) # 从SUMO获取配置的结束时间
    # 注意: getParameter 可能需要更具体的参数名，或者直接从sumocfg读取
    
    try:
        while traci.simulation.getTime() < simulation_end_time: # 当仿真时间小于结束时间时循环
            traci.simulationStep() # 执行一个仿真步
            current_sim_time_main = traci.simulation.getTime() # 获取当前仿真时间
            
            # 每隔 DECISION_INTERVAL 秒执行一次决策逻辑
            if step % int(DECISION_INTERVAL / float(traci.simulation.getDeltaT())) == 0 : # 判断是否到达决策时间点
                 # (假设 getDeltaT 返回的是秒，如果不是，需要调整)
                if TLS_ID_TO_CONTROL in g_adaptive_tls_definitions: # 如果要控制的信号灯在定义中
                    adaptive_tls_logic_step(TLS_ID_TO_CONTROL) # 执行自适应逻辑
    
            # (可以加入其他TraCI操作或数据记录)
            if step % 30 == 0 : # 每30步打印一次车辆总数作为心跳
                print(f"  主循环 心跳: 时间 {current_sim_time_main:.1f}s, 车辆数: {traci.vehicle.getIDCount()}") # 打印心跳信息
    
            step += 1 # 步数加1
            
            # if current_sim_time_main > 360: # 临时手动结束条件
            #     print("达到临时结束时间360s，退出。")
            #     break
    
    except traci.exceptions.TraCIException as e: # 捕获TraCI异常
        print(f"自适应仿真: 过程中发生 TraCI 错误: {e}") # 打印错误信息
    except KeyboardInterrupt: # 捕获键盘中断
        print("自适应仿真: 用户请求中断。") # 打印中断信息
    finally: # 最终执行块
        print("自适应仿真: 正在关闭 TraCI 连接...") # 打印关闭信息
        traci.close() # 关闭连接
        print("自适应仿真: TraCI 连接已关闭。") # 打印连接已关闭
    
    if __name__ == "__main__": # 如果是主程序
    if not os.path.exists(sumo_config_file): # 检查配置文件是否存在
        print(f"错误: SUMO 配置文件 '{sumo_config_file}' 未找到。") # 打印错误
    else:
        run_adaptive_simulation() # 自适应仿真
    
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/OTX0vJuZNWyg7cVfxsUon3tEYjGi.png)

 * 假设在一定的通信范围内，所有车辆都能即时、无误地获取其他车辆或设施的信息。
 * 在 Python 脚本中，可以通过 TraCI 查询所有相关对象的状态，然后将这些信息“分发”给需要它的车辆。
 * 例如，一辆车 `A` 可以通过 Python 脚本获取其周围一定范围内的其他车辆 `B`, `C` 的位置和速度，就如同它接收到了它们广播的 BSM 一样。
 * **优点** : 实现简单，专注于上层应用逻辑。
 * **缺点** : 忽略了通信的限制和不完美性。

 * 在 Python 脚本中，为每辆车定义一个通信范围 (例如 300 米)。
 * 在每个时间步，对于每辆车，计算其与其他车辆的距离。只有在通信范围内的车辆才被认为是“可通信”的。
 * 信息交换只在可通信的车辆之间进行。
 * 可以引入简单的概率模型来模拟丢包。
 * **优点** : 比理想模型更真实一些。
 * **缺点** : 仍然没有模拟具体的通信协议或信道竞争。

 * 例如 Veins (基于 OMNeT++ 和 SUMO) 或 Artery (基于 ETSI ITS-G5 协议栈)。这些工具提供了详细的无线通信层模拟 (如 IEEE 802.11p)。
 * SUMO 负责交通流模拟，OMNeT++ 负责通信模拟，两者通过 TraCI 或特定的接口耦合。
 * **优点** : 高度逼真的通信模拟。
 * **缺点** : 配置和使用更复杂，计算开销大。

    # (在 run_simulation 函数的主循环 while step < max_steps: 内部)
    # traci.simulationStep() # 已在循环开始处执行
    current_time = traci.simulation.getTime() # 获取当前仿真时间
    
    COMMUNICATION_RANGE = 100 # 米，定义车辆的V2V通信范围
    all_vehicle_ids = traci.vehicle.getIDList() # 获取当前所有车辆的ID
    
    # 构建一个字典来存储每辆车的“感知”信息
    # vehicle_knowledge_base = {veh_id: {"position": (x,y), "speed": float, "angle": float, 
    #                                     "type": str, "neighbors": {neighbor_id: neighbor_info_dict}}}
    vehicle_knowledge_base = {} # 初始化车辆知识库字典
    
    # 1. 每个车辆“广播”其自身状态 (即，我们从TraCI获取它们的状态)
    for veh_id in all_vehicle_ids: # 遍历所有车辆
    try:
        pos = traci.vehicle.getPosition(veh_id) # 获取车辆位置
        speed = traci.vehicle.getSpeed(veh_id) # 获取车辆速度
        angle = traci.vehicle.getAngle(veh_id) # 获取车辆角度
        veh_type = traci.vehicle.getTypeID(veh_id) # 获取车辆类型
        
        vehicle_knowledge_base[veh_id] = { # 在知识库中存储该车信息
            "id": veh_id, # 车辆ID
            "position": pos, # 位置
            "speed": speed, # 速度
            "angle": angle, # 角度
            "type": veh_type, # 类型
            "received_bsms": [] # 初始化一个列表，用于存放从邻车接收到的“BSM”信息
        }
    except traci.exceptions.TraCIException: # 捕获异常 (车辆可能刚离开)
        if veh_id in vehicle_knowledge_base: # 如果车辆仍在知识库中
            del vehicle_knowledge_base[veh_id] # 将其从知识库中移除
        continue # 继续下一个车辆
    
    # 2. 模拟信息交换：每个车辆“接收”其通信范围内的其他车辆的“BSM”
    #    在我们的Python脚本中，这意味着对于每个车辆，我们查找其范围内的其他车辆
    #    并将其他车辆的信息添加到它的 "received_bsms" 列表中。
    if current_time % 1 == 0: # 例如，每隔1秒进行一次信息交换的模拟 (BSM通常是10Hz，即0.1秒一次)
    print(f"\n--- 时间 {current_time:.1f}s: 模拟V2V BSM信息交换 (通信范围: {COMMUNICATION_RANGE}m) ---") # 打印信息交换标题
    for sender_id, sender_data in list(vehicle_knowledge_base.items()): # 遍历知识库中的每个发送者车辆
        if sender_id not in traci.vehicle.getIDList(): # 如果发送者已离开仿真
            if sender_id in vehicle_knowledge_base: del vehicle_knowledge_base[sender_id] # 从知识库中删除
            continue # 跳过
    
        # sender_data["received_bsms"].clear() # 清除上一轮接收的BSM (可选，取决于是否需要累积历史)
    
        for receiver_id, receiver_data in list(vehicle_knowledge_base.items()): # 遍历知识库中的每个接收者车辆
            if sender_id == receiver_id: # 如果发送者和接收者是同一辆车，则跳过
                continue
            if receiver_id not in traci.vehicle.getIDList(): # 如果接收者已离开仿真
                if receiver_id in vehicle_knowledge_base: del vehicle_knowledge_base[receiver_id] # 从知识库中删除
                continue # 跳过
    
            # 计算发送者和接收者之间的距离
            dist_sq = (sender_data["position"][0] - receiver_data["position"][0])**2 + \
                      (sender_data["position"][1] - receiver_data["position"][1])**2 # 计算距离的平方
            distance = dist_sq**0.5 # 计算实际距离
    
            if distance <= COMMUNICATION_RANGE: # 如果在通信范围内
                # 模拟接收者 receiver_id 接收到了发送者 sender_id 的BSM
                # 我们将发送者的完整信息（或选定信息）添加到接收者的 received_bsms 列表
                bsm_received_by_receiver = sender_data.copy() # 创建发送者数据的一个副本作为BSM
                del bsm_received_by_receiver["received_bsms"] # BSM本身不包含其他车的BSM列表
                
                # 确保 receiver_data 仍然有效并且有 "received_bsms" 键
                if receiver_id in vehicle_knowledge_base and "received_bsms" in vehicle_knowledge_base[receiver_id]:
                     vehicle_knowledge_base[receiver_id]["received_bsms"].append(bsm_received_by_receiver) # 将BSM添加到接收者的列表中
                     # print(f"    车辆 {receiver_id} 接收到来自 {sender_id} 的BSM。") # 打印接收信息
                # else:
                     # print(f"    警告: 车辆 {receiver_id} 在尝试接收BSM时已不在知识库中或结构不符。")
    
    
    # 3. 基于接收到的信息做出决策 (示例：如果前方近距离有慢车，则减速或准备换道)
    #    这部分是协同驾驶行为的核心，将在后续示例中展开。
    #    现在，我们先打印一些车的感知结果：
    if vehicle_knowledge_base and current_time % 5 == 0: # 每5秒打印一次部分车辆的感知信息
    print(f"  部分车辆在时间 {current_time:.1f}s 的感知情况:") # 打印感知情况标题
    for i, (veh_id, data) in enumerate(vehicle_knowledge_base.items()): # 遍历知识库中的车辆
        if i >= 3: break # 只打印前3辆车的信息以避免过多输出
        print(f"    车辆 {veh_id} (位置: ({data['position'][0]:.1f},{data['position'][1]:.1f}), 速度: {data['speed']:.1f} m/s):") # 打印车辆基本信息
        if data["received_bsms"]: # 如果该车接收到了BSM
            print(f"      接收到 {len(data['received_bsms'])} 条BSM:") # 打印接收到的BSM数量
            for bsm in data["received_bsms"][:2]: # 只打印前2条BSM的发送者信息
                print(f"        来自: {bsm['id']} (位置: ({bsm['position'][0]:.1f},{bsm['position'][1]:.1f}), 速度: {bsm['speed']:.1f} m/s)") # 打印BSM发送者信息
        else:
            print(f"      未接收到来自其他车辆的BSM。") # 打印未接收到BSM信息
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/5AubzBYSGdIVeTOKH4rZMhWiqmxk.png)

    # (在 run_simulation 函数的主循环，在上述BSM交换逻辑之后)
    
    # --- 协同紧急制动逻辑 ---
    BRAKING_DECELERATION_THRESHOLD = -3.0 # m/s^2，定义一个减速度阈值，低于此值认为是在紧急制动
    ANTICIPATORY_DECEL_FACTOR = 0.8 # 当收到前车紧急制动信号时，本车采取的减速度是前车减速度的百分比
    MIN_REACTION_GAP = 5.0 # 米，即使收到信号，也要保持的最小车头间距
    
    for ego_veh_id, ego_data in list(vehicle_knowledge_base.items()): # 遍历每辆车作为“本车”(ego vehicle)
    if ego_veh_id not in traci.vehicle.getIDList(): # 如果本车已离开
        if ego_veh_id in vehicle_knowledge_base: del vehicle_knowledge_base[ego_veh_id] # 从知识库删除
        continue # 跳过
    
    if not ego_data["received_bsms"]: # 如果本车没有收到任何BSM
        continue # 跳过，无法进行协同
    
    # 检查是否有前导车辆 (在同一车道，或者在即将进入的路径上)
    # 为了简化，我们只考虑从收到的BSM中判断“前方”车辆
    # 一个更精确的方法是结合 getLeader 和 BSM 信息
    
    potential_leaders_in_bsm = [] # 存储从BSM中识别出的潜在前车列表
    for bsm in ego_data["received_bsms"]: # 遍历本车收到的所有BSM
        # 判断 bsm['id'] 是否是 ego_veh_id 的前车
        # 简化判断：1. 在 ego_veh_id 的大致前方 (基于角度和相对位置)
        #           2. 距离较近
        # (这是一个非常简化的前方判断，实际需要更复杂的几何计算或使用SUMO的getLeader)
        
        # 假设我们通过某种方式 (例如，比较角度和相对位置，或者BSM中包含车道信息)
        # 确定了bsm['id']是ego_veh_id前方的一个车辆
        # 这里我们用一个占位符，实际应用需要实现 is_vehicle_ahead(ego_data, bsm_data)
        
        # 粗略的前方判断：sender 在 ego 前方大致相同方向
        # ego_angle_rad = math.radians(ego_data['angle'])
        # bsm_angle_rad = math.radians(bsm['angle'])
        # dx = bsm['position'][0] - ego_data['position'][0]
        # dy = bsm['position'][1] - ego_data['position'][1]
        # angle_to_bsm_veh_rad = math.atan2(dy, dx)
        # angle_diff = abs(ego_angle_rad - angle_to_bsm_veh_rad)
        # if angle_diff < math.pi / 4: # 角度差小于45度，认为大致同向
            # 并且 bsm 在 ego 的“前方投影”上
            # dot_product = dx * math.cos(ego_angle_rad) + dy * math.sin(ego_angle_rad)
            # if dot_product > 0: # 确保在前方
                # potential_leaders_in_bsm.append(bsm)
        
        # 更简单的方法：如果BSM消息中包含了发送者的加速度信息
        # 我们可以直接检查这个加速度。
        # 假设BSM中包含 "acceleration" 字段 (需要我们之前从TraCI获取并存入)
        # 例如，在填充 vehicle_knowledge_base 时:
        # 'acceleration': traci.vehicle.getAcceleration(veh_id)
    
        # 为了演示，我们假设BSM消息中已经有了 'acceleration'
        # 并且我们只关心那些正在急减速的BSM发送者
        # if "acceleration" in bsm and bsm["acceleration"] < BRAKING_DECELERATION_THRESHOLD:
        #    potential_leaders_in_bsm.append(bsm) # 将急减速的车辆视为潜在“危险”前车
        
        # 实际应用中，SUMO的 getLeader 更可靠
        leader_traci = traci.vehicle.getLeader(ego_veh_id, 0) # 获取ego的直接前车
        if leader_traci and leader_traci[0] is not None: # 如果通过TraCI找到了直接前车
            leader_id_traci = leader_traci[0] # 直接前车的ID
            # 找到该前车在BSM列表中的信息 (如果它在通信范围内并发送了BSM)
            for bsm_from_leader in ego_data["received_bsms"]: # 遍历收到的BSM
                if bsm_from_leader["id"] == leader_id_traci: # 如果BSM来自直接前车
                    # 获取前车的加速度 (假设已通过 TraCI 获取并存储在 BSM 模拟数据中)
                    # 如果没有在BSM中存，可以实时查询: leader_accel = traci.vehicle.getAcceleration(leader_id_traci)
                    # 这里假设bsm_from_leader中已经有 'acceleration' 字段
                    leader_accel = vehicle_knowledge_base.get(leader_id_traci, {}).get("acceleration", 0) # 从知识库获取前车加速度
                    
                    if leader_accel < BRAKING_DECELERATION_THRESHOLD: # 如果前车正在紧急制动
                        distance_to_leader = leader_traci[1] # 本车车头到前车车尾的距离
                        print(f"  协同制动警告: 车辆 {ego_veh_id} 的前车 {leader_id_traci} 正在紧急制动! "
                              f"(减速度: {leader_accel:.2f}m/s^2, 距离: {distance_to_leader:.2f}m)") # 打印警告信息
                        
                        # ego_veh_id 应该采取行动
                        # 1. 计算一个期望的减速度
                        ego_current_speed = ego_data["speed"] # 本车当前速度
                        # 注意：setSpeed 通常是设置期望速度，直接设置减速度更复杂
                        # SUMO的跟驰模型会自动处理减速，但这里我们想模拟更“主动”的协同减速
                        
                        # 简单的策略：如果本车速度大于0，尝试以一定比例跟随前车的减速度
                        # 但不能超过车辆自身的物理极限，并且要考虑安全距离
                        if ego_current_speed > 1.0 and distance_to_leader > MIN_REACTION_GAP: # 如果速度大于1且距离大于最小反应间隙
                            # 尝试设置一个比正常跟车更强的减速度，但不要直接碰撞
                            # SUMO 的 setSpeed 命令会使车辆尝试达到该速度，其内置模型会计算减速度
                            # 我们可能需要逐渐降低目标速度，或者使用 setSpeedMode 来允许更积极的减速
                            
                            # 一个更直接的方式是影响车辆的 "imperfection" (sigma) 或其他跟驰参数
                            # traci.vehicle.setImperfection(ego_veh_id, 0.0) # 使其驾驶更“完美”，反应更快
                            # traci.vehicle.setSpeedFactor(ego_veh_id, 0.8) # 降低其期望速度因子
                            
                            # 或者，我们可以计算一个非常低的目标速度，迫使其减速
                            # 比如，目标速度设为0，让跟驰模型去处理
                            traci.vehicle.setSpeed(ego_veh_id, 0.0) # 命令车辆尝试停车
                            # 也可以通过设置一个具体的减速度，但TraCI没有直接的setDeceleration命令
                            # 可以通过 traci.vehicle.setSpeedMode 配合 setStop 实现
                            # traci.vehicle.setStop(ego_veh_id, edgeID=traci.vehicle.getRoadID(ego_veh_id), 
                            #                      pos=traci.vehicle.getLanePosition(ego_veh_id) + 0.1, # 停在当前位置稍前
                            #                      duration=1, decel=abs(leader_accel * ANTICIPATORY_DECEL_FACTOR),
                            #                      flags=traci.constants.STOP_DEFAULT)
                            # 上面的setStop可能过于激进。setSpeed(ego_veh_id, 0) 更依赖于内置模型。
                            
                            print(f"    -> 车辆 {ego_veh_id} 因前车协同信号，尝试紧急减速 (目标速度设为0)。") # 打印减速尝试信息
                            
                            # 可以在这里改变车辆颜色以示警告状态
                            traci.vehicle.setColor(ego_veh_id, (255, 165, 0, 255)) # 设置为橙色
                        break # 已经处理了直接前车的紧急制动，跳出BSM循环
    
    # 清理上一轮的BSM数据 (如果不想累积)
    if ego_veh_id in vehicle_knowledge_base: # 确保车辆仍然存在
        vehicle_knowledge_base[ego_veh_id]["received_bsms"].clear() # 清空接收到的BSM列表
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/Y1XzkdtBIgNxyGFam7Jq52TUQfsl.png)

    # 在填充 vehicle_knowledge_base 时:

    # ...
    accel = traci.vehicle.getAcceleration(veh_id) # 获取车辆加速度
    vehicle_knowledge_base[veh_id] = {
    # ... 其他字段 ...
    "acceleration": accel, # 存储加速度
    "received_bsms": []
    }
    # ...
    
    
    python
    
    

    # (在 run_simulation 函数的主循环内)
    
    # --- GLOSA (Green Light Optimal Speed Advisory) 模拟 ---
    TLS_ID_FOR_GLOSA = "N2" # 我们关注的信号灯ID (与自适应控制的TLS可以相同或不同)
    GLOSA_COMM_RANGE = 200 # 米，车辆能够接收到该交叉口SPaT消息的最大距离
    DEFAULT_SPEED_LIMIT_GLOSA = 13.89 # m/s (50km/h)，用于计算的默认限速，实际应从车道获取
    
    # 1. I2V: 模拟交叉口广播SPaT消息
    #    我们将在每个仿真步获取N2的状态，并将其“提供”给范围内的车辆
    spat_message_for_n2 = None # 初始化SPaT消息为空
    if TLS_ID_FOR_GLOSA in traci.trafficlight.getIDList(): # 如果信号灯存在
    try:
        current_phase_idx_n2 = traci.trafficlight.getPhase(TLS_ID_FOR_GLOSA) # 获取N2当前相位索引
        current_phase_state_str_n2 = traci.trafficlight.getRedYellowGreenState(TLS_ID_FOR_GLOSA) # 获取当前灯色状态
        
        # 获取当前相位的剩余时间。这比较复杂，因为SUMO不直接提供“相位剩余时间”的TraCI命令
        # 如果是固定周期，可以从 getCompleteRedYellowGreenDefinition 获取各相位时长，然后自己计算。
        # 如果是自适应控制 (如我们之前实现的)，剩余时间取决于控制逻辑何时切换。
        
        # 简化：假设我们能通过某种方式获取或估计当前绿灯相位的剩余时间
        # For fixed-time or if our adaptive controller tracks phase end time:
        # time_to_next_switch_n2 = traci.trafficlight.getNextSwitch(TLS_ID_FOR_GLOSA) - current_time
        # 如果 getNextSwitch 返回的是绝对时间
        
        # 作为一个更简单的（但不完全准确的）代理，我们可以使用当前活动相位的定义信息
        # (这部分需要与我们之前的自适应控制逻辑或固定的信号灯逻辑结合)
        time_remaining_in_current_phase = -1 # 初始化为-1 (未知)
        next_phase_state_after_current = "UNKNOWN" # 初始化下一相位状态
        
        # 尝试从我们之前为自适应控制构建的 g_adaptive_tls_definitions 获取信息
        # 或者从 getCompleteRedYellowGreenDefinition 获取固定时序信息
        tls_logic_for_glosa = None
        if TLS_ID_FOR_GLOSA in g_adaptive_tls_definitions and g_adaptive_tls_definitions[TLS_ID_FOR_GLOSA]: # 如果在我们自适应控制的定义中
            tls_logic_for_glosa_phases = g_adaptive_tls_definitions[TLS_ID_FOR_GLOSA]
            active_phase_for_glosa_idx = g_active_phase_index.get(TLS_ID_FOR_GLOSA, -1) # 获取当前活动相位
            
            if active_phase_for_glosa_idx != -1 and active_phase_for_glosa_idx in tls_logic_for_glosa_phases: # 如果活动相位有效
                current_glosa_phase_info = tls_logic_for_glosa_phases[active_phase_for_glosa_idx] # 获取当前相位信息
                phase_start_for_glosa = g_current_phase_start_time.get(TLS_ID_FOR_GLOSA, current_time) # 获取相位开始时间
                elapsed_in_phase = current_time - phase_start_for_glosa # 计算已过时间
                
                # 估算剩余时间：如果是我们控制的绿灯，它会持续到max_green或队列清空
                # 这是一个粗略的估计，实际GLOSA需要更精确的SPaT
                if current_glosa_phase_info.get("is_green_phase"): # 如果是绿灯相位
                    # 假设它至少会持续到min_green，或者如果队列未清空，可能到max_green
                    # 这里的剩余时间计算很复杂，取决于自适应逻辑何时会切换
                    # 简化：假设当前绿灯会持续其“名义上”的下一个切换点（可能是max_green之后）
                    # 这是一个非常大的简化！
                    estimated_duration_of_current = current_glosa_phase_info.get("max_green", 30) # 获取最大绿灯时间或默认30秒
                    time_remaining_in_current_phase = max(0, estimated_duration_of_current - elapsed_in_phase) # 计算剩余时间
                
                # 获取下一个主要绿灯相位的状态 (这也需要对相位序列有了解)
                # (省略了复杂逻辑，实际SPaT会提供更完整的未来相位信息)
            
        elif traci.trafficlight.getProgram(TLS_ID_FOR_GLOSA) is not None: # 如果有固定程序
             # 如果是固定时序，可以解析 getCompleteRedYellowGreenDefinition
             # 并跟踪当前相位在周期中的位置来计算剩余时间
             # (这部分逻辑也比较复杂，暂时跳过精确计算)
             # 我们可以使用 traci.trafficlight.getNextSwitch(TLS_ID_FOR_GLOSA)
            next_switch_abs_time = traci.trafficlight.getNextSwitch(TLS_ID_FOR_GLOSA) # 获取下一次切换的绝对时间
            if next_switch_abs_time > current_time: # 如果切换时间在未来
                time_remaining_in_current_phase = next_switch_abs_time - current_time # 计算剩余时间
            # 获取切换后的相位状态 (需要知道切换后的相位索引，然后查其state)
            # (这也很复杂)
    
        # 构建SPaT消息 (简化版)
        spat_message_for_n2 = { # 创建SPaT消息字典
            "tls_id": TLS_ID_FOR_GLOSA, # 信号灯ID
            "current_phase_index": current_phase_idx_n2, # 当前相位索引
            "current_phase_state": current_phase_state_str_n2, # 当前灯色状态
            "time_remaining_current_phase": time_remaining_in_current_phase, # 当前相位剩余时间 (估算)
            "next_main_green_start_time": -1, # 下一个主绿灯开始时间 (估算, -1表示未知)
            "next_main_green_duration": -1    # 下一个主绿灯持续时间 (估算, -1表示未知)
        }
        # 一个真实的SPaT消息会包含未来多个相位的详细信息 (startTime, minEndTime, maxEndTime, state)
        
        # (打印SPaT消息用于调试)
        # if current_time % 10 == 0: 
        #     print(f"  SPaT for {TLS_ID_FOR_GLOSA} at {current_time:.1f}s: {spat_message_for_n2}")
    
    except traci.exceptions.TraCIException: # 捕获TraCI异常
        spat_message_for_n2 = None # 如果出错，SPaT消息设为None
        pass # 忽略错误，继续
    
    # 2. V2I: 车辆接收并处理SPaT，计算建议速度
    if spat_message_for_n2: # 如果成功获取/构建了SPaT消息
    for veh_id, veh_data in list(vehicle_knowledge_base.items()): # 遍历所有车辆
        if veh_id not in traci.vehicle.getIDList(): # 如果车辆已离开
             if veh_id in vehicle_knowledge_base: del vehicle_knowledge_base[veh_id] # 从知识库删除
             continue # 跳过
    
        try:
            # 检查车辆是否在GLOSA通信范围内，并且是否正在接近该交叉口
            # (需要知道交叉口的位置)
            tls_pos_n2 = traci.junction.getPosition(TLS_ID_FOR_GLOSA) # 获取N2交叉口的位置
            veh_pos = veh_data["position"] # 获取车辆位置
            dist_to_tls_sq = (veh_pos[0] - tls_pos_n2[0])**2 + (veh_pos[1] - tls_pos_n2[1])**2 # 计算到交叉口距离的平方
            dist_to_tls = dist_to_tls_sq**0.5 # 计算实际距离
    
            if dist_to_tls <= GLOSA_COMM_RANGE: # 如果车辆在通信范围内
                # 并且车辆的路径会经过这个交叉口 (简化：检查路径中是否有与交叉口相关的边)
                # (一个更精确的方法是检查车辆的当前路径是否会通过N2控制的某个进口道)
                is_approaching_n2 = False # 初始化为不接近N2
                current_veh_route = traci.vehicle.getRoute(veh_id) # 获取车辆当前路径
                # N2控制的进口边有E1, E3。出口边有E2, E4。
                # 如果车辆的路径包含 E1 或 E3，并且其后是 E2 或 E4 (通过N2)
                # 或者车辆当前就在 E1 或 E3 上，且未通过 N2
                current_road_id = traci.vehicle.getRoadID(veh_id) # 获取车辆当前所在边
                if current_road_id in ["E1", "E3"]: # 如果车辆在N2的进口道上
                    is_approaching_n2 = True # 标记为正在接近N2
                # (这个判断可以更完善)
    
                if is_approaching_n2: # 如果车辆正在接近N2
                    # print(f"  GLOSA: 车辆 {veh_id} (距离N2 {dist_to_tls:.1f}m) 正在处理SPaT。")
                    
                    # 获取车辆当前车道的限速 (作为建议速度的上限)
                    veh_lane_id = traci.vehicle.getLaneID(veh_id) # 获取车辆当前车道ID
                    speed_limit_on_lane = traci.lane.getMaxSpeed(veh_lane_id) # 获取车道限速
    
                    advised_speed = -1 # 初始化建议速度为-1 (不建议或无法计算)
                    current_phase_state = spat_message_for_n2["current_phase_state"] # 获取SPaT中的当前灯色状态
                    time_rem_current = spat_message_for_n2["time_remaining_current_phase"] # 获取当前相位剩余时间
    
                    # 判断车辆对于N2的进口道是哪个（例如，从E1来的，或从E3来的）
                    # 这需要知道当前信号灯状态字符串中，哪个字符对应本车的进口道
                    # 例如，如果车辆在E1_0上，而E1_0是 current_phase_state 字符串的第一个字符控制的
                    # my_signal_char = current_phase_state[0] (假设)
                    # 这是最复杂的部分：将车辆的当前运动与SPaT消息中的具体灯色关联起来。
                    # 简化：我们假设如果 `current_phase_state` 包含 'G' 或 'g'，就认为是绿灯。
                    # 这是一个非常粗略的假设，实际需要精确匹配。
    
                    is_current_green_for_veh = 'g' in current_phase_state.lower() # 简单判断当前是否为绿灯（可能对本车是红）
                                                                                # 需要更精确的判断：本车进口道的灯色是什么
                    
                    # 真正的判断：获取车辆当前车道连接到交叉口的link的索引，
                    # 然后用这个索引去查 current_phase_state 字符串中对应的字符。
                    # links = traci.vehicle.getLinks(veh_id) # 获取车辆下一个可能的连接 [(lane, via, toLane, isOpen, hasPrio, state, direction, length), ...]
                    # if links:
                    #    first_link_state = links[0][5] # 获取第一个连接的信号灯状态 (r,y,g,G,s,o)
                    #    is_current_green_for_veh = first_link_state.lower() == 'g'
                    # else: is_current_green_for_veh = False
    
                    # (为了示例的继续，我们还是用简化的 is_current_green_for_veh)
                    # 你需要用上述更精确的方法替换它
    
                    dist_to_stop_line = dist_to_tls # 简化：假设到交叉口中心的距离约等于到停车线的距离
                                                    # 精确值：traci.vehicle.getStopLine आगे की दूरी (需要SUMO 1.x)
                                                    # 或 traci.lane.getLength(veh_lane_id) - traci.vehicle.getLanePosition(veh_id)
                                                    # (如果停车线在车道末端)
                    
                    if is_current_green_for_veh and time_rem_current > 0: # 如果对本车是绿灯且还有剩余时间
                        # 尝试在绿灯结束前通过
                        time_to_pass_at_current_speed = dist_to_stop_line / veh_data["speed"] if veh_data["speed"] > 0.1 else float('inf') # 以当前速度通过所需时间
                        if time_to_pass_at_current_speed <= time_rem_current: # 如果能在绿灯结束前通过
                            advised_speed = veh_data["speed"] # 建议保持当前速度 (或略微调整)
                            # print(f"    GLOSA for {veh_id}: 绿灯，建议速度 {advised_speed:.2f} m/s (保持当前)。")
                        else:
                            # 无法在当前绿灯通过，计算在下一个绿灯开始时到达的速度
                            # (这需要SPaT提供下一个绿灯的信息，目前我们的简化SPaT没有)
                            # 简化：建议减速以安全停车
                            advised_speed = max(0, min(speed_limit_on_lane, dist_to_stop_line / (time_rem_current + 2.0))) # 尝试在绿灯结束+2秒缓冲内到达
                                                                                                                        # 这是一个非常粗略的减速建议
                            # print(f"    GLOSA for {veh_id}: 绿灯时间不足，建议减速至 {advised_speed:.2f} m/s。")
                    
                    elif not is_current_green_for_veh or time_rem_current <= 0: # 如果是红灯/黄灯，或当前绿灯即将结束
                        # 计算到达下一个绿灯周期开始时的建议速度
                        # 假设我们知道下一个绿灯将在 time_to_next_green 秒后开始
                        # (这需要SPaT提供，我们的简化SPaT没有。假设一个固定周期或从自适应逻辑获取)
                        
                        # 简化：假设下一个绿灯在 time_rem_current (如果是黄灯转红) + 一个固定红灯时长之后
                        estimated_time_to_next_green = -1 # 初始化为未知
                        if time_rem_current > 0 and 'y' in current_phase_state.lower(): # 如果是黄灯
                            estimated_time_to_next_green = time_rem_current + 20 # 黄灯时间 + 假设20秒红灯
                        elif 'r' in current_phase_state.lower(): # 如果是红灯
                            # 如果知道当前红灯相位何时结束 (例如，固定周期)
                            estimated_time_to_next_green = time_rem_current if time_rem_current > 0 else 20 # 当前红灯剩余或假设20秒
                        
                        if estimated_time_to_next_green > 0: # 如果能估计到下一个绿灯的时间
                            required_speed_to_arrive_on_green = dist_to_stop_line / estimated_time_to_next_green # 到达下一个绿灯所需速度
                            advised_speed = max(1.0, min(required_speed_to_arrive_on_green, speed_limit_on_lane)) # 确保速度不为0且不超过限速
                            # print(f"    GLOSA for {veh_id}: 红/黄灯，下一绿灯约在 {estimated_time_to_next_green:.1f}s 后，建议速度 {advised_speed:.2f} m/s。")
                        else:
                            # 无法预测下一绿灯，建议安全停车
                            advised_speed = 0.0 # 建议停车
                            # print(f"    GLOSA for {veh_id}: 红/黄灯，无法预测下一绿灯，建议停车。")
    
                    # 3. 应用建议速度 (如果有效)
                    if advised_speed >= 0: # 如果计算出了有效的建议速度
                        # 为了平滑，不要让建议速度变化太剧烈
                        current_actual_speed = veh_data["speed"] # 获取车辆当前实际速度
                        # final_speed_to_set = (current_actual_speed * 0.5 + advised_speed * 0.5) # 平滑处理
                        final_speed_to_set = advised_speed # 直接设置（跟驰模型会处理平滑性）
                        
                        # 确保不超过车辆自身最大速度
                        veh_max_speed_prop = traci.vehicletype.getMaxSpeed(veh_data["type"]) # 获取车辆类型的最大速度
                        final_speed_to_set = min(final_speed_to_set, veh_max_speed_prop, speed_limit_on_lane) # 不超过车辆最大速度和车道限速
                        
                        traci.vehicle.setSpeed(veh_id, final_speed_to_set) # 设置车辆的期望速度
                        # print(f"      -> 车辆 {veh_id} 应用GLOSA建议速度: {final_speed_to_set:.2f} m/s (原速度: {current_actual_speed:.2f} m/s)")
                        if abs(final_speed_to_set - current_actual_speed) > 1.0 : # 如果速度变化较大，改变颜色提示
                             traci.vehicle.setColor(veh_id, (0, 255, 255, 255)) # 设置为青色表示正在响应GLOSA
                    # else:
                        # print(f"    GLOSA for {veh_id}: 未能计算出建议速度。")
                        
        except traci.exceptions.TraCIException: # 捕获TraCI异常
            # print(f"  处理GLOSA或获取车辆 {veh_id} 信息时出错。")
            if veh_id in vehicle_knowledge_base: del vehicle_knowledge_base[veh_id] # 从知识库删除
            pass # 忽略错误，继续
    
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/COTasg5eVQZJNWAxqck8YL0UDrl7.png)

    # (在 run_simulation 函数的主循环内，BSM交换逻辑之后)
    
    # --- 协同并道/换道参数 ---
    MERGING_REQUEST_DISTANCE = 50 # 米, 匝道车在距离并道点多远时开始请求
    MIN_SAFE_GAP_FOR_MERGE = 10 # 米, 认为安全的最小并道间隙 (车头到车尾)
    COOP_ACCEL_RV = 1.0       # m/s^2, 主路车为配合而采取的（最大）加速度
    COOP_DECEL_RV = -1.5      # m/s^2, 主路车为配合而采取的（最大）减速度
    
    # 假设我们的路网中:
    # "E_ramp" 是一条匝道边，其末端连接到主路边 "E_main"
    # 车辆类型 "merging_car" 在匝道上行驶, "main_road_car" 在主路上行驶
    # (这些需要在 .rou.xml 或 .net.xml 中定义)
    
    # 1. 识别匝道车 (MV) 和主路车 (RV)
    merging_vehicles = {} # {mv_id: {"target_lane_on_main": lane_id, "dist_to_merge_point": float, "requesting_merge": bool}}
    main_road_vehicles_near_merge = {} # {rv_id: {"lane_id": str, "position": float, "speed": float}}
    
    # 获取匝道末端连接到主路的连接信息 (需要具体路网知识)
    # 简化：我们假设匝道 E_ramp 的最后一个节点是 merge_node_id
    # 并且从 merge_node_id 到主路 E_main 的连接是已知的
    # 或者，我们可以检查车辆的路径
    # 假设 E_ramp 的出口连接到 E_main 的车道 E_main_0 或 E_main_1
    
    RAMP_EDGE_ID = "E_ramp" # 示例匝道边ID
    MAIN_ROAD_EDGE_ID_AFTER_RAMP = "E_main" # 示例并道后的主路边ID
    
    for veh_id, veh_data in list(vehicle_knowledge_base.items()): # 遍历所有车辆
    if veh_id not in traci.vehicle.getIDList(): # 如果车辆已离开
        if veh_id in vehicle_knowledge_base: del vehicle_knowledge_base[veh_id] # 从知识库删除
        continue # 跳过
    
    try:
        current_road = traci.vehicle.getRoadID(veh_id) # 获取车辆当前所在边
        current_lane = traci.vehicle.getLaneID(veh_id) # 获取车辆当前所在车道
        
        if current_road == RAMP_EDGE_ID: # 如果车辆在匝道上
            # 计算到匝道末端(并道点)的距离
            lane_len = traci.lane.getLength(current_lane) # 获取匝道车道长度
            veh_pos_on_lane = traci.vehicle.getLanePosition(veh_id) # 获取车辆在车道上的位置
            dist_to_merge_pt = lane_len - veh_pos_on_lane # 计算到并道点的距离
            
            requesting = (dist_to_merge_pt <= MERGING_REQUEST_DISTANCE) # 如果距离小于请求距离，则标记为请求并道
            
            # 确定MV希望并入主路的哪条车道 (通常是最右侧)
            # 这需要知道匝道连接到主路的具体车道。假设它总是连接到主路的最右侧车道 (索引0)
            target_main_lane = f"{MAIN_ROAD_EDGE_ID_AFTER_RAMP}_0" # 假设目标是主路E_main的0号车道
    
            merging_vehicles[veh_id] = { # 存储匝道车信息
                "data": veh_data, # 原始车辆数据 (位置，速度等)
                "target_lane_on_main": target_main_lane, # 目标主路车道
                "dist_to_merge_point": dist_to_merge_pt, # 到并道点距离
                "requesting_merge": requesting, # 是否正在请求并道
                "is_merging_now": False # 标记是否正在执行并道动作
            }
            if requesting: # 如果正在请求并道
                print(f"  协同并道: 匝道车 {veh_id} (距并道点 {dist_to_merge_pt:.1f}m) 请求并入 {target_main_lane}。") # 打印请求信息
                traci.vehicle.setColor(veh_id, (255,0,255,255)) # 将匝道车设为品红色以突出显示
    
        elif current_road == MAIN_ROAD_EDGE_ID_AFTER_RAMP: # 如果车辆在并道点之后的主路上
            # 也可以考虑并道点之前的主路车辆
            # 这里简化为只考虑在目标主路段上的车
            main_road_vehicles_near_merge[veh_id] = { # 存储主路车信息
                "data": veh_data, # 原始数据
                "lane_id": current_lane, # 所在车道
                "position_on_lane": traci.vehicle.getLanePosition(veh_id) # 在车道上的位置
            }
            
    except traci.exceptions.TraCIException: # 捕获TraCI异常
        pass # 忽略错误，继续
    
    # 2. 主路车 (RV) 响应匝道车 (MV) 的请求
    for mv_id, mv_info in list(merging_vehicles.items()): # 遍历所有匝道车
    if not mv_info["requesting_merge"] or mv_info["is_merging_now"]: # 如果匝道车不请求或已在并道，则跳过
        continue
    if mv_id not in traci.vehicle.getIDList(): # 如果匝道车已离开
        if mv_id in merging_vehicles: del merging_vehicles[mv_id] # 从列表删除
        continue # 跳过
    
    # 获取匝道车的当前状态
    mv_data = mv_info["data"] # 匝道车数据
    mv_target_main_lane = mv_info["target_lane_on_main"] # 匝道车的目标主路车道
    
    # 查找在目标主路车道上，MV并道点附近的RV (前方和后方)
    # 我们需要估计MV到达并道点时，它会在主路上的哪个位置，以及它周围的RV
    # 这是一个复杂的预测问题。简化：我们关注当前在MV目标车道上的，且在MV通信范围内的RV。
    
    # 假设并道点在主路 E_main 的起点附近 (pos_on_main_road = 0)
    # MV到达并道点的时间约等于 mv_info["dist_to_merge_point"] / mv_data["speed"] (如果速度不变)
    
    rv_leader_in_target_lane = None # 初始化目标车道上的前导RV为空
    rv_follower_in_target_lane = None # 初始化目标车道上的跟随RV为空
    min_dist_to_leader_sq = float('inf') # 初始化到前导RV的最小距离平方
    min_dist_to_follower_sq = float('inf') # 初始化到跟随RV的最小距离平方
    
    for rv_id, rv_info in list(main_road_vehicles_near_merge.items()): # 遍历附近的主路车
        if rv_id not in traci.vehicle.getIDList(): # 如果主路车已离开
            if rv_id in main_road_vehicles_near_merge: del main_road_vehicles_near_merge[rv_id] # 从列表删除
            continue # 跳过
        
        rv_data = rv_info["data"] # 主路车数据
        # 检查是否在MV的通信范围内 (已经在BSM交换中隐式处理了，但可以再确认)
        dist_mv_rv_sq = (mv_data["position"][0] - rv_data["position"][0])**2 + \
                        (mv_data["position"][1] - rv_data["position"][1])**2 # 计算MV和RV距离的平方
        
        if dist_mv_rv_sq**0.5 <= COMMUNICATION_RANGE and rv_info["lane_id"] == mv_target_main_lane: # 如果在通信范围且在MV的目标车道上
            # 判断RV是在MV的“前方”还是“后方”（相对于并道点）
            # 简化：假设并道点在主路的 pos=0 处。MV的目标也是这个位置附近。
            # 如果RV的 position_on_lane 比MV“等效”在主路上的位置大，则为前车；小则为后车。
            # 这个等效位置很难精确。更简单：如果RV在MV的感知中 (通过BSM)，
            # 并且它在目标车道，我们需要判断它相对于MV并道后的位置。
    
            # 再次简化：我们只考虑当前就在MV目标车道上的RV，
            # 并看它们是否愿意为MV调整。
            
            # 假设MV想并入的位置是主路车道 mv_target_main_lane 的起点 (pos=0) 附近
            # rv_pos_on_main_lane = rv_info["position_on_lane"]
    
            # 这里需要一个更复杂的逻辑来判断RV相对于MV并道点的相对位置和时序
            # 为了演示协同行为，我们假设已经识别出“关键”的RV
            # 例如，一个在MV将要并入位置后方的RV (follower_candidate)
            # 和一个在前方的RV (leader_candidate)
            
            # --- 简化逻辑：如果RV在目标车道，且在MV“附近” ---
            # 我们让目标车道上的后方车辆尝试减速，前方车辆尝试加速
            
            # 模拟RV收到了MV的并道请求 (通过BSM中的特殊标志或消息类型)
            # if mv_info["requesting_merge"]: (已经检查过了)
    
            # 判断RV是MV并道后的潜在后车还是前车 (需要更精确的几何和速度投影)
            # 粗略判断：如果RV的当前位置比MV的当前位置更“下游”（在行驶方向上），则可能是前车
            # 这个判断非常依赖路网的朝向
            
            # --- 让我们关注目标车道上的关键后车 ---
            # 假设 rv_id 是 MV 并道后的潜在后车
            # 并且它在 mv_target_main_lane 上
            
            # 如果RV在MV并道点之后，且距离较近，它应该减速
            # (这个条件需要具体化)
            
            # 为了示例，我们让所有在目标主路车道上的，且在MV通信范围内的RV都尝试协同
            # 如果RV的速度比MV慢，且在MV“后方”某个区域，它可以考虑减速
            # 如果RV的速度比MV快，且在MV“前方”某个区域，它可以考虑加速
    
            # --- 更直接的协同：关键后车减速 ---
            # 查找MV并道后的直接后车 (这个在MV并道前很难精确确定)
            # 替代方案：MV请求后，在目标车道上的，且在并道点“后方”一定范围内的RV，如果它速度允许，则减速
            
            # 简化：如果RV在MV目标车道，并且其当前位置在并道点之前 (即MV会并到它前面)
            # 并且它离并道点不远，它应该考虑为MV减速
            # (这个逻辑需要知道并道点的精确位置在主路车道上的坐标)
    
            # --- 再次简化以演示行为调整 ---
            # 假设我们有一个RV (rv_id)，它决定为MV (mv_id) 减速
            if rv_data["speed"] > 3.0: # 如果RV不是太慢
                # 尝试减速，例如减到当前速度的80%，或一个固定值
                # new_rv_speed = rv_data["speed"] * 0.8 
                # 或者，如果知道与MV的期望间隙，计算能达到该间隙的速度
                
                # 这里的协同行为是：如果我是MV后方目标车道上的车，我稍微减速
                # (这需要判断“后方”。如果rv_info["position_on_lane"] < 预计的MV并入点位置)
                # 为了演示，我们让所有在目标车道的车都稍微减速，除非它们是远前方
                
                # 实际中，只有最关键的后车需要减速。
                # 我们假设 rv_id 就是那个关键后车
                if traci.vehicle.getSpeed(rv_id) > abs(COOP_DECEL_RV) * 1.0 : # 确保有足够速度来减速
                    # 应该用 setSpeed 来设置一个较低的目标速度，让模型自己减速
                    # 或者如果想更直接控制减速度，需要复杂方法
                    # 这里我们简单地降低其 setSpeed 的值
                    target_rv_speed_for_coop = max(1.0, traci.vehicle.getSpeed(rv_id) + COOP_DECEL_RV * 1.0) # 减速1秒的效果
                    traci.vehicle.setSpeed(rv_id, target_rv_speed_for_coop) # 设置RV的新目标速度
                    traci.vehicle.setColor(rv_id, (100,100,255,255)) # 设为浅蓝色表示正在协同
                    print(f"    协同并道: 主路车 {rv_id} 在车道 {mv_target_main_lane} 为 {mv_id} 减速至 {target_rv_speed_for_coop:.1f}m/s。") # 打印协同信息
                    # (应该只让最关键的后车做这个)
            # break # 假设我们只让一辆RV协同 (过于简化)
    
    # 3. MV尝试并道
    for mv_id, mv_info in list(merging_vehicles.items()): # 遍历所有匝道车
        if not mv_info["requesting_merge"] or mv_info["is_merging_now"]: # 如果不请求或已在并道，则跳过
            continue
        if mv_id not in traci.vehicle.getIDList(): # 如果匝道车已离开
            if mv_id in merging_vehicles: del merging_vehicles[mv_id] # 从列表删除
            continue # 跳过
    
        mv_data = mv_info["data"] # 获取匝道车数据
        mv_target_main_lane = mv_info["target_lane_on_main"] # 获取目标主路车道
    
        # MV 检查目标车道是否有安全间隙
        # 这需要模拟MV的感知：它需要知道目标车道上，其并道点附近的前车和后车
        # 可以使用 traci.vehicle.getLaneChangeState(mv_id, targetLaneIndexOnMainRoad)
        # 或者自己计算间隙
        
        # 简化：我们假设MV能“看到”目标车道上的情况 (通过之前的BSM交换或直接查询)
        # 并寻找一个大于 MIN_SAFE_GAP_FOR_MERGE 的间隙
        
        # 获取目标主路车道 (mv_target_main_lane) 上的车辆
        # 并找到在MV预估并道位置附近的前车和后车
        # (这部分逻辑非常复杂，涉及到预测MV和RV的轨迹以及它们在并道点的时空关系)
        
        # --- 简化间隙检查 ---
        # 假设MV到达并道点时，如果目标车道上，其前方一定距离内无车，
        # 且后方一定距离内无车，或者后方车辆速度远低于自己，则认为安全。
        # 这是一个非常粗略的检查。
        
        # SUMO 的内置换道模型 (LC2013, SL2015) 在车辆执行 changeLane 时会进行安全检查。
        # 我们可以尝试让MV执行换道，如果SUMO允许，就说明（根据其模型）是安全的。
        
        # 首先，MV需要能够“到达”主路车道。这意味着它的路径应该包含一个从匝道到主路的连接。
        # 如果MV的当前路径不直接导向主路目标车道，可能需要先 changeTarget 或 setRoute。
        # 假设MV的预设路径就是从匝道到主路。
        
        # 尝试让MV换道到目标主路车道
        # changeLane(vehID, laneIndex, duration) 
        # laneIndex 是相对于车辆当前所在边的车道索引。
        # 如果MV在匝道上，它不能直接 changeLane到主路的车道上。
        # 它需要通过一个 "连接(connection)" 从匝道车道到主路车道。
        # 当车辆自然行驶到匝道末端，如果连接允许，并且主路有空间，SUMO会自动处理并道。
        
        # 我们的“协同”目标是让主路车辆为这个自动并道过程创造更有利的条件。
        # MV本身的行为主要是：当它认为间隙可接受时，它会尝试维持速度或加速以完成并道。
        
        # 那么，MV如何判断间隙是否可接受呢？
        # 它需要感知目标主路车道上，在它将要进入的位置的前方车辆（leader_on_main）和后方车辆（follower_on_main）。
        
        # --- 再次简化：如果MV快到并道点了，并且它“感觉”主路车让了，它就加速并道 ---
        if mv_info["dist_to_merge_point"] < 15.0 : # 如果MV离并道点很近 (例如小于15米)
            # 在这里，MV应该做一个最终的并道决策
            # 它应该检查其目标车道 mv_target_main_lane 上的情况
            # 例如，使用 traci.vehicle.getNeighbors() 或分析已有的 BSM 数据
            # 来找到并道点附近的潜在前车和后车，并评估间隙。
            
            # 为了演示，我们假设如果MV离并道点很近，并且之前有RV为其减速了，
            # 那么MV就尝试加速并道 (SUMO的默认行为可能就会这样做)
            # 我们这里可以给它一个更明确的指令，比如略微提高速度或确保它不减速
            
            # 获取MV在主路上的“虚拟”前车和后车 (这需要高级感知逻辑)
            # 如果 gap_to_main_leader > MIN_SAFE_GAP_FOR_MERGE / 2 
            # 且 gap_to_main_follower > MIN_SAFE_GAP_FOR_MERGE / 2 (简化)
            
            # 一个非常简单的触发：如果快到并道点了，就尝试并道
            # 实际的并道是由SUMO的微观模型根据车辆在连接处的交互来处理的。
            # Python/TraCI的作用是影响这个交互环境。
            # 例如，让MV更“自信”：
            # traci.vehicle.setImperfection(mv_id, 0.1) # 让MV驾驶更果断 (sigma越小越果断)
            # traci.vehicle.setSpeedFactor(mv_id, 1.1) # 尝试比期望速度快10%
            
            # 标记MV正在尝试并道
            merging_vehicles[mv_id]["is_merging_now"] = True # 标记MV正在并道
            traci.vehicle.setColor(mv_id, (50,205,50,255)) # 设为亮绿色表示正在并道
            print(f"    协同并道: 匝道车 {mv_id} 正在尝试并入主路 (SUMO模型接管)。") # 打印尝试并道信息
            # 此时，我们主要依赖SUMO的并道/换道模型。
            # 我们的协同行为（RV减速）是为了给SUMO模型创造一个更容易并道的场景。
    
    # 清理已完成并道或离开的MV
    for mv_id in list(merging_vehicles.keys()): # 遍历匝道车列表
    if mv_id not in traci.vehicle.getIDList(): # 如果车辆已不在仿真中
        del merging_vehicles[mv_id] # 从列表中删除
    else:
        # 如果MV已不在匝道上，说明它可能已并入主路或去了别处
        if traci.vehicle.getRoadID(mv_id) != RAMP_EDGE_ID and merging_vehicles[mv_id]["is_merging_now"]: # 如果不在匝道且标记为正在并道
            print(f"    协同并道: 车辆 {mv_id} 已完成并道或离开匝道。") # 打印完成并道信息
            traci.vehicle.setColor(mv_id, vehicle_knowledge_base.get(mv_id, {}).get("original_color", (255,255,0,255))) # 恢复颜色 (假设有原始颜色)
            del merging_vehicles[mv_id] # 从列表中删除
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/GUb3veJyMtsElLpzqDnauTCX4ZS5.png)

    # (在 run_simulation 函数的主循环内)
    
    # --- 紧急车辆优先 (EVP) 参数 ---
    TLS_ID_FOR_EVP = "N2" # 我们关注的信号灯ID
    EVP_DETECTION_RANGE = 250 # 米, EV多远时信号灯开始响应
    EMERGENCY_VEHICLE_TYPE = "emergency_vtype" # 在.rou.xml中定义的紧急车辆类型ID
    g_evp_active_on_tls = {} # {tls_id: ev_id} 存储哪个EV正在被哪个TLS优先处理
    g_original_tls_program = {} # {tls_id: original_program_id} 存储优先前的原始程序
    
    # (需要在 .rou.xml 中定义一个车辆类型 id="emergency_vtype"，并让一些车辆使用它)
    # <vType id="emergency_vtype" vClass="emergency" guiShape="emergency" color="1,0,1,255" speedFactor="1.5" accel="4.0" decel="6.0">
    #    <param key="has.bluelight.device" value="true"/> <!-- 使得在GUI中有特殊显示 -->
    # </vType>
    # <vehicle id="ev0" type="emergency_vtype" route="route_east" depart="60" />
    
    
    # 1. 检测接近的紧急车辆
    active_evs_near_tls = {} # {ev_id: {"eta_to_tls": float, "approach_lane": str, "path_through_tls": list_of_edges}}
    
    for veh_id in traci.vehicle.getIDList(): # 遍历所有车辆
    try:
        if traci.vehicle.getTypeID(veh_id) == EMERGENCY_VEHICLE_TYPE: # 如果是紧急车辆类型
            tls_pos_evp = traci.junction.getPosition(TLS_ID_FOR_EVP) # 获取EVP信号灯位置
            ev_pos = traci.vehicle.getPosition(veh_id) # 获取EV位置
            dist_ev_to_tls = ((ev_pos[0]-tls_pos_evp[0])**2 + (ev_pos[1]-tls_pos_evp[1])**2)**0.5 # 计算EV到信号灯距离
    
            if dist_ev_to_tls <= EVP_DETECTION_RANGE: # 如果EV在检测范围内
                # 检查EV的路径是否会经过这个TLS
                ev_route = traci.vehicle.getRoute(veh_id) # 获取EV的路径
                ev_road = traci.vehicle.getRoadID(veh_id) # EV当前所在边
                
                # 确定EV从哪个进口道接近TLS_ID_FOR_EVP，以及它要往哪个出口道去
                # 这需要分析 ev_route 和 TLS_ID_FOR_EVP 控制的连接
                # 简化：假设我们能找到EV接近TLS的进口车道和离开的出口边
                approach_edge = None # EV接近的进口边
                departure_edge_after_tls = None # EV通过TLS后的出口边
                
                if ev_road in traci.trafficlight.getControlledLinks(TLS_ID_FOR_EVP, detailed=True).getIncomingEdges(): # 伪代码：检查EV是否在受控进口道
                     approach_edge = ev_road
                # (需要更复杂的逻辑来确定 approach_edge 和 departure_edge_after_tls)
                
                # 粗略估计ETA (不考虑信号灯延误)
                ev_speed = traci.vehicle.getSpeed(veh_id) # EV速度
                eta = dist_ev_to_tls / ev_speed if ev_speed > 0.1 else float('inf') # 估计到达时间
    
                # (为了演示，我们假设只要EV在范围内就记录它)
                active_evs_near_tls[veh_id] = { # 存储接近的EV信息
                    "eta": eta, # 预计到达时间
                    "current_road": ev_road, # 当前路边
                    "route": ev_route # 路径
                }
                if veh_id not in g_evp_active_on_tls.values(): # 如果该EV还未被任何TLS优先处理
                    print(f"  EVP: 紧急车辆 {veh_id} (类型: {EMERGENCY_VEHICLE_TYPE}) 正在接近 {TLS_ID_FOR_EVP} (距离 {dist_ev_to_tls:.1f}m, ETA {eta:.1f}s)") # 打印EV接近信息
    except traci.exceptions.TraCIException: # 捕获TraCI异常
        pass # 忽略错误
    
    # 2. 信号灯为检测到的EV执行优先逻辑
    if active_evs_near_tls: # 如果有EV正在接近
    # 选择一个EV进行优先 (例如，ETA最小的，或者已经有优先权的)
    # 简化：我们只处理第一个检测到的EV，并且一个TLS一次只为一个EV优先
    
    if TLS_ID_FOR_EVP not in g_evp_active_on_tls: # 如果该TLS当前没有正在处理的EVP
        # 选择一个EV (例如，ETA最小的)
        chosen_ev_id = min(active_evs_near_tls, key=lambda ev: active_evs_near_tls[ev]["eta"]) # 选择ETA最小的EV
        
        print(f"  EVP: 信号灯 {TLS_ID_FOR_EVP} 开始为紧急车辆 {chosen_ev_id} 执行优先操作。") # 打印开始优先操作信息
        
        # a. 保存当前信号灯程序 (如果它不是由我们完全控制的)
        if TLS_ID_FOR_EVP not in g_original_tls_program: # 如果未保存过原始程序
            g_original_tls_program[TLS_ID_FOR_EVP] = traci.trafficlight.getProgram(TLS_ID_FOR_EVP) # 保存当前程序ID
        
        # b. 找到能让 chosen_ev_id 通行的相位 (最关键的步骤)
        #    这需要知道 chosen_ev_id 的进口道和出口道，然后找到
        #    `getCompleteRedYellowGreenDefinition` 返回的相位中，哪个相位
        #    为这个 specific (进口道->出口道) 路径提供绿灯。
        target_green_phase_for_ev = -1 # 初始化目标绿灯相位索引为-1
        
        # --- !! 关键的相位识别逻辑 (高度依赖路网和信号灯定义) !! ---
        # 假设 chosen_ev_id 要从 E1 到 E2 (东西向)
        # 我们在 initialize_adaptive_tls_control 中识别过，假设相位0是EW_Green
        # if chosen_ev_id 的路径确实是 E1 -> N2 -> E2:
        #    target_green_phase_for_ev = 0 (EW_Green的索引)
        # 假设 chosen_ev_id 要从 E3 到 E4 (南北向)
        # if chosen_ev_id 的路径确实是 E3 -> N2 -> E4:
        #    target_green_phase_for_ev = 2 (NS_Green_E3的索引)
        
        # 这里需要一个函数: find_phase_for_vehicle_path(tls_id, ev_id, ev_route) -> phase_index
        # 这个函数会分析EV的路径和TLS的相位定义，返回最佳绿灯相位。
        # 为了演示，我们随机选择一个我们之前定义的“自适应绿灯相位”
        if TLS_ID_FOR_EVP in g_adaptive_tls_definitions: # 如果有自适应相位定义
            adaptive_green_phases_indices = [
                idx for idx, info in g_adaptive_tls_definitions[TLS_ID_FOR_EVP].items() if info.get("is_green_phase")
            ] # 获取所有自适应绿灯相位的索引
            if adaptive_green_phases_indices: # 如果存在这样的相位
                # 粗略匹配：看EV当前在哪条边，如果这条边是某个绿灯相位服务的，就选那个
                ev_current_road = active_evs_near_tls[chosen_ev_id]["current_road"] # 获取EV当前所在边
                for p_idx in adaptive_green_phases_indices: # 遍历自适应绿灯相位
                    if ev_current_road in g_adaptive_tls_definitions[TLS_ID_FOR_EVP][p_idx]["monitored_lanes"][0][:len(ev_current_road)]: # 简单匹配边ID前缀
                        target_green_phase_for_ev = p_idx # 设置目标绿灯相位
                        break # 退出循环
                if target_green_phase_for_ev == -1: # 如果没匹配到
                    target_green_phase_for_ev = adaptive_green_phases_indices[0] # 默认选第一个
        
        if target_green_phase_for_ev != -1: # 如果找到了目标绿灯相位
            print(f"    EVP: 为 {chosen_ev_id} 选定的优先相位是 {target_green_phase_for_ev} "
                  f"('{g_adaptive_tls_definitions[TLS_ID_FOR_EVP].get(target_green_phase_for_ev,{}).get('name','未知相位')}').") # 打印选定的优先相位
            
            # c. 切换到该优先相位 (同样，需要考虑黄灯/全红过渡)
            #    一个简单的方法是直接设置，然后让它持续一段时间直到EV通过。
            #    更安全的是通过一个预设的“清空”序列（例如，所有冲突方向黄灯->红灯）
            #    然后再给EV绿灯。
            
            # 简化：我们直接设置到目标绿灯相位，并标记EVP正在激活
            # (这忽略了安全清空阶段，实际中非常重要!)
            traci.trafficlight.setPhase(TLS_ID_FOR_EVP, target_green_phase_for_ev) # 设置到优先相位
            g_evp_active_on_tls[TLS_ID_FOR_EVP] = chosen_ev_id # 标记该TLS正在为该EV进行优先处理
            # 如果我们的自适应控制器也在，需要一种方式让它知道EVP正在激活，从而暂停其常规逻辑
            # 例如，可以设置一个全局标志 g_evp_mode_active = True
            if TLS_ID_FOR_EVP in g_active_phase_index: # 如果自适应控制器在跟踪这个TLS
                g_active_phase_index[TLS_ID_FOR_EVP] = target_green_phase_for_ev # 更新自适应控制器的活动相位
                g_current_phase_start_time[TLS_ID_FOR_EVP] = current_time # 更新相位开始时间
        else:
            print(f"    EVP: 未能为 {chosen_ev_id} 在 {TLS_ID_FOR_EVP} 上找到合适的优先相位。") # 打印未找到相位信息
    
    elif TLS_ID_FOR_EVP in g_evp_active_on_tls: # 如果该TLS当前正在执行EVP
    ev_being_served = g_evp_active_on_tls[TLS_ID_FOR_EVP] # 获取正在被服务的EV的ID
    # 检查EV是否已通过交叉口
    ev_has_passed = False # 初始化EV已通过标志为False
    if ev_being_served not in traci.vehicle.getIDList(): # 如果EV已不在仿真中
        ev_has_passed = True # 标记为已通过
    else:
        # 检查EV是否已经远离交叉口 (例如，在出口道上行驶了一段距离)
        ev_current_road_now = traci.vehicle.getRoadID(ev_being_served) # 获取EV当前所在边
        # 假设我们知道EVP的出口边 (例如，E2或E4)
        # if ev_current_road_now in ["E2", "E4"] and traci.vehicle.getLanePosition(ev_being_served) > 50: # 如果在出口道且已行驶超过50米
        #    ev_has_passed = True
        # 简化：如果EV不在其原始的几个进口道上，就认为它可能通过了
        # (这需要更精确的判断)
        if ev_current_road_now not in active_evs_near_tls.get(ev_being_served,{}).get("route",[]): # 一个非常粗略的判断
             # 或者如果它已经离交叉口中心很远了
            tls_pos_evp_check = traci.junction.getPosition(TLS_ID_FOR_EVP) # 获取TLS位置
            ev_pos_check = traci.vehicle.getPosition(ev_being_served) # 获取EV位置
            dist_now = ((ev_pos_check[0]-tls_pos_evp_check[0])**2 + (ev_pos_check[1]-tls_pos_evp_check[1])**2)**0.5 # 计算当前距离
            if dist_now > EVP_DETECTION_RANGE + 50 : # 如果EV已经驶离探测范围一段距离
                 ev_has_passed = True # 标记为已通过
                 
    if ev_has_passed: # 如果EV已通过
        print(f"  EVP: 紧急车辆 {ev_being_served} 已通过交叉口 {TLS_ID_FOR_EVP}。恢复正常信号控制。") # 打印恢复正常控制信息
        del g_evp_active_on_tls[TLS_ID_FOR_EVP] # 清除EVP激活状态
        
        # 恢复到原始的信号灯程序 (如果保存过)
        original_program = g_original_tls_program.get(TLS_ID_FOR_EVP) # 获取原始程序ID
        if original_program: # 如果有保存的原始程序
            try:
                traci.trafficlight.setProgram(TLS_ID_FOR_EVP, original_program) # 恢复到原始程序
                print(f"    TLS {TLS_ID_FOR_EVP} 已恢复到原始程序 '{original_program}'。") # 打印恢复信息
                # 可能还需要重置我们自适应控制器的状态
                if TLS_ID_FOR_EVP in g_active_phase_index: # 如果自适应控制器在跟踪
                    # 重新初始化或设置到一个默认相位
                    # (这里简单地让自适应逻辑在下一轮重新评估)
                    g_active_phase_index[TLS_ID_FOR_EVP] = traci.trafficlight.getPhase(TLS_ID_FOR_EVP) # 用当前实际相位更新
                    g_current_phase_start_time[TLS_ID_FOR_EVP] = current_time # 更新开始时间
            except traci.exceptions.TraCIException as e_restore: # 捕获恢复程序异常
                print(f"    恢复TLS {TLS_ID_FOR_EVP} 到原始程序时出错: {e_restore}") # 打印错误
        # else: # 如果没有原始程序，可能意味着它一直由TraCI控制，让自适应逻辑接管
            # initialize_adaptive_tls_control() # 或者重新初始化自适应逻辑的状态
    
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/FH3iN0mdE2pzAuSr9ch1BkGb7RPg.png)