• 从 Python 接口 TraCI
从 Python 接口 TraCI

TraCI 命令被划分为 busstop、calibrator、chargingstation、edge、gui、inductionloop、junction、lane、lanearea、meandata、multientryexit、overheadwire、parkingarea、person、poi、polygon、rerouter、route、routeprobe、simulation、trafficlight、variablespeedsign、vehicle 和 vehicletype 等域，它们对应于独立的模块。有关可用函数的详细列表，请参阅 pydoc 生成的文档。源代码可以在 [1] 找到。

请注意，如果性能是个问题且不需要 GUI，使用 libsumo 代替 traci 几乎总是更好的选择，它们具有相同的 API。

在脚本中导入 traci#

要使用该库，可以使用 pip install traci 进行安装，或者将 <SUMO_HOME>/tools 目录添加到 Python 的加载路径中。通常使用如下代码段完成：

import os
import sys
if 'SUMO_HOME' in os.environ:
    sys.path.append(os.path.join(os.environ['SUMO_HOME'], 'tools'))
import traci

这假设在运行脚本之前已经设置了环境变量 SUMO_HOME。或者，你可以直接声明 sumo/tools 的路径，如下所示：

sys.path.append(os.path.join('c:', os.sep, 'whatever', 'path', 'to', 'sumo', 'tools'))

如果你决定在后期切换到 libsumo 或者希望保持灵活性，可以稍后将导入行替换为：

import libsumo as traci

入门步骤#

通常，从 Python 接口与 SUMO 交互非常简单（以下示例修改自 tutorial/traci_tls）：

首先，编写启动 sumo 或 sumo-gui 的命令行（省略了 0.28.0 之前所需的选项）：

sumoBinary = "/path/to/sumo-gui"
sumoCmd = [sumoBinary, "-c", "yourConfiguration.sumocfg"]

然后启动模拟并使用脚本连接到它：

import traci
traci.start(sumoCmd)
step = 0
while step < 1000:
    traci.simulationStep()
    if traci.inductionloop.getLastStepVehicleNumber("0") > 0:
        traci.trafficlight.setRedYellowGreenState("0", "GrGr")
    step += 1

traci.close()

连接到模拟后，你可以发出各种命令并执行模拟步骤，直到你想通过关闭连接来结束。默认情况下，close 命令会等待直到 sumo 进程真正结束。你可以通过调用以下命令禁用此行为：

traci.close(False)

订阅 (Subscriptions)#

订阅可以被视为一种批量检索变量的模式。与其一遍又一遍地请求相同的变量，不如在每个时间步之后自动检索感兴趣的值。TraCI 订阅是按模块处理的。也就是说，你可以在每个时间步之后向模块请求所有当前订阅的结果。为了订阅变量，你需要知道它们的变量 ID，这些 ID 可以在 traci/constants.py 文件中查找。

import traci
import traci.constants as tc

traci.start(["sumo", "-c", "my.sumocfg"])
traci.vehicle.subscribe(vehID, (tc.VAR_ROAD_ID, tc.VAR_LANEPOSITION))
print(traci.vehicle.getSubscriptionResults(vehID))
for step in range(3):
    print("step", step)
    traci.simulationStep()
    print(traci.vehicle.getSubscriptionResults(vehID))
traci.close()

检索到的值始终是上一个时间步的值，无法检索更早的值。

Note

在 1.18.0 版本之前，traci.simulationStep() 会返回所有订阅结果，现在它返回 None。如果你需要旧行为，请使用 traci.simulationStepLegacy()。

上下文订阅 (Context Subscriptions)#

上下文订阅的工作方式类似于订阅，因为它们会自动为每个模拟停止检索变量列表。但是，它们通过设置一个参考对象和一个范围，然后检索该范围内给定类型的所有对象的变量来实现这一点。

TraCI 上下文订阅是按模块处理的。也就是说，你可以在每个时间步之后向模块请求所有当前订阅的结果。为了订阅变量，你需要获取要检索对象的域 ID 以及变量 ID，这些可以在 traci/constants.py 文件中查找。域 ID 始终具有 CMD_GET_<DOMAIN>_VARIABLE 的形式。以下代码检索一个路口（junction）范围内（42m）的所有车辆速度和等待时间（车辆 ID 是隐式检索的）。

import traci
import traci.constants as tc

traci.start(["sumo", "-c", "my.sumocfg"])
traci.junction.subscribeContext(junctionID, tc.CMD_GET_VEHICLE_VARIABLE, 42, [tc.VAR_SPEED, tc.VAR_WAITING_TIME])
print(traci.junction.getContextSubscriptionResults(junctionID))
for step in range(3):
    print("step", step)
    traci.simulationStep()
    print(traci.junction.getContextSubscriptionResults(junctionID))
traci.close()

检索到的值始终是上一个时间步的值，无法检索更早的值。

Note

在 1.18.0 版本之前，traci.junction.getAllContextSubscriptionResults 不会包含其上下文中没有对象的路口的 ID。现在它会将其 ID 映射到一个空字典（类似于 libsumo）。这同样适用于其他域。

上下文订阅过滤器 (Context Subscription Filters)#

对于车辆到车辆的上下文订阅（即，参考对象是车辆且请求的上下文对象也是车辆的上下文订阅），可以请求在服务器端应用额外的过滤器。一般过程是在调用 subscribeContext() 之后，通过连续调用 addSubscriptionFilter<FILTER_ID>() 来为请求的上下文订阅配备过滤器，例如以下代码段：

traci.vehicle.subscribeContext("ego", tc.CMD_GET_VEHICLE_VARIABLE, 0.0, [tc.VAR_SPEED])
traci.vehicle.addSubscriptionFilterLanes(lanes, noOpposite=True, downstreamDist=100, upstreamDist=50)

第一行请求对参考车辆（ID 为 "ego"）附近的车辆速度进行上下文订阅。上下文订阅的范围（指通常订阅机制的径向上下文区域）可以设置为 0.0，因为它将被第二行中调用 addSubscriptionFilterLanes() 时给定的 downstreamDist 和 upstreamDist 的选择性值所覆盖。调用 addSubscriptionFilter<FILTER_NAME>() 会自动对最后发出的上下文订阅生效，该订阅必须是车辆到车辆的形式才能成功应用。

以下是可用的过滤器类型：

  • Lanes：返回参考车辆指定车道上的周围车辆
  • CFManeuver：返回参考车辆所在车道的前导车和跟随车
  • LCManeuver：返回参考车辆所在车道及相邻车道上的前导车和跟随车
  • Turn：返回车辆沿路线行驶时在即将到来的路口上的冲突车辆
  • VType：仅返回指定 vTypes 的车辆
  • VClass：仅返回指定 vClasses 的车辆

有关详细规范，请参阅 pydoc 文档。

Caution

该过滤器仅在后续的模拟步骤中生效。发出订阅后直接返回的车辆值不受影响。

添加步骤监听器 (StepListener)#

通常需要在每次调用 traci.simulationStep() 时调用一个函数，为了自动实现这一点（总是在每次调用 simulationStep() 之后），可以添加一个 StepListener 对象 'listener'（更准确地说，是 traci.StepListener 子类的实例），即：

class ExampleListener(traci.StepListener):
    def step(self, t):
        # 在每次调用 simulationStep 后执行某些操作
        print("ExampleListener called with parameter %s." % t)
        # 指示步骤监听器应在下一步保持活动状态
        return True

listener = ExampleListener()
traci.addStepListener(listener)

请注意，监听器不是为每个模拟步骤激活，而是为每次调用 simulationStep 激活（这可能会执行多个步骤，直到给定的时间 t）。此外，参数 t 不是当前模拟时间，而是传递给 simulationStep 调用的（可选）参数（默认为 0）。

Caution

在一个 TraCI 客户端控制多个 SUMO 实例的情况下，不能使用 TraCI 步骤监听器。

从同一个 TraCI 脚本控制并行模拟#

TraCI python 库可用于通过单个脚本同时控制多个模拟。函数 traci.start() 有一个可选的标签参数，允许你使用不同的模拟实例和标签多次调用它。然后可以使用函数 traci.switch() 切换到任何已初始化的标签：

