还是第一个实验有意思,后面感觉都是框架了
存一个答案仓库:https://github.com/Yuichi1001/2024-AICS-EXP/tree/main
实验环境:
硬件平台:MLU 云平台环境。
软件环境:编程框架 PyTorch、计算加速库 CNNL、云端开发工具集 CNToolkit、Python3 及相关的扩展库 Pillow、Numpy、Scipy。
运行环境:
镜像收藏:mlu370_ubuntu22.04-student-1.22.1 :v4.2
基于 VGG19 实现图像分类
实验内容:
利用 PyTorch 的 API,实现基于 VGG19 进行图像分类的实验,运行平台包括 CPU 和 DLP,比较两种平台实现的差异。
实验目录:
/opt/code_chap_4/exp_4_1_VGG19/目录
代码补全
网络结构模块
generate_pth . py / evaluate_cpu . py / evaluate_cnnl_mfus . py这三个文件中的vgg19都一样的补全
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48
| def vgg19(): layers = [ 'conv1_1', 'relu1_1', 'conv1_2', 'relu1_2', 'pool1', 'conv2_1', 'relu2_1', 'conv2_2', 'relu2_2', 'pool2', 'conv3_1', 'relu3_1', 'conv3_2', 'relu3_2', 'conv3_3','relu3_3', 'conv3_4', 'relu3_4', 'pool3', 'conv4_1', 'relu4_1', 'conv4_2', 'relu4_2', 'conv4_3','relu4_3', 'conv4_4', 'relu4_4', 'pool4', 'conv5_1', 'relu5_1', 'conv5_2', 'relu5_2', 'conv5_3','relu5_3', 'conv5_4', 'relu5_4', 'pool5', 'flatten', 'fc6', 'relu6','fc7', 'relu7', 'fc8', 'softmax' ] layer_container = nn.Sequential() in_channels = 3 num_classes = 1000
chan_map = { '1': 64, '2': 128, '3': 256, '4': 512, '5': 512 } for i, layer_name in enumerate(layers): if layer_name.startswith('conv'): stage = layer_name[4] out_channels = chan_map[stage] layer_container.add_module(layer_name, nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1)) in_channels = out_channels elif layer_name.startswith('relu'): layer_container.add_module(layer_name, nn.ReLU(inplace=True)) elif layer_name.startswith('pool'): layer_container.add_module(layer_name, nn.MaxPool2d(kernel_size=2, stride=2)) elif layer_name == 'flatten': layer_container.add_module(layer_name, nn.Flatten()) elif layer_name == 'fc6': layer_container.add_module(layer_name, nn.Linear(512 * 7 * 7, 4096)) elif layer_name == 'fc7': layer_container.add_module(layer_name, nn.Linear(4096, 4096)) elif layer_name == 'fc8': layer_container.add_module(layer_name, nn.Linear(4096, num_classes)) elif layer_name == 'softmax': layer_container.add_module(layer_name, nn.Softmax(dim=1)) return layer_container
|
.pth 文件生成模块
generate_pth . py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
| if __name__ == '__main__': datas = scipy.io.loadmat(VGG_PATH)
model = vgg19() new_state_dict = OrderedDict() for i, param_name in enumerate(model.state_dict()): name = param_name.split('.') if name[-1] == 'weight': new_state_dict[param_name] = torch.from_numpy(datas[str(i)]).float() else: new_state_dict[param_name] = torch.from_numpy(datas[str(i)][0]).float() model.load_state_dict(new_state_dict) print("*** Start Saving pth ***") torch.save(model.state_dict(), 'models/vgg19.pth') print('Saving pth PASS.')
|
数据加载模块
evaluate_cpu . py / evaluate_cnnl_mfus . py
1 2 3 4 5 6 7 8 9 10 11 12 13
| def load_image(path): image = Image.open(path).convert('RGB') transform = transforms.Compose([transforms.Resize(256), transforms.CenterCrop(224), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])]) image = transform(image) image = image.unsqueeze(0) return image
|
CPU 推理模块
evaluate_cpu . py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
| if __name__ == '__main__': input_image = load_image(IMAGE_PATH) net = vgg19() net.load_state_dict(torch.load(VGG_PATH)) net.eval() st = time.time() prob = net(input_image) print("cpu infer time:{:.3f} s".format(time.time()-st)) with open('./labels/imagenet_classes.txt') as f: classes = [line.strip() for line in f.readlines()] _, indices = torch.sort(prob, descending=True) print("Classification result: id = %s, prob = %f " % (classes[indices[0][0]], prob[0][indices[0][0]].item())) if classes[indices[0][0]] == 'strawberry': print('TEST RESULT PASS.') else: print('TEST RESULT FAILED.') exit()
|
DLP 推理模块
evaluate_cnnl_mfus . py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36
| if __name__ == '__main__': input_image = load_image(IMAGE_PATH) net = vgg19() net.load_state_dict(torch.load(VGG_PATH)) net.eval() example_forward_input = torch.rand((1,3,224,224),dtype = torch.float) net_traced = torch.jit.trace(net, example_forward_input) input_image = input_image.to('mlu') net_traced = net_traced.to('mlu') st = time.time() prob = net_traced(input_image) print("mlu370<cnnl backend> infer time:{:.3f} s".format(time.time()-st)) prob = prob.to('cpu') with open('./labels/imagenet_classes.txt') as f: classes = [line.strip() for line in f.readlines()] _, indices = torch.sort(prob, descending=True) print("Classification result: id = %s, prob = %f " % (classes[indices[0][0]], prob[0][indices[0][0]].item())) if classes[indices[0][0]] == 'strawberry': print('TEST RESULT PASS.') else: print('TEST RESULT FAILED.')
|
实验步骤
补全 stu_upload 中的 generate_pth.py、evaluate_cpu.py、evaluate_cnnl_mfus.py 文件;
运行:
1 2 3 4
| bash run_cpu.sh
bash run_mlu.sh
|
实时风格迁移算法的推理
实验目的:
掌握如何使用 PyTorch 实现实时风格迁移算法中的图像转换网络,并基于其完成图像 风格迁移处理。具体包括:
- 掌握使用 PyTorch 定义完整网络结构的方法;
- 掌握使用 PyTorch 加载模型结构及参数的方法;
- 以实时风格迁移算法为例,掌握基于 PyTorch 在 CPU 平台上进行神经网络推理的 方法;
- 掌握基于 PyTorch 在 DLP 平台上进行神经网络推理的方法。
实验内容:
利用 PyTorch 的 API,实现基于 VGG19 进行图像分类的实验,运行平台包括 CPU 和 DLP,比较两种平台实现的差异。
实验目录:
/opt/code_chap_4/exp_4_2_SytleTransfer_infer/
实时风格迁移推理算法
训练
首先输入图像到图像转换网络生成风格化图像,再利用特征提取网络计算内容损失和风格损失,然后 迭代地更新图像转换网络参数以最小化损失。
推理
利用训练好的图像转 换网络对任意输入图像做网络推理来生成风格迁移后的图像,该过程基本达到实时。
网络结构
**下采样卷积层:**用于在特征提取的同时减少特征图尺寸, 由三个卷积层构成
**残差层:**通过残差连接缓解神经网络退化问题,由五个残差块构成,每 个残差块又包含两层卷积
**上采样卷积层:**用于在特征提取的同时逐渐恢复特征图尺寸,其不再使用传统的卷积,而是先执行上采样操作,再执行卷积操作,以避免“棋盘伪影”现象 的发生。
**输出层:**用于将前序网络输出的特征图映射到目标图像的通道数,由一个卷积层以 及 Sigmoid 激活函数构成。
除了输出层,所有非残差卷积层后面都加了实例归一化和 ReLU 操作。第一层和最后一层卷积使用 9 × 9 卷积核,其他卷积层均 使用 3 × 3 卷积核。

