【Al达人特训营】基于狗鼻子识别狗狗身份

!unzip /home/aistudio/data/data151224/dir_train.zip -d /home/aistudio/work/    #解压数据集到work下面,之后再开启时就可不运行该语句

import osimport paddleimport randomimport numpy as npfrom PIL import Imageimport matplotlib.pyplot as pltimport paddle.nn.functional as F

import osimport randomimport numpy as np
x_train = np.zeros((37459, 3, 224, 224))
class_dict = {}

class_id = 0 #一开始狗狗的编号是从0开始,也就是说狗狗是从第0类开始path='work/dir_train'file_list=os.listdir(path)
file_list.sort(key=lambda x: int(x[0:4]))for img_file in file_list:
    cur_path=os.path.join(path,img_file)    if os.path.isdir(cur_path):
        imgs=os.listdir(cur_path)        for img in imgs:            if 'jpg' in img:
                dog_id = img_file                if dog_id not in class_dict:
                    class_dict[dog_id]=class_id
                    class_id+=1fege=[] #将改变了的图片提取出来用于valother_imgs=[] #用于test的划分total=[]
total_end=[]for img_file in file_list:
    cur_path=os.path.join(path,img_file)    if os.path.isdir(cur_path):
        imgs=os.listdir(cur_path)        for img in imgs:            if 'jpg' in img:
                dog_id = img_file
                last_path =cur_path+'/'+img
                line = '%s %s\n' % (last_path, class_dict[dog_id])
                total.append(line)
                fege=img.split('.')                if(fege[0][-1]=='0'):
                    other_imgs.append(line)print('all',len(total))# random.seed(0)#保证每次产生相同的随机数# random.shuffle(other_imgs)#打乱名称次序length=len(other_imgs)
test= other_imgs[:int(length*0.06)]print('test',len(test))for index in total:    if index not in test:
        total_end.append(index)
train = total_endprint('train',len(total_end))with open('work/data/train.txt', 'w', encoding='UTF-8') as f:    for line in train:
        f.write(line)    
with open('work/data/test.txt', 'w', encoding='UTF-8') as f:    for line in test:
        f.write(line)

import osimport cv2import numpy as npfrom paddle.io import Datasetclass MyDataset(Dataset):
    def __init__(self,label_path, transform=None):#初始化数据集，将样本和标签映射到列表中
        super(MyDataset, self).__init__()
        self.data_list = []        with open(label_path,encoding='UTF-8') as f:            for line in f.readlines():
                image_path, label = line.strip().split(' ')
                self.data_list.append([image_path, label])
        self.transform = transform    def __getitem__(self, index):#定义指定index时如何获取数据,并返回单条数据（样本数据、对应的标签）
        global x_train,y_train        # 根据索引，从列表中取出一个图像
        image_path, label = self.data_list[index]        # 读取图片(以图片原本的形式加载)
        image = cv2.imread(image_path, cv2.IMREAD_UNCHANGED)        # 飞桨训练时内部数据格式默认为float32，将图像数据格式转换为 float32
        image = image.astype('float32')
        image = image /255.
        if self.transform is not None:
            image = self.transform(image)
        label = np.array([int(label)])        # 返回图像和对应标签
        return image, label    def __len__(self):#返回数据集的样本总数
        return len(self.data_list)

import paddle.vision.transforms as T'''
    对数据进行size修改,随机调整图像的亮度、对比度、饱和度和色调、基于概率来执行图片的水平翻转
'''transform = T.Compose([T.Resize(size=(224,224)),T.ColorJitter(),T.RandomHorizontalFlip(0.5),T.Transpose((2, 0, 1))])
train_dataset = MyDataset('work/data/train.txt', transform)
test_dataset = MyDataset('work/data/test.txt', transform)print('train_dataset images: ',len(train_dataset),' test_dataset images: ',len(test_dataset))

from PIL import Image
height_width = 224show_train = np.zeros((4, 3, 224, 224), dtype='float32')def show_image(img,length):
    box_size = height_width+4
    num_rows=1
    num_cols=length
    collage=Image.new(mode="RGB",
                      size=(num_cols * box_size,num_rows * box_size),
                      color=(255, 255, 255))    for col_idx in range(num_cols):
        array = (np.array(img[col_idx]) * 255).astype(np.uint8)
        array = array.transpose(1,2,0)        #将我们要显示的图贴在我们刚刚创建的collage版上
        collage.paste(  
                Image.fromarray(array), (col_idx* box_size,(num_rows-1)* box_size)#将显示的图片贴在第几行,这里设置以(列，行)为准
            )
    collage = collage.resize(( num_cols * box_size, num_rows * box_size))    return collage
random_images=np.random.randint(0,37459,size=4)
length=len(random_images)for index in range(length):
    put=random_images[index]
    show_train[index,:,:,:]=train_dataset[put][0]
show_image(show_train,length)