traci.start(["sumo", "-c", "sim1.sumocfg"], label="sim1")
traci.start(["sumo", "-c", "sim2.sumocfg"], label="sim2")
traci.switch("sim1")
traci.simulationStep() # 为 sim1 运行 1 步
traci.switch("sim2")
traci.simulationStep() # 为 sim2 运行 1 步
traci.switch("sim1")
traci.close()
traci.switch("sim2")
traci.close()

如果你更喜欢面向对象的方法，也可以使用连接对象与模拟通信。它们具有与静态 traci. 调用相同的接口，但你仍然需要手动启动模拟：

traci.start(["sumo", "-c", "sim1.sumocfg"], label="sim1")
traci.start(["sumo", "-c", "sim2.sumocfg"], label="sim2")
conn1 = traci.getConnection("sim1")
conn2 = traci.getConnection("sim2")
conn1.simulationStep() # 为 sim1 运行 1 步
conn2.simulationStep() # 为 sim2 运行 1 步

从多个客户端控制同一个模拟#

要连接多个客户端，必须提前知道客户端的数量，并使用 sumo 选项 --num-clients <INT> 指定。此外，连接端口必须对所有客户端都已知。确定端口后，可以通过参数或配置文件将其提供给客户端。可以通过以下方式获取空闲端口：

from sumolib.miscutils import getFreeSocketPort
port = sumolib.miscutils.getFreeSocketPort()

一个客户端可以使用方法 traci.start() 启动模拟并同时连接到它，而其他客户端只需要连接。建立客户端顺序后，每个客户端必须持续调用 simulationStep 以允许模拟推进：

# 客户端 1
# PORT = int(sys.argv[1]) # 示例
traci.start(["sumo", "-c", "sim.sumocfg", "--num-clients", "2"], port=PORT)
traci.setOrder(1) # 数字可以是任意值
while traci.simulation.getMinExpectedNumber() > 0:
    traci.simulationStep()
    # 更多 traci 命令
traci.close()

# 客户端 2
# PORT = int(sys.argv[1]) # 示例
traci.init(PORT)
traci.setOrder(2) # 数字可以是任意值，只要每个客户端都有自己的数字即可
while traci.simulation.getMinExpectedNumber() > 0:
    traci.simulationStep()
    # 更多 traci 命令
traci.close()

并发访问同一个 TraCI 连接#

在 SUMO 1.17.0 之前，纯 Python 客户端以及 libtraci 实现如果尝试从不同线程访问同一个连接，可能会崩溃。

目前实现了一些措施来防止访问套接字时的直接冲突。然而，仍然鼓励使用上一节中的多客户端方法。这是确保命令按预期顺序发送的唯一方法。如果只有一个线程发出 simulationStep 命令，而其他线程只查询模拟，那么多线程访问也应该可以工作。

附加功能#

使用 TraCI 时，有一些常见任务未被 traci 库涵盖，例如：

  • 分析路网
  • 解析模拟输出

对于此功能，建议使用 Tools/Sumolib

陷阱与解决方案#

  • 请注意，在 1.18.0 之前，交换字符串时必须是纯 ASCII。目前，所有字符串都应该支持 UTF-8。
  • 如果你使用 subprocess.Popen 从 python 脚本中启动 sumo，请务必在退出脚本之前在生成的进程对象上调用 wait()。否则可能会丢失输出。

确定加载的 traci 库#

当使用不同的 sumo 版本时，调用 import traci 可能会加载错误的库。调试此问题的最简单方法是在导入后添加以下行：

import traci
print("LOADPATH:", '\n'.join(sys.path))
print("TRACIPATH:", traci.__file__)
sys.exit()

确保 TRACIPATH 对应于你希望使用的 sumo 版本。如果不是，则必须更改 LOADPATH (sys.path) 中目录的顺序，或者从任何位于所需目录之前的目录中删除 SUMO 安装。

在 Linux 上调试 TraCI 会话#

有时 SUMO 可能在使用 TraCI 运行模拟时崩溃。以下步骤使得在调试器中运行带有 traci 的 sumo 变得简单：

1）在你的 traci 脚本中添加选项 --save-configuration：

traci.start([sumoBinary, '-c', 'run.sumocfg', '--save-configuration', 'debug.sumocfg'])

2）运行你的 traci 脚本。它不会启动 sumo，而是只写入带有选定端口的配置，但它仍会尝试重复连接。

3）运行

gdb --args sumoD -c debug.sumocfg

（其中 sumoD 是 以调试模式编译的 sumo）

生成所有 traci 命令的日志#

为了共享 traci 场景（例如在错误报告中），将 traci 脚本的逻辑与实际命令分开可能很有用。为此，函数 traci.start 接受可选参数 traceFile 和 traceGetters。 当调用 traci.start([<commands>], traceFile=<LOG_FILE_PATH>) 时，所有发送到 sumo 的 traci 命令都将写入给定的 LOG_FILE_PATH。这允许在没有原始运行脚本的情况下重新运行场景。

当设置选项 traceGetters=False 时，日志文件中仅包含更改模拟状态的函数。检索模拟数据的函数在技术上不需要重现场景，但如果数据检索函数本身是错误的原因，则包含它们可能很有用。

当设置选项 traceGetters="print" 时，所有获取器函数在将日志作为 python 程序运行时，会将其自身及其结果打印到标准输出。

Caution

避免使用选项 --random 运行模拟，因为这很可能会阻止你的 traceFile 被重复执行。

确定 TraCI 客户端无法连接的原因#

可能，传递给 traci.start 的参数在启动 SUMO 时产生了错误。这将表现为：

traci.exceptions.FatalTraCIError: Could not connect.

要诊断问题，请为 traci.start 添加写入日志文件的选项（即 traci.start(['sumo', '-c', 'example.sumocfg', '--log', 'logfile.txt'])） 在脚本启动失败后，查看写入的日志文件并修复其中报告的错误。

使用示例#

运行模拟直到所有车辆到达#

while traci.simulation.getMinExpectedNumber() > 0:
    traci.simulationStep()

动态添加行程（不完整的路线）#

定义一个由起点和终点边组成的路线：

traci.route.add("trip", ["startEdge", "endEdge"])

然后添加具有该路线的车辆

traci.vehicle.add("newVeh", "trip", typeID="reroutingType")

这将导致车辆根据出发时网络中的预计行驶时间计算从 startEdge 到 endEdge 的新路线。有关此机制的详细信息，请参阅 Demand/Automatic_Routing。

坐标转换#

x, y = traci.vehicle.getPosition(vehID)
lon, lat = traci.simulation.convertGeo(x, y)
x2, y2 = traci.simulation.convertGeo(lon, lat, fromGeo=True)

edgeID, lanePosition, laneIndex = traci.simulation.convertRoad(x3, y3)
edgeID, lanePosition, laneIndex = traci.simulation.convertRoad(lon2, lat2, True)

检索当前在网络中的所有车辆的 timeLoss#

import traci
import traci.constants as tc
traci.start(["sumo", "-c", "my.sumocfg"])
# 选择一个任意的路口
junctionID = traci.junction.getIDList()[0]
# 在该路口周围订阅，半径足够大以检索所有车辆在每一步的速度
traci.junction.subscribeContext(
    junctionID, tc.CMD_GET_VEHICLE_VARIABLE, 1000000,
    [tc.VAR_SPEED, tc.VAR_ALLOWED_SPEED]
)
stepLength = traci.simulation.getDeltaT()
while traci.simulation.getMinExpectedNumber() > 0:
    traci.simulationStep()
    scResults = traci.junction.getContextSubscriptionResults(junctionID)
    halting = 0
    if scResults:
        relSpeeds = [d[tc.VAR_SPEED] / d[tc.VAR_ALLOWED_SPEED] for d in scResults.values()]
        # 计算与 summary-output 对应的值
        running = len(relSpeeds)
        halting = len([1 for d in scResults.values() if d[tc.VAR_SPEED] < 0.1])
        meanSpeedRelative = sum(relSpeeds) / running
        timeLoss = (1 - meanSpeedRelative) * running * stepLength
    print(traci.simulation.getTime(), timeLoss, halting)
traci.close()

处理异常#

有时命令会引发（可恢复的）异常以指示错误（未知 ID、未找到路线等）。这些异常可以由你的代码按如下方式处理：

try:
    pos = traci.vehicle.getPosition(vehID)
except traci.TraCIException:
    pass # 或者做更智能的处理

更多资源#

  • 模块 Simpla 提供了一个用于车队功能的库，可以与用户客户端脚本集成。