基于Keras实现卷积神经网络CNN对图像的二分类识别

公众号：尤而小屋
作者：Peter
编辑：Peter

大家好，我是Peter~

本文是深度学习实战案例连载更新的第一篇文章：基于Keras实现卷积神经网络CNN对图像的二分类识别，包含完整的建模代码。

数据获取

数据主要是两个zip压缩文件：train.zip（569+M）和test.zip（284+M），获取方式：

1、文末提供基于百度网盘数据获取方式；

2、如果你可以科学上网，也可以在kaggle官网下载开源数据集：www.kaggle.com/competition…

3、直接添加小编好友获取

下面的文章都是基于你已经获取压缩文件，并且解压到了本地，生成了两个文件目录：/train和/test

因为代码中存在对目录的相关操作，务必保证按照文章中的操作进行；否则，代码可能无法正常运行

python操作文件目录

在这里先简单总结下python中操作文件和目录的3个包：os、shutil和pathlib

os包

1-处理文件和目录
os.access(path, mode) - 检查路径的访问权限
os.chdir(path) - 改变当前工作目录
os.getcwd() - 获取当前工作目录
os.listdir(path) - 列出指定路径下的文件和目录
os.mkdir(path) - 创建文件夹
os.remove(path) - 删除文件
os.rmdir(path) - 删除空文件夹
os.rename(src, dst) - 重命名文件或目录
os.stat(path) - 获取文件或目录信息
os.walk(top) - 遍历top目录下所有子目录和文件

2-执行系统命令
os.system(command) - 执行系统命令,直接返回命令执行状态
os.popen(command).read() - 执行系统命令,返回命令输出结果
os.getcwd() - 获取当前进程工作目录
os.chdir(path) - 改变当前进程工作目录
os.getpid() - 获取当前进程id
os.kill(pid, sig) - 向进程发送signal信号

3-获取环境变量
os.environ - 获取系统环境变量
os.getenv(key) - 获取指定环境变量的值
os.putenv(key, value) - 设置环境变量

4-其他功能
os.path - 用于路径处理的模块
os.linesep - 系统行分隔符
os.name - 当前系统名称
os.urandom(n) - 生成n字节长度的随机字节序列

shutil包

python中的shutil库提供了高级的文件、文件夹、压缩包处理功能,常用的主要有以下几个方面:

1. 复制文件/文件夹

- shutil.copy(src, dst) - 复制文件
- shutil.copy2(src, dst) - 复制文件和元数据
- shutil.copystat(src, dst) - 仅复制元数据
- shutil.copytree(src, dst) - 递归复制文件夹

2. 移动文件/文件夹

- shutil.move(src, dst) - 移动文件/文件夹

3. 删除文件/文件夹

- shutil.rmtree(path) - 递归删除文件夹
- shutil.unlink(path) - 删除文件

4. 打包压缩

- shutil.make_archive(base_name, format,...) - 创建压缩包
- shutil.unpack_archive(filename,...) - 解压压缩包

5. 其他

- shutil.disk_usage(path) - 返回路径占用的磁盘空间信息
- shutil.chown(path, user, group) - 修改文件权限
- shutil.which(cmd) - 返回命令路径

shutil库建立在os模块之上,提供了更易用的高级接口,可以通过import shutil来使用。

pathlib包

pathlib模块提供了面向对象的路径管理方法，可以更简单的处理文件系统路径。主要功能包括:

1. 创建Path对象
# 从一个路径字符串创建Path对象
from pathlib import Path
p = Path('example.txt')

2. 访问路径

- p.exists() - 检查路径是否存在
- p.is_file() - 检查路径是否是一个文件
- p.is_dir() - 检查路径是否是一个目录
- p.name - 文件名
- p.stem - 不包含扩展名的文件名
- p.suffix - 文件扩展名
- p.parent - 父目录路径

3. 操作文件系统

- p.mkdir() - 创建目录
- p.rmdir() - 删除目录
- p.unlink() - 删除文件
- p.rename() - 重命名路径
- p.open() - 打开文件并返回一个文件对象

4. 路径运算

可以用/拼接子路径,也支持os.path的运算:

- p / 'subpath' - 拼接子路径
- p.resolve() - 返回绝对路径
- p.is_absolute() - 检查是否是绝对路径

5.  glob模式匹配

- p.glob('*.py') - 匹配该路径下的py文件
- p.rglob('*.py') - 递归匹配所有子目录中的py文件

下面介绍完整的过程：

构建数据

由于个人PC配置有限，在这里使用部分的数据进行建模。从原始数据集图像中复制部分图像到指定目录下（构成少量数据集）。

在这里创建3个数据集：

• 训练集train
• 验证集validation
• 测试集test

In [1]:

import os
import shutil
import pathlib 

每次重新运行的时候，都要保证cats_dogs_small文件夹是不存在的

In [2]:

# 原始解压目录，已经存在
original_dir = pathlib.Path("train")  

# 保存小数据集的目录（直接生成，未提前创建）
new_base_dir = pathlib.Path("cats_dogs_small")  

In [3]:

new_base_dir

Out[3]:

WindowsPath('cats_dogs_small')

In [4]:

def make_subset(subset_name, start_index, end_index):
    """
    作用：从索引start_index到end_index，复制图像到子目录new_base_dir/{subset_name}/cat(dog)
    subset_name可以是train、validation或者test
    """
    
    for category in ["cat", "dog"]:  # cat或者dog同时遍历创建数据
        dir = new_base_dir / subset_name / category  #  拼接子目录完整路径
        os.makedirs(dir)  # 创建目录
        
        # 列表推导式生成图片名称
        fnames = [f"{category}.{i}.jpg" for i in range(start_index, end_index)]  
        
        # 对图片名称的循环复制
        for fname in fnames:
             # copyfile：从src路径复制到dst路径
            shutil.copyfile(src=original_dir / fname, 
                           dst=dir / fname)

In [5]:

分别创建训练集（索引号从0到999，不包含1000）、验证集（从1000到1499，不包含1500）和测试集（从1500到2499，不包含2500）

make_subset("train", start_index=0, end_index=1000)

In [6]:

make_subset("validation", start_index=1000, end_index=1500)

In [7]:

make_subset("test", start_index=1500, end_index=2500)

搭建卷积神经网络CNN

基于keras搭建卷积神经网络：卷积层和最大池化层的累加，再加上最后的展平层和密集连接输出层。

In [8]:

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

inputs = keras.Input(shape=(180,180,3))

# 像素尺寸缩放
x = layers.Rescaling(1. / 255)(inputs)

x = layers.Conv2D(filters=32, kernel_size=3, activation="relu")(x)  # 卷积层
x = layers.MaxPooling2D(pool_size=2)(x)  # 最大池化层
x = layers.Conv2D(filters=64, kernel_size=3, activation="relu")(x)
x = layers.MaxPooling2D(pool_size=2)(x)
x = layers.Conv2D(filters=128, kernel_size=3, activation="relu")(x)
x = layers.MaxPooling2D(pool_size=2)(x)
x = layers.Conv2D(filters=256, kernel_size=3, activation="relu")(x)
x = layers.MaxPooling2D(pool_size=2)(x)
x = layers.Conv2D(filters=256, kernel_size=3, activation="relu")(x)

x = layers.Flatten()(x)  # 展平层

outputs = layers.Dense(1, activation="sigmoid")(x)  # 密集连接层
model = keras.Model(inputs=inputs, outputs=outputs)

查看模型概要：

In [9]:

model.summary()
Model: "model"
_________________________________________________________________
 Layer (type)                Output Shape              Param #   
=================================================================
 input_1 (InputLayer)        [(None, 180, 180, 3)]     0         
                                                                 
 rescaling (Rescaling)       (None, 180, 180, 3)       0         
                                                                 
 conv2d (Conv2D)             (None, 178, 178, 32)      896       
                                                                 
 max_pooling2d (MaxPooling2D  (None, 89, 89, 32)       0         
 )                                                               
                                                                 
 conv2d_1 (Conv2D)           (None, 87, 87, 64)        18496     
                                                                 
 max_pooling2d_1 (MaxPooling  (None, 43, 43, 64)       0         
 2D)                                                             
                                                                 
 conv2d_2 (Conv2D)           (None, 41, 41, 128)       73856     
                                                                 
 max_pooling2d_2 (MaxPooling  (None, 20, 20, 128)      0         
 2D)                                                             
                                                                 
 conv2d_3 (Conv2D)           (None, 18, 18, 256)       295168    
                                                                 
 max_pooling2d_3 (MaxPooling  (None, 9, 9, 256)        0         
 2D)                                                             
                                                                 
 conv2d_4 (Conv2D)           (None, 7, 7, 256)         590080    
                                                                 
 flatten (Flatten)           (None, 12544)             0         
                                                                 
 dense (Dense)               (None, 1)                 12545     
                                                                 
=================================================================
Total params: 991,041
Trainable params: 991,041
Non-trainable params: 0
_________________________________________________________________

模型的编译，设置损失、优化器和评估指标：

In [10]:

# 模型编译compile

model.compile(loss="binary_crossentropy",  # 二分类使用binary_crossentropy
              optimizer="rmsprop", 
              metrics=["accuracy"])

数据预处理image_dataset_from_directory()

keras肯定是不能直接处理图像数据。因此数据在输入模型之前，应该将图像数据格式化为经过预处理的浮点数张量。将图片JPEG文件转成浮点数张量的步骤：

• 读取JPEG文件，并解码为RGB像素网格
• 将像素网格转为浮点数张量，并将张量的大小调节相同
• 将数据打包成批量

读取图像

Keras包含函数image_dataset_from_directory()，通过建立数据管道，将图片文件迅速转成张量批量

In [11]:

from tensorflow.keras.utils import image_dataset_from_directory

train_dataset = image_dataset_from_directory(
    new_base_dir / "train",  # 目录
    image_size=(180, 180),  # 图像大小
    batch_size=32)  # 批量大小

validation_dataset = image_dataset_from_directory(
    new_base_dir / "validation",
    image_size=(180, 180),
    batch_size=32)

test_dataset = image_dataset_from_directory(
    new_base_dir / "test",
    image_size=(180, 180),
    batch_size=32)
Found 2000 files belonging to 2 classes.
Found 1000 files belonging to 2 classes.
Found 2000 files belonging to 2 classes.

上面列出了每个目录下的文件数量和类别数目。

理解TensorFlow DataSet对象

TensorFlow提供了tf.data 这个API，用于为机器学习模型创建管道，最重要的类是tf.data.Dataset。

该对象类是一个迭代器，可以在for循环中使用，返回输入数据和标签组成的批量。可以将对象直接传入Keras的fit方法中。

In [12]:

# Dataset类还拥有一个用于修改数据集的函数式API


import numpy as np
import tensorflow as tf

random_numbers = np.random.normal(size=(1000,16))
random_numbers

Out[12]:

array([[-1.97578218, -0.90409304, -0.09623576, ...,  2.41728279,         0.91704091,  1.6140268 ],
       [-0.93297431,  0.22050653,  0.26043918, ..., -0.15646056,        -2.90198359, -2.43836605],
       [ 1.79131853, -0.87672883,  0.27288246, ...,  1.58597872,        -0.64024683,  0.52037157],
       ...,
       [-0.01100054, -0.69301064, -0.90628903, ...,  1.07197402,        -1.04802286, -0.25163522],
       [ 0.23100176, -0.59347372,  1.40955734, ...,  1.32233625,         0.0113596 , -1.25762504],
       [-1.32839193,  0.39171769, -0.2087147 , ...,  0.43798809,        -0.22259379,  0.98210044]])

from_tensor_slices该类方法可以利用numpy数组或者数组的元组或字典来创建一个Dataset对象：

In [13]:

dataset = tf.data.Dataset.from_tensor_slices(random_numbers)

dataset

Out[13]:

<_TensorSliceDataset element_spec=TensorSpec(shape=(16,), dtype=tf.float64, name=None)>

根据数据集生成单个样本：

In [14]:

for i, element in enumerate(dataset):
    print(element.shape)
    
    if i >= 2:
        break
(16,)
(16,)
(16,)

使用.batch方法来批量生成数据：

In [15]:

batched_dataset = dataset.batch(32)  # 调用batach方法批量生成数据

for i, element in enumerate(batched_dataset):
    print(element.shape)
    if i > 2:
        break
(32, 16)
(32, 16)
(32, 16)
(32, 16)

该对象还有其他方法：

• .shuffle(buffer_size)：打乱缓冲区元素。
• .prefetch(buffer_size)：将缓冲区元素预取到GPU内存中，以提高设备利用率。
• .map(callable)：对数据集的每个元素进行某项变换（函数callable的输入是数据集生成的单个元素）。

In [16]:

# map方法的使用

map_dataset = dataset.map(lambda x:tf.reshape(x,(4,4)))

for i, element in enumerate(map_dataset):
    print(element.shape)
    if i > 2:
        break
(4, 4)
(4, 4)
(4, 4)
(4, 4)

Dataset对象的输出

In [17]:

# 一个Dataset对象的输出

for data_batch, labels_batch in train_dataset:
    print("data_batch.shape: ", data_batch.shape)
    print("labels_batch.shape", labels_batch.shape)
    break
data_batch.shape:  (32, 180, 180, 3)
labels_batch.shape (32,)

利用Dataset对象训练模型（未正则化）

In [18]:

callbacks = [
    keras.callbacks.ModelCheckpoint(
        filepath="convnet_from_scratch.keras",  # 保存位置
        save_best_only=True,  # 只有当val_loss指标的当前值低于训练过程之前的所有值时，回调函数才会保存一个新文件
        monitor="val_loss"
    )
]

history = model.fit(
    train_dataset, 
    epochs=30,
    validation_data=validation_dataset,
    callbacks=callbacks
)

Epoch 1/30
63/63 [==============================] - 40s 617ms/step - loss: 0.6948 - accuracy: 0.5035 - val_loss: 0.6943 - val_accuracy: 0.5000
Epoch 2/30
63/63 [==============================] - 39s 615ms/step - loss: 0.6943 - accuracy: 0.5095 - val_loss: 0.6921 - val_accuracy: 0.5000
Epoch 3/30
63/63 [==============================] - 38s 600ms/step - loss: 0.6865 - accuracy: 0.5755 - val_loss: 0.6728 - val_accuracy: 0.5270
Epoch 4/30
63/63 [==============================] - 37s 594ms/step - loss: 0.6363 - accuracy: 0.6430 - val_loss: 0.6469 - val_accuracy: 0.6120
Epoch 5/30
63/63 [==============================] - 38s 609ms/step - loss: 0.6001 - accuracy: 0.6665 - val_loss: 0.9687 - val_accuracy: 0.5550
Epoch 6/30
63/63 [==============================] - 39s 621ms/step - loss: 0.5801 - accuracy: 0.7055 - val_loss: 0.6030 - val_accuracy: 0.6790
......
Epoch 29/30
63/63 [==============================] - 38s 605ms/step - loss: 0.0244 - accuracy: 0.9900 - val_loss: 2.0662 - val_accuracy: 0.7230
Epoch 30/30
63/63 [==============================] - 38s 609ms/step - loss: 0.0550 - accuracy: 0.9820 - val_loss: 2.2703 - val_accuracy: 0.7120

精度和损失可视化

对训练过程中精度和损失的可视化：

In [19]:

import matplotlib.pyplot as plt
%matplotlib inline

acc = history.history["accuracy"]
val_acc = history.history["val_accuracy"]
loss = history.history["loss"]
val_loss = history.history["val_loss"]

epochs = range(1, len(acc) + 1)

plt.figure()
plt.plot(epochs, acc, "bo", label="Training acc")
plt.plot(epochs, val_acc, "b", label="Validation acc")
plt.title("Training and validation accuracy")
plt.legend()

plt.figure()
plt.plot(epochs, loss, "bo", label="Training loss")
plt.plot(epochs, val_loss, "b", label="Validation loss")
plt.title("Training and validation loss")
plt.legend()
plt.show()

测试集上评估模型

将前面使用回调函数时保存的最佳模型直接导进来，然后用在测试集上进行评估：

In [20]:

test_model = keras.models.load_model("convnet_from_scratch.keras") 
test_loss, test_acc = test_model.evaluate(test_dataset)

print(f"Test accuracy: {test_acc:.3f}")
63/63 [==============================] - 7s 114ms/step - loss: 0.5807 - accuracy: 0.7395
Test accuracy: 0.739

数据增强

数据增强的目标是，模型在训练时不会操作两个完全相同图片，有助于观察到数据的更多内容，从而具有更强的泛化能力。

在模型的一开始添加数据增强层。

添加增强层

In [21]:

# 3个增强层
data_augment = keras.Sequential(  
    [layers.RandomFlip("horizontal"),  # 水平翻转应用于随机抽取的50%图像
     layers.RandomRotation(0.1),  # 将图像在[-10%,10%]的范围内随机旋转或者说[-36°, +36°]
     layers.RandomZoom(0.2)  # 放大或者缩小图像，在[-20%,+20%]范围内随机取值
    ]
)

显示增强后的图像

In [22]:

plt.figure(figsize=(10,10))

for images, _ in train_dataset.take(1):
    for i in range(9):
        augmented_images = data_augment(images)  # 将数据增强代码块用于图像批量
        ax = plt.subplot(3,3,i+1)
        plt.imshow(augmented_images[2].numpy().astype("uint8")) # 9次迭代：对同一个图像的增强
        plt.axis("off")

