POMDPPlanners.environments.nuplan_pomdp package

nuPlan POMDP environment: a forward-only world plus a swappable planner model.

This package adapts the nuPlan closed-loop planning simulator to the POMDPPlanners interface, following the same world/model split as the CARLA package:

  • nuplan_pomdp — the ground-truth world (NuPlanPOMDP), forward-only, emitting a raw {ego, agents} observation.

  • nuplan_perception — the swappable per-channel observation (encoder) models the planner degrades the raw reading with.

  • nuplan_generative_models — the planner-side generative model (policy.environment): dynamics + the composed observation model.

  • nuplan_belief — the particle belief that stamps the perceived agent block onto its particles.

class POMDPPlanners.environments.nuplan_pomdp.NuPlanPOMDP(discount_factor, scenario_loader=None, action_presets=None, max_tracked_agents=5, simulation_horizon=8.0, fixed_delta_seconds=0.1, reactive_agents=True, collision_distance=2.0, collision_penalty=100.0, desired_speed=8.0, out_lane_thresh=2.0, observation_extractor=None, seed=None, name=None, reward_range=None, output_dir=None, debug=False, use_queue_logger=False)[source]

Bases: Environment

Forward-only adapter exposing a nuPlan closed-loop session as a world POMDP.

The wrapper drives a nuPlan Simulation as the ground-truth world of an episode. It advances the simulator exactly one iteration per real interaction and serves the resulting next state, observation and reward from a small cache, because the POMDPPlanners episode loop requests those three quantities through separate method calls while nuPlan produces them atomically. The state is the ego vehicle’s ground-truth kinematics; the observation is the ego proprioception plus a tracked-object list, so the world is genuinely partially observed. See the module docstring for the exact state and observation variables, units, and frames.

Note

This is a world environment, not a generative model. It cannot sample a transition from an arbitrary state, so belief particle propagation and density queries are unsupported and raise NotImplementedError / RuntimeError. Pair it with a generative model environment on the planner (policy.environment).

Parameters:
action_presets

Discrete (acceleration, steering_angle) control pairs.

max_tracked_agents

Number of nearest agents carried in state/observation.

seed

Optional seed applied to the first reset for reproducibility.

Example

The environment is used as the forward-only world of an EpisodeRunner, paired with a separate generative model on the planner. It requires the nuPlan devkit and a scenario loader, so this snippet is illustrative rather than executed:

env = NuPlanPOMDP(discount_factor=0.95, scenario_loader=load_scenario)
state = env.initial_state_dist().sample()[0]
next_state, observation, reward = env.sample_next_step(state, 0)
compute_metrics(histories)[source]

Compute nuPlan driving-quality metrics from episode histories.

Parameters:

histories (List[History]) – Episode histories to summarise.

Returns:

  • collision_rate: fraction of episodes that ended in a collision.

  • average_progress: mean per-episode ground distance travelled (m).

  • average_speed: mean ego speed over the driven trajectory (m/s).

  • near_miss_count: mean number of near-miss events per episode.

  • min_vehicle_distance: mean over episodes of the closest the ego came to any agent (m); episodes that saw no agent are excluded.

Return type:

List[MetricValue]

get_metric_names()[source]

Names of the nuPlan-specific evaluation metrics.

Returns:

collision_rate, average_progress, average_speed, near_miss_count and min_vehicle_distance.

Return type:

List[str]

hash_action(action)[source]

Return a hashable key consistent with action equality.

Used by tree-search planners to index action children of a belief node in O(1). The returned key MUST satisfy:

action_a == action_b   (per env's notion of equality)
==> hash_action(action_a) == hash_action(action_b)

Subclasses with non-hashable actions (e.g. np.ndarray) must override to return a hashable surrogate (tobytes() is the standard choice for ndarray actions, which mirrors the np.array_equal semantics used by the linear-scan fallback).

Parameters:

action (Any) – Action to hash.

Return type:

Hashable

Returns:

A hashable key derived from action.

hash_observation(observation)[source]

Return a hashable key consistent with is_equal_observation().

Used by tree-search planners to index belief children by observation in O(1). The returned key MUST satisfy the contract:

is_equal_observation(a, b) implies hash_observation(a) == hash_observation(b)
Parameters:

observation (Any) – Observation to hash.

Returns:

the observation itself when it is already hashable).

