Use the Python API#
Drive ACRES from Python without the game. This page shows the two Python packages of the simulator: acres_core for the physics and the sensors, and acres_simulator for missions, skills and fleets.
Before you start
- Build ACRES Core and the run-time library. Refer to Install.
- Use the Python interpreter that you used for the build.
- Run each command from the repository root.
Set the Environment#
Put the packages and the build folders on the Python path.
export ACRES_ROOT=$PWD
export PYTHONPATH=$PWD/Core/Build:$PWD/Simulator:$PWD/Learning:$PWD/Harness:$PWD/Data
acres_simulator finds the run-time library in Runtime/Build. Set ACRES_RUNTIME_BUILD to use a different build folder.
Select the Layer#
| Package | You Give | ACRES Gives | Use It For |
|---|---|---|---|
acres_core |
The commands of each vehicle in each 0.1 s step | The vehicle state, the wheels, the LiDAR, the crop marks | Physics research, controllers, data |
acres_simulator |
A mission, a skill or a chunk of path | The drive with the vehicle software: route, driver, shield, checker | Planners, learned skills, many vehicles |
acres_harness |
A scenario and a model | A full run with logs, scores and a viewer | Mission planners and language models |
The harness tutorial shows acres_harness. This page shows the other two.
Drive a Batch of Vehicles#
acres_core.Batch holds many independent environments. Each environment has one Polaris on the ACRE farm.
-
Write this script into the file
batch_drive.py.batch_drive.py"""Drive 256 Polaris environments to goals with a simple steering law.""" import time import numpy as np import acres_core as ac N = 256 b = ac.Batch(".", num_envs=N) S = {name: i for i, name in enumerate(ac.STATE_COLUMNS)} # All environments start at one point in field 14. Each goal is on a circle of 50 m around the start. start = np.array([-585.0, 75.0]) bearing = np.linspace(0.0, 2.0 * np.pi, N, endpoint=False) goal = start + 50.0 * np.stack([np.cos(bearing), np.sin(bearing)], axis=1) b.reset(None, np.tile([start[0], start[1], np.pi / 2, 0.0], (N, 1))) state = b.state arrival = np.full(N, np.nan) t0 = time.perf_counter() for k in range(600): # 60 s at most to_goal = goal - state[:, [S["x"], S["y"]]] distance = np.hypot(to_goal[:, 0], to_goal[:, 1]) alpha = np.arctan2(to_goal[:, 1], to_goal[:, 0]) - state[:, S["yaw"]] alpha = (alpha + np.pi) % (2.0 * np.pi) - np.pi # the angle to the goal, -pi to pi # Steer to a point 6 m ahead in the direction of the goal. Turn fully when the goal is behind. curvature = np.where(np.abs(alpha) < np.pi / 2, 2.0 * np.sin(alpha) / 6.0, 0.15 * np.sign(alpha)) curvature = np.clip(curvature, -0.15, 0.15) speed = np.where(distance > 2.0, 3.0, 0.0) arrival = np.where(np.isnan(arrival) & (distance <= 2.0), k * 0.1, arrival) if not np.isnan(arrival).any(): break state = b.step_curvature_speed(curvature, speed) wall = time.perf_counter() - t0 simulated = N * state[0, S["time_s"]] print(f"{np.isfinite(arrival).sum()} of {N} environments arrived, mean time {np.nanmean(arrival):.1f} s") print(f"crushed crop: mean {state[:, S['crop_crushed_m2']].mean():.1f} m2, collisions {int(state[:, S['collision']].sum())}") print(f"{simulated:.0f} s of simulation in {wall:.2f} s of wall time: {simulated / wall:.0f} times real time") -
Run the script.
Expected result
All environments arrive. The crop counter shows the soybean that the wheels crushed in field 14.
256 of 256 environments arrived, mean time 21.8 s crushed crop: mean 115.1 m2, collisions 0 7168 s of simulation in 7.48 s of wall time: 959 times real timeThe speed depends on the processor. This result is from 4 threads.
The loop has three parts: read the state, compute the commands with NumPy, and call step_curvature_speed.
One call moves each environment forward by 0.1 s.
The environments do not see each other. Each environment has its own vehicle, crop marks and ruts.
Faster steps
Set farm=False in the Batch constructor when your task does not use the crop marks.
Send Drive-by-Wire Commands#
step_curvature_speed is a short form. step takes the full drive-by-wire command of each environment, as the real vehicle gets it. A NaN value means "no message".
import math
import numpy as np
import acres_core as ac
batch = ac.Batch(".", num_envs=1)
S = {name: i for i, name in enumerate(ac.STATE_COLUMNS)}
C = {name: i for i, name in enumerate(ac.COMMAND_COLUMNS)}
batch.reset([0], np.array([[-585.0, 75.0, math.pi / 2, 0.0]])) # field 14, heading north
command = np.full((1, len(ac.COMMAND_COLUMNS)), np.nan)
command[0, C["enable"]] = 1
command[0, C["steer_mode"]], command[0, C["steer_value"]] = 3, 0.0 # 3: curvature, 1/m
command[0, C["ulc_mode"]], command[0, C["ulc_value"]] = 1, 2.0 # 1: velocity, m/s
for _ in range(100): # 10 s
state = batch.step(command)
print(f"after 10 s: speed {state[0, S['speed']]:.2f} m/s, x {state[0, S['x']]:.1f}, y {state[0, S['y']]:.1f}")
ac.COMMAND_COLUMNS and ac.STATE_COLUMNS give the names of the columns. The ACRES Core Python API gives each column with its unit.
Scan with the LiDAR#
lidar casts the rays of a sensor from the pose of each environment. Continue the script above:
ranges, material = batch.lidar([0], pattern="planar") # 360 beams in one level ring
ranges, material = batch.lidar([0], pattern="helios") # 1800 columns x 32 rings
print("helios:", ranges.shape, "returns", int(np.isfinite(ranges).sum()))
A NaN range means no return. material gives the class of each hit. The options crops=True, intensity=True and sweep=True add the crops, the reflectivity and the motion during the sweep. The Sensors tutorial shows more.
Put an Object on the Path#
set_objects puts trucks and persons into one environment. A contact with the vehicle sets the state column collision.
O = {name: i for i, name in enumerate(ac.OBJECT_COLUMNS)}
x, y, yaw = state[0, S["x"]], state[0, S["y"]], state[0, S["yaw"]]
person = np.zeros((1, len(ac.OBJECT_COLUMNS)))
person[0, [O["kind"], O["x"], O["y"], O["yaw"], O["length_m"], O["width_m"], O["height_m"]]] = \
[ac.OBJECT_KINDS.index("person"), x + 8 * math.cos(yaw), y + 8 * math.sin(yaw), 0.0, 0.5, 0.5, 1.8]
batch.set_objects(0, person)
for k in range(60):
state = batch.step(command)
if state[0, S["collision"]]:
print(f"contact with the person after {k * 0.1:.1f} s")
break
This drive has no safety software: the commands go to the vehicle unchanged. The next section uses the vehicle software, which stops before the person.
Drive Skills with the Vehicle Software#
acres_simulator drives a vehicle through skills: goto, scout, return_home, wait, stop and resume. The run-time library plans the route and runs the vehicle software: the driver, the shield and the trajectory checker.
-
Write this script into the file
scout.py.scout.py"""Scout field F56 with the skill API, record the episode log, and print the events.""" import time import acres_simulator as acres world = acres.Core(num_envs=1, threads=2) skills = acres.Skills(world, agent=0) skills.start_log("scout_f56.mcap") t0 = time.perf_counter() result = skills.scout("F56") print(f"scout F56: {result.code} at {result.time_s:.0f} s of mission time ({time.perf_counter() - t0:.1f} s of wall time)") result = skills.return_home() print(f"return_home: {result.code} at {result.time_s:.0f} s") skills.stop_log() for event in skills.events(): print(f"{event['mission_time_s']:7.1f} {event['type']}") -
Run the script.
Expected result
The vehicle drives from the home pose H to field F56, scouts the loop of the field and comes back.
A skill ends with a result code: done, blocked, refused_plan, refused_limit, refused_request, stopped or failed. After blocked, result.counterexample gives the rule of the shield that held the vehicle and the obstacle distance.
The file scout_f56.mcap is an episode log. The Replay tutorial reads it and shows it in the game.
Run Many Vehicles#
A fleet drives a queue of missions on many vehicles at the same time. A driver class can change the path chunks that the vehicles drive. The trajectory checker checks each chunk before the vehicle uses it.
import acres_simulator as acres
world = acres.Core(num_envs=2, threads=2)
fleet = acres.Fleet(world, grip="unknown")
fleet.load(missions=["single-F56", "single-F04"], condition="dry")
class SlowerDriver(acres.Driver):
"""Proposes the chunks of the default driver at 80 % of their speed."""
def propose(self, obs):
chunks = fleet.default_chunks(n=16)
chunks["speed_mps"] *= 0.8
return chunks
acres.run(fleet, driver=SlowerDriver())
for mission, score in zip(fleet.missions, fleet.scores()):
print(mission, "success" if score["success"] else score["failure"], f'{score["time_s"]:.0f} s')
A mission name single-<field> is one field from H and back. The planner of the library plans it with the cost tables of the condition set (dry, wet or shift). The first plan of a map makes the cost tables. This takes some minutes, and the cache keeps them.
Select a Map and Plan a Route#
ACRES has the default map and seven map variants. One process uses one map. Select it before you make a world.
import acres_simulator as acres
from acres_simulator import maps, plans
maps.use("acre-mixed") # or set ACRES_MAP=acre-mixed before you start Python
route = plans.route("F28") # from H to field F28, its scout loop, and back to H
print(route["fields"], round(route["path_length_m"]), "m,", round(route["path_time_s"]), "s")
The Configure the Simulation tutorial gives the map variants and the other settings.
Next Steps#
-
Record episode logs and sensor data sets.
-
Read a log in Python and show it in the game.
-
Train a model that drives through the vehicle software.
-
Run scenarios with a planner or a language model.