factoriax.analysis.trajectory

Hold the recorded episodes that every analysis function reads.

Trajectory is the one input format of this package. It holds a batch of episodes, and each field is an array whose first two axes are the episode and the timestep.

Axis order

Every array starts with (B, T, ...), where B counts episodes and T counts timesteps. A field that belongs to one player adds a player axis after T, giving (B, T, P, ...). A map field adds the two map axes instead, giving (B, T, H, W, ...).

Only actions is required. Every other array field defaults to None, which means “not recorded” and never “empty”.

Scheme descriptors

Four dictionary fields record how a trajectory was produced: the observation scheme, the reward scheme, the cost scheme, and the engine parameters. Trajectory.save() writes them to the archive as JSON, under keys with a leading underscore, which keeps them apart from the array fields.

Examples

Build a trajectory from actions alone:

>>> import numpy as np
>>> from factoriax.analysis.trajectory import Trajectory
>>> actions = np.random.randint(0, 12, size=(16, 200, 2))
>>> traj = Trajectory(actions=actions)
>>> traj.num_episodes, traj.episode_length, traj.num_players
(16, 200, 2)

Record how it was produced:

>>> traj = Trajectory(
...     actions=actions,
...     observation_scheme={"type": "local", "radius": 7},
... )
>>> traj.observation_scheme["radius"]
7

Nothing runs these examples. pyproject.toml sets no --doctest-modules, so they are text and can go stale.

class factoriax.analysis.trajectory.Trajectory(actions: ndarray, positions: ndarray | None = None, player_directions: ndarray | None = None, player_inventory: ndarray | None = None, block_map: ndarray | None = None, block_resources: ndarray | None = None, machine_types: ndarray | None = None, tile_entity: ndarray | None = None, ent_y: ndarray | None = None, ent_x: ndarray | None = None, ent_type: ndarray | None = None, ent_direction: ndarray | None = None, ent_power: ndarray | None = None, ent_buf_type: ndarray | None = None, ent_buf_count: ndarray | None = None, ent_asm_in_type: ndarray | None = None, ent_asm_in_count: ndarray | None = None, ent_asm_out_type: ndarray | None = None, ent_asm_out_count: ndarray | None = None, ent_health: ndarray | None = None, selected_player: ndarray | None = None, achievements: ndarray | None = None, achievements_unlocked: ndarray | None = None, items_mined: ndarray | None = None, science_consumed_step: ndarray | None = None, rewards: ndarray | None = None, timesteps: ndarray | None = None, observation_scheme: dict[str, object] | None = None, reward_scheme: dict[str, object] | None = None, cost_scheme: dict[str, object] | None = None, env_params_scheme: dict[str, object] | None = None, _SCHEME_FIELDS: tuple[str, ...] = ('observation_scheme', 'reward_scheme', 'cost_scheme', 'env_params_scheme'))

Immutable container for batched episode data.

Every array field starts with (B, T, ...). B counts episodes and T counts timesteps. A player field adds a player axis after T. A map field adds the map axes instead.

The class is frozen, so a slice method returns a new trajectory and never changes the one it was called on. The arrays inside are not copied, so two trajectories can share memory and a caller that writes into one array affects both.

actions

The action taken at each step. Shape (B, T) for one player, or (B, T, P) for several. A 1-D array is accepted and gets a batch axis of 1.

Type:

numpy.ndarray

positions

Player positions as (B, T, P, 2), ordered (x, y) in tiles. Note the order: x is first, unlike the map fields, which are indexed [y, x] in row-major order.

Type:

numpy.ndarray | None

player_directions

The facing of each player as (B, T, P).

Type:

numpy.ndarray | None

player_inventory

Item counts as (B, T, P, NUM_ITEM_TYPES), indexed by ItemType value. This is a count for each item, not a list of slots.

Type:

numpy.ndarray | None

block_map

The tile type of each map cell as (B, T, H, W).

Type:

numpy.ndarray | None

block_resources

The resource left in each map cell as (B, T, H, W).

Type:

numpy.ndarray | None

machine_types

The machine on each map cell as (B, T, H, W).

Type:

numpy.ndarray | None

tile_entity

The entity index on each map cell as (B, T, H, W), or -1 for an empty cell.

Type:

numpy.ndarray | None

ent_y, ent_x

Entity positions as (B, T, MAX_M). A free slot holds a negative y.

ent_type, ent_direction, ent_power, ent_health

Per-entity fields as (B, T, MAX_M).

ent_buf_type, ent_buf_count

The buffer of each entity as (B, T, MAX_M).

ent_asm_in_type, ent_asm_in_count, ent_asm_out_type,

ent_asm_out_count

Assembler input and output slots as (B, T, MAX_M).

Type:

numpy.ndarray | None

selected_player

The index of the acting player as (B, T).

Type:

numpy.ndarray | None

achievements

The achievement mask at each step as (B, T, A), where A is the achievement count of the scenario.

Type:

numpy.ndarray | None

achievements_unlocked

The latched mask carried by the engine state, padded to MAX_ACHIEVEMENTS. This is wider than achievements and is not the same field.

Type:

numpy.ndarray | None

items_mined

Items mined so far as (B, T, ...).

Type:

numpy.ndarray | None

science_consumed_step

Science consumed at each step as (B, T, ...).

