Skip to content

Environments API

Environment APIs cover the PettingZoo-compatible environments, labelled wrappers, and episode logs used by training and experiment notebooks.

Shared Wrappers and Logs

src.environments.episode_log.EpisodeLog dataclass

src.environments.labelled_wrapper.LabelledEnv

Bases: BaseParallelWrapper

Generic labelled wrapper with episode logging + 'cost' tracking. Subclasses must implemented safe.

record_shield_raise(prev_floor, new_floor)

Called by an outer ShieldedEnv when its level rises.

Gridworld Environments

src.environments.gridworlds.markov_stag_hunt.env.MarkovStagHunt

Bases: ParallelEnv

state()

Global, agent-agnostic state (same content as obs), dtype=uint8.

num_cells()

Number of spatial cells in the layered observation grid.

channel_names()

Human-friendly names for each observation channel index.

Returns a list of length n_obs_types such that index i in the list corresponds to channel i in the (H, W, C) observation tensor.

action_names(action)

Human-friendly name for an action integer ID.

get_state()

Tuple layout

( agents: tuple[str, ...], results: tuple[int,int,int], # (stags_hunted, plants_harvested, maulings) damaged: tuple[int, ...], # ordered by possible_agents agent_positions: tuple[tuple[int,int]|None, ...] # ordered by possible_agents agent_counts: np.ndarray, # (H,W) int16 plants: tuple[tuple[int,int], ...], # sorted coords stags: tuple[tuple[int,int], ...], # sorted coords rng_state: object )

src.environments.gridworlds.ice_duel.env.IceDuel

Bases: ParallelEnv

state()

Return a global, agent-agnostic state (same structure as observations).

num_cells()

Number of spatial cells in the layered observation grid.

channel_names()

Human-friendly names for each observation channel index.

Index i of the returned list corresponds to channel i in the layered observation grid (H, W, C) or its flattened version.

action_names(action)

Human-friendly name for an action integer ID.

get_state()

Tuple layout

( agents: tuple[str, ...], agent_pos: tuple[tuple[int,int]|None, tuple[int,int]|None], # (p0, p1) rng_state: object )

src.environments.gridworlds.pursuit.env.Pursuit

Bases: ParallelEnv

Pursuit Evasion (Parallel PettingZoo)

  • Agents: 2 or 3 (0 is the Guard; 1 and 2 are normal).
  • Actions: {Left, Right, Up, Down, Stay}. Moves into walls/bounds are ignored.
  • Collision: any two agents share a cell after the simultaneous move.
  • Goal: true iff agent 0 OR 1 is in the goal region.

Rewards default to 0; optional shaping provided via constructor. Observations (globalised): binary channels shared by all agents: [Wall, Goal, A0, A1, (A2)] with dtype=np.float32. If flatten_observations is True, this is flattened to (HWC,).

state_space property

Describe the shape/dtype of state().

state()

Global, agent-agnostic state (same content as obs), matches state_space().

num_cells()

Number of spatial cells in the layered observation grid.

channel_names()

Human-friendly names for each observation channel index.

Channels are

0: Wall 1: Goal 2+: Agents (0 = guard, others = normal agents)

Hidden episode time is not part of the public state contract.

action_names(action)

Human-friendly name for an action integer ID.

get_state()

Tuple layout

( agents: tuple[str, ...], collision_ever: bool, agent_pos: tuple[Coord, ...] # ordered by possible_agents rng_state: object )

src.environments.gridworlds.gathering.env.Gathering

Bases: ParallelEnv

state_space property

Global, agent-agnostic state spec (same as obs), dtype=uint8.

state()

Global, agent-agnostic state (identical to obs), dtype=uint8.

num_cells()

Number of spatial cells in the layered observation grid.

channel_names()

Human-friendly names for each observation channel index.

Channels

0 .. n_agents-1 : player_i_active n_agents .. 2n_agents-1 : player_i_frozen 2n_agents : apple

action_names(action)

Human-friendly name for an action integer ID.

get_state()

Tuple layout

( agents: tuple[str, ...], agent_pos: tuple[Coord|None, ...] # ordered by possible_agents tag_hits: tuple[int, ...] # ordered by possible_agents frozen_timer: tuple[int, ...] # ordered by possible_agents apples: tuple[Coord, ...] # sorted coords rng_state: object )

Matrix Environments

src.environments.matrix.bertrand.env.BertrandMatrix

Bases: ParallelEnv

Repeated 2-player Bertrand (Parallel PettingZoo)

Stage game (Row vs Column) with payoffs: - High vs Low : (S, T) - Low vs High: (T, S) - High vs High: (R, R) - Low vs Low : (P, P) # price war

Defaults: T=8, R=5, S=0, P=0 (all floats).

Observations (global, binary channels), shape is either (C,) if flattened or (1,1,C): For each agent i in {0,1}: - ch 2i + 0 = 1 if agent_i's last action was High, else 0 - ch 2i + 1 = 1 if agent_i's last action was Low, else 0 Plus: - ch 2*num_agents = 1 if last outcome was a price war (both Low), else 0

On the first round (after reset), all channels are 0.

Attributes kept for compatibility with your other envs
  • n_agents
  • n_obs_types
  • flatten_observations
  • state() / state_space

num_cells()

Number of spatial cells in the layered observation grid.

BertrandMatrix is non-spatial: the entire game state is a single global cell. This returns 1 regardless of whether observations are flattened or shaped (1, 1, C).

channel_names()

Human-friendly names for each observation channel index.

Index i of the returned list corresponds to channel i in the observation vector (or the last axis of the (1, 1, C) tensor).

action_names(action)

Human-friendly name for an action integer ID.

get_state()

Fully restorable snapshot (no deep clone).

Tuple layout

( agents: tuple[str, ...], last_actions: tuple[int|None, int|None], # (player_0, player_1) last_war: bool, cum_rewards: tuple[float, float], # (player_0, player_1) rng_state: object # np.random.RandomState.get_state() )

set_state(state)

Restore a snapshot produced by get_state().

src.environments.matrix.chicken.env.ChickenMatrix

Bases: ParallelEnv

Repeated 2-player Chicken (Parallel PettingZoo)

Stage game (Row vs Column) with payoffs: - Straight vs Swerve: (T, S) - Swerve vs Straight: (S, T) - Swerve vs Swerve: (R, R) - Straight vs Straight:(P, P) # crash

Defaults: T=3, R=2, S=1, P=0 (all floats).

Observations (global, binary channels), shape is either (C,) if flattened or (1,1,C): For each agent i in {0,1}: - ch 2i + 0 = 1 if agent_i's last action was Swerve, else 0 - ch 2i + 1 = 1 if agent_i's last action was Straight, else 0 Plus: - ch 2*num_agents = 1 if last outcome was a crash (both Straight), else 0

On the first round (after reset), all channels are 0.

Attributes kept for compatibility with your other envs
  • n_agents
  • n_obs_types
  • flatten_observations
  • state() / state_space

num_cells()

Number of spatial cells in the layered observation grid.

ChickenMatrix is non-spatial: the entire game state is a single global cell. This returns 1 regardless of whether observations are flattened or shaped (1, 1, C).

channel_names()

Human-friendly names for each observation channel index.

Channels (by construction in _obs): ch0 = player_0 Swerve ch1 = player_0 Straight ch2 = player_1 Swerve ch3 = player_1 Straight ch4 = crash flag

action_names(action)

Human-friendly name for an action integer ID.

get_state()

Fully restorable snapshot (no deep clone).

Tuple layout

( agents: tuple[str, ...], last_actions: tuple[int|None, int|None], # (player_0, player_1) last_crash: bool, cum_rewards: tuple[float, float], # (player_0, player_1) rng_state: object # np.random.RandomState.get_state() )

set_state(state)

Restore a snapshot produced by get_state().

src.environments.matrix.congestion.env.CongestionMatrix

Bases: ParallelEnv

6-agent congestion routing with 2 roads (A/B) plus a safe detour action. Each agent picks one route per round. Per-road cost is linear in the number of users on that road: cost_A = base_A + slope_A * nA cost_B = base_B + slope_B * nB The detour has a fixed cost independent of congestion: cost_D = detour_cost Agent reward = - cost_(chosen road) (more congestion => lower reward) Optional jam penalty adds extra negative reward to agents on an overfull road.

6 agents, base_A=1, base_B=1, slope_A=1, slope_B=1,

detour_cost=6, jam_threshold=5, jam_penalty=5.

Observations (global, binary channels), shape is (C,) if flattened or (1,1,C): - For each agent i in {0..N-1}: ch 3i + 0 = 1 if last action was RoadA, else 0 ch 3i + 1 = 1 if last action was RoadB, else 0 ch 3*i + 2 = 1 if last action was Detour, else 0 - One-hot for last round load of RoadA over (N+1) buckets [0..N] - One-hot for last round load of RoadB over (N+1) buckets [0..N] - Jam flag (1 if max(nA, nB) >= jam_threshold else 0)

On the first round (after reset), all action bits are 0 and load one-hots reflect (0 users on A, 0 users on B); jam=0.

Attributes kept for compatibility
  • n_agents
  • n_obs_types
  • flatten_observations
  • state() / state_space
  • num_cells() -> 1

num_cells()

Non-spatial global state → single cell.

channel_names()

Human-friendly names for each observation channel index.

Layout (see _obs): 0..3N-1 : per-agent action bits (A/B/Detour) in blocks of 3 3N .. 3N+N : loadA one-hot over [0..N] 3N+N+1 .. 3N+2N+1 : loadB one-hot over [0..N] last : jam flag

action_names(action)

Human-friendly name for an action integer ID.

get_state()

Returns a lightweight, fully-restorable snapshot.

Tuple layout

( agents: tuple[str, ...], last_actions: tuple[int|None, ...] # len = n_agents, in possible_agents order loadA: int, loadB: int, jam: bool, cum_rewards: tuple[float, ...] # len = n_agents, in possible_agents order rng_state: object # np.random.RandomState.get_state() )

set_state(state)

Restore a snapshot produced by get_state().

src.environments.matrix.dpgg.env.DPGGMatrix

Bases: ParallelEnv

Dynamic Public Goods Game (Parallel PettingZoo), 2 players.

(Markov + binary obs): - Pot is discretized to a fixed step size pot_step. - Internal pot state is an integer index: self._pot_idx - Observations include the exact pot index encoded as binary bits (uint8 0/1)

Observation channels (binary): For each agent i in {0,1}: ch 2i + 0 = 1 if agent_i last action was Contribute ch 2i + 1 = 1 if agent_i last action was Withhold Plus: pot_bits bits encoding pot_idx (LSB-first), where pot_idx in [0, pot_levels-1] pot_value = pot_idx * pot_step pot_levels = floor(pot_cap / pot_step) + 1

action_names(action)

Human-friendly name for an action integer ID.

get_state()

Tuple layout

( agents: tuple[str, ...], last_actions: tuple[int|None, int|None], # (p0, p1) pot_idx: int, cum_rewards: tuple[float, float], # (p0, p1) rng_state: object )

src.environments.matrix.inspection.env.InspectionMatrix

Bases: ParallelEnv

Repeated 2-player Inspection game (Parallel PettingZoo)

Players
  • Row (player_0): Inspector — actions {NotInspect=0, Inspect=1}
  • Col (player_1): Inspectee — actions {Comply=0, Violate=1}

Stage payoffs (Inspector, Inspectee): (NotInspect, Comply) -> (0, 0) (NotInspect, Violate) -> (-h, b) # undetected violation (Inspect, Comply) -> (-c, 0) (Inspect, Violate) -> (v - c, -f)

Defaults: b=5, f=10, c=2, h=4, v=3 (all floats)

Observations (global, binary channels), shape is either (C,) if flattened or (1,1,C): For each agent i in {0,1}: - ch 2i + 0 = 1 if agent_i's last action was action 0, else 0 - ch 2i + 1 = 1 if agent_i's last action was action 1, else 0 Plus: - ch 2*num_agents = 1 if last outcome was an undetected violation ((NotInspect, Violate)), else 0

On the first round (after reset), all channels are 0.

Attributes kept for compatibility with your other envs
  • n_agents
  • n_obs_types
  • flatten_observations
  • state() / state_space

num_cells()

Number of spatial cells in the layered observation grid.

InspectionMatrix is non-spatial: the entire game state is a single global cell. This returns 1 regardless of whether observations are flattened or shaped (1, 1, C).

channel_names()

Human-friendly names for each observation channel index.

By construction in _obs

ch0 = p0 action==0 (NotInspect) ch1 = p0 action==1 (Inspect) ch2 = p1 action==0 (Comply) ch3 = p1 action==1 (Violate) ch4 = undetected flag ((NotInspect, Violate))

action_names(action)

Human-friendly name for an action integer ID.

get_state()

Tuple layout

( agents: tuple[str, ...], last_actions: tuple[int|None, int|None], # (p0, p1) last_undetected: bool, cum_rewards: tuple[float, float], # (p0, p1) rng_state: object )