POMDPPlanners.environments package
POMDP Environment Implementations.
This package contains concrete implementations of various POMDP environments used for testing and benchmarking planning algorithms. Each environment implements the core Environment interface with specific state spaces, action spaces, observation models, and reward functions.
- Available Environments:
TigerPOMDP: Classic tiger problem with discrete states and observations CartPolePOMDP: Pole balancing task with continuous states, discrete actions MountainCarPOMDP: Car climbing hill task with continuous state space PushPOMDP: Object manipulation task with spatial reasoning SafeAntVelocityPOMDP: Safety-constrained ant navigation SanityPOMDP: Simple test environment for debugging DiscreteLightDarkPOMDP: Grid-based light-dark navigation ContinuousLightDarkPOMDP: Continuous light-dark navigation problem LaserTagPOMDP: Pursuit-evasion problem with robot tagging opponent RockSamplePOMDP: Rock sampling problem with sensor-based rock quality evaluation
- Factory Functions:
get_environment: Create environment instances by name with parameters
- class POMDPPlanners.environments.CartPolePOMDP(discount_factor, noise_cov, state_transition_cov=None, name='CartPolePOMDP', output_dir=None, debug=False, use_queue_logger=False)[source]
Bases:
DiscreteActionsEnvironmentCartPole balancing task formulated as a POMDP.
This environment simulates the classic cart-pole balancing problem where an agent must apply left or right forces to keep a pole balanced on a moving cart. The challenge comes from noisy observations of the cart-pole state.
Problem Structure: - State: [cart_position, cart_velocity, pole_angle, pole_velocity] (continuous) - Actions: [left_force, right_force] (discrete) - Observations: Noisy state measurements (continuous) - Rewards: +1.0 per time step alive, 0.0 when terminated - Termination: Pole falls beyond angle threshold or cart moves too far
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> noise_cov = np.diag([0.1, 0.1, 0.1, 0.1]) >>> env = CartPolePOMDP(discount_factor=0.99, noise_cov=noise_cov) >>> >>> # 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:
- DEFAULT_STATE_TRANSITION_COV = array([[1.0e-04, 0.0e+00, 0.0e+00, 0.0e+00], [0.0e+00, 1.0e-04, 0.0e+00, 0.0e+00], [0.0e+00, 0.0e+00, 2.5e-05, 0.0e+00], [0.0e+00, 0.0e+00, 0.0e+00, 1.0e-04]])
- compute_metrics(histories)[source]
Compute CartPole POMDP specific metrics from simulation histories.
- Parameters:
- Return type:
- Returns:
List of MetricValue objects containing the computed metrics
- get_actions()[source]
Get all possible actions in the discrete action space.
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)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- initial_observation_dist()[source]
Get the initial observation distribution.
- Return type:
- Returns:
Distribution over initial observations
Note
Subclasses must implement this method to define initial observations.
- initial_state_dist()[source]
Get the initial state distribution.
- Return type:
- Returns:
Distribution over initial states
Note
Subclasses must implement this method to define the starting distribution.
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
- Parameters:
- Return type:
- 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)[source]
Check if a state is terminal.
- Parameters:
state (
ndarray) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
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:
- Return type:
- Returns:
1-D array of reward values with shape
(N,).
- sample_next_state(state, action, n_samples=1)[source]
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 lengthn_samples(numeric envs returnnp.ndarrayof shape(n_samples, *dim); structured envs returnList[T]).- Return type:
- Parameters:
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)[source]
Random rollout via native C++.
- Parameters:
state (
Any) – Current 4-D cart-pole state[x, x_dot, theta, theta_dot].action_sampler (
Any) – Object with asample()method (used only on 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:
- Returns:
Discounted sum of immediate rewards along the sampled trajectory.
- class POMDPPlanners.environments.ContinuousLaserTagPOMDP(discount_factor, name='ContinuousLaserTagPOMDP', grid_size=(11.0, 7.0), walls=None, robot_radius=0.3, opponent_radius=0.3, tag_radius=0.5, tag_reward=10.0, tag_penalty=10.0, step_cost=1.0, measurement_noise=1.0, robot_transition_cov_matrix=array([[0.1, 0.], [0., 0.1]]), opponent_transition_cov_matrix=array([[0.05, 0.], [0., 0.05]]), evasion_speed=0.6, dangerous_areas=None, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, dangerous_area_hit_probability=1.0, output_dir=None, debug=False, use_queue_logger=False, initial_state=None, opponent_policy=OpponentPolicy.EVADE)[source]
Bases:
EnvironmentContinuous LaserTag POMDP with continuous
[dx, dy, tag_flag]actions.A pursuit-evasion problem in continuous 2-D space where a robot must navigate to tag an opponent. The robot receives noisy 8-direction laser range observations.
- Stochasticity:
The dangerous-area penalty can be applied either deterministically (the default) or stochastically. When
dangerous_area_hit_probability == 1.0(default), the kernel’s deterministic deduction is preserved verbatim, matching legacy behavior. Whendangerous_area_hit_probability < 1.0, the accumulated dangerous-area deduction is applied to the reward only with that probability perreward()call, 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 makesreward(state, action)non-deterministic given a state-action pair, so any external caching that assumes deterministic rewards must be aware of this.transition_log_probabilityis unaffected.
Example
>>> import numpy as np >>> np.random.seed(42) >>> >>> # Initialize environment >>> env = ContinuousLaserTagPOMDP(discount_factor=0.95) >>> >>> # Get initial state >>> initial_state = env.initial_state_dist().sample()[0] >>> >>> # Sample complete step >>> action = np.array([1.0, 0.0, 0.0]) >>> next_state, observation, reward = env.sample_next_step(initial_state, action) >>> >>> # Check terminal condition >>> env.is_terminal(initial_state) False
Example
Risk-sensitive evaluation – a 10%-tail-risk environment suitable for benchmarking ICVaR-aware planners against expected-value MCTS:
>>> env = ContinuousLaserTagPOMDP( ... discount_factor=0.95, ... dangerous_area_penalty=150.0, ... dangerous_area_hit_probability=0.1, ... )
- Parameters:
discount_factor (float)
name (str)
robot_radius (float)
opponent_radius (float)
tag_radius (float)
tag_reward (float)
tag_penalty (float)
step_cost (float)
measurement_noise (float)
robot_transition_cov_matrix (np.ndarray)
opponent_transition_cov_matrix (np.ndarray)
evasion_speed (float)
dangerous_area_radius (float)
dangerous_area_penalty (float)
dangerous_area_hit_probability (float)
output_dir (Optional[Path])
debug (bool)
use_queue_logger (bool)
initial_state (Optional[np.ndarray])
opponent_policy (OpponentPolicy)
- cache_visualization(history, cache_path)[source]
Cache visualization data for an episode history.
This method can be overridden by subclasses to provide environment-specific visualization caching capabilities.
- compute_metrics(histories)[source]
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.
- get_metric_names()[source]
Get names of environment-specific metrics.
This method returns the names of custom metrics that this environment computes in the compute_metrics() method. It enables users to discover what metrics are available for hyperparameter optimization.
- Return type:
- Returns:
List of metric names that this environment produces. Default implementation returns empty list for environments without custom metrics.
Note
Subclasses that override compute_metrics() should also override this method to return the names of metrics they produce. Use an Enum to ensure consistency between the names returned here and the names used in compute_metrics().
- hash_action(action)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- hash_observation(observation)[source]
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:
- 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()[source]
Get the initial observation distribution.
- Return type:
- Returns:
Distribution over initial observations
Note
Subclasses must implement this method to define initial observations.
- initial_state_dist()[source]
Get the initial state distribution.
- Return type:
- Returns:
Distribution over initial states
Note
Subclasses must implement this method to define the starting distribution.
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
- Parameters:
- Return type:
- 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)[source]
Check if a state is terminal.
- Parameters:
state (
ndarray) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- observation_log_probability_single(next_state, action, observation)[source]
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 batchedobservation_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.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
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:
- Return type:
- Returns:
1-D array of reward values with shape
(N,).
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)[source]
Random rollout dispatched to native C++ via
cont_simulate_rollout.Pre-samples actions from
action_sampler, packs them into a(N, 3)buffer, and runs the full discounted-return loop inside C++. Results are numerically identical to theEnvironment.simulate_random_rollout()Python fallback.When
dangerous_area_hit_probability < 1.0, falls back to the Python rollout: the native kernel applies the dangerous-area penalty deterministically per step, which contradicts the stochastic semantics; routing through Pythonreward()keeps the per-step Bernoulli intact.Also falls back when
dangerous_areasis non-empty: the C++cont_simulate_rolloutkernel scores the danger penalty against the pre-transition robot position, while the Pythonreward()path (post-fix) consumes the realised post-transition position. Until the C++ kernel is rebuilt this is the only correctness-preserving path for configs with danger areas.
- class POMDPPlanners.environments.ContinuousLaserTagPOMDPDiscreteActions(discount_factor, name='ContinuousLaserTagPOMDPDiscreteActions', grid_size=(11.0, 7.0), walls=None, robot_radius=0.3, opponent_radius=0.3, tag_radius=0.5, tag_reward=10.0, tag_penalty=10.0, step_cost=1.0, measurement_noise=1.0, robot_transition_cov_matrix=array([[0.1, 0.], [0., 0.1]]), opponent_transition_cov_matrix=array([[0.05, 0.], [0., 0.05]]), evasion_speed=0.6, dangerous_areas=None, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, dangerous_area_hit_probability=1.0, output_dir=None, debug=False, use_queue_logger=False, initial_state=None, opponent_policy=OpponentPolicy.EVADE)[source]
Bases:
ContinuousLaserTagPOMDP,DiscreteActionsEnvironmentContinuous LaserTag POMDP with discrete string actions.
Actions:
"up","down","right","left","tag".Example
>>> import numpy as np >>> np.random.seed(42) >>> >>> env = ContinuousLaserTagPOMDPDiscreteActions(discount_factor=0.95) >>> >>> 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)
name (str)
robot_radius (float)
opponent_radius (float)
tag_radius (float)
tag_reward (float)
tag_penalty (float)
step_cost (float)
measurement_noise (float)
robot_transition_cov_matrix (np.ndarray)
opponent_transition_cov_matrix (np.ndarray)
evasion_speed (float)
dangerous_area_radius (float)
dangerous_area_penalty (float)
dangerous_area_hit_probability (float)
output_dir (Optional[Path])
debug (bool)
use_queue_logger (bool)
initial_state (Optional[np.ndarray])
opponent_policy (OpponentPolicy)
- get_actions()[source]
Get all possible actions in the discrete action space.
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)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- observation_log_probability(next_state, action, observations)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- observation_log_probability_single(next_state, action, observation)[source]
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 batchedobservation_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.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
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:
- Return type:
- Returns:
1-D array of reward values with shape
(N,).
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)[source]
Random rollout dispatched to native C++ via
cont_simulate_rollout.Pre-samples actions from
action_sampler, packs them into a(N, 3)buffer, and runs the full discounted-return loop inside C++. Results are numerically identical to theEnvironment.simulate_random_rollout()Python fallback.When
dangerous_area_hit_probability < 1.0, falls back to the Python rollout: the native kernel applies the dangerous-area penalty deterministically per step, which contradicts the stochastic semantics; routing through Pythonreward()keeps the per-step Bernoulli intact.Also falls back when
dangerous_areasis non-empty: the C++cont_simulate_rolloutkernel scores the danger penalty against the pre-transition robot position, while the Pythonreward()path (post-fix) consumes the realised post-transition position. Until the C++ kernel is rebuilt this is the only correctness-preserving path for configs with danger areas.
- class POMDPPlanners.environments.ContinuousLightDarkPOMDP(discount_factor, name='ContinuousLightDarkPOMDP', state_transition_cov_matrix=array([[0.05, 0.], [0., 0.05]]), observation_cov_matrix=array([[0.05, 0.], [0., 0.05]]), beacons=[(0, 0), (0, 5), (0, 10), (5, 0), (5, 5), (5, 10), (10, 0), (10, 5), (10, 10)], goal_state=array([10, 5]), start_state=array([0, 5]), obstacles=[(3, 7), (5, 5)], obstacle_hit_probability=0.2, obstacle_reward=-10.0, goal_reward=10.0, fuel_cost=2.0, grid_size=11, goal_state_radius=1.5, beacon_radius=1.0, obstacle_radius=1.5, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, observation_model_type=ObservationModelType.NORMAL_NOISE, penalty_decay=1.0, is_obstacle_hit_terminal=True)[source]
Bases:
BaseLightDarkPOMDPContinuous Light-Dark POMDP environment with continuous actions.
This environment extends the base Light-Dark problem to continuous 2D space with continuous action vectors. The agent navigates toward a goal while dealing with position-dependent observation noise and optional obstacles.
Key features: - Continuous 2D state and action spaces - Light beacons reduce observation noise when nearby - Multiple observation models available (normal noise, normal noise with no observation in dark) - Multiple reward models available (standard, decaying hit probability, high-variance states) - Optional obstacles with configurable hit penalties - Terminal conditions for goal reaching, obstacle hits, and boundary violations
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = ContinuousLightDarkPOMDP( ... discount_factor=0.95, ... goal_state=np.array([10, 5]), ... start_state=np.array([0, 5]) ... ) >>> >>> # Get initial state >>> initial_state = env.initial_state_dist().sample()[0] >>> >>> # Sample complete step (action must be provided based on environment type) >>> action = np.array([1.0, 0.0]) # Move right >>> next_state, observation, reward = env.sample_next_step(initial_state, action) >>> >>> # Check terminal condition >>> env.is_terminal(initial_state) False
- Parameters:
discount_factor (float)
name (str)
state_transition_cov_matrix (ndarray)
observation_cov_matrix (ndarray)
goal_state (ndarray)
start_state (ndarray)
obstacle_hit_probability (float)
obstacle_reward (float)
goal_reward (float)
fuel_cost (float)
grid_size (int)
goal_state_radius (float)
beacon_radius (float)
obstacle_radius (float)
reward_model_type (RewardModelType)
observation_model_type (ObservationModelType)
penalty_decay (float)
is_obstacle_hit_terminal (bool)
- compute_metrics(histories)[source]
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.
- hash_action(action)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- is_terminal(state)[source]
Check if a state is terminal.
- Parameters:
state (
ndarray) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- observation_log_probability_single(next_state, action, observation)[source]
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 batchedobservation_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.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
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:
- Return type:
- Returns:
1-D array of reward values with shape
(N,).
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- class POMDPPlanners.environments.ContinuousLightDarkPOMDPDiscreteActions(discount_factor, state_transition_cov_matrix=array([[1., 0.], [0., 1.]]), observation_cov_matrix=array([[1., 0.], [0., 1.]]), obstacle_hit_probability=0.2, obstacle_reward=-10.0, goal_reward=10.0, fuel_cost=2.0, grid_size=11, goal_state_radius=1.5, beacon_radius=1.0, obstacle_radius=1.5, name='ContinuousLightDarkPOMDPDiscreteActions', beacons=[(0, 0), (0, 5), (0, 10), (5, 0), (5, 5), (5, 10), (10, 0), (10, 5), (10, 10)], goal_state=array([10, 5]), start_state=array([0, 5]), obstacles=[(3, 7), (5, 5)], reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, observation_model_type=ObservationModelType.NORMAL_NOISE, penalty_decay=1.0, is_obstacle_hit_terminal=True)[source]
Bases:
ContinuousLightDarkPOMDP,DiscreteActionsEnvironmentContinuous Light-Dark POMDP environment with discrete actions.
This variant of the Continuous Light-Dark POMDP uses discrete directional actions (up, down, left, right) instead of continuous action vectors. The continuous state space and observation model are preserved.
Actions are mapped to unit vectors: - “up”: [0, 1] - “down”: [0, -1] - “right”: [1, 0] - “left”: [-1, 0]
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = ContinuousLightDarkPOMDPDiscreteActions( ... discount_factor=0.95, ... goal_state=np.array([10, 5]), ... start_state=np.array([0, 5]) ... ) >>> >>> # 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)
state_transition_cov_matrix (ndarray)
observation_cov_matrix (ndarray)
obstacle_hit_probability (float)
obstacle_reward (float)
goal_reward (float)
fuel_cost (float)
grid_size (int)
goal_state_radius (float)
beacon_radius (float)
obstacle_radius (float)
name (str)
goal_state (ndarray)
start_state (ndarray)
reward_model_type (RewardModelType)
observation_model_type (ObservationModelType)
penalty_decay (float)
is_obstacle_hit_terminal (bool)
- get_actions()[source]
Get all possible actions in the discrete action space.
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)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- observation_log_probability(next_state, action, observations)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
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:
- Return type:
- Returns:
1-D array of reward values with shape
(N,).
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)[source]
Random rollout via native C++.
The variant-aware
_native.simulate_rolloutkernel covers all three reward models in expectation (the rollout RNG draws come from the module-level C++ RNG and match the Python reward models sample-mean, not bit-exact), so no variant gate is required here. The Python fallback is retained only for the realised-position correctness case below.- Parameters:
state (
Any) – Current 2-D position[x, y].action_sampler (
Any) – Object with asample()method; used only for the Python fallback path. On the native path, action indices are pre-drawn inside this method.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:
- Returns:
Discounted sum of immediate rewards along the sampled trajectory.
- class POMDPPlanners.environments.DiscreteLightDarkPOMDP(discount_factor, name='DiscreteLightDarkPOMDP', transition_error_prob=0.05, observation_error_prob=0.05, beacons=[(0, 0), (0, 5), (0, 10), (5, 0), (5, 5), (5, 10), (10, 0), (10, 5), (10, 10)], goal_state=array([10, 5]), start_state=array([0, 5]), obstacles=[(3, 7), (5, 5)], obstacle_hit_probability=0.2, obstacle_reward=-10.0, goal_reward=10.0, beacon_radius=1.0, fuel_cost=2.0, grid_size=11, is_stochastic_reward=True, observation_model_type=ObservationModelType.NORMAL)[source]
Bases:
BaseLightDarkPOMDPDiscreteActions,DiscreteActionsEnvironmentDiscrete Light-Dark POMDP Environment for Robot Navigation with Observation Uncertainty.
This environment implements a discretized version of the classic Light-Dark POMDP problem, where a robot must navigate from a start position to a goal position in a grid world with beacons and obstacles. The key challenge is that the robot’s observation quality depends on its distance from beacons - closer to beacons means more accurate observations.
Problem Description: The robot operates in a discrete grid world where it can move in four cardinal directions. The environment includes: - Beacons: Fixed positions that provide location reference with varying accuracy - Obstacles: Grid cells that incur penalties when hit - Goal: Target position that provides high reward when reached - Observation uncertainty: Decreases with proximity to beacons (light areas)
Key Features: - Discrete state space: Robot positions are restricted to grid cells - Discrete action space: North, South, East, West movements - Multiple observation models available (normal, no observation in dark) - Distance-dependent observation accuracy: Closer to beacons = better observations - Stochastic transitions: Actions may fail with configurable probability - Obstacle avoidance: Penalties for hitting obstacles during navigation - Configurable environment parameters: Grid size, beacon positions, obstacles
State Space: - 2D grid coordinates (x, y) representing robot position - Bounded by grid_size parameter (default: 11x11 grid)
Action Space: - Discrete actions: [‘North’, ‘South’, ‘East’, ‘West’] - Each action moves robot one grid cell in the corresponding direction - Boundary conditions: Actions that would move outside grid are blocked
Observation Space: - Discrete observations based on beacon proximity and noise - Observation accuracy improves with proximity to beacons - Stochastic observation errors controlled by observation_error_prob
Reward Structure: - Goal reward: Large positive reward for reaching the goal state - Obstacle penalty: Negative reward for hitting obstacles - Fuel cost: Small negative reward for each movement action - Distance-based penalties: Encourage efficient navigation
- Parameters:
discount_factor (float)
name (str)
transition_error_prob (float)
observation_error_prob (float)
goal_state (ndarray)
start_state (ndarray)
obstacle_hit_probability (float)
obstacle_reward (float)
goal_reward (float)
beacon_radius (float)
fuel_cost (float)
grid_size (int)
is_stochastic_reward (bool)
observation_model_type (ObservationModelType)
- transition_error_prob
Probability that an action fails (results in different movement)
- observation_error_prob
Probability of observation noise/error
- is_stochastic_reward
Whether rewards include stochastic components
- beacons
List of (x, y) beacon positions that provide navigation references
- goal_state
Target position (x, y) that robot should reach
- start_state
Initial robot position (x, y)
- obstacles
List of (x, y) obstacle positions to avoid
- grid_size
Dimension of the square grid world
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = DiscreteLightDarkPOMDP( ... discount_factor=0.95, ... transition_error_prob=0.1, ... observation_error_prob=0.15, ... beacons=[(1, 1), (2, 2)], ... grid_size=11 ... ) >>> >>> # 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
References: - Platt, R., et al. “Belief space planning assuming maximum likelihood observations.” (2010) - Kurniawati, H., et al. “SARSOP: Efficient point-based POMDP planning by approximating optimally reachable belief spaces.” (2008) - Light-Dark domain: Classic POMDP benchmark for testing observation uncertainty
- compute_metrics(histories)[source]
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.
- hash_action(action)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- is_terminal(state)[source]
Check if a state is terminal.
- Parameters:
state (
ndarray) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.- Parameters:
- Return type:
- Returns:
ndarray of shape
(N,)with log-probabilities or log-PDFs.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
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:
- Return type:
- Returns:
1-D array of reward values with shape
(N,).
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.- Parameters:
- Returns:
np.ndarrayof shape(N, *dim). For structured envs (Tiger strings, Pacman tuples): a list of length N.- Return type:
- sample_next_step(state, action)[source]
Sample a complete state transition step.
This convenience method combines state transition, observation generation, and reward calculation in a single operation.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)[source]
Random rollout via native C++.
Pre-draws the per-step action indices on the Python side (so the
action_samplerinteraction stays observable for tests / hooks) and forwards to the native discrete rollout kernel. The kernel uses the module-level C++ RNG for the per-step obstacle-hit and transition-error draws.Falls back to the base-class Python loop when the env is configured for a non-NORMAL observation model only if the rollout would otherwise short-circuit at the wrong place — actually rollout reward and dynamics are independent of the observation model, so the native path is safe for all observation models.
- Parameters:
state (
Any) – Current 2-D position[x, y].action_sampler (
Any) – Object with asample()method; used only for the Python fallback path. On the native path, action indices are pre-drawn vianp.random.randint.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:
- Returns:
Discounted sum of immediate rewards along the sampled trajectory.
- class POMDPPlanners.environments.LaserTagPOMDP(discount_factor, name='LaserTagPOMDP', floor_shape=(11, 7), walls={(1, 2), (3, 0), (3, 4), (5, 0), (6, 4), (9, 1), (9, 4), (10, 6)}, tag_reward=10.0, tag_penalty=10.0, step_cost=1.0, measurement_noise=1.0, dangerous_areas={(2, 5), (5, 3), (7, 1)}, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, output_dir=None, debug=False, use_queue_logger=False, initial_state=None, transition_error_prob=0.0, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0, opponent_policy=OpponentPolicy.EVADE)[source]
Bases:
DiscreteActionsEnvironmentLaserTag POMDP environment implementation.
This is a pursuit-evasion problem where a robot must navigate a grid to tag an opponent. The robot receives noisy observations of the opponent’s position and must decide when and where to attempt tagging.
Problem Structure: - States: numpy array [robot_row, robot_col, opp_row, opp_col, terminal] - Actions: North(0), South(1), East(2), West(3), Tag(4) - Observations: 8-directional laser measurements (N,NE,E,SE,S,SW,W,NW) - Rewards: Tag success(+10), Tag failure(-10), Movement(-1)
- Parameters:
discount_factor (float)
name (str)
tag_reward (float)
tag_penalty (float)
step_cost (float)
measurement_noise (float)
dangerous_area_radius (float)
dangerous_area_penalty (float)
output_dir (Path | None)
debug (bool)
use_queue_logger (bool)
initial_state (ndarray | None)
transition_error_prob (float)
reward_model_type (RewardModelType)
penalty_decay (float)
opponent_policy (OpponentPolicy)
- floor_shape
Grid dimensions as (rows, cols)
- walls
Set of wall positions as (row, col) tuples
- tag_reward
Reward for successful tagging
- tag_penalty
Penalty for unsuccessful tagging
- step_cost
Cost per movement action
- measurement_noise
Standard deviation of observation noise
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = LaserTagPOMDP(discount_factor=0.95) >>> >>> # 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
- cache_visualization(history, cache_path)[source]
Cache visualization of the LaserTag episode as an animated GIF.
Creates an animated visualization showing: - Robot movement (red circle) - Opponent movement (blue circle) - Walls (black squares) - Dangerous areas (red circles) - Action arrows showing robot’s intended movement - Laser measurements (green rays from robot position) - Belief particles (if available) showing robot’s belief about opponent location - Grid boundaries and coordinate system
- Parameters:
- Raises:
ValueError – If history is empty or contains invalid data
TypeError – If cache_path is not a Path object or doesn’t end with .gif
- Return type:
- compute_metrics(histories)[source]
Compute LaserTag POMDP specific metrics from simulation histories.
- Return type:
- Parameters:
- hash_action(action)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
Observations are 8-dimensional laser measurements or terminal observations.
- observation_log_probability(next_state, action, observations)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action transition.
The wall / dangerous-area penalty is computed against the realised post-action robot position taken from
next_state. When the caller omitsnext_state(e.g., the open-loop scalar API path) the method resamples a transition viasample_next_state()so the penalty is always scored against an actual draw from the transition kernel — never against the open-loopstate + action_vectorintended position.Environment.sample_next_step()threads its samplednext_stateinto this method so trajectory and reward agree on the same realisation.
- reward_batch(states, action, next_states=None)[source]
Vectorised reward for a batch of states under a single action.
When
next_statesis supplied the danger-area / wall penalty is evaluated against the realised positions innext_states[:, :2](matching the contract honoured byEnvironment.sample_next_step()). When it isNonethe method resamples viasample_next_state_batch()whenever penalty terms exist, then delegates to the reward model so reward and trajectory remain consistent end-to-end.
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- class POMDPPlanners.environments.MountainCarPOMDP(discount_factor, state_transition_cov=None, name='MountainCarPOMDP', output_dir=None, debug=False, use_queue_logger=False)[source]
Bases:
DiscreteActionsEnvironmentMountain Car problem formulated as a POMDP.
This environment simulates an underpowered car trying to reach the top of a steep mountain. The car must build momentum by oscillating back and forth to gain enough energy to reach the goal, with noisy observations of its state.
Problem Structure: - State: [position, velocity] (continuous, position ∈ [-1.2, 0.6], velocity ∈ [-0.07, 0.07]) - Actions: [-1 (reverse), 0 (neutral), 1 (forward)] (discrete) - Observations: Noisy state measurements (continuous) - Rewards: 0 for reaching goal (position ≥ 0.5), -1 per time step otherwise - Goal: Drive car to position ≥ 0.5 (top of mountain)
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = MountainCarPOMDP(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:
- DEFAULT_STATE_TRANSITION_COV = array([[2.5e-05, 0.0e+00], [0.0e+00, 1.0e-06]])
- cache_visualization(history, cache_path)[source]
Cache visualization data for an episode history.
This method can be overridden by subclasses to provide environment-specific visualization caching capabilities.
- compute_metrics(histories)[source]
Compute Mountain Car POMDP specific metrics from simulation histories.
- Parameters:
- Return type:
- Returns:
List of MetricValue objects containing the computed metrics
- get_actions()[source]
Get all possible actions in the discrete action space.
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)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- initial_observation_dist()[source]
Get the initial observation distribution.
- Return type:
- Returns:
Distribution over initial observations
Note
Subclasses must implement this method to define initial observations.
- initial_state_dist()[source]
Get the initial state distribution.
- Return type:
- Returns:
Distribution over initial states
Note
Subclasses must implement this method to define the starting distribution.
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
- Parameters:
- Return type:
- 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)[source]
Check if a state is terminal.
- Parameters:
state (
Tuple[float,float]) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
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:
- Return type:
- Returns:
1-D array of reward values with shape
(N,).
- sample_next_state(state, action, n_samples=1)[source]
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 lengthn_samples(numeric envs returnnp.ndarrayof shape(n_samples, *dim); structured envs returnList[T]).- Return type:
- Parameters:
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)[source]
Random rollout via native C++.
- Parameters:
state (
Any) – Current 2-D car state[position, velocity].action_sampler (
Any) – Object with asample()method (kept for API parity with the baseEnvironmentcontract; unused on the native rollout path which draws indices directly via NumPy).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:
- Returns:
Discounted sum of immediate rewards along the sampled trajectory.
- class POMDPPlanners.environments.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)[source]
Bases:
DiscreteActionsEnvironmentPacMan 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:
num_ghosts (int)
pellet_reward (float)
ghost_collision_penalty (float)
step_penalty (float)
win_reward (float)
ghost_aggressiveness (float)
ghost_coordination (str)
observation_noise_factor (float)
max_observation_noise (float)
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_penaltywhenever 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
- get_metric_names()[source]
Get names of PacMan POMDP specific metrics.
- Return type:
- 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_encountersis the per-step sum of ghost-collision and dangerous-area-step events; a step that is both counts twice.
- get_transition_cpp_ctor_kwargs()[source]
Return the cached per-env kwargs dict passed to PacManTransitionCpp.
- hash_action(action)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- property initial_ghost_pos: Tuple[int, int]
returns first ghost position.
- Type:
Backward compatibility
- initial_observation_dist()[source]
Get the initial observation distribution.
Returns a live distribution that draws fresh noisy ghost-position observations from the true initial state on each
samplecall, 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:
- make_state(*, pacman_pos, ghost_positions, pellets=None, score=0.0, terminal=False)[source]
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 lengthnum_ghosts.pellets (
Optional[Tuple[Tuple[int,int],...]]) – Active pellet positions.Nonemeans every initial pellet is active (useful for constructing initial states).score (
float) – Current game score.terminal (
bool) – Whether the state is terminal.
- Return type:
- Returns:
1-D
float64array 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- reward(state, action, next_state=None)[source]
Calculate immediate reward.
Uses the realised
next_statewhen supplied (e.g. byEnvironment.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. Whennext_stateisNone, falls back to sampling one here.
- reward_batch(states, action, next_states=None)[source]
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. Whennext_statesis 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.
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)[source]
Estimate the value of
statevia 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 asample()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:
- Returns:
Discounted cumulative reward along the sampled trajectory.
- transition_log_probability(state, action, next_states)[source]
Log-probability of each candidate next state under
(state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate next states. Subclasses must implement.
- class POMDPPlanners.environments.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)[source]
Bases:
DiscreteActionsEnvironmentRobotic 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. Whenobstacle_hit_probability < 1.0, the penalty is applied only with that probability perreward()/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 makesreward(state, action)non- deterministic given a state-action pair, so any external caching that assumes deterministic rewards must be aware of this.transition_log_probabilityis unaffected; the obstacle still deterministically blocks movement. The native C++ rollout applies the Bernoulliobstacle_hit_probabilitydraw internally, sosimulate_random_rolloutalways routes through the native kernel.- Dangerous areas:
dangerous_areasis a separate, additive concept fromobstacles. Each entry is a circular region centred at(x, y)with radiusdangerous_area_radius. Entering a dangerous area appliesdangerous_area_penalty(a negative number, added to reward — same sign convention asobstacle_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 onedangerous_area_penaltyis applied per step even when multiple zones overlap. Like obstacles, the penalty supports a Bernoullidangerous_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)
obstacle_radius (float)
obstacle_penalty (float)
obstacle_hit_probability (float)
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)[source]
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:
- 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:
- compute_metrics(histories)[source]
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.
- get_actions()[source]
Get all possible actions in the discrete action space.
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)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- hash_observation(observation)[source]
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:
- 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()[source]
Get the initial observation distribution.
- Return type:
- Returns:
Distribution over initial observations
Note
Subclasses must implement this method to define initial observations.
- initial_state_dist()[source]
Get the initial state distribution.
- Return type:
- Returns:
Distribution over initial states
Note
Subclasses must implement this method to define the starting distribution.
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
- Parameters:
- Return type:
- 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)[source]
Check if a state is terminal.
- Parameters:
state (
ndarray) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- observation_log_probability_single(next_state, action, observation)[source]
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 batchedobservation_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.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
Calculate rewards for a batch of states given a single action.
When
next_statesis 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 cachedPushVectorizedUpdater. Per-particle rewards are computed in the C++push_reward_batchkernel (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.
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_next_step(state, action)[source]
Sample a complete state transition step.
This convenience method combines state transition, observation generation, and reward calculation in a single operation.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- class POMDPPlanners.environments.RockSamplePOMDP(map_size=(5, 5), rock_positions=None, init_pos=(0, 0), sensor_efficiency=10.0, bad_rock_penalty=-10.0, good_rock_reward=10.0, step_penalty=0.0, sensor_use_penalty=0.0, exit_reward=10.0, dangerous_areas=None, dangerous_area_radius=1.0, dangerous_area_penalty=-5.0, dangerous_area_hit_probability=1.0, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0, discount_factor=0.95, name='RockSample', output_dir=None, debug=False, use_queue_logger=False)[source]
Bases:
DiscreteActionsEnvironmentRockSample POMDP environment
This environment implements the classic rock sampling problem where a robot must navigate a grid, use sensors to evaluate rocks, and decide which ones to sample while balancing exploration costs and sampling rewards.
- Stochasticity:
The dangerous-area penalty can be applied either deterministically (the default) or stochastically. When
dangerous_area_hit_probability == 1.0(default), the penalty is applied every time the robot’s next position lies inside a dangerous area, matching legacy behavior. Whendangerous_area_hit_probability < 1.0, the penalty is applied only with that probability perreward()/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 makesreward(state, action)non-deterministic given a state-action pair, so any external caching that assumes deterministic rewards must be aware of this.transition_log_probabilityis unaffected.
- Parameters:
sensor_efficiency (float)
bad_rock_penalty (float)
good_rock_reward (float)
step_penalty (float)
sensor_use_penalty (float)
exit_reward (float)
dangerous_area_radius (float)
dangerous_area_penalty (float)
dangerous_area_hit_probability (float)
reward_model_type (RewardModelType)
penalty_decay (float)
discount_factor (float)
name (str)
output_dir (Path | None)
debug (bool)
use_queue_logger (bool)
- map_size
Grid dimensions as (rows, cols)
- rock_positions
List of rock positions as (row, col) tuples
- init_pos
Initial robot position
- sensor_efficiency
Sensor noise parameter (higher = less noise)
- bad_rock_penalty
Penalty for sampling a bad rock
- good_rock_reward
Reward for sampling a good rock
- step_penalty
Cost for each action
- sensor_use_penalty
Additional cost for using sensor
- exit_reward
Reward for reaching the exit
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = RockSamplePOMDP(map_size=(5, 5), rock_positions=[(0, 0), (2, 2), (3, 3)]) >>> >>> # 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
- hash_action(action)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- observation_log_probability(next_state, action, observations)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- reward(state, action, next_state=None)[source]
Calculate immediate reward.
Uses the realised
next_statewhen supplied (e.g. byEnvironment.sample_next_step()) so the dangerous-area penalty fires against the same outcome as the trajectory instead of a fresh draw.
- reward_batch(states, action, next_states=None)[source]
Calculate rewards for a batch of states given a single action.
Threads caller-supplied
next_statesthrough to the dangerous-area position check so the batch path agrees with the scalarreward()wheneverEnvironment.sample_next_step(or any other caller) pre-samples next states. Whennext_states is None, we fall back to closed-form reconstruction of the next robot position from(state, action); RockSample transitions are deterministic, so this fallback matches a fresh draw fromsample_next_state(). The per-call Bernoulli refund for the dangerous-area penalty is preserved in both branches.
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)[source]
Random rollout via native C++ deterministic transition and reward kernel.
The C++ kernel applies the variant-aware dangerous-area reward term directly, so no Python fallback is required when danger zones are configured.
- Parameters:
state (
Any) – Current RockSample state array.action_sampler (
Any) – Object with asample()method returning an integer action. Currently unused — actions are drawn uniformly by the native kernel.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:
- Returns:
Discounted sum of immediate rewards along the sampled trajectory.
- transition_log_probability(state, action, next_states)[source]
Log-probability of each candidate next state under
(state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate next states. Subclasses must implement.
- class POMDPPlanners.environments.SafeAntVelocityPOMDP(discount_factor, safe_velocity_threshold=2.0, max_force=1.0, dt=0.1, mass=1.0, damping=0.1, position_noise=0.1, velocity_noise=0.2, safety_violation_penalty=-100.0, movement_reward_scale=1.0, name='SafeVelocityPOMDP', output_dir=None, debug=False, use_queue_logger=False)[source]
Bases:
DiscreteActionsEnvironmentSafety-critical velocity control task formulated as a POMDP.
This environment presents a safety-critical control problem where an agent must navigate while keeping velocity below a safety threshold. The challenge comes from balancing exploration rewards with safety constraints under noisy velocity observations.
Problem Structure: - State: [position_x, position_y, velocity_x, velocity_y] (continuous) - Actions: [0=no force, 1=small, 2=medium, 3=large force] (discrete) - Observations: Noisy position and velocity measurements (continuous) - Rewards: Movement reward - safety violation penalty (if unsafe) - Safety constraint: velocity magnitude ≤ safe_velocity_threshold - Termination: Velocity exceeds 1.5x safety threshold
Safety Features: - Tracks safety and critical violation rates - Heavy penalties for constraint violations - Configurable safety thresholds and penalties - Physics simulation with uncertainty in force direction
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = SafeAntVelocityPOMDP(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:
- cache_visualization(history, cache_path)[source]
Cache animated visualization of the safety ant velocity episode.
Creates an animated GIF showing the ant’s movement trajectory with velocity vectors, safety zones, force applications, and safety constraint violations.
- Parameters:
- 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:
- compute_metrics(histories)[source]
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.
- get_actions()[source]
Get all possible actions in the discrete action space.
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)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- initial_observation_dist()[source]
Get the initial observation distribution.
- Return type:
- Returns:
Distribution over initial observations
Note
Subclasses must implement this method to define initial observations.
- initial_state_dist()[source]
Get the initial state distribution.
- Return type:
- Returns:
Distribution over initial states
Note
Subclasses must implement this method to define the starting distribution.
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
- Parameters:
- Return type:
- 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)[source]
Check if a state is terminal.
- Parameters:
state (
ndarray) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
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:
- Return type:
- Returns:
1-D array of reward values with shape
(N,).
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.
- sample_next_step(state, action)[source]
Sample a complete state transition step.
This convenience method combines state transition, observation generation, and reward calculation in a single operation.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)[source]
Random rollout via native C++ physics and reward kernel.
Pre-draws action indices and runs the full rollout in a single C++ call, avoiding per-step Python dispatch.
- Parameters:
state (
Any) – Current 4-D state[px, py, vx, vy].action_sampler (
Any) – Accepted for interface compatibility with the basesimulate_random_rolloutsignature; the native rollout draws action indices vianp.random.randintdirectly and never invokes the sampler.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:
- Returns:
Discounted sum of immediate rewards along the sampled trajectory.
- class POMDPPlanners.environments.SanityPOMDP(discount_factor=0.95, output_dir=None, debug=False, use_queue_logger=False)[source]
Bases:
DiscreteActionsEnvironmentSimple sanity check POMDP environment for testing and debugging.
This environment provides the simplest possible POMDP formulation with deterministic dynamics and perfect observability. It serves as a baseline for testing POMDP algorithms and ensuring correctness.
Problem Structure: - States: 0 (good), 1 (bad) - Actions: 0 (choose good), 1 (choose bad) - Observations: Same as states (perfect observability) - Rewards: 1.0 for good state, 0.0 for bad state - Dynamics: Deterministic state transitions based on action
The optimal policy is trivial: always choose action 0 to stay in the good state.
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = SanityPOMDP(discount_factor=0.95) >>> >>> # 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
- get_actions()[source]
Get all possible actions in the discrete action space.
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)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- initial_observation_dist()[source]
Get the initial observation distribution.
- Return type:
- Returns:
Distribution over initial observations
Note
Subclasses must implement this method to define initial observations.
- initial_state_dist()[source]
Get the initial state distribution.
- Return type:
- Returns:
Distribution over initial states
Note
Subclasses must implement this method to define the starting distribution.
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
- Parameters:
- Return type:
- 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)[source]
Check if a state is terminal.
- Parameters:
state (
int) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- class POMDPPlanners.environments.TigerPOMDP(discount_factor, name='TigerPOMDP', output_dir=None, debug=False, use_queue_logger=False)[source]
Bases:
DiscreteActionsEnvironmentTiger POMDP environment implementation.
This is the classic Tiger problem where an agent must decide which door to open to find treasure while avoiding the tiger. The agent can listen for acoustic cues but receives noisy observations.
Problem Structure: - States: tiger_left, tiger_right (tiger’s location) - Actions: listen, open_left, open_right - Observations: hear_left, hear_right, hear_nothing - Rewards: listen(-1), correct_door(+10), wrong_door(-100)
- Parameters:
- states
List of possible states
- actions
List of possible actions
- observations
List of possible observations
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> tiger = TigerPOMDP(discount_factor=0.95) >>> >>> # Get initial state and actions >>> initial_state = tiger.initial_state_dist().sample()[0] >>> actions = tiger.get_actions() >>> >>> # Sample complete step using convenience method >>> action = actions[0] >>> next_state, observation, reward = tiger.sample_next_step(initial_state, action) >>> >>> # Check terminal condition >>> tiger.is_terminal(initial_state) False
- compute_metrics(histories)[source]
Compute Tiger POMDP specific metrics from simulation histories.
- Parameters:
- Return type:
- Returns:
List of MetricValue objects containing the computed metrics
- get_actions()[source]
Get all possible actions in the discrete action space.
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)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- initial_observation_dist()[source]
Get the initial observation distribution.
- Return type:
- Returns:
Distribution over initial observations
Note
Subclasses must implement this method to define initial observations.
- initial_state_dist()[source]
Get the initial state distribution.
- Return type:
- Returns:
Distribution over initial states
Note
Subclasses must implement this method to define the starting distribution.
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
- Parameters:
- Return type:
- 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)[source]
Check if a state is terminal.
- Parameters:
state (
str) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.- Parameters:
- Return type:
- Returns:
ndarray of shape
(N,)with log-probabilities or log-PDFs.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
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:
- Return type:
- Returns:
1-D array of reward values with shape
(N,).
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.- Parameters:
- Returns:
np.ndarrayof shape(N, *dim). For structured envs (Tiger strings, Pacman tuples): a list of length N.- Return type:
Subpackages
- POMDPPlanners.environments.cartpole_pomdp package
CartPoleInitialObservationDistributionCartPoleInitialStateDistributionCartPolePOMDPCartPolePOMDP.DEFAULT_STATE_TRANSITION_COVCartPolePOMDP.compute_metrics()CartPolePOMDP.get_actions()CartPolePOMDP.get_metric_names()CartPolePOMDP.hash_action()CartPolePOMDP.initial_observation_dist()CartPolePOMDP.initial_state_dist()CartPolePOMDP.is_equal_observation()CartPolePOMDP.is_terminal()CartPolePOMDP.observation_log_probability()CartPolePOMDP.observation_log_probability_per_state()CartPolePOMDP.reward()CartPolePOMDP.reward_batch()CartPolePOMDP.sample_next_state()CartPolePOMDP.sample_next_state_batch()CartPolePOMDP.sample_observation()CartPolePOMDP.simulate_random_rollout()CartPolePOMDP.transition_log_probability()
CartPolePOMDPMetrics- Submodules
- POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp module
CartPoleInitialObservationDistributionCartPoleInitialStateDistributionCartPolePOMDPCartPolePOMDP.DEFAULT_STATE_TRANSITION_COVCartPolePOMDP.compute_metrics()CartPolePOMDP.get_actions()CartPolePOMDP.get_metric_names()CartPolePOMDP.hash_action()CartPolePOMDP.initial_observation_dist()CartPolePOMDP.initial_state_dist()CartPolePOMDP.is_equal_observation()CartPolePOMDP.is_terminal()CartPolePOMDP.observation_log_probability()CartPolePOMDP.observation_log_probability_per_state()CartPolePOMDP.reward()CartPolePOMDP.reward_batch()CartPolePOMDP.sample_next_state()CartPolePOMDP.sample_next_state_batch()CartPolePOMDP.sample_observation()CartPolePOMDP.simulate_random_rollout()CartPolePOMDP.transition_log_probability()
CartPolePOMDPMetrics
- POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp_beliefs module
CartPoleVectorizedUpdaterCartPoleVectorizedUpdater.state_transition_distCartPoleVectorizedUpdater.obs_distCartPoleVectorizedUpdater.force_magCartPoleVectorizedUpdater.gravityCartPoleVectorizedUpdater.masscartCartPoleVectorizedUpdater.masspoleCartPoleVectorizedUpdater.total_massCartPoleVectorizedUpdater.lengthCartPoleVectorizedUpdater.polemass_lengthCartPoleVectorizedUpdater.tauCartPoleVectorizedUpdater.kinematics_integratorCartPoleVectorizedUpdater.batch_observation_log_likelihood()CartPoleVectorizedUpdater.batch_transition()CartPoleVectorizedUpdater.config_idCartPoleVectorizedUpdater.from_environment()
create_cartpole_belief()
- POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp_gaussian_beliefs module
- POMDPPlanners.environments.environment_utils package
- POMDPPlanners.environments.laser_tag_pomdp package
ContinuousLaserTagPOMDPContinuousLaserTagPOMDP.cache_visualization()ContinuousLaserTagPOMDP.compute_metrics()ContinuousLaserTagPOMDP.get_metric_names()ContinuousLaserTagPOMDP.grid_sizeContinuousLaserTagPOMDP.hash_action()ContinuousLaserTagPOMDP.hash_observation()ContinuousLaserTagPOMDP.initial_observation_dist()ContinuousLaserTagPOMDP.initial_state_dist()ContinuousLaserTagPOMDP.is_equal_observation()ContinuousLaserTagPOMDP.is_terminal()ContinuousLaserTagPOMDP.observation_log_probability()ContinuousLaserTagPOMDP.observation_log_probability_per_state()ContinuousLaserTagPOMDP.observation_log_probability_single()ContinuousLaserTagPOMDP.reward()ContinuousLaserTagPOMDP.reward_batch()ContinuousLaserTagPOMDP.sample_next_state()ContinuousLaserTagPOMDP.sample_next_state_batch()ContinuousLaserTagPOMDP.sample_observation()ContinuousLaserTagPOMDP.simulate_random_rollout()ContinuousLaserTagPOMDP.transition_log_probability()ContinuousLaserTagPOMDP.walls
ContinuousLaserTagPOMDPDiscreteActionsContinuousLaserTagPOMDPDiscreteActions.get_actions()ContinuousLaserTagPOMDPDiscreteActions.hash_action()ContinuousLaserTagPOMDPDiscreteActions.observation_log_probability()ContinuousLaserTagPOMDPDiscreteActions.observation_log_probability_per_state()ContinuousLaserTagPOMDPDiscreteActions.observation_log_probability_single()ContinuousLaserTagPOMDPDiscreteActions.reward()ContinuousLaserTagPOMDPDiscreteActions.reward_batch()ContinuousLaserTagPOMDPDiscreteActions.sample_next_state()ContinuousLaserTagPOMDPDiscreteActions.sample_next_state_batch()ContinuousLaserTagPOMDPDiscreteActions.sample_observation()ContinuousLaserTagPOMDPDiscreteActions.simulate_random_rollout()ContinuousLaserTagPOMDPDiscreteActions.transition_log_probability()
LaserTagPOMDPLaserTagPOMDP.floor_shapeLaserTagPOMDP.wallsLaserTagPOMDP.tag_rewardLaserTagPOMDP.tag_penaltyLaserTagPOMDP.step_costLaserTagPOMDP.measurement_noiseLaserTagPOMDP.cache_visualization()LaserTagPOMDP.compute_metrics()LaserTagPOMDP.get_actions()LaserTagPOMDP.get_metric_names()LaserTagPOMDP.hash_action()LaserTagPOMDP.initial_observation_dist()LaserTagPOMDP.initial_state_dist()LaserTagPOMDP.is_equal_observation()LaserTagPOMDP.is_terminal()LaserTagPOMDP.observation_log_probability()LaserTagPOMDP.observation_log_probability_per_state()LaserTagPOMDP.reward()LaserTagPOMDP.reward_batch()LaserTagPOMDP.sample_next_state()LaserTagPOMDP.sample_next_state_batch()LaserTagPOMDP.sample_observation()LaserTagPOMDP.simulate_random_rollout()LaserTagPOMDP.transition_log_probability()
OpponentPolicy- Subpackages
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs package
ContinuousLaserTagVectorizedUpdaterLaserTagVectorizedUpdatercreate_continuous_laser_tag_belief()create_laser_tag_belief()- Submodules
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs.continuous_laser_tag_belief_factory module
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs.continuous_laser_tag_vectorized_updater module
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs.laser_tag_belief_factory module
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs.laser_tag_vectorized_updater module
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_utils package
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs package
- Submodules
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_geometry module
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_pomdp module
ContinuousLaserTagPOMDPContinuousLaserTagPOMDP.cache_visualization()ContinuousLaserTagPOMDP.compute_metrics()ContinuousLaserTagPOMDP.get_metric_names()ContinuousLaserTagPOMDP.grid_sizeContinuousLaserTagPOMDP.hash_action()ContinuousLaserTagPOMDP.hash_observation()ContinuousLaserTagPOMDP.initial_observation_dist()ContinuousLaserTagPOMDP.initial_state_dist()ContinuousLaserTagPOMDP.is_equal_observation()ContinuousLaserTagPOMDP.is_terminal()ContinuousLaserTagPOMDP.observation_log_probability()ContinuousLaserTagPOMDP.observation_log_probability_per_state()ContinuousLaserTagPOMDP.observation_log_probability_single()ContinuousLaserTagPOMDP.reward()ContinuousLaserTagPOMDP.reward_batch()ContinuousLaserTagPOMDP.sample_next_state()ContinuousLaserTagPOMDP.sample_next_state_batch()ContinuousLaserTagPOMDP.sample_observation()ContinuousLaserTagPOMDP.simulate_random_rollout()ContinuousLaserTagPOMDP.transition_log_probability()ContinuousLaserTagPOMDP.walls
ContinuousLaserTagPOMDPDiscreteActionsContinuousLaserTagPOMDPDiscreteActions.get_actions()ContinuousLaserTagPOMDPDiscreteActions.hash_action()ContinuousLaserTagPOMDPDiscreteActions.observation_log_probability()ContinuousLaserTagPOMDPDiscreteActions.observation_log_probability_per_state()ContinuousLaserTagPOMDPDiscreteActions.observation_log_probability_single()ContinuousLaserTagPOMDPDiscreteActions.reward()ContinuousLaserTagPOMDPDiscreteActions.reward_batch()ContinuousLaserTagPOMDPDiscreteActions.sample_next_state()ContinuousLaserTagPOMDPDiscreteActions.sample_next_state_batch()ContinuousLaserTagPOMDPDiscreteActions.sample_observation()ContinuousLaserTagPOMDPDiscreteActions.simulate_random_rollout()ContinuousLaserTagPOMDPDiscreteActions.transition_log_probability()
ContinuousLaserTagPOMDPMetricsContinuousLaserTagPOMDPMetrics.AVERAGE_ALL_DANGEROUS_ENCOUNTERSContinuousLaserTagPOMDPMetrics.AVERAGE_DANGEROUS_AREA_STEPSContinuousLaserTagPOMDPMetrics.AVERAGE_EPISODE_LENGTHContinuousLaserTagPOMDPMetrics.AVERAGE_FAILED_TAG_ATTEMPTSContinuousLaserTagPOMDPMetrics.AVERAGE_WALL_COLLISIONSContinuousLaserTagPOMDPMetrics.GOAL_REACHING_RATEContinuousLaserTagPOMDPMetrics.TAG_SUCCESS_RATE
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_visualizer module
ContinuousLaserTagVisualizerContinuousLaserTagVisualizer.grid_sizeContinuousLaserTagVisualizer.wallsContinuousLaserTagVisualizer.robot_radiusContinuousLaserTagVisualizer.opponent_radiusContinuousLaserTagVisualizer.dangerous_areasContinuousLaserTagVisualizer.dangerous_area_radiusContinuousLaserTagVisualizer.create_visualization()
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp module
LaserTagPOMDPLaserTagPOMDP.floor_shapeLaserTagPOMDP.wallsLaserTagPOMDP.tag_rewardLaserTagPOMDP.tag_penaltyLaserTagPOMDP.step_costLaserTagPOMDP.measurement_noiseLaserTagPOMDP.cache_visualization()LaserTagPOMDP.compute_metrics()LaserTagPOMDP.get_actions()LaserTagPOMDP.get_metric_names()LaserTagPOMDP.hash_action()LaserTagPOMDP.initial_observation_dist()LaserTagPOMDP.initial_state_dist()LaserTagPOMDP.is_equal_observation()LaserTagPOMDP.is_terminal()LaserTagPOMDP.observation_log_probability()LaserTagPOMDP.observation_log_probability_per_state()LaserTagPOMDP.reward()LaserTagPOMDP.reward_batch()LaserTagPOMDP.sample_next_state()LaserTagPOMDP.sample_next_state_batch()LaserTagPOMDP.sample_observation()LaserTagPOMDP.simulate_random_rollout()LaserTagPOMDP.transition_log_probability()
LaserTagPOMDPMetricsLaserTagPOMDPMetrics.AVERAGE_ALL_DANGEROUS_ENCOUNTERSLaserTagPOMDPMetrics.AVERAGE_DANGEROUS_AREA_STEPSLaserTagPOMDPMetrics.AVERAGE_EPISODE_LENGTHLaserTagPOMDPMetrics.AVERAGE_FAILED_TAG_ATTEMPTSLaserTagPOMDPMetrics.AVERAGE_OBSTACLE_COLLISIONSLaserTagPOMDPMetrics.GOAL_REACHING_RATELaserTagPOMDPMetrics.TAG_SUCCESS_RATE
RewardModelType
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_visualizer module
- POMDPPlanners.environments.light_dark_pomdp package
- Subpackages
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_beliefs package
ContinuousLightDarkDistanceBasedVectorizedUpdaterContinuousLightDarkNoObsInDarkVectorizedUpdaterContinuousLightDarkVectorizedUpdaterDiscreteLightDarkDistanceBasedVectorizedUpdaterDiscreteLightDarkNoObsInDarkVectorizedUpdaterDiscreteLightDarkVectorizedUpdaterGaussianBeliefUpdaterTypecreate_continuous_light_dark_belief()create_continuous_light_dark_gaussian_belief()create_discrete_light_dark_belief()- Submodules
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_beliefs.continuous_light_dark_belief_factory module
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_beliefs.continuous_light_dark_gaussian_beliefs module
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_beliefs.continuous_light_dark_vectorized_updater module
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_beliefs.discrete_light_dark_belief_factory module
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_beliefs.discrete_light_dark_vectorized_updater module
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils package
- Submodules
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.base_light_dark_pomdp module
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.light_dark_reward_models module
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.light_dark_visualizer module
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.numba_kernels module
- POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_beliefs package
- Submodules
- POMDPPlanners.environments.light_dark_pomdp.continuous_light_dark_pomdp module
ContinuousLightDarkPOMDPContinuousLightDarkPOMDP.compute_metrics()ContinuousLightDarkPOMDP.get_metric_names()ContinuousLightDarkPOMDP.hash_action()ContinuousLightDarkPOMDP.is_terminal()ContinuousLightDarkPOMDP.observation_log_probability()ContinuousLightDarkPOMDP.observation_log_probability_per_state()ContinuousLightDarkPOMDP.observation_log_probability_single()ContinuousLightDarkPOMDP.reward()ContinuousLightDarkPOMDP.reward_batch()ContinuousLightDarkPOMDP.sample_next_state()ContinuousLightDarkPOMDP.sample_next_state_batch()ContinuousLightDarkPOMDP.sample_observation()ContinuousLightDarkPOMDP.transition_log_probability()
ContinuousLightDarkPOMDPDiscreteActionsContinuousLightDarkPOMDPDiscreteActions.get_actions()ContinuousLightDarkPOMDPDiscreteActions.hash_action()ContinuousLightDarkPOMDPDiscreteActions.observation_log_probability()ContinuousLightDarkPOMDPDiscreteActions.observation_log_probability_per_state()ContinuousLightDarkPOMDPDiscreteActions.reward()ContinuousLightDarkPOMDPDiscreteActions.reward_batch()ContinuousLightDarkPOMDPDiscreteActions.sample_next_state()ContinuousLightDarkPOMDPDiscreteActions.sample_next_state_batch()ContinuousLightDarkPOMDPDiscreteActions.sample_observation()ContinuousLightDarkPOMDPDiscreteActions.simulate_random_rollout()ContinuousLightDarkPOMDPDiscreteActions.transition_log_probability()
ContinuousLightDarkPOMDPMetricsObservationModelTypeRewardModelType
- POMDPPlanners.environments.light_dark_pomdp.discrete_light_dark_pomdp module
DiscreteLightDarkPOMDPDiscreteLightDarkPOMDP.transition_error_probDiscreteLightDarkPOMDP.observation_error_probDiscreteLightDarkPOMDP.is_stochastic_rewardDiscreteLightDarkPOMDP.beaconsDiscreteLightDarkPOMDP.goal_stateDiscreteLightDarkPOMDP.start_stateDiscreteLightDarkPOMDP.obstaclesDiscreteLightDarkPOMDP.grid_sizeDiscreteLightDarkPOMDP.compute_metrics()DiscreteLightDarkPOMDP.get_metric_names()DiscreteLightDarkPOMDP.hash_action()DiscreteLightDarkPOMDP.is_terminal()DiscreteLightDarkPOMDP.observation_log_probability()DiscreteLightDarkPOMDP.observation_log_probability_per_state()DiscreteLightDarkPOMDP.reward()DiscreteLightDarkPOMDP.reward_batch()DiscreteLightDarkPOMDP.sample_next_state()DiscreteLightDarkPOMDP.sample_next_state_batch()DiscreteLightDarkPOMDP.sample_next_step()DiscreteLightDarkPOMDP.sample_observation()DiscreteLightDarkPOMDP.simulate_random_rollout()DiscreteLightDarkPOMDP.transition_log_probability()
DiscreteLightDarkPOMDPMetricsObservationModelType
- Subpackages
- POMDPPlanners.environments.mountain_car_pomdp package
MountainCarPOMDPMountainCarPOMDP.DEFAULT_STATE_TRANSITION_COVMountainCarPOMDP.cache_visualization()MountainCarPOMDP.compute_metrics()MountainCarPOMDP.get_actions()MountainCarPOMDP.get_metric_names()MountainCarPOMDP.hash_action()MountainCarPOMDP.initial_observation_dist()MountainCarPOMDP.initial_state_dist()MountainCarPOMDP.is_equal_observation()MountainCarPOMDP.is_terminal()MountainCarPOMDP.observation_log_probability()MountainCarPOMDP.observation_log_probability_per_state()MountainCarPOMDP.reward()MountainCarPOMDP.reward_batch()MountainCarPOMDP.sample_next_state()MountainCarPOMDP.sample_next_state_batch()MountainCarPOMDP.sample_observation()MountainCarPOMDP.simulate_random_rollout()MountainCarPOMDP.transition_log_probability()
MountainCarPOMDPMetrics- Submodules
- POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp module
MountainCarPOMDPMountainCarPOMDP.DEFAULT_STATE_TRANSITION_COVMountainCarPOMDP.cache_visualization()MountainCarPOMDP.compute_metrics()MountainCarPOMDP.get_actions()MountainCarPOMDP.get_metric_names()MountainCarPOMDP.hash_action()MountainCarPOMDP.initial_observation_dist()MountainCarPOMDP.initial_state_dist()MountainCarPOMDP.is_equal_observation()MountainCarPOMDP.is_terminal()MountainCarPOMDP.observation_log_probability()MountainCarPOMDP.observation_log_probability_per_state()MountainCarPOMDP.reward()MountainCarPOMDP.reward_batch()MountainCarPOMDP.sample_next_state()MountainCarPOMDP.sample_next_state_batch()MountainCarPOMDP.sample_observation()MountainCarPOMDP.simulate_random_rollout()MountainCarPOMDP.transition_log_probability()
MountainCarPOMDPMetrics
- POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp_beliefs module
MountainCarVectorizedUpdaterMountainCarVectorizedUpdater.state_transition_distMountainCarVectorizedUpdater.obs_distMountainCarVectorizedUpdater.powerMountainCarVectorizedUpdater.gravityMountainCarVectorizedUpdater.max_speedMountainCarVectorizedUpdater.min_positionMountainCarVectorizedUpdater.max_positionMountainCarVectorizedUpdater.batch_observation_log_likelihood()MountainCarVectorizedUpdater.batch_transition()MountainCarVectorizedUpdater.config_idMountainCarVectorizedUpdater.from_environment()
create_mountain_car_belief()
- POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp_gaussian_beliefs module
- POMDPPlanners.environments.pacman_pomdp package
PacManPOMDPPacManPOMDP.maze_sizePacManPOMDP.wallsPacManPOMDP.initial_pelletsPacManPOMDP.pellet_rewardPacManPOMDP.ghost_collision_penaltyPacManPOMDP.step_penaltyPacManPOMDP.win_rewardPacManPOMDP.ghost_aggressivenessPacManPOMDP.observation_noise_factorPacManPOMDP.max_observation_noisePacManPOMDP.dangerous_areasPacManPOMDP.dangerous_area_radiusPacManPOMDP.dangerous_area_penaltyPacManPOMDP.array_to_observation()PacManPOMDP.cache_visualization()PacManPOMDP.compute_metrics()PacManPOMDP.get_actions()PacManPOMDP.get_ghost_positions()PacManPOMDP.get_metric_names()PacManPOMDP.get_observation_cpp_ctor_kwargs()PacManPOMDP.get_pacman_pos()PacManPOMDP.get_pellets()PacManPOMDP.get_score()PacManPOMDP.get_terminal()PacManPOMDP.get_transition_cpp_ctor_kwargs()PacManPOMDP.hash_action()PacManPOMDP.initial_ghost_posPacManPOMDP.initial_observation_dist()PacManPOMDP.initial_state_dist()PacManPOMDP.is_equal_observation()PacManPOMDP.is_terminal()PacManPOMDP.make_state()PacManPOMDP.observation_log_probability()PacManPOMDP.observation_log_probability_per_state()PacManPOMDP.observation_to_array()PacManPOMDP.reward()PacManPOMDP.reward_batch()PacManPOMDP.sample_next_state()PacManPOMDP.sample_next_state_batch()PacManPOMDP.sample_observation()PacManPOMDP.simulate_random_rollout()PacManPOMDP.transition_log_probability()PacManPOMDP.visualize_path()
PacManVectorizedUpdaterPacManVectorizedUpdater.maze_sizePacManVectorizedUpdater.num_ghostsPacManVectorizedUpdater.num_pelletsPacManVectorizedUpdater.state_dimPacManVectorizedUpdater.ghost_aggressivenessPacManVectorizedUpdater.ghost_coordinationPacManVectorizedUpdater.ghost_strategiesPacManVectorizedUpdater.observation_noise_factorPacManVectorizedUpdater.max_observation_noisePacManVectorizedUpdater.batch_observation_log_likelihood()PacManVectorizedUpdater.batch_transition()PacManVectorizedUpdater.config_idPacManVectorizedUpdater.from_environment()
create_pacman_belief()create_simple_maze_pacman()- Subpackages
- POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_beliefs package
PacManVectorizedUpdatercreate_pacman_belief()- Submodules
- POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_beliefs.pacman_belief_factory module
- POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_beliefs.pacman_grid_utils module
- POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_beliefs.pacman_vectorized_updater module
- POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_utils package
- POMDPPlanners.environments.pacman_pomdp.pacman_pomdp_beliefs package
- Submodules
- POMDPPlanners.environments.pacman_pomdp.pacman_pomdp module
PacManPOMDPPacManPOMDP.maze_sizePacManPOMDP.wallsPacManPOMDP.initial_pelletsPacManPOMDP.pellet_rewardPacManPOMDP.ghost_collision_penaltyPacManPOMDP.step_penaltyPacManPOMDP.win_rewardPacManPOMDP.ghost_aggressivenessPacManPOMDP.observation_noise_factorPacManPOMDP.max_observation_noisePacManPOMDP.dangerous_areasPacManPOMDP.dangerous_area_radiusPacManPOMDP.dangerous_area_penaltyPacManPOMDP.array_to_observation()PacManPOMDP.cache_visualization()PacManPOMDP.compute_metrics()PacManPOMDP.get_actions()PacManPOMDP.get_ghost_positions()PacManPOMDP.get_metric_names()PacManPOMDP.get_observation_cpp_ctor_kwargs()PacManPOMDP.get_pacman_pos()PacManPOMDP.get_pellets()PacManPOMDP.get_score()PacManPOMDP.get_terminal()PacManPOMDP.get_transition_cpp_ctor_kwargs()PacManPOMDP.hash_action()PacManPOMDP.initial_ghost_posPacManPOMDP.initial_observation_dist()PacManPOMDP.initial_state_dist()PacManPOMDP.is_equal_observation()PacManPOMDP.is_terminal()PacManPOMDP.make_state()PacManPOMDP.observation_log_probability()PacManPOMDP.observation_log_probability_per_state()PacManPOMDP.observation_to_array()PacManPOMDP.reward()PacManPOMDP.reward_batch()PacManPOMDP.sample_next_state()PacManPOMDP.sample_next_state_batch()PacManPOMDP.sample_observation()PacManPOMDP.simulate_random_rollout()PacManPOMDP.transition_log_probability()PacManPOMDP.visualize_path()
PacManPOMDPMetricsRewardModelTypecreate_simple_maze_pacman()
- POMDPPlanners.environments.pacman_pomdp.pacman_visualizer module
- POMDPPlanners.environments.push_pomdp package
ContinuousPushPOMDPContinuousPushPOMDP.cache_visualization()ContinuousPushPOMDP.compute_metrics()ContinuousPushPOMDP.get_metric_names()ContinuousPushPOMDP.hash_action()ContinuousPushPOMDP.initial_observation_dist()ContinuousPushPOMDP.initial_state_dist()ContinuousPushPOMDP.is_equal_observation()ContinuousPushPOMDP.is_terminal()ContinuousPushPOMDP.observation_log_probability()ContinuousPushPOMDP.observation_log_probability_per_state()ContinuousPushPOMDP.reward()ContinuousPushPOMDP.reward_batch()ContinuousPushPOMDP.sample_next_state()ContinuousPushPOMDP.sample_next_state_batch()ContinuousPushPOMDP.sample_observation()ContinuousPushPOMDP.simulate_random_rollout()ContinuousPushPOMDP.transition_log_probability()
ContinuousPushPOMDPDiscreteActionsContinuousPushPOMDPDiscreteActions.get_actions()ContinuousPushPOMDPDiscreteActions.hash_action()ContinuousPushPOMDPDiscreteActions.observation_log_probability()ContinuousPushPOMDPDiscreteActions.observation_log_probability_per_state()ContinuousPushPOMDPDiscreteActions.reward()ContinuousPushPOMDPDiscreteActions.reward_batch()ContinuousPushPOMDPDiscreteActions.sample_next_state()ContinuousPushPOMDPDiscreteActions.sample_next_state_batch()ContinuousPushPOMDPDiscreteActions.sample_observation()ContinuousPushPOMDPDiscreteActions.transition_log_probability()
ContinuousPushPOMDPVisualizerPushPOMDPPushPOMDP.cache_visualization()PushPOMDP.compute_metrics()PushPOMDP.get_actions()PushPOMDP.get_metric_names()PushPOMDP.hash_action()PushPOMDP.hash_observation()PushPOMDP.initial_observation_dist()PushPOMDP.initial_state_dist()PushPOMDP.is_equal_observation()PushPOMDP.is_terminal()PushPOMDP.observation_log_probability()PushPOMDP.observation_log_probability_per_state()PushPOMDP.observation_log_probability_single()PushPOMDP.reward()PushPOMDP.reward_batch()PushPOMDP.sample_next_state()PushPOMDP.sample_next_state_batch()PushPOMDP.sample_next_step()PushPOMDP.sample_observation()PushPOMDP.simulate_random_rollout()PushPOMDP.transition_log_probability()
PushPOMDPVisualizer- Subpackages
- POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs package
ContinuousPushVectorizedUpdaterPushVectorizedUpdatercreate_continuous_push_belief()create_push_belief()- Submodules
- POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs.continuous_push_belief_factory module
- POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs.continuous_push_vectorized_updater module
- POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs.push_belief_factory module
- POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs.push_vectorized_updater module
- POMDPPlanners.environments.push_pomdp.push_pomdp_utils package
- POMDPPlanners.environments.push_pomdp.push_pomdp_beliefs package
- Submodules
- POMDPPlanners.environments.push_pomdp.continuous_push_geometry module
- POMDPPlanners.environments.push_pomdp.continuous_push_pomdp module
ContinuousPushPOMDPContinuousPushPOMDP.cache_visualization()ContinuousPushPOMDP.compute_metrics()ContinuousPushPOMDP.get_metric_names()ContinuousPushPOMDP.hash_action()ContinuousPushPOMDP.initial_observation_dist()ContinuousPushPOMDP.initial_state_dist()ContinuousPushPOMDP.is_equal_observation()ContinuousPushPOMDP.is_terminal()ContinuousPushPOMDP.observation_log_probability()ContinuousPushPOMDP.observation_log_probability_per_state()ContinuousPushPOMDP.reward()ContinuousPushPOMDP.reward_batch()ContinuousPushPOMDP.sample_next_state()ContinuousPushPOMDP.sample_next_state_batch()ContinuousPushPOMDP.sample_observation()ContinuousPushPOMDP.simulate_random_rollout()ContinuousPushPOMDP.transition_log_probability()
ContinuousPushPOMDPDiscreteActionsContinuousPushPOMDPDiscreteActions.get_actions()ContinuousPushPOMDPDiscreteActions.hash_action()ContinuousPushPOMDPDiscreteActions.observation_log_probability()ContinuousPushPOMDPDiscreteActions.observation_log_probability_per_state()ContinuousPushPOMDPDiscreteActions.reward()ContinuousPushPOMDPDiscreteActions.reward_batch()ContinuousPushPOMDPDiscreteActions.sample_next_state()ContinuousPushPOMDPDiscreteActions.sample_next_state_batch()ContinuousPushPOMDPDiscreteActions.sample_observation()ContinuousPushPOMDPDiscreteActions.transition_log_probability()
ContinuousPushPOMDPMetricsContinuousPushPOMDPMetrics.DANGEROUS_AREA_RATEContinuousPushPOMDPMetrics.GOAL_REACHING_RATEContinuousPushPOMDPMetrics.OBJECT_OBSTACLE_COLLISION_RATEContinuousPushPOMDPMetrics.ROBOT_OBSTACLE_COLLISION_RATEContinuousPushPOMDPMetrics.TOTAL_ALL_OBSTACLE_COLLISIONSContinuousPushPOMDPMetrics.TOTAL_DANGEROUS_AREA_STEPSContinuousPushPOMDPMetrics.TOTAL_OBJECT_OBSTACLE_COLLISIONSContinuousPushPOMDPMetrics.TOTAL_OBSTACLE_COLLISION_RATEContinuousPushPOMDPMetrics.TOTAL_ROBOT_OBSTACLE_COLLISIONS
- POMDPPlanners.environments.push_pomdp.continuous_push_pomdp_visualizer module
- POMDPPlanners.environments.push_pomdp.push_pomdp module
FixedStateDistributionPushPOMDPPushPOMDP.cache_visualization()PushPOMDP.compute_metrics()PushPOMDP.get_actions()PushPOMDP.get_metric_names()PushPOMDP.hash_action()PushPOMDP.hash_observation()PushPOMDP.initial_observation_dist()PushPOMDP.initial_state_dist()PushPOMDP.is_equal_observation()PushPOMDP.is_terminal()PushPOMDP.observation_log_probability()PushPOMDP.observation_log_probability_per_state()PushPOMDP.observation_log_probability_single()PushPOMDP.reward()PushPOMDP.reward_batch()PushPOMDP.sample_next_state()PushPOMDP.sample_next_state_batch()PushPOMDP.sample_next_step()PushPOMDP.sample_observation()PushPOMDP.simulate_random_rollout()PushPOMDP.transition_log_probability()
PushPOMDPMetricsPushPOMDPMetrics.DANGEROUS_AREA_RATEPushPOMDPMetrics.GOAL_REACHING_RATEPushPOMDPMetrics.OBJECT_OBSTACLE_COLLISION_RATEPushPOMDPMetrics.ROBOT_OBSTACLE_COLLISION_RATEPushPOMDPMetrics.TOTAL_ALL_OBSTACLE_COLLISIONSPushPOMDPMetrics.TOTAL_DANGEROUS_AREA_STEPSPushPOMDPMetrics.TOTAL_OBJECT_OBSTACLE_COLLISIONSPushPOMDPMetrics.TOTAL_OBSTACLE_COLLISION_RATEPushPOMDPMetrics.TOTAL_ROBOT_OBSTACLE_COLLISIONS
RandomInitialStateDistribution
- POMDPPlanners.environments.push_pomdp.push_pomdp_visualizer module
- POMDPPlanners.environments.rock_sample_pomdp package
RewardModelTypeRockSamplePOMDPRockSamplePOMDP.map_sizeRockSamplePOMDP.rock_positionsRockSamplePOMDP.init_posRockSamplePOMDP.sensor_efficiencyRockSamplePOMDP.bad_rock_penaltyRockSamplePOMDP.good_rock_rewardRockSamplePOMDP.step_penaltyRockSamplePOMDP.sensor_use_penaltyRockSamplePOMDP.exit_rewardRockSamplePOMDP.cache_visualization()RockSamplePOMDP.compute_metrics()RockSamplePOMDP.dangerous_areasRockSamplePOMDP.get_actions()RockSamplePOMDP.get_metric_names()RockSamplePOMDP.hash_action()RockSamplePOMDP.initial_observation_dist()RockSamplePOMDP.initial_state_dist()RockSamplePOMDP.is_equal_observation()RockSamplePOMDP.is_terminal()RockSamplePOMDP.observation_log_probability()RockSamplePOMDP.observation_log_probability_per_state()RockSamplePOMDP.reward()RockSamplePOMDP.reward_batch()RockSamplePOMDP.reward_modelRockSamplePOMDP.sample_next_state()RockSamplePOMDP.sample_next_state_batch()RockSamplePOMDP.sample_next_step()RockSamplePOMDP.sample_observation()RockSamplePOMDP.simulate_random_rollout()RockSamplePOMDP.transition_log_probability()RockSamplePOMDP.visualize_path()
RockSampleStateRockSampleVectorizedUpdaterRockSampleVectorizedUpdater.map_rowsRockSampleVectorizedUpdater.map_colsRockSampleVectorizedUpdater.num_rocksRockSampleVectorizedUpdater.rock_positionsRockSampleVectorizedUpdater.sensor_efficiencyRockSampleVectorizedUpdater.batch_observation_log_likelihood()RockSampleVectorizedUpdater.batch_transition()RockSampleVectorizedUpdater.config_idRockSampleVectorizedUpdater.from_environment()
RockSampleVisualizerRockSampleVisualizer.envRockSampleVisualizer.map_sizeRockSampleVisualizer.rock_positionsRockSampleVisualizer.action_namesRockSampleVisualizer.action_to_vectorRockSampleVisualizer.dangerous_areasRockSampleVisualizer.dangerous_area_radiusRockSampleVisualizer.create_visualization()RockSampleVisualizer.visualize_path()
create_random_rock_sample()create_rock_sample_state()create_rocksample_belief()get_robot_pos()get_rocks()states_equal()- Subpackages
- Submodules
- POMDPPlanners.environments.rock_sample_pomdp.rock_sample_pomdp module
RewardModelTypeRockSamplePOMDPRockSamplePOMDP.map_sizeRockSamplePOMDP.rock_positionsRockSamplePOMDP.init_posRockSamplePOMDP.sensor_efficiencyRockSamplePOMDP.bad_rock_penaltyRockSamplePOMDP.good_rock_rewardRockSamplePOMDP.step_penaltyRockSamplePOMDP.sensor_use_penaltyRockSamplePOMDP.exit_rewardRockSamplePOMDP.cache_visualization()RockSamplePOMDP.compute_metrics()RockSamplePOMDP.dangerous_areasRockSamplePOMDP.get_actions()RockSamplePOMDP.get_metric_names()RockSamplePOMDP.hash_action()RockSamplePOMDP.initial_observation_dist()RockSamplePOMDP.initial_state_dist()RockSamplePOMDP.is_equal_observation()RockSamplePOMDP.is_terminal()RockSamplePOMDP.observation_log_probability()RockSamplePOMDP.observation_log_probability_per_state()RockSamplePOMDP.reward()RockSamplePOMDP.reward_batch()RockSamplePOMDP.reward_modelRockSamplePOMDP.sample_next_state()RockSamplePOMDP.sample_next_state_batch()RockSamplePOMDP.sample_next_step()RockSamplePOMDP.sample_observation()RockSamplePOMDP.simulate_random_rollout()RockSamplePOMDP.transition_log_probability()RockSamplePOMDP.visualize_path()
RockSamplePOMDPMetricscreate_random_rock_sample()create_rock_sample_state()get_robot_pos()get_rocks()states_equal()
- POMDPPlanners.environments.rock_sample_pomdp.rock_sample_visualizer module
RockSampleVisualizerRockSampleVisualizer.envRockSampleVisualizer.map_sizeRockSampleVisualizer.rock_positionsRockSampleVisualizer.action_namesRockSampleVisualizer.action_to_vectorRockSampleVisualizer.dangerous_areasRockSampleVisualizer.dangerous_area_radiusRockSampleVisualizer.create_visualization()RockSampleVisualizer.visualize_path()
- POMDPPlanners.environments.safety_ant_velocity_pomdp package
SafeAntVelocityPOMDPSafeAntVelocityPOMDP.cache_visualization()SafeAntVelocityPOMDP.compute_metrics()SafeAntVelocityPOMDP.get_actions()SafeAntVelocityPOMDP.get_metric_names()SafeAntVelocityPOMDP.hash_action()SafeAntVelocityPOMDP.initial_observation_dist()SafeAntVelocityPOMDP.initial_state_dist()SafeAntVelocityPOMDP.is_equal_observation()SafeAntVelocityPOMDP.is_terminal()SafeAntVelocityPOMDP.observation_log_probability()SafeAntVelocityPOMDP.observation_log_probability_per_state()SafeAntVelocityPOMDP.reward()SafeAntVelocityPOMDP.reward_batch()SafeAntVelocityPOMDP.sample_next_state()SafeAntVelocityPOMDP.sample_next_state_batch()SafeAntVelocityPOMDP.sample_next_step()SafeAntVelocityPOMDP.sample_observation()SafeAntVelocityPOMDP.simulate_random_rollout()SafeAntVelocityPOMDP.transition_log_probability()
SafeAntVelocityVisualizer- Subpackages
- POMDPPlanners.environments.safety_ant_velocity_pomdp.safety_ant_velocity_pomdp_beliefs package
SafetyAntVelocityVectorizedUpdatercreate_safety_ant_velocity_belief()- Submodules
- POMDPPlanners.environments.safety_ant_velocity_pomdp.safety_ant_velocity_pomdp_beliefs.safety_ant_velocity_belief_factory module
- POMDPPlanners.environments.safety_ant_velocity_pomdp.safety_ant_velocity_pomdp_beliefs.safety_ant_velocity_vectorized_updater module
- POMDPPlanners.environments.safety_ant_velocity_pomdp.safety_ant_velocity_pomdp_beliefs package
- Submodules
- POMDPPlanners.environments.safety_ant_velocity_pomdp.safety_ant_velocity_pomdp module
SafeAntVelocityPOMDPSafeAntVelocityPOMDP.cache_visualization()SafeAntVelocityPOMDP.compute_metrics()SafeAntVelocityPOMDP.get_actions()SafeAntVelocityPOMDP.get_metric_names()SafeAntVelocityPOMDP.hash_action()SafeAntVelocityPOMDP.initial_observation_dist()SafeAntVelocityPOMDP.initial_state_dist()SafeAntVelocityPOMDP.is_equal_observation()SafeAntVelocityPOMDP.is_terminal()SafeAntVelocityPOMDP.observation_log_probability()SafeAntVelocityPOMDP.observation_log_probability_per_state()SafeAntVelocityPOMDP.reward()SafeAntVelocityPOMDP.reward_batch()SafeAntVelocityPOMDP.sample_next_state()SafeAntVelocityPOMDP.sample_next_state_batch()SafeAntVelocityPOMDP.sample_next_step()SafeAntVelocityPOMDP.sample_observation()SafeAntVelocityPOMDP.simulate_random_rollout()SafeAntVelocityPOMDP.transition_log_probability()
SafeAntVelocityPOMDPMetrics
- POMDPPlanners.environments.safety_ant_velocity_pomdp.safety_ant_velocity_visualizer module
Submodules
POMDPPlanners.environments.sanity_pomdp module
Sanity Check POMDP Environment Implementation.
This module implements a simple test environment used for debugging and sanity checking POMDP algorithms. The environment has deterministic dynamics and perfect observability, making it ideal for verifying algorithm correctness.
The Sanity POMDP features: - Two discrete states: 0 (good) and 1 (bad) - Two discrete actions: 0 (go to good state) and 1 (go to bad state) - Perfect observations: observation always equals the state - Simple reward structure: 1.0 for good state, 0.0 for bad state - No terminal states (infinite horizon)
This environment is primarily used for: - Testing POMDP algorithm implementations - Debugging belief updates and planning algorithms - Verifying that algorithms can solve trivial cases - Performance benchmarking baseline
- Classes:
SanityInitialStateDist: Always starts in good state SanityInitialObservationDist: Initial observation distribution SanityPOMDP: Main environment class for sanity testing
- class POMDPPlanners.environments.sanity_pomdp.SanityInitialObservationDist[source]
Bases:
DistributionInitial observation distribution for Sanity POMDP.
This distribution always returns observation 0 (corresponding to the good state) as the initial observation, maintaining consistency with the initial state distribution and perfect observability property.
Example
Using the initial observation distribution:
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> # Create initial observation distribution >>> initial_obs_dist = SanityInitialObservationDist() >>> >>> # Sample initial observation >>> initial_obs = initial_obs_dist.sample()[0] # Returns 0 >>> initial_obs == 0 True >>> >>> # Sample multiple observations >>> observations = initial_obs_dist.sample(n_samples=3) # Returns [0, 0, 0] >>> len(observations) == 3 True >>> all(obs == 0 for obs in observations) True >>> >>> # Check observation probabilities >>> prob = initial_obs_dist.probability([0]) # Returns [1.0] >>> bool(prob[0] == 1.0) True
- probability(values)[source]
Calculate probabilities for given values.
- Parameters:
values (
List[int]) – List of values to calculate probabilities for- Return type:
- Returns:
Numpy array of probabilities corresponding to input values
- Raises:
NotImplementedError – This method is not implemented by default. Subclasses should override if probability calculation is needed.
- class POMDPPlanners.environments.sanity_pomdp.SanityInitialStateDist[source]
Bases:
DistributionInitial state distribution for Sanity POMDP.
This distribution always returns state 0 (good state) as the initial state, providing a deterministic and predictable starting condition for testing.
Example
Using the initial state distribution:
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> # Create initial state distribution >>> initial_dist = SanityInitialStateDist() >>> >>> # Sample initial state (always returns good state) >>> initial_state = initial_dist.sample()[0] # Returns 0 >>> initial_state == 0 True >>> >>> # Sample multiple initial states >>> states = initial_dist.sample(n_samples=5) # Returns [0, 0, 0, 0, 0] >>> len(states) == 5 True >>> all(state == 0 for state in states) True >>> >>> # Check probability of initial states >>> prob_good = initial_dist.probability([0]) # Returns [1.0] >>> bool(prob_good[0] == 1.0) True >>> prob_bad = initial_dist.probability([1]) # Returns [0.0] >>> bool(prob_bad[0] == 0.0) True
- probability(values)[source]
Calculate probabilities for given values.
- Parameters:
values (
List[int]) – List of values to calculate probabilities for- Return type:
- Returns:
Numpy array of probabilities corresponding to input values
- Raises:
NotImplementedError – This method is not implemented by default. Subclasses should override if probability calculation is needed.
- class POMDPPlanners.environments.sanity_pomdp.SanityPOMDP(discount_factor=0.95, output_dir=None, debug=False, use_queue_logger=False)[source]
Bases:
DiscreteActionsEnvironmentSimple sanity check POMDP environment for testing and debugging.
This environment provides the simplest possible POMDP formulation with deterministic dynamics and perfect observability. It serves as a baseline for testing POMDP algorithms and ensuring correctness.
Problem Structure: - States: 0 (good), 1 (bad) - Actions: 0 (choose good), 1 (choose bad) - Observations: Same as states (perfect observability) - Rewards: 1.0 for good state, 0.0 for bad state - Dynamics: Deterministic state transitions based on action
The optimal policy is trivial: always choose action 0 to stay in the good state.
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = SanityPOMDP(discount_factor=0.95) >>> >>> # 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
- get_actions()[source]
Get all possible actions in the discrete action space.
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)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- initial_observation_dist()[source]
Get the initial observation distribution.
- Return type:
- Returns:
Distribution over initial observations
Note
Subclasses must implement this method to define initial observations.
- initial_state_dist()[source]
Get the initial state distribution.
- Return type:
- Returns:
Distribution over initial states
Note
Subclasses must implement this method to define the starting distribution.
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
- Parameters:
- Return type:
- 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)[source]
Check if a state is terminal.
- Parameters:
state (
int) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_observation(next_state, action, n_samples=1)[source]
Sample one or more observations for
(next_state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
POMDPPlanners.environments.tiger_pomdp module
Tiger POMDP Environment Implementation.
This module implements the classic Tiger problem, a benchmark POMDP environment where an agent must determine which of two doors conceals a treasure and which conceals a tiger, using only noisy acoustic observations.
The Tiger problem features: - Two doors (left and right) with a tiger behind one and treasure behind the other - Three actions: listen (to get information), open_left, open_right - Three observations: hear_left, hear_right, hear_nothing - Listening provides 85% accurate information about the tiger’s location - Opening the correct door yields +10 reward, opening wrong door yields -100 - Listening costs -1 per action
- Classes:
TigerStateTransition: State transition model for the Tiger problem TigerObservation: Observation model with noisy acoustic feedback TigerPOMDP: Main environment class implementing the Tiger problem
- class POMDPPlanners.environments.tiger_pomdp.TigerPOMDP(discount_factor, name='TigerPOMDP', output_dir=None, debug=False, use_queue_logger=False)[source]
Bases:
DiscreteActionsEnvironmentTiger POMDP environment implementation.
This is the classic Tiger problem where an agent must decide which door to open to find treasure while avoiding the tiger. The agent can listen for acoustic cues but receives noisy observations.
Problem Structure: - States: tiger_left, tiger_right (tiger’s location) - Actions: listen, open_left, open_right - Observations: hear_left, hear_right, hear_nothing - Rewards: listen(-1), correct_door(+10), wrong_door(-100)
- Parameters:
- states
List of possible states
- actions
List of possible actions
- observations
List of possible observations
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> tiger = TigerPOMDP(discount_factor=0.95) >>> >>> # Get initial state and actions >>> initial_state = tiger.initial_state_dist().sample()[0] >>> actions = tiger.get_actions() >>> >>> # Sample complete step using convenience method >>> action = actions[0] >>> next_state, observation, reward = tiger.sample_next_step(initial_state, action) >>> >>> # Check terminal condition >>> tiger.is_terminal(initial_state) False
- compute_metrics(histories)[source]
Compute Tiger POMDP specific metrics from simulation histories.
- Parameters:
- Return type:
- Returns:
List of MetricValue objects containing the computed metrics
- get_actions()[source]
Get all possible actions in the discrete action space.
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)[source]
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 thenp.array_equalsemantics used by the linear-scan fallback).
- initial_observation_dist()[source]
Get the initial observation distribution.
- Return type:
- Returns:
Distribution over initial observations
Note
Subclasses must implement this method to define initial observations.
- initial_state_dist()[source]
Get the initial state distribution.
- Return type:
- Returns:
Distribution over initial states
Note
Subclasses must implement this method to define the starting distribution.
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
- Parameters:
- Return type:
- 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)[source]
Check if a state is terminal.
- Parameters:
state (
str) – State to check for terminal condition- Return type:
- 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)[source]
Log-probability of each candidate observation under
(next_state, action).Returns
np.ndarrayof shape(N,)where N is the number of candidate observations. Subclasses must implement.
- observation_log_probability_per_state(next_states, action, observation)[source]
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 exposesbatch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.- Parameters:
- Return type:
- Returns:
ndarray of shape
(N,)with log-probabilities or log-PDFs.
- reward(state, action, next_state=None)[source]
Calculate the immediate reward for a state-action(-next_state) tuple.
next_stateis the realised post-transition state when known (e.g. threaded bysample_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 whenNone.- Parameters:
- Return type:
- Returns:
Immediate reward value.
Note
Subclasses must implement this method to define reward structure.
- reward_batch(states, action, next_states=None)[source]
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:
- Return type:
- Returns:
1-D array of reward values with shape
(N,).
- sample_next_state(state, action, n_samples=1)[source]
Sample one or more next states for
(state, action).Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.
- sample_next_state_batch(states, action)[source]
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 exposesbatch_sample(states_array)) should override to avoid the loop.- Parameters:
- Returns:
np.ndarrayof shape(N, *dim). For structured envs (Tiger strings, Pacman tuples): a list of length N.- Return type: