深夜：在？我用本地環境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小時內刪除侵權內容。

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。