POMDPPlanners.core.environment package

Public surface of the core.environment package.

Re-exports the same names that used to live in core/environment.py so that from POMDPPlanners.core.environment import Environment, SpaceInfo, ... continues to work after the package refactor. ConstrainedEnvironment is added here as the new sibling module’s public class.

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

Bases: Environment

Environment with a vector-valued constraint-cost channel.

Subclasses implement constraint_cost() to expose the per-transition cost vector that constrained planners (CPOMCPOW, C-PFT-DPW) read to compute Lagrangian Q-values and dual-ascent updates. Everything else on Environment — reward, transitions, observations, terminal handling — is inherited unchanged.

Note

This is an abstract base class and cannot be instantiated directly.

Parameters:
abstractmethod constraint_cost(state, action, next_state)[source]

Compute the per-transition constraint cost vector.

Parameters:

  • state (Any) – Current state.

  • action (Any) – Action executed from state.

  • next_state (Any) – Realised post-transition state.

Return type:

ndarray

Returns:

1-D array of shape (K,) where K is the number of constraint dimensions. For scalar constraints return a length-1 array; for chance-constrained planning with a binary failure indicator return np.array([1.0]) on unsafe transitions and np.array([0.0]) otherwise.

Note

Subclasses must implement this method.

constraint_cost_batch(states, action, next_states)[source]

Batched constraint cost for N transitions sharing one action.

Provides a loop-based default that subclasses with native / vectorized kernels can override.

Parameters:

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

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

  • next_states (Union[ndarray, Sequence[Any]]) – Sequence of N realised next states.

Return type:

ndarray

Returns:

2-D array of shape (N, K) where K is the number of constraint dimensions.

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

Bases: Environment

Abstract base class for POMDP environments with discrete action spaces.

This class extends the base Environment class with additional functionality specific to environments that have finite, enumerable action sets.

Note

This is an abstract base class and cannot be instantiated directly. Subclasses must implement all abstract methods from Environment plus the get_actions() method.

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

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.

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

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

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:
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:
class POMDPPlanners.core.environment.EnvironmentGenerator(name)[source]

Bases: ABC

Abstract base class for environment generators.

This class implements the factory pattern for creating environment instances. It’s useful for generating environments with randomized parameters or for creating multiple environment variants.

Note

This is an abstract base class and cannot be instantiated directly. Subclasses must implement the generate_environment() method.

Parameters:

name (str)

name

Identifier for the generator

abstractmethod generate_environment()[source]

Generate a new environment instance.

Return type:

Environment

Returns:

Newly created environment instance

Note

Subclasses must implement this method to define environment creation logic. This may involve randomization, parameter sampling, or deterministic generation.

class POMDPPlanners.core.environment.SpaceInfo(action_space, observation_space)[source]

Bases: object

Data class containing space type information for an environment.

This class encapsulates the space type classifications for both actions and observations in a POMDP environment.

Parameters:
action_space

The type of action space (discrete, continuous, or mixed)

observation_space

The type of observation space (discrete, continuous, or mixed)

Example

Creating space info for different environment types:

>>> # Discrete actions, continuous observations
>>> space_info = SpaceInfo(
...     action_space=SpaceType.DISCRETE,
...     observation_space=SpaceType.CONTINUOUS
... )
action_space: SpaceType
observation_space: SpaceType
class POMDPPlanners.core.environment.SpaceType(*values)[source]

Bases: Enum

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'

Submodules

POMDPPlanners.core.environment.constrained_environment module

Abstract base class for constrained POMDP environments.

Defines ConstrainedEnvironment, a subtype of POMDPPlanners.core.environment.Environment that adds a vector-valued constraint cost channel alongside the standard reward channel. Constrained planners (CPOMCPOW, C-PFT-DPW, future C-POMCP) read this method to compute Lagrangian Q-values and dual-ascent updates.

Why a separate name from reward: the word “cost” is overloaded in this repo. POMDPPlanners.core.cost defines belief_expectation_cost as the sign-flipped reward used by ICVaR-style planners. The constraint cost in a constrained POMDP is a different quantity — an inequality-constraint metric that is independent of reward. Naming the new method constraint_cost keeps the two unambiguous.

