【车道线+自动驾驶】用Paddle高层API实现车道线转角回归模型+部署

import osimport cv2import iofrom tqdm import tqdmimport numpy as npimport matplotlib.pyplot as pltfrom PIL import Image as PilImageimport paddlefrom paddle.nn import functional as F

paddle.__version__

!unzip -oq /home/aistudio/data/data46903/0728_carline.zip

img_folder_path = "data/img"dataset_path = "data"txt_path = "data/data.txt"IMAGE_SIZE = (224, 224)

# 检查数据集中图像和打角数据个数是否对应import os# 1、读取图像数据和角度数据angle_list = []
f = open(txt_path)for txt in f.readlines():
    txt.split("\n")
    angle_num = int(txt)
    angle_list.append(angle_num)
img_list = []for img_path in tqdm(os.listdir(img_folder_path)):
    img_path = os.path.join(img_folder_path, img_path)
    img_list.append(img_path)# 2、数据归一化（如果过于发散，模型不容易收敛）# 这里我的打角数据大小为：900-2100，中间值为1500for i, angle in enumerate(angle_list): 
    angle = (angle-1500)/600
    angle_list[i] = angletry:
    os.remove(os.path.join(dataset_path, "train.txt"))
    os.remove(os.path.join(dataset_path, "eval.txt"))except:    passtrain_txt = open(os.path.join(dataset_path, "train.txt"), "w")
eval_txt = open(os.path.join(dataset_path, "eval.txt"), "w")
n = 0for img_path, label in tqdm(zip(img_list, angle_list)):
    n += 1
    if n % 10 != 0:
        train_txt.write(img_path+" "+str(label))
        train_txt.write("\n")    else:
        eval_txt.write(img_path+" "+str(label))
        eval_txt.write("\n")
train_txt.close()
eval_txt.close()

import randomimport iofrom paddle.io import Datasetfrom paddle.vision.transforms import transforms as Tfrom PIL import Image as PilImageimport numpy as npclass MyDataset(Dataset):
    """
    数据集定义
    """
    def __init__(self, mode, dataset_path):
        """
        构造函数
        """
        self.image_size = IMAGE_SIZE
        self.mode = mode.lower()
        self.dataset_path = dataset_path        
        assert self.mode in ['train', 'test', 'eval'], \            "mode should be 'train' or 'test' or 'eval', but got {}".format(self.mode)
        
        self.train_images = []
        self.label_list = []        with open(os.path.join(self.dataset_path, ('{}.txt'.format(self.mode))), 'r') as f:            
            for line in tqdm(f.readlines()):
                image, label = line.strip().split(' ')
                
                img = PilImage.open(image)

                self.train_images.append(image)
                self.label_list.append(label)        

    def _load_img(self, path, color_mode='rgb', transforms=[]):
        """
        统一的图像处理接口封装，用于规整图像大小和通道
        """
        with open(path, 'rb') as f:
            img = PilImage.open(io.BytesIO(f.read()))            if color_mode == 'grayscale':                # if image is not already an 8-bit, 16-bit or 32-bit grayscale image
                # convert it to an 8-bit grayscale image.
                if img.mode not in ('L', 'I;16', 'I'):
                    img = img.convert('L')            elif color_mode == 'rgba':                if img.mode != 'RGBA':
                    img = img.convert('RGBA')            elif color_mode == 'rgb':                if img.mode != 'RGB':
                    img = img.convert('RGB')            else:                raise ValueError('color_mode must be "grayscale", "rgb", or "rgba"')            
            return T.Compose([
                T.Resize(self.image_size)
            ] + transforms)(img)    def __getitem__(self, idx):
        """
        返回 image, label
        """
        train_image = self._load_img(self.train_images[idx], 
                                     transforms=[
                                         T.Transpose(), 
                                         T.Normalize(mean=[127.5], std=[127.5]) # 归一化处理，将0~255归一化到-1~1
                                     ]) # 加载原始图像
        label = self.label_list[idx] # 加载Label
    
        # 返回image, label
        train_image = np.array(train_image, dtype='float32')
        label = np.array(label, dtype='float32')        return train_image, label        
    def __len__(self):
        """
        返回数据集总数
        """
        return len(self.train_images)# 定义训练集和验证集train_dataset = MyDataset(mode='train', dataset_path=dataset_path) # 训练数据集val_dataset = MyDataset(mode='eval', dataset_path=dataset_path) # 验证数据集

x, y = val_dataset.__getitem__(0)print(x, y)print(x.shape)print(train_dataset.__len__())with open(os.path.join(dataset_path, 'train.txt'), 'r') as f:
    i = 0

    for line in f.readlines():
        image_path, label = line.strip().split(' ')
        image = np.array(PilImage.open(image_path))    
        if i > 2:            break
        # 进行图片的展示
        plt.figure()

        plt.title(label)
        plt.imshow(image.astype('uint8'))
        plt.axis('off')

        plt.show()
        i = i + 1

import paddleimport paddle.nn as nnimport paddle.nn.functional as Fclass My_Model(nn.Layer):
    def __init__(self, num_classes):
        super(My_Model, self).__init__()

        self.conv1 = paddle.nn.Conv2D(in_channels=3, out_channels=32, kernel_size=(3, 3))
        self.pool1 = paddle.nn.MaxPool2D(kernel_size=2, stride=2)

        self.conv2 = paddle.nn.Conv2D(in_channels=32, out_channels=64, kernel_size=(3,3))
        self.pool2 = paddle.nn.MaxPool2D(kernel_size=2, stride=2)

        self.conv3 = paddle.nn.Conv2D(in_channels=64, out_channels=32, kernel_size=(3,3))
        self.pool3 = paddle.nn.MaxPool2D(kernel_size=2, stride=2)
        
        self.conv4 = paddle.nn.Conv2D(in_channels=32, out_channels=16, kernel_size=(3,3))

        self.flatten = paddle.nn.Flatten()

        self.linear1 = paddle.nn.Linear(in_features=9216, out_features=16)
        self.linear2 = paddle.nn.Linear(in_features=16, out_features=1)    def forward(self, x):
        x = self.conv1(x)
        x = F.relu(x)
        x = self.pool1(x)

        x = self.conv2(x)
        x = F.relu(x)
        x = self.pool2(x)

        x = self.conv3(x)
        x = F.relu(x)
        x = self.pool3(x)

        x= self.conv4(x)
        x = F.relu(x)

        x = self.flatten(x)
        x = self.linear1(x)
        x = F.relu(x)
        x = self.linear2(x)        return x

paddle.summary(My_Model(num_classes=1), input_size=(1,3, 224,224))

import paddleclass SaveBestModel(paddle.callbacks.Callback):
    def __init__(self, target=0.5, path='./best_model', verbose=0):
        self.target = target
        self.epoch = None
        self.path = path    def on_epoch_end(self, epoch, logs=None):
        self.epoch = epoch    def on_eval_end(self, logs=None):
        if logs.get('loss')[0] < self.target:
            self.target = logs.get('loss')[0]
            self.model.save(self.path)            print('best model is loss {} at epoch {}'.format(self.target, self.epoch))

callback_visualdl = paddle.callbacks.VisualDL(log_dir='./')
callback_savebestmodel = SaveBestModel(target=1, path='./model')
callbacks = [callback_visualdl, callback_savebestmodel]# 模型初始化model = paddle.Model(My_Model(num_classes=1)) # 线性回归模型# 优化器optim = paddle.optimizer.Momentum(learning_rate=0.0001, 
                                 momentum=0.9, 
                                 parameters=model.parameters())# 损失函数loss = paddle.nn.MSELoss()
model.prepare(optimizer=optim, loss=loss) # 线性回归模型一定不能用交叉熵，因为这个自带softmax，需要指定输出类别# 用 DataLoader 实现数据加载train_loader = paddle.io.DataLoader(train_dataset, places=paddle.CUDAPlace(0), batch_size=64)
eval_loader = paddle.io.DataLoader(val_dataset, places=paddle.CUDAPlace(0), batch_size= 64)

model.fit(train_loader, 
          eval_loader, 
          epochs=5, 
          callbacks=callbacks,
          verbose=1)

# training=True时，只会保存优化器参数和模型参数# training=False时，会保存模型结构、参数和优化器结构model.save("./model_dir/model", training=False)

import cv2import numpy as npfrom paddle.inference import Configfrom paddle.inference import create_predictor# ————————————————图像预处理函数————————————————def resize(img, target_size):
    """ resize """
    percent = float(target_size) / min(img.shape[0], img.shape[1])
    resized_width = int(round(img.shape[1] * percent))
    resized_height = int(round(img.shape[0] * percent))
    resized_short = cv2.resize(img, (resized_width, resized_height))
    resized = cv2.resize(resized_short, (target_size, target_size))    return resizeddef preprocess(img, target_size):
    mean = [127.5, 127.5, 127.5]
    std = [127.5, 127.5, 127.5]    # resize
    img = resize(img, target_size)    # bgr-> rgb && hwc->chw
    img = img[:, :, ::-1].astype('float32').transpose((2, 0, 1))
    img_mean = np.array(mean).reshape((3, 1, 1))
    img_std = np.array(std).reshape((3, 1, 1))
    img -= img_mean
    img /= img_std    return img[np.newaxis, :]#——————————————————————模型配置、预测相关函数——————————————————————————def predict_config(model_file, params_file):
    # 根据预测部署的实际情况，设置Config
    config = Config()    # 读取模型文件
    config.set_prog_file(model_file)
    config.set_params_file(params_file)    # Config默认是使用CPU预测，若要使用GPU预测，需要手动开启，设置运行的GPU卡号和分配的初始显存。
    config.enable_use_gpu(500, 0)    # 可以设置开启IR优化、开启内存优化。
    config.switch_ir_optim()
    config.enable_memory_optim()
    predictor = create_predictor(config)    return predictordef predict(image, predictor, target_size):
    img = preprocess(image, target_size)
    input_names = predictor.get_input_names()
    input_tensor = predictor.get_input_handle(input_names[0])
    input_tensor.reshape(img.shape)
    input_tensor.copy_from_cpu(img.copy())    # 执行Predictor
    predictor.run()    # 获取输出
    output_names = predictor.get_output_names()
    output_tensor = predictor.get_output_handle(output_names[0])
    output_data = output_tensor.copy_to_cpu()    print("output_names", output_names)    print("output_tensor", output_tensor)    print("output_data", output_data)    return output_dataif __name__ == '__main__':
    model_file = "model_dir/model.pdmodel"
    params_file = "model_dir/model.pdiparams"
    
    import random    # image = cv2.imread("data/img/7419.jpg")
    # image = cv2.imread("data/img/8891.jpg")
    image = cv2.imread("data/img/{}.jpg".format(random.randint(0,5000)))
    
    predictor = predict_config(model_file, params_file)
    res = predict(image, predictor, target_size=224)    # 进行图片的展示
    plt.figure()

    plt.title(res)
    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
    plt.imshow(image.astype('uint8'))
    plt.axis('on')

    plt.show()