Return type:

Hashable

Raises:

NotImplementedError – If the observation is not hashable and the subclass has not provided an override. Subclasses with non-hashable observations (e.g. np.ndarray) MUST override.

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_terminal(state)[source]

Check if a state is terminal.

Parameters:

state (Any) – State to check for terminal condition

Return type:

bool

Returns:

True if the state is terminal, False otherwise

Note

Subclasses must implement this method to define terminal conditions.

observation_log_probability(next_state, action, observations)[source]

Log-probability of each candidate observation under (next_state, action).

Returns np.ndarray of shape (N,) where N is the number of candidate observations. Subclasses must implement.

Return type:

ndarray

Parameters:

  • next_state (Any)

  • action (Any)

  • observations (Any)

reward(state, action, next_state=None)[source]

Calculate the immediate reward for a state-action(-next_state) tuple.

next_state is the realised post-transition state when known (e.g. threaded by sample_next_step()), allowing rewards that depend on stochastic transition outcomes to use the same draw as the trajectory instead of resampling. Subclasses whose reward is a pure function of (state, action) may ignore it; subclasses whose reward depends on the realised next state (collision penalties, win bonuses) should consume it when provided and fall back to drawing/computing one when None.

Parameters:

  • state (Any) – Current state.

  • action (Any) – Action executed from state.

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

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

sample_next_state(state, action, n_samples=1)[source]

Sample one or more next states for (state, action).

Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.

Returns:

a single next state of the env’s native type. When n_samples > 1: an array-like of length n_samples (numeric envs return np.ndarray of shape (n_samples, *dim); structured envs return List[T]).

Return type:

ndarray

Parameters:
sample_observation(next_state, action, n_samples=1)[source]

Sample one or more observations for (next_state, action).

Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.

Returns:

a single observation. When n_samples > 1: an array-like of length n_samples.

Return type:

Any

Parameters:

  • next_state (Any)

  • action (Any)

  • n_samples (int)

transition_log_probability(state, action, next_states)[source]

Log-probability of each candidate next state under (state, action).

Returns np.ndarray of shape (N,) where N is the number of candidate next states. Subclasses must implement.

Return type:

ndarray

Parameters:
class POMDPPlanners.environments.nuplan_pomdp.NuPlanPOMDPMetrics(*values)[source]

Bases: Enum

Metric names for the nuPlan POMDP environment.

AVERAGE_PROGRESS = 'average_progress'
AVERAGE_SPEED = 'average_speed'
COLLISION_RATE = 'collision_rate'
MIN_VEHICLE_DISTANCE = 'min_vehicle_distance'
NEAR_MISS_COUNT = 'near_miss_count'
class POMDPPlanners.environments.nuplan_pomdp.PerceivedAgentsBelief(particles, log_weights, max_tracked_agents=5, agent_pose_jitter=0.3, resampling=True, ess_factor=0.5)[source]

Bases: WeightedParticleBeliefReinvigoration

Weighted particle belief that stamps the observation’s agent block onto every particle.

After the standard particle-filter weight update and resample, the reinvigoration step writes the current observation’s agents block into every particle’s agent slots (plus optional per-particle jitter), leaving the ego block to the filter and any trailing light slot untouched. The belief holds no perception state — perception is the planner model’s, applied upstream by encode_observation — so a plain observation with a perceived agents block is all it needs.

Parameters:
max_tracked_agents

Number of fixed agent slots carried in each particle.

agent_pose_jitter

Std of Gaussian noise added to each stamped agent’s [rel_x, rel_y, rel_yaw, rel_speed] pose, for particle diversity.

Example

