深夜：在？我用本地环境pytest带你玩自定义算子

深夜：在？我用本地环境pytest带你玩自定义算子

module1：

多玩法Python调试框架pytest

初学入门

大家好python通用测试框架的是unittest+HTMLTestRunner，这段时间看到了pytest文档，发现这个框架和丰富的plugins很好用，所以来学习下pytest.

pytest是一个非常成熟的全功能的Python测试框架，主要有以下几个特点：

简单灵活，容易上手

支持参数化

能够支持简单的单元测试和复杂的功能测试，还可以用来做selenium/appnium等自动化测试、接口自动化测试（pytest+requests）

pytest具有很多第三方插件，并且可以自定义扩展，比较好用的如pytest-selenium（集成selenium）、pytest-html（完美html测试报告生成）、pytest-rerunfailures（失败case重复执行）、pytest-xdist（多CPU分发）等

测试用例的skip和xfail处理

可以很好的和jenkins集成

report框架----allure 也支持了pytest

安装pytest：

pip install -U pytest

验证安装的版本：

pytest --version

pytest documentation中的例子:

例1：

首先我们找一个目录，新建文件夹Pytest，然后新建test_sample.py，输入以下内容。

import pytest # content of test_sample.py def func(x): return x + 1 def test_answer(): assert func(3) == 5

如上图，在目录栏输入cmd（也可以直接在IDE中运行）：

回车，打开命令行，输入“pytest”。

pytest返回一个错误报告，因为func(3)不返回5。

例子2：

当需要编写多个测试样例的时候，我们可以将其放到一个测试类当中，新建一个目录test，在目录下新建test_class.py,输入一下内容：

class TestClass: def test_one(self): x = "this" assert 'h' in x def test_two(self): x = "hello" assert hasattr(x, 'check')

pytest -q test_class.py

结果：

从测试结果中可以看到，该测试共执行了两个测试样例，一个失败一个成功。同样，我们也看到失败样例的详细信息，和执行过程中的中间结果。-q即-quiet，作用是减少冗长，具体就是不再展示pytest的版本信息。

如何编写pytest测试样例

通过上面2个实例，我们发现编写pytest测试样例非常简单，只需要按照下面的规则：

测试文件以test_开头（以_test结尾也可以）

测试类以Test开头，并且不能带有 init 方法

测试函数以test_开头

断言使用基本的assert即可

运行模式

Pytest的多种运行模式，让测试和调试变得更加得心应手，下面介绍5种常用的模式。需要注意的是，运行测试的文件是有命名的要求的，否则需要额外的命令行指示。pytest命令会找当前目录及其子目录中的所有test_*.py 或 *_test.py格式的文件以及以test开头的方法或者class，不然就会提示找不到可以运行的case了。

1.运行后生成测试报告（htmlReport）

安装pytest-html：

pip install -U pytest-html

运行模式：

pytest --html=report.html

我们可以看到，目录下生成了html文件，点开看看：

报告效果：

我们可以看到排版非常整洁的错误报告，在以上报告中可以清晰的看到测试结果和错误原因，定位问题很容易。

2.运行指定的case

当我们写了较多的cases时，如果每次都要全部运行一遍，无疑是很浪费时间的，通过指定case来运行就很方便了。

测试用例，新建test2目录，目录下新建test_se.py文件：

class TestClassOne(object): def test_one(self): x = "this" assert 't'in x def test_two(self): x = "hello" assert hasattr(x, 'check') class TestClassTwo(object): def test_one(self): x = "iphone" assert 'p'in x def test_two(self): x = "apple" assert hasattr(x, 'check')

运行模式：

模式1：直接运行test_se.py文件中的所有cases:

pytest test_se.py

模式2：运行test_se.py文件中的TestClassOne这个class下的两个cases:

pytest test_se.py::TestClassOne

模式3：运行test_se.py文件中的TestClassTwo这个class下的test_one:

pytest test_se.py::TestClassTwo::test_one

注意：定义class时，需要以T开头，不然pytest是不会去运行该class的。

我们可以看到，三者的用时是递减的。

3.多进程运行cases

当cases量很多时，运行时间也会变的很长，如果想缩短脚本运行的时长，就可以用多进程来运行。

安装pytest-xdist：

pip install -U pytest-xdist

运行模式：

pytest test_se.py -n NUM

其中NUM填写并发的进程数。

对比最开始，我们这次用两个进程花了0.93s，一个进程花了0.83s，一百个进程花了十多秒。在我们测试代码量不大的情况下，多进程没有什么优势。

4.重试运行cases

在做接口测试时，有事会遇到503或短时的网络波动，导致case运行失败，而这并非是我们期望的结果，此时可以就可以通过重试运行cases的方式来解决。

安装pytest-rerunfailures：

pip install -U pytest-rerunfailures

运行模式：

pytest test_se.py --reruns NUM

NUM填写重试的次数。

5.显示print内容

在运行测试脚本时，为了调试或打印一些内容，我们会在代码中加一些print内容，但是在运行pytest时，这些内容不会显示出来。如果带上-s，就可以显示了。

运行模式：

pytest test_se.py -s

另外，pytest的多种运行模式是可以叠加执行的，比如说，你想同时运行4个进程，又想打印出print的内容。可以用：

pytest test_se.py -s -n 4

module2：

pytest带你玩转mindspore自定义算子

有关算子的前置知识，请阅读官网的参考文档：MindSpore算子

当开发网络遇到内置算子不足以满足需求时，你可以利用MindSpore的Python API和C++ API方便快捷地扩展CPU端的自定义算子。

自定义算子分为三步

算子原语：定义了算子在网络中的前端接口原型，也是组成网络模型的基础单元，主要包括算子的名称、属性（可选）、输入输出名称、输出shape推理方法、输出dtype推理方法等信息。

算子实现：利用框架提供的C++ API，结合算子具体特性实现算子内部计算逻辑。

算子信息：描述CPU算子的基本信息，如算子名称、支持的输入输出类型等。它是后端做算子选择和映射时的依据。

下面我们以自定义Transpose算子为例，介绍自定义算子和利用pytest进行测试。

注册算子原语

算子的原语是一个继承于PrimitiveWithInfer的子类，其类型名称即是算子名称。

CPU算子原语定义在mindspore/ops/operations路径下，根据算子类型选择适合的文件，接口定义如下：

属性由构造函数__init__的入参定义。本用例的算子没有init属性，因此__init__没有额外的入参。

输入输出的名称通过init_prim_io_names函数定义。

输出Tensor的shape和dtype(数据类型对象，可以参考本人这篇文章)检验在__infer__函数中实现。

以Transpose算子原语为例，给出如下示例代码，这里用到了装饰器语法糖。

Transpose算子原语中参数“perm”作为输入传入，但是在解析时元组类型的“perm”实际被认为是算子的属性。

from mindspore.ops import PrimitiveWithInfer class Transpose(PrimitiveWithInfer): """ The definition of the Transpose primitive. """ @prim_attr_register def __init__(self): """Initialize Transpose""" self.init_prim_io_names(inputs=['x', 'perm'], outputs=['output']) def __infer__(self, x, perm): x_shape = x['shape'] p_value = perm['value'] if len(x_shape) != len(p_value): raise ValueError('The dimension of x and perm must be equal.') out_shapes = [] for i in p_value: out_shapes.append(x_shape[i]) out = {'shape': tuple(out_shapes), 'dtype': x['dtype'], 'value': None} return out

实现CPU算子和注册算子信息（此处仅供简单了解，详见官网）

实现CPU算子

通常一个CPU算子的实现，需要编写一个头文件和一个源文件，文件路径为mindspore/ccsrc/backend/kernel_compiler/cpu，如果算子的逻辑实现是通过调用第三方库MKL-DNN，则放在子目录mkldnn下。详细介绍请参考oneMKL和oneDNN 。

算子的头文件中包括算子的注册信息和类的声明。算子类继承于CPUKernel父类，重载InitKernel和Launch两个成员函数。

算子的源文件是类的实现，主要是重载InitKernel和Launch两个函数，InitKernel中AnfRuntimeAlgorithm类中的函数实现了各种对算子节点的操作，shape_表示算子第1个输入的shape，axis_表示算子的属性perm。

AnfRuntimeAlgorithm类的详细内容可参考MindSpore源码中mindspore/ccsrc/backend/session/anf_runtime_algorithm.h下的声明。

源文件中Launch函数首先依次获取每个输入输出的地址，然后根据axis_变换维度，把值赋给输出地址指向的空间。

这里对实现过程就不详细阐述了。

注册算子信息

算子信息是指导后端选择算子实现的关键信息，MS_REG_CPU_KERNEL中第一个参数是注册算子的名称，和原语中算子名称一致，第二个参数依次指明每个输入输出的类型，最后一个参数是算子实现的类名。Transpose算子注册代码如下：

MS_REG_CPU_KERNEL(Transpose, KernelAttr().AddInputAttr(kNumberTypeFloat32).AddOutputAttr(kNumberTypeFloat32), TransposeCPUFwdKernel);

算子信息中定义输入输出信息的个数和顺序、算子实现中的输入输出信息的个数和顺序、算子原语中输入输出名称列表的个数和顺序，三者要完全一致。

编译MindSpore

写好自定义CPU算子后，需要重新编译安装MindSpore，具体请参考安装文档。

使用自定义CPU算子并用pytest测试

编译并安装完成后，自定义CPU算子可以通过导入原语直接使用。下面以Transpose的单算子网络测试为例进行说明。

在test_transpose.py文件中定义网络。

import numpy as np import mindspore.nn as nn import mindspore.context as context from mindspore import Tensor import mindspore.ops as ops context.set_context(mode=context.GRAPH_MODE, device_target="CPU") class Net(nn.Cell): def __init__(self): super(Net, self).__init__() self.transpose = ops.Transpose() def construct(self, data): return self.transpose(data, (1, 0)) def test_net(): x = np.arange(2 * 3).reshape(2, 3).astype(np.float32) transpose = Net() output = transpose(Tensor(x)) print("output: ", output)

执行用例（这里我们可以看到，利用pytest是测试其中一个类中的一个函数，并且均以test_开头，可以回到上文的pytest用法复习一下）:

pytest -s test_transpose.py::test_net

执行结果:

output: [[0, 3] [1, 4] [2, 5]]

我们把报告输出html：

pytest --html=report.html -s test_transpose.py::test_net

可以查看html报告。

定义算子反向传播函数

如果算子要支持自动微分，需要在其原语中定义其反向传播函数（bprop）。你需要在bprop中描述利用正向输入、正向输出和输出梯度得到输入梯度的反向计算逻辑。反向计算逻辑可以使用内置算子或自定义反向算子构成。

定义算子反向传播函数时需注意以下几点：

bprop函数的入参顺序约定为正向的输入、正向的输出、输出梯度。若算子为多输出算子，正向输出和输出梯度将以元组的形式提供。

bprop函数的返回值形式约定为输入梯度组成的元组，元组中元素的顺序与正向输入参数顺序一致。即使只有一个输入梯度，返回值也要求是元组的形式。

例如，Transpose的反向原语为：

import mindspore as ms import mindspore.ops as ops from mindspore.ops._grad.grad_base import bprop_getters fill = ops.Fill() invert_permutation = ops.InvertPermutation() transpose = ops.Transpose() @bprop_getters.register(ops.Transpose) def get_bprop_transpose(self): """Generate bprop for Transpose""" def bprop(x, perm, out, dout): return transpose(dout, invert_permutation(perm)), fill(ms.int32, (len(perm), ), 0) return bprop

Transpose反向算子中用到了InvertPermutation算子，该算子和Transpose算子开发一样，需要有算子的原语，注册，实现等完整的流程。

在test_transpose.py文件中加入一下内容，定义反向用例。

import mindspore.ops as ops class Grad(nn.Cell): def __init__(self, network): super(Grad, self).__init__() self.grad = ops.GradOperation(sens_param=True) self.network = network def construct(self, input_data, sens): gout = self.grad(self.network)(input_data, sens) return gout def test_grad_net(): x = np.arange(2 * 3).reshape(2, 3).astype(np.float32) sens = np.arange(2 * 3).reshape(3, 2).astype(np.float32) grad = Grad(Net()) dx = grad(Tensor(x), Tensor(sens)) print("dx: ", dx.asnumpy())

执行用例:

pytest -s test_transpose.py::test_grad_net

执行结果:

dx: [[0. 2. 4.] [1. 3. 5.]]

开源代码

qmckw/CPUcustom_opByPytest (gitee.com)

参考资料

pytest documentation

好用的Pytest单元测试框架（《51测试天地》四十九（下）- 44）

Pytest学习笔记

pytest单元测试框架

MindSpore参考文档

MindSpore Python 单元测试

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。