POMDPPlanners.environments.light_dark_pomdp package

Subpackages

Submodules

POMDPPlanners.environments.light_dark_pomdp.continuous_light_dark_pomdp module

Continuous Light-Dark POMDP Environment Implementation.

This module implements the continuous Light-Dark domain, a classic POMDP benchmark where an agent must navigate to a goal position in a continuous 2D space while dealing with position-dependent observation noise.

The Continuous Light-Dark POMDP features: - Continuous 2D state space representing agent position - Discrete or continuous action space for movement - Light source at a specific location that affects observation quality - Observation noise that decreases closer to the light source - Goal region that agent must reach to maximize reward - Optional obstacles that cause negative rewards when hit

Key characteristics: - State: [x, y] position in continuous 2D space - Actions: Movement vectors or discrete directions - Observations: Noisy position estimates (noise depends on distance from light) - Rewards: Goal reaching bonus, movement costs, obstacle penalties - Multiple reward model variants available

Classes:

RewardModelType: Enumeration of available reward model types ContinuousLightDarkPOMDP: Main environment class ContinuousLightDarkPOMDPDiscreteActions: Discrete action variant

class POMDPPlanners.environments.light_dark_pomdp.continuous_light_dark_pomdp.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: BaseLightDarkPOMDP

Continuous 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:
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.

Parameters:

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

Return type:

List[MetricValue]

Returns:

List of computed metrics with confidence intervals

get_metric_names()[source]

Get names of Continuous Light-Dark POMDP specific metrics.

Returns:

goal_reaching_rate, obstacle_hit_rate, avg_obstacle_hit_counter, out_of_grid_rate, and avg_high_variance_states_counter

Return type:

List[str]

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 the np.array_equal semantics used by the linear-scan fallback).

Parameters:

action (Any) – Action to hash.

Return type:

Hashable

Returns:

A hashable key derived from action.

is_terminal(state)[source]

Check if a state is terminal.

Parameters:

state (ndarray) – State to check for terminal condition

Return type:

bool

Returns:

True if the state is terminal, False otherwise

Note

Subclasses must implement this method to define terminal conditions.

observation_log_probability(next_state, action, observations)[source]

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

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

Return type:

ndarray

Parameters:
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 exposes batch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.

Parameters:

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

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

  • observation (Any) – A single observation.

Return type:

ndarray

Returns:

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

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 batched observation_log_probability() path on a singleton input.

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

Return type:

float

Parameters:

  • next_state (Any)

  • action (Any)

  • observation (Any)

reward(state, action, next_state=None)[source]

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

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

Parameters:

  • state (ndarray) – Current state.

  • action (ndarray) – Action executed from state.

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

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

reward_batch(states, action, next_states=None)[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:

ndarray

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 length n_samples (numeric envs return np.ndarray of shape (n_samples, *dim); structured envs return List[T]).

Return type:

ndarray

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 exposes batch_sample(states_array)) should override to avoid the loop.

Parameters:

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

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

Returns:

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

Return type:

ndarray

sample_observation(next_state, action, n_samples=1)[source]

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

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

Returns:

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

Return type:

Any

Parameters:
transition_log_probability(state, action, next_states)[source]

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

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

Return type:

ndarray

Parameters:
class POMDPPlanners.environments.light_dark_pomdp.continuous_light_dark_pomdp.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, DiscreteActionsEnvironment

Continuous 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:
get_actions()[source]

Get all possible actions in the discrete action space.

Return type:

List[Any]

Returns:

List containing all valid actions that can be executed

Note

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

hash_action(action)[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 the np.array_equal semantics used by the linear-scan fallback).

Parameters:

action (Any) – Action to hash.

Return type:

Hashable

Returns:

A hashable key derived from action.

observation_log_probability(next_state, action, observations)[source]

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

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

Return type:

ndarray

Parameters:
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 exposes batch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.

Parameters:

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

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

  • observation (Any) – A single observation.

Return type:

ndarray

Returns:

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

reward(state, action, next_state=None)[source]

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

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

Parameters:

  • state (ndarray) – Current state.

  • action (Any) – Action executed from state.

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

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

reward_batch(states, action, next_states=None)[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:

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

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

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

Return type:

ndarray

Returns:

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

sample_next_state(state, action, n_samples=1)[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 length n_samples (numeric envs return np.ndarray of shape (n_samples, *dim); structured envs return List[T]).

Return type:

ndarray

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 exposes batch_sample(states_array)) should override to avoid the loop.

Parameters:

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

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

Returns:

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

Return type:

ndarray

sample_observation(next_state, action, n_samples=1)[source]

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

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

Returns:

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

Return type:

Any

Parameters:
simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)[source]

Random rollout via native C++.

The variant-aware _native.simulate_rollout kernel 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 a sample() 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:

float

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.ndarray of shape (N,) where N is the number of candidate next states. Subclasses must implement.

Return type:

ndarray

