Scenario Generation

Scenario Generation#

Rail Generators, Line Generators and Timetable Generators#

The separation between rail generation and schedule generation reflects the organisational separation in the railway domain

  • Infrastructure Manager (IM): is responsible for the layout and maintenance of tracks simulated by rail_generator.

  • Railway Undertaking (RU): operates trains on the infrastructure Usually, there is a third organisation, which ensures discrimination-free access to the infrastructure for concurrent requests for the infrastructure in a schedule planning phase simulated by line_generator and timetable_generator. However, in the Flatland challenge, we focus on the re-scheduling problem during live operations.

We can produce RailGenerators by completing the following:

from typing import Mapping, List

from numpy.random.mtrand import RandomState

from flatland.envs.line_generators import LineGenerator
from flatland.envs.distance_map import DistanceMap
from flatland.envs.agent_utils import EnvAgent
from flatland.envs.timetable_utils import Timetable
from flatland.envs.rail_env import RailEnv
from flatland.envs import rail_generators as rail_gen
from flatland.envs import line_generators as line_gen
import flatland.envs.timetable_generators as ttg
from flatland.utils import seeding

def sparse_rail_generator(max_num_cities=5, grid_mode=False, max_rails_between_cities=4,
                          max_rail_pairs_in_city=4, seed=0):
    def generator(width, height, num_agents, num_resets=0):
        # generate the grid and (optionally) some hints for the line_generator
        ...

        return grid_map, {'agents_hints': {
            'num_agents': num_agents,
            'city_positions': city_positions,
            'train_stations': train_stations,
            'city_orientations': city_orientations
        }}

    return generator

Similarly, LineGenerators:

def sparse_line_generator(speed_ratio_map: Mapping[float, float] = None) -> LineGenerator:
    def generator(rail: GridTransitionMap, num_agents: int, hints: Any = None):
        # place agents:
        # - initial position
        # - initial direction
        # - targets
        # - speed data
        # - malfunction data
        ...

        return agents_position, agents_direction, agents_target, speeds, agents_malfunction

    return generator

And finally, timetable_generator is called within the RailEnv’s reset() during line generation to create a time table for the trains.

def timetable_generator(agents: List[EnvAgent], distance_map: DistanceMap,
                        agents_hints: dict, np_random: RandomState = None) -> Timetable:
    # specify:
    # - earliest departures
    # - latest arrivals
    # - max episode steps
    ...

    return Timetable(earliest_departures, latest_arrivals, max_episode_steps)

env = RailEnv(
    width=30,
    height=30,
    rail_generator=rail_gen.sparse_rail_generator(
            max_num_cities=2,
            seed=42,
            grid_mode=False,
            max_rails_between_cities=2,
            max_rail_pairs_in_city=2
        ),
        line_generator=line_gen.sparse_line_generator(speed_ratio_map={1.0: 0.25, 0.5: 0.25, 0.33: 0.25, 0.25: 0.25}, seed=42),
        timetable_generator=ttg.timetable_generator,
)
obs, info = env.reset()
info

{'action_required': {0: False, 1: False},
 'malfunction': {0: 0, 1: 0},
 'speed': {0: 0.3333333333333333, 1: 0.3333333333333333},
 'state': {0: <TrainState.WAITING: 0>, 1: <TrainState.WAITING: 0>}}

Inside reset(), rail, line and timetable generators are called as follows:

rail, optionals = rail_gen.sparse_rail_generator(
            max_num_cities=2,
            seed=42,
            grid_mode=False,
            max_rails_between_cities=2,
            max_rail_pairs_in_city=2
        )(30,30,2)
optionals

{'agents_hints': {'city_positions': [(10, 7), (24, 12)],
  'train_stations': [[((10, 7), 0), ((10, 8), 1)],
   [((24, 12), 0), ((24, 13), 1)]],
  'city_orientations': [<Grid4TransitionsEnum.SOUTH: 2>,
   <Grid4TransitionsEnum.NORTH: 0>]},
 'level_free_positions': set()}

line_gen.sparse_line_generator(speed_ratio_map={1.0: 0.25, 0.5: 0.25, 0.33: 0.25, 0.25: 0.25}, seed=42)(rail, 2, optionals["agents_hints"], np_random=seeding.np_random(42)[0])

Line(agent_positions=[[(10, 7)], [(24, 12)]], agent_directions=[[<Grid4TransitionsEnum.NORTH: 0>], [<Grid4TransitionsEnum.NORTH: 0>]], agent_targets=[(24, 13), (10, 8)], agent_speeds=[0.33, 1.0])

Notice that the rail_generator may pass agents_hints to the line_generator and timetable_generator which the latter may interpret. For instance, the way the sparse_rail_generator generates the grid, it already determines the agent’s goal and target. Hence, rail_generator, line_generator and timetable_generator have to match if line_generator presupposes some specific agents_hints. Currently, the only one used are the sparse_rail_generator, sparse_line_generator and the timetable_generator which works in conjunction with these.

Rail Generator#

Available Rail Generators#

Flatland provides the sparse_rail_generator, which generates realistic-looking railway networks.

Sparse rail generator#

The idea behind the sparse rail generator is to mimic classic railway structures where dense nodes (cities) are sparsely connected to each other and where you have to manage traffic flow between the nodes efficiently. The cities in this level generator are much simplified in comparison to real city networks but they mimic parts of the problems faced in daily operations of any railway company.

sparse rail

There are a number of parameters you can tune to build your own map and test different complexity levels of the levels.

Note

Some combinations of parameters do not go well together and will lead to infeasible level generation. In the worst case, the level generator will issue a warning when it cannot build the environment according to the parameters provided.

You can see that you now need both a rail_generator and a line_generator to generate a level. These need to work nicely together. The rail_generator will generate the railway infrastructure and provide hints to the line_generator about where to place agents. The line_generator will then generate a Line by placing agents at different train stations and providing them with individual targets.

You can tune the following parameters in the sparse_rail_generator:

  • max_num_cities: Maximum number of cities to build. The generator tries to achieve this numbers given all the other parameters. Cities are the only nodes that can host start and end points for agent tasks (train stations).

  • grid_mode: How to distribute the cities in the path, either equally in a grid or randomly.

  • max_rails_between_cities: Maximum number of rails connecting cities. This is only the number of connection points at city border. The number of tracks drawn in-between cities can still vary.

  • max_rails_in_city: Maximum number of parallel tracks inside the city. This represents the number of tracks in the train stations.

  • seed: The random seed used to initialize the random generator. Can be used to generate reproducible networks.

🎢 Over- and underpasses (aka. level-free diamond crossings)#

This feature was introduced in 4.0.5

Description#

Introduce level-free crossings. This reflects core railway domain features.

In particular, Diamond crossing can be defined to be level-free, which allows two trains to occupy the cell if one runs horizontal and the other vertical.

Implementation#

SparseRailGen has a new option

        p_level_free : float
            Percentage of diamond-crossings which are level-free.

RailEnv keeps tracks of level-free diamond crossings:

    self.level_free_positions: Set[Vector2D] = set()

The RailEnv will then allow two agents to be in the same cell concurrently if one is running horizontally and the other is running vertically.

Line Generator#

🚄 Speed profiles (aka. Multi-Speed)#

This feature was introduced in 3.0.0

Finally, trains in real railway networks don’t all move at the same speed. A freight train will for example be slower than a passenger train. This is an important consideration, as you want to avoid scheduling a fast train behind a slow train!

Agents can have speed profiles, reflecting different train classes (passenger, freight, etc.).

One of the main contributions to the complexity of railway network operations stems from the fact that all trains travel at different speeds while sharing a very limited railway network. In Flatland 3 this feature will be enabled as well and will lead to much more complex configurations. Here we count on your support if you find bugs or improvements :).

The different speed profiles can be generated using the schedule_generator, where you can actually chose as many different speeds as you like. Keep in mind that the fastest speed is 1 and all slower speeds must be between 1 and 0. For the submission scoring you can assume that there will be no more than 5 speed profiles.

Later versions of Flatland might have varying speeds during episodes. Therefore, we return the agent speeds. Notice that we do not guarantee that the speed will be computed at each step, but if not costly we will return it at each step. In your controller, you can get the agents’ speed from the info returned by step:

obs, rew, done, info = env.step(actions)
...
for a in range(env.get_num_agents()):
    speed = info['speed'][a]

import inspect

from flatland.envs.agent_utils import EnvAgent
from flatland.envs.rail_env import RailEnv
from flatland.envs.rail_env_action import RailEnvActions
from flatland.env_generation.env_generator import env_generator

env, _, _ = env_generator()
for _ in range(25):
    obs, rew, done, info = env.step({i: RailEnvActions.MOVE_FORWARD for i in env.get_agent_handles()})
print("after 25 steps")
for a in range(env.get_num_agents()):
    speed = info['speed'][a]
    print(f"\tagent {a} has speed {speed} in state {env.agents[a].state.name}")
print("after 26 steps")
obs, rew, done, info = env.step({i: RailEnvActions.STOP_MOVING for i in env.get_agent_handles()})
for a in range(env.get_num_agents()):
    speed = info['speed'][a]
    print(f"\tagent {a} has speed {speed} in state {env.agents[a].state.name}")

/opt/hostedtoolcache/Python/3.10.17/x64/lib/python3.10/site-packages/flatland/envs/rail_generators.py:321: UserWarning: Could not set all required cities! Created 1/2
  warnings.warn(city_warning)
/opt/hostedtoolcache/Python/3.10.17/x64/lib/python3.10/site-packages/flatland/envs/rail_generators.py:217: UserWarning: [WARNING] Changing to Grid mode to place at least 2 cities.
  warnings.warn("[WARNING] Changing to Grid mode to place at least 2 cities.")

after 25 steps
	agent 0 has speed 0.25 in state MOVING
	agent 1 has speed 1.0 in state WAITING
	agent 2 has speed 0.25 in state MOVING
	agent 3 has speed 1.0 in state WAITING
	agent 4 has speed 1.0 in state WAITING
	agent 5 has speed 0.25 in state WAITING
	agent 6 has speed 0.5 in state WAITING
after 26 steps
	agent 0 has speed 0 in state STOPPED
	agent 1 has speed 1.0 in state WAITING
	agent 2 has speed 0 in state STOPPED
	agent 3 has speed 1.0 in state WAITING
	agent 4 has speed 1.0 in state WAITING
	agent 5 has speed 0.25 in state WAITING
	agent 6 has speed 0.5 in state WAITING

📉 Multi-stop Schedules (w/o alternatives/routing flexibility)#

This feature was introduced in 4.0.5

Description#

Introduce intermediate targets in schedule and reward function. This reflects core railway domain features.

In particular, Flatland timetable can have several intermediate targets with time window earliest, latest. (Negative) rewards for not serving intermediate targets or not respecting earliest/latest window can be configured. Schedule generator can be configured with number of intermediate targets.

Implementation#

flatland.envs.line_generators.SparseLineGen takes an additional option

        line_length : int
            The length of the lines. Defaults to 2.

A Line now allows for multiple intermediate positions/directions and a Timetable contains a time window for each stop:

import inspect
import flatland.envs.timetable_utils
print("".join(inspect.getsourcelines(flatland.envs.timetable_utils)[0]))

from typing import List, NamedTuple

from flatland.core.grid.grid4 import Grid4TransitionsEnum
from flatland.core.grid.grid_utils import IntVector2DArray, IntVector2DArrayArray

Line = NamedTuple('Line', [
    # positions and directions without target (which has no direction)
    ('agent_positions', IntVector2DArrayArray),
    ('agent_directions', List[List[Grid4TransitionsEnum]]),
    ('agent_targets', IntVector2DArray),
    ('agent_speeds', List[float]),
])

Timetable = NamedTuple('Timetable', [
    # earliest departures and latest arrivals including None for latest arrival at initial and None for earliest departure at target
    ('earliest_departures', List[List[int]]),
    ('latest_arrivals', List[List[int]]),
    ('max_episode_steps', int)
])

In addition, Rewards introduces 3 new penalties for intermediate stops:

    - intermediate_not_served_penalty = -1
    - intermediate_late_arrival_penalty_factor = 0.2
    - intermediate_early_departure_penalty_factor = 0.5

Note that earliest_departure at the initial position is enforced by the RailEnv (i.e. an agent cannot start before that timestep) whereas the time windows for intermediate stops are not enforced by the RailEnv but penalized only by the Rewards configuration.

Timetable Generator#

This feature was introduced in flatland 3.0.0

Background#

Up until this point, the trains in Flatland were allowed to depart and arrive whenever they desired, the only goal was to make every train reach its destination as fast as possible. However, things are quite different in the real world. Timing and punctuality are crucial to railways. Trains have specific schedules. They are expected to depart and arrive at particular times.

This concept has been introduced to the environment in Flatland 3.0. Trains now have a time window within which they are expected to start and reach their destination.

Contents