flatland.trajectories.trajectories module

flatland.trajectories.trajectories module#

class flatland.trajectories.trajectories.Trajectory(data_dir: Path, ep_id: str = NOTHING, trains_positions: DataFrame | None = None, actions: DataFrame | None = None, trains_arrived: DataFrame | None = None, trains_rewards_dones_infos: DataFrame | None = None)[source]#

Bases: object

Encapsulates episode data (actions, positions etc.) for one or multiple episodes for further analysis/evaluation.

Aka. Episode Aka. Recording

In contrast to rllib (ray-project/ray), we use a tabular approach (tsv-backed) instead of `dict`s.

Directory structure: - event_logs

ActionEvents.discrete_action – holds set of action to be replayed for the related episodes. TrainMovementEvents.trains_arrived – holds success rate for the related episodes. TrainMovementEvents.trains_positions – holds the positions for the related episodes. TrainMovementEvents.trains_rewards_dones_infos – holds the rewards for the related episodes.

  • serialised_state

    <ep_id>.pkl – Holds the pickled environment version for the episode.

Indexing:

  • actions for step i are index i-1 (i.e. starting at 0)

  • positions before step i are indexed i-1 (i.e. starting at 0)

  • positions after step are indexed i (i.e. starting at 1)

action_collect(env_time: int, agent_id: int, action: RailEnvActions)[source]#
action_lookup(env_time: int, agent_id: int) → RailEnvActions[source]#

Method used to retrieve the stored action (if available). Defaults to 2 = MOVE_FORWARD.

Parameters#

env_time: int

action going into step env_time

agent_id: int

agent ID

Returns#

RailEnvActions

The action to step the env.

arrived_collect(env_time: int, success_rate: float, normalized_reward: float)[source]#
build_cache() → Tuple[dict, dict, dict][source]#
compare_actions(other: Trajectory, start_step: int | None = None, end_step: int | None = None, ignoring_waiting=False) → DataFrame[source]#
compare_arrived(other: Trajectory, start_step: int | None = None, end_step: int | None = None, skip_normalized_reward: bool = True) → DataFrame[source]#
compare_positions(other: Trajectory, start_step: int | None = None, end_step: int | None = None) → DataFrame[source]#
compare_rewards_dones_infos(other: Trajectory, start_step: int | None = None, end_step: int | None = None, ignoring_rewards: bool = False) → DataFrame[source]#
static create_empty(data_dir: Path, ep_id: str | None = None) → Trajectory[source]#

Create a new empty trajectory.

Parameters#

data_dirPath

the data dir backing the trajectory. Must be empty.

ep_id

the episode ID for the new trajectory. If not provided, a new UUID is generated.

Returns#

Trajectory

fork(data_dir: Path, start_step: int, ep_id: str | None = None) → Trajectory[source]#

Fork a trajectory to a new location and a new episode ID.

Parameters#

data_dirPath

the data dir backing the forked trajectory.

ep_idstr

the new episode ID for the fork. If not provided, a new UUID is generated.

start_stepint

where to start the fork

Returns#

Trajectory

load_env(start_step: int | None = None, inexact: bool = False, rewards: Rewards | None = None) → RailEnv | None[source]#

Restore an episode’s env.

Parameters#

start_stepOptional[int]

start from snapshot (if it exists)

inexactbool

allows returning the last snapshot before start_step

rewardsRewards

rewards for the loaded env. If not provided, defaults to the loaded env’s rewards.

Returns#

RailEnv

the rail env or None if the snapshot at the step does not exist

static load_existing(data_dir: Path, ep_id: str) → Trajectory[source]#

Load existing trajectory from disk.

Parameters#

data_dirPath

the data dir backing the trajectory.

ep_id

the ep_id - the data dir may contain multiple trajectories in the same data frames.

Returns#

Trajectory

property outputs_dir: Path#
persist()[source]#
position_collect(env_time: int, agent_id: int, position: Tuple[Tuple[int, int], int])[source]#
position_lookup(env_time: int, agent_id: int) → Tuple[Tuple[int, int], int][source]#

Method used to retrieve the stored position (if available).

Parameters#

env_time: int

position before (!) step env_time

agent_id: int

agent ID

Returns#

Tuple[Tuple[int, int], int]

The position in the format ((row, column), direction).

rewards_dones_infos_collect(env_time: int, agent_id: int, reward: float, info: Any, done: bool)[source]#
trains_arrived_lookup() → Series[source]#

Method used to retrieve the trains arrived for the episode.

Returns#

pd.Series

The trains arrived data.

trains_rewards_dones_infos_lookup(env_time: int, agent_id: int) → Tuple[float, bool, Dict][source]#

Method used to retrieve the rewards for the episode.

Parameters#

env_time: int

action going into step env_time

agent_id: int

agent ID

Returns#

pd.DataFrame

The trains arrived data.

Contents