Parameters:
class POMDPPlanners.environments.light_dark_pomdp.continuous_light_dark_pomdp.ContinuousLightDarkPOMDPMetrics(*values)[source]

Bases: Enum

Metric names for Continuous Light-Dark POMDP environment.

AVG_HIGH_VARIANCE_STATES_COUNTER = 'avg_high_variance_states_counter'
AVG_OBSTACLE_HIT_COUNTER = 'avg_obstacle_hit_counter'
GOAL_REACHING_RATE = 'goal_reaching_rate'
OBSTACLE_HIT_RATE = 'obstacle_hit_rate'
OUT_OF_GRID_RATE = 'out_of_grid_rate'
class POMDPPlanners.environments.light_dark_pomdp.continuous_light_dark_pomdp.ObservationModelType(*values)[source]

Bases: Enum

DISTANCE_BASED = 'distance_based'
NORMAL_NOISE = 'normal_noise'
NORMAL_NOISE_NO_OBS_IN_DARK = 'normal_noise_no_obs_in_dark'
class POMDPPlanners.environments.light_dark_pomdp.continuous_light_dark_pomdp.RewardModelType(*values)[source]

Bases: Enum

CONSTANT_HAZARD_PENALTY = 'constant_hazard_penalty'
DISTANCE_DECAYED_HAZARD_PENALTY = 'distance_decayed_hazard_penalty'
ZERO_MEAN_HAZARD_SHOCK = 'zero_mean_hazard_shock'

POMDPPlanners.environments.light_dark_pomdp.discrete_light_dark_pomdp module

class POMDPPlanners.environments.light_dark_pomdp.discrete_light_dark_pomdp.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, DiscreteActionsEnvironment

Discrete 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:
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.

Parameters:

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

Return type:

List[MetricValue]

Returns:

List of computed metrics with confidence intervals

get_metric_names()[source]

Get names of Discrete Light-Dark POMDP specific metrics.

Returns:

goal_reaching_rate, obstacle_hit_rate, avg_obstacle_hit_counter, out_of_grid_rate, and avg_high_variance_states_counter

Return type:

List[str]

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 the np.array_equal semantics used by the linear-scan fallback).

Parameters:

action (Any) – Action to hash.

Return type:

Hashable

Returns:

A hashable key derived from action.

is_terminal(state)[source]

Check if a state is terminal.

Parameters:

state (ndarray) – State to check for terminal condition

Return type:

bool

Returns:

True if the state is terminal, False otherwise

Note

Subclasses must implement this method to define terminal conditions.

observation_log_probability(next_state, action, observations)[source]

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

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

Return type:

ndarray

Parameters:
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 exposes batch_log_likelihood(next_states_array, observation_array)) should override to avoid the loop.

Parameters:

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

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

  • observation (ndarray) – A single observation.

Return type:

ndarray

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

Parameters:

  • state (ndarray) – Current state.

  • action (Any) – Action executed from state.

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

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

reward_batch(states, action, next_states=None)[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:

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

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

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

Return type:

ndarray

Returns:

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

sample_next_state(state, action, n_samples=1)[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 length n_samples (numeric envs return np.ndarray of shape (n_samples, *dim); structured envs return List[T]).

Return type:

Any

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 exposes batch_sample(states_array)) should override to avoid the loop.

Parameters:

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

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

Returns:

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

Return type:

ndarray

sample_next_step(state, action)[source]

Sample a complete state transition step.

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

Parameters:

  • state (ndarray) – Current state

  • action (Any) – Action to execute

Returns:

  • next_state: Sampled next state

  • next_observation: Sampled observation

  • reward: Immediate reward

Return type:

Tuple[Any, Any, float]

sample_observation(next_state, action, n_samples=1)[source]

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

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

Returns:

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

Return type:

Any

Parameters:
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_sampler interaction 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 a sample() method; used only for the Python fallback path. On the native path, action indices are pre-drawn via np.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:

float

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.ndarray of shape (N,) where N is the number of candidate next states. Subclasses must implement.

Return type:

ndarray

Parameters:
class POMDPPlanners.environments.light_dark_pomdp.discrete_light_dark_pomdp.DiscreteLightDarkPOMDPMetrics(*values)[source]

Bases: Enum

Metric names for Discrete Light-Dark POMDP environment.

AVG_OBSTACLE_HIT_COUNTER = 'avg_obstacle_hit_counter'
AVG_ZERO_MEAN_HAZARD_SHOCK_COUNTER = 'avg_high_variance_states_counter'
GOAL_REACHING_RATE = 'goal_reaching_rate'
OBSTACLE_HIT_RATE = 'obstacle_hit_rate'
OUT_OF_GRID_RATE = 'out_of_grid_rate'
class POMDPPlanners.environments.light_dark_pomdp.discrete_light_dark_pomdp.ObservationModelType(*values)[source]

Bases: Enum

DISTANCE_BASED = 'distance_based'
NORMAL = 'normal'
NO_OBS_IN_DARK = 'no_obs_in_dark'