Why vector-valued: matches the standard constrained-POMDP definition (costs(s, a, s') -> R^K) so multi-constraint problems are first-class without an interface change. Scalar constraints return a length-1 array.

Standard constrained-POMDP definition: the CPOMDP tuple augments the POMDP with a vector-valued cost function C(s, a, s') -> R^K and a cost-budget vector; an optimal policy maximises expected discounted reward subject to a budget on each dimension of the expected discounted cost. This ABC exposes the cost-vector channel on the env; the budget lives on the constrained planner.

Classes:
ConstrainedEnvironment: ABC extending Environment with one

abstract method constraint_cost and one default-implemented batched helper constraint_cost_batch.

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

Bases: Environment

Environment with a vector-valued constraint-cost channel.

Subclasses implement constraint_cost() to expose the per-transition cost vector that constrained planners (CPOMCPOW, C-PFT-DPW) read to compute Lagrangian Q-values and dual-ascent updates. Everything else on Environment — reward, transitions, observations, terminal handling — is inherited unchanged.

Note

This is an abstract base class and cannot be instantiated directly.

Parameters:
abstractmethod constraint_cost(state, action, next_state)[source]

Compute the per-transition constraint cost vector.

Parameters:

  • state (Any) – Current state.

  • action (Any) – Action executed from state.

  • next_state (Any) – Realised post-transition state.

Return type:

ndarray

Returns:

1-D array of shape (K,) where K is the number of constraint dimensions. For scalar constraints return a length-1 array; for chance-constrained planning with a binary failure indicator return np.array([1.0]) on unsafe transitions and np.array([0.0]) otherwise.

Note

Subclasses must implement this method.

constraint_cost_batch(states, action, next_states)[source]

Batched constraint cost for N transitions sharing one action.

Provides a loop-based default that subclasses with native / vectorized kernels can override.

Parameters:

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

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

  • next_states (Union[ndarray, Sequence[Any]]) – Sequence of N realised next states.

Return type:

ndarray

Returns:

2-D array of shape (N, K) where K is the number of constraint dimensions.

POMDPPlanners.core.environment.environment module

Module for POMDP environment abstractions.

This module provides the foundational classes and interfaces for defining POMDP environments, including abstract base classes for state transitions, observation models, and reward functions.

Classes:

Environment: Abstract base class for POMDP environments DiscreteActionsEnvironment: Specialized for discrete action spaces ObservationModel: Abstract observation model interface StateTransitionModel: Abstract state transition interface EnvironmentGenerator: Factory pattern for environment creation SpaceType: Enumeration for action/observation space types SpaceInfo: Data class containing space type information

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

Bases: Environment

Abstract base class for POMDP environments with discrete action spaces.

This class extends the base Environment class with additional functionality specific to environments that have finite, enumerable action sets.

Note

This is an abstract base class and cannot be instantiated directly. Subclasses must implement all abstract methods from Environment plus the get_actions() method.

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

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.

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

class POMDPPlanners.core.environment.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

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:
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:
class POMDPPlanners.core.environment.environment.EnvironmentGenerator(name)[source]

Bases: ABC

Abstract base class for environment generators.

This class implements the factory pattern for creating environment instances. It’s useful for generating environments with randomized parameters or for creating multiple environment variants.

Note

This is an abstract base class and cannot be instantiated directly. Subclasses must implement the generate_environment() method.

Parameters:

name (str)

name

Identifier for the generator

abstractmethod generate_environment()[source]

Generate a new environment instance.

Return type:

Environment

Returns:

Newly created environment instance

Note

Subclasses must implement this method to define environment creation logic. This may involve randomization, parameter sampling, or deterministic generation.

class POMDPPlanners.core.environment.environment.SpaceInfo(action_space, observation_space)[source]

Bases: object

Data class containing space type information for an environment.

This class encapsulates the space type classifications for both actions and observations in a POMDP environment.

Parameters:
action_space

The type of action space (discrete, continuous, or mixed)

observation_space

The type of observation space (discrete, continuous, or mixed)

Example

Creating space info for different environment types:

>>> # Discrete actions, continuous observations
>>> space_info = SpaceInfo(
...     action_space=SpaceType.DISCRETE,
...     observation_space=SpaceType.CONTINUOUS
... )
action_space: SpaceType
observation_space: SpaceType
class POMDPPlanners.core.environment.environment.SpaceType(*values)[source]

Bases: Enum

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'