Traffic environment¶

class routerl.environment.TrafficEnvironment(seed: int = 23423, create_agents: bool = True, create_paths: bool = True, save_detectors_info: bool = False, action_masks: dict = None, generate_asgn_data: bool = False, agents: list = None, **kwargs)[source]

A PettingZoo AECEnv interface for optimal route choice using SUMO simulator. This environment is designed for the training of human agents (rational decision-makers) and machine agents (reinforcement learning agents).

See SUMO for details on SUMO.

See PettingZoo for details on PettingZoo.

Note

Users can configure the experiment with keyword arguments, see the structure below. Moreover, users can provide custom demand data in agents.csv inside the configured records folder (training_records by default). You can refer to the structure of such a file here.

Parameters:

seed (int, optional) – Random seed for reproducibility. Defaults to 23423.
create_agents (bool, optional) – Whether to create agent data. Defaults to True.
create_paths (bool, optional) – Whether to generate paths. Defaults to True.
save_detectors_info (bool, optional) – Whether to collect and save detector data. Defaults to False.
action_masks (dict[tuple[int, int], np.ndarray] | None, optional) – Optional mapping from (origin, destination) pairs to binary action masks. Each mask is a 1D NumPy array of 0/1 values with length equal to the action space size. Masks are used for route handling, human-agent actions, free-flow times, and compatible observations. Defaults to None.
generate_asgn_data (bool, optional) – Whether to generate per-timestep departure and snapshot files in SUMO_output/. Defaults to False.
agents (list | None, optional) – Agents used in the environment. When None, agents are generated or read from agents.csv according to create_agents. Defaults to None.
**kwargs (dict, optional) – User-defined parameter overrides. These override default values from defaults.json and allow experiment configuration.

Keyword arguments (see the usage below):

agent_parameters (dict, optional):
Agent settings.

num_agents (int, default=100):
Total number of agents.

new_machines_after_mutation (int, default=25):
Number of humans converted to machines.

machine_parameters (dict):
Machine agent settings.

behavior (str, default=”selfish”):
Route choice behavior. Options: selfish, competitive, collaborative, cooperative, social, altruistic, malicious, collectivist, militant. Alternatively, passing an iterable of 4 floats (e.g., [0.1, -0.2, 0.3, -0.4]) will be interpreted as a custom behavior vector.

observed_span (int, default=300):
Time window considered for observations.

observation_type (str, default=”trip_info_eta”):
Type of observation. Options: previous_agents, previous_agents_plus_start_time, previous_agents_plus_start_time_detector_data, trip_info_eta, trip_info_eta_mask_norm, trip_info_eta_route_congestion, route_congestion, trip_info_eta_sumo. previous_agents_plus_start_time_detector_data requires save_detectors_info=True.

group_vicinity (bool, default=False):
Whether group-based machine rewards only consider agents departing within observed_span of the machine agent.

human_parameters (dict):
Human agent settings.

model (str, default=”gawron”):
Decision-making model (options: aon, gawron, culo, random, weighted).

beta (float, default=1.5):
Positive value, multiplier of reward (travel time) used in utility, determines sensitivity.

beta_randomness (float, default=0.1):
Agent-specific randomness in beta.

alpha (float, default=0.2):
Human learning rate.

deterministic (bool, default=False):
Whether gawron selects the highest-utility path deterministically instead of sampling stochastically.

remember (int, default=5):
Number of previous actions to remember for learning, used in weighted model.

environment_parameters (dict, optional):
Environment settings.

number_of_days (int, default=1):
Number of days in the scenario.

save_every (int, default=1):
Save the episode data to disk every X days.

simulator_parameters (dict, optional):
SUMO simulator settings.

network_name (str, default=”csomor”):
Network name (e.g., arterial, cologne, grid)

custom_network_folder (str, default=”NA”):
Path to the folder containing a custom network.

simulation_timesteps (int, default=3600):
Total simulation time in seconds.

sumo_type (str, default=”sumo”):
SUMO execution mode (sumo or sumo-gui).

stuck_time (int, default=600):
Number of seconds to tolerate before teleporting a stopped vehicle to resolve gridlocks.

daily_reseed (bool, default=False):
Whether to change SUMO seed in each reset. If False, the seed will remain constant throughout the simulation.

use_libsumo (bool, default=False):
Whether to use libsumo instead of TraCI. Avoid using both use_libsumo=True and sumo_type=sumo-gui at the same time. Visit https://sumo.dlr.de/docs/Libsumo.html for more insight.

use_sumo_teleport (bool, default=False):
If set to True teleport logic will be handled by SUMO. Otherwise custom python logic will be used.

path_generation_parameters (dict, optional):
Path generation settings.

number_of_paths (int, default=3):
Number of routes per OD.

beta (float, default=-3.0):
Sensitivity to travel time in path generation.

weight (str, default=”time”):
Optimization criterion.

num_samples (int, default=100):
Number of samples for path generation.

path_gen_workers (int, default=4):
Maximum number of worker processes used for parallel path generation and path visualization.

origins (str | list[str], default=”default”):
Origin points from the network. (e.g., ["-25166682#0", "-4936412"])

destinations (str | list[str], default=”default”):
Destination points from the network. (e.g., ["-115604057#1", "-279952229#4"])

visualize_paths (bool, default=True):
Whether to visualize generated paths. Visuals will be saved in the plotter_parameters/plots_folder.

plotter_parameters (dict, optional):
Plotting & logging settings.

records_folder (str, default=”training_records”):
Directory for training records.

plots_folder (str, default=”plots”):
Directory for plots.

plot_choices (str, default=”all”):
Selection of plots to be generated. Options: none, basic, all.

smooth_by (int, default=50):
Smoothing parameter for plots.

phases (list[int], default=[0, 100]):
X-axis positions for phase markers.

phase_names (list[str], default=[“Human learning”, “Mutation - Machine learning”]):
Phase names for labeling phase markers.

clear_records (bool, default=True):
Whether to clear existing episode, detector, and SUMO output records during initialization.

Usage:

Case 1

% Your file structure in the beginning
project_directory/
|-- your_script.py

>>> # Environment initialization
... env = TrafficEnvironment(
...     seed=42,
...     agent_parameters={
...         "num_agents": 5,
...         "new_machines_after_mutation": 1,
...         "machine_parameters": {
...             "behavior": "selfish",
...             }},
...     simulator_parameters={"sumo_type": "sumo-gui"},
...     path_generation_parameters={"number_of_paths": 2}
... )

% File structure after the initialization:
project_directory/
|-- your_script.py
|-- training_records/
|   |-- agents.csv
|   |-- routes.csv
|   |-- route.rou.xml
|   |-- detector/
|   |   |--             % to be populated during simulation
|   |-- episodes/
|   |   |--             % to be populated during simulation
|-- plots/
|   |-- 0_0.png
|   |-- ...             % visuals of generated paths for each OD
|   |-- ...             % to be populated after the experiment

Case 2

% Your file structure in the beginning
project_directory/
|-- your_script.py
|-- training_records/
|   |-- agents.csv      % your custom demand, conforming to the structure

Warning

Demand data in agents.csv should be aligned with the specified experiment settings (e.g., number of agents, number of origins and destinations, etc.).

>>> env = TrafficEnvironment(
...     create_agents=False, # Environment will use your agent data
...     agent_parameters={
...         "new_machines_after_mutation": 10,
...         "machine_parameters": {
...             "behavior": "selfish",
...             }},
...     simulator_parameters={"network_name": "arterial"},
...     path_generation_parameters={"number_of_paths": 3}
... )

% File structure after the initialization:
project_directory/
|-- your_script.py
|-- training_records/
|   |-- agents.csv      % stays the same, used for agent generation
|   |-- routes.csv
|   |-- route.rou.xml
|   |-- detector/
|   |   |--             % to be populated during simulation
|   |-- episodes/
|   |   |--             % to be populated during simulation
|-- plots/
|   |-- 0_0.png
|   |-- ...             % visuals of generated paths for each OD
|   |-- ...             % to be populated after the experiment

Warning

Setting create_paths=False requires an existing routes.csv and route.rou.xml in the configured records folder. routes.csv supplies route and free-flow metadata, while route.rou.xml supplies the routes to SUMO. It is otherwise advised to generate paths for each experiment.

Variables:

day (int) – Current day index in the simulation.
human_learning (bool) – Whether human agents are learning.
number_of_days (int) – Number of days to simulate.
action_space_size (int) – Size of the action space.
recorder (Recorder) – Object for recording simulation data.
simulator (SumoSimulator) – SUMO simulator instance.
all_agents (list) – List of all agent objects.
machine_agents (list) – List of all machine agent objects.
human_agents (list) – List of all human agent objects.
last_episode_had_teleports (bool) – Whether any agents were teleported in the last episode.
last_episode_travel_times (list) – List of all agents’ travel-time records from the last episode.

action_space(agent: str)[source]

Method that returns the action space of the agent.

Parameters:: agent (str) – The agent name.
Returns:: self._action_spaces[agent] (Any) – The action space of the agent.

close() → None[source]

Not implemented.

Returns:: None

get_free_flow_times(invalid_pad: float = 1000000000.0) → dict[source]

Retrieve free flow times for all origin-destination pairs from the simulator paths data.

Parameters:: invalid_pad (float, optional) – Value used for missing clustered-route slots. Defaults to 1e9.
Returns:: ff_dict (dict) – A dictionary where keys are tuples of origin and destination, and values are lists of free flow times.

get_observation() → tuple[source]

Retrieve the current observation from the simulator.

This method returns the current timestep of the simulation and the values of the episode actions.

Returns:: tuple – A tuple containing the current timestep and the episode actions.

get_observation_function() → Observations[source]

Returns an observation object based on the provided parameters.

Returns:: Observations – An observation object.
Raises:: ValueError – If the configured observation type is unknown.

multisync_env_factories(env_wrapper, count: int = 0) → list[source]

Create factories for environments used by TorchRL’s MultiSyncDataCollector.

Each episode is assumed to be one day long, and human agents are assumed not to learn.

Parameters:

env_wrapper (Callable) – Callable used for wrapping the environment. Should take env as an argument and return the wrapped environment.
count (int, optional) – Number of factories to return. 0 uses one fewer than the available CPU count. Defaults to 0.

Returns:

list – Environment factory callables.

mutation(disable_human_learning: bool = True, mutation_start_percentile: int = 25) → None[source]

Perform mutation by converting selected human agents into machine agents.

This method identifies human agents whose start times are above the configured percentile, removes a specified number of them, and replaces them with machine agents.

Parameters:

disable_human_learning (bool, default=True) – Whether to disable further learning by human agents.
mutation_start_percentile (int, default=25) – The percentile threshold for selecting human agents for mutation. Set to -1 to disable this filter.

Returns:

None

Raises:

ValueError – If there are insufficient human agents available for mutation.

observation_space(agent: str)[source]

Method that returns the observation space of the agent.

Parameters:: agent (str) – The agent name.
Returns:: self._observation_spaces[agent] (Any) – The observation space of the agent.

observe(agent: str) → ndarray[source]

Retrieve the observations for a specific agent.

Parameters:: agent (str) – The identifier for the agent whose observations are to be retrieved.
Returns:: np.ndarray – The current observation for the specified agent.

plot_results() → None[source]

Plot the recorded simulation results.

Returns:: None

reset(seed: int = None, options: dict = None) → tuple[source]

Resets the environment.

Parameters:

seed (int, optional) – Reserved for PettingZoo API compatibility; currently unused. Defaults to None.
options (dict, optional) – Reserved for PettingZoo API compatibility; currently unused. Defaults to None.

Returns:

observations (dict) – observations.
infos (dict) – dictionary of information for the agents.

simulation_loop(machine_action: int, machine_id: int | str) → None[source]

This function contains the integration of the agent’s actions to SUMO.

We iterate through all the time steps of the simulation. For each timestep there are none, one or more than one agents type (humans, machines) that start. If more than one machine agents have the same start time, we break from this function because we need to take the agent’s action from the STEP function.

Parameters:

machine_action (int) – Route index selected by the machine agent.
machine_id (int | str) – Identifier of the machine agent whose action is to be performed. Human-only simulations use 0.

Returns:

None

start() → None[source]

Start the connection with SUMO.

Returns:: None

step(machine_action: int = None) → None[source]

Step method.

Takes an action for the current agent (specified by agent_selection) and updates various parameters including rewards, cumulative rewards, terminations, truncations, infos, and agent_selection. Also updates any internal state used by observe().

Parameters:: machine_action (int, optional) – Route index selected for the current machine agent. None is used for human-only simulations and PettingZoo dead steps. Defaults to None.
Returns:: None

stop_simulation() → None[source]

End the simulation.

Returns:: None