POMDPPlanners.environments.push_pomdp package

Push POMDP Environment Module.

This module provides the Push POMDP environment implementation and related components for robotic manipulation tasks.

Classes:

PushPOMDP: Main POMDP environment for robotic push tasks PushPOMDPVisualizer: Visualization utilities for Push POMDP episodes ContinuousPushPOMDP: Continuous-action Push POMDP environment ContinuousPushPOMDPDiscreteActions: Discrete-action wrapper

class POMDPPlanners.environments.push_pomdp.ContinuousPushPOMDP(discount_factor, grid_size=10, push_threshold=1.0, friction_coefficient=0.3, max_push=2.0, observation_noise=0.1, obstacles=None, obstacle_penalty=-10.0, obstacle_hit_probability=1.0, dangerous_areas=None, dangerous_area_radius=0.5, dangerous_area_penalty=-10.0, dangerous_area_hit_probability=1.0, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0, robot_radius=0.3, state_transition_cov_matrix=array([[0.1, 0.], [0., 0.1]]), initial_state=None, name='ContinuousPushPOMDP', output_dir=None, debug=False, use_queue_logger=False)

Bases: Environment

Continuous-action Push POMDP environment.

A robot (circle) must push an object (point) to a target location on a 2D grid. The robot moves via continuous 2D action vectors with Gaussian noise; obstacles are axis-aligned squares.

State: [robot_x, robot_y, object_x, object_y, target_x, target_y] Actions: 2D numpy vectors Observations: [robot_x, robot_y, noisy_obj_x, noisy_obj_y, target_x, target_y]

Stochasticity:

The obstacle-collision penalty can be applied either deterministically (the default) or stochastically. When obstacle_hit_probability == 1.0 (default), the penalty is applied every time the post-action robot position overlaps an obstacle AABB, matching legacy behavior. When obstacle_hit_probability < 1.0, the penalty is applied only with that probability per reward() / reward_batch() call (one Bernoulli draw per state), producing a heavy-tailed return distribution suitable for benchmarking risk-sensitive planners (e.g. ICVaR-aware MCTS) against expected-value MCTS on the same env. Note that this makes reward(state, action) non- deterministic given a state-action pair, so any external caching that assumes deterministic rewards must be aware of this. transition_log_probability is unaffected. The native C++ rollout applies the Bernoulli obstacle_hit_probability draw internally, so simulate_random_rollout always routes through the native kernel.

Dangerous areas:

dangerous_areas is a separate, additive concept from obstacles. Each entry is a circular region centred at (x, y) with radius dangerous_area_radius. Entering a dangerous area applies dangerous_area_penalty (a negative number, added to reward) but does NOT block movement (unlike obstacles, which act as walls in the continuous variant). The penalty fires when the post-action robot position lies inside any dangerous area; the object position is ignored. At most one dangerous_area_penalty is applied per step even when multiple zones overlap. Like obstacles, the penalty supports a Bernoulli dangerous_area_hit_probability (default 1.0) for risk-sensitive planning. The native C++ rollout applies the Bernoulli draw internally, so all rollouts route through the native kernel regardless of the configured probability.

Example

>>> import numpy as np
>>> np.random.seed(42)
>>>
>>> env = ContinuousPushPOMDP(discount_factor=0.99)
>>>
>>> initial_state = env.initial_state_dist().sample()[0]
>>>
>>> action = np.array([1.0, 0.0])
>>> next_state, observation, reward = env.sample_next_step(initial_state, action)
>>>
>>> env.is_terminal(initial_state)
False

Parameters:

• discount_factor (float)

• grid_size (int)

• push_threshold (float)

• friction_coefficient (float)

• max_push (float)

• observation_noise (float)

• obstacles (List[Tuple[float, float, float]] | None)

• obstacle_penalty (float)

• obstacle_hit_probability (float)

• dangerous_areas (List[Tuple[float, float]] | None)

• dangerous_area_radius (float)

• dangerous_area_penalty (float)

• dangerous_area_hit_probability (float)

• reward_model_type (RewardModelType)

• penalty_decay (float)

• robot_radius (float)

• state_transition_cov_matrix (ndarray)

• initial_state (ndarray | None)

• name (str)

• output_dir (Path | None)

• debug (bool)

• use_queue_logger (bool)

cache_visualization(history, cache_path)

Cache animated visualization of the continuous push episode.

Creates an animated GIF showing the robot pushing the object toward the target, with rectangular obstacles, collision detection, distance indicators, and success feedback.

Parameters:

• history (List[StepData]) – Episode history containing states, actions, and rewards.

• cache_path (Path) – Path where to save the visualization (must end with .gif).

Raises:

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

• TypeError – If cache_path is not a Path object.

Return type:

None

compute_metrics(histories)

Compute environment-specific metrics from episode histories.

This method can be overridden by subclasses to provide custom metric calculations beyond standard return and episode length.

Parameters:

histories (List[History]) – List of episode histories to analyze

Return type:

List[MetricValue]

Returns:

List of computed metrics with confidence intervals

get_metric_names()

Get names of Continuous Push POMDP specific metrics.

Return type:

List[str]

Returns:

List of metric name strings.

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.

initial_observation_dist()

Get the initial observation distribution.

Return type:

Distribution

Returns:

Distribution over initial observations

Note

Subclasses must implement this method to define initial observations.

initial_state_dist()

Get the initial state distribution.

Return type:

Distribution

Returns:

Distribution over initial states

Note

Subclasses must implement this method to define the starting distribution.

is_equal_observation(observation1, observation2)

Check if two observations are equal.

Parameters:

• observation1 (ndarray) – First observation to compare

• observation2 (ndarray) – Second observation to compare

Return type:

bool

Returns:

True if observations are considered equal, False otherwise

Note

Subclasses must implement this method to define observation equality. This is particularly important for discrete observation spaces.

is_terminal(state)

Check if a state is terminal.

Parameters:

state (ndarray) – State to check for terminal condition

Return type:

bool

Returns:

True if the state is terminal, False otherwise

Note

Subclasses must implement this method to define terminal conditions.

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 (ndarray)

• 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 (ndarray) – The action that was executed.

• observation (Any) – A single observation.

Return type:

ndarray

Returns:

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

reward(state, action, next_state=None)

Calculate the immediate reward for a state-action(-next_state) tuple.

next_state is the realised post-transition state when known (e.g. threaded by sample_next_step()), allowing rewards that depend on stochastic transition outcomes to use the same draw as the trajectory instead of resampling. Subclasses whose reward is a pure function of (state, action) may ignore it; subclasses whose reward depends on the realised next state (collision penalties, win bonuses) should consume it when provided and fall back to drawing/computing one when None.

Parameters:

• state (ndarray) – Current state.

• action (ndarray) – Action executed from state.

• next_state (Any) – Realised next state, or None if the caller did not pre-sample one. Defaults to None.

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

reward_batch(states, action, next_states=None)

Calculate rewards for a batch of states given a single action.

Provides a loop-based default that subclasses can override with vectorized numpy implementations for better performance.

Parameters:

• states (Union[ndarray, Sequence[Any]]) – Sequence of states of length N.

• action (ndarray) – Action executed from each state.

• next_states (Union[ndarray, Sequence[Any], None]) – Optional realised next states (length N) threaded through to reward(). Defaults to None.

Return type:

ndarray

Returns:

1-D array of reward values with shape (N,).

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 (ndarray)

• 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 (ndarray) – 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 (ndarray)

• n_samples (int)

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

Random rollout dispatched to native C++ when a fixed action set is available.

Uses the cont_simulate_rollout native kernel when self has an action_to_vector mapping (i.e. the discrete-action subclass). Falls back to the Python base-class loop for pure continuous-action environments where no finite action set exists.

Parameters:

• state (Any) – Current 6-D state [rx, ry, ox, oy, tx, ty].

• action_sampler (Any) – Object with a sample() method; used only for the Python fallback path.

• max_depth (int) – Maximum rollout depth.

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

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

Return type:

float

Returns:

Discounted sum of immediate rewards 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 (ndarray)

• next_states (Any)

class POMDPPlanners.environments.push_pomdp.ContinuousPushPOMDPDiscreteActions(discount_factor, grid_size=10, push_threshold=1.0, friction_coefficient=0.3, max_push=2.0, observation_noise=0.1, obstacles=None, obstacle_penalty=-10.0, obstacle_hit_probability=1.0, dangerous_areas=None, dangerous_area_radius=0.5, dangerous_area_penalty=-10.0, dangerous_area_hit_probability=1.0, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0, robot_radius=0.3, state_transition_cov_matrix=array([[0.1, 0.], [0., 0.1]]), initial_state=None, name='ContinuousPushPOMDPDiscreteActions', output_dir=None, debug=False, use_queue_logger=False)

Bases: ContinuousPushPOMDP, DiscreteActionsEnvironment

Discrete-action wrapper for the Continuous Push POMDP.

Maps string actions ["up", "down", "right", "left"] to unit vectors and delegates to the continuous parent.

Example

>>> import numpy as np
>>> np.random.seed(42)
>>>
>>> env = ContinuousPushPOMDPDiscreteActions(discount_factor=0.99)
>>>
>>> initial_state = env.initial_state_dist().sample()[0]
>>> actions = env.get_actions()
>>>
>>> action = actions[0]
>>> next_state, observation, reward = env.sample_next_step(initial_state, action)
>>>
>>> env.is_terminal(initial_state)
False

Parameters:

• discount_factor (float)

• grid_size (int)

• push_threshold (float)

• friction_coefficient (float)

• max_push (float)

• observation_noise (float)

• obstacles (List[Tuple[float, float, float]] | None)

• obstacle_penalty (float)

• obstacle_hit_probability (float)

• dangerous_areas (List[Tuple[float, float]] | None)

• dangerous_area_radius (float)

• dangerous_area_penalty (float)

• dangerous_area_hit_probability (float)

• reward_model_type (RewardModelType)

• penalty_decay (float)

• robot_radius (float)

• state_transition_cov_matrix (ndarray)

• initial_state (ndarray | None)

• name (str)

• output_dir (Path | None)

• debug (bool)

• use_queue_logger (bool)

get_actions()

Get all possible actions in the discrete action space.

Return type:

List[str]

Returns:

List containing all valid actions that can be executed

Note

Subclasses must implement this method to enumerate all possible actions. This is used by planning algorithms that need to iterate over actions.

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.

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 (Any)

• 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 (Any) – The action that was executed.

• observation (Any) – A single observation.

Return type:

ndarray

Returns:

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

reward(state, action, next_state=None)

Calculate the immediate reward for a state-action(-next_state) tuple.

next_state is the realised post-transition state when known (e.g. threaded by sample_next_step()), allowing rewards that depend on stochastic transition outcomes to use the same draw as the trajectory instead of resampling. Subclasses whose reward is a pure function of (state, action) may ignore it; subclasses whose reward depends on the realised next state (collision penalties, win bonuses) should consume it when provided and fall back to drawing/computing one when None.

Parameters:

• state (ndarray) – Current state.

• action (Any) – Action executed from state.

• next_state (Any) – Realised next state, or None if the caller did not pre-sample one. Defaults to None.

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

reward_batch(states, action, next_states=None)

Calculate rewards for a batch of states given a single action.

Provides a loop-based default that subclasses can override with vectorized numpy implementations for better performance.

Parameters:

• states (Union[ndarray, Sequence[Any]]) – Sequence of states of length N.

• action (Any) – Action executed from each state.

• next_states (Union[ndarray, Sequence[Any], None]) – Optional realised next states (length N) threaded through to reward(). Defaults to None.

Return type:

ndarray

Returns:

1-D array of reward values with shape (N,).

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 (Any)

• 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 (Any) – 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 (Any)

• n_samples (int)

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 (Any)

• next_states (Any)

class POMDPPlanners.environments.push_pomdp.ContinuousPushPOMDPVisualizer(env)

Bases: object

Handles visualization and animation for Continuous Push POMDP environments.

This class encapsulates all visualization logic for Continuous Push POMDP episodes, creating animated GIFs showing robot movement (with circular body), object pushing, rectangular obstacle collisions, and task completion.

Parameters:

env (ContinuousPushPOMDP)

env

Reference to the ContinuousPushPOMDP environment instance.

grid_size

Size of the grid environment.

push_threshold

Distance threshold for robot to push object.

obstacles

Shape (M, 4) AABB array (cx, cy, hx, hy).

robot_radius

Radius of the robot body.

create_visualization(history, cache_path)

Create animated visualization of a Continuous Push POMDP episode.

Creates an animated GIF showing the robot pushing the object toward the target, with rectangular obstacles, collision detection, distance indicators, and success feedback.

Parameters:

• history (List[StepData]) – Episode history containing states, actions, and rewards.

• cache_path (Path) – Path where to save the visualization (must end with .gif).

Raises:

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

• TypeError – If cache_path is not a Path object.

Return type:

None

class POMDPPlanners.environments.push_pomdp.PushPOMDP(discount_factor, grid_size=10, push_threshold=1.0, friction_coefficient=0.3, observation_noise=0.1, obstacles=None, obstacle_radius=0.5, obstacle_penalty=-10.0, obstacle_hit_probability=1.0, dangerous_areas=None, dangerous_area_radius=0.5, dangerous_area_penalty=-10.0, dangerous_area_hit_probability=1.0, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0, initial_state=None, transition_error_prob=0.0, name='PushPOMDP', output_dir=None, debug=False, use_queue_logger=False)

Bases: DiscreteActionsEnvironment

Robotic push task formulated as a POMDP.

This environment simulates a robot that must push an object to a target location on a 2D grid. The robot can move in four directions and pushes objects when close enough, with partial observability through noisy object position measurements.

Problem Structure: - State: [robot_x, robot_y, object_x, object_y, target_x, target_y] (continuous) - Actions: [“up”, “down”, “left”, “right”] (discrete) - Observations: [robot_x, robot_y, noisy_object_x, noisy_object_y, target_x, target_y] - Rewards: -distance_to_target + 100 (when object reaches target) - Termination: Object within 0.5 units of target position

Key Features: - Physics-based pushing with configurable friction - Distance-based pushing threshold - Noisy observations of object position only - Dense reward signal based on object-target distance - Obstacle collision detection with configurable penalties - Obstacles prevent robot and object movement through them

Stochasticity:

The obstacle-collision penalty can be applied either deterministically (the default) or stochastically. When obstacle_hit_probability == 1.0 (default), the penalty is applied every time the robot’s intended next position lies inside an obstacle, matching legacy behavior. When obstacle_hit_probability < 1.0, the penalty is applied only with that probability per reward() / reward_batch() call (one Bernoulli draw per state), producing a heavy-tailed return distribution suitable for benchmarking risk-sensitive planners (e.g. ICVaR-aware MCTS) against expected-value MCTS on the same env. Note that this makes reward(state, action) non- deterministic given a state-action pair, so any external caching that assumes deterministic rewards must be aware of this. transition_log_probability is unaffected; the obstacle still deterministically blocks movement. The native C++ rollout applies the Bernoulli obstacle_hit_probability draw internally, so simulate_random_rollout always routes through the native kernel.

