CARLA数据标注功能实战指南：从零开始构建自动驾驶数据集

1次阅读

没有评论

共计 2201 个字符，预计需要花费 6 分钟才能阅读完成。

自动驾驶模型训练离不开高质量标注数据，但真实场景数据采集和标注存在明显瓶颈：

采集成本高：需配备专业传感器车队，覆盖多样天气 / 路况，单小时采集成本可达数千美元
标注效率低：人工标注一帧激光雷达点云平均耗时 30 分钟，100 小时视频需 2000+ 人工小时
长尾问题难覆盖：极端场景（如交通事故）数据难以通过常规采集获取

启动 CARLA 服务后，通过 WorldObserver 模块可实时获取以下标注类型：

语义分割
调用 carla.Camera.listen() 时设置 semantic_segmentation 标签
每个像素值对应 carla.CityObjectLabel 枚举（如 10= 车辆，22= 行人）

2D 边界框

for actor in world.get_actors():
    if actor.type_id.startswith('vehicle'):
        bbox = actor.bounding_box
        vertices = [carla.Location(x,y,z) for x,y,z in bbox.get_world_vertices(actor.get_transform())]

深度图
通过 camera.listen(lambda image: image.convert(carla.ColorConverter.Depth)) 获取
需使用 image.float_to_depth() 转换为米制单位

典型数据采集脚本结构：

初始化客户端并加载地图
设置传感器（相机 /LiDAR）及标注参数
定义异步数据回调存储函数
控制车辆运动路径

关键代码结构：

# 传感器配置示例
blueprint = world.get_blueprint_library().find('sensor.camera.rgb')
blueprint.set_attribute('image_size_x', '1920')
blueprint.set_attribute('image_size_y', '1080')

# 数据回调示例
def sensor_callback(data, queue):
    array = np.frombuffer(data.raw_data, dtype=np.uint8)
    array = np.reshape(array, (data.height, data.width, 4))
    queue.put((data.frame, array[:,:,:3]))

# 主循环
while True:
    world.tick()
    vehicle.apply_control(carla.VehicleControl(throttle=0.3))

转换关键步骤：

构建 categories 列表对应 CARLA 标签体系
将边界框顶点转换为 [x,y,width,height] 格式
计算 area 字段（bbox 面积）

示例片段：

coco_annotation = {
    "id": frame_id,
    "image_id": image_id,
    "category_id": city_label_to_coco[actor.semantic_tags[0]],
    "bbox": [x_min, y_min, width, height],
    "area": width * height,
    "iscrowd": 0
}

LiDAR 数据转换要点：

将 CARLA 点云（List[carla.Location]）转换为 Nx4 矩阵（x,y,z,intensity）
写入二进制文件时使用 tofile() 保持 little-endian 格式
标注文件需包含 truncated/occluded 等 KITTI 特有字段

标注不一致：
解决方案：固定随机种子(world = client.get_world(settings=carla.WorldSettings(random_seed=42)))
检查 actor.semantic_tags 是否为空（部分静态物体需手动标记）
内存泄漏：
每个 sensor.destroy() 后执行gc.collect()
使用 weakref 管理传感器引用

批量处理：
使用 world.tick() 替代 time.sleep() 保持同步
单次 tick 内获取多传感器数据

分布式标注：

with concurrent.futures.ThreadPoolExecutor() as executor:
    futures = [executor.submit(process_sensor, sensor) for sensor in sensors]
    for future in concurrent.futures.as_completed(futures):
        save_data(*future.result())

仿真数据虽能快速生成，但需注意：