nuScenes数据实战：用Python脚本一键提取Lidar点云和未标注的Sweeps帧（附完整代码）

nuScenes数据高效处理指南：Python脚本实现点云与Sweeps帧自动化提取

自动驾驶算法开发中，数据准备往往占据70%以上的时间成本。本文将分享一套完整的Python解决方案，帮助开发者快速从nuScenes数据集中提取LIDAR点云和未标注的sweeps帧数据，大幅提升算法原型开发效率。

1. 环境配置与数据准备

1.1 安装必要的工具链

处理nuScenes数据集需要以下核心组件：

pip install nuscenes-devkit matplotlib numpy open3d

推荐使用conda创建独立环境以避免依赖冲突：

conda create -n nuscenes python=3.8 conda activate nuscenes

1.2 数据集目录结构规范

正确的数据集存放结构对后续处理至关重要：

nuscenes/ ├── maps/ ├── samples/ ├── sweeps/ └── v1.0-trainval/ ├── attribute.json ├── calibrated_sensor.json ├── category.json └── ...（其他元数据文件）

提示：建议使用符号链接将数据目录统一到固定路径，便于多项目共享数据集

2. 核心数据提取流程

2.1 LIDAR点云数据批量提取

以下脚本实现了从指定场景中提取所有关键帧的点云数据：

from nuscenes.nuscenes import NuScenes from nuscenes.utils.data_classes import LidarPointCloud import os import numpy as np def export_lidar_frames(nusc, output_dir, sensor='LIDAR_TOP'): os.makedirs(output_dir, exist_ok=True) for scene in nusc.scene: sample_token = scene['first_sample_token'] while sample_token: sample = nusc.get('sample', sample_token) lidar_data = nusc.get('sample_data', sample['data'][sensor]) # 加载并保存点云 pc = LidarPointCloud.from_file( os.path.join(nusc.dataroot, lidar_data['filename']) ) points = np.transpose(pc.points) output_path = os.path.join( output_dir, f"{lidar_data['token']}.npy" ) np.save(output_path, points) sample_token = sample['next']

2.2 未标注Sweeps帧智能采集

利用令牌链式访问机制获取连续时序数据：

def export_sweeps_sequence(nusc, output_dir, sensor='LIDAR_TOP', num_sweeps=5): os.makedirs(output_dir, exist_ok=True) for scene in nusc.scene: sample_token = scene['first_sample_token'] while sample_token: sample = nusc.get('sample', sample_token) lidar_data = nusc.get('sample_data', sample['data'][sensor]) # 回溯获取历史sweeps current_token = lidar_data['prev'] for i in range(num_sweeps): if not current_token: break sweep_data = nusc.get('sample_data', current_token) if not sweep_data['is_key_frame']: pc = LidarPointCloud.from_file( os.path.join(nusc.dataroot, sweep_data['filename']) ) np.save( os.path.join(output_dir, f"{sweep_data['token']}.npy"), np.transpose(pc.points) ) current_token = sweep_data['prev'] sample_token = sample['next']

3. 高级功能实现

3.1 数据可视化校验模块

为确保数据提取正确性，集成Open3D可视化：

import open3d as o3d def visualize_pointcloud(nusc, sample_token, sensor='LIDAR_TOP'): sample = nusc.get('sample', sample_token) lidar_data = nusc.get('sample_data', sample['data'][sensor]) pc = LidarPointCloud.from_file( os.path.join(nusc.dataroot, lidar_data['filename']) ) points = np.transpose(pc.points[:, :3]) # 取XYZ坐标 pcd = o3d.geometry.PointCloud() pcd.points = o3d.utility.Vector3dVector(points) # 可视化配置 vis = o3d.visualization.Visualizer() vis.create_window() vis.add_geometry(pcd) # 设置视角 ctr = vis.get_view_control() ctr.set_front([0, 0, -1]) ctr.set_up([0, -1, 0]) vis.run() vis.destroy_window()

3.2 元数据关联导出

为后续半监督学习准备完整的元信息：

def export_metadata(nusc, output_file): import json metadata = { 'scenes': [], 'samples': [], 'sweeps': [] } for scene in nusc.scene: scene_info = { 'token': scene['token'], 'name': scene['name'], 'sample_count': scene['nbr_samples'] } metadata['scenes'].append(scene_info) sample_token = scene['first_sample_token'] while sample_token: sample = nusc.get('sample', sample_token) sample_info = { 'token': sample_token, 'timestamp': sample['timestamp'], 'scene': scene['token'], 'next': sample['next'], 'prev': sample['prev'] } metadata['samples'].append(sample_info) sample_token = sample['next'] with open(output_file, 'w') as f: json.dump(metadata, f, indent=2)

4. 工程化实践建议

4.1 性能优化技巧

处理大规模数据集时的关键优化点：

4.2 典型应用场景

这套工具链在以下场景中表现优异：

• 时序目标检测：连续sweeps帧提供运动线索
• 点云配准：高频率采集数据提升ICP精度
• 半监督学习：利用大量未标注sweeps数据
• 传感器标定：多模态数据时间对齐验证

# 示例：构建时序数据加载器 class TemporalLoader: def __init__(self, nusc, scene_token, num_sweeps=5): self.nusc = nusc self.scene = nusc.get('scene', scene_token) self.num_sweeps = num_sweeps def __iter__(self): sample_token = self.scene['first_sample_token'] while sample_token: sample = self.nusc.get('sample', sample_token) lidar_data = self.nusc.get('sample_data', sample['data']['LIDAR_TOP']) sequence = self._get_sweep_sequence(lidar_data) yield sequence sample_token = sample['next'] def _get_sweep_sequence(self, lidar_data): sequence = [] current_token = lidar_data['token'] for _ in range(self.num_sweeps): if not current_token: break data = self.nusc.get('sample_data', current_token) pc = LidarPointCloud.from_file( os.path.join(self.nusc.dataroot, data['filename']) ) sequence.append(np.transpose(pc.points)) current_token = data['prev'] return sequence[::-1] # 按时间顺序返回

实际项目中，建议将提取的数据转换为更高效的存储格���如HDF5，特别是当需要处理完整trainval集时。以下是将多个.npy文件合并为HDF5的实用代码片段：

import h5py def convert_to_hdf5(npy_dir, hdf5_path): files = [f for f in os.listdir(npy_dir) if f.endswith('.npy')] with h5py.File(hdf5_path, 'w') as hf: for i, f in enumerate(files): data = np.load(os.path.join(npy_dir, f)) hf.create_dataset( f'frame_{i}', data=data, compression='gzip' )