Dangerous areas:

dangerous_areas is a separate, additive concept from obstacles. Each entry is a circular region centred at (x, y) with radius dangerous_area_radius. Entering a dangerous area applies dangerous_area_penalty (a negative number, added to reward — same sign convention as obstacle_penalty) but does NOT block movement. Penalty fires when the robot’s intended next position lies inside any dangerous area; the object position is ignored. At most one dangerous_area_penalty is applied per step even when multiple zones overlap. Like obstacles, the penalty supports a Bernoulli dangerous_area_hit_probability (default 1.0) for risk-sensitive planning. The native C++ rollout applies the Bernoulli draw internally, so all rollouts route through the native kernel regardless of the configured probability.

Example

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>>
>>> # Initialize environment
>>> env = PushPOMDP(discount_factor=0.99)
>>>
>>> # 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

Parameters:

• discount_factor (float)

• grid_size (int)

• push_threshold (float)

• friction_coefficient (float)

• observation_noise (float)

• obstacles (List[Tuple[float, float]] | None)

• obstacle_radius (float)

• obstacle_penalty (float)

• obstacle_hit_probability (float)

• dangerous_areas (List[Tuple[float, float]] | None)

• dangerous_area_radius (float)

• dangerous_area_penalty (float)

• dangerous_area_hit_probability (float)

• reward_model_type (RewardModelType)

• penalty_decay (float)

• initial_state (ndarray | None)

• transition_error_prob (float)

• name (str)

• output_dir (Path | None)

• debug (bool)

• use_queue_logger (bool)

cache_visualization(history, cache_path)

Cache animated visualization of the push episode.

Creates an animated GIF showing the robot pushing the object toward the target, with obstacles, collision detection, distance indicators, and success feedback.

Parameters:

• history (List[StepData]) – Episode history containing states, actions, and rewards

• cache_path (Path) – Path where to save the visualization (must end with .gif)

Raises:

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

• TypeError – If cache_path is not a Path object

Return type:

None

compute_metrics(histories)

Compute environment-specific metrics from episode histories.

This method can be overridden by subclasses to provide custom metric calculations beyond standard return and episode length.

Parameters:

histories (List[History]) – List of episode histories to analyze

Return type:

List[MetricValue]

Returns:

List of computed metrics with confidence intervals

get_actions()

Get all possible actions in the discrete action space.

Return type:

List[str]

Returns:

List containing all valid actions that can be executed

Note

Subclasses must implement this method to enumerate all possible actions. This is used by planning algorithms that need to iterate over actions.

get_metric_names()

Get names of Push POMDP specific metrics.

Return type:

List[str]

Returns:

List containing collision-related metric names

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.

hash_observation(observation)

Return a hashable key consistent with is_equal_observation().

Used by tree-search planners to index belief children by observation in O(1). The returned key MUST satisfy the contract:

is_equal_observation(a, b) implies hash_observation(a) == hash_observation(b)

Parameters:

observation (Any) – Observation to hash.

Returns:

the observation itself when it is already hashable).

Return type:

Hashable

Raises:

NotImplementedError – If the observation is not hashable and the subclass has not provided an override. Subclasses with non-hashable observations (e.g. np.ndarray) MUST override.

initial_observation_dist()

Get the initial observation distribution.

Return type:

Distribution

Returns:

Distribution over initial observations

Note

Subclasses must implement this method to define initial observations.

initial_state_dist()

Get the initial state distribution.

Return type:

Distribution

Returns:

Distribution over initial states

Note

Subclasses must implement this method to define the starting distribution.

is_equal_observation(observation1, observation2)

Check if two observations are equal.

Parameters:

• observation1 (ndarray) – First observation to compare

• observation2 (ndarray) – Second observation to compare

Return type:

bool

Returns:

True if observations are considered equal, False otherwise

Note

Subclasses must implement this method to define observation equality. This is particularly important for discrete observation spaces.

is_terminal(state)

Check if a state is terminal.

Parameters:

state (ndarray) – State to check for terminal condition

Return type:

bool

Returns:

True if the state is terminal, False otherwise

Note

Subclasses must implement this method to define terminal conditions.

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 (str)

• 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 (str) – 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_log_probability_single(next_state, action, observation)

Scalar log-likelihood for one (next_state, observation) pair.

Per-state fast-path used by incremental belief updates (e.g. POMCPOW’s WeightedParticleBeliefStateUpdate.inplace_update()) to skip the per-call numpy setup overhead of the batched observation_log_probability() path on a singleton input.

The default falls back to the batched method with a one-element observations list. Envs with cheap scalar likelihoods (e.g. the 2-D Gaussian on Push or the cached-inverse-cov path on ContinuousLightDark) should override to skip array allocation.

Return type:

float

Parameters:

• next_state (Any)

• action (Any)

• observation (Any)

reward(state, action, next_state=None)

Calculate the immediate reward for a state-action(-next_state) tuple.

next_state is the realised post-transition state when known (e.g. threaded by sample_next_step()), allowing rewards that depend on stochastic transition outcomes to use the same draw as the trajectory instead of resampling. Subclasses whose reward is a pure function of (state, action) may ignore it; subclasses whose reward depends on the realised next state (collision penalties, win bonuses) should consume it when provided and fall back to drawing/computing one when None.

Parameters:

• state (ndarray) – Current state.

• action (str) – Action executed from state.

• next_state (Any) – Realised next state, or None if the caller did not pre-sample one. Defaults to None.

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

reward_batch(states, action, next_states=None)

Calculate rewards for a batch of states given a single action.

When next_states is supplied (e.g. by a caller that has already sampled the realised batch transition), it is used directly; otherwise N next states are drawn here via the cached PushVectorizedUpdater. Per-particle rewards are computed in the C++ push_reward_batch kernel (variant-aware: CONSTANT_HAZARD_PENALTY, ZERO_MEAN_HAZARD_SHOCK, DISTANCE_DECAYED_HAZARD_PENALTY) so the batch cost is a single round-trip into native code.

Return type:

ndarray

Parameters:

• states (Any)

• action (str)

• next_states (Any)

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 (str)

• 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 (str) – 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_next_step(state, action)

Sample a complete state transition step.

This convenience method combines state transition, observation generation, and reward calculation in a single operation.

Parameters:

• state (Any) – Current state

• action (Any) – Action to execute

Returns:

• next_state: Sampled next state

• next_observation: Sampled observation

• reward: Immediate reward

Return type:

Tuple[Any, Any, float]

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 (str)

• n_samples (int)

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

Return type:

float

Parameters:

• state (Any)

• action_sampler (Any)

• max_depth (int)

• discount_factor (float)

• depth (int)

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 (str)

• next_states (Any)

class POMDPPlanners.environments.push_pomdp.PushPOMDPVisualizer(env)

Bases: object

Handles visualization and animation for Push POMDP environments.

This class encapsulates all visualization logic for Push POMDP episodes, creating animated GIFs showing robot movement, object pushing, obstacle collisions, and task completion.

Parameters:

env (PushPOMDP)

env

Reference to the PushPOMDP environment instance

grid_size

Size of the grid environment

push_threshold

Distance threshold for robot to push object

obstacles

List of obstacle positions

obstacle_radius

Radius of obstacles for collision detection

create_visualization(history, cache_path)

Create animated visualization of a Push POMDP episode.

Creates an animated GIF showing the robot pushing the object toward the target, with obstacles, collision detection, distance indicators, and success feedback.

Parameters:

• history (List[StepData]) – Episode history containing states, actions, and rewards

• cache_path (Path) – Path where to save the visualization (must end with .gif)

Raises:

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

• TypeError – If cache_path is not a Path object

Return type:

None

Subpackages

• POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs package
  • ContinuousPushVectorizedUpdater
    • ContinuousPushVectorizedUpdater.obs_dist
    • ContinuousPushVectorizedUpdater.grid_size
    • ContinuousPushVectorizedUpdater.push_threshold
    • ContinuousPushVectorizedUpdater.friction_coefficient
    • ContinuousPushVectorizedUpdater.max_push
    • ContinuousPushVectorizedUpdater.obstacles
    • ContinuousPushVectorizedUpdater.robot_radius
    • ContinuousPushVectorizedUpdater.batch_observation_log_likelihood()
    • ContinuousPushVectorizedUpdater.batch_transition()
    • ContinuousPushVectorizedUpdater.config_id
    • ContinuousPushVectorizedUpdater.from_environment()
  • PushVectorizedUpdater
    • PushVectorizedUpdater.obs_dist
    • PushVectorizedUpdater.grid_size
    • PushVectorizedUpdater.push_threshold
    • PushVectorizedUpdater.friction_coefficient
    • PushVectorizedUpdater.obstacles
    • PushVectorizedUpdater.obstacle_radius
    • PushVectorizedUpdater.transition_error_prob
    • PushVectorizedUpdater.ACTION_NAME_TO_INDEX
    • PushVectorizedUpdater.ACTION_VECTORS
    • PushVectorizedUpdater.batch_observation_log_likelihood()
    • PushVectorizedUpdater.batch_transition()
    • PushVectorizedUpdater.config_id
    • PushVectorizedUpdater.from_environment()
  • create_continuous_push_belief()
  • create_push_belief()
  • Submodules
  • POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs.continuous_push_belief_factory module
    • create_continuous_push_belief()
  • POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs.continuous_push_vectorized_updater module
    • ContinuousPushVectorizedUpdater
      • ContinuousPushVectorizedUpdater.obs_dist
      • ContinuousPushVectorizedUpdater.grid_size
      • ContinuousPushVectorizedUpdater.push_threshold
      • ContinuousPushVectorizedUpdater.friction_coefficient
      • ContinuousPushVectorizedUpdater.max_push
      • ContinuousPushVectorizedUpdater.obstacles
      • ContinuousPushVectorizedUpdater.robot_radius
      • ContinuousPushVectorizedUpdater.batch_observation_log_likelihood()
      • ContinuousPushVectorizedUpdater.batch_transition()
      • ContinuousPushVectorizedUpdater.config_id
      • ContinuousPushVectorizedUpdater.from_environment()
  • POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs.push_belief_factory module
    • create_push_belief()
  • POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs.push_vectorized_updater module
    • PushVectorizedUpdater
      • PushVectorizedUpdater.obs_dist
      • PushVectorizedUpdater.grid_size
      • PushVectorizedUpdater.push_threshold
      • PushVectorizedUpdater.friction_coefficient
      • PushVectorizedUpdater.obstacles
      • PushVectorizedUpdater.obstacle_radius
      • PushVectorizedUpdater.transition_error_prob
      • PushVectorizedUpdater.ACTION_NAME_TO_INDEX
      • PushVectorizedUpdater.ACTION_VECTORS
      • PushVectorizedUpdater.batch_observation_log_likelihood()
      • PushVectorizedUpdater.batch_transition()
      • PushVectorizedUpdater.config_id
      • PushVectorizedUpdater.from_environment()
• POMDPPlanners.environments.push_pomdp.push_pomdp_utils package
  • Submodules
  • POMDPPlanners.environments.push_pomdp.push_pomdp_utils.push_reward_models module
    • BasePushRewardModel
      • BasePushRewardModel.compute_reward()
      • BasePushRewardModel.compute_reward_batch()
    • ContinuousPushDistanceDecayedHazardPenaltyRewardModel
    • ContinuousPushRewardModel
      • ContinuousPushRewardModel.compute_reward()
      • ContinuousPushRewardModel.compute_reward_batch()
    • ContinuousPushZeroMeanHazardShockRewardModel
    • DiscretePushDistanceDecayedHazardPenaltyRewardModel
    • DiscretePushRewardModel
      • DiscretePushRewardModel.compute_reward()
      • DiscretePushRewardModel.compute_reward_batch()
    • DiscretePushZeroMeanHazardShockRewardModel
    • RewardModelType
      • RewardModelType.CONSTANT_HAZARD_PENALTY
      • RewardModelType.DISTANCE_DECAYED_HAZARD_PENALTY
      • RewardModelType.ZERO_MEAN_HAZARD_SHOCK

Submodules

POMDPPlanners.environments.push_pomdp.continuous_push_geometry module

Geometry utilities for the Continuous Push POMDP.

Provides circle-AABB overlap tests, point-in-AABB tests, collision resolution and grid clamping used by the continuous push environment and its vectorized belief updater.

Wall AABBs (obstacles) are stored as rows (cx, cy, hx, hy) where (cx, cy) is the center and (hx, hy) the half-extents. For square obstacles hx == hy.

Functions:

circle_aabb_overlap: Boolean circle-AABB overlap test. point_inside_aabb: Boolean point-inside-AABB test. resolve_circle_wall_collision: Push a circle out of overlapping AABBs. clamp_circle_to_grid: Clamp a circle center so the circle stays in-bounds. clamp_point_to_grid: Clamp a point to [0, grid_size-1]. batch_resolve_circle_wall_collision: Vectorized circle-AABB resolution. batch_clamp_circle_to_grid: Vectorized circle grid clamping. batch_point_inside_aabb: Vectorized point-inside-AABB test. batch_clamp_point_to_grid: Vectorized point grid clamping.

POMDPPlanners.environments.push_pomdp.continuous_push_geometry.batch_clamp_circle_to_grid(positions, radius, grid_size)

Clamp an array of circle centers so circles stay in-bounds.

Parameters:

• positions (ndarray) – Shape (N, 2).

• radius (float) – Circle radius.

• grid_size (float) – Grid dimension.

Return type:

ndarray

Returns:

Shape (N, 2) clamped positions.

POMDPPlanners.environments.push_pomdp.continuous_push_geometry.batch_clamp_point_to_grid(positions, grid_size)

Clamp an array of points to [0, grid_size-1].

Parameters:

• positions (ndarray) – Shape (N, 2).

• grid_size (float) – Grid dimension.

Return type:

ndarray

Returns:

Shape (N, 2) clamped positions.

POMDPPlanners.environments.push_pomdp.continuous_push_geometry.batch_point_inside_aabb(points, walls)

Test whether each point lies inside any AABB.

Parameters:

• points (ndarray) – Shape (N, 2).

• walls (ndarray) – Shape (M, 4) – AABBs.

Return type:

ndarray

Returns:

Shape (N,) boolean array – True where the point is inside at least one AABB.

POMDPPlanners.environments.push_pomdp.continuous_push_geometry.batch_resolve_circle_wall_collision(positions, radius, walls)

Resolve circle-AABB collisions for an array of positions.

Parameters:

• positions (ndarray) – Shape (N, 2).

• radius (float) – Circle radius.

• walls (ndarray) – Shape (M, 4) – wall AABBs.

Return type:

ndarray

Returns:

Shape (N, 2) resolved positions.

POMDPPlanners.environments.push_pomdp.continuous_push_geometry.circle_aabb_overlap(center, radius, wall)

Test whether a circle overlaps an axis-aligned bounding box.

Parameters:

• center (ndarray) – Shape (2,) – circle center (x, y).

