图像分类概述
概述
什么是图像分类
- 图像分类就是将不同的图像划分到不同类别，实现最小分类误差、最高精度。手写体识别就是一个经典的图像分类问题，它将输入图像分为0~9某个数字中，实际就是将输入图像分为10类
图像分类粒度
- 跨物种级图像分类：在不同物种层次上识别不同对象，如猫狗分类
- 子类细粒度图像分类：同一大类下，不同子类的分类。如不同的鸟分类，不同的狗分类
- 实例级图像分类：区分不同的个体。如人脸识别
图像分类发展历程
- 图像分类任务从传统的方法到基于深度学习的方法，经历了几十年的发展
图像分类问题的挑战
- 虽然图像分类大赛正确率已经接近极限，但在实际工程应用中，面临诸多挑战与难题：
- 类别不均衡
- 数据集小
- 巨大的类内差异
- 实际应用复杂情况：光照、遮挡、模糊、角度变化、干扰
常用数据集介绍
MNIST数据集
- 手写数字的数据集，来自美国国家标准与技术研究所（National Institute of Standards and Technology，NIST），发布与1998年
- 样本来自250个不同人的手写数字，50%高中学生，50%是人口普查局的工作人员
- 数字从0～9，图片大小是28×28像素，训练数据集包含60000个样本，测试数据集包含10000个样本
- 下载地址：http://yann.lecun.com/exdb/mnist/
CIFAR10数据集
- CIFAR10数据集由Alex Krizhevsky、Vinod Nair和Geoffrey Hinton等人收集
- 包含6万张彩色图像，图像大小是32×32，共有10个类，每类有6000张图。其中，5万张图组成训练集合，训练集合中的每一类均等，都有5000张图；剩余1万张图作为测试集合，测试集合中的每一类也均等，各有1000张图
- 10个类别是：airplane、automobile、bird、cat、deer、dog、frog、horse、ship和truck
- 下载地址：http://www.cs.toronto.edu/~kriz/cifar.html
ImageNet数据集
- ImageNet数据集由李飞飞实验室发布于2009年- 1400多万幅图片，涵盖2万多个类别的标注与超过百万的边界框标注，每一个类别大约有500～1000张图片
- ImageNet竞赛使用的是ImageNet完整数据集的一个子类，包括1000类，其中大部分是动物。在深度学习任务中，我们经常会使用ImageNet预训练的模型
- 下载地址：http://www.image-net.org/
FDDB人脸数据集
- 发布于2010年，是被广泛用于人脸检测方法评测的一个数据集
- 共2845张图像，包含有5171张人脸图像，大部是自然条件下拍摄的名人
- 下载地址：http://vis-www.cs.umass.edu/fddb/index.html#download
WIDER Face数据集
- 2015年由香港中文大学发布
- 32203张图像，共有393703张人脸图像，比FDDB数据集大10倍，而且在面部的尺寸、姿势、遮挡、表情、妆容和光照上都有很大的变化，自发布后广泛应用于评估性能比传统方法更强大的卷积神经网络
- 下载地址：http://mmlab.ie.cuhk.edu.hk/projects/WIDERFace/
图像分类的应用
图像分类的应用
- 交通违章识别
- 安检系统
- 人脸识别
- 生物种群数量统计
- 工业质检
- 工地安全监测
- 病虫害识别
- 医疗诊断
利用CNN实现图片分类
思路及实现
数据集介绍
- 来源：爬虫从百度图片搜索结果爬取
- 内容：包含1036张水果图片，共5个类别（苹果288张、香蕉275张、葡萄216张、橙子276张、梨251张）
- 图像预处理时，将其中10%作为测试数据，90%作为训练数据
总体步骤
- 数据预处理：建立分类文件，建立训练集、测试集
- 训练与模型评估
- 读取测试图片，进行预测
数据预处理
- 图片位于5个目录，遍历每个目录，将其中90%写入训练集文件，10%写入测试集文件，文件中记录了图片的路径，用于数据读取器进行读取
- 生成3个文件：readme.json（汇总文件）、trainer.list（训练集）、test.list（测试集）
模型结构
利用CNN实现图片分类
#利用cnn实现水果分类
###################预处理标注测试集训练集###################
import os
name_dict = {'apple':0,'banana':1,'grape':2,'orange':3,'pear':4,}
data_root_path = 'data/fruits/'#数据集目录
test_file_path = data_root_path+'test.txt'#测试集目录
train_file_path = data_root_path+'train.txt'#训练集目录
name_data_list = {}#字典 key 类别 value路径列表
# 将图片路径存入name_data_list字典
def add_to_dict(path,name):#path图片路径 name类别名称
if name not in name_data_list:#判断类别是否存在
name_data_list[name] = []#不存在就新增
name_data_list[name].append(path)#然后将路径追加到字典
#遍历数据集下的子目录
dirs = os.listdir(data_root_path)
print(dirs)#['banana', 'grape', 'orange', 'apple', 'pear']
for d in dirs:#遍历目录下的所有文件
# 简洁版本
# sub_dir_path = os.listdir(data_root_path+d)
# for i in sub_dir_path:
# add_to_dict(data_root_path+d+'/'+i,d)
# 教学版本
sub_dir_path = data_root_path+d
#判断是否为目录
if os.path.isdir(sub_dir_path):
imgs = os.listdir(data_root_path+d)#列出所有图片
for img in imgs:#遍历所有图片
#拼接图片路径
img_path = data_root_path + d + '/' + img
#添加到字典
add_to_dict(img_path, d)
else:
#如果为文件
pass
# 清空训练集和测试集
with open(test_file_path,'w') as f:
pass
with open(train_file_path,'w') as f:
pass
#遍历字典 分配测试集和训练集 1:9
for name,img_list in name_data_list.items():
i = 0
num = len(img_list)#图片数量
print('%s,%d张'%(name,num))
for img in img_list:#遍历所有图片路径
#处理文本格式
line = "%s\t%d\n"%(img,name_dict[name])
# print(line) # data/fruits/banana/15.jpg 1
if i % 10 == 0:#测试集
with open(test_file_path, 'a') as f:
f.write(line)
else:#训练集
with open(train_file_path, 'a') as f:
f.write(line)
i+=1
###################模型定义训练保存###################
import paddle
import paddle.fluid as fluid
import time
import matplotlib.pyplot as plt
import sys
import os
import numpy as np
from multiprocessing import cpu_count # 并发处理模块
def train_mapper(sample):
"""
接收文本，根据图像路径读取图像，返回图像数据
:param sample: 文本中的一行（图像路径，类别）
:return: 返回图像数据（张量），类别
"""
img_path,label = sample #img_path路径 label类别
#读取图像数据
img = paddle.dataset.image.load_image(img_path)
#缩放（图像统一大小），裁剪（缓解过拟合）
img = paddle.dataset.image.simple_transform(
im=img, # 原图像
resize_size=128, # 缩放到128*128
crop_size=128, # 裁剪到128*128
is_train=True, # 是否训练模式（随机裁剪）
is_color=True) # 是否彩色图像
# 归一化（将每个像素值缩放到0～1之间）
# 归一化优点 缓解过拟合 缓解梯度消失 加快收敛速度
img = img.astype('float32')/255.0
return img,label # 返回归一化处理的数据和标签
#从训练集读取图片路径，类别
def train_r(train_list,buffered_size=1024):
def reader():
with open(train_list,'r') as f:
lines = f.readlines()
for i in lines: #遍历每行
i = i.replace("\n",'')#替换换行符
img_list,label = i.split("\t")#以tab分隔
yield img_list,int(label)
return paddle.reader.xmap_readers(
train_mapper,#下一步处理的函数
reader,#原始读取器
cpu_count(),#线程数量（和逻辑cpu数量一直）
buffered_size#缓冲区大小（预分配内存）
)
def create_CNN(image,type_size):
"""
定义CNN模型
:param image: 输入图像数据
:param type_size: 类别数量
:return: 模型预测值（一组概率）
"""
#第一组卷积/激活/池化/dropout
conv_pool_1 = fluid.nets.simple_img_conv_pool(
input=image,#输入第一层为原始图像
filter_size=3,#卷积核大小3*3
num_filters=32,#卷积核数量32
pool_size=2,#池化区域2*2
pool_stride=2,#池化步长
act='relu')#激活函数relu
drop = fluid.layers.dropout(x=conv_pool_1,#输入
dropout_prob=0.5)#丢弃率
#第二组卷积/激活/池化/dropout
conv_pool_2 = fluid.nets.simple_img_conv_pool(
input=drop,#上一层输出作为输入
filter_size=3,#卷积核大小3*3
num_filters=64,#卷积核数量64
pool_size=2,#池化区域2*2
pool_stride=2,#池化步长
act='relu')#激活函数relu
drop = fluid.layers.dropout(x=conv_pool_2,#输入
dropout_prob=0.5)#丢弃率
#第三组卷积/激活/池化/dropout
conv_pool_3 = fluid.nets.simple_img_conv_pool(
input=drop,#上一层输出作为输入
filter_size=3,#卷积核大小3*3
num_filters=64,#卷积核数量64
pool_size=2,#池化区域2*2
pool_stride=2,#池化步长
act='relu')#激活函数relu
drop = fluid.layers.dropout(x=conv_pool_3,#输入
dropout_prob=0.5)#丢弃率
#第一层fc
fc = fluid.layers.fc(input=drop,#输入
size=512,#神经元数量（输出这个数）
act='relu')#激活函数
drop = fluid.layers.dropout(x=fc,#输入
dropout_prob=0.5)#丢弃率
#第二层fc（输出层）
predict = fluid.layers.fc(input=drop,#输入
size=type_size,#输出值为类别数
act='softmax')#输出层采用softmax
return predict
BATCH_SIZE = 32#一个批次32个样本
random_train_reader = paddle.reader.shuffle(#随机读取器
train_r(train_file_path), # 原reader读取器
buf_size=1300) # 打乱范围
batch_train_reader = paddle.batch(#批量读取器
random_train_reader,
BATCH_SIZE) # 每个批次32个
#占位符
image = fluid.layers.data('image',#名称
shape=[3,128,128],#形状
dtype='float32')#类型
label = fluid.layers.data('label',#名称
shape=[1],#形状
dtype='int64')#标签为整数
#调用函数创建CNN
predict = create_CNN(image=image,type_size=5)
#损失函数分类问题使用交叉熵
cost = fluid.layers.cross_entropy(input=predict,#预测值
label=label)#标签
# 求均值
avg_cost = fluid.layers.mean(cost)
# 优化器
optimizer = fluid.optimizer.Adam(learning_rate=0.001)
optimizer.minimize(avg_cost)# 指定优化的目标函数
# 计算准确率
accurcy = fluid.layers.accuracy(input=predict,#预测值
label=label)#标签
# 计算准确率
# correct_pred = fluid.layers.equal(
# fluid.layers.argmax(predict, 1), # 返回预测结果中概率最大的索引
# fluid.layers.argmax(label, 1)) # 标签值中概率最大的索引
# accurcy2 = fluid.layers.reduce_mean(fluid.layers.cast(x=correct_pred, dtype='float32'))
# 执行器
place = fluid.CUDAPlace(0) # 第一个GPU
# place = fluid.CPUPlace() # CPU
exe = fluid.Executor(place) # 执行器
exe.run(fluid.default_startup_program()) # 初始化
# feeder
feeder = fluid.DataFeeder(feed_list=[image,label], # 喂入哪些参数
place=place)# 在哪个设备上运行
costs = [] # 存放损失值
accs = []# 存放准确率
times = 0#计数器
batches = [] # 存放迭代次数
for epoch in range(10):
for bat_id,data in enumerate(batch_train_reader()): # 内层循环控制批次
times += 1
c,a = exe.run(
program=fluid.default_main_program(),
feed=feeder.feed(data), # feed()方法返回字典
fetch_list=[avg_cost,accurcy]) # 要返回的值
if bat_id % 20 == 0:
print("epoch: %d,bat_id: %d,cost: %f,acc: %f"%(epoch,bat_id,c[0],a[0]))
accs.append(a[0]) # 记录准确率
costs.append(c[0]) # 记录损失值
batches.append(times) # 存放迭代次数
# 保存模型
model_save_dir = 'model/fruits/' # 模型保存目录
if not os.path.exists(model_save_dir):
os.makedirs(model_save_dir) # 不存在就创建
# 保存用于可以增量训练的模型
# fluid.io.save_persistables()
# 保存用于推理的模型
fluid.io.save_inference_model(model_save_dir, # 模型保存目录
['image'], # 模型预测时需要喂入的参数名称
[predict], # 模型预测结果
exe) # 执行器
# 训练过程可视化
plt.figure("Training Cost")
plt.title("Training", fontsize=24)
plt.xlabel("iter", fontsize=14)
plt.ylabel("cost/acc", fontsize=14)
plt.plot(batches, costs, color="red", label="Training Cost")
plt.plot(batches, accs, color="green", label="Training Acc")
plt.legend()
plt.grid()
plt.savefig("train.png")
plt.show()
###################预测###################
from PIL import Image
#读取图像缩放，归一化
def load_img(path):
# load_and_transform封装了以下2个方法
# paddle.dataset.image.load_image
# paddle.dataset.image.simple_transform
img = paddle.dataset.image.load_and_transform(
filename=path,#图片路径
resize_size = 128, # 缩放到128*128
crop_size = 128, # 裁剪到128*128
is_train = False# 是否训练模式（随机裁剪）
).astype('float32')
img = img/255.0#归一化
return img
#执行器
# place = fluid.CUDAPlace(0)
place = fluid.CPUPlace()#测试计算量小用cpu
infer_exe = fluid.Executor(place) # 执行器
model_save_dir = 'model/fruits/' # 模型保存目录
infer_prog,feed_vars,fetch_targets = fluid.io.load_inference_model(model_save_dir,infer_exe)
## 测试集reader
infer_reader = paddle.batch(
paddle.dataset.uci_housing.test(),#测试集
batch_size=200) # 只有102笔数据,实际是一次性读取
test_data = next(infer_reader())#读取一批数据
#读取测试图片
test_img = 'apple_1.png'#测试图片路径
infer_imgs = []#存放测试图片
infer_imgs.append(load_img(test_img))#读取图像，添加到列表
infer_imgs = np.array(infer_imgs)
# 参数字典
params = {feed_vars[0]:infer_imgs}
# 执行推理
results = infer_exe.run(
program=infer_prog,
feed=params, # 喂入的参数
fetch_list=[fetch_targets]) # 要返回的值
name_dict = {'apple':0,'banana':1,'grape':2,'orange':3,'pear':4,}
print(results)
index = np.argmax(results[0][0])
for k,v in name_dict.items():
if v == index:
print("预测结果",k)
break
img = Image.open(test_img)
plt.imshow(img)
plt.show()利用VGG实现图片分类
#利用VGG实现水果分类
###################预处理标注测试集训练集###################
import os
name_dict = {'apple':0,'banana':1,'grape':2,'orange':3,'pear':4,}
data_root_path = 'data/fruits/'#数据集目录
test_file_path = data_root_path+'test.txt'#测试集目录
train_file_path = data_root_path+'train.txt'#训练集目录
name_data_list = {}#字典 key 类别 value路径列表
# 将图片路径存入name_data_list字典
def add_to_dict(path,name):#path图片路径 name类别名称
if name not in name_data_list:#判断类别是否存在
name_data_list[name] = []#不存在就新增
name_data_list[name].append(path)#然后将路径追加到字典
#遍历数据集下的子目录
dirs = os.listdir(data_root_path)
print(dirs)#['banana', 'grape', 'orange', 'apple', 'pear']
for d in dirs:#遍历目录下的所有文件
# 简洁版本
# sub_dir_path = os.listdir(data_root_path+d)
# for i in sub_dir_path:
# add_to_dict(data_root_path+d+'/'+i,d)
# 教学版本
sub_dir_path = data_root_path+d
#判断是否为目录
if os.path.isdir(sub_dir_path):
imgs = os.listdir(data_root_path+d)#列出所有图片
for img in imgs:#遍历所有图片
#拼接图片路径
img_path = data_root_path + d + '/' + img
#添加到字典
add_to_dict(img_path, d)
else:
#如果为文件
pass
# 清空训练集和测试集
with open(test_file_path,'w') as f:
pass
with open(train_file_path,'w') as f:
pass
#遍历字典 分配测试集和训练集 1:9
for name,img_list in name_data_list.items():
i = 0
num = len(img_list)#图片数量
print('%s,%d张'%(name,num))
for img in img_list:#遍历所有图片路径
#处理文本格式
line = "%s\t%d\n"%(img,name_dict[name])
# print(line) # data/fruits/banana/15.jpg 1
if i % 10 == 0:#测试集
with open(test_file_path, 'a') as f:
f.write(line)
else:#训练集
with open(train_file_path, 'a') as f:
f.write(line)
i+=1
###################模型定义训练保存###################
import paddle
import paddle.fluid as fluid
import time
import matplotlib.pyplot as plt
import sys
import os
import numpy as np
from multiprocessing import cpu_count # 并发处理模块
def train_mapper(sample):
"""
接收文本，根据图像路径读取图像，返回图像数据
:param sample: 文本中的一行（图像路径，类别）
:return: 返回图像数据（张量），类别
"""
img_path,label = sample #img_path路径 label类别
#读取图像数据
img = paddle.dataset.image.load_image(img_path)
#缩放（图像统一大小），裁剪（缓解过拟合）
img = paddle.dataset.image.simple_transform(
im=img, # 原图像
resize_size=128, # 缩放到128*128
crop_size=128, # 裁剪到128*128
is_train=True, # 是否训练模式（随机裁剪）
is_color=True) # 是否彩色图像
# 归一化（将每个像素值缩放到0～1之间）
# 归一化优点 缓解过拟合 缓解梯度消失 加快收敛速度
img = img.astype('float32')/255.0
return img,label # 返回归一化处理的数据和标签
#从训练集读取图片路径，类别
def train_r(train_list,buffered_size=1024):
def reader():
with open(train_list,'r') as f:
lines = f.readlines()
for i in lines: #遍历每行
i = i.replace("\n",'')#替换换行符
img_list,label = i.split("\t")#以tab分隔
yield img_list,int(label)
return paddle.reader.xmap_readers(
train_mapper,#下一步处理的函数
reader,#原始读取器
cpu_count(),#线程数量（和逻辑cpu数量一直）
buffered_size#缓冲区大小（预分配内存）
)
def vgg_bn_drop(image,type_size):#VGG模型
#内部函数
def conv_block(x,num_filter,groups,dropouts):
"""
创建连续n个卷积和1个池化层
:param x: 输入
:param num_filter: 卷积核数量
:param groups: 连续几个卷积层
:param dropouts: 每个卷积层的丢弃率
:return: 卷积池化的结果
"""
#batch normal：每一次进行归一化处理，可以理解成对特征数据整体的缩放和平移 能缓解梯度消失 过拟合 增加模型的稳定性
return fluid.nets.img_conv_group(
input=x,#输入
conv_num_filter=[num_filter]*groups,#卷积核数量
# [2]*3 ==> [2,2,2] [64]*2 ==> [64,64]
conv_filter_size=3,#卷积核大小 3*3
pool_type='max',#池化类型 最大池化
pool_size=2,#池化大小 2*2
pool_stride=2,#池化步长 2
conv_act='relu',#激活函数
conv_with_batchnorm=True)# 是否使用batch normal
# 连续5组卷积池化
conv1=conv_block(image,64,2,[0.0,0.0])
conv2=conv_block(conv1,128,2,[0.0,0.0])
conv3=conv_block(conv2,256,3,[0.0,0.0,0.0])
conv4=conv_block(conv3,512,3,[0.0,0.0,0.0])
conv5=conv_block(conv4,512,3,[0.0,0.0,0.0])
#dropout
drop = fluid.layers.dropout(x=conv5,#输入
dropout_prob=0.5)#丢弃率
#fc1 全连接层
fc1 = fluid.layers.fc(input=drop,#输入
size=512,#大小
act=None)#不使用激活函数
#bn层
bn = fluid.layers.batch_norm(input=fc1,#输入
act='relu')#使用激活函数relu
#dropout
drop2 = fluid.layers.dropout(x=bn,#输入
dropout_prob=0.0)#丢弃率
# fc2 全连接层
fc2 = fluid.layers.fc(input=drop2,#输入
size=512,#大小
act=None)#不使用激活函数
#输出层
predict = fluid.layers.fc(input=fc2,#输入
size=type_size,#大小
act='softmax')#使用激活函数softmax
return predict
BATCH_SIZE = 32#一个批次32个样本
random_train_reader = paddle.reader.shuffle(#随机读取器
train_r(train_file_path), # 原reader读取器
buf_size=1300) # 打乱范围
batch_train_reader = paddle.batch(#批量读取器
random_train_reader,
BATCH_SIZE) # 每个批次32个
#占位符
image = fluid.layers.data('image',#名称
shape=[3,128,128],#形状
dtype='float32')#类型
label = fluid.layers.data('label',#名称
shape=[1],#形状
dtype='int64')#标签为整数
#调用函数创建vgg
predict = vgg_bn_drop(image=image,type_size=5)
#损失函数分类问题使用交叉熵
cost = fluid.layers.cross_entropy(input=predict,#预测值
label=label)#标签
# 求均值
avg_cost = fluid.layers.mean(cost)
# 优化器
optimizer = fluid.optimizer.Adam(learning_rate=0.000001)
optimizer.minimize(avg_cost)# 指定优化的目标函数
# 计算准确率
accurcy = fluid.layers.accuracy(input=predict,#预测值
label=label)#标签
# 计算准确率
# correct_pred = fluid.layers.equal(
# fluid.layers.argmax(predict, 1), # 返回预测结果中概率最大的索引
# fluid.layers.argmax(label, 1)) # 标签值中概率最大的索引
# accurcy2 = fluid.layers.reduce_mean(fluid.layers.cast(x=correct_pred, dtype='float32'))
# 执行器
place = fluid.CUDAPlace(0) # 第一个GPU
# place = fluid.CPUPlace() # CPU
exe = fluid.Executor(place) # 执行器
exe.run(fluid.default_startup_program()) # 初始化
# feeder
feeder = fluid.DataFeeder(feed_list=[image,label], # 喂入哪些参数
place=place)# 在哪个设备上运行
costs = [] # 存放损失值
accs = []# 存放准确率
times = 0#计数器
batches = [] # 存放迭代次数
for epoch in range(10):
for bat_id,data in enumerate(batch_train_reader()): # 内层循环控制批次
times += 1
c,a = exe.run(
program=fluid.default_main_program(),
feed=feeder.feed(data), # feed()方法返回字典
fetch_list=[avg_cost,accurcy]) # 要返回的值
if bat_id % 20 == 0:
print("epoch: %d,bat_id: %d,cost: %f,acc: %f"%(epoch,bat_id,c[0],a[0]))
accs.append(a[0]) # 记录准确率
costs.append(c[0]) # 记录损失值
batches.append(times) # 存放迭代次数
# 保存模型
model_save_dir = 'model/fruits/' # 模型保存目录
if not os.path.exists(model_save_dir):
os.makedirs(model_save_dir) # 不存在就创建
# 保存用于可以增量训练的模型
# fluid.io.save_persistables()
# 保存用于推理的模型
fluid.io.save_inference_model(model_save_dir, # 模型保存目录
['image'], # 模型预测时需要喂入的参数名称
[predict], # 模型预测结果
exe) # 执行器
# 训练过程可视化
plt.figure("Training Cost")
plt.title("Training", fontsize=24)
plt.xlabel("iter", fontsize=14)
plt.ylabel("cost/acc", fontsize=14)
plt.plot(batches, costs, color="red", label="Training Cost")
plt.plot(batches, accs, color="green", label="Training Acc")
plt.legend()
plt.grid()
plt.savefig("train.png")
plt.show()
###################预测###################
from PIL import Image
#读取图像缩放，归一化
def load_img(path):
# load_and_transform封装了以下2个方法
# paddle.dataset.image.load_image
# paddle.dataset.image.simple_transform
img = paddle.dataset.image.load_and_transform(
filename=path,#图片路径
resize_size = 128, # 缩放到128*128
crop_size = 128, # 裁剪到128*128
is_train = False# 是否训练模式（随机裁剪）
).astype('float32')
img = img/255.0#归一化
return img
#执行器
# place = fluid.CUDAPlace(0)
place = fluid.CPUPlace()#测试计算量小用cpu
infer_exe = fluid.Executor(place) # 执行器
model_save_dir = 'model/fruits/' # 模型保存目录
infer_prog,feed_vars,fetch_targets = fluid.io.load_inference_model(model_save_dir,infer_exe)
## 测试集reader
infer_reader = paddle.batch(
paddle.dataset.uci_housing.test(),#测试集
batch_size=200) # 只有102笔数据,实际是一次性读取
test_data = next(infer_reader())#读取一批数据
#读取测试图片
test_img = 'apple_1.png'#测试图片路径
infer_imgs = []#存放测试图片
infer_imgs.append(load_img(test_img))#读取图像，添加到列表
infer_imgs = np.array(infer_imgs)
# 参数字典
params = {feed_vars[0]:infer_imgs}
# 执行推理
results = infer_exe.run(
program=infer_prog,
feed=params, # 喂入的参数
fetch_list=[fetch_targets]) # 要返回的值
name_dict = {'apple':0,'banana':1,'grape':2,'orange':3,'pear':4,}
print(results)
index = np.argmax(results[0][0])
for k,v in name_dict.items():
if v == index:
print("预测结果",k)
break
img = Image.open(test_img)
plt.imshow(img)
plt.show()瓷砖检测
通过瑕疵检测对不同瑕疵的瓷砖进行分类
通过旋转图片的方式进行数据增强
# 图像样本预处理
import cv2
import numpy as np
import os
import random
import matplotlib.pyplot as plt
from global_var import *
from math import *
# 不切边旋转
def remote(img, angle):
h, w = img.shape[:2]
h_new = int(w * fabs(sin(radians(angle))) + h * fabs(cos(radians(angle))))
w_new = int(h * fabs(sin(radians(angle))) + w * fabs(cos(radians(angle))))
matRotation = cv2.getRotationMatrix2D((w / 2, h / 2), angle, 1)
matRotation[0, 2] += (w_new - w) / 2
matRotation[1, 2] += (h_new - h) / 2
imgRotation = cv2.warpAffine(img, matRotation, (w_new, h_new), borderValue=(255, 255, 255))
return imgRotation
def do_rotate(im, angle, center=None, scale=1.0):
"""
图像旋转变换
:param im: 原始图像数据
:param angle: 旋转角度
:param center: 旋转中心，如果为None则以原图中心为旋转中心
:param scale: 缩放比例，默认为1
:return: 返回旋转后的图像
"""
h, w = im.shape[:2] # 获取图像高、宽
# 旋转中心默认为图像中心
if center is None:
center = (w / 2, h / 2)
# 计算旋转矩阵
M = cv2.getRotationMatrix2D(center, angle, scale)
# 使用openCV仿射变换实现函数旋转
rotated = cv2.warpAffine(im, M, (w, h))
return rotated # 返回旋转后的矩阵
def rotate_all():
dirs = os.listdir(data_root_path) # 列出所有子目录
for d in dirs:
dir_path = os.path.join(data_root_path, d) # 拼接路径
if not os.path.isdir(dir_path): # 不是目录
continue
sub_dir_path = os.path.join(dir_path, "Imgs") # 子目录下的Imgs目录
imgs = os.listdir(sub_dir_path) # 列出所有子目录下的原始样本
for img_file in imgs: # 遍历
img_full_path = os.path.join(sub_dir_path, img_file) # 拼接完整路径
print(img_full_path)
im = cv2.imread(img_full_path) # img_full_path
pos = img_file.find(".") # 返回.的位置
name = img_file[0:pos] # 取出名称部分
suffix = img_file[pos:] # 取出后缀名
# 旋转45/90/135/180/225/270/315度
for i in range(1, 8):
img_new = remote(im, 45 * i)
# 拼一个新的文件名，格式如：AIBJ-KG-00001_rotate_1.jpg
img_new_name = "%s_rotate_%d%s" % (name, i, suffix)
cv2.imwrite(os.path.join(sub_dir_path, img_new_name), img_new) # 将裁剪后的图片保存至新文件
print("save ok:", os.path.join(sub_dir_path, img_new_name))
if __name__ == "__main__":
# 图像旋转
rotate_all()
print("图像预处理结束")对数据进行处理,并生成数据集
# 人脸（水果）识别示例：数据预处理
import paddle.fluid as fluid
import os
import json
from global_var import *
name_data_list = {} # 记录每个人多少张训练图片、多少张测试图片
def get_file_lines(file_path, type): # 获取文件行数
with open(file_path) as f:
i = 0
for line in f.readlines():
line = line.strip().replace("\n", "")
tmp = line.split("\t")
if len(tmp) < 2:
continue
else:
if int(tmp[1]) == type:
i += 1
return i
def save_train_test_file(path, name):
if name not in name_data_list: # 未在字典中
img_list = []
img_list.append(path) # 将图片添加到列表
name_data_list[name] = img_list # 存入字典
else: # 已经在字典中
name_data_list[name].append(path) # 加入
# 获取所有类别保存的文件夹名称
dirs = os.listdir(data_root_path)
for d in dirs:
full_path = os.path.join(data_root_path, d) # 完整路径
if os.path.isdir(full_path): # 目录
full_path = os.path.join(full_path, "Imgs")
imgs = os.listdir(full_path)
for img in imgs:
# print(img + "," + d)
save_train_test_file(os.path.join(full_path, img), d)
else: # 文件
pass
# 清空数据文件
with open(test_file_path, "w") as f:
pass
with open(train_file_path, "w") as f:
pass
for name, img_list in name_data_list.items():
i = 0
num = len(img_list)
print("%s: %d张" % (name, num))
for img in img_list:
if i % 10 == 0: # 每10笔取一笔测试数据
with open(test_file_path, "a") as f:
line = "%s\t%d\n" % (img, name_dict[name])
# print(line)
f.write(line)
else: # 其它作为训练数据
with open(train_file_path, "a") as f:
line = "%s\t%d\n" % (img, name_dict[name])
# print(line)
f.write(line)
i += 1
print('生成数据列表完成！')进行训练
# 人脸(水果)识别示例：训练
import paddle
import paddle.fluid as fluid
import numpy
import sys
import os
from multiprocessing import cpu_count
import time
import matplotlib.pyplot as plt
from global_var import *
import logging
from paddle.fluid.param_attr import ParamAttr
from paddle.fluid.regularizer import L2Decay
train_img_size = 200 # 训练图像大小
label_dict = {} # 标签字典
BUF_SIZE = 10000
BATCH_SIZE = 16 # 批次大小
EPOCH_NUM = 20 # 迭代次数
learning_rate=0.0001 # 学习率
# 定义训练的mapper
def train_mapper(sample):
img, lable = sample
if not os.path.exists(img):
print(img, "文件不存在")
# 进行图片读取，由于数据集的像素和维度不同，需要进一步对图像进行变换
img = paddle.dataset.image.load_image(img)
# 对图像进行简单变换，对图像进行crop修剪操作，输出img的维度(3, 240, 240)
img = paddle.dataset.image.simple_transform(im=img,
resize_size=train_img_size, # 剪裁图片
crop_size=train_img_size,
is_color=True, # 彩色图像
is_train=False)
# 将img数组进行归一化处理，得到0~1之间的数值
img = img.flatten().astype("float32") / 255.0
return img, lable
# 对自定义的数据集创建训练集train和reader
def train_r(train_list, buffered_size=BUF_SIZE):
def reader():
with open(train_list, "r") as f: # 打开训练样本
lines = [line.strip() for line in f]
for line in lines:
img_path, lab = line.strip().split("\t")
if not os.path.exists(img_path): # 图片可能空白太多被移走
continue
# print(img_path, ":", int(lab))
yield img_path, int(lab)
return paddle.reader.xmap_readers(train_mapper, reader, cpu_count(), buffered_size)
def test_mapper(sample):
img, label = sample
img = paddle.dataset.image.load_image(img)
img = paddle.dataset.image.simple_transform(im=img,
resize_size=train_img_size,
crop_size=train_img_size,
is_color=True,
is_train=False)
img = img.flatten().astype("float32") / 255.0
return img, label
def test_r(test_list, buffered_size=BUF_SIZE):
def reader():
with open(test_list, "r") as f:
# 将train.list里面的标签和图片放到一个list列表中，中间用\t隔开
lines = [line.strip() for line in f]
for line in lines:
img_path, lab = line.strip().split("\t")
if not os.path.exists(img_path): # 图片可能空白太多被移走
print("图片不存在:", img_path)
continue
yield img_path, int(lab)
return paddle.reader.xmap_readers(test_mapper, reader, cpu_count(), buffered_size)
def init_log_config(): # 初始化日志相关配置
global logger
logger = logging.getLogger() # 创建日志对象
logger.setLevel(logging.INFO) # 设置日志级别
log_path = os.path.join(os.getcwd(), 'logs')
if not os.path.exists(log_path): # 创建日志路径
os.makedirs(log_path)
log_name = os.path.join(log_path, 'train.log') # 训练日志文件
fh = logging.FileHandler(log_name, mode='w') # 打开文件句柄
fh.setLevel(logging.DEBUG) # 设置级别
formatter = logging.Formatter("%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s")
fh.setFormatter(formatter)
logger.addHandler(fh)
############################ 程序开始 ################################
init_log_config() # 初始化日期工具
print("开始执行:", time.time())
trainer_reader = train_r(train_list=train_file_path)
train_batch_reader = paddle.batch(paddle.reader.shuffle(reader=trainer_reader, buf_size=BUF_SIZE), # buf_size=300
batch_size=BATCH_SIZE)
tester_reader = test_r(test_list=test_file_path)
test_batch_reader = paddle.batch(tester_reader, batch_size=BATCH_SIZE)
image = fluid.layers.data(name="image", shape=[3, train_img_size, train_img_size],
dtype="float32") # [3, 400, 400]表示三通道RGB图像
label = fluid.layers.data(name="label", shape=[1], dtype="int64")
print("image_shape:", image.shape)
# 搭建CNN网络
# 输入层 --> 卷积/池化/dropout --> 卷积/池化/dropout --> 卷积/池化/dropout --> 全连接 --> dropout --> 输出层
def convolution_neural_network(image, type_size):
# 第一个卷积-池化层
conv_pool_1 = fluid.nets.simple_img_conv_pool(input=image, # 输入image
filter_size=3, # 滤波器大小
num_filters=32, # filter数量，与输出通道相同
pool_size=2, # 池化层大小2*2
pool_stride=2, # 池化层步长
act="relu") # 激活函数
# Dropout主要作用是减少过拟合，随机让某些权重不更新
drop = fluid.layers.dropout(x=conv_pool_1, dropout_prob=0.5)
# 第二个卷积-池化层
conv_pool_2 = fluid.nets.simple_img_conv_pool(input=drop,
filter_size=3,
num_filters=64,
pool_size=2,
pool_stride=2,
act="relu")
drop = fluid.layers.dropout(x=conv_pool_2, dropout_prob=0.5)
# 第三个卷积-池化层
conv_pool_3 = fluid.nets.simple_img_conv_pool(input=drop,
filter_size=3,
num_filters=64,
pool_size=2,
pool_stride=2,
act="relu")
drop = fluid.layers.dropout(x=conv_pool_3, dropout_prob=0.5)
# 全连接层
fc = fluid.layers.fc(input=drop, size=512, act="relu")
# dropout层
drop = fluid.layers.dropout(x=fc, dropout_prob=0.5)
# 输出层
predict = fluid.layers.fc(input=drop, size=type_size, act="softmax")
return predict
# 搭建VGG网络
def vgg_bn_drop(image, type_size):
def conv_block(ipt, num_filter, groups, dropouts):
return fluid.nets.img_conv_group(input=ipt, # 具有[N，C，H，W]格式的输入图像
pool_size=2,
pool_stride=2,
conv_num_filter=[num_filter] * groups, # 过滤器个数
conv_filter_size=3, # 过滤器大小
conv_act='relu',
conv_with_batchnorm=True, # 表示在 Conv2d Layer 之后是否使用 BatchNorm
conv_batchnorm_drop_rate=dropouts, # 表示 BatchNorm 之后的 Dropout Layer 的丢弃概率
pool_type='max') # 最大池化
conv1 = conv_block(image, 64, 2, [0.0, 0])
conv2 = conv_block(conv1, 128, 2, [0.0, 0])
conv3 = conv_block(conv2, 256, 3, [0.0, 0.0, 0])
conv4 = conv_block(conv3, 512, 3, [0.0, 0.0, 0])
conv5 = conv_block(conv4, 512, 3, [0.0, 0.0, 0])
drop = fluid.layers.dropout(x=conv5, dropout_prob=0.5)
fc1 = fluid.layers.fc(input=drop, size=512, act=None)
bn = fluid.layers.batch_norm(input=fc1, act='relu')
drop2 = fluid.layers.dropout(x=bn, dropout_prob=0.0)
fc2 = fluid.layers.fc(input=drop2, size=512, act=None)
predict = fluid.layers.fc(input=fc2, size=type_size, act='softmax')
return predict
type_size = 6
# 获取分类器，用cnn或vgg网络进行分类type_size要和训练时的类别一致
predict = convolution_neural_network(image=image, type_size=type_size)
# predict = vgg_bn_drop(image=image, type_size=type_size)
# 获取损失函数和准确率
cost = fluid.layers.cross_entropy(input=predict, label=label)
# 计算cost中所有元素的平均值
avg_cost = fluid.layers.mean(cost)
# 计算准确率
accuracy = fluid.layers.accuracy(input=predict, label=label)
test_program = fluid.default_main_program().clone(for_test=True)
# 定义优化器
optimizer = fluid.optimizer.Adam(learning_rate)
opt = optimizer.minimize(avg_cost)
# 执行训练
# place = fluid.CPUPlace()
place = fluid.CUDAPlace(0)
exe = fluid.Executor(place)
exe.run(fluid.default_startup_program())
# 定义输入数据的维度, DataFeeder负责将reader返回的数据转成一种特殊结构，输入到Executor
feeder = fluid.DataFeeder(feed_list=[image, label], place=place)
# 训练5个Pass, 每个Pass训练结束后，使用验证集进行验证，并求出相应的损失值cost和准确度acc
model_save_dir = "./model/"
costs = [] # 损失值,可视化使用
accs = [] # 准确率,可视化使用
times = 0
batches = []
if os.path.exists(model_save_dir): # 先加载模型执行增量训练
fluid.io.load_persistables(exe, model_save_dir, fluid.default_main_program())
print("加载增量模型成功.")
print("开始训练......")
for pass_id in range(EPOCH_NUM):
train_cost = 0
for batch_id, data in enumerate(train_batch_reader()):
times += 1
train_cost, train_acc = exe.run(program=fluid.default_main_program(), # 运行主程序
feed=feeder.feed(data), # 喂入一个batch的数据
fetch_list=[avg_cost, accuracy]) # fetch均方误差和准确率
if batch_id % 50 == 0:
tmp_str = "Pass:%d, Step:%d, Cost:%.6f, Acc:%.6f" % (pass_id, batch_id, train_cost[0], train_acc[0])
# logger.info(tmp_str)
print(tmp_str)
accs.append(train_acc[0])
costs.append(train_cost[0])
batches.append(times)
# 开始测试
test_accs = []
test_costs = []
for batch_id, data in enumerate(test_batch_reader()):
test_cost, test_acc = exe.run(program=test_program,
feed=feeder.feed(data),
fetch_list=[avg_cost, accuracy])
test_accs.append(test_acc[0])
test_costs.append(test_cost[0])
test_cost = (sum(test_costs) / len(test_costs))
test_acc = (sum(test_accs) / len(test_accs))
tmp_str = "Test:%d, Cost:%.6f, ACC:%.6f" % (pass_id, test_cost, test_acc)
# logger.info(tmp_str)
print(tmp_str)
# 保存增量模型
if not os.path.exists(model_save_dir): # 如果存储模型的目录不存在，则创建
os.makedirs(model_save_dir)
fluid.io.save_persistables(exe, model_save_dir, fluid.default_main_program())
print("保存增量模型成功!")
# 保存固化模型
model_freeze_dir = "model_freeze/"
if not os.path.exists(model_freeze_dir):
os.makedirs(model_freeze_dir)
fluid.io.save_inference_model(dirname=model_freeze_dir,
feeded_var_names=["image"],
target_vars=[predict],
executor=exe)
print("保存固化模型成功!")
# 训练过程可视化
plt.figure("training", facecolor="lightgray")
plt.title("training", fontsize=24)
plt.xlabel("iter", fontsize=20)
plt.ylabel("cost/acc", fontsize=20)
plt.plot(batches, costs, color='red', label="Training Cost")
plt.plot(batches, accs, color='green', label="Training Acc")
plt.legend()
plt.grid()
plt.savefig("tarin.png")
plt.show()验证训练结果
# 人脸(水果)识别示例：预测
import paddle
import paddle.fluid as fluid
import numpy as np
import sys
import os
import matplotlib.pyplot as plt
from PIL import Image
from global_var import *
place = fluid.CPUPlace()
infer_exe = fluid.Executor(place)
inference_scope = fluid.core.Scope()
model_freeze_dir = "model_freeze/"
# 加载数据
def load_image(path):
img = paddle.dataset.image.load_and_transform(path, 200, 200, False).astype("float32")
img = img / 255.0
return img
infer_imgs = []
# 类别0
#test_img = "data/MagneticTile/MT_Blowhole/Imgs/exp6_num_4841.jpg"
# 类别1
# test_img = "data/MagneticTile/MT_Break/Imgs/exp2_num_271384.jpg"
# 类别2
# test_img = "data/MagneticTile/MT_Crack/Imgs/exp1_num_32128.jpg"
# 类别3
# test_img = "data/MagneticTile/MT_Fray/Imgs/exp1_num_20362.jpg"
# 类别4
# test_img = "data/MagneticTile/MT_Free/Imgs/exp3_num_344580.jpg"
test_img = "data/MagneticTile/MT_Free/Imgs/exp6_num_293912.jpg"
# 类别5
# test_img = "data/MagneticTile/MT_Uneven/Imgs/exp1_num_45007.jpg"
infer_imgs.append(load_image(test_img))
infer_imgs = np.array(infer_imgs)
print("infer_imgs.shape:", infer_imgs.shape)
with fluid.scope_guard(inference_scope):
[inference_program, feed_target_names, fetch_targets] = \
fluid.io.load_inference_model(model_freeze_dir, infer_exe)
# 开始预测
results = infer_exe.run(inference_program,
feed={feed_target_names[0]: infer_imgs},
fetch_list=fetch_targets)
print("results:", results)
result = results[0]
print(result.shape)
max_index = np.argmax(result)
for k, v in name_dict.items():
if max_index == v:
print("预测结果: 类别编号[%d], 名称[%s], 概率[%.4f]" % (max_index, k, result[0][max_index] * 100))
# 显示原图
img = Image.open(test_img)
plt.imshow(img)
plt.show()将模型部署在django上,可以方便访问
- tile_server.zip
- 在aistudio上运行paddle-1.8.4的模型运行在paddle-1.5.2报错
图像分类优化手段
样本优化
- 增大样本数量
- 数据增强
- 形态变化：翻转、平移、随机修剪、尺度变换、旋转
- 色彩变化：色彩抖动（错位的位移对图像产生的一种特殊效果）、图像白化（将图像本身归一化成Gaussian(0,1) 分布）
- 加入噪声：噪声扰动
参数优化
- 丢弃学习：按照一定比率丢弃神经元输出Ø 权重衰减：通过为模型损失函数添加惩罚项使得训练的模型参数较小
- 批量正则化：在网络的每一层输入之前增加归一化处理，使输入的均值为0，标准差为 1。目的是将数据限制在统一的分布下
- 变化学习率：学习率由固定调整为变化，例如由固定0.001调整为0.1, 0.001,0.0005
- 加深网络：加深网络可能提高准确率，也可能降低准确率，视具体情况而定
模型优化
- 更换更复杂、精度更高的网络模型。如由简单CNN更换为VGG、GooLeNet、ResNet
Alipay
Wechat
最后一次更新于2023-08-02 16:02
0 条评论