POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils package

Submodules

POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.base_light_dark_pomdp module

class POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.base_light_dark_pomdp.BaseLightDarkPOMDP(discount_factor, name, space_info, reward_range=None, beacons=[(0, 0), (0, 5), (0, 10), (5, 0), (5, 5), (5, 10), (10, 0), (10, 5), (10, 10)], goal_state=array([10, 5]), start_state=array([0, 5]), obstacles=[(3, 7), (5, 5)], obstacle_hit_probability=0.2, obstacle_reward=-10.0, obstacle_radius=1.0, goal_reward=10.0, beacon_radius=1.0, fuel_cost=2.0, grid_size=11)[source]

Bases: Environment, ABC

Parameters:
cache_visualization(history, cache_path)[source]

Cache visualization of agent’s path and belief.

Parameters:

  • history (List[StepData]) – List of step data from an episode.

  • cache_path (Path) – Path where to save the visualization.

Raises:

  • TypeError – If history is not a List or contains non-StepData objects, or if cache_path is not a Path object.

  • ValueError – If history is empty or contains invalid data.

Return type:

None

abstractmethod 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. This implementation ensures that the config_id is invariant to the order of beacons and obstacles.

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.

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.

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.

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.

is_state_near_beacon(state)[source]

Return True if state lies within beacon_radius of any beacon.

Used by tests and observation-model selection logic. Replaces the previous wrapper-attribute access pattern (model.near_beacon) with an env-level method that does not require the wrapper class.

Return type:

bool

Parameters:

state (ndarray)

abstractmethod is_terminal(state)[source]

Check if a state is terminal.

Parameters:

state (ndarray) – State to check for terminal condition

Return type:

bool

Returns:

True if the state is terminal, False otherwise

Note

Subclasses must implement this method to define terminal conditions.

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 (ndarray) – Current state.

  • action (Any) – Action executed from state.

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

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

visualize_path(path, agent_belief_path, actions, cache_path)[source]

Create and save an animated visualization of the agent’s path.

Parameters:

  • path (List[ndarray]) – List of state positions (2D numpy arrays) along the agent’s trajectory.

  • agent_belief_path (List[DiscreteDistribution]) – List of belief distributions at each step.

  • actions (List[str]) – List of actions taken at each step.

  • cache_path (Path) – Path where to save the visualization (must end with .gif).

Raises:

  • TypeError – If cache_path is not a Path object.

  • ValueError – If cache_path doesn’t end with .gif.

Return type:

None

class POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.base_light_dark_pomdp.BaseLightDarkPOMDPDiscreteActions(discount_factor, name, is_discrete_observations, reward_range=None, beacons=[(0, 0), (0, 5), (0, 10), (5, 0), (5, 5), (5, 10), (10, 0), (10, 5), (10, 10)], goal_state=array([10, 5]), start_state=array([0, 5]), obstacles=[(3, 7), (5, 5)], obstacle_hit_probability=0.2, obstacle_reward=-10.0, goal_reward=10.0, beacon_radius=1.0, fuel_cost=2.0, grid_size=11)[source]

Bases: BaseLightDarkPOMDP

Parameters:
get_actions()[source]
Return type:

List[Any]

POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.light_dark_reward_models module

class POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.light_dark_reward_models.BaseLightDarkRewardModel[source]

Bases: ABC

compute_reward(state, action, next_state=None)[source]
Return type:

float

Parameters:
compute_reward_batch(states, action, next_states=None)[source]
Return type:

ndarray

Parameters:
class POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.light_dark_reward_models.ContinuousLDZeroMeanHazardShockRewardModel(goal_state, obstacles, goal_state_radius, obstacle_radius, grid_size, obstacle_hit_probability, obstacle_reward, goal_reward, fuel_cost)[source]

Bases: ContinuousLightDarkRewardModel

Parameters:
class POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.light_dark_reward_models.ContinuousLightDarkDistanceDecayedHazardPenaltyRewardModel(goal_state, obstacles, goal_state_radius, obstacle_radius, grid_size, obstacle_hit_probability, obstacle_reward, goal_reward, fuel_cost, penalty_decay)[source]

Bases: BaseLightDarkRewardModel

Parameters:
compute_reward_batch(states, action, next_states=None)[source]
Return type:

ndarray

Parameters:
class POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.light_dark_reward_models.ContinuousLightDarkRewardModel(goal_state, obstacles, goal_state_radius, obstacle_radius, grid_size, obstacle_hit_probability, obstacle_reward, goal_reward, fuel_cost)[source]

