POMDPPlanners.environments.nuplan_pomdp.nuplan_generative_models package

Planner-side generative models paired with the forward-only nuPlan world.

While nuplan_pomdp is the ground-truth world (forward-only, no densities), a planner carries a generative model as policy.environment. This subpackage holds that model: the abstract interface and its concrete implementations, all sharing the nuPlan state/observation schema defined by the world.

Classes:

NuPlanModelPOMDP: Abstract generative-model interface over the nuPlan schema. FactoredNuPlanModelPOMDP: Concrete nuPlan model with a factored observation model. KinematicNuPlanModelPOMDP: Factored model with a kinematic bicycle transition.

class POMDPPlanners.environments.nuplan_pomdp.nuplan_generative_models.FactoredNuPlanModelPOMDP(discount_factor, action_presets=None, max_tracked_agents=5, perception_range=50.0, occlusion_radius=1.5, pose_std=0.5, ego_std=0.01, detect_prob=0.95, desired_speed=8.0, out_lane_thresh=2.0, collision_penalty=100.0, observation=None, name=None)

Bases: NuPlanModelPOMDP

Concrete nuPlan generative model pairing placeholder dynamics with factored perception.

The observation is composed per channel (held as self.observation_models): a FactoredAgentObservationModel on agents (detection gated by perception range and geometric occlusion, additive Gaussian pose noise) and an EgoObservationModel on ego. The reward is the shared gym-carla driving-quality reward. The transition is a documented identity placeholder to be replaced with real dynamics per study.

Parameters:

• discount_factor (float)

• action_presets (Sequence[Tuple[float, float]] | None)

• max_tracked_agents (int)

• perception_range (float | None)

• occlusion_radius (float)

• pose_std (float)

• ego_std (float)

• detect_prob (float)

• desired_speed (float)

• out_lane_thresh (float)

• collision_penalty (float)

• observation (Dict[str, str] | None)

• name (str | None)

observation_models

The per-channel {channel: NuPlanObservationModel} map carrying the observation parameters (perception_range, occlusion_radius, pose_std, ego_std, detect_prob).

desired_speed

Target longitudinal speed (m/s) used by the reward.

out_lane_thresh

Lateral offset (m) beyond which the reward penalises off-route.

collision_penalty

Penalty scale applied on a terminal collision in the reward.

Example

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>>
>>> from POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp import (
...     AGENT_SLOT_WIDTH, EGO_STATE_WIDTH, LIGHT_SLOT_WIDTH)
>>> env = FactoredNuPlanModelPOMDP(discount_factor=0.95)
>>>
>>> width = (
...     EGO_STATE_WIDTH + env.max_tracked_agents * AGENT_SLOT_WIDTH + LIGHT_SLOT_WIDTH)
>>> state = np.zeros(width)
>>> action = env.get_actions()[0]
>>>
>>> next_state, observation, reward = env.sample_next_step(state, action)
>>> sorted(observation)
['agents', 'ego']
>>> env.is_terminal(state)
False

initial_observation_dist()

Get the initial observation distribution.

Return type:

Distribution

Returns:

Distribution over initial observations

Note

Subclasses must implement this method to define initial observations.

initial_state_dist()

Get the initial state distribution.

Return type:

Distribution

Returns:

Distribution over initial states

Note

Subclasses must implement this method to define the starting distribution.

is_terminal(state)

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.

reward(state, action, next_state=None)

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

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

Parameters:

• state (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)

Sample n_samples next states for (state, action).

Return type:

ndarray

Parameters:

• state (Any)

• action (Any)

• n_samples (int)

transition_log_probability(state, action, next_states)

Log-density of next_states under the transition model for (state, action).

Return type:

ndarray

Parameters:

• state (Any)

• action (Any)

• next_states (Any)

class POMDPPlanners.environments.nuplan_pomdp.nuplan_generative_models.KinematicNuPlanModelPOMDP(discount_factor, dt=0.1, action_presets=None, max_tracked_agents=5, wheelbase=2.8, drag=0.05, collision_gap=5.0, collision_halfwidth=1.2, safe_distance=12.0, stop_gap=0.0, name=None, **kwargs)

Bases: FactoredNuPlanModelPOMDP

Factored nuPlan model whose transition is a kinematic bicycle propagation.

Parameters:

• discount_factor (float)

• dt (float)

• action_presets (Sequence[Tuple[float, float]] | None)

• max_tracked_agents (int)

• wheelbase (float)

• drag (float)

• collision_gap (float)

• collision_halfwidth (float)

• safe_distance (float)

• stop_gap (float)

• name (str | None)

• kwargs (Any)

dt

Integration step (seconds); must match the world’s fixed_delta_seconds.

wheelbase

Bicycle-model wheelbase (m) mapping steering angle to yaw rate.

drag

Linear speed-proportional deceleration coefficient (1/s).

collision_gap

Forward ego-frame distance (m) within which a present agent ahead is treated as a predicted collision by is_terminal().

collision_halfwidth

Lateral ego-frame half-corridor (m) within which a present agent ahead is treated as a predicted collision by is_terminal().

safe_distance

Lead gap (m) at/above which the reward targets the full desired_speed; the obstacle-aware target ramps down below it.

stop_gap

Lead gap (m) at/below which the obstacle-aware target speed is zero; 0.0 keeps the flat desired_speed.

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp import (
...     AGENT_SLOT_WIDTH, EGO_STATE_WIDTH, LIGHT_SLOT_WIDTH)
>>> env = KinematicNuPlanModelPOMDP(discount_factor=0.95, dt=0.1)
>>>
>>> width = (
...     EGO_STATE_WIDTH + env.max_tracked_agents * AGENT_SLOT_WIDTH + LIGHT_SLOT_WIDTH)
>>> state = np.zeros(width)
>>> accelerate_action = 0  # (1.5, 0.0) accelerate straight
>>> next_state = env.sample_next_state(state, accelerate_action)
>>>
>>> bool(next_state[3] > 0.0)  # acceleration produced forward velocity
True

is_terminal(state)

Whether a present agent occupies the ego’s footprint just ahead.

Because this model does predict ego and agent motion, it can foresee running into the vehicle ahead. Any present agent slot within collision_gap metres forward and collision_halfwidth metres laterally (ego frame) is treated as a collision, which the inherited reward turns into the terminal collision_penalty (driving_quality_reward()).

Return type:

bool

Parameters:

state (Any)

reward(state, action, next_state=None)

Driving-quality reward whose target speed adapts to the nearest lead obstacle.

The bare parent reward tracks a fixed desired_speed and only charges the collision once an agent is inside the terminal box — a cliff that cannot be braked for at speed, while a large fixed penalty instead freezes the ego in traffic. This override follows an obstacle-aware desired speed: the target equals the full desired_speed when the lead gap is at least safe_distance, ramps linearly to zero at stop_gap, and is zero closer in. The ego is thus rewarded for driving when the road is clear and for slowing as an obstacle nears, without a separate penalty term that traps it at a standstill. stop_gap == 0 keeps the flat parent behaviour.

Return type:

float

Parameters:

• state (Any)

• action (Any)

• next_state (Any)

sample_next_state(state, action, n_samples=1)

Sample n_samples next states for (state, action).

Return type:

ndarray

Parameters:

• state (Any)

• action (Any)

• n_samples (int)

sample_next_state_batch(states, action)

Sample one next state per input state, all under the same action.

Used by particle filters: given N current particles and one action, draw N next states (one per particle) in a single vectorized call.

The default implementation falls back to a per-state Python loop delegating to sample_next_state(). Native-backed envs (those whose state-transition kernel exposes batch_sample(states_array)) should override to avoid the loop.

Parameters:

• states (Any) – A sequence (length N) or ndarray of shape (N, *dim) of input particles.

• action (Any) – A single action to apply to every particle.

Returns:

np.ndarray of shape (N, *dim). For structured envs (Tiger strings, Pacman tuples): a list of length N.

Return type:

ndarray

class POMDPPlanners.environments.nuplan_pomdp.nuplan_generative_models.NuPlanModelPOMDP(discount_factor, action_presets=None, max_tracked_agents=5, observation_models=None, name=None)

Bases: DiscreteActionsEnvironment

Abstract generative-model interface paired with the forward-only nuPlan world.

Concrete subclasses supply the dynamics — sample_next_state(), transition_log_probability(), reward(), is_terminal(), and the initial-distribution hooks — while this base owns the fixed nuPlan schema shared by every model (the ego + agent-slot state layout, the discrete action set, and the observation-dict hashing/equality) and the observation model. The observation is factored by channel: this base holds a {channel: NuPlanObservationModel} map (observation_models) and composes it — sample_observation() perceives each channel of the clean observation built from the state, observation_log_probability() sums the per-channel densities, and encode_observation() (the single raw-observation seam) perceives each channel of the forward-only world’s raw reading into the same perceived space, so the belief filter and planner search operate on one consistent (encoded) observation. Two models differ in their observation only by the channel models they hold; a model that scores observations for a belief update needs every channel to provide a density (supports_density). A subclass whose observation is not a per-channel clean transform (e.g. a learned latent decoder) may instead override sample_observation(), observation_log_probability() and encode_observation() directly.

Parameters:

• discount_factor (float)

• action_presets (Sequence[Tuple[float, float]] | None)

• max_tracked_agents (int)

• observation_models (Dict[str, NuPlanObservationModel] | None)

• name (str | None)

action_presets

Discrete (acceleration, steering_angle) control pairs; the discrete action set is the indices into this list.

max_tracked_agents

Number of fixed agent slots in the state/observation.

observation_models

The per-channel {channel: NuPlanObservationModel} map composed to produce the observation, or None for a subclass that overrides the observation methods directly.

Note

This is an abstract base class and cannot be instantiated directly. See FactoredNuPlanModelPOMDP for a concrete reference implementation.

encode_observation(observation)

Perceive the world’s raw observation into the belief/planner observation space.

The forward-only world emits a raw, fully-detected observation; each channel model degrades its channel (per-slot detection gating, occlusion, additive noise) into the perceived observation the belief filter and planner search operate in. This is the single raw-observation seam — every other observation method works in the perceived (encoded) space. A subclass without channel models (observation_models is None, e.g. a learned latent model) inherits the identity default.

Parameters:

observation (Any) – The raw {ego, agents} observation emitted by the world.

Return type:

Any

Returns:

The perceived observation the belief and planner consume.

get_actions()

Discrete action set: indices into action_presets.

Return type:

List[int]

hash_action(action)

Return a hashable key consistent with action equality.

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

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

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

Parameters:

action (Any) – Action to hash.

Return type:

Hashable

Returns:

A hashable key derived from action.

hash_observation(observation)

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.

is_equal_observation(observation1, observation2)

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.

observation_log_probability(next_state, action, observations)

Log-density of observations under the per-channel perception given next_state.

Return type:

ndarray

Parameters:

• next_state (Any)

• action (Any)

• observations (Any)

abstractmethod sample_next_state(state, action, n_samples=1)

Sample n_samples next states for (state, action).

Return type:

ndarray

Parameters:

• state (Any)

• action (Any)

• n_samples (int)

sample_observation(next_state, action, n_samples=1)

Sample n_samples observations by perceiving next_state’s clean reading.

Return type:

Any

Parameters:

• next_state (Any)

• action (Any)

• n_samples (int)

abstractmethod transition_log_probability(state, action, next_states)

Log-density of next_states under the transition model for (state, action).

Return type:

ndarray

Parameters:

• state (Any)

• action (Any)

• next_states (Any)

Submodules

POMDPPlanners.environments.nuplan_pomdp.nuplan_generative_models.nuplan_factored_model_pomdp module

Reference concrete nuPlan generative model pairing dynamics with factored perception.

FactoredNuPlanModelPOMDP implements the NuPlanModelPOMDP interface by composing per-channel observation models — a FactoredAgentObservationModel on the agents channel (per-slot detection with range + occlusion gating and additive Gaussian pose noise) and an EgoObservationModel on the ego channel — the observation methods themselves are inherited from the base and driven by that map, together with the same gym-carla driving-quality reward the world uses. The transition dynamics are a documented identity placeholder for a specific study to replace with real (e.g. learned) motion.

State/observation layout and the shared reward are imported from nuplan_pomdp so world and model agree by construction.

Classes:

FactoredNuPlanModelPOMDP: Concrete nuPlan model with a factored-perception observation model.

class POMDPPlanners.environments.nuplan_pomdp.nuplan_generative_models.nuplan_factored_model_pomdp.FactoredNuPlanModelPOMDP(discount_factor, action_presets=None, max_tracked_agents=5, perception_range=50.0, occlusion_radius=1.5, pose_std=0.5, ego_std=0.01, detect_prob=0.95, desired_speed=8.0, out_lane_thresh=2.0, collision_penalty=100.0, observation=None, name=None)

Bases: NuPlanModelPOMDP

Concrete nuPlan generative model pairing placeholder dynamics with factored perception.

The observation is composed per channel (held as self.observation_models): a FactoredAgentObservationModel on agents (detection gated by perception range and geometric occlusion, additive Gaussian pose noise) and an EgoObservationModel on ego. The reward is the shared gym-carla driving-quality reward. The transition is a documented identity placeholder to be replaced with real dynamics per study.

Parameters:

• discount_factor (float)

• action_presets (Sequence[Tuple[float, float]] | None)

• max_tracked_agents (int)

• perception_range (float | None)

• occlusion_radius (float)

• pose_std (float)

• ego_std (float)

• detect_prob (float)

• desired_speed (float)

• out_lane_thresh (float)

• collision_penalty (float)

• observation (Dict[str, str] | None)

• name (str | None)

observation_models

The per-channel {channel: NuPlanObservationModel} map carrying the observation parameters (perception_range, occlusion_radius, pose_std, ego_std, detect_prob).

desired_speed

Target longitudinal speed (m/s) used by the reward.

out_lane_thresh

Lateral offset (m) beyond which the reward penalises off-route.

collision_penalty

Penalty scale applied on a terminal collision in the reward.

Example

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>>
>>> from POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp import (
...     AGENT_SLOT_WIDTH, EGO_STATE_WIDTH, LIGHT_SLOT_WIDTH)
>>> env = FactoredNuPlanModelPOMDP(discount_factor=0.95)
>>>
>>> width = (
...     EGO_STATE_WIDTH + env.max_tracked_agents * AGENT_SLOT_WIDTH + LIGHT_SLOT_WIDTH)
>>> state = np.zeros(width)
>>> action = env.get_actions()[0]
>>>
>>> next_state, observation, reward = env.sample_next_step(state, action)
>>> sorted(observation)
['agents', 'ego']
>>> env.is_terminal(state)
False

initial_observation_dist()

Get the initial observation distribution.

Return type:

Distribution

Returns:

Distribution over initial observations

Note

Subclasses must implement this method to define initial observations.

initial_state_dist()

Get the initial state distribution.

Return type:

Distribution

Returns:

Distribution over initial states

Note

Subclasses must implement this method to define the starting distribution.

is_terminal(state)

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.

reward(state, action, next_state=None)

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

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

Parameters:

• state (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)

Sample n_samples next states for (state, action).

Return type:

ndarray

Parameters:

• state (Any)

• action (Any)

• n_samples (int)

transition_log_probability(state, action, next_states)

Log-density of next_states under the transition model for (state, action).

Return type:

ndarray

Parameters:

• state (Any)

• action (Any)

• next_states (Any)

POMDPPlanners.environments.nuplan_pomdp.nuplan_generative_models.nuplan_kinematic_model_pomdp module

Concrete nuPlan model with a kinematic bicycle transition under the control preset.

KinematicNuPlanModelPOMDP replaces the identity-placeholder transition of FactoredNuPlanModelPOMDP with a real ego-motion model: it propagates the ego [x, y, yaw, vx, vy, lat, heading_err] forward one iteration under the selected (acceleration, steering_angle) control using a point-mass longitudinal model plus a bicycle yaw model, and closes the range on tracked agents by the distance the ego travelled. The factored observation model, reward, and terminal check are inherited unchanged.

This is what gives a planner a gradient toward accelerating: because acceleration now visibly increases the along-route speed the reward rewards, POMCPOW picks controls that actually move the car (the identity placeholder made every action look motionless).

Classes:

KinematicNuPlanModelPOMDP: Factored nuPlan model with a kinematic ego transition.

class POMDPPlanners.environments.nuplan_pomdp.nuplan_generative_models.nuplan_kinematic_model_pomdp.KinematicNuPlanModelPOMDP(discount_factor, dt=0.1, action_presets=None, max_tracked_agents=5, wheelbase=2.8, drag=0.05, collision_gap=5.0, collision_halfwidth=1.2, safe_distance=12.0, stop_gap=0.0, name=None, **kwargs)

Bases: FactoredNuPlanModelPOMDP

Factored nuPlan model whose transition is a kinematic bicycle propagation.

Parameters:

• discount_factor (float)

• dt (float)

• action_presets (Sequence[Tuple[float, float]] | None)

• max_tracked_agents (int)

• wheelbase (float)

• drag (float)

• collision_gap (float)

• collision_halfwidth (float)

• safe_distance (float)

• stop_gap (float)

• name (str | None)

• kwargs (Any)

dt

Integration step (seconds); must match the world’s fixed_delta_seconds.

wheelbase

Bicycle-model wheelbase (m) mapping steering angle to yaw rate.

drag

Linear speed-proportional deceleration coefficient (1/s).

collision_gap

Forward ego-frame distance (m) within which a present agent ahead is treated as a predicted collision by is_terminal().

collision_halfwidth

Lateral ego-frame half-corridor (m) within which a present agent ahead is treated as a predicted collision by is_terminal().

safe_distance

Lead gap (m) at/above which the reward targets the full desired_speed; the obstacle-aware target ramps down below it.

stop_gap

Lead gap (m) at/below which the obstacle-aware target speed is zero; 0.0 keeps the flat desired_speed.

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.nuplan_pomdp.nuplan_pomdp import (
...     AGENT_SLOT_WIDTH, EGO_STATE_WIDTH, LIGHT_SLOT_WIDTH)
>>> env = KinematicNuPlanModelPOMDP(discount_factor=0.95, dt=0.1)
>>>
>>> width = (
...     EGO_STATE_WIDTH + env.max_tracked_agents * AGENT_SLOT_WIDTH + LIGHT_SLOT_WIDTH)
>>> state = np.zeros(width)
>>> accelerate_action = 0  # (1.5, 0.0) accelerate straight
>>> next_state = env.sample_next_state(state, accelerate_action)
>>>
>>> bool(next_state[3] > 0.0)  # acceleration produced forward velocity
True

is_terminal(state)

Whether a present agent occupies the ego’s footprint just ahead.

Because this model does predict ego and agent motion, it can foresee running into the vehicle ahead. Any present agent slot within collision_gap metres forward and collision_halfwidth metres laterally (ego frame) is treated as a collision, which the inherited reward turns into the terminal collision_penalty (driving_quality_reward()).

Return type:

bool

Parameters:

state (Any)

reward(state, action, next_state=None)

Driving-quality reward whose target speed adapts to the nearest lead obstacle.

The bare parent reward tracks a fixed desired_speed and only charges the collision once an agent is inside the terminal box — a cliff that cannot be braked for at speed, while a large fixed penalty instead freezes the ego in traffic. This override follows an obstacle-aware desired speed: the target equals the full desired_speed when the lead gap is at least safe_distance, ramps linearly to zero at stop_gap, and is zero closer in. The ego is thus rewarded for driving when the road is clear and for slowing as an obstacle nears, without a separate penalty term that traps it at a standstill. stop_gap == 0 keeps the flat parent behaviour.

Return type:

float

Parameters:

• state (Any)

• action (Any)

• next_state (Any)

sample_next_state(state, action, n_samples=1)

Sample n_samples next states for (state, action).

Return type:

ndarray

Parameters:

• state (Any)

• action (Any)

• n_samples (int)

sample_next_state_batch(states, action)

Sample one next state per input state, all under the same action.

Used by particle filters: given N current particles and one action, draw N next states (one per particle) in a single vectorized call.

The default implementation falls back to a per-state Python loop delegating to sample_next_state(). Native-backed envs (those whose state-transition kernel exposes batch_sample(states_array)) should override to avoid the loop.

Parameters:

• states (Any) – A sequence (length N) or ndarray of shape (N, *dim) of input particles.

• action (Any) – A single action to apply to every particle.

Returns:

np.ndarray of shape (N, *dim). For structured envs (Tiger strings, Pacman tuples): a list of length N.

Return type:

ndarray

POMDPPlanners.environments.nuplan_pomdp.nuplan_generative_models.nuplan_model_pomdp module

Abstract planner-side generative-model interface for the nuPlan world.

NuPlanPOMDP is a forward-only world (no densities, no state injection). A planner instead carries a generative model as policy.environment — one that can sample transitions from an arbitrary state, score an observation against a state, and supply a reward. This module defines the interface that model must satisfy: NuPlanModelPOMDP owns only the nuPlan state/observation schema (agent-slot layout, discrete action set, observation-dict hashing/equality) and the observation model, and leaves every dynamic quantity — transition, observation, reward, terminal — abstract for a study- or task-specific subclass (e.g. a learned model) to fill in.

A runnable reference implementation with a fixed factored observation model lives in nuplan_factored_model_pomdp.

The schema (state/observation layout, action presets) is imported from nuplan_pomdp so world and model agree by construction.

Classes:

NuPlanModelPOMDP: Abstract generative-model interface over the nuPlan schema.

class POMDPPlanners.environments.nuplan_pomdp.nuplan_generative_models.nuplan_model_pomdp.NuPlanModelPOMDP(discount_factor, action_presets=None, max_tracked_agents=5, observation_models=None, name=None)

Bases: DiscreteActionsEnvironment

Abstract generative-model interface paired with the forward-only nuPlan world.

Concrete subclasses supply the dynamics — sample_next_state(), transition_log_probability(), reward(), is_terminal(), and the initial-distribution hooks — while this base owns the fixed nuPlan schema shared by every model (the ego + agent-slot state layout, the discrete action set, and the observation-dict hashing/equality) and the observation model. The observation is factored by channel: this base holds a {channel: NuPlanObservationModel} map (observation_models) and composes it — sample_observation() perceives each channel of the clean observation built from the state, observation_log_probability() sums the per-channel densities, and encode_observation() (the single raw-observation seam) perceives each channel of the forward-only world’s raw reading into the same perceived space, so the belief filter and planner search operate on one consistent (encoded) observation. Two models differ in their observation only by the channel models they hold; a model that scores observations for a belief update needs every channel to provide a density (supports_density). A subclass whose observation is not a per-channel clean transform (e.g. a learned latent decoder) may instead override sample_observation(), observation_log_probability() and encode_observation() directly.

Parameters:

• discount_factor (float)

• action_presets (Sequence[Tuple[float, float]] | None)

• max_tracked_agents (int)

• observation_models (Dict[str, NuPlanObservationModel] | None)

• name (str | None)

action_presets

Discrete (acceleration, steering_angle) control pairs; the discrete action set is the indices into this list.

max_tracked_agents

Number of fixed agent slots in the state/observation.

observation_models

The per-channel {channel: NuPlanObservationModel} map composed to produce the observation, or None for a subclass that overrides the observation methods directly.

Note

This is an abstract base class and cannot be instantiated directly. See FactoredNuPlanModelPOMDP for a concrete reference implementation.

encode_observation(observation)

Perceive the world’s raw observation into the belief/planner observation space.

The forward-only world emits a raw, fully-detected observation; each channel model degrades its channel (per-slot detection gating, occlusion, additive noise) into the perceived observation the belief filter and planner search operate in. This is the single raw-observation seam — every other observation method works in the perceived (encoded) space. A subclass without channel models (observation_models is None, e.g. a learned latent model) inherits the identity default.

Parameters:

observation (Any) – The raw {ego, agents} observation emitted by the world.

Return type:

Any

Returns:

The perceived observation the belief and planner consume.

get_actions()

Discrete action set: indices into action_presets.

Return type:

List[int]

hash_action(action)

Return a hashable key consistent with action equality.

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

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

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

Parameters:

action (Any) – Action to hash.

Return type:

Hashable

Returns:

A hashable key derived from action.

hash_observation(observation)

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.

is_equal_observation(observation1, observation2)

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.

observation_log_probability(next_state, action, observations)

Log-density of observations under the per-channel perception given next_state.

Return type:

ndarray

Parameters:

• next_state (Any)

• action (Any)

• observations (Any)

abstractmethod sample_next_state(state, action, n_samples=1)

Sample n_samples next states for (state, action).

Return type:

ndarray

Parameters:

• state (Any)

• action (Any)

• n_samples (int)

sample_observation(next_state, action, n_samples=1)

Sample n_samples observations by perceiving next_state’s clean reading.

Return type:

Any

Parameters:

• next_state (Any)

• action (Any)

• n_samples (int)

abstractmethod transition_log_probability(state, action, next_states)

Log-density of next_states under the transition model for (state, action).

Return type:

ndarray

Parameters:

• state (Any)

• action (Any)

• next_states (Any)