POMDPPlanners.environments.pacman_pomdp package

PacMan POMDP package with sprite-based visualization.

class POMDPPlanners.environments.pacman_pomdp.PacManPOMDP(maze_size=(7, 7), walls=None, initial_pellets=None, initial_pacman_pos=(0, 0), num_ghosts=1, initial_ghost_positions=None, initial_ghost_pos=None, pellet_reward=10.0, ghost_collision_penalty=-100.0, step_penalty=-1.0, win_reward=100.0, ghost_aggressiveness=2.0, ghost_coordination='independent', ghost_strategies=None, observation_noise_factor=0.3, max_observation_noise=1.5, dangerous_areas=None, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, discount_factor=0.95, name='PacManPOMDP', output_dir=None, debug=False, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0)

Bases: DiscreteActionsEnvironment

PacMan POMDP environment inspired by the classic arcade game.

This environment implements a simplified PacMan game where PacMan must collect pellets while avoiding a single ghost. The ghost position is only partially observable through noisy sensor readings.

Parameters:

• maze_size (Tuple[int, int])

• walls (Set[Tuple[int, int]] | None)

• initial_pellets (List[Tuple[int, int]] | None)

• initial_pacman_pos (Tuple[int, int])

• num_ghosts (int)

• initial_ghost_positions (List[Tuple[int, int]] | None)

• initial_ghost_pos (Tuple[int, int] | None)

• pellet_reward (float)

• ghost_collision_penalty (float)

• step_penalty (float)

• win_reward (float)

• ghost_aggressiveness (float)

• ghost_coordination (str)

• ghost_strategies (List[str] | None)

• observation_noise_factor (float)

• max_observation_noise (float)

• dangerous_areas (Set[Tuple[int, int]] | None)

• dangerous_area_radius (float)

• dangerous_area_penalty (float)

• discount_factor (float)

• name (str)

• output_dir (Path | None)

• debug (bool)

• reward_model_type (RewardModelType)

• penalty_decay (float)

maze_size

Grid dimensions as (rows, cols)

walls

Set of wall positions as (row, col) tuples

initial_pellets

List of initial pellet positions

pellet_reward

Reward for collecting a pellet

ghost_collision_penalty

Penalty for collision with ghost

step_penalty

Cost per action

win_reward

Reward for collecting all pellets

ghost_aggressiveness

Temperature parameter for ghost movement policy

observation_noise_factor

Multiplier for observation noise based on distance

max_observation_noise

Maximum noise standard deviation

dangerous_areas

List of (row, col) centers of circular hazard zones

dangerous_area_radius

Radius (in grid cells) defining each hazard zone

dangerous_area_penalty

Penalty subtracted when PacMan ends a step inside a zone

Example

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>>
>>> # Initialize environment
>>> env = PacManPOMDP(maze_size=(7, 7))
>>>
>>> # Get initial state and actions
>>> initial_state = env.initial_state_dist().sample()[0]
>>> actions = env.get_actions()
>>>
>>> # Sample complete step using convenience method
>>> action = actions[0]
>>> next_state, observation, reward = env.sample_next_step(initial_state, action)
>>>
>>> # Check terminal condition
>>> env.is_terminal(initial_state)
False

Example

Construct an env with a circular hazard zone — PacMan is penalised by dangerous_area_penalty whenever its realised next position lies inside the zone, but the zone does not block movement or terminate.

>>> import numpy as np
>>> np.random.seed(0)
>>> env = PacManPOMDP(
...     maze_size=(7, 7),
...     dangerous_areas={(3, 3)},
...     dangerous_area_radius=1.0,
...     dangerous_area_penalty=5.0,
... )
>>> state = env.initial_state_dist().sample()[0]
>>> _ = env.sample_next_step(state, env.get_actions()[0])
>>> env.dangerous_area_penalty
5.0

array_to_observation(arr)

Convert a flat numpy array back to a PacMan observation tuple.

Parameters:

arr (ndarray) – 1-D array of shape (2 * num_ghosts,).

Return type:

Tuple[Tuple[int, int], ...]

Returns:

Observation as tuple of (row, col) tuples.

cache_visualization(history, cache_path)

Cache visualization of episode history.

Parameters:

• history (List[StepData]) – List of StepData objects representing the episode

• cache_path (Path) – Path where the GIF should be saved

Return type:

None

compute_metrics(histories)

Compute environment-specific metrics.

Return type:

List[MetricValue]

Parameters:

histories (List[History])

get_actions()

Get all available actions.

Return type:

List[int]

get_ghost_positions(state)

Return ghost positions as a tuple of (row, col) pairs.

Return type:

Tuple[Tuple[int, int], ...]

Parameters:

state (ndarray)

get_metric_names()

Get names of PacMan POMDP specific metrics.

Return type:

List[str]

Returns:

List containing metric names including standard metrics (win_rate, avg_pellets_collected, avg_episode_length, avg_pacman_closest_ghost_distance, avg_collision_encounters, avg_dangerous_area_steps, avg_all_dangerous_encounters) and dynamically generated per-ghost distance metrics for multi-ghost scenarios (avg_pacman_ghost_0_distance, avg_pacman_ghost_1_distance, etc.). avg_all_dangerous_encounters is the per-step sum of ghost-collision and dangerous-area-step events; a step that is both counts twice.

get_observation_cpp_ctor_kwargs()

Return the kwargs dict passed to PacManObservationCpp.

Return type:

Dict[str, Any]

get_pacman_pos(state)

Return PacMan’s (row, col) position from a state array.

Return type:

Tuple[int, int]

Parameters:

state (ndarray)

get_pellets(state)

Return the tuple of active pellet positions.

Return type:

Tuple[Tuple[int, int], ...]

Parameters:

state (ndarray)

get_score(state)

Return the state’s score as a Python float.

Return type:

float

Parameters:

state (ndarray)

get_terminal(state)

Return whether the state is terminal.

Return type:

bool

Parameters:

state (ndarray)

get_transition_cpp_ctor_kwargs()

Return the cached per-env kwargs dict passed to PacManTransitionCpp.

Return type:

Dict[str, Any]

hash_action(action)

Return a hashable key consistent with action equality.

Used by tree-search planners to index action children of a belief node in O(1). The returned key MUST satisfy:

action_a == action_b   (per env's notion of equality)
==> hash_action(action_a) == hash_action(action_b)

Subclasses with non-hashable actions (e.g. np.ndarray) must override to return a hashable surrogate (tobytes() is the standard choice for ndarray actions, which mirrors the np.array_equal semantics used by the linear-scan fallback).

Parameters:

action (Any) – Action to hash.

Return type:

Hashable

Returns:

A hashable key derived from action.

property initial_ghost_pos: Tuple[int, int]

returns first ghost position.

Type:

Backward compatibility

initial_observation_dist()

Get the initial observation distribution.

Returns a live distribution that draws fresh noisy ghost-position observations from the true initial state on each sample call, instead of the previous Dirac wrapper around a single pre-drawn sample (which collapsed the entire initial-belief observation prior to a point mass).

Return type:

Distribution

initial_state_dist()

Get initial state distribution.

Return type:

DiscreteDistribution

is_equal_observation(observation1, observation2)

Check if two observations are equal.

Return type:

bool

Parameters:

• observation1 (Any)

• observation2 (Any)

is_terminal(state)

Check if state is terminal.

Return type:

bool

Parameters:

state (ndarray)

make_state(*, pacman_pos, ghost_positions, pellets=None, score=0.0, terminal=False)

Build a PacMan state array in the canonical layout.

The array layout is [pac_row, pac_col, g0_row, g0_col, ..., pellet_mask[0..P-1], score, terminal].

Parameters:

• pacman_pos (Tuple[int, int]) – PacMan grid position (row, col).

• ghost_positions (Tuple[Tuple[int, int], ...]) – Per-ghost positions as a tuple of length num_ghosts.

• pellets (Optional[Tuple[Tuple[int, int], ...]]) – Active pellet positions. None means every initial pellet is active (useful for constructing initial states).

• score (float) – Current game score.

• terminal (bool) – Whether the state is terminal.

Return type:

ndarray

Returns:

1-D float64 array of shape (self._state_dim,).

Raises:

ValueError – If any argument has the wrong type or length, or if a pellet position was not registered at env construction.

observation_log_probability(next_state, action, observations)

Log-probability of each candidate observation under (next_state, action).

Returns np.ndarray of shape (N,) where N is the number of candidate observations. Subclasses must implement.

Return type:

ndarray

Parameters:

• next_state (ndarray)

• action (int)

• observations (Any)

observation_log_probability_per_state(next_states, action, observation)

Log-probability of one observation under each candidate next-state.

Used by particle filters: given N candidate next-states and ONE observation, return N log-likelihoods.

The default implementation falls back to a per-state Python loop delegating to observation_log_probability(). Native-backed envs (those whose observation kernel exposes batch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.

Parameters:

• next_states (Any) – A sequence (length N) or ndarray of shape (N, *dim) of candidate next-states.

• action (int) – The action that was executed.

• observation (Any) – A single observation.

Return type:

ndarray

Returns:

ndarray of shape (N,) with log-probabilities or log-PDFs.

observation_to_array(obs)

Convert a PacMan observation tuple to a flat numpy array.

Parameters:

obs (Tuple[Tuple[int, int], ...]) – Observation as tuple of ghost (row, col) positions.

Return type:

ndarray

Returns:

1-D array of shape (2 * num_ghosts,).

reward(state, action, next_state=None)

Calculate immediate reward.

Uses the realised next_state when supplied (e.g. by Environment.sample_next_step()) so the collision penalty and win bonus reflect the same stochastic ghost transition as the trajectory rather than a fresh independent draw. When next_state is None, falls back to sampling one here.

Return type:

float

Parameters:

• state (ndarray)

• action (int)

• next_state (Any)

reward_batch(states, action, next_states=None)

Calculate rewards for a batch of states.

Accepts a 2-D numpy array of shape (N, state_dim) on the fast vectorized path, or a sequence of 1-D state arrays on the fallback per-particle path.

Without next_states, computes deterministic reward components only (step penalty, pellet collection, win bonus); ghost collision penalty is excluded because it depends on the stochastic ghost transition. When next_states is supplied (e.g. by a caller that already realised the batch transition), the collision penalty is included against those realised draws so the per- particle batch reward agrees with the trajectory-driven single-state path.

Return type:

ndarray

Parameters:

• states (ndarray | Sequence[Any])

• action (int)

• next_states (ndarray | Sequence[Any] | None)

sample_next_state(state, action, n_samples=1)

Sample one or more next states for (state, action).

Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.

Returns:

a single next state of the env’s native type. When n_samples > 1: an array-like of length n_samples (numeric envs return np.ndarray of shape (n_samples, *dim); structured envs return List[T]).

Return type:

Any

Parameters:

• state (ndarray)

• action (int)

• n_samples (int)

sample_next_state_batch(states, action)

Sample one next state per input state, all under the same action.

Used by particle filters: given N current particles and one action, draw N next states (one per particle) in a single vectorized call.

The default implementation falls back to a per-state Python loop delegating to sample_next_state(). Native-backed envs (those whose state-transition kernel exposes batch_sample(states_array)) should override to avoid the loop.

Parameters:

• states (Any) – A sequence (length N) or ndarray of shape (N, *dim) of input particles.

• action (int) – A single action to apply to every particle.

Returns:

np.ndarray of shape (N, *dim). For structured envs (Tiger strings, Pacman tuples): a list of length N.

Return type:

ndarray

sample_observation(next_state, action, n_samples=1)

Sample one or more observations for (next_state, action).

Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.

Returns:

a single observation. When n_samples > 1: an array-like of length n_samples.

Return type:

Any

Parameters:

• next_state (ndarray)

• action (int)

• n_samples (int)

simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)

Estimate the value of state via a native C++ random rollout.

Pre-draws all action indices in NumPy, then delegates the entire trajectory (transition + reward accumulation) to the C++ kernel. This avoids per-step Python frame overhead for the common path.

Parameters:

• state (Any) – Current state ndarray.

• action_sampler (Any) – Object with a sample() method returning a random action; only used to pre-draw action integers.

• max_depth (int) – Maximum rollout depth.

• discount_factor (float) – Per-step discount factor.

• depth (int) – Current depth consumed by the search tree. Defaults to 0.

Return type:

float

Returns:

Discounted cumulative reward along the sampled trajectory.

transition_log_probability(state, action, next_states)

Log-probability of each candidate next state under (state, action).

Returns np.ndarray of shape (N,) where N is the number of candidate next states. Subclasses must implement.

Return type:

ndarray

Parameters:

• state (ndarray)

• action (int)

• next_states (Any)

visualize_path(path, actions, cache_path)

Visualize PacMan path through the maze using sprite-based rendering.

Parameters:

• path (List[ndarray]) – List of state arrays representing the path through the maze.

• actions (List[int]) – List of actions taken at each step.

• cache_path (Path) – Path where the GIF should be saved.

class POMDPPlanners.environments.pacman_pomdp.PacManVectorizedUpdater(maze_size, num_ghosts, num_pellets, state_dim, ghost_aggressiveness, ghost_coordination, ghost_strategies, observation_noise_factor, max_observation_noise, transition_ctor_kwargs, observation_ctor_kwargs, patrol_dir_state)

Bases: VectorizedParticleBeliefUpdater

Vectorized particle belief updater for PacMan POMDP, native-backed.

Performs all-particle transitions and observation log-likelihoods by dispatching to the PacManTransitionCpp and PacManObservationCpp native kernels.

Parameters:

• maze_size (Tuple[int, int])

• num_ghosts (int)

• num_pellets (int)

• state_dim (int)

• ghost_aggressiveness (float)

• ghost_coordination (str)

• ghost_strategies (List[str])

• observation_noise_factor (float)

• max_observation_noise (float)

• transition_ctor_kwargs (Dict[str, Any])

• observation_ctor_kwargs (Dict[str, Any])

• patrol_dir_state (np.ndarray)

maze_size

Grid dimensions (rows, cols).

num_ghosts

Number of ghosts.

num_pellets

Number of initial pellets.

state_dim

Dimensionality of the array state.

ghost_aggressiveness

Softmax temperature for ghost pursuit.

ghost_coordination

Ghost coordination mode.

ghost_strategies

Per-ghost strategy list.

observation_noise_factor

Multiplier for observation noise.

max_observation_noise

Maximum observation noise std.

batch_observation_log_likelihood(next_particles, action, observation)

Compute observation log-likelihoods for all particles at once.

Parameters:

• next_particles (ndarray) – Transitioned particle states of shape (N, d).

• action (ndarray) – Action vector.

• observation (ndarray) – Observed value.

Return type:

ndarray

Returns:

Log-likelihoods of shape (N,).

batch_transition(particles, action)

Transition all particles in a single batched operation.

Parameters:

• particles (ndarray) – Current particle states of shape (N, d).

• action (ndarray) – Action vector.

Return type:

ndarray

Returns:

Next-state particles of shape (N, d).

property config_id: str

Return a deterministic identifier for this updater configuration.

classmethod from_environment(env)

Construct an updater from a PacManPOMDP instance.

Return type:

PacManVectorizedUpdater

Parameters:

env (PacManPOMDP)

POMDPPlanners.environments.pacman_pomdp.create_pacman_belief(env, belief_type=BeliefType.VECTORIZED_PARTICLE, n_particles=200, **kwargs)

Create a ready-to-use belief for the PacMan POMDP.

Parameters:

• env (PacManPOMDP) – PacManPOMDP environment instance.

• belief_type (BeliefType) – Desired belief representation. Defaults to BeliefType.VECTORIZED_PARTICLE.

• n_particles (int) – Number of particles. Defaults to 200.

• **kwargs (Any) – Extra arguments (reserved for future use).

Return type:

Belief

Returns:

A configured Belief object.

Raises:

ValueError – If belief_type is not supported.

Example

>>> import numpy as np
>>> np.random.seed(42)
>>> from POMDPPlanners.environments.pacman_pomdp import PacManPOMDP
>>> env = PacManPOMDP(discount_factor=0.95)
>>> belief = create_pacman_belief(env, n_particles=50)
>>> belief.sample().shape[0] > 0
True

POMDPPlanners.environments.pacman_pomdp.create_simple_maze_pacman(maze_size=7, num_walls=5, num_ghosts=1, seed=None)

Create a simple PacMan instance with random walls and multiple ghosts.

Parameters:

• maze_size (int) – Size of square maze. Defaults to 7.

• num_walls (int) – Number of walls to place randomly. Defaults to 5.

• num_ghosts (int) – Number of ghosts in the game. Defaults to 1.

• seed (Optional[int]) – Random seed. Defaults to None.

Return type:

PacManPOMDP

Returns:

Randomly configured PacMan POMDP with multi-ghost support

Subpackages

• POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_beliefs package
  • PacManVectorizedUpdater
    • PacManVectorizedUpdater.maze_size
    • PacManVectorizedUpdater.num_ghosts
    • PacManVectorizedUpdater.num_pellets
    • PacManVectorizedUpdater.state_dim
    • PacManVectorizedUpdater.ghost_aggressiveness
    • PacManVectorizedUpdater.ghost_coordination
    • PacManVectorizedUpdater.ghost_strategies
    • PacManVectorizedUpdater.observation_noise_factor
    • PacManVectorizedUpdater.max_observation_noise
    • PacManVectorizedUpdater.batch_observation_log_likelihood()
    • PacManVectorizedUpdater.batch_transition()
    • PacManVectorizedUpdater.config_id
    • PacManVectorizedUpdater.from_environment()
  • create_pacman_belief()
  • Submodules
  • POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_beliefs.pacman_belief_factory module
    • create_pacman_belief()
  • POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_beliefs.pacman_grid_utils module
    • precompute_neighbor_table()
    • precompute_neighbor_validity()
    • precompute_valid_cell_mask()
  • POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_beliefs.pacman_vectorized_updater module
    • PacManVectorizedUpdater
      • PacManVectorizedUpdater.maze_size
      • PacManVectorizedUpdater.num_ghosts
      • PacManVectorizedUpdater.num_pellets
      • PacManVectorizedUpdater.state_dim
      • PacManVectorizedUpdater.ghost_aggressiveness
      • PacManVectorizedUpdater.ghost_coordination
      • PacManVectorizedUpdater.ghost_strategies
      • PacManVectorizedUpdater.observation_noise_factor
      • PacManVectorizedUpdater.max_observation_noise
      • PacManVectorizedUpdater.batch_observation_log_likelihood()
      • PacManVectorizedUpdater.batch_transition()
      • PacManVectorizedUpdater.config_id
      • PacManVectorizedUpdater.from_environment()
• POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_utils package
  • Submodules
  • POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_utils.numba_kernels module
    • Conventions
    • compute_reward_batch_kernel()
    • compute_reward_scalar_kernel()
  • POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_utils.pacman_reward_models module
    • BasePacManRewardModel
      • BasePacManRewardModel.compute_reward()
      • BasePacManRewardModel.compute_reward_batch()
    • PacManDistanceDecayedHazardPenaltyRewardModel
      • PacManDistanceDecayedHazardPenaltyRewardModel.compute_reward()
      • PacManDistanceDecayedHazardPenaltyRewardModel.compute_reward_batch()
    • PacManRewardModel
      • PacManRewardModel.compute_reward()
      • PacManRewardModel.compute_reward_batch()
    • PacManZeroMeanHazardShockRewardModel
      • PacManZeroMeanHazardShockRewardModel.compute_reward()
      • PacManZeroMeanHazardShockRewardModel.compute_reward_batch()

Submodules

POMDPPlanners.environments.pacman_pomdp.pacman_pomdp module

Module for PacMan POMDP environment.

This module provides the PacMan POMDP environment implementation inspired by the classic arcade game. The environment features a grid world where PacMan must collect pellets while avoiding ghosts, with partial observability of ghost positions.

The environment involves PacMan navigating a maze with walls, collecting pellets, and avoiding ghosts that move according to stochastic policies. PacMan receives noisy observations about nearby ghost positions. The state is a flat float64 ndarray in the canonical layout [pac_row, pac_col, g0_row, g0_col, ..., pellet_mask[0..P-1], score, terminal]; build states via PacManPOMDP.make_state() and read fields back with get_pacman_pos / get_ghost_positions / get_pellets / get_score / get_terminal.

Classes:

PacManPOMDP: The main POMDP environment implementation

class POMDPPlanners.environments.pacman_pomdp.pacman_pomdp.PacManPOMDP(maze_size=(7, 7), walls=None, initial_pellets=None, initial_pacman_pos=(0, 0), num_ghosts=1, initial_ghost_positions=None, initial_ghost_pos=None, pellet_reward=10.0, ghost_collision_penalty=-100.0, step_penalty=-1.0, win_reward=100.0, ghost_aggressiveness=2.0, ghost_coordination='independent', ghost_strategies=None, observation_noise_factor=0.3, max_observation_noise=1.5, dangerous_areas=None, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, discount_factor=0.95, name='PacManPOMDP', output_dir=None, debug=False, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0)

Bases: DiscreteActionsEnvironment

PacMan POMDP environment inspired by the classic arcade game.

This environment implements a simplified PacMan game where PacMan must collect pellets while avoiding a single ghost. The ghost position is only partially observable through noisy sensor readings.

Parameters:

• maze_size (Tuple[int, int])

• walls (Set[Tuple[int, int]] | None)

• initial_pellets (List[Tuple[int, int]] | None)

• initial_pacman_pos (Tuple[int, int])

• num_ghosts (int)

• initial_ghost_positions (List[Tuple[int, int]] | None)

• initial_ghost_pos (Tuple[int, int] | None)

• pellet_reward (float)

• ghost_collision_penalty (float)

• step_penalty (float)

• win_reward (float)

• ghost_aggressiveness (float)

• ghost_coordination (str)

• ghost_strategies (List[str] | None)

• observation_noise_factor (float)

• max_observation_noise (float)

• dangerous_areas (Set[Tuple[int, int]] | None)

• dangerous_area_radius (float)

• dangerous_area_penalty (float)

• discount_factor (float)

• name (str)

• output_dir (Path | None)

• debug (bool)

• reward_model_type (RewardModelType)

• penalty_decay (float)

maze_size

Grid dimensions as (rows, cols)

walls

Set of wall positions as (row, col) tuples

initial_pellets

List of initial pellet positions

pellet_reward

Reward for collecting a pellet

ghost_collision_penalty

Penalty for collision with ghost

step_penalty

Cost per action

win_reward

Reward for collecting all pellets

ghost_aggressiveness

Temperature parameter for ghost movement policy

observation_noise_factor

Multiplier for observation noise based on distance

max_observation_noise

Maximum noise standard deviation

dangerous_areas

List of (row, col) centers of circular hazard zones

dangerous_area_radius

Radius (in grid cells) defining each hazard zone

dangerous_area_penalty

Penalty subtracted when PacMan ends a step inside a zone

Example

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>>
>>> # Initialize environment
>>> env = PacManPOMDP(maze_size=(7, 7))
>>>
>>> # Get initial state and actions
>>> initial_state = env.initial_state_dist().sample()[0]
>>> actions = env.get_actions()
>>>
>>> # Sample complete step using convenience method
>>> action = actions[0]
>>> next_state, observation, reward = env.sample_next_step(initial_state, action)
>>>
>>> # Check terminal condition
>>> env.is_terminal(initial_state)
False

Example

Construct an env with a circular hazard zone — PacMan is penalised by dangerous_area_penalty whenever its realised next position lies inside the zone, but the zone does not block movement or terminate.

>>> import numpy as np
>>> np.random.seed(0)
>>> env = PacManPOMDP(
...     maze_size=(7, 7),
...     dangerous_areas={(3, 3)},
...     dangerous_area_radius=1.0,
...     dangerous_area_penalty=5.0,
... )
>>> state = env.initial_state_dist().sample()[0]
>>> _ = env.sample_next_step(state, env.get_actions()[0])
>>> env.dangerous_area_penalty
5.0

array_to_observation(arr)

Convert a flat numpy array back to a PacMan observation tuple.

Parameters:

arr (ndarray) – 1-D array of shape (2 * num_ghosts,).

Return type:

Tuple[Tuple[int, int], ...]

Returns:

Observation as tuple of (row, col) tuples.

cache_visualization(history, cache_path)

Cache visualization of episode history.

Parameters:

• history (List[StepData]) – List of StepData objects representing the episode

• cache_path (Path) – Path where the GIF should be saved

Return type:

None

compute_metrics(histories)

Compute environment-specific metrics.

Return type:

List[MetricValue]

Parameters:

histories (List[History])

get_actions()

Get all available actions.

Return type:

List[int]

get_ghost_positions(state)

Return ghost positions as a tuple of (row, col) pairs.

Return type:

Tuple[Tuple[int, int], ...]

Parameters:

state (ndarray)

get_metric_names()

Get names of PacMan POMDP specific metrics.

Return type:

List[str]

Returns:

List containing metric names including standard metrics (win_rate, avg_pellets_collected, avg_episode_length, avg_pacman_closest_ghost_distance, avg_collision_encounters, avg_dangerous_area_steps, avg_all_dangerous_encounters) and dynamically generated per-ghost distance metrics for multi-ghost scenarios (avg_pacman_ghost_0_distance, avg_pacman_ghost_1_distance, etc.). avg_all_dangerous_encounters is the per-step sum of ghost-collision and dangerous-area-step events; a step that is both counts twice.

get_observation_cpp_ctor_kwargs()

Return the kwargs dict passed to PacManObservationCpp.

Return type:

Dict[str, Any]

get_pacman_pos(state)

Return PacMan’s (row, col) position from a state array.

Return type:

Tuple[int, int]

Parameters:

state (ndarray)

get_pellets(state)

Return the tuple of active pellet positions.

Return type:

Tuple[Tuple[int, int], ...]

Parameters:

state (ndarray)

get_score(state)

Return the state’s score as a Python float.

Return type:

float

Parameters:

state (ndarray)

get_terminal(state)

Return whether the state is terminal.

Return type:

bool

Parameters:

state (ndarray)

get_transition_cpp_ctor_kwargs()

Return the cached per-env kwargs dict passed to PacManTransitionCpp.

Return type:

Dict[str, Any]

hash_action(action)

Return a hashable key consistent with action equality.

Used by tree-search planners to index action children of a belief node in O(1). The returned key MUST satisfy:

action_a == action_b   (per env's notion of equality)
==> hash_action(action_a) == hash_action(action_b)

Subclasses with non-hashable actions (e.g. np.ndarray) must override to return a hashable surrogate (tobytes() is the standard choice for ndarray actions, which mirrors the np.array_equal semantics used by the linear-scan fallback).

Parameters:

action (Any) – Action to hash.

Return type:

Hashable

Returns:

A hashable key derived from action.

property initial_ghost_pos: Tuple[int, int]

returns first ghost position.

Type:

Backward compatibility

initial_observation_dist()

Get the initial observation distribution.

Returns a live distribution that draws fresh noisy ghost-position observations from the true initial state on each sample call, instead of the previous Dirac wrapper around a single pre-drawn sample (which collapsed the entire initial-belief observation prior to a point mass).

Return type:

Distribution

initial_state_dist()

Get initial state distribution.

Return type:

DiscreteDistribution

is_equal_observation(observation1, observation2)

Check if two observations are equal.

Return type:

bool

Parameters:

• observation1 (Any)

• observation2 (Any)

is_terminal(state)

Check if state is terminal.

Return type:

bool

Parameters:

state (ndarray)

make_state(*, pacman_pos, ghost_positions, pellets=None, score=0.0, terminal=False)

Build a PacMan state array in the canonical layout.

The array layout is [pac_row, pac_col, g0_row, g0_col, ..., pellet_mask[0..P-1], score, terminal].

Parameters:

• pacman_pos (Tuple[int, int]) – PacMan grid position (row, col).

• ghost_positions (Tuple[Tuple[int, int], ...]) – Per-ghost positions as a tuple of length num_ghosts.

• pellets (Optional[Tuple[Tuple[int, int], ...]]) – Active pellet positions. None means every initial pellet is active (useful for constructing initial states).

• score (float) – Current game score.

• terminal (bool) – Whether the state is terminal.

Return type:

ndarray

Returns:

1-D float64 array of shape (self._state_dim,).

Raises:

ValueError – If any argument has the wrong type or length, or if a pellet position was not registered at env construction.

observation_log_probability(next_state, action, observations)

Log-probability of each candidate observation under (next_state, action).

Returns np.ndarray of shape (N,) where N is the number of candidate observations. Subclasses must implement.

Return type:

ndarray

Parameters:

• next_state (ndarray)

• action (int)

• observations (Any)

observation_log_probability_per_state(next_states, action, observation)

Log-probability of one observation under each candidate next-state.

Used by particle filters: given N candidate next-states and ONE observation, return N log-likelihoods.

The default implementation falls back to a per-state Python loop delegating to observation_log_probability(). Native-backed envs (those whose observation kernel exposes batch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.

Parameters:

• next_states (Any) – A sequence (length N) or ndarray of shape (N, *dim) of candidate next-states.

• action (int) – The action that was executed.

• observation (Any) – A single observation.

Return type:

ndarray

Returns:

ndarray of shape (N,) with log-probabilities or log-PDFs.

observation_to_array(obs)

Convert a PacMan observation tuple to a flat numpy array.

Parameters:

obs (Tuple[Tuple[int, int], ...]) – Observation as tuple of ghost (row, col) positions.

Return type:

ndarray

Returns:

1-D array of shape (2 * num_ghosts,).

reward(state, action, next_state=None)

Calculate immediate reward.

Uses the realised next_state when supplied (e.g. by Environment.sample_next_step()) so the collision penalty and win bonus reflect the same stochastic ghost transition as the trajectory rather than a fresh independent draw. When next_state is None, falls back to sampling one here.

Return type:

float

Parameters:

• state (ndarray)

• action (int)

• next_state (Any)

reward_batch(states, action, next_states=None)

Calculate rewards for a batch of states.

Accepts a 2-D numpy array of shape (N, state_dim) on the fast vectorized path, or a sequence of 1-D state arrays on the fallback per-particle path.

Without next_states, computes deterministic reward components only (step penalty, pellet collection, win bonus); ghost collision penalty is excluded because it depends on the stochastic ghost transition. When next_states is supplied (e.g. by a caller that already realised the batch transition), the collision penalty is included against those realised draws so the per- particle batch reward agrees with the trajectory-driven single-state path.

Return type:

ndarray

Parameters:

• states (ndarray | Sequence[Any])

• action (int)

• next_states (ndarray | Sequence[Any] | None)

sample_next_state(state, action, n_samples=1)

Sample one or more next states for (state, action).

Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.

Returns:

a single next state of the env’s native type. When n_samples > 1: an array-like of length n_samples (numeric envs return np.ndarray of shape (n_samples, *dim); structured envs return List[T]).

Return type:

Any

Parameters:

• state (ndarray)

• action (int)

• n_samples (int)

sample_next_state_batch(states, action)

Sample one next state per input state, all under the same action.

Used by particle filters: given N current particles and one action, draw N next states (one per particle) in a single vectorized call.

The default implementation falls back to a per-state Python loop delegating to sample_next_state(). Native-backed envs (those whose state-transition kernel exposes batch_sample(states_array)) should override to avoid the loop.

Parameters:

• states (Any) – A sequence (length N) or ndarray of shape (N, *dim) of input particles.

• action (int) – A single action to apply to every particle.

Returns:

np.ndarray of shape (N, *dim). For structured envs (Tiger strings, Pacman tuples): a list of length N.

Return type:

ndarray

sample_observation(next_state, action, n_samples=1)

Sample one or more observations for (next_state, action).

Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.

Returns:

a single observation. When n_samples > 1: an array-like of length n_samples.

Return type:

Any

Parameters:

• next_state (ndarray)

• action (int)

• n_samples (int)

simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)

Estimate the value of state via a native C++ random rollout.

Pre-draws all action indices in NumPy, then delegates the entire trajectory (transition + reward accumulation) to the C++ kernel. This avoids per-step Python frame overhead for the common path.

Parameters:

• state (Any) – Current state ndarray.

• action_sampler (Any) – Object with a sample() method returning a random action; only used to pre-draw action integers.

• max_depth (int) – Maximum rollout depth.

• discount_factor (float) – Per-step discount factor.

• depth (int) – Current depth consumed by the search tree. Defaults to 0.

Return type:

float

Returns:

Discounted cumulative reward along the sampled trajectory.

transition_log_probability(state, action, next_states)

Log-probability of each candidate next state under (state, action).

Returns np.ndarray of shape (N,) where N is the number of candidate next states. Subclasses must implement.

Return type:

ndarray

Parameters:

• state (ndarray)

• action (int)

• next_states (Any)

visualize_path(path, actions, cache_path)

Visualize PacMan path through the maze using sprite-based rendering.

Parameters:

• path (List[ndarray]) – List of state arrays representing the path through the maze.

• actions (List[int]) – List of actions taken at each step.

• cache_path (Path) – Path where the GIF should be saved.

class POMDPPlanners.environments.pacman_pomdp.pacman_pomdp.PacManPOMDPMetrics(*values)

Bases: Enum

Metric names for PacMan POMDP environment.

AVG_ALL_DANGEROUS_ENCOUNTERS = 'avg_all_dangerous_encounters'

AVG_COLLISION_ENCOUNTERS = 'avg_collision_encounters'

AVG_DANGEROUS_AREA_STEPS = 'avg_dangerous_area_steps'

AVG_EPISODE_LENGTH = 'avg_episode_length'

AVG_PACMAN_CLOSEST_GHOST_DISTANCE = 'avg_pacman_closest_ghost_distance'

AVG_PELLETS_COLLECTED = 'avg_pellets_collected'

WIN_RATE = 'win_rate'

class POMDPPlanners.environments.pacman_pomdp.pacman_pomdp.RewardModelType(*values)

Bases: Enum

Reward-model variants selectable on PacManPOMDP.

CONSTANT_HAZARD_PENALTY = 'constant_hazard_penalty'

DISTANCE_DECAYED_HAZARD_PENALTY = 'distance_decayed_hazard_penalty'

ZERO_MEAN_HAZARD_SHOCK = 'zero_mean_hazard_shock'

POMDPPlanners.environments.pacman_pomdp.pacman_pomdp.create_simple_maze_pacman(maze_size=7, num_walls=5, num_ghosts=1, seed=None)

Create a simple PacMan instance with random walls and multiple ghosts.

Parameters:

• maze_size (int) – Size of square maze. Defaults to 7.

• num_walls (int) – Number of walls to place randomly. Defaults to 5.

• num_ghosts (int) – Number of ghosts in the game. Defaults to 1.

• seed (Optional[int]) – Random seed. Defaults to None.

Return type:

PacManPOMDP

Returns:

Randomly configured PacMan POMDP with multi-ghost support

POMDPPlanners.environments.pacman_pomdp.pacman_visualizer module

Visualization module for PacMan POMDP environment.

This module provides sprite-based visualization capabilities for PacMan POMDP episodes, rendering animated GIFs of agent behavior and game state.

Classes:

PacManVisualizer: Handles sprite-based rendering and GIF generation

class POMDPPlanners.environments.pacman_pomdp.pacman_visualizer.PacManVisualizer(environment, tile_size=32)

Bases: object

Handles visualization for PacMan POMDP environments.

This class manages sprite loading, frame rendering, and GIF generation for visualizing PacMan POMDP episodes. It renders the maze, PacMan, ghosts, pellets, and game state information.

Parameters:

• environment (PacManPOMDP)

• tile_size (int)

env

Reference to the PacMan POMDP environment

tile_size

Size of each tile in pixels

sprites

Dictionary of loaded sprite images

cache_visualization(history, cache_path)

Cache visualization of episode history.

Parameters:

• history (List[StepData]) – List of StepData objects representing the episode

• cache_path (Path) – Path where the GIF should be saved

Raises:

• TypeError – If history or cache_path have wrong types

• ValueError – If history is empty or cache_path doesn’t end with .gif

Return type:

None

visualize_path(path, actions, cache_path, beliefs=None)

Visualize PacMan path through the maze using sprite-based rendering.

Parameters:

• path (List[ndarray]) – List of state arrays representing the path through the maze.

• actions (List[int]) – List of actions taken at each step.

• cache_path (Path) – Path where the GIF should be saved.

• beliefs (Optional[List[Optional[Belief]]]) – Optional per-frame beliefs. When supplied, each frame overlays a translucent red heatmap over the cells the belief assigns non-zero ghost-occupation probability.

Raises:

TypeError – If cache_path is not a Path object.

Return type:

None