数据增强 +dropout正则化

基于数据增强和dropout正则化构建卷积神经网络：

In [23]:

inputs = keras.Input(shape=(180,180,3))

x = data_augment(inputs)  # 添加数据增强
x = layers.Rescaling(1./255)(x)
x = layers.Conv2D(filters=32, kernel_size=3, activation="relu")(x)
x = layers.MaxPooling2D(pool_size=2)(x)
x = layers.Conv2D(filters=64, kernel_size=3, activation="relu")(x)
x = layers.MaxPooling2D(pool_size=2)(x)
x = layers.Conv2D(filters=128, kernel_size=3, activation="relu")(x)
x = layers.MaxPooling2D(pool_size=2)(x)
x = layers.Conv2D(filters=256, kernel_size=3, activation="relu")(x)
x = layers.MaxPooling2D(pool_size=2)(x)
x = layers.Conv2D(filters=256, kernel_size=3, activation="relu")(x)
x = layers.Flatten()(x)

x = layers.Dropout(0.5)(x)  # drop比例为0.5

outputs = layers.Dense(1, activation="sigmoid")(x)

In [24]:

model = keras.Model(inputs=inputs, outputs=outputs)
model.compile(loss="binary_crossentropy",
             optimizer="rmsprop",
             metrics=["accuracy"]
             )

基于正则化的卷积神经网络

In [25]:

callbacks = [
    keras.callbacks.ModelCheckpoint(
    filepath="convnet_from_scratch_with_augmentation.keras",
    save_best_only=True,
    monitor="val_loss")
]

history = model.fit(train_dataset,  # 此时的模型model已经是基于数据增强的
                   epochs=100,
                   validation_data=validation_dataset,
                   callbacks=callbacks)

Epoch 1/100
63/63 [==============================] - 40s 626ms/step - loss: 0.7104 - accuracy: 0.4965 - val_loss: 0.6924 - val_accuracy: 0.5000
Epoch 2/100
63/63 [==============================] - 40s 633ms/step - loss: 0.6934 - accuracy: 0.5195 - val_loss: 0.6902 - val_accuracy: 0.5810
Epoch 3/100
63/63 [==============================] - 40s 631ms/step - loss: 0.6938 - accuracy: 0.5285 - val_loss: 0.6742 - val_accuracy: 0.5780
Epoch 4/100
63/63 [==============================] - 39s 626ms/step - loss: 0.6606 - accuracy: 0.6110 - val_loss: 0.6832 - val_accuracy: 0.5270
Epoch 5/100
63/63 [==============================] - 40s 629ms/step - loss: 0.6434 - accuracy: 0.6250 - val_loss: 0.6279 - val_accuracy: 0.6250
......省略
Epoch 98/100
63/63 [==============================] - 39s 624ms/step - loss: 0.1888 - accuracy: 0.9365 - val_loss: 1.0649 - val_accuracy: 0.8340
Epoch 99/100
63/63 [==============================] - 39s 617ms/step - loss: 0.1616 - accuracy: 0.9480 - val_loss: 0.5868 - val_accuracy: 0.8350
Epoch 100/100
63/63 [==============================] - 40s 642ms/step - loss: 0.1545 - accuracy: 0.9470 - val_loss: 0.8994 - val_accuracy: 0.8220

模型评估

In [26]:

test_model = keras.models.load_model("convnet_from_scratch_with_augmentation.keras")
test_loss, test_acc = test_model.evaluate(test_dataset)

print(f"Test accuracy: {test_acc:.3f}")
63/63 [==============================] - 7s 112ms/step - loss: 0.5057 - accuracy: 0.7990
Test accuracy: 0.799

精度和损失可视化

In [27]:

import matplotlib.pyplot as plt
%matplotlib inline

acc = history.history["accuracy"]
val_acc = history.history["val_accuracy"]
loss = history.history["loss"]
val_loss = history.history["val_loss"]

epochs = range(1, len(acc) + 1)

plt.figure()
plt.plot(epochs, acc, "bo", label="Training acc")
plt.plot(epochs, val_acc, "b", label="Validation acc")
plt.title("Training and validation accuracy")
plt.legend()

plt.figure()
plt.plot(epochs, loss, "bo", label="Training loss")
plt.plot(epochs, val_loss, "b", label="Validation loss")
plt.title("Training and validation loss")
plt.legend()
plt.show()