POMDPPlanners.environments.mountain_car_pomdp package
Mountain Car POMDP Environment Module.
This module provides the Mountain Car POMDP environment implementation and related components for hill-climbing tasks with noisy observations.
- Classes:
MountainCarPOMDP: Main Mountain Car environment with POMDP formulation MountainCarPOMDPMetrics: Metric names for Mountain Car POMDP environment
- class POMDPPlanners.environments.mountain_car_pomdp.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.mountain_car_pomdp.MountainCarPOMDPMetrics(*values)[source]
Bases:
EnumMetric names for Mountain Car POMDP environment.
- GOAL_REACHING_RATE = 'goal_reaching_rate'
Submodules
POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp module
Mountain Car POMDP Environment Implementation.
This module implements the classic Mountain Car problem as a POMDP, where an agent must drive an underpowered car up a steep mountain by building momentum through oscillating motion, with noisy observations of the car’s state.
The Mountain Car POMDP features: - Continuous 2D state space: [position, velocity] - Discrete action space: [-1 (reverse), 0 (neutral), 1 (forward)] - Noisy continuous observations of position and velocity - Physics-based dynamics with gravity and momentum - Sparse reward: 0 for reaching goal, -1 per time step otherwise
The key challenge is that the car’s engine is too weak to drive directly up the mountain, so the agent must learn to build momentum by first moving away from the goal.
- Classes:
MountainCarPOMDP: Main Mountain Car environment with POMDP formulation
- class POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp.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.
POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp_beliefs module
Vectorized particle belief updater for the Mountain Car POMDP.
This module implements a concrete
VectorizedParticleBeliefUpdater
that performs batched state transitions and observation log-likelihood
evaluations for the Mountain Car environment, replacing per-particle
Python loops with NumPy array operations.
- Classes:
MountainCarVectorizedUpdater: Batched updater for the Mountain Car POMDP.
- Functions:
create_mountain_car_belief: Factory producing a configured belief for MountainCarPOMDP.
- class POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp_beliefs.MountainCarVectorizedUpdater(state_transition_dist, obs_dist, power, gravity, max_speed, min_position, max_position)[source]
Bases:
VectorizedParticleBeliefUpdaterVectorized particle belief updater for the Mountain Car POMDP.
Performs all-particle transitions and observation log-likelihood evaluations using vectorized NumPy operations, replacing per-particle Python loops with batched array operations.
batch_transitionapplies the deterministic cart physics to all particles, then adds a per-particle Gaussian process-noise sample drawn fromstate_transition_dist(mirroring the native_native.MountainCarTransitionCppsamplepath), and finally re-applies the position/velocity clipping and wall-stop boundary rule. Observations follow a single Gaussian centred on the true state.- Parameters:
state_transition_dist (CovarianceParameterizedMultivariateNormal)
obs_dist (CovarianceParameterizedMultivariateNormal)
power (float)
gravity (float)
max_speed (float)
min_position (float)
max_position (float)
- state_transition_dist
Process-noise distribution added after the deterministic physics step.
- obs_dist
Observation noise distribution.
- power
Engine power scaling factor.
- gravity
Gravitational force constant.
- max_speed
Maximum velocity magnitude.
- min_position
Minimum position boundary.
- max_position
Maximum position boundary.
Example
>>> import numpy as np >>> np.random.seed(42) >>> from POMDPPlanners.environments.mountain_car_pomdp import MountainCarPOMDP >>> env = MountainCarPOMDP(discount_factor=0.99) >>> updater = MountainCarVectorizedUpdater.from_environment(env) >>> particles = np.column_stack([ ... np.random.uniform(-0.6, -0.4, 50), ... np.zeros(50), ... ]) >>> action = 1 >>> next_p = updater.batch_transition(particles, action) >>> next_p.shape (50, 2) >>> obs = np.array([-0.5, 0.0]) >>> ll = updater.batch_observation_log_likelihood(next_p, action, obs) >>> ll.shape (50,)
- batch_observation_log_likelihood(next_particles, action, observation)[source]
Compute observation log-likelihoods for all particles at once.
- batch_transition(particles, action)[source]
Transition all particles in a single batched operation.
- classmethod from_environment(env)[source]
Construct an updater from a MountainCarPOMDP instance.
- Parameters:
env (
MountainCarPOMDP) – Environment to extract parameters from.- Return type:
- Returns:
A new
MountainCarVectorizedUpdaterinstance.
- POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp_beliefs.create_mountain_car_belief(env, belief_type=BeliefType.VECTORIZED_PARTICLE, n_particles=200, **kwargs)[source]
Create a ready-to-use belief for the Mountain Car POMDP.
For
BeliefType.GAUSSIAN, the following keyword arguments are forwarded tocreate_mountain_car_gaussian_belief():updater_type(GaussianBeliefUpdaterType): defaults toGaussianBeliefUpdaterType.UKF.initial_covariance(np.ndarray): defaults tonp.diag([0.2**2 / 12, 1e-4]).process_noise_scale(float): defaults to1e-4.
- Parameters:
env (
MountainCarPOMDP) – MountainCarPOMDP environment instance.belief_type (
BeliefType) – Desired belief representation. Defaults toBeliefType.VECTORIZED_PARTICLE.n_particles (
int) – Number of particles (ignored for GAUSSIAN). Defaults to 200.**kwargs (
Any) – Extra arguments forwarded to the Gaussian factory.
- Return type:
- Returns:
A configured
Beliefobject.- Raises:
ValueError – If belief_type is not supported.
Example
>>> import numpy as np >>> np.random.seed(42) >>> from POMDPPlanners.environments.mountain_car_pomdp import MountainCarPOMDP >>> env = MountainCarPOMDP(discount_factor=0.99) >>> belief = create_mountain_car_belief(env, n_particles=50) >>> belief.sample().shape (2,)
POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp_gaussian_beliefs module
Factory for pre-configured Gaussian beliefs for the Mountain Car POMDP.
This module provides a single factory function that creates a
GaussianBelief instance
pre-configured for the
MountainCarPOMDP
environment, with an enum-based selector for the updater type (EKF or UKF).
The Mountain Car POMDP has nonlinear dynamics (velocity depends on
cos(3 * position)) with a linear-Gaussian observation model (identity
plus additive noise). Because the dynamics are nonlinear, a standard
linear Kalman filter is not applicable; only EKF (which requires analytical
Jacobians) and UKF (Jacobian-free sigma-point propagation) are supported.
- Classes:
GaussianBeliefUpdaterType: Enum selecting the Gaussian updater variant.
- Functions:
create_mountain_car_gaussian_belief: Factory producing a configured GaussianBelief.
- class POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp_gaussian_beliefs.GaussianBeliefUpdaterType(*values)[source]
Bases:
EnumSelector for the Gaussian belief updater variant.
- EKF
Extended Kalman filter (linearised via analytical Jacobians).
- UKF
Unscented Kalman filter (sigma-point propagation).
- EKF = 'ekf'
- UKF = 'ukf'
- POMDPPlanners.environments.mountain_car_pomdp.mountain_car_pomdp_gaussian_beliefs.create_mountain_car_gaussian_belief(env, updater_type, initial_covariance=None, process_noise_scale=0.0001)[source]
Create a GaussianBelief configured for a MountainCarPOMDP.
The Mountain Car POMDP has nonlinear dynamics:
v_{t+1} = clip(v_t + action * power + cos(3 * p_t) * (-gravity)) p_{t+1} = clip(p_t + v_{t+1}) z_t = [p_{t+1}, v_{t+1}] + w, w ~ N(0, R)
where R is
env.cov_matrix. A small process noise Q is added for numerical stability of the Kalman covariance updates.- Parameters:
env (
MountainCarPOMDP) – MountainCarPOMDP instance.updater_type (
GaussianBeliefUpdaterType) – Which Gaussian updater to use (EKF or UKF).initial_covariance (
Optional[ndarray]) – Initial belief covariance of shape (2, 2). Defaults tonp.diag([0.2**2 / 12, 1e-4])(variance of Uniform(-0.6, -0.4) for position, small for velocity).process_noise_scale (
float) – Diagonal scaling for the process noise covariance Q. Defaults to 1e-4.
- Return type:
- Returns:
A
GaussianBeliefwith the selected updater.
Example
>>> import numpy as np >>> from POMDPPlanners.environments.mountain_car_pomdp import MountainCarPOMDP >>> env = MountainCarPOMDP(discount_factor=0.99) >>> belief = create_mountain_car_gaussian_belief( ... env=env, ... updater_type=GaussianBeliefUpdaterType.EKF, ... ) >>> belief.mean.shape (2,)