import paddle# 定义并初始化数据读取器train_loader = paddle.io.DataLoader(train_dataset, batch_size=128, shuffle=True, num_workers=3, drop_last=True)# 调用DataLoader迭代读取数据for batch_id, data in enumerate(train_loader()):
    imgs, labels = data    print("batch_id: {}, 训练数据shape: {}, 标签数据shape: {}".format(batch_id, imgs.shape, labels.shape))    break

#由于一共有6000个狗狗的id，所以在这我们的类别为6000resnet50 = paddle.vision.models.resnet50(num_classes=6000)
paddle.summary(resnet50,(1,3, 224, 224)) # 可视化模型组网结构和参数

import matplotlib.pyplot as pltdef draw_train_process(title,iters,loss,accs,label_cost,lable_acc):
    plt.title(title, fontsize=24)
    plt.xlabel("iter", fontsize=20)
    plt.ylabel("loss/acc", fontsize=20)
    plt.plot(iters, loss,color='red',label=label_cost) 
    plt.plot(iters, accs,color='green',label=lable_acc) 
    plt.legend()
    plt.grid()
    plt.show()

height_width=224show_test = np.zeros((2, 3, 224, 224), dtype='float32')def show_result(img,length):
    box_size = height_width+4
    num_rows=1
    num_cols=length
    collage=Image.new(mode="RGB",
                      size=(num_cols * box_size,num_rows * box_size),
                      color=(255, 255, 255))    for col_idx in range(num_cols):
        array = (np.array(img[col_idx]) * 255).astype(np.uint8)
        array = array.transpose(1,2,0)        #将我们要显示的图贴在我们刚刚创建的collage版上
        collage.paste(  
                Image.fromarray(array), (col_idx* box_size,(num_rows-1)* box_size)#将显示的图片贴在第几行,这里设置以(列，行)为准
            )
    collage = collage.resize(( num_cols * box_size, num_rows * box_size))    return collage

batch_size=128def train(model):
    all_train_iter=0
    all_train_iters=[]
    all_train_loss=[]
    all_train_accs=[]    print('start training ... ')
    model.train()

    epoch_num = 25
    
    opt = paddle.optimizer.Adam(learning_rate=0.0001,parameters=model.parameters())
    loss_fn = paddle.nn.CrossEntropyLoss()    for epoch in range(epoch_num):        for batch_id, data in enumerate(train_loader()):
            trains_data = data[0]            # 训练数据
            labels_data = data[1]            # 训练数据标签
            predicts = model(trains_data)    # 预测结果
            loss=loss_fn(predicts,labels_data)#计算loss
            all_train_iter=all_train_iter+batch_size
            acc=paddle.metric.accuracy(predicts, labels_data) #计算acc
            all_train_iters.append(all_train_iter)
            all_train_loss.append(loss)
            all_train_accs.append(acc)            # 反向传播 
            loss.backward()            if batch_id % 500 == 0:                print("epoch: {}, batch_id: {}, loss is: {}, acc is: {}".format(epoch, batch_id, loss.numpy(), acc.numpy()))            # 更新参数 
            opt.step()            # 梯度清零
            opt.clear_grad()
    draw_train_process("training",all_train_iters,all_train_loss,all_train_accs,"trainning loss","trainning acc")

train(resnet50)

# 加载测试数据集  from PIL import Image 
test_loader = paddle.io.DataLoader(test_dataset, batch_size=64, drop_last=True)
loss_fn = paddle.nn.CrossEntropyLoss()# 将该模型及其所有子层设置为预测模式resnet50.eval()for batch_id, data in enumerate(test_loader()):#     # 取出测试数据
    test_data = data[0] 
    test_label_data = data[1]
    box_size=228#     # 获取预测结果
    predicts = resnet50(test_data)print("predict finished")
loss=loss_fn(predicts,test_label_data)
acc=paddle.metric.accuracy(predicts,test_label_data) 
print("loss is: {}, acc is: {}".format(loss.numpy(), acc.numpy()))# # 从测试集中取出一组数据img, label = test_loader().next()# # 执行推理并打印结果pred_label = resnet50(img)[0].argmax()print('true label: {}, pred label: {}'.format(label[0].item(), pred_label[0].item()))# # 可视化图片show_test[0,:,:,:]=train_dataset[label[0].item()][0]#存放真实值show_test[1,:,:,:]=img[0]#存放预测值show_result(show_test,2)

y_train= np.zeros((37459, 1), dtype='int64')#这里我们取37459是因为我们划分的训练集大小为37459for i in range(len(train_dataset)):
    y_train[i, 0] = train_dataset[i][1]

y_test= np.zeros((1177, 1), dtype='int64')#这里我们取1177是因为我们划分的测试集大小为1177for i in range(len(test_dataset)):
    y_test[i, 0] = test_dataset[i][1]

y_train = np.squeeze(y_train)#删除y_train数组的单维度y_test = np.squeeze(y_test)

print('y_train',y_train)print('y_test',y_test)

from collections import defaultdict
class_to_train = defaultdict(list)for train_idx, label in enumerate(y_train):
    class_to_train[label].append(train_idx)
class_to_test = defaultdict(list)for test_idx, label in enumerate(y_test):
    class_to_test[label].append(test_idx)

import random
height_width=224num_classes = 40def read_iterator(num_batches):
    iter_step = 0
    while True:        if iter_step > num_batches:            break
        iter_step+=1
        # random_images=np.random.randint(0,100,size=num_classes)
        x = np.empty((2, num_classes, 3, height_width, height_width), dtype=np.float32)        for index in range(num_classes):            # examples_for_class = class_to_train[random_images[index]]
            examples_for_class = class_to_train[index]
            anchor_idx = random.choice(examples_for_class)
            positive_idx = random.choice(examples_for_class)            while positive_idx == anchor_idx:
                positive_idx = random.choice(examples_for_class)
            x[0, index] = train_dataset[anchor_idx][0]
            x[1, index] = train_dataset[positive_idx][0]            # x[0, random_images[index]] = train_dataset[anchor_idx][0]
            # x[1, random_images[index]] = train_dataset[positive_idx][0]
        yield x

examples = next(read_iterator(num_batches=1000))print(examples.shape)

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

        self.conv1 = paddle.nn.Conv2D(in_channels=3, 
                                      out_channels=224, 
                                      kernel_size=(3, 3),
                                      stride=2)
        self.conv2 = paddle.nn.Conv2D(in_channels=224, 
                                      out_channels=112, 
                                      kernel_size=(3,3), 
                                      stride=2) 
        self.conv3 = paddle.nn.Conv2D(in_channels=112, 
                                      out_channels=56, 
                                      kernel_size=(3,3), 
                                      stride=2)       
        self.conv4 = paddle.nn.Conv2D(in_channels=56, 
                                      out_channels=64, 
                                      kernel_size=(3,3), 
                                      stride=2)
        self.conv5 = paddle.nn.Conv2D(in_channels=64, 
                                      out_channels=128, 
                                      kernel_size=(3,3),
                                      stride=2)
       
        self.gloabl_pool = paddle.nn.AdaptiveAvgPool2D((1,1))

        self.fc = paddle.nn.Linear(in_features=128, out_features=8)    def forward(self, x):
        x = self.conv1(x)
        x = F.relu(x)
        x = self.conv2(x)
        x = F.relu(x)
        x = self.conv3(x)
        x = F.relu(x)
        x = self.conv4(x)
        x = F.relu(x)
        x = self.conv5(x)
        x = F.relu(x)
        x = self.gloabl_pool(x)
        x = paddle.squeeze(x, axis=[2, 3])
        x = self.fc(x)
        x = x / paddle.norm(x, axis=1, keepdim=True)        return x

model=MyNet()
model.train()
paddle.summary(model,(1,3, 224, 224)) # 可视化模型组网结构和参数

def train(model):
    print('start training ... ')
    model.train()
    epoch_num = 20
    num_batches=100
    opt = paddle.optimizer.Adam(learning_rate=0.0001,parameters=model.parameters())    
    for epoch in range(epoch_num):        for batch_id, data in enumerate(read_iterator(num_batches=100)):
            anchors_data, positives_data = data[0], data[1]
            anchors = paddle.to_tensor(anchors_data)
            positives = paddle.to_tensor(positives_data)
            anchor = model(anchors)
            positive = model(positives)            #通过余弦相似度进行图片间的相似度计算
            similarities = paddle.matmul(anchor, positive, transpose_y=True) 
            sparse_labels = paddle.arange(0, num_classes, dtype='int64')
            labels=np.array(sparse_labels).reshape(len(sparse_labels),1)
            labels = paddle.to_tensor(labels)
            loss = F.cross_entropy(similarities, sparse_labels)
            acc = paddle.metric.accuracy(similarities,labels)            if batch_id % 100 == 0:                print("epoch: {}, batch_id: {}, loss is: {}".format(epoch, batch_id, loss.numpy()))
            loss.backward()# 反向传播 
            opt.step()# 更新参数 
            opt.clear_grad()# 梯度清零

train(model)

test_pp=[]
train_pp=[]for data in train_dataset:
    img,label=data    if label<=[40]:
        test_pp.append(img)for data in test_dataset:
    img,label=data    if label<=[40]:
        train_pp.append(img)
test_pp=paddle.to_tensor(test_pp)
train_pp=paddle.to_tensor(train_pp)
test_pp=model(test_pp)
train_pp=model(train_pp)
similarities = paddle.matmul(test_pp, train_pp, transpose_y=True) 
sparse_labels = paddle.arange(0, len(test_pp), dtype='int64')
labels=np.array(sparse_labels).reshape(len(sparse_labels),1)
labels = paddle.to_tensor(labels)
loss = F.cross_entropy(similarities, sparse_labels)
acc = paddle.metric.accuracy(similarities,labels)print("loss is: {}, acc is: {}".format(loss.numpy(), acc.numpy()))