哈喽大家好，最近刚开始接触FPGA，便要在上面部署神经网络，特此记录一下整体过程。

我用的FPGA板子是xc7z020clg400-2，部署的神经网络是LeNet-5，主要就是走通一下整体过程吧，为之后可能部署更复杂的网络打一下基础。

我用的相关软件是Pycharm，HLS，Vivado，Vitis(2020.2)，以下是一个整体的流程图。

ok，废话不多说，直接上代码：

关于训练模型相关的代码，我看的是炮哥带你学的LeNet相关内容，可以直接用的：

model.py

# 导入pytorch库
import torch
# 导入torch.nn模块
from torch import nn
from torchsummary import summary


# 定义这个LeNet网络模型class def super 三行，约定俗成
# MyLeNet5（子类）继承nn.Module（父类）
class MyLeNet5(nn.Module):
    # 子类继承中重新定义Module类的__init__()和forward()函数
    # init()函数：进行初始化.....，申明模型中各层的定义
    def __init__(self):
        # super：引入父类的初始化方法给子类进行初始化
        super(MyLeNet5, self).__init__()
        # 定义卷积层，输入大小为28*28，输出大小为28*28，输入通道为1（灰度图），输出为6，卷积核为5即卷积核大小5*5，扩充边缘为2
        self.c1 = nn.Conv2d(in_channels=1, out_channels=6, kernel_size=5, padding=2)
        # 使用sigmoid作为激活函数
        self.Sigmoid = nn.Sigmoid()
        # AvgPool2d：二维平均池化操作
        # 池化层，输入大小为28*28，输出大小为14*14，输入通道为6，输出为6，卷积核为2，步长为2
        self.s2 = nn.AvgPool2d(kernel_size=2, stride=2)
        # 卷积层，输入大小为14*14，输出大小为10*10，输入通道为6，输出为16，卷积核为5
        self.c3 = nn.Conv2d(in_channels=6, out_channels=16, kernel_size=5)
        # 池化层，输入大小为10*10，输出大小为5*5，输入通道为16，输出为16，卷积核为2，步长为2
        self.s4 = nn.AvgPool2d(kernel_size=2, stride=2)


        # Flatten()：将张量（多维数组）平坦化处理，张量的第0维表示的是batch_size（数量），所以Flatten()默认从第二维开始平坦化
        self.flatten = nn.Flatten()
        # 全连接层
        # Linear（in_features，out_features）
        # in_features指的是[batch_size, size]中的size,即样本的大小
        # out_features指的是[batch_size，output_size]中的output_size，样本输出的维度大小，也代表了该全连接层的神经元个数
        self.f5 = nn.Linear(400, 120)
        self.f6 = nn.Linear(120, 84)
        self.f7 = nn.Linear(84, 10)


    # forward()：定义前向传播过程,描述了各层之间的连接关系
    def forward(self, x):
        # x输入为28*28*1， 输出为28*28*6
        x = self.Sigmoid(self.c1(x))
        # x输入为28*28*6，输出为14*14*6
        x = self.s2(x)
        # x输入为14*14*6，输出为10*10*16
        x = self.Sigmoid(self.c3(x))
        # x输入为10*10*16，输出为5*5*16
        x = self.s4(x)
        # x输入为5*5*16，输出为1*1*120
        x = self.flatten(x)
        x = self.f5(x)
        x = self.f6(x)
        x = self.f7(x)

        return x


# 测试代码
# 每个python模块（python文件）都包含内置的变量 __name__，当该模块被直接执行的时候，__name__ 等于文件名（包含后缀 .py ）
# 如果该模块 import 到其他模块中，则该模块的 __name__ 等于模块名称（不包含后缀.py）
# “__main__” 始终指当前执行模块的名称（包含后缀.py）
# if确保只有单独运行该模块时，此表达式才成立，才可以进入此判断语法，执行其中的测试代码，反之不行
###
if __name__ == "__main__":
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    print(device)
    model = MyLeNet5().to(device)
    print(summary(model,(1,28,28)))

train.py

import copy
import time

import pandas as pd
from torchvision.datasets import FashionMNIST
from torchvision import transforms
import torch.utils.data as Data
import numpy as np
import matplotlib.pyplot as plt
from model import MyLeNet5
import torch
import torch.nn as nn

def train_val_data_process():
    train_data = FashionMNIST(root='./data',
                              train = True,
                              transform = transforms.Compose([transforms.Resize(size=28), transforms.ToTensor()]),
                              download = True)
    train_data,val_data = Data.random_split(train_data,[round(0.8*len(train_data)),round(0.2*len(train_data))])#划分训练集验证集

    train_dataloader = Data.DataLoader(dataset=train_data,
                                       batch_size=16,
                                       shuffle=True,
                                       num_workers=2)
    val_dataloader = Data.DataLoader(dataset=val_data,
                                       batch_size=16,
                                       shuffle=True,
                                       num_workers=2)
    return train_dataloader,val_dataloader

def train_model_process(model,train_dataloader,val_dataloader,num_epochs):
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    optimizer = torch.optim.Adam(model.parameters(),lr=0.001) #优化器，模型参数更新，梯度下降法的优化
    criterion = nn.CrossEntropyLoss()  #交叉熵损失函数
    model = model.to(device)

    best_model_wts = copy.deepcopy(model.state_dict())

    #初始化参数
    best_acc = 0.0 #最高准确度
    train_loss_all = []#损失列表
    val_loss_all = []
    train_acc_all = []
    val_acc_all = []

    since = time.time()#当前时间

    for epoch in range(num_epochs):
        print("Epoch {}/{}".format( epoch,num_epochs-1))
        print("-"*10)

        train_loss = 0.0
        train_corrects = 0.0
        val_loss = 0.0
        val_corrects = 0.0
        train_num = 0.0
        val_num = 0.0

        for step,(b_x,b_y) in enumerate(train_dataloader):
            b_x = b_x.to(device)
            b_y = b_y.to(device)
            model.train()

            output = model(b_x)

            pre_lab = torch.argmax(output,dim = 1)#查找每一行中最大值对应的行标
            loss = criterion(output,b_y)

            optimizer.zero_grad()#将梯度置为零
            loss.backward()#反向传播计算
            optimizer.step()#根据网络反向传播的梯度信息来更新网络的参数，以起到降低loss函数计算值的作用
            train_loss += loss.item()* b_x.size(0)#对损失函数进行累加
            train_corrects += torch.sum(pre_lab == b_y.data)
            train_num += b_x.size(0)

        for step, (b_x, b_y) in enumerate(val_dataloader):
            b_x = b_x.to(device)
            b_y = b_y.to(device)

            model.eval()
            output = model(b_x)
            pre_lab = torch.argmax(output, dim=1)  # 查找每一行中最大值对应的行标
            loss = criterion(output, b_y)
            val_loss += loss.item() * b_x.size(0)  # 对损失函数进行累加
            val_corrects += torch.sum(pre_lab == b_y.data)
            val_num += b_x.size(0)

        train_loss_all.append(train_loss / train_num)
        train_acc_all.append(train_corrects.double().item() / train_num)

        val_loss_all.append(val_loss / val_num)
        val_acc_all.append(val_corrects.double().item() / val_num)

        print('{} Train Loss: {:.4f} Train Acc: {:.4f} '.format(epoch,train_loss_all[-1], train_acc_all[-1]))
        print('{} Val Loss: {:.4f} Val Acc: {:.4f} '.format(epoch, val_loss_all[-1], val_acc_all[-1]))

        if val_acc_all[-1] > best_acc:#寻找最高准确度
            best_acc = val_acc_all[-1]

            best_model_wts = copy.deepcopy(model.state_dict())

        time_use = time.time() - since#训练时间
        print("训练和验证耗费的时间{:.0f}m{:.0f}s".format(time_use//60,time_use%60 ))

        #选择最优参数
        #加载最高准确率下的模型参数
        torch.save(best_model_wts,'F:/Python_Net/LeNet-5/best_model1.pth')

    train_process = pd.DataFrame(data={"epoch":range(num_epochs),
                                       "train_loss_all":train_loss_all,
                                       "val_loss_all": val_loss_all,
                                       "train_acc_all": train_acc_all,
                                       "val_acc_all": val_acc_all,})
    return train_process

def matplot_acc_loss(train_process):
    plt.figure(figsize=(12,4))
    plt.subplot(1,2,1)
    plt.plot(train_process["epoch"],train_process.train_loss_all,'ro-',label= "train loss")
    plt.plot(train_process["epoch"], train_process.val_loss_all, 'bs-', label="val loss")
    plt.legend()
    plt.xlabel("epoch")
    plt.ylabel("loss")

    plt.subplot(1, 2, 2)
    plt.plot(train_process["epoch"], train_process.train_acc_all, 'ro-', label="train loss")
    plt.plot(train_process["epoch"], train_process.val_acc_all, 'bs-', label="val loss")
    plt.legend()
    plt.xlabel("epoch")
    plt.ylabel("acc")

    plt.legend()
    plt.show()

if __name__ == "__main__":

    MyLeNet5 = MyLeNet5()
    train_dataloader,val_dataloader = train_val_data_process()
    train_process = train_model_process(MyLeNet5, train_dataloader, val_dataloader, 20)
    matplot_acc_loss(train_process)

test.py

import torch
import torch.utils.data as Data
from torchvision import transforms
from torchvision.datasets import FashionMNIST
from model import MyLeNet5

def test_data_process():
    test_data = FashionMNIST(root='./data',
                              train = False,
                              transform = transforms.Compose([transforms.Resize(size=28), transforms.ToTensor()]),
                              download = True)

    test_dataloader = Data.DataLoader(dataset=test_data,
                                       batch_size=1,
                                       shuffle=True,
                                       num_workers=0)

    return test_dataloader

def test_model_process(model,test_dataloader):
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = model.to(device)

    test_corrects = 0.0
    test_num = 0.0

    with torch.no_grad():
        for test_data_x,test_data_y in test_dataloader:
            test_data_x = test_data_x.to(device)
            test_data_y = test_data_y.to(device)

            model.eval()
            output = model(test_data_x)
            pre_lab = torch.argmax(output,dim=1)
            test_corrects += torch.sum(pre_lab == test_data_y.data)
            test_num += test_data_x.size(0)


    test_acc = test_corrects.double().item() / test_num
    print("测试的准确率:",test_acc)

if __name__ == "__main__":
    model = MyLeNet5()
    model.load_state_dict(torch.load('best_model.pth'))
    test_dataloader = test_data_process()
    #test_model_process(model,test_dataloader)
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = model.to(device)

    classes = ['T-shirt/top', 'Trouser', 'Pullover', 'Dress', 'Coat', 'Sandal', 'Shirt', 'Sneaker', 'Bag', 'Ankle boot']
    with torch.no_grad():
        for b_x,b_y in test_dataloader:
            b_x = b_x.to(device)
            b_y = b_y.to(device)

            model.eval()
            output = model(b_x)
            pre_lab = torch.argmax(output, dim=1)
            result = pre_lab.item()
            label = b_y.item()
            print("预测值:",classes[result],"----------""真实值:",classes[label])

经过训练后会得到一个best_model.pth，这里面包括了卷积全连接层相关的权重和偏置参数，需要把这些数据分别保存成相关文件，我根据一个b站的高手学习资料，把他们分别保存成dat文件，这里我是手动保存的：
 

import torch
import numpy as np
from model import MyLeNet5

# 加载训练好的模型
model = MyLeNet5()
model.load_state_dict(torch.load('best_model1.pth'))
model.eval()  # 设置为评估模式

# 提取模型的各层权重
weights = {}
for name, param in model.named_parameters():
    if 'weight' in name:
        weights[name] = param.detach().numpy()
    elif 'bias' in name:
        weights[name] = param.detach().numpy()


# 将权重保存为 C 语言源文件
def save_weights_as_c_file(weight_dict, output_filename):
    with open(output_filename, 'w') as f:
        f.write("#include <stdio.h>\n\n")

        # 遍历所有权重并格式化为 C 数组
        for name, weight in weight_dict.items():
            # 判断是卷积层还是全连接层，并格式化成对应的 C 数组
            if 'weight' in name:
                # 提取层的维度信息
                layer_name = name.split('.')[0]
                dims = weight.shape

                f.write(f"// {layer_name} weights\n")

                if len(dims) == 4:  # 对于卷积层，权重是四维的 (out_channels, in_channels, kernel_height, kernel_width)
                    f.write(f"float {layer_name}_weights[{dims[0]}][{dims[1]}][{dims[2]}][{dims[3]}] = {{\n")
                    for i in range(dims[0]):  # 遍历输出通道
                        f.write("    {\n")
                        for j in range(dims[1]):  # 遍历输入通道
                            f.write("        { ")
                            for k in range(dims[2]):  # 遍历卷积核的高度
                                for l in range(dims[3]):  # 遍历卷积核的宽度
                                    f.write(f"{weight[i, j, k, l]:.7f}, ")
                            f.write("},\n")
                        f.write("    },\n")
                    f.write("};\n\n")

                elif len(dims) == 2:  # 对于全连接层，权重是二维的 (out_features, in_features)
                    f.write(f"float {layer_name}_weights[{dims[0]}][{dims[1]}] = {{\n")
                    for i in range(dims[0]):
                        f.write("    { ")
                        for j in range(dims[1]):
                            f.write(f"{weight[i, j]:.7f}, ")
                        f.write("},\n")
                    f.write("};\n\n")
                else:
                    print(f"Unrecognized weight dimensions for {layer_name}: {dims}")

            elif 'bias' in name:
                # 提取层的维度信息
                layer_name = name.split('.')[0]
                dims = weight.shape

                f.write(f"// {layer_name} biases\n")
                f.write(f"float {layer_name}_biases[{dims[0]}] = {{\n")

                # 写入偏置值
                f.write("    ")
                for i, bias in enumerate(weight):
                    f.write(f"{bias:.7f}")
                    if i < len(weight) - 1:
                        f.write(", ")
                f.write("\n};\n\n")

    print(f"Saved all weights to {output_filename}")


# 保存所有权重到单一的 C 文件
save_weights_as_c_file(weights, 'all_weights.c')

具体就是把参数都保存到一个.c文件中，然后从这个文件中复制数据到对应的dat文件。

之后需要再提取一下图像数据，用于在HLS中tb仿真：
 

import torch
import numpy as np
import matplotlib.pyplot as plt
from torchvision import transforms
from model import MyLeNet5  # 假设你已经在 model.py 中定义了 MyLeNet5 模型
import struct

# 读取 Fashion-MNIST 数据并归一化的部分代码
file_path = "F:/Python_Net/LeNet-5/data/FashionMNIST/raw/t10k-images-idx3-ubyte"
with open(file_path, 'rb') as f:
    magic, num_images, rows, cols = struct.unpack('>IIII', f.read(16))  # 解包数据
    print(f"Magic Number: {magic}, 图像数量: {num_images}, 尺寸: {rows}x{cols}")
    image_data = f.read(rows * cols * num_images)
    images = np.frombuffer(image_data, dtype=np.uint8)
    images = images.reshape(num_images, rows, cols)

# 选择第 1000 张图像
first_image = images[3000]

# 归一化处理
normalized_image = first_image / 255.0

# 将归一化后的图像转换为 int 类型，并将其放回到 [0, 255] 范围内
int_image = (normalized_image * 255).astype(np.int32)  # 转换为整数类型，并恢复到 0-255 范围

# 将图像数据转换为16进制，并保存到文件
output_path = "F:/Python_Net/LeNet-5/data/FashionMNIST/first_image_hex.dat"
with open(output_path, 'w') as f:
    for row in int_image:
        for pixel in row:
            # 将每个整数像素值转换为16进制格式
            hex_value = f"0x{pixel:02x}"
            f.write(f"{hex_value},\n")  # 每个像素值存储为16进制格式，并加逗号
print(f"✅ 16进制格式的图像数据已保存到: {output_path}")

# 显示图像
plt.imshow(int_image, cmap='gray')
plt.title("Fashion-MNIST (Hex) ")
plt.axis('off')
plt.show()

# 转换图像为Tensor，并确保形状为 [1, 1, 28, 28]
# 这里你可以将数据类型更改为 int32，如果你需要的是 int 类型的 Tensor
# 然后将其转换为 float32 类型，这是 LeNet 需要的格式
transform = transforms.Compose([transforms.ToTensor()])
normalized_image_tensor = transform(int_image).unsqueeze(0).float()  # 这里将其转换为 float32 类型，因为 LeNet 处理的是浮点数输入
print(f"输入图像的形状: {normalized_image_tensor.shape}")  # 确认形状为 [1, 1, 28, 28]

# 载入模型
model = MyLeNet5()  # 假设你已经定义了MyLeNet5模型

# 加载训练好的模型权重
model_weights_path = 'best_model1.pth'
try:
    model.load_state_dict(torch.load(model_weights_path))
    print(f"✅ 成功加载模型权重: {model_weights_path}")
except Exception as e:
    print(f"❌ 加载模型时发生错误: {e}")

# 确认模型结构
print(model)

# 确定设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)
normalized_image_tensor = normalized_image_tensor.to(device)

# 用于存储F7层输出的钩子
f7_output = []

# 定义钩子函数来捕获f7层的输出
def hook_fn(module, input, output):
    global f7_output
    f7_output = output.detach()
    print("F7层输出的形状:", f7_output.shape)  # 打印F7输出的形状
    print("F7层输出的值:", f7_output)  # 打印F7层的值

# 打印模型层结构，确认F7层位置
print(model)

# 这里你需要根据你模型的实际结构来注册钩子
# 假设f7是最后一个卷积层或全连接层，可以修改这部分代码以匹配正确的层
# 注册f7层的钩子（假设模型有名为f7的层）
if hasattr(model, 'f7'):
    model.f7.register_forward_hook(hook_fn)
else:
    print("❌ 模型没有名为f7的层，请检查模型结构")

# 将模型设置为评估模式
model.eval()

# 推理：通过模型获得预测
with torch.no_grad():
    output = model(normalized_image_tensor)
    predicted_label = torch.argmax(output, dim=1).item()

# 定义类别名称
classes = ['T-shirt/top', 'Trouser', 'Pullover', 'Dress', 'Coat', 'Sandal', 'Shirt', 'Sneaker', 'Bag', 'Ankle boot']

# 输出预测类别
print(f"预测类别: {classes[predicted_label]}")

# 打印F7层的输出（最后10个神经元的值）
if f7_output.numel() > 0:  # 使用 .numel() 检查 Tensor 是否为空
    print(f7_output[-1, :10])  # 打印 F7 输出的前 10 个值

可能有些输出冗余，都是调试的时候加的，至此pycharm中的工作就做完了，要转到HLS中编程。