factoriax.analysis

Tools for reading and drawing recorded episodes.

Trajectory is the shared input format. Every plot in this package takes one.

Importing this package imports the five submodules listed in __all__, and therefore matplotlib. The other modules (build_progression, categories, curriculum_strip, graph_layout, inventory, recipe_graph, recorder, and utils) are not re-exported here. Import them by their full path.

Modules

actions	Find temporal patterns in what an agent did, and draw them.
build_progression	Draw build snapshots as progression figures.
categories	Group the engine's actions and items into colored categories.
curriculum_strip	Render a graded-achievement curriculum as a horizontal phase strip.
eval	Hold one eval episode and draw its summary plots.
graph_layout	Place the nodes of an acyclic graph on a layered grid.
inventory	Draw a player inventory as a standalone RGB panel.
milestones	Find when key events first happen in an episode, and plot them.
recipe_graph	Render a recipe DAG as a layered crafting diagram.
recorder	Collect rollout chunks from a training loop into a trajectory.
state	Read and draw how the world changed over an episode.
trajectory	Hold the recorded episodes that every analysis function reads.
utils	Helpers that every plot in the analysis package shares.
video	Build video frames from env states and encode them to MP4.

class factoriax.analysis.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