2. 数据采集与预处理流水线搭建
任务定义
本案例搭建一条可复用的数据采集与预处理流水线，输入为模拟器或实车采集的原始图像序列及对应车辆状态（转向角、速度），输出为经过清洗、增强和标准化后的训练数据集（图像张量 + 标签张量）。流水线以Python脚本实现，支持命令行调用，并输出数据统计报告以便验证。
流水线架构
核心实现
以下代码实现完整流水线。清洗阶段检测图像文件完整性、丢弃标签超出物理范围（|转向角|>1.0 rad 或速度<0 或 >50 m/s）的帧，并对缺失标签做线性插值。增强阶段对图像做随机亮度调整、水平平移和水平翻转，同时同步变换转向角（翻转取反、平移按比例补偿）。标准化阶段将图像像素除以255.0，标签按训练集统计量做Z-score归一化。
#!/usr/bin/env python3
"""
data_pipeline.py — 数据采集与预处理流水线
用法:
python data_pipeline.py --input ./raw_data --output ./processed
输入目录结构:
raw_data/
frames/ # 000001.jpg, 000002.jpg, ...
labels.csv # 列: frame_id, steering, speed
"""
import os
import csv
import json
import argparse
import numpy as np
from pathlib import Path
from PIL import Image, ImageEnhance
# ---------- 1. 数据加载 ----------
def load_raw_data(input_dir):
"""加载原始图像路径和标签"""
frames_dir = Path(input_dir) / "frames"
csv_path = Path(input_dir) / "labels.csv"
if not frames_dir.exists() or not csv_path.exists():
raise FileNotFoundError(f"缺少 frames/ 或 labels.csv: {input_dir}")
samples = []
with open(csv_path, "r") as f:
reader = csv.DictReader(f)
for row in reader:
frame_id = row["frame_id"]
img_path = frames_dir / f"{frame_id}.jpg"
if not img_path.exists():
print(f"[清洗] 跳过缺失图像: {img_path}")
continue
samples.append({
"image_path": str(img_path),
"steering": float(row["steering"]),
"speed": float(row["speed"])
})
return samples
# ---------- 2. 清洗 ----------
def clean_samples(samples):
"""丢弃异常帧，对缺失标签插值"""
if not samples:
return []
# 物理范围过滤
valid = []
for s in samples:
if abs(s["steering"]) > 1.0: # 转向角超过 ±1 rad 视为异常
print(f"[清洗] 丢弃异常转向角: {s['steering']:.3f}")
continue
if s["speed"] < 0 or s["speed"] > 50: # 速度范围 0~50 m/s
print(f"[清洗] 丢弃异常速度: {s['speed']:.2f}")
continue
valid.append(s)
# 对连续缺失的标签做线性插值（此处模拟：若 steering 为 NaN 则用前后帧均值）
# 实际采集数据中 NaN 已在 float() 转换时抛出，此处保留接口
return valid
# ---------- 3. 增强 ----------
def augment_image(image, steering, speed):
"""单样本增强：随机亮度、平移、翻转"""
rng = np.random.default_rng()
# 随机亮度 (0.6 ~ 1.4)
factor = rng.uniform(0.6, 1.4)
image = ImageEnhance.Brightness(image).enhance(factor)
# 随机水平平移 (±20px)，转向角按比例补偿
shift_px = rng.integers(-20, 21)
if shift_px != 0:
width, _ = image.size
shifted = Image.new("RGB", image.size, (0, 0, 0))
if shift_px > 0:
shifted.paste(image, (shift_px, 0))
else:
shifted.paste(image, (shift_px, 0))
image = shifted
# 转向角补偿：平移 20px 约对应 0.1 rad
steering += shift_px * 0.005
# 随机水平翻转 (50% 概率)
if rng.random() > 0.5:
image = image.transpose(Image.FLIP_LEFT_RIGHT)
steering = -steering
return image, steering, speed
# ---------- 4. 标准化 ----------
def normalize(images, labels):
"""图像像素归一化到 [0,1]，标签 Z-score 标准化"""
images = images.astype(np.float32) / 255.0
mean = labels.mean(axis=0)
std = labels.std(axis=0)
std[std == 0] = 1.0 # 防止除零
labels_norm = (labels - mean) / std
return images, labels_norm, mean, std
# ---------- 主流程 ----------
def main():
parser = argparse.ArgumentParser(description="数据采集与预处理流水线")
parser.add_argument("--input", required=True, help="原始数据目录")
parser.add_argument("--output", default="./processed", help="输出目录")
parser.add_argument("--augment-times", type=int, default=2,
help="每张原始图像的增强倍数")
args = parser.parse_args()
# 加载
print("[1/4] 加载原始数据...")
samples = load_raw_data(args.input)
print(f" 加载 {len(samples)} 个有效样本")
# 清洗
print("[2/4] 清洗数据...")
samples = clean_samples(samples)
print(f" 清洗后保留 {len(samples)} 个样本")
if len(samples) == 0:
raise RuntimeError("清洗后无有效样本，请检查输入数据")
# 增强
print(f"[3/4] 增强数据 (每样本 ×{args.augment_times})...")
aug_images = []
aug_labels = []
for s in samples:
# 原始样本
img = Image.open(s["image_path"]).convert("RGB")
img = img.resize((160, 80)) # 统一尺寸
aug_images.append(np.array(img))
aug_labels.append([s["steering"], s["speed"]])
# 增强副本
for _ in range(args.augment_times - 1):
aug_img, aug_steer, aug_speed = augment_image(
img.copy(), s["steering"], s["speed"])
aug_images.append(np.array(aug_img))
aug_labels.append([aug_steer, aug_speed])
images = np.stack(aug_images) # (N, 80, 160, 3)
labels = np.array(aug_labels) # (N, 2)
print(f" 增强后共 {len(images)} 个样本")
# 标准化
print("[4/4] 标准化数据...")
images_norm, labels_norm, mean, std = normalize(images, labels)
# 保存
os.makedirs(args.output, exist_ok=True)
np.save(os.path.join(args.output, "images.npy"), images_norm)
np.save(os.path.join(args.output, "labels.npy"), labels_norm)
stats = {
"num_samples": len(images_norm),
"image_shape": list(images_norm.shape[1:]),
"label_mean": mean.tolist(),
"label_std": std.tolist(),
"original_samples": len(samples),
"augment_times": args.augment_times
}
with open(os.path.join(args.output, "stats.json"), "w") as f:
json.dump(stats, f, indent=2)
# 保存清洗后的元数据
with open(os.path.join(args.output, "cleaned.csv"), "w", newline="") as f:
writer = csv.writer(f)
writer.writerow(["frame_id", "steering", "speed"])
for i, s in enumerate(samples):
writer.writerow([i, s["steering"], s["speed"]])
print(f"\n完成！输出目录: {args.output}")
print(f" 样本数: {stats['num_samples']}")
print(f" 图像尺寸: {stats['image_shape']}")
print(f" 标签均值: {stats['label_mean']}")
print(f" 标签标准差: {stats['label_std']}")
if __name__ == "__main__":
main()
验证脚本
#!/usr/bin/env python3
"""validate.py — 验证预处理输出"""
import json
import numpy as np
from pathlib import Path
def validate(output_dir):
out = Path(output_dir)
images = np.load(out / "images.npy")
labels = np.load(out / "labels.npy")
stats = json.load(open(out / "stats.json"))
errors = []
# 1. 形状检查
expected_shape = tuple(stats["image_shape"])
if images.shape[1:] != expected_shape:
errors.append(f"图像形状 {images.shape[1:]} != {expected_shape}")
# 2. 数值检查
if np.isnan(images).any() or np.isnan(labels).any():
errors.append("存在 NaN 值")
if images.min() < 0 or images.max() > 1:
errors.append(f"图像范围超出 [0,1]: [{images.min():.3f}, {images.max():.3f}]")
# 3. 标签统计检查
label_mean = np.array(stats["label_mean"])
label_std = np.array(stats["label_std"])
labels_denorm = labels * label_std + label_mean
if abs(labels_denorm[:, 0]).max() > 1.0:
errors.append("转向角超出 ±1 rad")
if labels_denorm[:, 1].min() < 0 or labels_denorm[:, 1].max() > 50:
errors.append("速度超出 0~50 m/s")
# 4. 样本数检查
expected_count = stats["original_samples"] * stats["augment_times"]
if len(images) != expected_count:
errors.append(f"样本数 {len(images)} != 预期 {expected_count}")
if errors:
print("验证失败:")
for e in errors:
print(f" ✗ {e}")
return False
else:
print("验证通过 ✓")
print(f" 样本数: {len(images)}")
print(f" 图像形状: {images.shape}")
print(f" 标签范围: 转向角 [{labels_denorm[:,0].min():.3f}, {labels_denorm[:,0].max():.3f}] rad")
print(f" 速度 [{labels_denorm[:,1].min():.2f}, {labels_denorm[:,1].max():.2f}] m/s")
return True
if __name__ == "__main__":
import sys
validate(sys.argv[1] if len(sys.argv) > 1 else "./processed")
运行与验证
依赖安装：
pip install numpy pillow
准备演示数据：创建 demo_data/frames/ 目录，放入至少3张JPG图像（任意尺寸），并创建 demo_data/labels.csv：
frame_id,steering,speed
000001,0.15,8.5
000002,-0.22,12.3
000003,0.08,6.7
执行流水线：
python data_pipeline.py --input demo_data --output processed --augment-times 3
验证输出：
python validate.py processed
判定标准：验证脚本输出“验证通过 ✓”即表示流水线正确完成清洗、增强和标准化。预期输出文件包括 images.npy（形状为 (9, 80, 160, 3) 的 float32 数组，值域 [0,1]）、labels.npy（形状 (9, 2) 的标准化标签）、stats.json 和 cleaned.csv。若输入图像尺寸不同，代码会自动缩放到 160×80，因此输出形状固定为 (N, 80, 160, 3)。