• radius (float) – Circle radius.

• wall (ndarray) – Shape (4,) – AABB (cx, cy, hx, hy).

Return type:

bool

Returns:

True if the circle and AABB overlap.

POMDPPlanners.environments.push_pomdp.continuous_push_geometry.clamp_circle_to_grid(pos, radius, grid_size)

Clamp a circle center so the full circle stays within [0, grid_size-1].

Parameters:

• pos (ndarray) – Shape (2,) – circle center.

• radius (float) – Circle radius.

• grid_size (float) – Grid dimension (positions valid in [0, grid_size-1]).

Return type:

ndarray

Returns:

Clamped position as shape (2,) array.

POMDPPlanners.environments.push_pomdp.continuous_push_geometry.clamp_point_to_grid(pos, grid_size)

Clamp a point to [0, grid_size-1].

Parameters:

• pos (ndarray) – Shape (2,) – point position.

• grid_size (float) – Grid dimension.

Return type:

ndarray

Returns:

Clamped position as shape (2,) array.

POMDPPlanners.environments.push_pomdp.continuous_push_geometry.point_inside_aabb(point, wall)

Test whether a point lies inside an axis-aligned bounding box.

Parameters:

• point (ndarray) – Shape (2,) – point (x, y).

• wall (ndarray) – Shape (4,) – AABB (cx, cy, hx, hy).

Return type:

bool

Returns:

True if the point is inside the AABB.

POMDPPlanners.environments.push_pomdp.continuous_push_geometry.resolve_circle_wall_collision(pos, radius, walls)

Push a circular entity out of any overlapping wall AABBs.

For each wall, if the entity circle overlaps the AABB, the entity is pushed along the axis of minimum penetration.

Parameters:

• pos (ndarray) – Shape (2,) – entity center.

• radius (float) – Entity body radius.

• walls (ndarray) – Shape (M, 4) – wall AABBs.

Return type:

ndarray

Returns:

Resolved position as shape (2,) array.

POMDPPlanners.environments.push_pomdp.continuous_push_pomdp module

Continuous Push POMDP Environment Implementation.

This module implements a continuous-action variant of the Push POMDP where the robot moves via 2D action vectors with Gaussian noise, has a configurable radius, and obstacles are axis-aligned squares.

The Continuous Push POMDP features: - Continuous 2D state space: [robot_x, robot_y, object_x, object_y, target_x, target_y] - Continuous action space (2D movement vectors) - Robot modelled as a circle with configurable radius - Object modelled as a point - Square obstacles defined as axis-aligned bounding boxes - Gaussian transition noise on robot movement - Capped push force with friction - Noisy observations of object position

Classes:

ContinuousPushPOMDP: Main environment with continuous actions. ContinuousPushPOMDPDiscreteActions: Discrete action wrapper.

class POMDPPlanners.environments.push_pomdp.continuous_push_pomdp.ContinuousPushPOMDP(discount_factor, grid_size=10, push_threshold=1.0, friction_coefficient=0.3, max_push=2.0, observation_noise=0.1, obstacles=None, obstacle_penalty=-10.0, obstacle_hit_probability=1.0, dangerous_areas=None, dangerous_area_radius=0.5, dangerous_area_penalty=-10.0, dangerous_area_hit_probability=1.0, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0, robot_radius=0.3, state_transition_cov_matrix=array([[0.1, 0.], [0., 0.1]]), initial_state=None, name='ContinuousPushPOMDP', output_dir=None, debug=False, use_queue_logger=False)

Bases: Environment

Continuous-action Push POMDP environment.

A robot (circle) must push an object (point) to a target location on a 2D grid. The robot moves via continuous 2D action vectors with Gaussian noise; obstacles are axis-aligned squares.

State: [robot_x, robot_y, object_x, object_y, target_x, target_y] Actions: 2D numpy vectors Observations: [robot_x, robot_y, noisy_obj_x, noisy_obj_y, target_x, target_y]

Stochasticity:

The obstacle-collision penalty can be applied either deterministically (the default) or stochastically. When obstacle_hit_probability == 1.0 (default), the penalty is applied every time the post-action robot position overlaps an obstacle AABB, matching legacy behavior. When obstacle_hit_probability < 1.0, the penalty is applied only with that probability per reward() / reward_batch() call (one Bernoulli draw per state), producing a heavy-tailed return distribution suitable for benchmarking risk-sensitive planners (e.g. ICVaR-aware MCTS) against expected-value MCTS on the same env. Note that this makes reward(state, action) non- deterministic given a state-action pair, so any external caching that assumes deterministic rewards must be aware of this. transition_log_probability is unaffected. The native C++ rollout applies the Bernoulli obstacle_hit_probability draw internally, so simulate_random_rollout always routes through the native kernel.

Dangerous areas:

dangerous_areas is a separate, additive concept from obstacles. Each entry is a circular region centred at (x, y) with radius dangerous_area_radius. Entering a dangerous area applies dangerous_area_penalty (a negative number, added to reward) but does NOT block movement (unlike obstacles, which act as walls in the continuous variant). The penalty fires when the post-action robot position lies inside any dangerous area; the object position is ignored. At most one dangerous_area_penalty is applied per step even when multiple zones overlap. Like obstacles, the penalty supports a Bernoulli dangerous_area_hit_probability (default 1.0) for risk-sensitive planning. The native C++ rollout applies the Bernoulli draw internally, so all rollouts route through the native kernel regardless of the configured probability.

Example

>>> import numpy as np
>>> np.random.seed(42)
>>>
>>> env = ContinuousPushPOMDP(discount_factor=0.99)
>>>
>>> initial_state = env.initial_state_dist().sample()[0]
>>>
>>> action = np.array([1.0, 0.0])
>>> next_state, observation, reward = env.sample_next_step(initial_state, action)
>>>
>>> env.is_terminal(initial_state)
False

Parameters:

• discount_factor (float)

• grid_size (int)

• push_threshold (float)

• friction_coefficient (float)

• max_push (float)

• observation_noise (float)

• obstacles (List[Tuple[float, float, float]] | None)

• obstacle_penalty (float)

• obstacle_hit_probability (float)

• dangerous_areas (List[Tuple[float, float]] | None)

• dangerous_area_radius (float)

• dangerous_area_penalty (float)

• dangerous_area_hit_probability (float)

• reward_model_type (RewardModelType)

• penalty_decay (float)

• robot_radius (float)

• state_transition_cov_matrix (ndarray)

• initial_state (ndarray | None)

• name (str)

• output_dir (Path | None)

• debug (bool)

• use_queue_logger (bool)

cache_visualization(history, cache_path)

Cache animated visualization of the continuous push episode.

Creates an animated GIF showing the robot pushing the object toward the target, with rectangular obstacles, collision detection, distance indicators, and success feedback.

Parameters:

• history (List[StepData]) – Episode history containing states, actions, and rewards.

• cache_path (Path) – Path where to save the visualization (must end with .gif).

Raises:

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

• TypeError – If cache_path is not a Path object.

Return type:

None

compute_metrics(histories)

Compute environment-specific metrics from episode histories.

This method can be overridden by subclasses to provide custom metric calculations beyond standard return and episode length.

Parameters:

histories (List[History]) – List of episode histories to analyze

Return type:

List[MetricValue]

Returns:

List of computed metrics with confidence intervals

get_metric_names()

Get names of Continuous Push POMDP specific metrics.

Return type:

List[str]

Returns:

List of metric name strings.

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.

initial_observation_dist()

Get the initial observation distribution.

Return type:

Distribution

Returns:

Distribution over initial observations

Note

Subclasses must implement this method to define initial observations.

initial_state_dist()

Get the initial state distribution.

Return type:

Distribution

Returns:

Distribution over initial states

Note

Subclasses must implement this method to define the starting distribution.

is_equal_observation(observation1, observation2)

Check if two observations are equal.

Parameters:

• observation1 (ndarray) – First observation to compare

• observation2 (ndarray) – Second observation to compare

Return type:

bool

Returns:

True if observations are considered equal, False otherwise

Note

Subclasses must implement this method to define observation equality. This is particularly important for discrete observation spaces.

is_terminal(state)

Check if a state is terminal.

Parameters:

state (ndarray) – State to check for terminal condition

Return type:

bool

Returns:

True if the state is terminal, False otherwise

Note

Subclasses must implement this method to define terminal conditions.

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 (ndarray)

• 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 (ndarray) – The action that was executed.

• observation (Any) – A single observation.

Return type:

ndarray

Returns:

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

reward(state, action, next_state=None)

Calculate the immediate reward for a state-action(-next_state) tuple.

next_state is the realised post-transition state when known (e.g. threaded by sample_next_step()), allowing rewards that depend on stochastic transition outcomes to use the same draw as the trajectory instead of resampling. Subclasses whose reward is a pure function of (state, action) may ignore it; subclasses whose reward depends on the realised next state (collision penalties, win bonuses) should consume it when provided and fall back to drawing/computing one when None.

Parameters:

• state (ndarray) – Current state.

• action (ndarray) – Action executed from state.

• next_state (Any) – Realised next state, or None if the caller did not pre-sample one. Defaults to None.

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

reward_batch(states, action, next_states=None)

Calculate rewards for a batch of states given a single action.

Provides a loop-based default that subclasses can override with vectorized numpy implementations for better performance.

Parameters:

• states (Union[ndarray, Sequence[Any]]) – Sequence of states of length N.

• action (ndarray) – Action executed from each state.

• next_states (Union[ndarray, Sequence[Any], None]) – Optional realised next states (length N) threaded through to reward(). Defaults to None.

Return type:

ndarray

Returns:

1-D array of reward values with shape (N,).

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 (ndarray)

• 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 (ndarray) – 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 (ndarray)

• n_samples (int)

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

Random rollout dispatched to native C++ when a fixed action set is available.

Uses the cont_simulate_rollout native kernel when self has an action_to_vector mapping (i.e. the discrete-action subclass). Falls back to the Python base-class loop for pure continuous-action environments where no finite action set exists.

Parameters:

• state (Any) – Current 6-D state [rx, ry, ox, oy, tx, ty].

• action_sampler (Any) – Object with a sample() method; used only for the Python fallback path.

• max_depth (int) – Maximum rollout depth.

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

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

Return type:

float

Returns:

Discounted sum of immediate rewards 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 (ndarray)

• next_states (Any)

class POMDPPlanners.environments.push_pomdp.continuous_push_pomdp.ContinuousPushPOMDPDiscreteActions(discount_factor, grid_size=10, push_threshold=1.0, friction_coefficient=0.3, max_push=2.0, observation_noise=0.1, obstacles=None, obstacle_penalty=-10.0, obstacle_hit_probability=1.0, dangerous_areas=None, dangerous_area_radius=0.5, dangerous_area_penalty=-10.0, dangerous_area_hit_probability=1.0, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0, robot_radius=0.3, state_transition_cov_matrix=array([[0.1, 0.], [0., 0.1]]), initial_state=None, name='ContinuousPushPOMDPDiscreteActions', output_dir=None, debug=False, use_queue_logger=False)

Bases: ContinuousPushPOMDP, DiscreteActionsEnvironment

Discrete-action wrapper for the Continuous Push POMDP.

Maps string actions ["up", "down", "right", "left"] to unit vectors and delegates to the continuous parent.

Example

>>> import numpy as np
>>> np.random.seed(42)
>>>
>>> env = ContinuousPushPOMDPDiscreteActions(discount_factor=0.99)
>>>
>>> initial_state = env.initial_state_dist().sample()[0]
>>> actions = env.get_actions()
>>>
>>> action = actions[0]
>>> next_state, observation, reward = env.sample_next_step(initial_state, action)
>>>
>>> env.is_terminal(initial_state)
False

Parameters:

• discount_factor (float)

• grid_size (int)

• push_threshold (float)

• friction_coefficient (float)

• max_push (float)

• observation_noise (float)

• obstacles (List[Tuple[float, float, float]] | None)

• obstacle_penalty (float)

• obstacle_hit_probability (float)

• dangerous_areas (List[Tuple[float, float]] | None)

• dangerous_area_radius (float)

• dangerous_area_penalty (float)

• dangerous_area_hit_probability (float)

• reward_model_type (RewardModelType)

• penalty_decay (float)

• robot_radius (float)

• state_transition_cov_matrix (ndarray)

• initial_state (ndarray | None)

• name (str)

• output_dir (Path | None)

• debug (bool)

• use_queue_logger (bool)

get_actions()

Get all possible actions in the discrete action space.

Return type:

List[str]

Returns:

List containing all valid actions that can be executed

Note

Subclasses must implement this method to enumerate all possible actions. This is used by planning algorithms that need to iterate over actions.

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.

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 (Any)

• 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 (Any) – The action that was executed.

• observation (Any) – A single observation.

Return type:

ndarray

Returns:

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

reward(state, action, next_state=None)

Calculate the immediate reward for a state-action(-next_state) tuple.

next_state is the realised post-transition state when known (e.g. threaded by sample_next_step()), allowing rewards that depend on stochastic transition outcomes to use the same draw as the trajectory instead of resampling. Subclasses whose reward is a pure function of (state, action) may ignore it; subclasses whose reward depends on the realised next state (collision penalties, win bonuses) should consume it when provided and fall back to drawing/computing one when None.

Parameters:

• state (ndarray) – Current state.

• action (Any) – Action executed from state.

• next_state (Any) – Realised next state, or None if the caller did not pre-sample one. Defaults to None.

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

reward_batch(states, action, next_states=None)

Calculate rewards for a batch of states given a single action.

Provides a loop-based default that subclasses can override with vectorized numpy implementations for better performance.

Parameters:

• states (Union[ndarray, Sequence[Any]]) – Sequence of states of length N.

• action (Any) – Action executed from each state.

• next_states (Union[ndarray, Sequence[Any], None]) – Optional realised next states (length N) threaded through to reward(). Defaults to None.

Return type:

ndarray

Returns:

1-D array of reward values with shape (N,).

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 (Any)

• 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 (Any) – 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 (Any)

• n_samples (int)

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 (Any)

• next_states (Any)

class POMDPPlanners.environments.push_pomdp.continuous_push_pomdp.ContinuousPushPOMDPMetrics(*values)

Bases: Enum

Metric names for Continuous Push POMDP environment.

DANGEROUS_AREA_RATE = 'dangerous_area_rate'

GOAL_REACHING_RATE = 'goal_reaching_rate'

OBJECT_OBSTACLE_COLLISION_RATE = 'object_obstacle_collision_rate'

ROBOT_OBSTACLE_COLLISION_RATE = 'robot_obstacle_collision_rate'

TOTAL_ALL_OBSTACLE_COLLISIONS = 'total_all_obstacle_collisions'

TOTAL_DANGEROUS_AREA_STEPS = 'total_dangerous_area_steps'

TOTAL_OBJECT_OBSTACLE_COLLISIONS = 'total_object_obstacle_collisions'

TOTAL_OBSTACLE_COLLISION_RATE = 'total_obstacle_collision_rate'

