pytorch

1

2

3

4

5

6
from caffe2.python import core, workspace
7

8
import caffe2.python.hypothesis_test_util as hu
9
import caffe2.python.serialized_test.serialized_test_util as serial
10
import hypothesis.strategies as st
11
import numpy as np
12

13

14
class TestScaleOps(serial.SerializedTestCase):
15
    @serial.given(dim=st.sampled_from([[1, 386, 1], [386, 1, 1],
16
                                       [1, 256, 1], [256, 1, 1],
17
                                       [1024, 256, 1], [1, 1024, 1],
18
                                       [1, 1, 1]]),
19
                    scale=st.floats(0.0, 10.0),
20
                    num_tensors=st.integers(1, 10),
21
                    **hu.gcs)
22
    def test_scale_ops(self, dim, scale, num_tensors, gc, dc):
23
        in_tensors = []
24
        in_tensor_ps = []
25
        out_tensors = []
26
        out_ref_tensors = []
27
        # initialize tensors
28
        for i in range(num_tensors):
29
            tensor = "X_{}".format(i)
30
            X = np.random.rand(*dim).astype(np.float32) - 0.5
31
            in_tensors.append(tensor)
32
            in_tensor_ps.append(X)
33
            out_tensor = "O_{}".format(i)
34
            out_tensors.append(out_tensor)
35
            workspace.FeedBlob(tensor, X, device_option=gc)
36

37
        # run ScaleBlobs operator
38
        scale_blobs_op = core.CreateOperator(
39
            "ScaleBlobs",
40
            in_tensors,
41
            out_tensors,
42
            scale=scale,
43
        )
44
        scale_blobs_op.device_option.CopyFrom(gc)
45
        workspace.RunOperatorOnce(scale_blobs_op)
46

47
        # run Scale op for each tensor and compare with ScaleBlobs
48
        for i in range(num_tensors):
49
            tensor = "X_{}".format(i)
50
            out_ref_tensor = "O_ref_{}".format(i)
51
            scale_op = core.CreateOperator(
52
                "Scale",
53
                [tensor],
54
                [out_ref_tensor],
55
                scale=scale,
56
            )
57
            scale_op.device_option.CopyFrom(gc)
58
            workspace.RunOperatorOnce(scale_op)
59
            o_ref = workspace.FetchBlob(out_ref_tensor)
60
            o = workspace.FetchBlob(out_tensors[i])
61
            np.testing.assert_allclose(o, o_ref)
62

63
if __name__ == '__main__':
64
    import unittest
65

66
    unittest.main()
67