Type:

numpy.ndarray | None

rewards

The reward at each step as (B, T).

Type:

numpy.ndarray | None

timesteps

The engine timestep of each step as (B, T). This can differ from the position along the time axis when a recording skips steps.

Type:

numpy.ndarray | None

observation_scheme, reward_scheme, cost_scheme, env_params_scheme

Plain dictionaries that record how the trajectory was produced. None means “not recorded”.

Raises:

ValueError – When actions has more than three axes, or fewer than one.

Notes

Nothing checks that the fields agree with each other. A trajectory whose rewards are shorter than its actions is built without complaint and fails later, inside whichever function reads both.

episode(idx: int) → Trajectory

Return one episode as a trajectory of batch size 1.

Parameters:

idx – The episode to keep. A negative index counts from the end, as in a Python list.

Returns:

A new trajectory with B of 1. The time axis is kept, so the result is not squeezed to a single episode shape.

Return type:

Trajectory

property episode_length: int

Number of timesteps per episode.

episodes(s: slice | ndarray) → Trajectory

Return the selected episodes as a new trajectory.

Parameters:

s – Anything NumPy accepts on the first axis: a slice, an array of indexes, or a boolean mask.

Returns:

A new trajectory holding those episodes. Every recorded field is sliced the same way, and the scheme dictionaries are carried over unchanged.

Return type:

Trajectory

property is_multi_player: bool

Whether the trajectory has more than one player.

classmethod load(path: str) → Trajectory

Read a trajectory back from a .npz archive.

A field absent from the archive stays None. Each JSON entry with a leading underscore is read back into its scheme dictionary, so the values return as plain JSON types.

An old archive that stored _obs_type and _obs_radius as scalars is folded into observation_scheme. That fold runs only when the archive holds no observation_scheme of its own.

Parameters:

path – The archive to read.

Returns:

The trajectory the archive holds.

Return type:

Trajectory

Notes

The load runs with allow_pickle=True, which is what lets the JSON entries come back. Read only archives you trust, and treat one from outside the project as untrusted input.

property num_episodes: int

Number of episodes in the batch.

property num_players: int

Number of players (1 if single-player).

player(idx: int) → Trajectory

Return the view of one player as a new trajectory.

Parameters:

idx – The player to keep.

Returns:

A new trajectory with the player axis dropped from the player fields. A trajectory that is not multi-player is returned unchanged, including one of shape (B, T, 1), which therefore keeps its player axis. Use factoriax.analysis.utils.resolve_player_actions() when the result must always have two axes.

Return type:

Trajectory

save(path: str) → None

Write the trajectory to a compressed .npz archive.

Each array field is stored under its own name. Each scheme dictionary is stored as JSON bytes under its name with a leading underscore, such as _observation_scheme.

A field that is None is left out of the archive. An unrecorded field is therefore absent, and not present as a null, so load() can tell the two apart.

Parameters:

path – Where to write the archive. NumPy appends .npz when the name does not end with it. Parent directories are not created.

time_slice(start: int, end: int) → Trajectory

Return the timesteps from start to end as a trajectory.

Parameters:

• start – The first timestep to keep.

• end – The timestep after the last one to keep, as in a Python slice.

Returns:

A new trajectory holding that window of every recorded field. The timesteps field is sliced too, so it keeps the original engine timesteps and does not restart at 0.

Return type:

Trajectory

factoriax.analysis.trajectory.states_to_trajectory(states: list[EnvState], actions: np.ndarray | None = None, rewards: np.ndarray | None = None, params: EnvParams | None = None) → Trajectory

Stack a list of engine states into a one-episode trajectory.

Every array field of the states is stacked along a new time axis, and a batch axis of 1 is added in front. The result therefore has B of 1 and T equal to the number of states.

Parameters:

• states – The states, one for each timestep, in order.

• actions – The action at each step, of shape (T,) or (T, P), or None for all zeros. Nothing compares its length against the state count, so a mismatch gives a trajectory whose action axis and time axis disagree.

• rewards – The reward at each step, of shape (T,), or None to record none.

• params – The engine parameters that produced the states, or None. When given, their dictionary form is stored in env_params_scheme so replay code can rebuild them.

Returns:

A trajectory with B of 1. Its timesteps field counts from 0 to T - 1, whatever the states themselves report.

Return type:

Trajectory

Raises:

ValueError – When states is empty.

factoriax.analysis.trajectory.trajectory_to_states(traj: Trajectory, episode: int = 0) → list[EnvState]

Rebuild the engine states of one episode from a trajectory.

This is the inverse of states_to_trajectory(). Only the fields the trajectory recorded are set. A field that EnvState requires but the trajectory lacks raises inside EnvState, and not here.

The achievements field is skipped. It is per-step and only as wide as the achievement count of the scenario, while EnvState carries the latched achievements_unlocked at MAX_ACHIEVEMENTS.

Parameters:

• traj – The recording to read.

• episode – Which episode to rebuild.

Returns:

One EnvState for each timestep. The timestep of each state comes from the timesteps field, or from the position along the time axis when that field was not recorded.

Return type:

list

Notes

A trajectory from RolloutRecorder.finish() is padded with zeros to a common length. The states rebuilt from those pad steps look like real states at the origin with empty inventories.