TOTAL_ROBOT_OBSTACLE_COLLISIONS = 'total_robot_obstacle_collisions'

POMDPPlanners.environments.push_pomdp.continuous_push_pomdp_visualizer module

Visualization module for Continuous Push POMDP Environment.

This module provides visualization capabilities for Continuous Push POMDP episodes, creating animated GIFs showing robot movement, object pushing, obstacle collisions, and task completion.

Obstacles are axis-aligned bounding boxes rendered as rectangles. The robot is drawn as a circle with its configured radius. Actions are displayed as formatted 2D vectors.

Classes:

ContinuousPushPOMDPVisualizer: Handles all visualization logic for

Continuous Push POMDP

class POMDPPlanners.environments.push_pomdp.continuous_push_pomdp_visualizer.ContinuousPushPOMDPVisualizer(env)

Bases: object

Handles visualization and animation for Continuous Push POMDP environments.

This class encapsulates all visualization logic for Continuous Push POMDP episodes, creating animated GIFs showing robot movement (with circular body), object pushing, rectangular obstacle collisions, and task completion.

Parameters:

env (ContinuousPushPOMDP)

env

Reference to the ContinuousPushPOMDP environment instance.

grid_size

Size of the grid environment.

push_threshold

Distance threshold for robot to push object.

obstacles

Shape (M, 4) AABB array (cx, cy, hx, hy).

robot_radius

Radius of the robot body.

create_visualization(history, cache_path)

Create animated visualization of a Continuous Push POMDP episode.

Creates an animated GIF showing the robot pushing the object toward the target, with rectangular obstacles, collision detection, distance indicators, and success feedback.

Parameters:

• history (List[StepData]) – Episode history containing states, actions, and rewards.

• cache_path (Path) – Path where to save the visualization (must end with .gif).

Raises:

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

• TypeError – If cache_path is not a Path object.

Return type:

None

POMDPPlanners.environments.push_pomdp.push_pomdp module

Push POMDP Environment Implementation.

This module implements a robotic push task as a POMDP, where a robot must push an object to a target location on a 2D grid. The robot can move in four directions and pushes objects when within range, with noisy observations of the object’s position.

The Push POMDP features: - Continuous 2D state space: [robot_x, robot_y, object_x, object_y, target_x, target_y] - Discrete action space: [“up”, “down”, “left”, “right”] - Noisy observations of object position (robot and target positions are known) - Physics-based pushing mechanics with friction - Distance-based rewards encouraging object movement toward target

Key mechanics: - Robot must be within push_threshold distance to move objects - Friction reduces the effectiveness of pushes - Object position observations include Gaussian noise - Episode terminates when object reaches target

Classes:

PushPOMDP: Main push task environment with POMDP formulation

class POMDPPlanners.environments.push_pomdp.push_pomdp.FixedStateDistribution(state)

Bases: Distribution

Deterministic distribution that always returns the same fixed state.

Parameters:

state (ndarray)

sample(n_samples=1)

Sample values from the distribution.

Parameters:

n_samples (int) – Number of samples to return. Defaults to 1.

Return type:

List[Any]

Returns:

List of n_samples independent samples from the distribution

Note

Subclasses must implement this method according to their specific distribution type and parameters.

class POMDPPlanners.environments.push_pomdp.push_pomdp.PushPOMDP(discount_factor, grid_size=10, push_threshold=1.0, friction_coefficient=0.3, observation_noise=0.1, obstacles=None, obstacle_radius=0.5, obstacle_penalty=-10.0, obstacle_hit_probability=1.0, dangerous_areas=None, dangerous_area_radius=0.5, dangerous_area_penalty=-10.0, dangerous_area_hit_probability=1.0, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0, initial_state=None, transition_error_prob=0.0, name='PushPOMDP', output_dir=None, debug=False, use_queue_logger=False)

Bases: DiscreteActionsEnvironment

Robotic push task formulated as a POMDP.

This environment simulates a robot that must push an object to a target location on a 2D grid. The robot can move in four directions and pushes objects when close enough, with partial observability through noisy object position measurements.

Problem Structure: - State: [robot_x, robot_y, object_x, object_y, target_x, target_y] (continuous) - Actions: [“up”, “down”, “left”, “right”] (discrete) - Observations: [robot_x, robot_y, noisy_object_x, noisy_object_y, target_x, target_y] - Rewards: -distance_to_target + 100 (when object reaches target) - Termination: Object within 0.5 units of target position

Key Features: - Physics-based pushing with configurable friction - Distance-based pushing threshold - Noisy observations of object position only - Dense reward signal based on object-target distance - Obstacle collision detection with configurable penalties - Obstacles prevent robot and object movement through them

Stochasticity:

The obstacle-collision penalty can be applied either deterministically (the default) or stochastically. When obstacle_hit_probability == 1.0 (default), the penalty is applied every time the robot’s intended next position lies inside an obstacle, matching legacy behavior. When obstacle_hit_probability < 1.0, the penalty is applied only with that probability per reward() / reward_batch() call (one Bernoulli draw per state), producing a heavy-tailed return distribution suitable for benchmarking risk-sensitive planners (e.g. ICVaR-aware MCTS) against expected-value MCTS on the same env. Note that this makes reward(state, action) non- deterministic given a state-action pair, so any external caching that assumes deterministic rewards must be aware of this. transition_log_probability is unaffected; the obstacle still deterministically blocks movement. The native C++ rollout applies the Bernoulli obstacle_hit_probability draw internally, so simulate_random_rollout always routes through the native kernel.

Dangerous areas:

dangerous_areas is a separate, additive concept from obstacles. Each entry is a circular region centred at (x, y) with radius dangerous_area_radius. Entering a dangerous area applies dangerous_area_penalty (a negative number, added to reward — same sign convention as obstacle_penalty) but does NOT block movement. Penalty fires when the robot’s intended next position lies inside any dangerous area; the object position is ignored. At most one dangerous_area_penalty is applied per step even when multiple zones overlap. Like obstacles, the penalty supports a Bernoulli dangerous_area_hit_probability (default 1.0) for risk-sensitive planning. The native C++ rollout applies the Bernoulli draw internally, so all rollouts route through the native kernel regardless of the configured probability.

Example

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>>
>>> # Initialize environment
>>> env = PushPOMDP(discount_factor=0.99)
>>>
>>> # 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

Parameters:

• discount_factor (float)

• grid_size (int)

• push_threshold (float)

• friction_coefficient (float)

• observation_noise (float)

• obstacles (List[Tuple[float, float]] | None)

• obstacle_radius (float)

• obstacle_penalty (float)

• obstacle_hit_probability (float)

• dangerous_areas (List[Tuple[float, float]] | None)

• dangerous_area_radius (float)

• dangerous_area_penalty (float)

• dangerous_area_hit_probability (float)

• reward_model_type (RewardModelType)

• penalty_decay (float)

• initial_state (ndarray | None)

• transition_error_prob (float)

• name (str)

• output_dir (Path | None)

• debug (bool)

• use_queue_logger (bool)

cache_visualization(history, cache_path)

Cache animated visualization of the push episode.

Creates an animated GIF showing the robot pushing the object toward the target, with obstacles, collision detection, distance indicators, and success feedback.

Parameters:

• history (List[StepData]) – Episode history containing states, actions, and rewards

• cache_path (Path) – Path where to save the visualization (must end with .gif)

Raises:

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

• TypeError – If cache_path is not a Path object

Return type:

None

compute_metrics(histories)

Compute environment-specific metrics from episode histories.

This method can be overridden by subclasses to provide custom metric calculations beyond standard return and episode length.

Parameters:

histories (List[History]) – List of episode histories to analyze

Return type:

List[MetricValue]

Returns:

List of computed metrics with confidence intervals

get_actions()

Get all possible actions in the discrete action space.

Return type:

List[str]

Returns:

List containing all valid actions that can be executed

Note

Subclasses must implement this method to enumerate all possible actions. This is used by planning algorithms that need to iterate over actions.

get_metric_names()

Get names of Push POMDP specific metrics.

Return type:

List[str]

Returns:

List containing collision-related metric names

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.

hash_observation(observation)

Return a hashable key consistent with is_equal_observation().

Used by tree-search planners to index belief children by observation in O(1). The returned key MUST satisfy the contract:

is_equal_observation(a, b) implies hash_observation(a) == hash_observation(b)

Parameters:

observation (Any) – Observation to hash.

Returns:

the observation itself when it is already hashable).

Return type:

Hashable

Raises:

NotImplementedError – If the observation is not hashable and the subclass has not provided an override. Subclasses with non-hashable observations (e.g. np.ndarray) MUST override.

initial_observation_dist()

Get the initial observation distribution.

Return type:

Distribution

Returns:

Distribution over initial observations

Note

Subclasses must implement this method to define initial observations.

initial_state_dist()

Get the initial state distribution.

Return type:

Distribution

Returns:

Distribution over initial states

Note

Subclasses must implement this method to define the starting distribution.

is_equal_observation(observation1, observation2)

Check if two observations are equal.

Parameters:

• observation1 (ndarray) – First observation to compare

• observation2 (ndarray) – Second observation to compare

Return type:

bool

Returns:

True if observations are considered equal, False otherwise

Note

Subclasses must implement this method to define observation equality. This is particularly important for discrete observation spaces.

is_terminal(state)

Check if a state is terminal.

Parameters:

state (ndarray) – State to check for terminal condition

Return type:

bool

Returns:

True if the state is terminal, False otherwise

Note

Subclasses must implement this method to define terminal conditions.

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 (str)

• 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 (str) – 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_log_probability_single(next_state, action, observation)

Scalar log-likelihood for one (next_state, observation) pair.

Per-state fast-path used by incremental belief updates (e.g. POMCPOW’s WeightedParticleBeliefStateUpdate.inplace_update()) to skip the per-call numpy setup overhead of the batched observation_log_probability() path on a singleton input.

The default falls back to the batched method with a one-element observations list. Envs with cheap scalar likelihoods (e.g. the 2-D Gaussian on Push or the cached-inverse-cov path on ContinuousLightDark) should override to skip array allocation.

Return type:

float

Parameters:

• next_state (Any)

• action (Any)

• observation (Any)

reward(state, action, next_state=None)

Calculate the immediate reward for a state-action(-next_state) tuple.

next_state is the realised post-transition state when known (e.g. threaded by sample_next_step()), allowing rewards that depend on stochastic transition outcomes to use the same draw as the trajectory instead of resampling. Subclasses whose reward is a pure function of (state, action) may ignore it; subclasses whose reward depends on the realised next state (collision penalties, win bonuses) should consume it when provided and fall back to drawing/computing one when None.

Parameters:

• state (ndarray) – Current state.

• action (str) – Action executed from state.

• next_state (Any) – Realised next state, or None if the caller did not pre-sample one. Defaults to None.

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

reward_batch(states, action, next_states=None)

Calculate rewards for a batch of states given a single action.

When next_states is supplied (e.g. by a caller that has already sampled the realised batch transition), it is used directly; otherwise N next states are drawn here via the cached PushVectorizedUpdater. Per-particle rewards are computed in the C++ push_reward_batch kernel (variant-aware: CONSTANT_HAZARD_PENALTY, ZERO_MEAN_HAZARD_SHOCK, DISTANCE_DECAYED_HAZARD_PENALTY) so the batch cost is a single round-trip into native code.

Return type:

ndarray

Parameters:

• states (Any)

• action (str)

• next_states (Any)

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 (str)

• 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 (str) – 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_next_step(state, action)

Sample a complete state transition step.

This convenience method combines state transition, observation generation, and reward calculation in a single operation.

Parameters:

• state (Any) – Current state

• action (Any) – Action to execute

Returns:

• next_state: Sampled next state

• next_observation: Sampled observation

• reward: Immediate reward

Return type:

Tuple[Any, Any, float]

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 (str)

• n_samples (int)

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

Return type:

float

Parameters:

• state (Any)

• action_sampler (Any)

• max_depth (int)

• discount_factor (float)

• depth (int)

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 (str)

• next_states (Any)

class POMDPPlanners.environments.push_pomdp.push_pomdp.PushPOMDPMetrics(*values)

Bases: Enum

Metric names for Push POMDP environment.

DANGEROUS_AREA_RATE = 'dangerous_area_rate'

GOAL_REACHING_RATE = 'goal_reaching_rate'

OBJECT_OBSTACLE_COLLISION_RATE = 'object_obstacle_collision_rate'

ROBOT_OBSTACLE_COLLISION_RATE = 'robot_obstacle_collision_rate'

TOTAL_ALL_OBSTACLE_COLLISIONS = 'total_all_obstacle_collisions'

TOTAL_DANGEROUS_AREA_STEPS = 'total_dangerous_area_steps'

TOTAL_OBJECT_OBSTACLE_COLLISIONS = 'total_object_obstacle_collisions'

TOTAL_OBSTACLE_COLLISION_RATE = 'total_obstacle_collision_rate'

TOTAL_ROBOT_OBSTACLE_COLLISIONS = 'total_robot_obstacle_collisions'

class POMDPPlanners.environments.push_pomdp.push_pomdp.RandomInitialStateDistribution(grid_size, target_pos, obstacles, obstacle_radius, parent)

Bases: Distribution

Random initial state distribution for Push POMDP.

Parameters:

• grid_size (int)

• target_pos (ndarray)

• obstacles (List[Tuple[float, float]])

• obstacle_radius (float)

• parent (PushPOMDP)

sample(n_samples=1)

Sample values from the distribution.

Parameters:

n_samples (int) – Number of samples to return. Defaults to 1.

Return type:

List[Any]

Returns:

List of n_samples independent samples from the distribution

Note

Subclasses must implement this method according to their specific distribution type and parameters.

POMDPPlanners.environments.push_pomdp.push_pomdp_visualizer module

Visualization module for Push POMDP Environment.

This module provides visualization capabilities for Push POMDP episodes, creating animated GIFs showing robot movement, object pushing, obstacle collisions, and task completion.

Classes:

PushPOMDPVisualizer: Handles all visualization logic for Push POMDP

class POMDPPlanners.environments.push_pomdp.push_pomdp_visualizer.PushPOMDPVisualizer(env)

Bases: object

Handles visualization and animation for Push POMDP environments.

This class encapsulates all visualization logic for Push POMDP episodes, creating animated GIFs showing robot movement, object pushing, obstacle collisions, and task completion.

Parameters:

env (PushPOMDP)

env

Reference to the PushPOMDP environment instance

grid_size

Size of the grid environment

push_threshold

Distance threshold for robot to push object

obstacles

List of obstacle positions

obstacle_radius

Radius of obstacles for collision detection

create_visualization(history, cache_path)

Create animated visualization of a Push POMDP episode.

Creates an animated GIF showing the robot pushing the object toward the target, with obstacles, collision detection, distance indicators, and success feedback.

Parameters:

• history (List[StepData]) – Episode history containing states, actions, and rewards

• cache_path (Path) – Path where to save the visualization (must end with .gif)

Raises:

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

• TypeError – If cache_path is not a Path object

Return type:

None