反卷积神经网络

import os

os.environ["KERAS_BACKEND"] = "tensorflow"

import tensorflow as tf
import keras
import matplotlib.pyplot as plt

# 设置种子以保证可重现性。
tf.random.set_seed(42)

class Involution(keras.layers.Layer):
    def __init__(
        self, channel, group_number, kernel_size, stride, reduction_ratio, name
    ):
        super().__init__(name=name)

        # 初始化参数。
        self.channel = channel
        self.group_number = group_number
        self.kernel_size = kernel_size
        self.stride = stride
        self.reduction_ratio = reduction_ratio

    def build(self, input_shape):
        # 获取输入的形状。
        (_, height, width, num_channels) = input_shape

        # 根据步幅缩放高度和宽度。
        height = height // self.stride
        width = width // self.stride

        # 定义一个层，该层对输入张量进行平均池化
        # 如果步幅大于1。
        self.stride_layer = (
            keras.layers.AveragePooling2D(
                pool_size=self.stride, strides=self.stride, padding="same"
            )
            if self.stride > 1
            else tf.identity
        )
        # 定义核生成层。
        self.kernel_gen = keras.Sequential(
            [
                keras.layers.Conv2D(
                    filters=self.channel // self.reduction_ratio, kernel_size=1
                ),
                keras.layers.BatchNormalization(),
                keras.layers.ReLU(),
                keras.layers.Conv2D(
                    filters=self.kernel_size * self.kernel_size * self.group_number,
                    kernel_size=1,
                ),
            ]
        )
        # 定义重塑层
        self.kernel_reshape = keras.layers.Reshape(
            target_shape=(
                height,
                width,
                self.kernel_size * self.kernel_size,
                1,
                self.group_number,
            )
        )
        self.input_patches_reshape = keras.layers.Reshape(
            target_shape=(
                height,
                width,
                self.kernel_size * self.kernel_size,
                num_channels // self.group_number,
                self.group_number,
            )
        )
        self.output_reshape = keras.layers.Reshape(
            target_shape=(height, width, num_channels)
        )

    def call(self, x):
        # 根据输入张量生成核。
        # B, H, W, K*K*G
        kernel_input = self.stride_layer(x)
        kernel = self.kernel_gen(kernel_input)

        # 重塑核
        # B, H, W, K*K, 1, G
        kernel = self.kernel_reshape(kernel)

        # 提取输入补丁。
        # B, H, W, K*K*C
        input_patches = tf.image.extract_patches(
            images=x,
            sizes=[1, self.kernel_size, self.kernel_size, 1],
            strides=[1, self.stride, self.stride, 1],
            rates=[1, 1, 1, 1],
            padding="SAME",
        )

        # 重塑输入补丁以与后续操作对齐。
        # B, H, W, K*K, C//G, G
        input_patches = self.input_patches_reshape(input_patches)

        # 计算核和补丁的乘加操作。
        # B, H, W, K*K, C//G, G
        output = tf.multiply(kernel, input_patches)
        # B, H, W, C//G, G
        output = tf.reduce_sum(output, axis=3)

        # 重塑输出核。
        # B, H, W, C
        output = self.output_reshape(output)

        # 返回输出张量和核。
        return output, kernel

# 定义输入张量。
input_tensor = tf.random.normal((32, 256, 256, 3))

# 计算步幅为1的反卷积。
output_tensor, _ = Involution(
    channel=3, group_number=1, kernel_size=5, stride=1, reduction_ratio=1, name="inv_1"
)(input_tensor)
print(f"步幅为1的输出形状: {output_tensor.shape}")

# 计算步幅为2的反卷积。
output_tensor, _ = Involution(
    channel=3, group_number=1, kernel_size=5, stride=2, reduction_ratio=1, name="inv_2"
)(input_tensor)
print(f"步幅为2的输出形状: {output_tensor.shape}")

# 计算步幅为1，通道数为16，降维比为2的反卷积。
output_tensor, _ = Involution(
    channel=16, group_number=1, kernel_size=5, stride=1, reduction_ratio=2, name="inv_3"
)(input_tensor)
print(
    "通道数为16，降维比为2的输出形状: {}".format(output_tensor.shape)
)

# 加载CIFAR10数据集。
print("加载CIFAR10数据集...")
(
    (train_images, train_labels),
    (
        test_images,
        test_labels,
    ),
) = keras.datasets.cifar10.load_data()

# 将像素值标准化到0和1之间。
(train_images, test_images) = (train_images / 255.0, test_images / 255.0)

# 打乱并分批数据集。
train_ds = (
    tf.data.Dataset.from_tensor_slices((train_images, train_labels))
    .shuffle(256)
    .batch(256)
)
test_ds = tf.data.Dataset.from_tensor_slices((test_images, test_labels)).batch(256)

class_names = [
    "飞机",
    "汽车",
    "鸟",
    "猫",
    "鹿",
    "狗",
    "青蛙",
    "马",
    "船",
    "卡车",
]

plt.figure(figsize=(10, 10))
for i in range(25):
    plt.subplot(5, 5, i + 1)
    plt.xticks([])
    plt.yticks([])
    plt.grid(False)
    plt.imshow(train_images[i])
    plt.xlabel(class_names[train_labels[i][0]])
plt.show()

# 构建卷积模型。
print("构建卷积模型...")
conv_model = keras.Sequential(
    [
        keras.layers.Conv2D(32, (3, 3), input_shape=(32, 32, 3), padding="same"),
        keras.layers.ReLU(name="relu1"),
        keras.layers.MaxPooling2D((2, 2)),
        keras.layers.Conv2D(64, (3, 3), padding="same"),
        keras.layers.ReLU(name="relu2"),
        keras.layers.MaxPooling2D((2, 2)),
        keras.layers.Conv2D(64, (3, 3), padding="same"),
        keras.layers.ReLU(name="relu3"),
        keras.layers.Flatten(),
        keras.layers.Dense(64, activation="relu"),
        keras.layers.Dense(10),
    ]
)

# 使用必要的损失函数和优化器编译模型。
print("编译卷积模型...")
conv_model.compile(
    optimizer="adam",
    loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
    metrics=["accuracy"],
)

# 训练模型。
print("卷积模型训练中...")
conv_hist = conv_model.fit(train_ds, epochs=20, validation_data=test_ds)

# 构建反卷积模型。
print("building the involution model...")

inputs = keras.Input(shape=(32, 32, 3))
x, _ = Involution(
    channel=3, group_number=1, kernel_size=3, stride=1, reduction_ratio=2, name="inv_1"
)(inputs)
x = keras.layers.ReLU()(x)
x = keras.layers.MaxPooling2D((2, 2))(x)
x, _ = Involution(
    channel=3, group_number=1, kernel_size=3, stride=1, reduction_ratio=2, name="inv_2"
)(x)
x = keras.layers.ReLU()(x)
x = keras.layers.MaxPooling2D((2, 2))(x)
x, _ = Involution(
    channel=3, group_number=1, kernel_size=3, stride=1, reduction_ratio=2, name="inv_3"
)(x)
x = keras.layers.ReLU()(x)
x = keras.layers.Flatten()(x)
x = keras.layers.Dense(64, activation="relu")(x)
outputs = keras.layers.Dense(10)(x)

inv_model = keras.Model(inputs=[inputs], outputs=[outputs], name="inv_model")

# 使用必要的损失函数和优化器编译模型。
print("compiling the involution model...")
inv_model.compile(
    optimizer="adam",
    loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
    metrics=["accuracy"],
)

# 训练模型
print("inv model training...")
inv_hist = inv_model.fit(train_ds, epochs=20, validation_data=test_ds)

conv_model.summary()

inv_model.summary()

plt.figure(figsize=(20, 5))

plt.subplot(1, 2, 1)
plt.title("卷积损失")
plt.plot(conv_hist.history["loss"], label="损失")
plt.plot(conv_hist.history["val_loss"], label="验证损失")
plt.legend()

plt.subplot(1, 2, 2)
plt.title("反卷积损失")
plt.plot(inv_hist.history["loss"], label="损失")
plt.plot(inv_hist.history["val_loss"], label="验证损失")
plt.legend()

plt.show()

plt.figure(figsize=(20, 5))

plt.subplot(1, 2, 1)
plt.title("卷积准确率")
plt.plot(conv_hist.history["accuracy"], label="准确率")
plt.plot(conv_hist.history["val_accuracy"], label="验证准确率")
plt.legend()

plt.subplot(1, 2, 2)
plt.title("反卷积准确率")
plt.plot(inv_hist.history["accuracy"], label="准确率")
plt.plot(inv_hist.history["val_accuracy"], label="验证准确率")
plt.legend()

plt.show()

layer_names = ["inv_1", "inv_2", "inv_3"]
outputs = [inv_model.get_layer(name).output[1] for name in layer_names]
vis_model = keras.Model(inv_model.input, outputs)

fig, axes = plt.subplots(nrows=10, ncols=4, figsize=(10, 30))

for ax, test_image in zip(axes, test_images[:10]):
    (inv1_kernel, inv2_kernel, inv3_kernel) = vis_model.predict(test_image[None, ...])
    inv1_kernel = tf.reduce_sum(inv1_kernel, axis=[-1, -2, -3])
    inv2_kernel = tf.reduce_sum(inv2_kernel, axis=[-1, -2, -3])
    inv3_kernel = tf.reduce_sum(inv3_kernel, axis=[-1, -2, -3])

    ax[0].imshow(keras.utils.array_to_img(test_image))
    ax[0].set_title("输入图像")

    ax[1].imshow(keras.utils.array_to_img(inv1_kernel[0, ..., None]))
    ax[1].set_title("反卷积核 1")

    ax[2].imshow(keras.utils.array_to_img(inv2_kernel[0, ..., None]))
    ax[2].set_title("反卷积核 2")

    ax[3].imshow(keras.utils.array_to_img(inv3_kernel[0, ..., None]))
    ax[3].set_title("反卷积核 3")

 1/1 ━━━━━━━━━━━━━━━━━━━━ 1s 503ms/步
 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 11ms/步
 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 11ms/步
 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 9ms/步
 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 11ms/步
 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 9ms/步
 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 9ms/步
 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 9ms/步
 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 10ms/步
 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 9ms/步