Bases: BaseLightDarkRewardModel

Parameters:
compute_reward_batch(states, action, next_states=None)[source]
Return type:

ndarray

Parameters:

POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.light_dark_visualizer module

class POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.light_dark_visualizer.LightDarkPOMDPVisualizer(environment)[source]

Bases: object

Visualizer for Light-Dark POMDP environments.

Handles all visualization and animation logic for Light-Dark POMDP environments, including path visualization, belief particle rendering, and animation generation.

Parameters:

environment (Any)

environment

The Light-Dark POMDP environment instance to visualize.

cache_visualization(history, cache_path)[source]

Cache visualization of agent’s path and belief.

Parameters:

  • history (List[StepData]) – List of step data from an episode.

  • cache_path (Path) – Path where to save the visualization.

Raises:

  • TypeError – If history is not a List or contains non-StepData objects, or if cache_path is not a Path object.

  • ValueError – If history is empty or contains invalid data.

Return type:

None

visualize_path(path, agent_belief_path, actions, cache_path)[source]

Create and save an animated visualization of the agent’s path.

Parameters:

  • path (List[ndarray]) – List of state positions (2D numpy arrays) along the agent’s trajectory.

  • agent_belief_path (List[DiscreteDistribution]) – List of belief distributions at each step.

  • actions (List[str]) – List of actions taken at each step.

  • cache_path (Path) – Path where to save the visualization (must end with .gif).

Raises:

  • TypeError – If cache_path is not a Path object.

  • ValueError – If cache_path doesn’t end with .gif.

POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.numba_kernels module

Light-Dark-specific Numba-JIT kernels.

Holds kernels whose signatures or logic hardcode light-dark concepts (goal+obstacles+out-of-grid shape, standard / high-variance-states / decaying- hit-probability reward formulas). Generic geometric / numerical primitives used here also by other envs live in POMDPPlanners.utils.numba_kernels instead.

Conventions match the shared module: contiguous float64 arrays, scalar floats/ints/bools, no Python objects, all RNG draws happen in the caller and are passed in as parameters.

Public kernels

POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.numba_kernels.compute_reward_base_batch_kernel(next_states, goal_state, obstacles, goal_state_radius, obstacle_radius, grid_size, fuel_cost, goal_reward, obstacle_reward)[source]

Batched form of compute_reward_base_kernel().

next_states is shape (N, 2) and contains the realised post-transition positions threaded by the Python caller (or the deterministic states + action fallback when no realised batch is available). Returns (rewards, obstacle_mask) both of length N. rewards already includes fuel, goal-distance, goal bonus, and the out-of-grid penalty. The Python caller must add the stochastic obstacle-hit contribution at indices where obstacle_mask is True, drawing its own RNG so seeded tests stay bit-identical to the per-state path.

Return type:

Tuple[ndarray, ndarray]

Parameters:
POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.numba_kernels.compute_reward_base_kernel(next_state, goal_state, obstacles, goal_state_radius, obstacle_radius, grid_size, fuel_cost, goal_reward, obstacle_reward)[source]

Return (base_reward, is_obstacle_hit_region).

base_reward already includes fuel, goal-distance, and the out-of-grid penalty. The caller (Python) must add the stochastic obstacle-hit contribution when is_obstacle_hit_region is True, using its own np.random.rand() draw so seeded tests stay bit-identical. Used by Standard and ZeroMeanHazardShock reward models.

next_state is the realised post-transition position. The Python caller threads either the draw from Environment.sample_next_step() or the deterministic state + action fallback when no realised draw is available.

Return type:

Tuple[float, bool]

Parameters:
POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.numba_kernels.compute_reward_decaying_hit_prob_kernel(next_state, goal_state, obstacles, goal_state_radius, grid_size, fuel_cost, goal_reward, obstacle_reward, penalty_decay, uniform)[source]

Full reward for the Decaying-Hit-Probability model.

next_state is the realised post-transition position threaded by the Python caller. The caller draws a single uniform ~ U[0,1) (via np.random.rand()) and passes it in — this matches the pre-refactor RNG call pattern, which unconditionally drew one uniform in _obstacle_reward.

Return type:

float

Parameters:
POMDPPlanners.environments.light_dark_pomdp.light_dark_pomdp_utils.numba_kernels.is_terminal_kernel(state, goal_state, obstacles, goal_state_radius, obstacle_radius, is_obstacle_hit_terminal)[source]
Return type:

bool

Parameters: