mindspore的softmax回归模型

1、导入模块

import mindspore
import os
import struct
import numpy as np
import mindspore.dataset as ds
from mindspore import nn
from mindspore.common.initializer import Normal
import mindspore.dataset.vision.c_transforms as cv
import mindspore.dataset.transforms.c_transforms as C
from IPython import display
from matplotlib import pyplot as plt

2、下载数据集

将下载好的Fashion-MNIST数据集放在“…\data”文件夹中

def load_mnist(path, kind='train'):
    """导入已下载的数据集"""
    #os.path.join()函数用于路径拼接文件路径
    labels_path = os.path.join(path,'%s-labels-idx1-ubyte'% kind)
    images_path = os.path.join(path,'%s-images-idx3-ubyte'% kind)
    
    with open(labels_path, 'rb') as lbpath:
        magic, n = struct.unpack('>II',
                                 lbpath.read(8))
        labels = np.fromfile(lbpath,
                             dtype=np.uint8)

    with open(images_path, 'rb') as imgpath:
        magic, num, rows, cols = struct.unpack('>IIII',
                                               imgpath.read(16))
        images = np.fromfile(imgpath,
                             dtype=np.uint8).reshape(len(labels),28,28,1)   # 设置图像形状，高度宽度均为28，通道数为1

    return images, labels



class FashionMnist():
    """创建一个迭代器类,作为GeneratorDataset的数据源"""
    def __init__(self, path, kind):
        self.data, self.label = load_mnist(path, kind)
    
    def __getitem__(self, index):
        return self.data[index], self.label[index]
    
    def __len__(self):
        return len(self.data)
    
    
    
def load_data_fashion_mnist(data_path, batch_size, resize=None, works=1):  
    """将Fashion-MNIST数据集加载到内存中。"""
    mnist_train = FashionMnist(data_path, kind='train')
    mnist_test = FashionMnist(data_path, kind='t10k')

    mnist_train = ds.GeneratorDataset(source=mnist_train, column_names=['image', 'label'], 
                                      shuffle=False,python_multiprocessing=False)
    mnist_test = ds.GeneratorDataset(source=mnist_test, column_names=['image', 'label'], 
                                     shuffle=False,python_multiprocessing=False)
    
    # 数据变换
    trans = [cv.Rescale(1.0 / 255.0, 0), cv.HWC2CHW()]   # 调整图像的像素大小Rescale变换用于调整图像像素值的大小，包括两个参数：
                                                         # rescale：缩放因子。shift：平移因子。图像的每个像素将根据这两个参数
                                                         # 进行调整，输出的像素值为outputi=inputi∗rescale+shift
                                                         # HWC2CWH变换用于转换图像格式，(height, width, channel)转为
                                                         # (channel, height, width)
    type_cast_op = C.TypeCast(mindspore.int32)           # 将输入的Tensor转换为指定的数据类型
    if resize:
        trans.insert(0, cv.Resize(resize))               # 调整为给定的尺寸大小
    mnist_train = mnist_train.map(trans, input_columns=["image"])
    mnist_test = mnist_test.map(trans, input_columns=["image"])
    mnist_train = mnist_train.map(type_cast_op, input_columns=['label'])
    mnist_test = mnist_test.map(type_cast_op, input_columns=['label'])
    
    mnist_train = mnist_train.batch(batch_size, num_parallel_workers=works)
    mnist_test = mnist_test.batch(batch_size, num_parallel_workers=works)
    
    return mnist_train, mnist_test

batch_size = 256
mnist_train, mnist_test = load_data_fashion_mnist('../data' ,batch_size)

3、构建模型

# 骨干网络模型
net = nn.SequentialCell([nn.Flatten(), nn.Dense(784, 10, weight_init=Normal(0.01, 0), bias_init='zero')])  
# nn.Flatten将输入的X维度从[256,1,28,28]变成[256,784]，则一个样本数据成一行

# 损失函数SoftmaxCrossEntropyWithLogits，交叉熵损失函数中传递未规范化的预测，并同时计算softmax及其损失
loss = nn.SoftmaxCrossEntropyWithLogits(sparse=True, reduction='mean')
# 这里需要设置reduction为mean，才能实现gradient做sum和div(batch_size)的相同运算

# 优化器SGD，学习率为0.1的随机梯度下降
optim = nn.SGD(net.trainable_params(), learning_rate=0.1)

class Accumulator:  
    """累加器"""
    def __init__(self, n):
        self.data = [0.0] * n

    def add(self, *args):
        self.data = [a + float(b) for a, b in zip(self.data, args)]

    def reset(self):
        self.data = [0.0] * len(self.data)

    def __getitem__(self, idx):
        return self.data[idx]

def accuracy(y_hat, y):  
    """计算预测正确的数量"""
    if len(y_hat.shape) > 1 and y_hat.shape[1] > 1:   # 判断y_hat是不是矩阵
        y_hat = y_hat.argmax(axis=1)                  # 得到每样本预测概率最大所属分类的下标
    cmp = y_hat.asnumpy() == y.asnumpy()              # y_hat.asnumpy() == y.asnumpy()返回的是一个布尔数组
    return float(cmp.sum())



def evaluate_accuracy(net, data_iter):  
    """计算在指定数据集上模型的精度"""
    metric = Accumulator(2)         # 累加器,metric[0]记录正确预测数，metric[1]记录预测总数
    for X, y in data_iter:
        metric.add(accuracy(net(X), y), y.size)
    return metric[0] / metric[1]    # 正确预测数 / 预测总数

def train_epoch(net, train_iter, loss, optim):  
    """训练模型一个迭代周期"""
    net_with_loss = nn.WithLossCell(net, loss)                # 将net与loss连接
    net_train = nn.TrainOneStepCell(net_with_loss, optim)     # 将net,loss,optim连接,生成训练模型
    metric = Accumulator(3)
    for X, y in train_iter:
        l = net_train(X, y)
        y_hat = net(X)
        metric.add(float(l.sum().asnumpy()), accuracy(y_hat, y), y.size)
    return metric[0] / metric[2], metric[1] / metric[2]       # 误差 / 预测总数 ，正确预测数 / 预测总数

def trainer(net, train_iter, test_iter, loss, num_epochs, optim):  
    """训练模型"""
    # 动画设置
    animator = Animator(xlabel='epoch', xlim=[1, num_epochs], ylim=[0.3, 0.9],
                        legend=['train loss', 'train acc', 'test acc'])
    
    for epoch in range(num_epochs):
        train_metrics = train_epoch(net, train_iter, loss, optim)
        test_acc = evaluate_accuracy(net, test_iter)
        animator.add(epoch + 1, train_metrics + (test_acc,))
    train_loss, train_acc = train_metrics
    
    # 检测
    assert train_loss < 0.6, train_loss
    assert train_acc <= 1 and train_acc > 0.7, train_acc
    assert test_acc <= 1 and test_acc > 0.7, test_acc

def set_axes(axes, xlabel, ylabel, xlim, ylim, xscale, yscale, legend):
    """设置matplotlib的轴。"""
    axes.set_xlabel(xlabel)
    axes.set_ylabel(ylabel)
    axes.set_xscale(xscale)
    axes.set_yscale(yscale)
    axes.set_xlim(xlim)
    axes.set_ylim(ylim)
    if legend:
        axes.legend(legend)
    axes.grid()

class Animator:  
    """在动画中绘制数据"""
    def __init__(self, xlabel=None, ylabel=None, legend=None, xlim=None,
                 ylim=None, xscale='linear', yscale='linear',
                 fmts=('-', 'm--', 'g-.', 'r:'), nrows=1, ncols=1,
                 figsize=(3.5, 2.5)):
        if legend is None:
            legend = []
        display.set_matplotlib_formats('svg')
        self.fig, self.axes =plt.subplots(nrows, ncols, figsize=figsize)
        if nrows * ncols == 1:
            self.axes = [self.axes, ]
        self.config_axes = lambda: set_axes(
            self.axes[0], xlabel, ylabel, xlim, ylim, xscale, yscale, legend)
        self.X, self.Y, self.fmts = None, None, fmts

    def add(self, x, y):
        if not hasattr(y, "__len__"):
            y = [y]
        n = len(y)
        if not hasattr(x, "__len__"):
            x = [x] * n
        if not self.X:
            self.X = [[] for _ in range(n)]
        if not self.Y:
            self.Y = [[] for _ in range(n)]
        for i, (a, b) in enumerate(zip(x, y)):
            if a is not None and b is not None:
                self.X[i].append(a)
                self.Y[i].append(b)
        self.axes[0].cla()
        for x, y, fmt in zip(self.X, self.Y, self.fmts):
            self.axes[0].plot(x, y, fmt)
        self.config_axes()
        display.display(self.fig)
        display.clear_output(wait=True)

4、训练模型

num_epochs = 10
trainer(net, mnist_train, mnist_test, loss, num_epochs, optim)