残差块
基本结构:
输入 x 经过一 个卷积层(即图4.2中的权重层),再做 ReLU,然后经过另一个卷积层得到 F (x),再加上 x 得到输出 H (x) = F (x) + x,然后做 ReLU 得到残差块的最终输出 y。

与卷积神经网络相比,残差块增加了从输入到输出的直连(Shortcut Connection),其拟合的是输出与输入的差(即残差)。由于输入和输出都做了批归一化,符合正态分布,因此 输入和输出可以做减法。
残差网络的优点是对数据波动更灵敏,更容易求得最优解
归一化
**批归一化:**对批次内的样本 进行归一化,将输入特征归一化到标准正态分布
实例归一化:批归一化方法是对图像的一个批次做归一化以保证数据分布的一致性,而 在实时风格迁移算法中,由于迁移后的结果主要依赖单张图像实例,所以对输入批次整体 归一化并不适合。
实例归一化保持每个图 像实例之间的独立性。公式:
$$
y_{tijk} = \frac{x_{tijk} - \mu_{ti}}{\sqrt{\sigma_{ti}^2 + \epsilon}}, \quad
\mu_{ti} = \frac{1}{HW} \sum_{j=1}^{W} \sum_{k=1}^{H} x_{tijk}, \quad
\sigma_{ti}^2 = \frac{1}{HW} \sum_{j=1}^{W} \sum_{k=1}^{H} (x_{tijk} - \mu_{ti})^2
$$
代码补全
数据加载模块
evaluate_cpu . py / evaluate_cnnl_mfus . py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28
| class COCODataSet(Dataset):
def __init__(self): super(COCODataSet, self).__init__() self.zip_files = ZipFile('./data/train2014_small.zip') self.data_set = [] for file_name in self.zip_files.namelist(): if file_name.endswith('.jpg'): self.data_set.append(file_name)
def __len__(self): return len(self.data_set)
def __getitem__(self, item): file_path = self.data_set[item] image = self.zip_files.read(file_path) image = numpy.asarray(bytearray(image), dtype='uint8') image = cv2.imdecode(image, cv2.IMREAD_COLOR) image = cv2.resize(image, (512, 512), interpolation=cv2.INTER_AREA) image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) image = torch.from_numpy(image).float() / 255.0 image = image.permute(2, 0, 1) return image
|
网络结构模块
ResBlock是残差块
风格迁移网络(Fast Style Transfer)为了配合 InstanceNorm2d 使用,通常会将卷积层的 bias 设置为 False,因为归一化操作会抵消掉偏置的作用。前面需要设置bias=False,最后一个卷积层(输出层)不设
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80
| class ResBlock(nn.Module):
def __init__(self, c): super(ResBlock, self).__init__() self.layer = nn.Sequential( nn.Conv2d(c, c, kernel_size=3, stride=1, padding=1, bias=False), nn.InstanceNorm2d(c), nn.ReLU(inplace=True), nn.Conv2d(c, c, kernel_size=3, stride=1, padding=1, bias=False), nn.InstanceNorm2d(c), )
def forward(self, x): return torch.nn.functional.relu(x + self.layer(x))
class TransNet(nn.Module):
def __init__(self): super(TransNet, self).__init__() self.layer = nn.Sequential( nn.Conv2d(3, 32, kernel_size=9, stride=1, padding=4, bias=False), nn.InstanceNorm2d(32), nn.ReLU(inplace=True), nn.Conv2d(32, 64, kernel_size=3, stride=2, padding=1, bias=False), nn.InstanceNorm2d(64), nn.ReLU(inplace=True), nn.Conv2d(64, 128, kernel_size=3, stride=2, padding=1, bias=False), nn.InstanceNorm2d(128), nn.ReLU(inplace=True),
ResBlock(128), ResBlock(128), ResBlock(128), ResBlock(128), ResBlock(128),
nn.Upsample(scale_factor=2, mode='nearest'), nn.Conv2d(128, 64, kernel_size=3, stride=1, padding=1, bias=False), nn.InstanceNorm2d(64), nn.ReLU(inplace=True),
nn.Upsample(scale_factor=2, mode='nearest'), nn.Conv2d(64, 32, kernel_size=3, stride=1, padding=1, bias=False), nn.InstanceNorm2d(32), nn.ReLU(inplace=True),
nn.Conv2d(32, 3, kernel_size=9, stride=1, padding=4), nn.Sigmoid() )
|
CPU 推理模块
evaluate_cpu . py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25
| if __name__ == '__main__': g_net = TransNet().cpu() g_net.load_state_dict(torch.load('./models/fst.pth', map_location='cpu')) print("g_net build PASS!\n") data_set = COCODataSet() print("load COCODataSet PASS!\n")
batch_size = 1 data_group = DataLoader(data_set,batch_size,True,drop_last=True)
for i, image in enumerate(data_group): image_c = image.cpu() start = time.time() image_g = g_net(image_c) end = time.time() delta_time = end - start print("Inference (CPU) processing time: %s" % delta_time) save_image(torch.cat((image_c, image_g), dim=3), f'./out/cpu/{i}.jpg') print("TEST RESULT PASS!\n")
|
DLP 推理模块
evaluate_cnnl_mfus . py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33
| if __name__ == '__main__': g_net = TransNet().cpu() g_net.load_state_dict(torch.load('./models/g_net.pth', map_location='cpu')) print("g_net build PASS!\n") g_net.eval().float() data_set = COCODataSet() print("load COCODataSet PASS!\n") batch_size = 1 data_group = DataLoader(data_set,batch_size,True,drop_last=True) example_forward_input = torch.rand((1,3,512,512),dtype = torch.float) net_traced = torch.jit.trace(g_net, example_forward_input) for i, image in enumerate(data_group): print(f"The {i} image will be predicted.") image_c = image.cpu() input_image_c = image_c.to('mlu') net_mlu = net_traced.to('mlu') start = time.time() image_g_mlu = net_mlu(input_image_c) image_g_mlu = image_g_mlu.cpu() end = time.time() delta_time = end - start print("Inference (mfus) processing time: %s" % delta_time) save_image(torch.cat((image_c, image_g_mlu), dim=3), f'./out/mlu_cnnl_mfus/{i}.jpg')
print("TEST RESULT PASS!\n")
|
实验步骤
先补全两个文件,然后同上运行脚本
实时风格迁移算法的训练
实验目的:
掌握如何使用 PyTorch 实现实时风格迁移算法的训练。具体包括:
- 掌握使用 PyTorch 定义损失函数的方法;
- 掌握使用 PyTorch 存储网络模型的方法;
- 以实时风格迁移算法为例,掌握使用 PyTorch 进行神经网络训练的方法
实验内容:
基于 PyTorch 的 API 实现卷积层、残差块等基本单元,构建实时风格 迁移网络,通过特征提取网络构建损失函数,并基于该损失函数来迭代地训练图像转换网 络,最终取得较好的训练效果。同时,比较 CPU 与 DLP 在实时风格迁移训练中的实现 差异。
实验目录:
/opt/code_chap_4/exp_4_3_StyleTransfer_train/
实时风格迁移训练算法
输入的内容图像 x 经过图像转换网 络输出风格化图像(生成图像)yˆ;其次,利用特征提取网络(原始网络为在 ImageNet 数据 集上预训练好的 VGG16,这里改为 VGG19)分别提取生成图像 yˆ、风格图像 ys 和内容图 像 yc 的特征,并利用这些特征计算损失函数;然后,通过迭代地调整图像转换网络的参数 来最小化损失函数,最终完成对图像转换网络的训练。

损失函数
由特征重建损失 $\mathcal{L}{feat}$ 和风格重建损失 $\mathcal{L}{style}$ 两部分组成:
$$
\mathcal{L} = \mathbb{E}{x} \left[ \lambda_1 \mathcal{L}{feat}(f_W(x), y_c) + \lambda_2 \mathcal{L}_{style}(f_W(x), y_s) \right] \tag{4.3}
$$
其中, $\lambda_1$ 和 $\lambda_2$ 是权重参数。特征重建损失用卷积输出的特征计算视觉损失:
$$
\mathcal{L}^j_{feat}(\hat{y}, y) = \frac{1}{C_j H_j W_j} | \varphi_j(\hat{y}) - \varphi_j(y) |_2^2 \tag{4.4}
$$
其中, $C_j$、$H_j$、$W_j$ 分别表示第 $j$ 层卷积输出特征图的通道数、高度和宽度, $\varphi(y)$ 是特征提取网络中第 $j$ 层卷积输出的特征图,实际中选择第 $4$ 层卷积的特征计算特征重建损失。
而第 $j$ 层卷积后的风格重建损失为输出图像和目标图像的格拉姆矩阵(Gram Matrix)的差的 F-范数:
$$
\mathcal{L}^j_{style}(\hat{y}, y) = | G_j(\hat{y}) - G_j(y) |_F^2 \tag{4.5}
$$
其中,格拉姆矩阵 $G_j(x)$ 为 $C_j \times C_j$ 大小的矩阵,它衡量了特征之间的相关性。
代码补全
数据加载模块
t r a i n . py / train −mlu . py
1 2 3 4 5 6 7 8 9 10 11 12
| def load_image(path): image = cv2.imread(path) image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) image = cv2.resize(image, (512, 512), interpolation=cv2.INTER_AREA) image = torch.from_numpy(image).float() / 255.0 image = image.permute(2, 0, 1).unsqueeze(0) return image
|
网络结构模块
1 2 3 4 5 6 7 8 9 10 11 12 13 14
| class VGG19(nn.Module): def __init__(self): super(VGG19, self).__init__() a = vgg19(True) a = a.features self.layer1 = nn.Sequential(a[:4]) self.layer2 = nn.Sequential(a[4:9]) self.layer3 = nn.Sequential(a[9:18]) self.layer4 = nn.Sequential(a[18:27])
|
CPU 训练模块
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67
| if __name__ == '__main__': image_style = load_image('./data/udnie.jpg').cpu() net = VGG19().cpu() g_net = TransNet().cpu() print("g_net build PASS!\n") optimizer = torch.optim.Adam(g_net.parameters()) loss_func = nn.MSELoss() print("build loss PASS!\n") data_set = COCODataSet() print("load COCODataSet PASS!\n") batch_size = 1 data_loader = DataLoader(data_set, batch_size, True, drop_last=True) s1, s2, s3, s4 = net(image_style) s1 = get_gram_matrix(s1).detach() s2 = get_gram_matrix(s2).detach() s3 = get_gram_matrix(s3).detach() s4 = get_gram_matrix(s4).detach() j = 0 count = 0 epochs = 0 while j <= epochs: for i, image in enumerate(data_loader): """生成图片,计算损失""" image_c = image.cpu() image_g = g_net(image_c) out1, out2, out3, out4 = net(image_g) """计算风格损失""" loss_s1 = loss_func(get_gram_matrix(out1), s1) loss_s2 = loss_func(get_gram_matrix(out2), s2) loss_s3 = loss_func(get_gram_matrix(out3), s3) loss_s4 = loss_func(get_gram_matrix(out4), s4) loss_s = loss_s1 + loss_s2 + loss_s3 + loss_s4
"""计算内容损失""" c1, c2, c3, c4 = net(image_c) loss_c2 = loss_func(out2, c2.detach()) loss_c = loss_c2 """总损失""" loss = loss_c + 0.000000005 * loss_s
"""清空梯度、计算梯度、更新参数""" optimizer.zero_grad() loss.backward() optimizer.step() print('j:',j, 'i:',i, 'loss:',loss.item(), 'loss_c:',loss_c.item(), 'loss_s:',loss_s.item()) count += 1 if i % 10 == 0: torch.save(g_net.state_dict(), './models/fst_train.pth') save_image(torch.cat([image_c, image_g], dim=3), './out/train/{}.jpg'.format(count)) j += 1
print("TRAIN RESULT PASS!\n")
|
DLP 训练模块
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82
| if __name__ == '__main__': image_style = load_image('./data/udnie.jpg').cpu() mlu_image_style = image_style.to('mlu') net = VGG19().cpu() g_net = TransNet().cpu() mlu_g_net = g_net.to('mlu') mlu_net = net.to('mlu') print("mlu_net build PASS!\n") optimizer = torch.optim.Adam(mlu_g_net.parameters()) loss_func = nn.MSELoss() mlu_loss_func = loss_func.to('mlu') print("build loss PASS!\n") data_set = COCODataSet() print("load COCODataSet PASS!\n") batch_size = 1 data_loader = DataLoader(data_set, batch_size, True, drop_last=True) s1, s2, s3, s4 = mlu_net(mlu_image_style) s1 = get_gram_matrix(s1).detach() s2 = get_gram_matrix(s2).detach() s3 = get_gram_matrix(s3).detach() s4 = get_gram_matrix(s4).detach() j = 0 count = 0 epochs = 0 while j <= epochs: for i, image in enumerate(data_loader): """生成图片,计算损失""" image_c = image.cpu() mlu_image_c = image_c.to('mlu') mlu_image_g = mlu_g_net(mlu_image_c) out1, out2, out3, out4 = mlu_net(mlu_image_g) """计算风格损失""" loss_s1 = mlu_loss_func(get_gram_matrix(out1), s1) loss_s2 = mlu_loss_func(get_gram_matrix(out2), s2) loss_s3 = mlu_loss_func(get_gram_matrix(out3), s3) loss_s4 = mlu_loss_func(get_gram_matrix(out4), s4) loss_s = loss_s1 + loss_s2 + loss_s3 + loss_s4
"""计算内容损失""" c1, c2, c3, c4 = mlu_net(mlu_image_c)
c2 = c2.detach() loss_c2 = mlu_loss_func(out2, c2) loss_c = loss_c2
"""总损失""" loss = loss_c + 0.000000005 * loss_s
"""清空梯度、计算梯度、更新参数""" optimizer.zero_grad() loss.backward() optimizer.step() print('j:',j, 'i:',i, 'loss:',loss.item(), 'loss_c:',loss_c.item(), 'loss_s:',loss_s.item()) count += 1 mlu_image_g = mlu_image_g.cpu() mlu_image_c = mlu_image_c.cpu() if i % 10 == 0: torch.save(mlu_g_net.state_dict(), 'models/fst_train_mlu.pth') save_image(torch.cat([mlu_image_c, mlu_image_g], dim=3), f'out/train_mlu/{j}_{i}.jpg') j += 1
print("MLU TRAIN RESULT PASS!\n")
|
实验步骤
先补全两个文件(除了上面提到的代码其他的按照实验4.2补全),然后同上运行脚本