Environments

POMDP environments define the problem structure, including states, actions, observations, transitions, and rewards. POMDPPlanners provides both classic benchmark problems and modern challenging environments.

Core Environment Types

Classic Benchmark Problems

Control & Navigation

Manipulation

Information Gathering & Pursuit

Environment Interface

All environments inherit from the base Environment class:

class POMDPPlanners.core.environment.Environment(discount_factor, name, space_info, reward_range=None, output_dir=None, debug=False, use_queue_logger=False)[source]

Bases: ABC

Abstract base class for POMDP environments.

This is the core abstract class that all POMDP environments must inherit from. It defines the essential interface for POMDP environments including state transitions, observations, rewards, and terminal conditions.

Note

This is an abstract base class and cannot be instantiated directly. Subclasses must implement all abstract methods.

Parameters:

discount_factor (float)
name (str)
space_info (SpaceInfo)
reward_range (Tuple[float, float] | None)
output_dir (Path | None)
debug (bool)
use_queue_logger (bool)

discount_factor: Discount factor for future rewards

name: Environment identifier string

space_info: Information about action and observation space types

reward_range: Optional tuple containing (min_reward, max_reward)

output_dir: Optional directory for logging output

debug: Flag to enable debug logging

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.

Parameters:

history (List[StepData]) – List of step data from an episode
cache_path (Path) – Path where visualization data should be cached

Return type:

None

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

property config_id: str: Generate a deterministic identifier based on environment configuration.

Note

Uses custom serialization logic (not centralized serialize_value) to ensure: - Deterministic dict key ordering for consistent hashing - Compact format without __type__ markers - Recursive handling of nested objects Changing this serialization format would invalidate all cached results.

classmethod from_dict(data)[source]

Reconstruct environment from dictionary.

Dynamically imports the environment class and instantiates it with the saved parameters.

Parameters:

data (Dict[str, Any]) – Dictionary containing environment serialization data with keys: class, module, params, config_id

Return type:

Environment

Returns:

Reconstructed environment instance

Raises:

ImportError – If environment class cannot be imported
ValueError – If required data fields are missing
TypeError – If parameters are invalid for environment constructor

Example

>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> env = TigerPOMDP(discount_factor=0.95)
>>> env_dict = env.to_dict()
>>> reconstructed_env = Environment.from_dict(env_dict)
>>> reconstructed_env.discount_factor
0.95

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:: List[str]
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().

abstractmethod 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.

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:: Hashable
Raises:: NotImplementedError – If the observation is not hashable and the subclass has not provided an override. Subclasses with non-hashable observations (e.g. np.ndarray) MUST override.

abstractmethod initial_observation_dist()[source]

Get the initial observation distribution.

Return type:: Distribution
Returns:: Distribution over initial observations

Note

Subclasses must implement this method to define initial observations.

abstractmethod initial_state_dist()[source]

Get the initial state distribution.

Return type:: Distribution
Returns:: Distribution over initial states

Note

Subclasses must implement this method to define the starting distribution.

abstractmethod is_equal_observation(observation1, observation2)[source]

Check if two observations are equal.

Parameters:

observation1 (Any) – First observation to compare
observation2 (Any) – Second observation to compare

Return type:

bool

Returns:

True if observations are considered equal, False otherwise

Note

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

abstractmethod is_terminal(state)[source]

Check if a state is terminal.

Parameters:: state (Any) – 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.

property logger: Logger

Get logger instance for this environment.

The logger is implemented as a property to maintain pickle compatibility, as logger objects cannot be pickled directly.

Returns:: Configured logger instance with hierarchical naming

abstractmethod 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:

next_state (Any)
action (Any)
observations (Any)

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.

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

abstractmethod 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:

state (Any)
action (Any)
n_samples (int)

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:

Any

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 (Any) – Current state
action (Any) – Action to execute

Returns:

next_state: Sampled next state
next_observation: Sampled observation
reward: Immediate reward

Return type:

Tuple[Any, Any, float]

abstractmethod 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:

next_state (Any)
action (Any)
n_samples (int)

to_dict()[source]

Serialize environment to dictionary format.

Extracts environment class information and constructor parameters to enable JSON serialization and reconstruction.

Returns:

class: Full class path (module.ClassName)
module: Module name
params: Constructor parameters
config_id: Deterministic configuration identifier

Return type:

Dict[str, Any]

Example

>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> env = TigerPOMDP(discount_factor=0.95)
>>> env_dict = env.to_dict()
>>> 'class' in env_dict and 'params' in env_dict
True

Note

Uses centralized serialization system with registered SpaceInfo handler.

abstractmethod 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:

state (Any)
action (Any)
next_states (Any)

Space Types

Environments can have different action and observation space types:

class POMDPPlanners.core.environment.SpaceType(*values)[source]

Enumeration for categorizing action and observation spaces.

This enum is used to classify the mathematical structure of action and observation spaces in POMDP environments.

DISCRETE: Finite, countable spaces (e.g., {0, 1, 2, …})

CONTINUOUS: Real-valued continuous spaces (e.g., R^n)

MIXED: Combination of discrete and continuous elements

CONTINUOUS = 'continuous'

DISCRETE = 'discrete'

MIXED = 'mixed'

Key Environment Features

State Spaces

Discrete states (Tiger, Sanity)
Continuous states (CartPole, Mountain Car, Light-Dark)
Mixed representations

Action Spaces

Discrete actions (Tiger: listen/open doors)
Continuous actions (Light-Dark: movement vectors)
Hybrid approaches

Observation Models

Perfect observability (Sanity POMDP)
Noisy observations (Tiger: 85% accuracy)
Position-dependent noise (Light-Dark)
Sensor noise models (CartPole, Mountain Car)

Reward Structures

Goal-reaching rewards
Action costs
Safety penalties
Shaped rewards for learning

Creating Custom Environments

To create a custom environment, inherit from the base Environment class and implement the required methods:

from POMDPPlanners.core.environment import Environment
import numpy as np

class MyCustomPOMDP(Environment):
    def __init__(self, discount_factor=0.95):
        super().__init__(discount_factor)
        # Initialize your environment

    def get_states(self):
        # Return list of possible states
        return ["state1", "state2", "state3"]

    def get_actions(self):
        # Return list of possible actions
        return ["action1", "action2"]

    def initial_state_dist(self):
        # Return initial state distribution
        pass

    def state_transition_model(self, state, action):
        # Return state transition model
        pass

    def observation_model(self, next_state, action):
        # Return observation model
        pass

    def reward(self, state, action):
        # Return reward for state-action pair
        pass

    def is_terminal(self, state):
        # Return whether state is terminal
        return False

Environments

Core Environment Types

Environment Interface

Space Types

Key Environment Features

Creating Custom Environments

See Also