Traffic Example

Note

Example directory: [SimOne installation directory]\SimOneAPI\Tutorial\Traffic

Algorithm Overview

The Traffic example demonstrates how to connect to SimOne's traffic flow provider interface via Socket and push custom traffic flow data (vehicle, bicycle, and pedestrian positions) to the simulation engine in real time using JSON format, enabling external traffic flow takeover control.

Both C++ and Python implementations are provided; they are logically equivalent.

---

C++ Example

Core flow:

1. Establish a Socket connection to the SimOne traffic flow interface (127.0.0.1:21568)
2. Load waypoints (Trail) from a JSON trajectory file
3. Compute the current position of each traffic object per frame
4. Serialize the Task → Simulation data packet as JSON and send it

Key class descriptions:

Class	Role
Trail	Loads a .json trajectory file and interpolates positions based on travel distance
TrafficObject	A single traffic participant, bound to a trajectory and speed
Task	Single-frame data packet containing a timestamp and the current state of all objects
Simulation	Manages the Socket connection and sends a Task each frame

Trajectory loading and sending example:

// Establish a Socket connection (port 21568 is the SimOne traffic flow interface)
int sock = Utils::Connect("127.0.0.1", 21568);
Simulation simulation(sock);

// Load trajectory file (JSON format with a waypoints array)
Trail trail;
trail.Create("vehicleWayPoints.json");

// Create a traffic object: id="10000", asset="1000000", type=Vehicle, speed=20 m/s
TrafficObject* car = new TrafficObject("10000", "1000000", "Vehicle", 20, "trail_car", 0);
traffic.AppendTrail("trail_car", &trail);
traffic.AppendTrafficObject(car);

// Main loop: send one frame every 0.1 s
while (true) {
    Task task;
    task.Create(simulation.GetTimestamp());
    task.Append(car);
    simulation.Append(&task);
    traffic.UpdateObjects();   // Update each object's position based on speed and time step
    simulation.Run();          // Serialize to JSON and send
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
}

---

Python Example

Two scripts are provided for different control requirements:

TrafficFlow.py — Trajectory-following Mode

Drives traffic flow along preset waypoints (waypoints JSON). Suitable for batch-creating vehicles, bicycles, and pedestrians traveling along a path.

import socket, json, time, math, os

# Connect to the SimOne traffic flow interface
trafficProviderSocket = socket.socket()
trafficProviderSocket.connect(("127.0.0.1", 21568))

simulation = Simulation(trafficProviderSocket)
traffic = Traffic()

# Load three trajectories
trail_car = Trail()
trail_car.Create("vehicleWayPoints.json")
traffic.AppendTrail("trail_car", trail_car)

# Distribute vehicles evenly along the trajectory (one every 20 m, speed 20 m/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)

# Main loop
while True:
    task = Task()
    task.Create(simulation.timestamp)
    for obj in traffic.trafficObjects:
        task.Append(obj)
    simulation.Append(task)
    traffic.UpdateObjects()   # Position interpolation update
    simulation.Run(traffic)   # Send JSON data packet
    time.sleep(0.1)

Generalization.py — Kinematic Mode

Drives a single object using speed + acceleration (kinematic equations). Suitable for precise control of a specific opponent vehicle's behavior (e.g., cut-in, car-following scenarios).

from SimOneServiceAPI import *
from SimOnePNCAPI import *

TIME_STEP = 0.03
EGO = '0'

# Initialize SimOne API (to read ego vehicle status)
if SoInitSimOneAPI(EGO, 0, "127.0.0.1") == 1:
    print("SimOneAPI init successed!!!")

case = Case()
episode = Episode(case)

# Create an opponent vehicle: initial position (-200.978, -3.0), initial speed 11.1 m/s, acceleration -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()

# Run for 5 seconds, then stop and reset the ego vehicle
time.sleep(5)
episode.Stop()
case.Stop()

Build/Run Environment

• C++ project: Same as other Tutorial modules — use CMakeLists.txt to generate a VS project or Linux Makefile.
• Python scripts: Run python TrafficFlow.py or python Generalization.py directly. SimOne must be running before starting the scripts.

Usage

1. Start SimOne and create a new use case.

2. Configure the traffic flow interface in the use case (confirm that port 21568 is open).

3. Run the use case.

4. Run the example program or script and observe traffic objects moving along the preset trajectories in the simulation.

5. API Feature Support: API Feature Support