Traffic 示例¶

注意

示例目录：[SimOne 安装目录]\SimOneAPI\Tutorial\Traffic

算法构思

Traffic 示例演示如何通过 Socket 连接到 SimOne 的交通流提供接口，以 JSON 格式将自定义交通流数据（车辆、自行车、行人的位置）实时推送给仿真引擎，实现外部交通流接管控制。

支持 C++ 和 Python 两套实现，逻辑等价。

C++ 示例¶

核心流程：

建立 Socket 连接至 SimOne 交通流接口（127.0.0.1:21568）
从 JSON 轨迹文件加载路径点（Trail）
按帧计算各交通对象的当前位置
将 Task → Simulation 数据包序列化为 JSON 后发送

关键类说明：

类	职责
`Trail`	加载 `.json` 轨迹文件，根据行驶距离插值计算位置
`TrafficObject`	单个交通参与者，绑定轨迹、速度
`Task`	单帧数据包，包含时间戳和所有对象的当前状态
`Simulation`	管理 Socket 连接，逐帧发送 Task

轨迹加载与发送示例：

// 建立 Socket 连接（端口 21568 为 SimOne 交通流接口）
int sock = Utils::Connect("127.0.0.1", 21568);
Simulation simulation(sock);

// 加载轨迹文件（JSON 格式，含 waypoints 数组）
Trail trail;
trail.Create("vehicleWayPoints.json");

// 创建交通对象：id="10000", 资产="1000000", 类型=Vehicle, 速度=20m/s
TrafficObject* car = new TrafficObject("10000", "1000000", "Vehicle", 20, "trail_car", 0);
traffic.AppendTrail("trail_car", &trail);
traffic.AppendTrafficObject(car);

// 主循环：每 0.1s 发送一帧
while (true) {
    Task task;
    task.Create(simulation.GetTimestamp());
    task.Append(car);
    simulation.Append(&task);
    traffic.UpdateObjects();   // 根据速度和时间步长更新各对象位置
    simulation.Run();          // 序列化为 JSON 并发送
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
}

Python 示例¶

提供两个脚本，适用于不同控制需求：

TrafficFlow.py — 轨迹跟随模式¶

按预设轨迹（waypoints JSON）驱动交通流，适合批量创建沿路径行驶的车辆、自行车、行人。

import socket, json, time, math, os

# 连接 SimOne 交通流接口
trafficProviderSocket = socket.socket()
trafficProviderSocket.connect(("127.0.0.1", 21568))

simulation = Simulation(trafficProviderSocket)
traffic = Traffic()

# 加载三条轨迹
trail_car = Trail()
trail_car.Create("vehicleWayPoints.json")
traffic.AppendTrail("trail_car", trail_car)

# 按间距在轨迹上均布车辆（每 20m 一辆，速度 20m/s）
carCount = int(trail_car.GetTrailLength() / 20)
for index in range(carCount):
    car = TrafficObject(str(10000 + index), "1000000", "Vehicle",
                        speed=20, trailName="trail_car", distance=index * 20)
    traffic.AppendTrafficObject(car)

# 主循环
while True:
    task = Task()
    task.Create(simulation.timestamp)
    for obj in traffic.trafficObjects:
        task.Append(obj)
    simulation.Append(task)
    traffic.UpdateObjects()   # 位置插值更新
    simulation.Run(traffic)   # 发送 JSON 数据包
    time.sleep(0.1)

Generalization.py — 物理运动学模式¶

通过速度 + 加速度（运动学方程）驱动单个对象，适合精确控制特定对手车行为（如切入、跟车等场景）。

from SimOneServiceAPI import *
from SimOnePNCAPI import *

TIME_STEP = 0.03
EGO = '0'

# 初始化 SimOne API（用于读取主车状态）
if SoInitSimOneAPI(EGO, 0, "127.0.0.1") == 1:
    print("SimOneAPI init successed!!!")

case = Case()
episode = Episode(case)

# 创建对手车：起始位置 (-200.978, -3.0)，初速 11.1 m/s，加速度 -2.0 m/s²
veh = TrafficObject("10000", "1000000", "Vehicle",
                    x=-200.978073, y=-3.000088,
                    speedX=11.1111, accelX=-2.0,
                    speedY=0.0, accelY=0.0)
episode.AddTrafficObject(veh)
episode.Start()

# 运行 5 秒后停止并重置主车
time.sleep(5)
episode.Stop()
case.Stop()

编译/运行环境

C++ 工程：同其他 Tutorial 模块，使用 CMakeLists.txt 生成 VS 工程或 Linux Makefile
Python 脚本：直接运行 python TrafficFlow.py 或 python Generalization.py，需确保 SimOne 仿真已启动

使用流程

启动 SimOne，新建测试用例
在用例中配置交通流接口（确认端口 21568 已开启）——详见外接第三方交通流 (Vissim-API)
运行用例
运行示例程序/脚本，观察仿真中交通对象按预设轨迹运动
API 功能支持：API 功能支持