>>> import numpy as np
>>> np.random.seed(0)
>>> from POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp import (
...     AGENT_SLOT_WIDTH, EGO_STATE_WIDTH)
>>> width = EGO_STATE_WIDTH + 1 * AGENT_SLOT_WIDTH
>>> particles = [np.zeros(width) for _ in range(4)]
>>> belief = PerceivedAgentsBelief(
...     particles=particles,
...     log_weights=np.log(np.ones(4) / 4),
...     max_tracked_agents=1,
... )
>>> observation = {  # a perceived agent 8 m ahead
...     "ego": np.zeros(7),
...     "agents": np.array([1.0, 8.0, 0.0, 0.0, 5.0]),
... }
>>> base = WeightedParticleBelief(particles=particles, log_weights=belief.log_weights)
>>> refreshed = belief.reinvigorate("noop", observation, None, base)
>>> bool(np.asarray(refreshed.particles)[0, EGO_STATE_WIDTH] == 1.0)  # slot now present
True
reinvigorate(action, observation, pomdp, belief)[source]

Stamp the observation’s perceived agent block onto every particle.

Return type:

PerceivedAgentsBelief

Parameters:
POMDPPlanners.environments.nuplan_pomdp.assemble_state(ego_row, agent_rows, max_tracked_agents, light_row=None)[source]

Concatenate an ego row, padded agent slots, and a light slot into a state vector.

Pure numeric assembly of the nuPlan state layout, factored out of the live session so the state geometry can be exercised without a nuPlan installation. Agent rows are written into the nearest fixed slots (already ego-frame); missing slots are padded with zeros (present == 0).

Parameters:

  • ego_row (Union[Sequence[float], ndarray]) – The EGO_STATE_WIDTH ego block [x, y, yaw, vx, vy, lat, heading_err].

  • agent_rows (Union[Sequence[Sequence[float]], Sequence[ndarray]]) – Zero or more ego-frame agent rows [present, rel_x, rel_y, rel_yaw, rel_speed]; only the first max_tracked_agents are kept.

  • max_tracked_agents (int) – Number of fixed agent slots to emit.

  • light_row (Union[Sequence[float], ndarray, None]) – Optional LIGHT_SLOT_WIDTH traffic-light slot; a zero (absent) slot is emitted when None.

Return type:

ndarray

Returns:

The full state vector of width EGO_STATE_WIDTH + max_tracked_agents * AGENT_SLOT_WIDTH + LIGHT_SLOT_WIDTH.

POMDPPlanners.environments.nuplan_pomdp.driving_quality_reward(next_state, steering_angle, terminated, desired_speed, out_lane_thresh, collision_penalty)[source]

Score a transition with a gym-carla-style driving-quality reward.

Rewards along-route progress and penalises overspeed, drifting off the route baseline, harsh / high-speed steering, each elapsed step, and a terminal collision. Shared by the NuPlanPOMDP world and the planner-side factored model so the two score a transition identically by construction.

Parameters:

  • next_state (ndarray) – Resulting ego state [x, y, yaw(rad), vx, vy, lat, heading_err].

  • steering_angle (float) – Steering command applied on the transition (from the action preset).

  • terminated (bool) – Whether the transition ended in a terminal collision.

  • desired_speed (float) – Target longitudinal speed (m/s); exceeding it is penalised.

  • out_lane_thresh (float) – Lateral offset (m) beyond which the ego is treated as off-route.

  • collision_penalty (float) – Penalty scale applied on a terminal collision.

Return type:

float

Returns:

The scalar reward for the transition.

POMDPPlanners.environments.nuplan_pomdp.relative_agent_row(ego_x, ego_y, ego_yaw_rad, other_x, other_y, other_yaw_rad, other_speed)[source]

Express another agent’s pose/speed in the ego frame as a present slot row.

Returns [1.0, rel_x, rel_y, rel_yaw, rel_speed] with rel_x pointing along the ego heading, rel_y to its left, and rel_yaw wrapped to [-pi, pi].

Parameters:

  • ego_x (float) – Ego x position in the map frame (m).

  • ego_y (float) – Ego y position in the map frame (m).

  • ego_yaw_rad (float) – Ego heading (rad).

  • other_x (float) – Other agent x position in the map frame (m).

  • other_y (float) – Other agent y position in the map frame (m).

  • other_yaw_rad (float) – Other agent heading (rad).

  • other_speed (float) – Other agent speed (m/s).

Return type:

ndarray

Returns:

The ego-frame present slot row for the agent.

Subpackages

Submodules

POMDPPlanners.environments.nuplan_pomdp.nuplan_belief module

Plain particle belief that stamps the observed agent block onto its particles.

Perception lives on the planner’s generative model, not here and not in the world: the forward-only NuPlanPOMDP emits a raw, ground-truth observation, and the model’s encode_observation() degrades it into the perceived observation the belief receives, so the agents channel the belief sees is already the tracked object list. The belief therefore does no perception at all.

It still cannot be a bare particle filter, though: a weight-only update can reweight and propagate the agents a particle was seeded with, but it can never acquire a vehicle that appears mid-episode, and a slot seeded empty stays empty forever. PerceivedAgentsBelief closes that gap in the minimal way — after the ordinary particle-filter weight update and resample, it replaces every particle’s ego-frame agent block with the observation’s agents block (plus optional per-particle pose jitter for diversity), leaving the ego block to the filter and any trailing light slot untouched. The perceived agents are the observation’s estimate, trusted rather than re-filtered as a per-particle latent.

The returned belief is itself a PerceivedAgentsBelief, so the stamping repeats on every step of the episode.

Note

The belief’s max_tracked_agents must match the width of the observation’s agents block, since that block is written straight into each particle’s fixed agent slots.

Classes:

PerceivedAgentsBelief: Particle belief that stamps the observed agent block onto particles.

class POMDPPlanners.environments.nuplan_pomdp.nuplan_belief.PerceivedAgentsBelief(particles, log_weights, max_tracked_agents=5, agent_pose_jitter=0.3, resampling=True, ess_factor=0.5)[source]

Bases: WeightedParticleBeliefReinvigoration

Weighted particle belief that stamps the observation’s agent block onto every particle.

After the standard particle-filter weight update and resample, the reinvigoration step writes the current observation’s agents block into every particle’s agent slots (plus optional per-particle jitter), leaving the ego block to the filter and any trailing light slot untouched. The belief holds no perception state — perception is the planner model’s, applied upstream by encode_observation — so a plain observation with a perceived agents block is all it needs.

Parameters:
max_tracked_agents

Number of fixed agent slots carried in each particle.

agent_pose_jitter

Std of Gaussian noise added to each stamped agent’s [rel_x, rel_y, rel_yaw, rel_speed] pose, for particle diversity.

Example

>>> import numpy as np
>>> np.random.seed(0)
>>> from POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp import (
...     AGENT_SLOT_WIDTH, EGO_STATE_WIDTH)
>>> width = EGO_STATE_WIDTH + 1 * AGENT_SLOT_WIDTH
>>> particles = [np.zeros(width) for _ in range(4)]
>>> belief = PerceivedAgentsBelief(
...     particles=particles,
...     log_weights=np.log(np.ones(4) / 4),
...     max_tracked_agents=1,
... )
>>> observation = {  # a perceived agent 8 m ahead
...     "ego": np.zeros(7),
...     "agents": np.array([1.0, 8.0, 0.0, 0.0, 5.0]),
... }
>>> base = WeightedParticleBelief(particles=particles, log_weights=belief.log_weights)
>>> refreshed = belief.reinvigorate("noop", observation, None, base)
>>> bool(np.asarray(refreshed.particles)[0, EGO_STATE_WIDTH] == 1.0)  # slot now present
True
reinvigorate(action, observation, pomdp, belief)[source]

Stamp the observation’s perceived agent block onto every particle.

Return type:

PerceivedAgentsBelief

Parameters:

POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp module

nuPlan POMDP world environment.

This module adapts the nuPlan closed-loop planning simulator to the POMDPPlanners Environment interface so it can serve as the ground-truth world in an EpisodeRunner.

nuPlan is forward-only: a Simulation advances a single true state one iteration per call, propagating the ego under a planned trajectory while reactive background agents (IDM) respond. It cannot be queried for a transition/observation density nor re-run from an arbitrary injected state, so it cannot act as a planner’s generative model. In the two-environment episode design the planner keeps its own generative model (policy.environment) and this wrapper only advances the single true state forward, one step per real interaction. Consequently NuPlanPOMDP.transition_log_probability() and NuPlanPOMDP.observation_log_probability() intentionally raise NotImplementedError — in the intended world/model split they are never called.

Unlike a fully-observed gym wrapper (observation equals state), nuPlan is genuinely partially observed: the ego reads its own proprioception plus a tracked-object list (DetectionsTracks) of the nearest agents, not the world’s full ground truth.

The state is the ego vehicle’s ground-truth kinematics and lane pose, [x, y, yaw, vx, vy, lat, heading_err], followed by fixed slots for the ``max_tracked_agents`` nearest other agents (ground truth). Each agent slot is [present, rel_x, rel_y, rel_yaw, rel_speed] expressed in the ego frame (rel_x forward, rel_y left, rel_yaw in radians, rel_speed in m/s); present is 1 for a filled slot and 0 for padding. The ego part is read straight from the simulator, where:

  • x, y: ego rear-axle position in the map frame, in metres.

  • yaw: ego heading about the map Z axis, in radians (nuPlan convention).

  • vx, vy: ego linear-velocity components in the map frame, in metres per second.

  • lat: signed lateral offset from the centre of the ego’s route baseline, in metres (positive to the baseline’s left).

  • heading_err: ego heading minus the route-baseline direction, wrapped to [-pi, pi], in radians.

The state ends with one traffic-light slot, [present, rel_x, rel_y, state_code, time_to_change] (ego frame; state_code is a TRAFFIC_LIGHT_* code, time_to_change in seconds), carrying the light governing the ego lane as ground truth (present == 0 when none affects it). It is always in the state and is independent of whether the observation exposes the light.

(The vertical axis z and roll/pitch are intentionally omitted; the ego is modelled on the ground plane.)

The lane-relative terms (lat, heading_err) drive a gym-carla-style driving-quality reward: it rewards longitudinal progress along the route while penalising overspeed, drifting off the baseline, and harsh / high-speed steering, plus a per-step time cost and a terminal collision penalty. See REWARD_SPEED_WEIGHT and the sibling weights.

The observation is a multi-modal dict of native nuPlan payloads:

  • "ego" (always present): the ego’s proprioceptive kinematics [x, y, yaw, vx, vy, lat, heading_err] — nuPlan gives the ego near-perfect self-localisation, so this is the ego measurement channel.

  • "agents" (always present): the max_tracked_agents agent slots of the state flattened, reported raw at their true ego-frame poses. The world applies no perception, so this is the ground-truth channel; range-gating, occlusion and sensor noise are the planner model’s observation model, not the world’s.

Any measurement noise is the planner model’s; the wrapper adds none.

Classes:

NuPlanPOMDP: Forward-only adapter exposing a nuPlan session as a world Environment.

class POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp.NuPlanPOMDP(discount_factor, scenario_loader=None, action_presets=None, max_tracked_agents=5, simulation_horizon=8.0, fixed_delta_seconds=0.1, reactive_agents=True, collision_distance=2.0, collision_penalty=100.0, desired_speed=8.0, out_lane_thresh=2.0, observation_extractor=None, seed=None, name=None, reward_range=None, output_dir=None, debug=False, use_queue_logger=False)[source]

Bases: Environment

Forward-only adapter exposing a nuPlan closed-loop session as a world POMDP.

The wrapper drives a nuPlan Simulation as the ground-truth world of an episode. It advances the simulator exactly one iteration per real interaction and serves the resulting next state, observation and reward from a small cache, because the POMDPPlanners episode loop requests those three quantities through separate method calls while nuPlan produces them atomically. The state is the ego vehicle’s ground-truth kinematics; the observation is the ego proprioception plus a tracked-object list, so the world is genuinely partially observed. See the module docstring for the exact state and observation variables, units, and frames.

Note

This is a world environment, not a generative model. It cannot sample a transition from an arbitrary state, so belief particle propagation and density queries are unsupported and raise NotImplementedError / RuntimeError. Pair it with a generative model environment on the planner (policy.environment).

Parameters:
action_presets

Discrete (acceleration, steering_angle) control pairs.

max_tracked_agents

Number of nearest agents carried in state/observation.

seed

Optional seed applied to the first reset for reproducibility.

Example

The environment is used as the forward-only world of an EpisodeRunner, paired with a separate generative model on the planner. It requires the nuPlan devkit and a scenario loader, so this snippet is illustrative rather than executed:

env = NuPlanPOMDP(discount_factor=0.95, scenario_loader=load_scenario)
state = env.initial_state_dist().sample()[0]
next_state, observation, reward = env.sample_next_step(state, 0)
action_presets: List[Tuple[float, float]]
compute_metrics(histories)[source]

Compute nuPlan driving-quality metrics from episode histories.

Parameters:

histories (List[History]) – Episode histories to summarise.

Returns:

  • collision_rate: fraction of episodes that ended in a collision.

  • average_progress: mean per-episode ground distance travelled (m).

  • average_speed: mean ego speed over the driven trajectory (m/s).

  • near_miss_count: mean number of near-miss events per episode.

  • min_vehicle_distance: mean over episodes of the closest the ego came to any agent (m); episodes that saw no agent are excluded.

Return type:

List[MetricValue]

get_metric_names()[source]

Names of the nuPlan-specific evaluation metrics.

Returns:

collision_rate, average_progress, average_speed, near_miss_count and min_vehicle_distance.

Return type:

List[str]

hash_action(action)[source]

Return a hashable key consistent with action equality.

Used by tree-search planners to index action children of a belief node in O(1). The returned key MUST satisfy:

action_a == action_b   (per env's notion of equality)
==> hash_action(action_a) == hash_action(action_b)

Subclasses with non-hashable actions (e.g. np.ndarray) must override to return a hashable surrogate (tobytes() is the standard choice for ndarray actions, which mirrors the np.array_equal semantics used by the linear-scan fallback).

Parameters:

action (Any) – Action to hash.

Return type:

Hashable

Returns:

A hashable key derived from action.

hash_observation(observation)[source]

Return a hashable key consistent with is_equal_observation().

Used by tree-search planners to index belief children by observation in O(1). The returned key MUST satisfy the contract:

is_equal_observation(a, b) implies hash_observation(a) == hash_observation(b)
Parameters:

observation (Any) – Observation to hash.

Returns:

the observation itself when it is already hashable).

Return type:

Hashable

Raises:

NotImplementedError – If the observation is not hashable and the subclass has not provided an override. Subclasses with non-hashable observations (e.g. np.ndarray) MUST override.

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_terminal(state)[source]

Check if a state is terminal.

Parameters:

state (Any) – State to check for terminal condition

Return type:

bool

Returns:

True if the state is terminal, False otherwise

Note

Subclasses must implement this method to define terminal conditions.

observation_log_probability(next_state, action, observations)[source]

Log-probability of each candidate observation under (next_state, action).

Returns np.ndarray of shape (N,) where N is the number of candidate observations. Subclasses must implement.

Return type:

ndarray

Parameters:

  • next_state (Any)

  • action (Any)

  • observations (Any)

reward(state, action, next_state=None)[source]

Calculate the immediate reward for a state-action(-next_state) tuple.

next_state is the realised post-transition state when known (e.g. threaded by sample_next_step()), allowing rewards that depend on stochastic transition outcomes to use the same draw as the trajectory instead of resampling. Subclasses whose reward is a pure function of (state, action) may ignore it; subclasses whose reward depends on the realised next state (collision penalties, win bonuses) should consume it when provided and fall back to drawing/computing one when None.

Parameters:

  • state (Any) – Current state.

  • action (Any) – Action executed from state.

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

Return type:

float

Returns:

Immediate reward value.

Note

Subclasses must implement this method to define reward structure.

sample_next_state(state, action, n_samples=1)[source]

Sample one or more next states for (state, action).

Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.

Returns:

a single next state of the env’s native type. When n_samples > 1: an array-like of length n_samples (numeric envs return np.ndarray of shape (n_samples, *dim); structured envs return List[T]).

Return type:

ndarray

Parameters:
sample_observation(next_state, action, n_samples=1)[source]

Sample one or more observations for (next_state, action).

Hot-path entry point used by MCTS planners and particle filters. Subclasses must implement.

Returns:

a single observation. When n_samples > 1: an array-like of length n_samples.

Return type:

Any

Parameters:

  • next_state (Any)

  • action (Any)

  • n_samples (int)

transition_log_probability(state, action, next_states)[source]

Log-probability of each candidate next state under (state, action).

Returns np.ndarray of shape (N,) where N is the number of candidate next states. Subclasses must implement.

Return type:

ndarray

Parameters:
class POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp.NuPlanPOMDPMetrics(*values)[source]

Bases: Enum

Metric names for the nuPlan POMDP environment.

AVERAGE_PROGRESS = 'average_progress'
AVERAGE_SPEED = 'average_speed'
COLLISION_RATE = 'collision_rate'
MIN_VEHICLE_DISTANCE = 'min_vehicle_distance'
NEAR_MISS_COUNT = 'near_miss_count'
POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp.assemble_state(ego_row, agent_rows, max_tracked_agents, light_row=None)[source]

Concatenate an ego row, padded agent slots, and a light slot into a state vector.

Pure numeric assembly of the nuPlan state layout, factored out of the live session so the state geometry can be exercised without a nuPlan installation. Agent rows are written into the nearest fixed slots (already ego-frame); missing slots are padded with zeros (present == 0).

Parameters:

  • ego_row (Union[Sequence[float], ndarray]) – The EGO_STATE_WIDTH ego block [x, y, yaw, vx, vy, lat, heading_err].

  • agent_rows (Union[Sequence[Sequence[float]], Sequence[ndarray]]) – Zero or more ego-frame agent rows [present, rel_x, rel_y, rel_yaw, rel_speed]; only the first max_tracked_agents are kept.

  • max_tracked_agents (int) – Number of fixed agent slots to emit.

  • light_row (Union[Sequence[float], ndarray, None]) – Optional LIGHT_SLOT_WIDTH traffic-light slot; a zero (absent) slot is emitted when None.

Return type:

ndarray

Returns:

The full state vector of width EGO_STATE_WIDTH + max_tracked_agents * AGENT_SLOT_WIDTH + LIGHT_SLOT_WIDTH.

POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp.driving_quality_reward(next_state, steering_angle, terminated, desired_speed, out_lane_thresh, collision_penalty)[source]

Score a transition with a gym-carla-style driving-quality reward.

Rewards along-route progress and penalises overspeed, drifting off the route baseline, harsh / high-speed steering, each elapsed step, and a terminal collision. Shared by the NuPlanPOMDP world and the planner-side factored model so the two score a transition identically by construction.

Parameters:

  • next_state (ndarray) – Resulting ego state [x, y, yaw(rad), vx, vy, lat, heading_err].

  • steering_angle (float) – Steering command applied on the transition (from the action preset).

  • terminated (bool) – Whether the transition ended in a terminal collision.

  • desired_speed (float) – Target longitudinal speed (m/s); exceeding it is penalised.

  • out_lane_thresh (float) – Lateral offset (m) beyond which the ego is treated as off-route.

  • collision_penalty (float) – Penalty scale applied on a terminal collision.

Return type:

float

Returns:

The scalar reward for the transition.

POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp.relative_agent_row(ego_x, ego_y, ego_yaw_rad, other_x, other_y, other_yaw_rad, other_speed)[source]

Express another agent’s pose/speed in the ego frame as a present slot row.

Returns [1.0, rel_x, rel_y, rel_yaw, rel_speed] with rel_x pointing along the ego heading, rel_y to its left, and rel_yaw wrapped to [-pi, pi].

Parameters:

  • ego_x (float) – Ego x position in the map frame (m).

  • ego_y (float) – Ego y position in the map frame (m).

  • ego_yaw_rad (float) – Ego heading (rad).

  • other_x (float) – Other agent x position in the map frame (m).

  • other_y (float) – Other agent y position in the map frame (m).

  • other_yaw_rad (float) – Other agent heading (rad).

  • other_speed (float) – Other agent speed (m/s).

Return type:

ndarray

Returns:

The ego-frame present slot row for the agent.

POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp.wrap_to_pi(angle)[source]

Wrap an angle in radians to the [-pi, pi] interval.

Every angle in the nuPlan state/observation layout (ego yaw and heading_err, each agent slot’s rel_yaw) carries this invariant, so world, model and belief all route their angle arithmetic through here.

Parameters:

angle (float) – Angle in radians.

Return type:

float

Returns:

The equivalent angle in [-pi, pi].