POMDPPlanners.environments.carla_pomdp.carla_perception package

Returns:

The updated (N, 5) track set.

class POMDPPlanners.environments.carla_pomdp.carla_perception.CarlaObservationModel[source]

Bases: ABC

Abstract single-channel observation model: clean channel -> perceived channel.

A concrete perception maps one observation channel’s clean, fully-detected value (built from a state by a model, or taken from the world’s raw reading) to the degraded value a planner sees. Implementations declare which channel they handle via channel and set supports_density to True when they also provide log_probability().

channel: The observation-dict key this model handles (e.g. "gnss" or "agents").

supports_density: Whether log_probability() is implemented. Sample-only channels leave this False and are usable only where sampling is needed.

Note

This is an abstract base class and cannot be instantiated directly.

channel: str = ''

log_probability(clean_channel, channel_observation)[source]

Log-density of channel_observation given the clean channel value.

Parameters:

clean_channel (Any) – The noise-free value of this channel built from a state.
channel_observation (Any) – The channel value whose likelihood is scored.

Return type:

Returns:

The channel’s observation log-probability.

Raises:

NotImplementedError – If this is a sample-only channel without a density.

abstractmethod perceive(clean_channel)[source]

Sample this channel’s perceived value from its clean, fully-detected value.

Parameters:: clean_channel (Any) – The noise-free value of this channel, built from a state or taken from the world’s raw reading.
Return type:: Any
Returns:: The perceived value of the same channel.

supports_density: bool = False

class POMDPPlanners.environments.carla_pomdp.carla_perception.CarlaPerceptionPipeline(max_tracked_agents=5, perception=None, tracker=None, sensor_fusion=True, stop_for_traffic_lights=True, obstacle_detection_range=30.0, dt=0.05, lidar_corridor_halfwidth=1.5, tracks=None)[source]

Bases: object

Standalone perception + prediction stage: raw observation -> agent slots + obstacle.

Composes a PerceptionModel (single-frame) and a MotionTracker (temporal), owns the tracker state, and produces the ego-frame agent block a belief stamps onto its particles plus a fused forward-obstacle distance. Immutable: process() returns a PerceptionOutput carrying a successor pipeline with the advanced tracks.

It is a whole-observation sensor-fusion stage (fusing lidar/camera into the agent block), not a per-channel CarlaObservationModel: it perceives a whole observation and exposes no observation density. The world threads its tracker state forward via process(); the perceive() convenience method returns a single perceived observation without carrying the advanced tracks, for callers that only need one reading.

Parameters:

max_tracked_agents (int)
perception (PerceptionModel | None)
tracker (MotionTracker | None)
sensor_fusion (bool)
stop_for_traffic_lights (bool)
obstacle_detection_range (float)
dt (float)
lidar_corridor_halfwidth (float)
tracks (ndarray | None)

max_tracked_agents: Number of agent slots produced in the agent block.

perception: The single-frame PerceptionModel.

tracker: The temporal MotionTracker.

sensor_fusion: Whether the fused lidar/camera forward obstacle is reported.

stop_for_traffic_lights: Whether a red/amber light is reported as a forward obstacle.

obstacle_detection_range: Only obstacles nearer than this (m) are reported.

dt: Tracker time step (s).

Example

>>> import numpy as np
>>> pipeline = CarlaPerceptionPipeline(max_tracked_agents=1)
>>> obs = {"lidar": np.zeros((0, 4)), "camera": np.zeros((8, 8, 3), dtype=np.uint8)}
>>> output = pipeline.process(obs)
>>> output.agent_rows.shape
(1, 5)

perceive(clean_observation)[source]

Perceive one observation, replacing its agents block with tracked slots.

Unlike process() this discards the advanced tracker state: it yields a single perceived observation for callers that only need one reading. Use process() when the successor pipeline (advanced tracks) must be threaded forward.

Parameters:: clean_observation (Mapping[str, Any]) – The world’s raw observation dict.
Return type:: dict
Returns:: The perceived observation dict with the tracked agents block.

process(observation)[source]

Perceive and track one observation into an agent block and forward obstacle.

Parameters:: observation (Mapping[str, Any]) – The world’s raw observation dict.
Return type:: PerceptionOutput
Returns:: A PerceptionOutput with the agent block, the fused obstacle distance (or None), and the successor pipeline carrying the advanced tracks.

class POMDPPlanners.environments.carla_pomdp.carla_perception.Detections(vehicle_positions, forward_clearance, traffic_light)[source]

Bases: object

Single-frame perception output.

Parameters:

vehicle_positions (ndarray)
forward_clearance (float)
traffic_light (ndarray)

vehicle_positions: (M, 3) ego-frame vehicle detections [rel_x, rel_y, confidence].

forward_clearance: Fused lidar+camera forward-obstacle distance (m); large when clear.

traffic_light: [should_stop, distance_m] stop signal for a red/amber light.

forward_clearance: float

traffic_light: ndarray

vehicle_positions: ndarray

class POMDPPlanners.environments.carla_pomdp.carla_perception.FactoredAgentObservationModel(max_tracked_agents=5, perception_range=50.0, occlusion_radius=1.5, pose_std=0.5, detect_prob=0.95)[source]

Bases: CarlaObservationModel

Reference agent perception: per-slot detection gating plus additive Gaussian pose noise.

Each agent slot is detected only when in perception range and not geometrically occluded by another agent on the ego->target sight line; a detected agent’s pose is corrupted with additive Gaussian noise. Provides both a sampler and a matching density, so it can back a scoring generative model.

Parameters:

max_tracked_agents (int)
perception_range (float | None)
occlusion_radius (float)
pose_std (float)
detect_prob (float)

max_tracked_agents: Number of fixed agent slots in the agents block.

perception_range: Metres beyond which an agent is undetectable (None disables the range gate).

occlusion_radius: Sight-line blocking radius among agents.

pose_std: Std of Gaussian noise on a detected agent’s pose measurement.

detect_prob: Probability of detecting a visible agent; 1 - detect_prob is the miss rate scored by the density.

Example

>>> import numpy as np
>>> np.random.seed(0)
>>> model = FactoredAgentObservationModel(max_tracked_agents=1, perception_range=50.0)
>>> agents = np.array([1.0, 10.0, 0.0, 0.0, 0.0])
>>> float(model.perceive(agents)[0])  # near agent detected
1.0

channel: str = 'agents'

log_probability(clean_channel, channel_observation)[source]

Log-density of channel_observation given the clean channel value.

Parameters:

clean_channel (Any) – The noise-free value of this channel built from a state.
channel_observation (Any) – The channel value whose likelihood is scored.

Return type:

Returns:

The channel’s observation log-probability.

Raises:

NotImplementedError – If this is a sample-only channel without a density.

perceive(clean_channel)[source]

Sample this channel’s perceived value from its clean, fully-detected value.

Parameters:: clean_channel (Any) – The noise-free value of this channel, built from a state or taken from the world’s raw reading.
Return type:: ndarray
Returns:: The perceived value of the same channel.

render(clean_channel, noisy)[source]

Gate the clean agent block per slot, optionally adding Gaussian pose noise.

Parameters:

clean_channel (Any) – The noise-free flat agents block.
noisy (bool) – When True, add pose Gaussian noise (the sampler path); when False, return the gated but noise-free block.

Return type:

Returns:

The perceived flat agents block.

supports_density: bool = True

class POMDPPlanners.environments.carla_pomdp.carla_perception.GnssObservationModel(gnss_std=1e-05)[source]

Bases: CarlaObservationModel

GNSS channel corrupted by additive Gaussian noise, with a matching density.

Parameters:: gnss_std (float)

gnss_std: Std of the zero-mean Gaussian noise added to the 2-D gnss reading.

Example

>>> import numpy as np
>>> np.random.seed(0)
>>> model = GnssObservationModel(gnss_std=1e-5)
>>> perceived = model.perceive(np.zeros(2))
>>> perceived.shape
(2,)

channel: str = 'gnss'

log_probability(clean_channel, channel_observation)[source]

Log-density of channel_observation given the clean channel value.

Parameters:

clean_channel (Any) – The noise-free value of this channel built from a state.
channel_observation (Any) – The channel value whose likelihood is scored.

Return type:

Returns:

The channel’s observation log-probability.

Raises:

NotImplementedError – If this is a sample-only channel without a density.

perceive(clean_channel)[source]

Sample this channel’s perceived value from its clean, fully-detected value.

Parameters:: clean_channel (Any) – The noise-free value of this channel, built from a state or taken from the world’s raw reading.
Return type:: ndarray
Returns:: The perceived value of the same channel.

supports_density: bool = True

class POMDPPlanners.environments.carla_pomdp.carla_perception.LidarCameraPerceptionModel(lidar_corridor_halfwidth=1.5, traffic_light_source='camera')[source]

Default CARLA perception: lidar vehicle clustering + camera obstacle/traffic-light cues.

Vehicles are clustered from the lidar cloud, the forward obstacle fuses the lidar corridor clearance with the camera looming cue, and the traffic light is inferred from the camera image (no ground-truth channel is consulted) unless traffic_light_source='channel'.

Parameters:

lidar_corridor_halfwidth (float)
traffic_light_source (str)

lidar_corridor_halfwidth: Half-width (m) of the forward corridor scanned for an obstacle.

traffic_light_source: 'camera' infers the light from the RGB image; 'channel' reads it from the observation’s traffic_light key.

Example

>>> import numpy as np
>>> model = LidarCameraPerceptionModel()
>>> obs = {"lidar": np.zeros((0, 4)), "camera": np.zeros((8, 8, 3), dtype=np.uint8)}
>>> model.detect(obs).vehicle_positions.shape
(0, 3)

detect(observation)[source]

Perceive vehicles, a forward obstacle, and a traffic light from one observation.

Parameters:: observation (Mapping[str, Any]) – The world’s raw observation dict (lidar/camera/agents/ traffic_light keys, any subset present).
Return type:: Detections
Returns:: The single-frame Detections for this observation.

class POMDPPlanners.environments.carla_pomdp.carla_perception.MotionTracker[source]

Bases: ABC

Abstract temporal prediction interface: prior tracks + detections -> tracks with velocity.

Note

This is an abstract base class and cannot be instantiated directly.

abstractmethod update(tracks, vehicle_positions, dt)[source]

Advance the tracker one step and return the (N, 5) track set with velocity.

Parameters:

tracks (Optional[ndarray]) – Prior (N, 5) tracks [rel_x, rel_y, vx, vy, confidence], or None.
vehicle_positions (ndarray) – (M, 3) detections [rel_x, rel_y, confidence].
dt (float) – Time step (s) since the previous update.

Return type:

Returns:

The updated (N, 5) track set.

class POMDPPlanners.environments.carla_pomdp.carla_perception.OracleAgentPerceptionModel(max_tracked_agents=5, lidar_corridor_halfwidth=1.5)[source]

Ground-truth-agent perception for studies/tests: vehicles from the agents channel.

Reads the observation’s ground-truth agents rows as vehicle detections (position with confidence 1.0), still fusing lidar/camera for the forward obstacle and reading the light from the traffic_light channel. Use when a study wants exact agent positions rather than inferred ones; the tracker then re-estimates velocity from the position stream.

Parameters:

max_tracked_agents (int)
lidar_corridor_halfwidth (float)

max_tracked_agents: Number of fixed agent slots in the ground-truth agents channel.

lidar_corridor_halfwidth: Half-width (m) of the forward obstacle corridor.

Example

>>> import numpy as np
>>> model = OracleAgentPerceptionModel(max_tracked_agents=1)
>>> obs = {"agents": np.array([1.0, 8.0, 0.0, 0.0, 5.0])}
>>> model.detect(obs).vehicle_positions.shape
(1, 3)

detect(observation)[source]

Perceive vehicles, a forward obstacle, and a traffic light from one observation.

Parameters:: observation (Mapping[str, Any]) – The world’s raw observation dict (lidar/camera/agents/ traffic_light keys, any subset present).
Return type:: Detections
Returns:: The single-frame Detections for this observation.

class POMDPPlanners.environments.carla_pomdp.carla_perception.PerceptionModel[source]

Bases: ABC

Abstract single-frame perception interface: raw sensor dict -> Detections.

Note

This is an abstract base class and cannot be instantiated directly.

abstractmethod detect(observation)[source]

Perceive vehicles, a forward obstacle, and a traffic light from one observation.

Parameters:: observation (Mapping[str, Any]) – The world’s raw observation dict (lidar/camera/agents/ traffic_light keys, any subset present).
Return type:: Detections
Returns:: The single-frame Detections for this observation.

class POMDPPlanners.environments.carla_pomdp.carla_perception.PerceptionOutput(agent_rows, obstacle_distance, pipeline)[source]

Bases: object

Per-step pipeline output consumed by the belief.

Parameters:

agent_rows (ndarray)
obstacle_distance (float | None)
pipeline (CarlaPerceptionPipeline)

agent_rows: (K, AGENT_SLOT_WIDTH) ego-frame agent slots [present, rel_x, rel_y, rel_yaw, rel_speed].

obstacle_distance: Nearest fused forward-obstacle distance (m), or None when clear.

pipeline: Successor pipeline carrying the advanced tracker state.

agent_rows: ndarray

obstacle_distance: float | None

pipeline: CarlaPerceptionPipeline

POMDPPlanners.environments.carla_pomdp.carla_perception.available_observation_models(channel)[source]

Return the catalog names registered for channel, sorted.

Return type:: List[str]
Parameters:: channel (str)

POMDPPlanners.environments.carla_pomdp.carla_perception.build_observation_model(channel, name, **kwargs)[source]

Instantiate the observation model registered under (channel, name).

Parameters:

channel (str) – The observation-dict key to resolve the model for.
name (str) – The registered catalog name within that channel.
**kwargs (Any) – Forwarded to the registered factory.

Return type:

CarlaObservationModel

Returns:

The instantiated per-channel observation model.

Raises:

KeyError – If no model is registered under (channel, name).

POMDPPlanners.environments.carla_pomdp.carla_perception.camera_looming_cue(image)[source]

Fraction in [0, 1] of the lower-centre view filled by a large, dark near object.

A close vehicle or wall directly ahead darkens and flattens the lower-centre of the front camera. This coarse cue corroborates a lidar detection; it is deliberately conservative so it cannot, on its own, manufacture phantom braking (see fuse_forward_obstacle()).

Parameters:: image (Optional[ndarray]) – (H, W, 3) RGB frame, or None when no frame is available.
Return type:: float
Returns:: The fraction of lower-centre pixels below a darkness threshold.

POMDPPlanners.environments.carla_pomdp.carla_perception.fuse_forward_obstacle(lidar_clearance, camera_cue, max_range=50.0, camera_trigger=0.35, camera_range=12.0)[source]

Fuse the lidar clearance and camera cue into a single forward-obstacle distance.

Lidar is authoritative: whenever it reports an in-range obstacle that distance is used. The camera only acts when the lidar corridor is clear (lidar_clearance >= max_range) yet the looming cue is strong, in which case a conservative camera_range obstacle is assumed so a lidar miss (sparse returns on a dark, close surface) still triggers caution.

Parameters:

lidar_clearance (float) – Forward clearance from lidar_forward_clearance().
camera_cue (float) – Looming fraction from camera_looming_cue().
max_range (float) – Sensor range that denotes “lidar corridor clear”.
camera_trigger (float) – Cue value at/above which the camera flags a near obstacle.
camera_range (float) – Distance assumed for a camera-only detection.

Return type:

Returns:

Distance (m) to the nearest forward obstacle; max_range when nothing is detected.

POMDPPlanners.environments.carla_pomdp.carla_perception.lidar_forward_clearance(points, corridor_halfwidth=1.5, z_min=-2.0, z_max=1.0, max_range=50.0)[source]

Distance (m) to the nearest forward in-corridor lidar return, or max_range if clear.

Parameters:

points (Optional[ndarray]) – (N, 4) lidar cloud [x, y, z, intensity] in the sensor frame (x forward, y right, z up), or None/empty when no scan is available.
corridor_halfwidth (float) – Half-width (m) of the forward corridor watched for obstacles.
z_min (float) – Lower height bound (m); returns below it (the ground) are ignored.
z_max (float) – Upper height bound (m); returns above it (overhead structure) are ignored.
max_range (float) – Value returned when the corridor holds no qualifying return.

Return type:

Returns:

The minimum forward x among corridor returns, else max_range.

POMDPPlanners.environments.carla_pomdp.carla_perception.lidar_vehicle_detections(points, cell_size=0.5, z_min=-2.0, z_max=1.0, detection_range=50.0, min_points=4, max_extent=8.0, confident_points=40)[source]

Cluster a lidar cloud into ego-frame vehicle detections [rel_x, rel_y, confidence].

A single-frame perception stage — the first step of a real perception pipeline that replaces the ground-truth agents oracle. It drops ground / overhead returns, groups the rest into BEV clusters (8-connected occupied cells), and keeps the vehicle-sized ones (rejecting walls / buildings by extent and noise by point count). Confidence rises with the return count. Positions are in the ego frame (rel_x forward, rel_y left); velocity and heading are left to a downstream tracker.

Parameters:

points (Optional[ndarray]) – (N, 4) cloud [x, y, z, intensity] in the sensor frame, or None.
cell_size (float) – BEV grid cell size (m); adjacent occupied cells merge into one cluster.
z_min (float) – Lower height bound (m); returns below it (the ground) are dropped.
z_max (float) – Upper height bound (m); returns above it (overhead structure) are dropped.
detection_range (float) – Clusters beyond this range (m) are dropped.
min_points (int) – Clusters with fewer returns are discarded as noise.
max_extent (float) – Clusters wider than this (m) are discarded as walls / buildings.
confident_points (int) – Return count at/above which confidence saturates to 1.0.

Return type:

Returns:

An (M, 3) array of [rel_x, rel_y, confidence] detections (empty if none).

POMDPPlanners.environments.carla_pomdp.carla_perception.register_observation_model(channel, name)[source]

Register an observation-model factory under (channel, name) for user selection.

Parameters:

channel (str) – The observation-dict key the model handles (e.g. "gnss", "agents").
name (str) – The catalog name the user selects the model by within that channel.

Return type:

Callable[[TypeVar(_FactoryT, bound= Callable[..., CarlaObservationModel])], TypeVar(_FactoryT, bound= Callable[..., CarlaObservationModel])]

Returns:

A decorator that registers the factory (a class or callable returning a CarlaObservationModel) and returns it unchanged (its type is preserved).

POMDPPlanners.environments.carla_pomdp.carla_perception.traffic_light_from_camera(image, focal_pixels=200.0, bulb_diameter=0.3, roi_fraction=0.6, min_bulb_pixels=4, min_distance=3.0, max_distance=50.0)[source]

Infer [should_stop, distance_m] from a red/amber bulb in the upper camera frame.

A lightweight classic-vision stand-in for a learned traffic-light detector, so the light is perceived from the image rather than read from a ground-truth channel. It thresholds the upper region of interest for red and amber bulbs, takes the largest qualifying blob, and estimates its forward distance from the pinhole relation distance = focal_pixels * bulb_diameter / bulb_pixel_size. A green or absent bulb yields [0, 0] (no stop).

Note

Distance from a single low-resolution frame is inherently coarse; focal_pixels is a calibration constant that should match the camera intrinsics, and the result is clamped to [min_distance, max_distance]. The output matches the world’s traffic_light channel layout so it is a drop-in replacement for traffic_light_stop_distance().

Parameters:

image (Optional[ndarray]) – (H, W, 3) RGB frame, or None when no frame is available.
focal_pixels (float) – Camera focal length (px) used by the pinhole distance estimate.
bulb_diameter (float) – Physical traffic-light bulb diameter (m).
roi_fraction (float) – Fraction of the frame height (from the top) searched for a bulb.
min_bulb_pixels (int) – Blobs with fewer pixels than this are rejected as noise.
min_distance (float) – Lower clamp (m) on the estimated stop distance.
max_distance (float) – Upper clamp (m) on the estimated stop distance.

Return type:

Returns:

[1.0, distance_m] when a red/amber bulb is detected, else [0.0, 0.0].

POMDPPlanners.environments.carla_pomdp.carla_perception.traffic_light_stop_distance(traffic_light, max_range=50.0)[source]

Forward distance (m) to a red/yellow stop line, or max_range if clear.

Turns the world’s traffic_light observation into an obstacle distance so a red light can be injected as a virtual stop-obstacle, exactly like a stopped vehicle.

Parameters:

traffic_light (Optional[ndarray]) – [should_stop, distance_m] from the observation, or None.
max_range (float) – Value returned when there is no active stop signal.

Return type:

Returns:

The stop-line distance when should_stop is set and positive, else max_range.

POMDPPlanners.environments.carla_pomdp.carla_perception.update_tracks(tracks, detections, dt, gate=4.0, alpha=0.5, beta=0.3, new_confidence=0.5, confidence_gain=0.25, confidence_decay=0.25, min_confidence=0.1, track_range=50.0)[source]

Advance the tracker one step and return the new (K, 5) track set.

Parameters:

tracks (Optional[ndarray]) – Prior (N, 5) tracks [rel_x, rel_y, vx, vy, confidence], or None.
detections (Optional[ndarray]) – (M, 3) detections [rel_x, rel_y, confidence] from the detector.
dt (float) – Time step (s) since the previous update.
gate (float) – Max distance (m) at which a detection is associated with a predicted track.
alpha (float) – Position correction gain toward the measurement.
beta (float) – Velocity correction gain from the position residual.
new_confidence (float) – Confidence a newly spawned track starts with.
confidence_gain (float) – Confidence added when a track is matched to a detection.
confidence_decay (float) – Confidence removed when a track is missed (coasted).
min_confidence (float) – Tracks below this confidence are evicted.
track_range (float) – Tracks beyond this ego-frame range (m) are evicted.

Return type:

Returns:

The updated (K, 5) track set (empty (0, 5) when there is nothing to track).

Subpackages

POMDPPlanners.environments.carla_pomdp.carla_perception.observation_models package

Submodules

POMDPPlanners.environments.carla_pomdp.carla_perception.carla_pipeline module

Standalone, swappable perception + prediction pipeline for the CARLA planner.

In the intended architecture the belief is a plain particle filter over the ego and does not run perception itself. This module holds that perception, decoupled from both the world and the belief, so a user can swap in a different perception model without touching either. It turns the world’s raw multi-modal observation (lidar/camera/traffic_light/agents) into the two things a planner-side belief needs each step:

the ego-frame agent block — the max_tracked_agents nearest vehicles as [present, rel_x, rel_y, rel_yaw, rel_speed] slots — perceived and tracked (with velocity) from the sensors, and
a single fused forward-obstacle distance (lidar corridor + camera looming cue, optionally a red/amber traffic light), or None when the way ahead is clear.

Two swappable interfaces compose the pipeline:

PerceptionModel — the single-frame stage (raw sensors -> Detections).
MotionTracker — the temporal stage that estimates agent velocity over time.

The defaults (LidarCameraPerceptionModel + AlphaBetaTracker) reconstruct a real autonomous-driving perception stack: vehicles are clustered from the lidar cloud, their velocity is estimated by a constant-velocity tracker whose coasting carries a briefly occluded vehicle through a dropout, and the traffic light is inferred from the camera image rather than read from a ground-truth channel. OracleAgentPerceptionModel is provided for studies/tests that want exact agent positions instead of inferred ones.

The pipeline is immutable and owns the tracker state: CarlaPerceptionPipeline.process() returns a PerceptionOutput carrying a successor pipeline with the advanced tracks, so a belief can thread perception forward without holding any perception state of its own.

Classes:: PerceptionModel: Abstract single-frame perception interface. MotionTracker: Abstract temporal (velocity-estimating) tracking interface. LidarCameraPerceptionModel: Default lidar+camera perception with camera traffic lights. OracleAgentPerceptionModel: Ground-truth-agent perception for studies/tests. AlphaBetaTracker: Default constant-velocity multi-object tracker. CarlaPerceptionPipeline: Composed, immutable perception + prediction stage.

class POMDPPlanners.environments.carla_pomdp.carla_perception.carla_pipeline.AlphaBetaTracker[source]

Bases: MotionTracker

Default constant-velocity multi-object tracker (alpha-beta) over vehicle detections.

A thin MotionTracker wrapper around update_tracks(); its coasting of undetected tracks is what carries a briefly occluded vehicle through a sensor dropout.

Example

>>> import numpy as np
>>> tracker = AlphaBetaTracker()
>>> tracks = tracker.update(None, np.array([[8.0, 0.0, 1.0]]), dt=0.05)
>>> tracks.shape
(1, 5)

update(tracks, vehicle_positions, dt)[source]

Advance the tracker one step and return the (N, 5) track set with velocity.

Parameters:

tracks (Optional[ndarray]) – Prior (N, 5) tracks [rel_x, rel_y, vx, vy, confidence], or None.
vehicle_positions (ndarray) – (M, 3) detections [rel_x, rel_y, confidence].
dt (float) – Time step (s) since the previous update.

Return type:

Returns:

The updated (N, 5) track set.

class POMDPPlanners.environments.carla_pomdp.carla_perception.carla_pipeline.CarlaPerceptionPipeline(max_tracked_agents=5, perception=None, tracker=None, sensor_fusion=True, stop_for_traffic_lights=True, obstacle_detection_range=30.0, dt=0.05, lidar_corridor_halfwidth=1.5, tracks=None)[source]

Bases: object

Standalone perception + prediction stage: raw observation -> agent slots + obstacle.

Composes a PerceptionModel (single-frame) and a MotionTracker (temporal), owns the tracker state, and produces the ego-frame agent block a belief stamps onto its particles plus a fused forward-obstacle distance. Immutable: process() returns a PerceptionOutput carrying a successor pipeline with the advanced tracks.

It is a whole-observation sensor-fusion stage (fusing lidar/camera into the agent block), not a per-channel CarlaObservationModel: it perceives a whole observation and exposes no observation density. The world threads its tracker state forward via process(); the perceive() convenience method returns a single perceived observation without carrying the advanced tracks, for callers that only need one reading.

Parameters:

max_tracked_agents (int)
perception (PerceptionModel | None)
tracker (MotionTracker | None)
sensor_fusion (bool)
stop_for_traffic_lights (bool)
obstacle_detection_range (float)
dt (float)
lidar_corridor_halfwidth (float)
tracks (ndarray | None)

max_tracked_agents: Number of agent slots produced in the agent block.

perception: The single-frame PerceptionModel.

tracker: The temporal MotionTracker.

sensor_fusion: Whether the fused lidar/camera forward obstacle is reported.

stop_for_traffic_lights: Whether a red/amber light is reported as a forward obstacle.

obstacle_detection_range: Only obstacles nearer than this (m) are reported.

dt: Tracker time step (s).

Example

>>> import numpy as np
>>> pipeline = CarlaPerceptionPipeline(max_tracked_agents=1)
>>> obs = {"lidar": np.zeros((0, 4)), "camera": np.zeros((8, 8, 3), dtype=np.uint8)}
>>> output = pipeline.process(obs)
>>> output.agent_rows.shape
(1, 5)

perceive(clean_observation)[source]

Perceive one observation, replacing its agents block with tracked slots.

Unlike process() this discards the advanced tracker state: it yields a single perceived observation for callers that only need one reading. Use process() when the successor pipeline (advanced tracks) must be threaded forward.

Parameters:: clean_observation (Mapping[str, Any]) – The world’s raw observation dict.
Return type:: dict
Returns:: The perceived observation dict with the tracked agents block.

process(observation)[source]

Perceive and track one observation into an agent block and forward obstacle.

Parameters:: observation (Mapping[str, Any]) – The world’s raw observation dict.
Return type:: PerceptionOutput
Returns:: A PerceptionOutput with the agent block, the fused obstacle distance (or None), and the successor pipeline carrying the advanced tracks.

class POMDPPlanners.environments.carla_pomdp.carla_perception.carla_pipeline.Detections(vehicle_positions, forward_clearance, traffic_light)[source]

Bases: object

Single-frame perception output.

Parameters:

vehicle_positions (ndarray)
forward_clearance (float)
traffic_light (ndarray)

vehicle_positions: (M, 3) ego-frame vehicle detections [rel_x, rel_y, confidence].

forward_clearance: Fused lidar+camera forward-obstacle distance (m); large when clear.

traffic_light: [should_stop, distance_m] stop signal for a red/amber light.

forward_clearance: float

traffic_light: ndarray

vehicle_positions: ndarray

class POMDPPlanners.environments.carla_pomdp.carla_perception.carla_pipeline.LidarCameraPerceptionModel(lidar_corridor_halfwidth=1.5, traffic_light_source='camera')[source]

Default CARLA perception: lidar vehicle clustering + camera obstacle/traffic-light cues.

Vehicles are clustered from the lidar cloud, the forward obstacle fuses the lidar corridor clearance with the camera looming cue, and the traffic light is inferred from the camera image (no ground-truth channel is consulted) unless traffic_light_source='channel'.

Parameters:

lidar_corridor_halfwidth (float)
traffic_light_source (str)

lidar_corridor_halfwidth: Half-width (m) of the forward corridor scanned for an obstacle.

traffic_light_source: 'camera' infers the light from the RGB image; 'channel' reads it from the observation’s traffic_light key.

Example

>>> import numpy as np
>>> model = LidarCameraPerceptionModel()
>>> obs = {"lidar": np.zeros((0, 4)), "camera": np.zeros((8, 8, 3), dtype=np.uint8)}
>>> model.detect(obs).vehicle_positions.shape
(0, 3)

detect(observation)[source]

Perceive vehicles, a forward obstacle, and a traffic light from one observation.

Parameters:: observation (Mapping[str, Any]) – The world’s raw observation dict (lidar/camera/agents/ traffic_light keys, any subset present).
Return type:: Detections
Returns:: The single-frame Detections for this observation.

class POMDPPlanners.environments.carla_pomdp.carla_perception.carla_pipeline.MotionTracker[source]

Bases: ABC

Abstract temporal prediction interface: prior tracks + detections -> tracks with velocity.

Note

This is an abstract base class and cannot be instantiated directly.

abstractmethod update(tracks, vehicle_positions, dt)[source]

Advance the tracker one step and return the (N, 5) track set with velocity.

Parameters:

tracks (Optional[ndarray]) – Prior (N, 5) tracks [rel_x, rel_y, vx, vy, confidence], or None.
vehicle_positions (ndarray) – (M, 3) detections [rel_x, rel_y, confidence].
dt (float) – Time step (s) since the previous update.

Return type:

Returns:

The updated (N, 5) track set.

class POMDPPlanners.environments.carla_pomdp.carla_perception.carla_pipeline.OracleAgentPerceptionModel(max_tracked_agents=5, lidar_corridor_halfwidth=1.5)[source]

Ground-truth-agent perception for studies/tests: vehicles from the agents channel.

Reads the observation’s ground-truth agents rows as vehicle detections (position with confidence 1.0), still fusing lidar/camera for the forward obstacle and reading the light from the traffic_light channel. Use when a study wants exact agent positions rather than inferred ones; the tracker then re-estimates velocity from the position stream.

Parameters:

max_tracked_agents (int)
lidar_corridor_halfwidth (float)

max_tracked_agents: Number of fixed agent slots in the ground-truth agents channel.

lidar_corridor_halfwidth: Half-width (m) of the forward obstacle corridor.

Example

>>> import numpy as np
>>> model = OracleAgentPerceptionModel(max_tracked_agents=1)
>>> obs = {"agents": np.array([1.0, 8.0, 0.0, 0.0, 5.0])}
>>> model.detect(obs).vehicle_positions.shape
(1, 3)

detect(observation)[source]

Perceive vehicles, a forward obstacle, and a traffic light from one observation.

Parameters:: observation (Mapping[str, Any]) – The world’s raw observation dict (lidar/camera/agents/ traffic_light keys, any subset present).
Return type:: Detections
Returns:: The single-frame Detections for this observation.

class POMDPPlanners.environments.carla_pomdp.carla_perception.carla_pipeline.PerceptionModel[source]

Bases: ABC

Abstract single-frame perception interface: raw sensor dict -> Detections.

Note

This is an abstract base class and cannot be instantiated directly.

abstractmethod detect(observation)[source]

Perceive vehicles, a forward obstacle, and a traffic light from one observation.

Parameters:: observation (Mapping[str, Any]) – The world’s raw observation dict (lidar/camera/agents/ traffic_light keys, any subset present).
Return type:: Detections
Returns:: The single-frame Detections for this observation.

class POMDPPlanners.environments.carla_pomdp.carla_perception.carla_pipeline.PerceptionOutput(agent_rows, obstacle_distance, pipeline)[source]

Bases: object

Per-step pipeline output consumed by the belief.

Parameters:

agent_rows (ndarray)
obstacle_distance (float | None)
pipeline (CarlaPerceptionPipeline)

agent_rows: (K, AGENT_SLOT_WIDTH) ego-frame agent slots [present, rel_x, rel_y, rel_yaw, rel_speed].

obstacle_distance: Nearest fused forward-obstacle distance (m), or None when clear.

pipeline: Successor pipeline carrying the advanced tracker state.

agent_rows: ndarray

obstacle_distance: float | None

pipeline: CarlaPerceptionPipeline

POMDPPlanners.environments.carla_pomdp.carla_perception.carla_sensors module

Forward-obstacle perception from the CARLA lidar and front camera.

The planner-side models carry other vehicles as fixed ego-frame agent slots, but those slots are a nearest-K-vehicle abstraction: they miss non-vehicle geometry, vehicles beyond the tracked count, and — because the kinematic model never propagates agent lateral motion — a car cutting into the lane until it is already dead ahead. The raycast lidar does not share those blind spots: it measures the true distance to whatever solid is in front of the ego right now, so it is the reliable signal for “is something about to be hit.”

This module turns the raw sensors into a single scalar — the forward clearance (metres to the nearest in-corridor obstacle, or the sensor range when clear) — with pure, testable helpers:

lidar_forward_clearance() — the authoritative geometric range from the (N, 4) point cloud, gated to a forward driving corridor and a vehicle-height band (so ground and overhead returns are ignored).
camera_looming_cue() — a lightweight secondary cue in [0, 1] from the front RGB frame that rises when a large near object fills the lower-centre view; it corroborates a lidar detection and can flag a near obstacle when the lidar returns are sparse.
fuse_forward_obstacle() — combines the two into the reported obstacle distance, lidar taking precedence and the camera only shortening the range when the lidar sees nothing.
traffic_light_stop_distance() — turns the traffic_light observation into a stop distance, so a red light can be injected as a virtual obstacle just like a stopped car.

These helpers are composed by CarlaPerceptionPipeline, whose output the belief stamps as agent slots so the planner’s terminal-collision and headway logic brake for them — i.e. the sensors are used by the planner, not by a hidden actuation override.

POMDPPlanners.environments.carla_pomdp.carla_perception.carla_sensors.camera_looming_cue(image)[source]

Fraction in [0, 1] of the lower-centre view filled by a large, dark near object.

A close vehicle or wall directly ahead darkens and flattens the lower-centre of the front camera. This coarse cue corroborates a lidar detection; it is deliberately conservative so it cannot, on its own, manufacture phantom braking (see fuse_forward_obstacle()).

Parameters:: image (Optional[ndarray]) – (H, W, 3) RGB frame, or None when no frame is available.
Return type:: float
Returns:: The fraction of lower-centre pixels below a darkness threshold.

POMDPPlanners.environments.carla_pomdp.carla_perception.carla_sensors.fuse_forward_obstacle(lidar_clearance, camera_cue, max_range=50.0, camera_trigger=0.35, camera_range=12.0)[source]

Fuse the lidar clearance and camera cue into a single forward-obstacle distance.

Lidar is authoritative: whenever it reports an in-range obstacle that distance is used. The camera only acts when the lidar corridor is clear (lidar_clearance >= max_range) yet the looming cue is strong, in which case a conservative camera_range obstacle is assumed so a lidar miss (sparse returns on a dark, close surface) still triggers caution.

Parameters:

lidar_clearance (float) – Forward clearance from lidar_forward_clearance().
camera_cue (float) – Looming fraction from camera_looming_cue().
max_range (float) – Sensor range that denotes “lidar corridor clear”.
camera_trigger (float) – Cue value at/above which the camera flags a near obstacle.
camera_range (float) – Distance assumed for a camera-only detection.

Return type:

Returns:

Distance (m) to the nearest forward obstacle; max_range when nothing is detected.

POMDPPlanners.environments.carla_pomdp.carla_perception.carla_sensors.lidar_forward_clearance(points, corridor_halfwidth=1.5, z_min=-2.0, z_max=1.0, max_range=50.0)[source]

Distance (m) to the nearest forward in-corridor lidar return, or max_range if clear.

Parameters:

points (Optional[ndarray]) – (N, 4) lidar cloud [x, y, z, intensity] in the sensor frame (x forward, y right, z up), or None/empty when no scan is available.
corridor_halfwidth (float) – Half-width (m) of the forward corridor watched for obstacles.
z_min (float) – Lower height bound (m); returns below it (the ground) are ignored.
z_max (float) – Upper height bound (m); returns above it (overhead structure) are ignored.
max_range (float) – Value returned when the corridor holds no qualifying return.

Return type:

Returns:

The minimum forward x among corridor returns, else max_range.

POMDPPlanners.environments.carla_pomdp.carla_perception.carla_sensors.lidar_vehicle_detections(points, cell_size=0.5, z_min=-2.0, z_max=1.0, detection_range=50.0, min_points=4, max_extent=8.0, confident_points=40)[source]

Cluster a lidar cloud into ego-frame vehicle detections [rel_x, rel_y, confidence].

A single-frame perception stage — the first step of a real perception pipeline that replaces the ground-truth agents oracle. It drops ground / overhead returns, groups the rest into BEV clusters (8-connected occupied cells), and keeps the vehicle-sized ones (rejecting walls / buildings by extent and noise by point count). Confidence rises with the return count. Positions are in the ego frame (rel_x forward, rel_y left); velocity and heading are left to a downstream tracker.

Parameters:

points (Optional[ndarray]) – (N, 4) cloud [x, y, z, intensity] in the sensor frame, or None.
cell_size (float) – BEV grid cell size (m); adjacent occupied cells merge into one cluster.
z_min (float) – Lower height bound (m); returns below it (the ground) are dropped.
z_max (float) – Upper height bound (m); returns above it (overhead structure) are dropped.
detection_range (float) – Clusters beyond this range (m) are dropped.
min_points (int) – Clusters with fewer returns are discarded as noise.
max_extent (float) – Clusters wider than this (m) are discarded as walls / buildings.
confident_points (int) – Return count at/above which confidence saturates to 1.0.

Return type:

Returns:

An (M, 3) array of [rel_x, rel_y, confidence] detections (empty if none).

POMDPPlanners.environments.carla_pomdp.carla_perception.carla_sensors.traffic_light_from_camera(image, focal_pixels=200.0, bulb_diameter=0.3, roi_fraction=0.6, min_bulb_pixels=4, min_distance=3.0, max_distance=50.0)[source]

Infer [should_stop, distance_m] from a red/amber bulb in the upper camera frame.

A lightweight classic-vision stand-in for a learned traffic-light detector, so the light is perceived from the image rather than read from a ground-truth channel. It thresholds the upper region of interest for red and amber bulbs, takes the largest qualifying blob, and estimates its forward distance from the pinhole relation distance = focal_pixels * bulb_diameter / bulb_pixel_size. A green or absent bulb yields [0, 0] (no stop).

Note

Distance from a single low-resolution frame is inherently coarse; focal_pixels is a calibration constant that should match the camera intrinsics, and the result is clamped to [min_distance, max_distance]. The output matches the world’s traffic_light channel layout so it is a drop-in replacement for traffic_light_stop_distance().

Parameters:

image (Optional[ndarray]) – (H, W, 3) RGB frame, or None when no frame is available.
focal_pixels (float) – Camera focal length (px) used by the pinhole distance estimate.
bulb_diameter (float) – Physical traffic-light bulb diameter (m).
roi_fraction (float) – Fraction of the frame height (from the top) searched for a bulb.
min_bulb_pixels (int) – Blobs with fewer pixels than this are rejected as noise.
min_distance (float) – Lower clamp (m) on the estimated stop distance.
max_distance (float) – Upper clamp (m) on the estimated stop distance.

Return type:

Returns:

[1.0, distance_m] when a red/amber bulb is detected, else [0.0, 0.0].

POMDPPlanners.environments.carla_pomdp.carla_perception.carla_sensors.traffic_light_stop_distance(traffic_light, max_range=50.0)[source]

Forward distance (m) to a red/yellow stop line, or max_range if clear.

Turns the world’s traffic_light observation into an obstacle distance so a red light can be injected as a virtual stop-obstacle, exactly like a stopped vehicle.

Parameters:

traffic_light (Optional[ndarray]) – [should_stop, distance_m] from the observation, or None.
max_range (float) – Value returned when there is no active stop signal.

Return type:

Returns:

The stop-line distance when should_stop is set and positive, else max_range.

POMDPPlanners.environments.carla_pomdp.carla_perception.carla_tracking module

Constant-velocity multi-object tracker for lidar vehicle detections.

The temporal (prediction) stage of the perception pipeline: it turns the per-frame detections from lidar_vehicle_detections() into persistent tracks that carry an estimated ego-frame velocity, using an alpha-beta filter (a simplified constant-velocity Kalman) with greedy nearest-neighbour association. Undetected tracks coast on their velocity and decay in confidence until evicted; unmatched detections spawn new tracks. This is what supplies the velocity a downstream planner needs and lets the belief shrink to a thin snapshot of the tracker’s estimate rather than doing the tracking itself.

A track is a row [rel_x, rel_y, vx, vy, confidence] in the ego frame (rel_x forward, rel_y left; vx, vy the agent’s velocity relative to the ego; confidence in [0, 1]). The tracker is a pure function of (prior_tracks, detections, dt) so it composes with an immutable belief that carries the track set forward.

POMDPPlanners.environments.carla_pomdp.carla_perception.carla_tracking.update_tracks(tracks, detections, dt, gate=4.0, alpha=0.5, beta=0.3, new_confidence=0.5, confidence_gain=0.25, confidence_decay=0.25, min_confidence=0.1, track_range=50.0)[source]

Advance the tracker one step and return the new (K, 5) track set.

Parameters:

tracks (Optional[ndarray]) – Prior (N, 5) tracks [rel_x, rel_y, vx, vy, confidence], or None.
detections (Optional[ndarray]) – (M, 3) detections [rel_x, rel_y, confidence] from the detector.
dt (float) – Time step (s) since the previous update.
gate (float) – Max distance (m) at which a detection is associated with a predicted track.
alpha (float) – Position correction gain toward the measurement.
beta (float) – Velocity correction gain from the position residual.
new_confidence (float) – Confidence a newly spawned track starts with.
confidence_gain (float) – Confidence added when a track is matched to a detection.
confidence_decay (float) – Confidence removed when a track is missed (coasted).
min_confidence (float) – Tracks below this confidence are evicted.
track_range (float) – Tracks beyond this ego-frame range (m) are evicted.

Return type:

Returns:

The updated (K, 5) track set (empty (0, 5) when there is nothing to track).

POMDPPlanners.environments.carla_pomdp.carla_perception.observation_model module

Per-channel observation model: one clean observation channel in, one perceived channel out.

The forward-only world emits a raw, ground-truth observation; the planner-side generative model degrades it into the reading a planner actually sees. That degradation is factored by observation channel — each channel (gnss, agents, and, in future, image / lidar) is handled by its own CarlaObservationModel, and the generative model composes a {channel: CarlaObservationModel} map. This module holds the single-channel interface; concrete per-channel models live in the observation_models catalog.

Two capabilities, with different reach:

CarlaObservationModel.perceive() — sample this channel’s perceived value from its clean one. Required; used by a model to generate a tree observation (sample_observation) and to encode the world’s raw channel (encode_observation).
CarlaObservationModel.log_probability() — this channel’s observation density. Optional; a sample-only channel (e.g. a learned encoder) may leave it unimplemented and is still usable to generate observations, but is rejected by a generative model that must score observations for a belief update.

Classes:: CarlaObservationModel: Abstract single-channel clean -> perceived observation interface.

class POMDPPlanners.environments.carla_pomdp.carla_perception.observation_model.CarlaObservationModel[source]

Bases: ABC

Abstract single-channel observation model: clean channel -> perceived channel.

A concrete perception maps one observation channel’s clean, fully-detected value (built from a state by a model, or taken from the world’s raw reading) to the degraded value a planner sees. Implementations declare which channel they handle via channel and set supports_density to True when they also provide log_probability().

channel: The observation-dict key this model handles (e.g. "gnss" or "agents").

supports_density: Whether log_probability() is implemented. Sample-only channels leave this False and are usable only where sampling is needed.

Note

This is an abstract base class and cannot be instantiated directly.

channel: str = ''

log_probability(clean_channel, channel_observation)[source]

Log-density of channel_observation given the clean channel value.

Parameters:

clean_channel (Any) – The noise-free value of this channel built from a state.
channel_observation (Any) – The channel value whose likelihood is scored.

Return type: