POMDPPlanners.environments.laser_tag_pomdp package
LaserTag POMDP Environment Package.
This package implements the LaserTag pursuit-evasion POMDP environment in both discrete-grid and continuous-space variants.
Note
LaserTagState is now represented as numpy arrays with shape (5,). See laser_tag_pomdp.py for state vector structure documentation.
- class POMDPPlanners.environments.laser_tag_pomdp.ContinuousLaserTagPOMDP(discount_factor, name='ContinuousLaserTagPOMDP', grid_size=(11.0, 7.0), walls=None, robot_radius=0.3, opponent_radius=0.3, tag_radius=0.5, tag_reward=10.0, tag_penalty=10.0, step_cost=1.0, measurement_noise=1.0, robot_transition_cov_matrix=array([[0.1, 0.], [0., 0.1]]), opponent_transition_cov_matrix=array([[0.05, 0.], [0., 0.05]]), evasion_speed=0.6, dangerous_areas=None, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, dangerous_area_hit_probability=1.0, output_dir=None, debug=False, use_queue_logger=False, initial_state=None, opponent_policy=OpponentPolicy.EVADE)[source]
Bases:
EnvironmentContinuous LaserTag POMDP with continuous
[dx, dy, tag_flag]actions.A pursuit-evasion problem in continuous 2-D space where a robot must navigate to tag an opponent. The robot receives noisy 8-direction laser range observations.
- Stochasticity:
The dangerous-area penalty can be applied either deterministically (the default) or stochastically. When
dangerous_area_hit_probability == 1.0(default), the kernel’s deterministic deduction is preserved verbatim, matching legacy behavior. Whendangerous_area_hit_probability < 1.0, the accumulated dangerous-area deduction is applied to the reward only with that probability perreward()call, producing a heavy-tailed return distribution suitable for benchmarking risk-sensitive planners (e.g. ICVaR-aware MCTS) against expected-value MCTS on the same env. Note that this makesreward(state, action)non-deterministic given a state-action pair, so any external caching that assumes deterministic rewards must be aware of this.transition_log_probabilityis unaffected.
Example
>>> import numpy as np >>> np.random.seed(42) >>> >>> # Initialize environment >>> env = ContinuousLaserTagPOMDP(discount_factor=0.95) >>> >>> # Get initial state >>> initial_state = env.initial_state_dist().sample()[0] >>> >>> # Sample complete step >>> action = np.array([1.0, 0.0, 0.0]) >>> next_state, observation, reward = env.sample_next_step(initial_state, action) >>> >>> # Check terminal condition >>> env.is_terminal(initial_state) False
Example
Risk-sensitive evaluation – a 10%-tail-risk environment suitable for benchmarking ICVaR-aware planners against expected-value MCTS:
>>> env = ContinuousLaserTagPOMDP( ... discount_factor=0.95, ... dangerous_area_penalty=150.0, ... dangerous_area_hit_probability=0.1, ... )
- Parameters:
discount_factor (float)
name (str)
robot_radius (float)
opponent_radius (float)
tag_radius (float)
tag_reward (float)
tag_penalty (float)
step_cost (float)
measurement_noise (float)
robot_transition_cov_matrix (np.ndarray)
opponent_transition_cov_matrix (np.ndarray)
evasion_speed (float)
dangerous_area_radius (float)
dangerous_area_penalty (float)
dangerous_area_hit_probability (float)
output_dir (Optional[Path])
debug (bool)
use_queue_logger (bool)
initial_state (Optional[np.ndarray])
opponent_policy (OpponentPolicy)
- 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 environment-specific metrics from episode histories.
This method can be overridden by subclasses to provide custom metric calculations beyond standard return and episode length.
- 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:
- 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().
- 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).
- 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:
- 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:
- 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 (
ndarray) – 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.
- 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 batchedobservation_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.
- 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.
- 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 dispatched to native C++ via
cont_simulate_rollout.Pre-samples actions from
action_sampler, packs them into a(N, 3)buffer, and runs the full discounted-return loop inside C++. Results are numerically identical to theEnvironment.simulate_random_rollout()Python fallback.When
dangerous_area_hit_probability < 1.0, falls back to the Python rollout: the native kernel applies the dangerous-area penalty deterministically per step, which contradicts the stochastic semantics; routing through Pythonreward()keeps the per-step Bernoulli intact.Also falls back when
dangerous_areasis non-empty: the C++cont_simulate_rolloutkernel scores the danger penalty against the pre-transition robot position, while the Pythonreward()path (post-fix) consumes the realised post-transition position. Until the C++ kernel is rebuilt this is the only correctness-preserving path for configs with danger areas.
- class POMDPPlanners.environments.laser_tag_pomdp.ContinuousLaserTagPOMDPDiscreteActions(discount_factor, name='ContinuousLaserTagPOMDPDiscreteActions', grid_size=(11.0, 7.0), walls=None, robot_radius=0.3, opponent_radius=0.3, tag_radius=0.5, tag_reward=10.0, tag_penalty=10.0, step_cost=1.0, measurement_noise=1.0, robot_transition_cov_matrix=array([[0.1, 0.], [0., 0.1]]), opponent_transition_cov_matrix=array([[0.05, 0.], [0., 0.05]]), evasion_speed=0.6, dangerous_areas=None, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, dangerous_area_hit_probability=1.0, output_dir=None, debug=False, use_queue_logger=False, initial_state=None, opponent_policy=OpponentPolicy.EVADE)[source]
Bases:
ContinuousLaserTagPOMDP,DiscreteActionsEnvironmentContinuous LaserTag POMDP with discrete string actions.
Actions:
"up","down","right","left","tag".Example
>>> import numpy as np >>> np.random.seed(42) >>> >>> env = ContinuousLaserTagPOMDPDiscreteActions(discount_factor=0.95) >>> >>> initial_state = env.initial_state_dist().sample()[0] >>> actions = env.get_actions() >>> >>> action = actions[0] >>> next_state, observation, reward = env.sample_next_step(initial_state, action) >>> >>> env.is_terminal(initial_state) False
- Parameters:
discount_factor (float)
name (str)
robot_radius (float)
opponent_radius (float)
tag_radius (float)
tag_reward (float)
tag_penalty (float)
step_cost (float)
measurement_noise (float)
robot_transition_cov_matrix (np.ndarray)
opponent_transition_cov_matrix (np.ndarray)
evasion_speed (float)
dangerous_area_radius (float)
dangerous_area_penalty (float)
dangerous_area_hit_probability (float)
output_dir (Optional[Path])
debug (bool)
use_queue_logger (bool)
initial_state (Optional[np.ndarray])
opponent_policy (OpponentPolicy)
- 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).
- 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.
- 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 batchedobservation_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.
- 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.
- 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 dispatched to native C++ via
cont_simulate_rollout.Pre-samples actions from
action_sampler, packs them into a(N, 3)buffer, and runs the full discounted-return loop inside C++. Results are numerically identical to theEnvironment.simulate_random_rollout()Python fallback.When
dangerous_area_hit_probability < 1.0, falls back to the Python rollout: the native kernel applies the dangerous-area penalty deterministically per step, which contradicts the stochastic semantics; routing through Pythonreward()keeps the per-step Bernoulli intact.Also falls back when
dangerous_areasis non-empty: the C++cont_simulate_rolloutkernel scores the danger penalty against the pre-transition robot position, while the Pythonreward()path (post-fix) consumes the realised post-transition position. Until the C++ kernel is rebuilt this is the only correctness-preserving path for configs with danger areas.
- class POMDPPlanners.environments.laser_tag_pomdp.LaserTagPOMDP(discount_factor, name='LaserTagPOMDP', floor_shape=(11, 7), walls={(1, 2), (3, 0), (3, 4), (5, 0), (6, 4), (9, 1), (9, 4), (10, 6)}, tag_reward=10.0, tag_penalty=10.0, step_cost=1.0, measurement_noise=1.0, dangerous_areas={(2, 5), (5, 3), (7, 1)}, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, output_dir=None, debug=False, use_queue_logger=False, initial_state=None, transition_error_prob=0.0, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0, opponent_policy=OpponentPolicy.EVADE)[source]
Bases:
DiscreteActionsEnvironmentLaserTag POMDP environment implementation.
This is a pursuit-evasion problem where a robot must navigate a grid to tag an opponent. The robot receives noisy observations of the opponent’s position and must decide when and where to attempt tagging.
Problem Structure: - States: numpy array [robot_row, robot_col, opp_row, opp_col, terminal] - Actions: North(0), South(1), East(2), West(3), Tag(4) - Observations: 8-directional laser measurements (N,NE,E,SE,S,SW,W,NW) - Rewards: Tag success(+10), Tag failure(-10), Movement(-1)
- Parameters:
discount_factor (float)
name (str)
tag_reward (float)
tag_penalty (float)
step_cost (float)
measurement_noise (float)
dangerous_area_radius (float)
dangerous_area_penalty (float)
output_dir (Path | None)
debug (bool)
use_queue_logger (bool)
initial_state (ndarray | None)
transition_error_prob (float)
reward_model_type (RewardModelType)
penalty_decay (float)
opponent_policy (OpponentPolicy)
- floor_shape
Grid dimensions as (rows, cols)
- walls
Set of wall positions as (row, col) tuples
- tag_reward
Reward for successful tagging
- tag_penalty
Penalty for unsuccessful tagging
- step_cost
Cost per movement action
- measurement_noise
Standard deviation of observation noise
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = LaserTagPOMDP(discount_factor=0.95) >>> >>> # 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
- cache_visualization(history, cache_path)[source]
Cache visualization of the LaserTag episode as an animated GIF.
Creates an animated visualization showing: - Robot movement (red circle) - Opponent movement (blue circle) - Walls (black squares) - Dangerous areas (red circles) - Action arrows showing robot’s intended movement - Laser measurements (green rays from robot position) - Belief particles (if available) showing robot’s belief about opponent location - Grid boundaries and coordinate system
- Parameters:
- Raises:
ValueError – If history is empty or contains invalid data
TypeError – If cache_path is not a Path object or doesn’t end with .gif
- Return type:
- compute_metrics(histories)[source]
Compute LaserTag POMDP specific metrics from simulation histories.
- Return type:
- Parameters:
- 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).
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
Observations are 8-dimensional laser measurements or terminal observations.
- 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 transition.
The wall / dangerous-area penalty is computed against the realised post-action robot position taken from
next_state. When the caller omitsnext_state(e.g., the open-loop scalar API path) the method resamples a transition viasample_next_state()so the penalty is always scored against an actual draw from the transition kernel — never against the open-loopstate + action_vectorintended position.Environment.sample_next_step()threads its samplednext_stateinto this method so trajectory and reward agree on the same realisation.
- reward_batch(states, action, next_states=None)[source]
Vectorised reward for a batch of states under a single action.
When
next_statesis supplied the danger-area / wall penalty is evaluated against the realised positions innext_states[:, :2](matching the contract honoured byEnvironment.sample_next_step()). When it isNonethe method resamples viasample_next_state_batch()whenever penalty terms exist, then delegates to the reward model so reward and trajectory remain consistent end-to-end.
- 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.
- 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.
- class POMDPPlanners.environments.laser_tag_pomdp.OpponentPolicy(*values)[source]
Bases:
EnumOpponent transition behaviour selectable on the LaserTag environments.
Three policies are offered:
EVADE(default): the opponent flees the robot, placing its directional probability mass on the cell that increases distance, and reacts to the robot’s current (pre-move) position. This matches JuliaPOMDP/LaserTag.jl.PURSUE: the opponent chases the robot, placing its directional mass on the cell that decreases distance, and reacts to the robot’s post-move position. This restores the behaviour used before the evader alignment fix.EVADE_WHEN_SPOTTED: a partially-observed reactive opponent. When the robot has a clear line of sight to the opponent (the opponent lies on one of the robot’s unoccluded laser rays, evaluated from the robot’s pre-move position), it behaves exactly likeEVADE. Otherwise the unspotted behaviour is environment-specific: the discrete grid env moves randomly (uniformly over the moves, with the usual stay/wall handling), while the continuous env holds its position (only the Gaussian opponent noise jitters it). The opponent is memoryless — visibility is recomputed each step from the current state.
EVADEandPURSUEcouple both the directional choice and the reference-position choice, so they are mutually exclusive opposites.- EVADE = 'evade'
- EVADE_WHEN_SPOTTED = 'evade_when_spotted'
- PURSUE = 'pursue'
Subpackages
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs package
ContinuousLaserTagVectorizedUpdaterContinuousLaserTagVectorizedUpdater.wallsContinuousLaserTagVectorizedUpdater.grid_sizeContinuousLaserTagVectorizedUpdater.robot_radiusContinuousLaserTagVectorizedUpdater.opponent_radiusContinuousLaserTagVectorizedUpdater.tag_radiusContinuousLaserTagVectorizedUpdater.evasion_speedContinuousLaserTagVectorizedUpdater.measurement_noiseContinuousLaserTagVectorizedUpdater.robot_covarianceContinuousLaserTagVectorizedUpdater.opponent_covarianceContinuousLaserTagVectorizedUpdater.batch_observation_log_likelihood()ContinuousLaserTagVectorizedUpdater.batch_transition()ContinuousLaserTagVectorizedUpdater.config_idContinuousLaserTagVectorizedUpdater.from_environment()
LaserTagVectorizedUpdaterLaserTagVectorizedUpdater.floor_shapeLaserTagVectorizedUpdater.valid_cellLaserTagVectorizedUpdater.wall_dist_tableLaserTagVectorizedUpdater.measurement_noiseLaserTagVectorizedUpdater.transition_error_probLaserTagVectorizedUpdater.batch_observation_log_likelihood()LaserTagVectorizedUpdater.batch_transition()LaserTagVectorizedUpdater.config_idLaserTagVectorizedUpdater.from_environment()
create_continuous_laser_tag_belief()create_laser_tag_belief()- Submodules
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs.continuous_laser_tag_belief_factory module
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs.continuous_laser_tag_vectorized_updater module
ContinuousLaserTagVectorizedUpdaterContinuousLaserTagVectorizedUpdater.wallsContinuousLaserTagVectorizedUpdater.grid_sizeContinuousLaserTagVectorizedUpdater.robot_radiusContinuousLaserTagVectorizedUpdater.opponent_radiusContinuousLaserTagVectorizedUpdater.tag_radiusContinuousLaserTagVectorizedUpdater.evasion_speedContinuousLaserTagVectorizedUpdater.measurement_noiseContinuousLaserTagVectorizedUpdater.robot_covarianceContinuousLaserTagVectorizedUpdater.opponent_covarianceContinuousLaserTagVectorizedUpdater.batch_observation_log_likelihood()ContinuousLaserTagVectorizedUpdater.batch_transition()ContinuousLaserTagVectorizedUpdater.config_idContinuousLaserTagVectorizedUpdater.from_environment()
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs.laser_tag_belief_factory module
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_beliefs.laser_tag_vectorized_updater module
LaserTagVectorizedUpdaterLaserTagVectorizedUpdater.floor_shapeLaserTagVectorizedUpdater.valid_cellLaserTagVectorizedUpdater.wall_dist_tableLaserTagVectorizedUpdater.measurement_noiseLaserTagVectorizedUpdater.transition_error_probLaserTagVectorizedUpdater.batch_observation_log_likelihood()LaserTagVectorizedUpdater.batch_transition()LaserTagVectorizedUpdater.config_idLaserTagVectorizedUpdater.from_environment()
- POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp_utils package
Submodules
POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_geometry module
Geometry utilities for the Continuous LaserTag POMDP.
Provides ray-AABB intersection, ray-circle intersection, wall collision resolution and grid clamping used by the continuous laser-tag environment and its vectorized belief updater.
Wall AABBs are stored as rows (cx, cy, hx, hy) where (cx, cy) is
the center and (hx, hy) the half-extents. Entity radii are used for
circle-AABB overlap tests during collision resolution.
- Functions:
- ray_aabb_distances: Vectorized ray-AABB slab intersection for multiple
rays originating from a single point against an array of AABBs.
- ray_circle_distance: Distance along a ray to the nearest intersection
with a circle.
compute_laser_measurements: Full 8-direction laser scan from a position. resolve_wall_collision: Push a circular entity out of overlapping AABBs. clamp_to_grid: Clamp a 2-D position to the grid boundaries.
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_geometry.batch_clamp_to_grid(positions, entity_radius, grid_size)[source]
Clamp an array of positions to the grid.
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_geometry.batch_laser_measurements(robot_positions, opponent_positions, opponent_radius, walls, grid_size)[source]
Compute 8-direction laser measurements for many particles.
- Parameters:
- Return type:
- Returns:
Shape
(N, 8)measurement array.
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_geometry.batch_resolve_wall_collision(positions, entity_radius, walls)[source]
Resolve wall collisions for an array of positions.
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_geometry.clamp_to_grid(position, entity_radius, grid_size)[source]
Clamp a position so the entity circle stays within
[0, w] x [0, h].
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_geometry.compute_laser_measurements(robot_pos, opponent_pos, opponent_radius, walls, grid_size)[source]
Compute 8-direction laser measurements from the robot.
Each measurement is the distance to the nearest obstacle (wall AABB, opponent circle, or grid boundary) along the corresponding ray in
LASER_DIRECTIONS.- Parameters:
- Return type:
- Returns:
Shape
(8,)array of distances.
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_geometry.ray_aabb_distances(origin, directions, walls)[source]
Compute distances from origin along each ray to the nearest wall AABB.
Uses the slab method. For each of the D directions the minimum positive intersection distance across all M walls is returned. If a ray does not hit any wall before
_RAY_MAXthe returned distance is_RAY_MAX.
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_geometry.ray_circle_distance(origin, direction, center, radius)[source]
Distance along a ray to the nearest intersection with a circle.
- POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_geometry.resolve_wall_collision(position, entity_radius, walls)[source]
Push a circular entity out of any overlapping wall AABBs.
For each wall, if the entity circle overlaps the AABB, the entity is pushed along the axis of minimum penetration.
POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_pomdp module
Continuous LaserTag POMDP Environment Implementation.
This module implements a continuous-space variant of the LaserTag pursuit-evasion POMDP where a robot must navigate to tag an opponent that moves stochastically through continuous 2-D space.
Two environment classes are provided:
ContinuousLaserTagPOMDP– continuous actions[dx, dy, tag_flag]ContinuousLaserTagPOMDPDiscreteActions– five string actions"up","down","right","left","tag"
- State representation:
np.ndarrayshape(5,)–[robot_x, robot_y, opponent_x, opponent_y, terminal_flag]- Observation:
np.ndarrayshape(8,)– noisy 8-direction laser range measurements. Terminal observation isnp.full(8, -1.0).
Opponent behaviour is selectable via opponent_policy (see
OpponentPolicy):
EVADE (default) flees the robot’s pre-move position at evasion_speed;
PURSUE chases the robot’s post-move position. EVADE_WHEN_SPOTTED flees
only while the robot has line of sight to it and otherwise holds its position
(in this continuous env; the discrete grid env moves randomly instead).
evasion_speed is a direction-neutral step magnitude under all policies.
- Classes:
ContinuousLaserTagPOMDP: Continuous-action environment. ContinuousLaserTagPOMDPDiscreteActions: Discrete-action variant.
- class POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_pomdp.ContinuousLaserTagPOMDP(discount_factor, name='ContinuousLaserTagPOMDP', grid_size=(11.0, 7.0), walls=None, robot_radius=0.3, opponent_radius=0.3, tag_radius=0.5, tag_reward=10.0, tag_penalty=10.0, step_cost=1.0, measurement_noise=1.0, robot_transition_cov_matrix=array([[0.1, 0.], [0., 0.1]]), opponent_transition_cov_matrix=array([[0.05, 0.], [0., 0.05]]), evasion_speed=0.6, dangerous_areas=None, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, dangerous_area_hit_probability=1.0, output_dir=None, debug=False, use_queue_logger=False, initial_state=None, opponent_policy=OpponentPolicy.EVADE)[source]
Bases:
EnvironmentContinuous LaserTag POMDP with continuous
[dx, dy, tag_flag]actions.A pursuit-evasion problem in continuous 2-D space where a robot must navigate to tag an opponent. The robot receives noisy 8-direction laser range observations.
- Stochasticity:
The dangerous-area penalty can be applied either deterministically (the default) or stochastically. When
dangerous_area_hit_probability == 1.0(default), the kernel’s deterministic deduction is preserved verbatim, matching legacy behavior. Whendangerous_area_hit_probability < 1.0, the accumulated dangerous-area deduction is applied to the reward only with that probability perreward()call, producing a heavy-tailed return distribution suitable for benchmarking risk-sensitive planners (e.g. ICVaR-aware MCTS) against expected-value MCTS on the same env. Note that this makesreward(state, action)non-deterministic given a state-action pair, so any external caching that assumes deterministic rewards must be aware of this.transition_log_probabilityis unaffected.
Example
>>> import numpy as np >>> np.random.seed(42) >>> >>> # Initialize environment >>> env = ContinuousLaserTagPOMDP(discount_factor=0.95) >>> >>> # Get initial state >>> initial_state = env.initial_state_dist().sample()[0] >>> >>> # Sample complete step >>> action = np.array([1.0, 0.0, 0.0]) >>> next_state, observation, reward = env.sample_next_step(initial_state, action) >>> >>> # Check terminal condition >>> env.is_terminal(initial_state) False
Example
Risk-sensitive evaluation – a 10%-tail-risk environment suitable for benchmarking ICVaR-aware planners against expected-value MCTS:
>>> env = ContinuousLaserTagPOMDP( ... discount_factor=0.95, ... dangerous_area_penalty=150.0, ... dangerous_area_hit_probability=0.1, ... )
- Parameters:
discount_factor (float)
name (str)
robot_radius (float)
opponent_radius (float)
tag_radius (float)
tag_reward (float)
tag_penalty (float)
step_cost (float)
measurement_noise (float)
robot_transition_cov_matrix (np.ndarray)
opponent_transition_cov_matrix (np.ndarray)
evasion_speed (float)
dangerous_area_radius (float)
dangerous_area_penalty (float)
dangerous_area_hit_probability (float)
output_dir (Optional[Path])
debug (bool)
use_queue_logger (bool)
initial_state (Optional[np.ndarray])
opponent_policy (OpponentPolicy)
- 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 environment-specific metrics from episode histories.
This method can be overridden by subclasses to provide custom metric calculations beyond standard return and episode length.
- 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:
- 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().
- 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).
- 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:
- 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:
- 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 (
ndarray) – 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.
- 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 batchedobservation_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.
- 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.
- 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 dispatched to native C++ via
cont_simulate_rollout.Pre-samples actions from
action_sampler, packs them into a(N, 3)buffer, and runs the full discounted-return loop inside C++. Results are numerically identical to theEnvironment.simulate_random_rollout()Python fallback.When
dangerous_area_hit_probability < 1.0, falls back to the Python rollout: the native kernel applies the dangerous-area penalty deterministically per step, which contradicts the stochastic semantics; routing through Pythonreward()keeps the per-step Bernoulli intact.Also falls back when
dangerous_areasis non-empty: the C++cont_simulate_rolloutkernel scores the danger penalty against the pre-transition robot position, while the Pythonreward()path (post-fix) consumes the realised post-transition position. Until the C++ kernel is rebuilt this is the only correctness-preserving path for configs with danger areas.
- class POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_pomdp.ContinuousLaserTagPOMDPDiscreteActions(discount_factor, name='ContinuousLaserTagPOMDPDiscreteActions', grid_size=(11.0, 7.0), walls=None, robot_radius=0.3, opponent_radius=0.3, tag_radius=0.5, tag_reward=10.0, tag_penalty=10.0, step_cost=1.0, measurement_noise=1.0, robot_transition_cov_matrix=array([[0.1, 0.], [0., 0.1]]), opponent_transition_cov_matrix=array([[0.05, 0.], [0., 0.05]]), evasion_speed=0.6, dangerous_areas=None, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, dangerous_area_hit_probability=1.0, output_dir=None, debug=False, use_queue_logger=False, initial_state=None, opponent_policy=OpponentPolicy.EVADE)[source]
Bases:
ContinuousLaserTagPOMDP,DiscreteActionsEnvironmentContinuous LaserTag POMDP with discrete string actions.
Actions:
"up","down","right","left","tag".Example
>>> import numpy as np >>> np.random.seed(42) >>> >>> env = ContinuousLaserTagPOMDPDiscreteActions(discount_factor=0.95) >>> >>> initial_state = env.initial_state_dist().sample()[0] >>> actions = env.get_actions() >>> >>> action = actions[0] >>> next_state, observation, reward = env.sample_next_step(initial_state, action) >>> >>> env.is_terminal(initial_state) False
- Parameters:
discount_factor (float)
name (str)
robot_radius (float)
opponent_radius (float)
tag_radius (float)
tag_reward (float)
tag_penalty (float)
step_cost (float)
measurement_noise (float)
robot_transition_cov_matrix (np.ndarray)
opponent_transition_cov_matrix (np.ndarray)
evasion_speed (float)
dangerous_area_radius (float)
dangerous_area_penalty (float)
dangerous_area_hit_probability (float)
output_dir (Optional[Path])
debug (bool)
use_queue_logger (bool)
initial_state (Optional[np.ndarray])
opponent_policy (OpponentPolicy)
- 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).
- 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.
- 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 batchedobservation_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.
- 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.
- 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 dispatched to native C++ via
cont_simulate_rollout.Pre-samples actions from
action_sampler, packs them into a(N, 3)buffer, and runs the full discounted-return loop inside C++. Results are numerically identical to theEnvironment.simulate_random_rollout()Python fallback.When
dangerous_area_hit_probability < 1.0, falls back to the Python rollout: the native kernel applies the dangerous-area penalty deterministically per step, which contradicts the stochastic semantics; routing through Pythonreward()keeps the per-step Bernoulli intact.Also falls back when
dangerous_areasis non-empty: the C++cont_simulate_rolloutkernel scores the danger penalty against the pre-transition robot position, while the Pythonreward()path (post-fix) consumes the realised post-transition position. Until the C++ kernel is rebuilt this is the only correctness-preserving path for configs with danger areas.
- class POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_pomdp.ContinuousLaserTagPOMDPMetrics(*values)[source]
Bases:
EnumMetric names for Continuous LaserTag POMDP.
- AVERAGE_ALL_DANGEROUS_ENCOUNTERS = 'average_all_dangerous_encounters'
- AVERAGE_DANGEROUS_AREA_STEPS = 'average_dangerous_area_steps'
- AVERAGE_EPISODE_LENGTH = 'average_episode_length'
- AVERAGE_FAILED_TAG_ATTEMPTS = 'average_failed_tag_attempts'
- AVERAGE_WALL_COLLISIONS = 'average_wall_collisions'
- GOAL_REACHING_RATE = 'goal_reaching_rate'
- TAG_SUCCESS_RATE = 'tag_success_rate'
POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_visualizer module
Continuous LaserTag POMDP Visualization Module.
This module provides visualization for the continuous-space LaserTag environment, creating animated GIF visualizations of episodes.
- class POMDPPlanners.environments.laser_tag_pomdp.continuous_laser_tag_visualizer.ContinuousLaserTagVisualizer(grid_size, walls, robot_radius, opponent_radius, dangerous_areas, dangerous_area_radius)[source]
Bases:
objectHandles visualization for the Continuous LaserTag POMDP.
Creates animated GIF visualizations showing robot and opponent movement as rendered icons, rectangular walls, laser rays, belief particles, and tag indicators. The robot is shown as a red humanoid and the opponent as a blue wheeled rover.
- Parameters:
- grid_size
Arena dimensions
(width, height)as ndarray.
- walls
Shape
(M, 4)wall AABB array.
- robot_radius
Robot body radius.
- opponent_radius
Opponent body radius.
- dangerous_areas
Dangerous area centers as
(x, y)tuples.
- dangerous_area_radius
Radius of dangerous areas.
POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp module
LaserTag POMDP Environment Implementation.
This module implements the LaserTag problem, a pursuit-evasion POMDP environment where an agent must navigate a grid to tag an opponent that moves stochastically. The agent has noisy observations of the opponent’s location.
The LaserTag problem features: - A grid-based environment (default 7x11) with optional walls - Robot and opponent moving on discrete grid cells - 5 possible actions: North, South, East, West, Tag - 8-directional laser range measurements with Gaussian noise - Positive reward for successful tagging, negative reward for failed tag attempts - Step cost for each movement action - Opponent moves with 0.4 prob in x-dir, 0.4 prob in y-dir, 0.2 prob stay; the
direction of the 0.4 mass is set by
opponent_policy(seeOpponentPolicy):EVADE(default) moves away from the robot’s pre-move position,PURSUEmoves toward the robot’s post-move position
When aligned on an axis, the 0.4 budget is split equally (0.2/0.2) between both directions, regardless of policy
- Classes:
LaserTagState: State representation with robot and opponent positions LaserTagPOMDP: Main environment class implementing the LaserTag problem OpponentPolicy: Selectable opponent transition behaviour (evade vs pursue)
- class POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp.LaserTagPOMDP(discount_factor, name='LaserTagPOMDP', floor_shape=(11, 7), walls={(1, 2), (3, 0), (3, 4), (5, 0), (6, 4), (9, 1), (9, 4), (10, 6)}, tag_reward=10.0, tag_penalty=10.0, step_cost=1.0, measurement_noise=1.0, dangerous_areas={(2, 5), (5, 3), (7, 1)}, dangerous_area_radius=1.0, dangerous_area_penalty=5.0, output_dir=None, debug=False, use_queue_logger=False, initial_state=None, transition_error_prob=0.0, reward_model_type=RewardModelType.CONSTANT_HAZARD_PENALTY, penalty_decay=1.0, opponent_policy=OpponentPolicy.EVADE)[source]
Bases:
DiscreteActionsEnvironmentLaserTag POMDP environment implementation.
This is a pursuit-evasion problem where a robot must navigate a grid to tag an opponent. The robot receives noisy observations of the opponent’s position and must decide when and where to attempt tagging.
Problem Structure: - States: numpy array [robot_row, robot_col, opp_row, opp_col, terminal] - Actions: North(0), South(1), East(2), West(3), Tag(4) - Observations: 8-directional laser measurements (N,NE,E,SE,S,SW,W,NW) - Rewards: Tag success(+10), Tag failure(-10), Movement(-1)
- Parameters:
discount_factor (float)
name (str)
tag_reward (float)
tag_penalty (float)
step_cost (float)
measurement_noise (float)
dangerous_area_radius (float)
dangerous_area_penalty (float)
output_dir (Path | None)
debug (bool)
use_queue_logger (bool)
initial_state (ndarray | None)
transition_error_prob (float)
reward_model_type (RewardModelType)
penalty_decay (float)
opponent_policy (OpponentPolicy)
- floor_shape
Grid dimensions as (rows, cols)
- walls
Set of wall positions as (row, col) tuples
- tag_reward
Reward for successful tagging
- tag_penalty
Penalty for unsuccessful tagging
- step_cost
Cost per movement action
- measurement_noise
Standard deviation of observation noise
Example
>>> import numpy as np >>> np.random.seed(42) # For reproducible results >>> >>> # Initialize environment >>> env = LaserTagPOMDP(discount_factor=0.95) >>> >>> # 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
- cache_visualization(history, cache_path)[source]
Cache visualization of the LaserTag episode as an animated GIF.
Creates an animated visualization showing: - Robot movement (red circle) - Opponent movement (blue circle) - Walls (black squares) - Dangerous areas (red circles) - Action arrows showing robot’s intended movement - Laser measurements (green rays from robot position) - Belief particles (if available) showing robot’s belief about opponent location - Grid boundaries and coordinate system
- Parameters:
- Raises:
ValueError – If history is empty or contains invalid data
TypeError – If cache_path is not a Path object or doesn’t end with .gif
- Return type:
- compute_metrics(histories)[source]
Compute LaserTag POMDP specific metrics from simulation histories.
- Return type:
- Parameters:
- 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).
- is_equal_observation(observation1, observation2)[source]
Check if two observations are equal.
Observations are 8-dimensional laser measurements or terminal observations.
- 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 transition.
The wall / dangerous-area penalty is computed against the realised post-action robot position taken from
next_state. When the caller omitsnext_state(e.g., the open-loop scalar API path) the method resamples a transition viasample_next_state()so the penalty is always scored against an actual draw from the transition kernel — never against the open-loopstate + action_vectorintended position.Environment.sample_next_step()threads its samplednext_stateinto this method so trajectory and reward agree on the same realisation.
- reward_batch(states, action, next_states=None)[source]
Vectorised reward for a batch of states under a single action.
When
next_statesis supplied the danger-area / wall penalty is evaluated against the realised positions innext_states[:, :2](matching the contract honoured byEnvironment.sample_next_step()). When it isNonethe method resamples viasample_next_state_batch()whenever penalty terms exist, then delegates to the reward model so reward and trajectory remain consistent end-to-end.
- 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.
- 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.
- class POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp.LaserTagPOMDPMetrics(*values)[source]
Bases:
EnumMetric names for LaserTag POMDP environment.
- AVERAGE_ALL_DANGEROUS_ENCOUNTERS = 'average_all_dangerous_encounters'
- AVERAGE_DANGEROUS_AREA_STEPS = 'average_dangerous_area_steps'
- AVERAGE_EPISODE_LENGTH = 'average_episode_length'
- AVERAGE_FAILED_TAG_ATTEMPTS = 'average_failed_tag_attempts'
- AVERAGE_OBSTACLE_COLLISIONS = 'average_obstacle_collisions'
- GOAL_REACHING_RATE = 'goal_reaching_rate'
- TAG_SUCCESS_RATE = 'tag_success_rate'
- class POMDPPlanners.environments.laser_tag_pomdp.laser_tag_pomdp.RewardModelType(*values)[source]
Bases:
EnumReward-model variants selectable on
LaserTagPOMDP.- CONSTANT_HAZARD_PENALTY = 'constant_hazard_penalty'
- DISTANCE_DECAYED_HAZARD_PENALTY = 'distance_decayed_hazard_penalty'
- ZERO_MEAN_HAZARD_SHOCK = 'zero_mean_hazard_shock'
POMDPPlanners.environments.laser_tag_pomdp.laser_tag_visualizer module
LaserTag POMDP Visualization Module.
This module provides visualization functionality for LaserTag POMDP environments, creating animated GIF visualizations of episodes.
- class POMDPPlanners.environments.laser_tag_pomdp.laser_tag_visualizer.LaserTagVisualizer(floor_shape, walls, dangerous_areas, dangerous_area_radius)[source]
Bases:
objectHandles visualization for LaserTag POMDP environments.
Creates animated GIF visualizations showing robot movement, opponent movement, walls, laser measurements, belief particles, and action indicators.
- Parameters:
- floor_shape
Grid dimensions as (rows, cols)
- walls
Set of wall positions as (row, col) tuples
- dangerous_areas
List of dangerous area center positions
- dangerous_area_radius
Radius around dangerous area centers
- create_visualization(history, cache_path)[source]
Create animated GIF visualization of a LaserTag episode.
Creates an animated visualization showing: - Robot movement (red circle with path trail) - Opponent movement (blue circle with path trail) - Walls (black squares) - Dangerous areas (red circles) - Action arrows showing robot’s intended movement - Laser measurements (green rays from robot position) - Belief particles (if available) showing robot’s belief about opponent location - Grid boundaries and coordinate system - Step counter and action labels
- Parameters:
- Raises:
ValueError – If history is empty or contains invalid data
TypeError – If cache_path is not a Path object or doesn’t end with .gif
- Return type: