POMDPPlanners.environments.cartpole_pomdp package

CartPole POMDP Environment Module.

This module provides the CartPole POMDP environment implementation and related components for pole-balancing tasks with noisy observations.

Classes:

CartPolePOMDP: Main CartPole environment with POMDP formulation CartPoleInitialStateDistribution: Initial state sampling distribution CartPolePOMDPMetrics: Metric names for CartPole POMDP environment

class POMDPPlanners.environments.cartpole_pomdp.CartPoleInitialObservationDistribution(noise_cov)

Bases: Distribution

Initial observation distribution: state prior convolved with obs noise.

Marginal ∫ p(o|s) p_0(s) ds produced by drawing a state from CartPoleInitialStateDistribution and adding zero-mean Gaussian noise with the env’s observation covariance.

Parameters:

noise_cov (ndarray[tuple[int, ...], dtype[floating[Any]]])

sample(n_samples=1)

Sample values from the distribution.

Parameters:

n_samples (int) – Number of samples to return. Defaults to 1.

Return type:

List[ndarray]

Returns:

List of n_samples independent samples from the distribution

Note

Subclasses must implement this method according to their specific distribution type and parameters.

class POMDPPlanners.environments.cartpole_pomdp.CartPoleInitialStateDistribution

Bases: Distribution

Initial state distribution for CartPole POMDP.

This distribution generates random initial states for the cart-pole system by sampling uniformly from a small range around the equilibrium position. All state variables (position, velocity, angle, angular velocity) are initialized close to zero with small random perturbations.

Example

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>> # Create initial state distribution
>>> initial_dist = CartPoleInitialStateDistribution()

>>> # Sample initial state
>>> initial_state = initial_dist.sample()[0]
>>> len(initial_state) == 4
True
>>> all(-0.05 <= x <= 0.05 for x in initial_state)  # Values in valid range
True

>>> # Sample multiple initial states
>>> states = initial_dist.sample(n_samples=3)
>>> len(states) == 3
True
>>> all(len(state) == 4 for state in states)
True

>>> # Each state has 4 components: [cart_pos, cart_vel, pole_angle, pole_ang_vel]
>>> position, velocity, angle, angular_velocity = initial_state
>>> isinstance(position, (int, float, np.floating))
True

sample(n_samples=1)

Sample values from the distribution.

Parameters:

n_samples (int) – Number of samples to return. Defaults to 1.

Return type:

List[ndarray]

Returns:

List of n_samples independent samples from the distribution

Note

Subclasses must implement this method according to their specific distribution type and parameters.

class POMDPPlanners.environments.cartpole_pomdp.CartPolePOMDP(discount_factor, noise_cov, state_transition_cov=None, name='CartPolePOMDP', output_dir=None, debug=False, use_queue_logger=False)

Bases: DiscreteActionsEnvironment

CartPole balancing task formulated as a POMDP.

This environment simulates the classic cart-pole balancing problem where an agent must apply left or right forces to keep a pole balanced on a moving cart. The challenge comes from noisy observations of the cart-pole state.

Problem Structure: - State: [cart_position, cart_velocity, pole_angle, pole_velocity] (continuous) - Actions: [left_force, right_force] (discrete) - Observations: Noisy state measurements (continuous) - Rewards: +1.0 per time step alive, 0.0 when terminated - Termination: Pole falls beyond angle threshold or cart moves too far

Example

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>>
>>> # Initialize environment
>>> noise_cov = np.diag([0.1, 0.1, 0.1, 0.1])
>>> env = CartPolePOMDP(discount_factor=0.99, noise_cov=noise_cov)
>>>
>>> # Get initial state and actions
>>> initial_state = env.initial_state_dist().sample()[0]
>>> actions = env.get_actions()
>>>
>>> # Sample complete step using convenience method
>>> action = actions[0]
>>> next_state, observation, reward = env.sample_next_step(initial_state, action)
>>>
>>> # Check terminal condition
>>> env.is_terminal(initial_state)
False

Parameters:

• discount_factor (float)

• noise_cov (ndarray[tuple[int, ...], dtype[floating[Any]]])

• state_transition_cov (ndarray[tuple[int, ...], dtype[floating[Any]]] | None)

• name (str)

• output_dir (Path | None)

• debug (bool)

• use_queue_logger (bool)

DEFAULT_STATE_TRANSITION_COV = array([[1.0e-04, 0.0e+00, 0.0e+00, 0.0e+00],        [0.0e+00, 1.0e-04, 0.0e+00, 0.0e+00],        [0.0e+00, 0.0e+00, 2.5e-05, 0.0e+00],        [0.0e+00, 0.0e+00, 0.0e+00, 1.0e-04]])

compute_metrics(histories)

Compute CartPole POMDP specific metrics from simulation histories.

Parameters:

histories (List[History]) – List of simulation histories

Return type:

List[MetricValue]

Returns:

List of MetricValue objects containing the computed metrics

get_actions()

Get all possible actions in the discrete action space.

Return type:

List[int]

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.

get_metric_names()

Get names of CartPole POMDP specific metrics.

Returns:

goal_reaching_rate

Return type:

List[str]

hash_action(action)

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.

initial_observation_dist()

Get the initial observation distribution.

Return type:

Distribution

Returns:

Distribution over initial observations

Note

Subclasses must implement this method to define initial observations.

initial_state_dist()

Get the initial state distribution.

Return type:

Distribution

Returns:

Distribution over initial states

Note

Subclasses must implement this method to define the starting distribution.

is_equal_observation(observation1, observation2)

Check if two observations are equal.

Parameters:

• observation1 (ndarray) – First observation to compare

• observation2 (ndarray) – 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.

is_terminal(state)

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)

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 (ndarray)

• action (int)

• observations (Sequence[Any] | ndarray)

observation_log_probability_per_state(next_states, action, observation)

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 (int) – 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)

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 (int) – 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)

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 (int) – 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)

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[tuple[int, ...], dtype[double]]

Parameters:

• state (ndarray)

• action (int)

• n_samples (int)

sample_next_state_batch(states, action)

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 (int) – 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)

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:

ndarray[tuple[int, ...], dtype[double]]

Parameters:

• next_state (ndarray)

• action (int)

• n_samples (int)

simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)

Random rollout via native C++.

Parameters:

• state (Any) – Current 4-D cart-pole state [x, x_dot, theta, theta_dot].

• action_sampler (Any) – Object with a sample() method (used only on the Python fallback path).

• max_depth (int) – Maximum rollout depth.

• discount_factor (float) – Per-step discount factor.

• depth (int) – Depth already consumed by the search tree. Defaults to 0.

Return type:

float

Returns:

Discounted sum of immediate rewards along the sampled trajectory.

transition_log_probability(state, action, next_states)

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 (ndarray)

• action (int)

• next_states (Sequence[Any] | ndarray)

class POMDPPlanners.environments.cartpole_pomdp.CartPolePOMDPMetrics(*values)

Bases: Enum

Metric names for CartPole POMDP environment.

GOAL_REACHING_RATE = 'goal_reaching_rate'

Submodules

POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp module

CartPole POMDP Environment Implementation.

This module implements a CartPole balancing task as a POMDP, where an agent must balance a pole on a cart using discrete left/right force actions, with noisy observations of the cart-pole state.

The CartPole POMDP features: - Continuous 4D state space: [cart_position, cart_velocity, pole_angle, pole_velocity] - Discrete binary action space: [left_force, right_force] - Noisy continuous observations of the state - Physics-based dynamics simulation - Episode termination when pole falls beyond threshold or cart moves too far

Classes:

CartPolePOMDP: Main CartPole environment with POMDP formulation

class POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp.CartPoleInitialObservationDistribution(noise_cov)

Bases: Distribution

Initial observation distribution: state prior convolved with obs noise.

Marginal ∫ p(o|s) p_0(s) ds produced by drawing a state from CartPoleInitialStateDistribution and adding zero-mean Gaussian noise with the env’s observation covariance.

Parameters:

noise_cov (ndarray[tuple[int, ...], dtype[floating[Any]]])

sample(n_samples=1)

Sample values from the distribution.

Parameters:

n_samples (int) – Number of samples to return. Defaults to 1.

Return type:

List[ndarray]

Returns:

List of n_samples independent samples from the distribution

Note

Subclasses must implement this method according to their specific distribution type and parameters.

class POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp.CartPoleInitialStateDistribution

Bases: Distribution

Initial state distribution for CartPole POMDP.

This distribution generates random initial states for the cart-pole system by sampling uniformly from a small range around the equilibrium position. All state variables (position, velocity, angle, angular velocity) are initialized close to zero with small random perturbations.

Example

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>> # Create initial state distribution
>>> initial_dist = CartPoleInitialStateDistribution()

>>> # Sample initial state
>>> initial_state = initial_dist.sample()[0]
>>> len(initial_state) == 4
True
>>> all(-0.05 <= x <= 0.05 for x in initial_state)  # Values in valid range
True

>>> # Sample multiple initial states
>>> states = initial_dist.sample(n_samples=3)
>>> len(states) == 3
True
>>> all(len(state) == 4 for state in states)
True

>>> # Each state has 4 components: [cart_pos, cart_vel, pole_angle, pole_ang_vel]
>>> position, velocity, angle, angular_velocity = initial_state
>>> isinstance(position, (int, float, np.floating))
True

sample(n_samples=1)

Sample values from the distribution.

Parameters:

n_samples (int) – Number of samples to return. Defaults to 1.

Return type:

List[ndarray]

Returns:

List of n_samples independent samples from the distribution

Note

Subclasses must implement this method according to their specific distribution type and parameters.

class POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp.CartPolePOMDP(discount_factor, noise_cov, state_transition_cov=None, name='CartPolePOMDP', output_dir=None, debug=False, use_queue_logger=False)

Bases: DiscreteActionsEnvironment

CartPole balancing task formulated as a POMDP.

This environment simulates the classic cart-pole balancing problem where an agent must apply left or right forces to keep a pole balanced on a moving cart. The challenge comes from noisy observations of the cart-pole state.

Problem Structure: - State: [cart_position, cart_velocity, pole_angle, pole_velocity] (continuous) - Actions: [left_force, right_force] (discrete) - Observations: Noisy state measurements (continuous) - Rewards: +1.0 per time step alive, 0.0 when terminated - Termination: Pole falls beyond angle threshold or cart moves too far

Example

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>>
>>> # Initialize environment
>>> noise_cov = np.diag([0.1, 0.1, 0.1, 0.1])
>>> env = CartPolePOMDP(discount_factor=0.99, noise_cov=noise_cov)
>>>
>>> # Get initial state and actions
>>> initial_state = env.initial_state_dist().sample()[0]
>>> actions = env.get_actions()
>>>
>>> # Sample complete step using convenience method
>>> action = actions[0]
>>> next_state, observation, reward = env.sample_next_step(initial_state, action)
>>>
>>> # Check terminal condition
>>> env.is_terminal(initial_state)
False

Parameters:

• discount_factor (float)

• noise_cov (ndarray[tuple[int, ...], dtype[floating[Any]]])

• state_transition_cov (ndarray[tuple[int, ...], dtype[floating[Any]]] | None)

• name (str)

• output_dir (Path | None)

• debug (bool)

• use_queue_logger (bool)

DEFAULT_STATE_TRANSITION_COV = array([[1.0e-04, 0.0e+00, 0.0e+00, 0.0e+00],        [0.0e+00, 1.0e-04, 0.0e+00, 0.0e+00],        [0.0e+00, 0.0e+00, 2.5e-05, 0.0e+00],        [0.0e+00, 0.0e+00, 0.0e+00, 1.0e-04]])

compute_metrics(histories)

Compute CartPole POMDP specific metrics from simulation histories.

Parameters:

histories (List[History]) – List of simulation histories

Return type:

List[MetricValue]

Returns:

List of MetricValue objects containing the computed metrics

get_actions()

Get all possible actions in the discrete action space.

Return type:

List[int]

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.

get_metric_names()

Get names of CartPole POMDP specific metrics.

Returns:

goal_reaching_rate

Return type:

List[str]

hash_action(action)

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.

initial_observation_dist()

Get the initial observation distribution.

Return type:

Distribution

Returns:

Distribution over initial observations

Note

Subclasses must implement this method to define initial observations.

initial_state_dist()

Get the initial state distribution.

Return type:

Distribution

Returns:

Distribution over initial states

Note

Subclasses must implement this method to define the starting distribution.

is_equal_observation(observation1, observation2)

Check if two observations are equal.

Parameters:

• observation1 (ndarray) – First observation to compare

• observation2 (ndarray) – 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.

is_terminal(state)

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)

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 (ndarray)

• action (int)

• observations (Sequence[Any] | ndarray)

observation_log_probability_per_state(next_states, action, observation)

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 (int) – 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)

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 (int) – 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)

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 (int) – 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)

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[tuple[int, ...], dtype[double]]

Parameters:

• state (ndarray)

• action (int)

• n_samples (int)

sample_next_state_batch(states, action)

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 (int) – 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)

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:

ndarray[tuple[int, ...], dtype[double]]

Parameters:

• next_state (ndarray)

• action (int)

• n_samples (int)

simulate_random_rollout(state, action_sampler, max_depth, discount_factor, depth=0)

Random rollout via native C++.

Parameters:

• state (Any) – Current 4-D cart-pole state [x, x_dot, theta, theta_dot].

• action_sampler (Any) – Object with a sample() method (used only on the Python fallback path).

• max_depth (int) – Maximum rollout depth.

• discount_factor (float) – Per-step discount factor.

• depth (int) – Depth already consumed by the search tree. Defaults to 0.

Return type:

float

Returns:

Discounted sum of immediate rewards along the sampled trajectory.

transition_log_probability(state, action, next_states)

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 (ndarray)

• action (int)

• next_states (Sequence[Any] | ndarray)

class POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp.CartPolePOMDPMetrics(*values)

Bases: Enum

Metric names for CartPole POMDP environment.

GOAL_REACHING_RATE = 'goal_reaching_rate'

POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp_beliefs module

Vectorized particle belief updater for the CartPole POMDP.

This module implements a concrete VectorizedParticleBeliefUpdater that performs batched state transitions and observation log-likelihood evaluations for the CartPole environment, replacing per-particle Python loops with NumPy array operations.

Classes:

CartPoleVectorizedUpdater: Batched updater for the CartPole POMDP.

Functions:

create_cartpole_belief: Factory producing a configured belief for CartPolePOMDP.

class POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp_beliefs.CartPoleVectorizedUpdater(state_transition_dist, obs_dist, force_mag, gravity, masscart, masspole, total_mass, length, polemass_length, tau, kinematics_integrator)

Bases: VectorizedParticleBeliefUpdater

Vectorized particle belief updater for the CartPole POMDP.

Performs all-particle transitions and observation log-likelihood evaluations using vectorized NumPy operations, replacing per-particle Python loops with batched array operations.

batch_transition applies the deterministic cart-pole physics to all particles and then adds a per-particle Gaussian process-noise sample drawn from state_transition_dist (mirroring CartPoleTransition.sample()). Observations follow a single Gaussian centred on the true state.

Parameters:

• state_transition_dist (CovarianceParameterizedMultivariateNormal)

• obs_dist (CovarianceParameterizedMultivariateNormal)

• force_mag (float)

• gravity (float)

• masscart (float)

• masspole (float)

• total_mass (float)

• length (float)

• polemass_length (float)

• tau (float)

• kinematics_integrator (str)

state_transition_dist

Process-noise distribution added after the deterministic physics step.

obs_dist

Observation noise distribution.

force_mag

Magnitude of force applied to the cart.

gravity

Gravitational acceleration constant.

masscart

Mass of the cart.

masspole

Mass of the pole.

total_mass

Combined mass of cart and pole.

length

Half the pole’s length.

polemass_length

Pole mass times pole half-length.

tau

Integration time step.

kinematics_integrator

Integration method (“euler” or “semi-implicit euler”).

Example

>>> import numpy as np
>>> np.random.seed(42)
>>> from POMDPPlanners.environments.cartpole_pomdp import CartPolePOMDP
>>> noise_cov = np.diag([0.1, 0.1, 0.1, 0.1])
>>> env = CartPolePOMDP(discount_factor=0.99, noise_cov=noise_cov)
>>> updater = CartPoleVectorizedUpdater.from_environment(env)
>>> particles = np.random.uniform(-0.05, 0.05, (50, 4))
>>> action = 1
>>> next_p = updater.batch_transition(particles, action)
>>> next_p.shape
(50, 4)
>>> obs = np.array([0.0, 0.0, 0.0, 0.0])
>>> ll = updater.batch_observation_log_likelihood(next_p, action, obs)
>>> ll.shape
(50,)

batch_observation_log_likelihood(next_particles, action, observation)

Compute observation log-likelihoods for all particles at once.

Parameters:

• next_particles (ndarray) – Transitioned particle states of shape (N, d).

• action (ndarray) – Action vector.

• observation (ndarray) – Observed value.

Return type:

ndarray

Returns:

Log-likelihoods of shape (N,).

batch_transition(particles, action)

Transition all particles in a single batched operation.

Parameters:

• particles (ndarray) – Current particle states of shape (N, d).

• action (ndarray) – Action vector.

Return type:

ndarray

Returns:

Next-state particles of shape (N, d).

property config_id: str

Return a deterministic identifier for this updater configuration.

classmethod from_environment(env)

Construct an updater from a CartPolePOMDP instance.

Parameters:

env (CartPolePOMDP) – Environment to extract parameters from.

Return type:

CartPoleVectorizedUpdater

Returns:

A new CartPoleVectorizedUpdater instance.

POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp_beliefs.create_cartpole_belief(env, belief_type=BeliefType.VECTORIZED_PARTICLE, n_particles=200, **kwargs)

Create a ready-to-use belief for the CartPole POMDP.

For BeliefType.GAUSSIAN, the following keyword arguments are forwarded to create_cartpole_gaussian_belief():

• updater_type (GaussianBeliefUpdaterType): defaults to GaussianBeliefUpdaterType.UKF.

• initial_covariance (np.ndarray): defaults to np.eye(4) * (0.1**2 / 12).

• process_noise_scale (float): defaults to 1e-4.

Parameters:

• env (CartPolePOMDP) – CartPolePOMDP environment instance.

• belief_type (BeliefType) – Desired belief representation. Defaults to BeliefType.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:

Belief

Returns:

A configured Belief object.

Raises:

ValueError – If belief_type is not supported.

Example

>>> import numpy as np
>>> np.random.seed(42)
>>> from POMDPPlanners.environments.cartpole_pomdp import CartPolePOMDP
>>> env = CartPolePOMDP(discount_factor=0.99,
...                    noise_cov=np.diag([0.1, 0.1, 0.1, 0.1]))
>>> belief = create_cartpole_belief(env, n_particles=50)
>>> belief.sample().shape
(4,)

POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp_gaussian_beliefs module

Factory for pre-configured Gaussian beliefs for the CartPole POMDP.

This module provides a single factory function that creates a GaussianBelief instance pre-configured for the CartPolePOMDP environment, with an enum-based selector for the updater type (EKF or UKF).

The CartPole POMDP has nonlinear dynamics (coupled cart-pole physics) 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_cartpole_gaussian_belief: Factory producing a configured GaussianBelief.

class POMDPPlanners.environments.cartpole_pomdp.cartpole_pomdp_gaussian_beliefs.GaussianBeliefUpdaterType(*values)

Bases: Enum

Selector 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.cartpole_pomdp.cartpole_pomdp_gaussian_beliefs.create_cartpole_gaussian_belief(env, updater_type, initial_covariance=None, process_noise_scale=0.0001)

Create a GaussianBelief configured for a CartPolePOMDP.

The CartPole POMDP has nonlinear dynamics:

x_{t+1} = f(x_t, u_t)      (deterministic cart-pole physics)
z_t     = x_{t+1} + v,      v ~ N(0, R)

where R is env.noise_cov. A small process noise Q is added for numerical stability of the Kalman covariance updates.

Parameters:

• env (CartPolePOMDP) – CartPolePOMDP instance.

• updater_type (GaussianBeliefUpdaterType) – Which Gaussian updater to use (EKF or UKF).

• initial_covariance (Optional[ndarray]) – Initial belief covariance of shape (4, 4). Defaults to np.eye(4) * (0.1**2 / 12) (variance of Uniform(-0.05, 0.05)).

• process_noise_scale (float) – Diagonal scaling for the process noise covariance Q. Defaults to 1e-4.

Return type:

GaussianBelief

Returns:

A GaussianBelief with the selected updater.

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.cartpole_pomdp import CartPolePOMDP
>>> noise_cov = np.diag([0.1, 0.1, 0.1, 0.1])
>>> env = CartPolePOMDP(discount_factor=0.99, noise_cov=noise_cov)
>>> belief = create_cartpole_gaussian_belief(
...     env=env,
...     updater_type=GaussianBeliefUpdaterType.EKF,
... )
>>> belief.mean.shape
(4,)