前言

开源，是为了让自己的智能车生涯不留遗憾！
该文章旨在提供除遥控数据训练循迹模型外的另一种方式，算法循迹获取数据集，相关代码请前往仓库https://github.com/SugarSong404/baidu_car
循迹算法有关的文件都在lane_idea目录下，不过没有整理过，还请自行理清文件用途，或者查看下方文章了解大概的思路

如何制作基础数据集

我的想法是，一个好的循迹模型，表现出来的效果就如同一个扛强光等干扰因素能力很强的循迹算法
正如同蓝白赛道的竞速车循迹算法一样，传入的是图像，输出的是偏差
偏差作为输入传入转向环，得到的yaw角速度再传入速度环，就可以实现循迹转向效果
所以，我先实现了一个在普通光照下可以完赛的循迹算法，并让车在跑赛道时记录图像和当前偏差，大致流程如下图所示

数据集目录：

• images文件夹包含了640x480的RGB赛道图像
• label.csv文件记录图像对应的偏差，格式为
  filename, deviation
  image/0001.jpg -14.440048217773436
  image/0002.jpg -16.307037353515625
  image/0003.jpg -19.632350921630856
  …

算法循迹

主循迹算法

这里直接贴出我的循迹算法，当然也可以去我的仓库里找到lane_idea目录，里面有更详细的思路与我用过的工具，不过我懒得整理了

为什么需要一个search_img的库呢，因为最开始我用py写的这个函数，运行一帧居然需要逆天的400ms，后面换成c后只需2ms，所以我直接在py中调用c的函数即可提升效率了

import cv2
import os
import json
import time
import numpy as np
from collections import deque

import ctypes
lib = ctypes.CDLL(os.path.dirname(__file__)+'/my_models/lane/search_img.so')
lib.search_img.argtypes = [
    np.ctypeslib.ndpointer(dtype=np.uint8, ndim=3, flags='C_CONTIGUOUS'), 
    np.ctypeslib.ndpointer(dtype=np.uint8, ndim=2, flags='C_CONTIGUOUS'), 
    ctypes.c_int,
    ctypes.c_int 
]
lib.search_img.restype = None

resize_size = 120
view_limit = 400
YMAX = 116
YMIN = 3
XMAX = 117
XMIN = 2

class Lane:
    def __init__(self, model_path = os.path.dirname(__file__)+"/my_models/lane"):
        standard_path = model_path + "/standard_binary.jpg"
        perspective_path = model_path + "/perspective.json"
        self.last_dev = 0
        self.ann_status = 0
        self.ann_devs = []
        self.draw_f = False
        self.standard_lr = get_standard(standard_path)
        self.perspective = get_perspective(perspective_path, self.standard_lr)

    def draw_basemap(self, img):
        return get_basemap(img)

    def cal_dev(self, img, kk, bb, ann_f=0):
        self.flip_img = get_basemap(img)[:, :, 0]
        left_scan, right_scan = get_scan_line(self.flip_img)
        left_line, right_line = get_lr_line(self.flip_img, left_scan, right_scan)

        if self.ann_status==0 and ann_f==1:
            self.ann_status = 1
        elif self.ann_status==1 and self.draw_f:
            self.ann_status = 2
        elif self.ann_status==2 and ann_f==2:
            self.ann_status = 3
            self.timer = time.time()
        elif self.ann_status == 3 and (time.time() - self.timer) >= 4:
            self.ann_status = 4

        if self.ann_status==1 or self.ann_status==2:
            right_line = [-1]*119
            temp = deal_annulus(self.flip_img)
            if temp is not None:
                left_line = temp
                self.draw_f = True
        elif self.ann_status==3:
            return sum(self.ann_devs)/len(self.ann_devs)

        struct_lr = [process_line(left_line), process_line(right_line)]
        struct_lr = delete_line(self.flip_img, [left_scan, right_scan], struct_lr)
        dev = calculate_dev(self.standard_lr, struct_lr, self.perspective, [kk, bb])
        if dev is None:dev = self.last_dev

        if self.ann_status==2:
            self.ann_devs.append(dev)

        self.last_dev = dev
        return dev

    def get_order(self):
        if self.ann_status == 0:return 1
        elif self.ann_status <=3:return 2
        elif self.ann_status == 4: return 3 

def delete_line(image, scan_lr, struct_lr):
    if (struct_lr[0][2]!=0 and struct_lr[1][2]!=0) and (abs(scan_lr[1]-scan_lr[0]) < 30) and (image[0][scan_lr[0]] > 128 and image[0][scan_lr[1]] > 128):
        arr_l, start_l, len_l = struct_lr[0]
        arr_r, start_r, len_r = struct_lr[1]
        x_l = arr_l[start_l] - arr_l[start_l+len_l-1]
        k_l = abs(len_l / x_l) if x_l!=0 else float('inf')
        x_r = arr_r[start_r] - arr_r[start_r+len_r-1]
        k_r = abs(len_r / x_r) if x_r!=0 else float('inf')
        if k_l-k_r>0:struct_lr[1] = arr_r, 0, 0
        else: struct_lr[0] = arr_r, 0, 0
    return struct_lr

def get_basemap(image):
    def preprocess(image):
        flip_img = cv2.flip(image, 0)

        gray = cv2.cvtColor(flip_img, cv2.COLOR_BGR2GRAY)
        _, binary = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY | cv2.THRESH_OTSU)
        binary_bgr = cv2.cvtColor(binary, cv2.COLOR_GRAY2BGR)
        cropped_img = binary_bgr[:view_limit, :, :]
        binary_resize = cv2.resize(cropped_img, (resize_size, resize_size), interpolation=cv2.INTER_AREA)
        return binary_resize

    def search_img(image, start_x, start_y):
        basemap = np.ones((119, 119), dtype=np.uint8)
        lib.search_img(image, basemap, start_x, start_y)
        return basemap
        
    def get_start_point(image):
        max_len = 0  
        start_point = -1  
        current_len = 0  
        current_start = -1  

        for X in range(0, 119):
            if image[0][X][0] < 128:
                if current_len == 0:
                    current_start = X
                current_len += 1
            else:
                if current_len > max_len:
                    max_len = current_len
                    start_point = current_start
                current_len = 0

        if current_len > max_len:
            max_len = current_len
            start_point = current_start

        return start_point if max_len > 0 else -1

    image = preprocess(image)
    point = get_start_point(image)
    basemap = search_img(image, point, 0)
    gray_image = np.where(basemap == 0, 0, 255).astype(np.uint8)
    gray_image = cv2.cvtColor(gray_image, cv2.COLOR_GRAY2BGR)
    final_img = cv2.flip(gray_image, 0)
    return final_img

def fix_line_array(arr):
    arr = np.array(arr, dtype=float)
    y_coords = np.arange(len(arr))

    valid_mask = arr != -1
    x_valid = arr[valid_mask]
    y_valid = y_coords[valid_mask]

    k, b = np.polyfit(y_valid, x_valid, deg=1)

    for i in range(len(arr)):
        if arr[i] == -1:
            arr[i] = int(k * i + b)
    return arr


def get_lr_line(image, scan_left, scan_right):
    image = np.asarray(image, dtype=np.uint8)
    H = 119
    W = 119

    left = np.full(H, -1, dtype=np.int32)
    right = np.full(H, -1, dtype=np.int32)

    if scan_left >= 2:
        x_left_range = np.arange(scan_left, 1, -1)
        for i, x in enumerate(x_left_range):
            cond = (image[:, x] < 128) & (image[:, x-1] > 128) & (image[:, x-2] > 128)
            update_mask = (left == -1) & cond
            left[update_mask] = x
    if scan_right <= 116:
        x_right_range = np.arange(scan_right, 117)
        for i, x in enumerate(x_right_range):
            cond = (image[:, x] < 128) & (image[:, x+1] > 128) & (image[:, x+2] > 128)
            update_mask = (right == -1) & cond
            right[update_mask] = x

    return left.tolist(), right.tolist()


def get_standard(path):
    standard = cv2.imread(path)[:, :, 0]
    standard_left, standard_right = get_lr_line(standard, 60 ,60)
    return [fix_line_array(standard_left), fix_line_array(standard_right)]

def get_scan_line(image):
    image = np.asarray(image, dtype=np.uint8)
    h, w = image.shape
    col_y      = image[2:, :]
    col_y_m1   = image[1:-1, :]
    col_y_m2   = image[0:-2, :]
    condition  = (col_y < 128) & (col_y_m1 > 128) & (col_y_m2 > 128)

    first_edge = np.argmax(condition, axis=0) + 2
    has_edge = condition.any(axis=0)
    edge_y = np.where(has_edge, first_edge, h)

    edge_y = np.where((edge_y == h) & (image[0, :] < 128), 2, edge_y)
    min_edge_y = edge_y.min()
    candidates = np.where(edge_y == min_edge_y)[0]
    if len(candidates) > 0:
        scan_left = int(candidates[0])
        scan_right = int(candidates[-1])
    else:
        scan_left, scan_right = 118, 0

    return scan_left, scan_right

def process_line(line, max_gap=10):
    max_len = 0
    max_start = 0
    current_start = None
    current_len = 0
    prev_val = None

    for i, val in enumerate(line):
        if val == -1:
            if current_start is not None:
                if current_len > max_len:
                    max_len = current_len
                    max_start = current_start
                current_start = None
                current_len = 0
                prev_val = None
            continue

        if current_start is None:
            current_start = i
            current_len = 1
            prev_val = val
        else:
            if abs(val - prev_val) > max_gap:
                if current_len > max_len:
                    max_len = current_len
                    max_start = current_start
                current_start = i
                current_len = 1
            else:
                current_len += 1
            prev_val = val
    if current_start is not None and current_len > max_len:
        max_len = current_len
        max_start = current_start
    filled = line.copy()
    prev_val = None
    for i in range(max_start, max_start + max_len):
        if filled[i] != -1:
            prev_val = filled[i]
        elif prev_val is not None:
            filled[i] = prev_val
            
    
    if max_len < 10:return filled, 0, 0
    return filled, max_start, max_len


def deal_noneline(image):
    high_line = [-1] * 119
    num=0
    for x in range(0, 119):
        for y in range(117, 0, -1):
            if image[y+1][x] < 128 and image[y][x] < 128 and image[y-1][x]>128:
                num+=1
                high_line[x] = y
    if num<10:return None
    high_line = list(filter(lambda x: x != -1, high_line))
    x = np.arange(len(high_line))
    slope, _ = np.polyfit(x, high_line, 1) 
    if slope > 0:return -40
    elif slope < -0:return 40
    else:return None

def calculate_dev(standard_lr, struct_lr, pers, kb):
    left_struct = struct_lr[0]
    right_struct = struct_lr[1]
    standard_left = standard_lr[0]
    standard_right = standard_lr[1]
    dev_l = None
    dev_r = None

    if right_struct[2] > 10 :dev_r = get_kb(standard_right, right_struct, pers, kb)
    if left_struct[2] > 10 :dev_l = get_kb(standard_left, left_struct, pers, kb)

    if dev_l is not None and dev_r is not None:return (dev_r + dev_l)/2
    if dev_l is not None: return dev_l
    if dev_r is not None: return dev_r
    return None
        
def get_kb(standard, struct, pers, kb):
    sumx = 0
    sumy = 0
    sumxy = 0
    sumx2 = 0
    n = 0 
    x = 0
    y = 0
    for j in range(struct[1], struct[1] + struct[2]):
        n += 1
        y = (struct[0][j] - standard[j])*pers[1][j]
        x = pers[0][j]
        sumx += x
        sumy += y
        sumxy += x * y
        sumx2 += x * x
    k = (n * sumxy - sumx * sumy) / (n * sumx2 - sumx * sumx)
    b = sumy / n - k * sumx / n
    return k * kb[0] + b * kb[1]

def get_perspective(path, standard_lr):
    with open(path, "r") as f:
        data = json.load(f)
    arr = data["arr"]
    wid = data["wid"]
    arr2 = [0] * 119
    for y in range(0, 119):
        arr2[y] = wid / float(standard_lr[1][y] - standard_lr[0][y] + 1)
    return [arr, arr2]

def deal_annulus(image):
    def has_black_white_black(seq, start, end):
        first_white_index = -1
        for i in range(start, end, -1):
            prev, current = seq[i + 1], seq[i]
            if prev < 128 and current >= 128 and first_white_index == -1:
                first_white_index = i 
            elif first_white_index != -1 and prev >= 128 and current < 128:
                return first_white_index
        return -1 

    def change_line(image, corner):
        left_line = [-1]*119
        k = (corner[0]-108)/(corner[1]-0)
        b = corner[0] - k * corner[1] 
        if k==0: return None
        for y in range(108, corner[0]-1, -1):
            left_line[y] = int((y - b)/k)
        for y in range(corner[0]+1, -1, -1):
            for x in range(117, corner[1], -1):
                if image[y][x+1] <128 and image[y][x] < 128 and image[y][x-1] >128:
                    left_line[y] = x
                    break
        return left_line

    num = 0
    start_y = 0
    for y in range(114,5,-1):
        if abs(int(image[y][118])-int(image[y-1][118])) > 128:
            num+=1
            start_y = y
    if image[114][118] > 128 or num!=1:return None
    num = 0
    corner_y = -1
    corner_x = -1
    for y in range(start_y, 118):
        x = has_black_white_black(image[y], 117, -1)
        if x!=-1:
            num += 1
            corner_x = x
            corner_y = y
        else:break
    if num>=3 and corner_x!=-1:return change_line(image, [corner_y, corner_x])
    else:return None

其它文件

这里是search_img.c，还请自行编译为so文件方便py调用

#include <stdint.h>
#include <stdlib.h>

#define YY 118
#define XX 118
#define SIZE 119
void search_img(uint8_t image[SIZE+1][SIZE+1][3], uint8_t basemap[SIZE][SIZE],
                int start_x, int start_y)
{
    int max_size = SIZE * SIZE;
    int *q_x = (int*)malloc(max_size * sizeof(int));
    int *q_y = (int*)malloc(max_size * sizeof(int));
    int head = 0, tail = 0;

    uint8_t index_bm[YY+1][XX+1];

    for(int y = 0; y < YY+1; y++) {
        for(int x = 0; x < XX+1; x++) {
            basemap[y][x] = 1;
            index_bm[y][x] = 1;
        }
    }

    if (start_x < 0 || start_x > XX || start_y < 0 || start_y > YY) {
        free(q_x); free(q_y);
        return;
    }

    q_x[tail] = start_x;
    q_y[tail] = start_y;
    tail++;
    index_bm[start_y][start_x] = 0;

    int directions[4][2] = {{0,-1},{0,1},{-1,0},{1,0}};

    while(head < tail){
        int a_x = q_x[head];
        int a_y = q_y[head];
        head++;

        uint8_t val = image[a_y][a_x][0];

        if(val < 128){
            if(basemap[a_y][a_x] == 1)
                basemap[a_y][a_x] = 0;

            for(int d = 0; d < 4; d++){
                int nx = a_x + directions[d][0];
                int ny = a_y + directions[d][1];
                if(nx >= 0 && nx <= XX && ny >= 0 && ny <= YY && index_bm[ny][nx] == 1){
                    index_bm[ny][nx] = 0;
                    q_x[tail] = nx;
                    q_y[tail] = ny;
                    tail++;
                }
            }
        } else {
            basemap[a_y][a_x] = 2;
        }
    }
    free(q_x);
    free(q_y);
}

一个标准直道图像standard_binary.jpg也是需要的，得到的边线用作标准左右线，如下图

还有需要的就是一个逆透视数组perspective.json，格式如下，如果不知道什么是逆透视以及如何测量可以去问蓝白赛道竞速组

wid是赛道宽度，arr是图像第n层对应的真实距离

data = {
  "wid": 40,
  "arr": [
    82.9, 80.6, 78.0, 75.6, 73.8, 71.6, 70.2, 68.2, 66.5, 64.3, 63.4, 61.7,
    60.1, 58.7, 57.5, 56.2, 55.0, 53.7, 52.7, 51.5, 50.3, 49.2, 48.3, 46.6,
    46.1, 45.1, 44.2, 43.1, 42.6, 42.0, 41.2, 40.4, 39.6, 38.6, 37.1, 36.2,
    35.3, 35.0, 34.4, 33.6, 32.3, 31.5, 31.0, 30.6, 30.2, 29.9, 29.1, 28.7,
    28.2, 27.7, 27.4, 26.9, 26.5, 26.2, 25.9, 25.2, 25.0, 24.7, 24.2, 23.9,
    23.6, 23.3, 22.9, 22.7, 22.2, 21.5, 21.5, 21.0, 20.5, 20.4, 20.2, 20.0,
    19.8, 19.5, 19.4, 19.1, 18.9, 18.6, 18.5, 18.3, 18.0, 17.6, 17.4, 17.3,
    17.0, 16.8, 16.4, 16.2, 16.0, 15.9, 15.7, 15.5, 15.3, 14.5, 14.4, 14.2,
    14.0, 13.9, 13.7, 13.5, 13.2, 13.1, 13.0, 12.9, 12.7, 12.6, 12.5, 12.3,
    12.0, 12.0, 11.7, 11.5, 11.3, 11.2, 11.1, 11.0, 10.9, 10.9, 10.8
  ]
}

训练出基础模型

初次训练的数据集大概跑个20次循迹，10000张左右，录制时建议的循迹速度是0.3，这样数据集的质量不错而且数量也够

在训练前可以预载一下数据集，减下训练时的加载与预处理压力

import os
import numpy as np
from PIL import Image
import paddle.vision.transforms as T

IMG_DIR = './dataset/images'
OUT_DIR = './preprocessed_npy'
os.makedirs(OUT_DIR, exist_ok=True)

transform = T.Compose([
    T.Resize((384, 512)),
    T.ToTensor(),
    T.Normalize(mean=[0.485, 0.456, 0.406],
                std=[0.229, 0.224, 0.225])
])

for fname in os.listdir(IMG_DIR):
    if not fname.endswith('.jpg'): continue
    img_path = os.path.join(IMG_DIR, fname)
    img = Image.open(img_path).convert('RGB')
    img_tensor = transform(img).numpy()
    np.save(os.path.join(OUT_DIR, fname.replace('.jpg', '.npy')), img_tensor)

然后就是很常规的cnn训练脚本，用resnet18作为基模型收敛更加的快

loss对应的是均方误差，所以一般降到0.25就相当于偏差只有0.5的误差了，此时模型的精度足够循迹使用了

训练一次的时间大概在2~3个小时，还请耐心等候

import os
import pandas as pd
from PIL import Image
import matplotlib.pyplot as plt

import paddle
import paddle.nn as nn
import paddle.vision.transforms as T
from paddle.io import Dataset, DataLoader
from paddle.vision.models import resnet18


paddle.set_device('gpu:0')
print(paddle.is_compiled_with_cuda())
print(paddle.get_device())
ROOT_DIR = '/dataset'
CSV_PATH = os.path.join(ROOT_DIR, 'labels.csv')
IMG_DIR = os.path.join(ROOT_DIR, 'images')


IMG_HEIGHT = 384
IMG_WIDTH = 512
BATCH_SIZE = 32
EPOCHS = 150
LR = 1e-4
WEIGHT_DECAY = 1e-5

class LaneDeviationNPYDataset(Dataset):
    def __init__(self, csv_path, npy_dir):
        self.data = pd.read_csv(csv_path)
        self.npy_dir = npy_dir

    def __getitem__(self, idx):
        row = self.data.iloc[idx]
        npy_path = os.path.join(self.npy_dir, row['filename'].replace('.jpg', '.npy'))
        image = np.load(npy_path)
        label = float(row['deviation'])
        return paddle.to_tensor(image, dtype='float32'), paddle.to_tensor(label, dtype='float32')

    def __len__(self):
        return len(self.data)


NPP_DIR = './dataset/preprocessed_npy'

full_dataset = LaneDeviationNPYDataset(CSV_PATH, NPP_DIR)
train_size = int(0.8 * len(full_dataset))
val_size = len(full_dataset) - train_size
train_dataset, val_dataset = paddle.io.random_split(full_dataset, [train_size, val_size])

train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True, num_workers=3)
val_loader = DataLoader(val_dataset, batch_size=64, num_workers=3)

class LaneRegressor(nn.Layer):
    def __init__(self):
        super().__init__()
        self.backbone = resnet18(pretrained=True)
        self.backbone.fc = nn.Sequential(
            nn.Linear(512, 256),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(256, 1)
        )

    def forward(self, x):
        return self.backbone(x)

model = LaneRegressor()


criterion = nn.SmoothL1Loss()
optimizer = paddle.optimizer.Adam(
    parameters=model.parameters(),
    learning_rate=LR,
    weight_decay=WEIGHT_DECAY
)


train_losses = []
val_losses = []
best_val_loss = float('inf')
best_model_path = os.path.join(ROOT_DIR, 'lane_deviation_best.pdparams')

for epoch in range(EPOCHS):
    model.train()
    train_loss = 0.0
    for images, labels in train_loader:
        labels = labels.unsqueeze(1)
        outputs = model(images)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        optimizer.clear_grad()
        train_loss += loss.item() * images.shape[0]

    train_loss /= len(train_loader.dataset)
    train_losses.append(train_loss)

    model.eval()
    val_loss = 0.0
    with paddle.no_grad():
        for images, labels in val_loader:
            labels = labels.unsqueeze(1)
            outputs = model(images)
            loss = criterion(outputs, labels)
            val_loss += loss.item() * images.shape[0]

    val_loss /= len(val_loader.dataset)
    val_losses.append(val_loss)

    print(f"[{epoch+1:02d}/{EPOCHS}] Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}")

    if val_loss < best_val_loss:
        best_val_loss = val_loss
        paddle.save(model.state_dict(), best_model_path)
        print(f"保存最佳模型权重，Val Loss: {best_val_loss:.4f}")
    if (epoch + 1) % 20 == 0:
        LR *= 0.5
        optimizer.set_lr(LR)
        print(f"学习率下降为：{LR:.2e}")

final_model_path = os.path.join(ROOT_DIR, 'lane_deviation_final.pdparams')
paddle.save(model.state_dict(), final_model_path)
print(f"训练结束，最终模型权重保存至：{final_model_path}")

plt.figure(figsize=(8, 6))
plt.plot(range(1, EPOCHS + 1), train_losses, label='Train Loss')
plt.plot(range(1, EPOCHS + 1), val_losses, label='Val Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.title('Training and Validation Loss')
plt.legend()
plt.grid(True)
loss_fig_path = os.path.join(ROOT_DIR, 'loss_curve.png')
plt.savefig(loss_fig_path)
print(f"Loss 曲线图已保存至：{loss_fig_path}")
plt.close()

基础模型反刍抗干扰模型

如果循迹算法抗反光、阴影能力不足怎么办，那么这些情况的数据集就没法录制了，如何加强模型在这种环境下的鲁棒性呢？据说我们上一届学长手标了上万张左右线，而我没有这种毅力，所以换了种方法。

上面训练的基础模型虽然在不同光照表现不一，但仍然具有循迹能力，只不过是内切或外切的区别，这一点比算法循迹要好上不少。

而外切或内切我们只要对偏差乘上一个比例即可解决，所以我们何不用基础模型跑的循迹制作数据集呢，在不同的情况下使用不同的比例，只要跑在赛道中间的就是好数据，可以保留。

于是，不同的光照环境我们都录到了一定量的优秀数据集，大概共20000张左右，重新用上述脚本训练，最终得到的是一个抗光照干扰能力强大的模型，而且，基础模型反刍的数据集抖动更少，比算法更加线性。

多个循迹模型

同样的图像，不同的偏差出现在数据集中非常不合理，但是在赛道中有很多这样的情况出现：如三叉口既可能向左也可能向右转、圆环入口处既可能进圆环也可能是已经进过圆环而直接驶过。

像三叉完全可以先固定打角再循迹，但圆环我则是用了第二个循迹模型，来分开类似的圆环入口图像对应的两种偏差。
其实三叉我后来也打算练一个单独的循迹模型防止在汉诺塔任务时过早因为出口处的弧度而转弯，只不过实在有点懒，看着固定路径效果不错就直接凑合着用了。

循迹模型后话

这样一套流程练下来的循迹模型的能力上限已经超过了官模的速度上限，是完全够用的，而且想加速也只需要调整偏差比例即可。根本不需要重新训练模型，我个人认为比遥控循迹更加丝滑且好用。有更好想法的大佬也可以和我交流讨论:)