目录
MindSpore 环境配置、MNIST 数据集下载及处理展开。
MindSpore 环境配置、MNIST 数据集下载及处理展开。
首先，通过命令行操作安装特定版本的 MindSpore 库，并查看其版本。接着，从指定 URL 下载 MNIST 数据集，然后定义数据处理函数 data_load，对数据集进行随机打乱、数据增强、指定输出列、批量操作等处理。最后，对训练数据集调用该函数，并获取处理后的数据集迭代次数并打印输出。整体涵盖了环境准备、数据获取与处理等关键步骤，为后续基于 MNIST 数据集的模型训练奠定基础。
代码如下：
%%capture captured_output
# 这行代码用于捕获后续单元格的输出，防止其直接显示在输出区域。
# 实验环境已经预装了 mindspore==2.2.14，如需更换 mindspore 版本，可更改下面 mindspore 的版本号
!pip uninstall mindspore -y
!pip install -i https://pypi.mirrors.ustc.edu.cn/simple mindspore==2.2.14
# 先卸载已安装的 MindSpore 库，然后通过指定的国内镜像源安装特定版本（2.2.14）的 MindSpore 库。
# 查看当前 mindspore 版本
!pip show mindspore
# 查看当前安装的 MindSpore 版本信息。
# 数据下载
from download import download
url = "https://mindspore-website.obs.cn-north-4.myhuaweicloud.com/notebook/datasets/MNIST_Data.zip"
download(url, ".", kind="zip", replace=True)
# 从给定的 URL 下载 MNIST 数据集的 ZIP 文件到当前目录，如果已存在则替换。
import numpy as np
import mindspore.dataset as ds
batch_size = 64
latent_size = 100 # 隐码的长度
# 导入 NumPy 库和 MindSpore 的数据集模块，并设置批量大小和隐码长度。
train_dataset = ds.MnistDataset(dataset_dir='./MNIST_Data/train')
test_dataset = ds.MnistDataset(dataset_dir='./MNIST_Data/test')
# 从指定目录加载 MNIST 训练集和测试集的数据。
def data_load(dataset):
dataset1 = ds.GeneratorDataset(dataset, ["image", "label"], shuffle=True, python_multiprocessing=False,num_samples=10000)
# 基于输入的数据集创建一个新的 GeneratorDataset，指定包含的列名为"image"和"label"，随机打乱数据，并采样 10000 个样本。
# 数据增强
mnist_ds = dataset1.map(
operations=lambda x: (x.astype("float32"), np.random.normal(size=latent_size).astype("float32")),
output_columns=["image", "latent_code"])
mnist_ds = mnist_ds.project(["image", "latent_code"])
# 对数据进行映射操作，将数据类型转换为 float32，并生成指定大小的随机隐码，同时指定输出的列名为"image"和"latent_code"。
# 批量操作
mnist_ds = mnist_ds.batch(batch_size, True)
# 将处理后的数据按照指定的批量大小进行分组。
return mnist_ds
# 定义了一个名为 data_load 的函数，用于对输入的数据集进行一系列处理和操作。
mnist_ds = data_load(train_dataset)
# 调用 data_load 函数对训练数据集进行处理，并将结果存储在 mnist_ds 变量中。
iter_size = mnist_ds.get_dataset_size()
print('Iter size: %d' % iter_size)
# 获取处理后的训练数据集的迭代次数，并将其打印输出。
数据集可视化
首先，导入了 matplotlib 库的 pyplot 模块，并将其重命名为 plt ，从而方便后续调用该模块中的绘图函数。接下来，从 mnist_ds 这个数据源获取一个迭代器，借助 next 函数获取下一个元素，把它存放在 data_iter 变量中，同时明确指定输出为 numpy 格式的数据。接着，创建了一个新的图形，将其大小设定为 3x3 。之后，定义了列数和行数，二者均为 5 。在循环部分：从 data_iter 里获取图像数据；运用 figure.add_subplot 函数于图形中添加子图；通过 plt.axis("off") 来关闭坐标轴的显示；使用 plt.imshow 函数来展示图像，其中 image.squeeze 或许是对图像进行了某种维度的压缩操作，而 cmap="gray" 意味着采用灰度颜色映射。最后，展示绘制完成的图形。
代码如下：
import matplotlib.pyplot as plt
data_iter = next(mnist_ds.create_dict_iterator(output_numpy=True))
figure = plt.figure(figsize=(3, 3))
cols, rows = 5, 5
for idx in range(1, cols * rows + 1):
image = data_iter['image'][idx]
figure.add_subplot(rows, cols, idx)
plt.axis("off")
plt.imshow(image.squeeze(), cmap="gray")
plt.show()
分析：例如，倘若 mnist_ds 中存储的是手写数字的图像数据，那么这段代码的作用便是从数据集中提取出 25 个图像，并以 5x5 的布局予以呈现，而且每个图像均不显示坐标轴，以灰度图的形式展现在我们眼前。
运行结果：
隐码构造
首先，导入了必需的模块与库，涵盖了随机模块 random 、numpy 库 np 、mindspore 中的 Tensor 类以及 mindspore.common 中的 dtype 。接着，为 numpy 设置了随机数种子为 2323 ，如此一来，能够确保每次运行代码时所生成的随机数序列保持一致。之后，运用 numpy 的 random.normal 函数生成了一个形状为 (25, 100) 的正态分布的随机数数组，紧接着将其转换为 mindspore 的 Tensor 类型，并且指定数据类型为 float32 。最后，使用 random 模块的 shuffle 函数对 test_noise 进行了随机打乱操作。
代码如下：
import random
import numpy as np
from mindspore import Tensor
from mindspore.common import dtype
# 利用随机种子创建一批隐码
np.random.seed(2323)
test_noise = Tensor(np.random.normal(size=(25, 100)), dtype.float32)
random.shuffle(test_noise)
分析：例如，如果这段代码在一个模型训练的上下文中，设置固定的随机种子可以保证实验的可重复性，生成的随机隐码 test_noise 经过打乱后可能用于模型的不同输入或处理步骤。
模型构建
导入了 mindspore 中的 nn 模块和 ops 模块。将训练图像的边长定义为 28 。定义了一个名为 Generator 的类，此类别继承自 nn.Cell 。在 __init__ 方法里构建了生成器的网络架构，涵盖了一连串的线性层、批量归一化层、激活函数等等。在 construct 方法中确立了前向传播的计算逻辑，对输入加以处理后借助 ops.reshape 来调整形状。创建了生成器的实例 net_g ，并且更新了其参数的名称。定义了一个名为 Discriminator 的判别器类，同样继承自 nn.Cell ，构建了判别器的网络结构以及前向传播的逻辑。创建了判别器的实例 net_d ，同时更新了其参数的名称。设定了学习率，定义了二分类交叉熵损失函数，创建了生成器和判别器的优化器，并对优化器参数的名称进行了更新。
代码如下：
from mindspore import nn
import mindspore.ops as ops
img_size = 28 # 训练图像长（宽）
class Generator(nn.Cell):
def __init__(self, latent_size, auto_prefix=True):
super(Generator, self).__init__(auto_prefix=auto_prefix)
self.model = nn.SequentialCell()
# [N, 100] -> [N, 128]
# 输入一个100维的0～1之间的高斯分布，然后通过第一层线性变换将其映射到256维
self.model.append(nn.Dense(latent_size, 128))
self.model.append(nn.ReLU())
# [N, 128] -> [N, 256]
self.model.append(nn.Dense(128, 256))
self.model.append(nn.BatchNorm1d(256))
self.model.append(nn.ReLU())
# [N, 256] -> [N, 512]
self.model.append(nn.Dense(256, 512))
self.model.append(nn.BatchNorm1d(512))
self.model.append(nn.ReLU())
# [N, 512] -> [N, 1024]
self.model.append(nn.Dense(512, 1024))
self.model.append(nn.BatchNorm1d(1024))
self.model.append(nn.ReLU())
# [N, 1024] -> [N, 784]
# 经过线性变换将其变成784维
self.model.append(nn.Dense(1024, img_size * img_size))
# 经过Tanh激活函数是希望生成的假的图片数据分布能够在-1～1之间
self.model.append(nn.Tanh())
def construct(self, x):
img = self.model(x)
return ops.reshape(img, (-1, 1, 28, 28))
net_g = Generator(latent_size)
net_g.update_parameters_name('generator')
# 判别器
class Discriminator(nn.Cell):
def __init__(self, auto_prefix=True):
super().__init__(auto_prefix=auto_prefix)
self.model = nn.SequentialCell()
# [N, 784] -> [N, 512]
self.model.append(nn.Dense(img_size * img_size, 512)) # 输入特征数为784，输出为512
self.model.append(nn.LeakyReLU()) # 默认斜率为0.2的非线性映射激活函数
# [N, 512] -> [N, 256]
self.model.append(nn.Dense(512, 256)) # 进行一个线性映射
self.model.append(nn.LeakyReLU())
# [N, 256] -> [N, 1]
self.model.append(nn.Dense(256, 1))
self.model.append(nn.Sigmoid()) # 二分类激活函数，将实数映射到[0,1]
def construct(self, x):
x_flat = ops.reshape(x, (-1, img_size * img_size))
return self.model(x_flat)
net_d = Discriminator()
net_d.update_parameters_name('discriminator')
lr = 0.0002 # 学习率
# 损失函数
adversarial_loss = nn.BCELoss(reduction='mean')
# 优化器
optimizer_d = nn.Adam(net_d.trainable_params(), learning_rate=lr, beta1=0.5, beta2=0.999)
optimizer_g = nn.Adam(net_g.trainable_params(), learning_rate=lr, beta1=0.5, beta2=0.999)
optimizer_g.update_parameters_name('optim_g')
optimizer_d.update_parameters_name('optim_d')
分析：例如，如果这是一个生成对抗网络（GAN）的代码片段，生成器 Generator 尝试生成逼真的图像来欺骗判别器 Discriminator ，而判别器则努力区分真实图像和生成器生成的假图像。通过不断优化生成器和判别器的参数，使得生成器能够生成越来越逼真的图像，而判别器的判别能力也不断提高。
模型训练
导入了所需的模块与函数，涵盖了与操作系统相关的 os 、与时间相关的 time 、用于绘图的 matplotlib.pyplot 、mindspore 库以及有关保存检查点的相关功能。对训练的周期数和训练集的批量大小进行了定义。设定了是否加载预训练模型的标识以及预训练模型的路径，同时也确立了结果和图像的保存路径。定义了生成器用于计算损失的函数。定义了判别器用于计算损失的函数。创建了生成器和判别器的梯度计算方式。定义了训练的步骤函数，其中包含了损失的计算、梯度的获取，并运用优化器来更新参数。定义了用于保存生成的测试图像的函数。创建了保存参数和生成图像的路径。将生成器和判别器设置为训练模式。对存储生成器和判别器损失的列表进行了初始化。这是训练的主循环，在每个周期当中，对数据集中的数据进行遍历，执行训练步骤，计算并打印损失等信息，存储损失值，生成并保存测试图像，依据周期数来保存模型权重。
代码如下：
import os
import time
import matplotlib.pyplot as plt
import mindspore as ms
from mindspore import Tensor, save_checkpoint
total_epoch = 12 # 训练周期数
batch_size = 64 # 用于训练的训练集批量大小
# 加载预训练模型的参数
pred_trained = False
pred_trained_g = './result/checkpoints/Generator99.ckpt'
pred_trained_d = './result/checkpoints/Discriminator99.ckpt'
checkpoints_path = "./result/checkpoints" # 结果保存路径
image_path = "./result/images" # 测试结果保存路径
# 生成器计算损失过程
def generator_forward(test_noises):
fake_data = net_g(test_noises)
fake_out = net_d(fake_data)
loss_g = adversarial_loss(fake_out, ops.ones_like(fake_out))
return loss_g
# 判别器计算损失过程
def discriminator_forward(real_data, test_noises):
fake_data = net_g(test_noises)
fake_out = net_d(fake_data)
real_out = net_d(real_data)
real_loss = adversarial_loss(real_out, ops.ones_like(real_out))
fake_loss = adversarial_loss(fake_out, ops.zeros_like(fake_out))
loss_d = real_loss + fake_loss
return loss_d
# 梯度方法
grad_g = ms.value_and_grad(generator_forward, None, net_g.trainable_params())
grad_d = ms.value_and_grad(discriminator_forward, None, net_d.trainable_params())
def train_step(real_data, latent_code):
# 计算判别器损失和梯度
loss_d, grads_d = grad_d(real_data, latent_code)
optimizer_d(grads_d)
loss_g, grads_g = grad_g(latent_code)
optimizer_g(grads_g)
return loss_d, loss_g
# 保存生成的test图像
def save_imgs(gen_imgs1, idx):
for i3 in range(gen_imgs1.shape[0]):
plt.subplot(5, 5, i3 + 1)
plt.imshow(gen_imgs1[i3, 0, :, :] / 2 + 0.5, cmap="gray")
plt.axis("off")
plt.savefig(image_path + "/test_{}.png".format(idx))
# 设置参数保存路径
os.makedirs(checkpoints_path, exist_ok=True)
# 设置中间过程生成图片保存路径
os.makedirs(image_path, exist_ok=True)
net_g.set_train()
net_d.set_train()
# 储存生成器和判别器loss
losses_g, losses_d = [], []
for epoch in range(total_epoch):
start = time.time()
for (iter, data) in enumerate(mnist_ds):
start1 = time.time()
image, latent_code = data
image = (image - 127.5) / 127.5 # [0, 255] -> [-1, 1]
image = image.reshape(image.shape[0], 1, image.shape[1], image.shape[2])
d_loss, g_loss = train_step(image, latent_code)
end1 = time.time()
if iter % 10 == 10:
print(f"Epoch:[{int(epoch):>3d}/{int(total_epoch):>3d}], "
f"step:[{int(iter):>4d}/{int(iter_size):>4d}], "
f"loss_d:{d_loss.asnumpy():>4f} , "
f"loss_g:{g_loss.asnumpy():>4f} , "
f"time:{(end1 - start1):>3f}s, "
f"lr:{lr:>6f}")
end = time.time()
print("time of epoch {} is {:.2f}s".format(epoch + 1, end - start))
losses_d.append(d_loss.asnumpy())
losses_g.append(g_loss.asnumpy())
# 每个epoch结束后，使用生成器生成一组图片
gen_imgs = net_g(test_noise)
save_imgs(gen_imgs.asnumpy(), epoch)
# 根据epoch保存模型权重文件
if epoch % 1 == 0:
save_checkpoint(net_g, checkpoints_path + "/Generator%d.ckpt" % (epoch))
save_checkpoint(net_d, checkpoints_path + "/Discriminator%d.ckpt" % (epoch))
运行结果：
分析：例如，如果这是一个生成对抗网络（GAN）的训练代码，它会在多个周期内不断优化生成器和判别器，生成并保存图像以及模型参数，以观察训练效果和模型的进展。
效果展示
首先创建了一个图形，对其大小进行了设定并赋予了标题。接着绘制了生成器和判别器的损失曲线，分别以蓝色和橙色予以呈现，同时还设置了坐标轴的范围、标签以及图例，最终将图形展示出来。创建了一个空的列表 image_list ，随后通过循环读取指定路径下每个周期所生成的测试图像，并将这些图像添加至该列表当中。创建了另一个空列表 show_list 以及一个图形对象 fig ，然后通过循环以步长为 5 从 image_list 中提取图像，设置坐标轴处于关闭状态，并将图像添加到 show_list 里。运用 matplotlib.animation 模块创建了一个动画对象 ani ，并对动画的相关参数进行了设置，例如间隔时间、重复延迟等。最后将动画保存为 gif 格式的文件 train_test.gif ，指定使用 pillow 作为写入器，帧率设定为 1 。
代码如下：
plt.figure(figsize=(6, 4))
plt.title("Generator and Discriminator Loss During Training")
plt.plot(losses_g, label="G", color='blue')
plt.plot(losses_d, label="D", color='orange')
plt.xlim(-5,15)
plt.ylim(0, 3.5)
plt.xlabel("iterations")
plt.ylabel("Loss")
plt.legend()
plt.show()
import cv2
import matplotlib.animation as animation
# 将训练过程中生成的测试图转为动态图
image_list = []
for i in range(total_epoch):
image_list.append(cv2.imread(image_path + "/test_{}.png".format(i), cv2.IMREAD_GRAYSCALE))
show_list = []
fig = plt.figure(dpi=70)
for epoch in range(0, len(image_list), 5):
plt.axis("off")
show_list.append([plt.imshow(image_list[epoch], cmap='gray')])
ani = animation.ArtistAnimation(fig, show_list, interval=1000, repeat_delay=1000, blit=True)
ani.save('train_test.gif', writer='pillow', fps=1)
分析：例如，如果这是一个关于图像生成模型的训练过程，上述代码可以直观地展示训练过程中生成器和判别器的损失变化，以及生成的测试图像的动态变化，帮助我们更好地理解模型的训练效果和生成图像的质量演变。
运行结果：
模型推理
导入 mindspore 库，并将其简称为 ms 。被注释掉的这部分代码原本旨在加载指定的检查点文件，并且把其中的参数加载到生成器网络 net_g 当中。创建了一个测试数据 test_data ，其为一个形状是 (25, 100) 的随机正态分布张量，同时转换为 float32 类型。将测试数据输入到生成器网络 net_g 中加以处理，从而获取生成的结果，接着对结果实施维度变换，并转换为 numpy 数组。创建了一个图形，对其大小和分辨率进行了设定。通过循环在图形里添加 25 个子图，每个子图用于展示生成的图像，关闭坐标轴，并且以灰度图的形式予以呈现。最后将图形展示出来。
代码如下：
import mindspore as ms
# test_ckpt = './result/checkpoints/Generator199.ckpt'
# parameter = ms.load_checkpoint(test_ckpt)
# ms.load_param_into_net(net_g, parameter)
# 模型生成结果
test_data = Tensor(np.random.normal(0, 1, (25, 100)).astype(np.float32))
images = net_g(test_data).transpose(0, 2, 3, 1).asnumpy()
# 结果展示
fig = plt.figure(figsize=(3, 3), dpi=120)
for i in range(25):
fig.add_subplot(5, 5, i + 1)
plt.axis("off")
plt.imshow(images[i].squeeze(), cmap="gray")
plt.show()
分析：例如，如果这是一个图像生成模型的测试代码，那么这段代码的作用是使用随机生成的数据作为输入，通过模型生成图像，并以 5x5 的布局展示出来，以便直观地观察生成的效果。
运行结果：
打印时间：
