caffe2/python/ideep/copy_op_test.py

import unittest
import numpy as np
from random import randint
from caffe2.proto import caffe2_pb2
from caffe2.python import core, workspace


@unittest.skipIf(not workspace.C.use_mkldnn, "No MKLDNN support.")
class CopyTest(unittest.TestCase):
    def _get_deep_device(self):
        return caffe2_pb2.DeviceOption(device_type=caffe2_pb2.IDEEP)

    def test_copy_to_ideep(self):
        op = core.CreateOperator(
            "CopyCPUToIDEEP",
            ["X"],
            ["X_ideep"],
        )
        op.device_option.CopyFrom(self._get_deep_device())
        n = randint(1, 128)
        c = randint(1, 64)
        h = randint(1, 128)
        w = randint(1, 128)
        X = np.random.rand(n, c, h, w).astype(np.float32)
        workspace.FeedBlob("X", X)
        workspace.RunOperatorOnce(op)
        X_ideep = workspace.FetchBlob("X_ideep")
        np.testing.assert_allclose(X, X_ideep)

    def test_copy_to_ideep_zero_dim(self):
        op = core.CreateOperator(
            "CopyCPUToIDEEP",
            ["X"],
            ["X_ideep"],
        )
        op.device_option.CopyFrom(self._get_deep_device())
        n = 0
        c = randint(1, 128)
        X = np.random.rand(n, c).astype(np.float32)
        workspace.FeedBlob("X", X)
        workspace.RunOperatorOnce(op)
        X_ideep = workspace.FetchBlob("X_ideep")
        np.testing.assert_allclose(X, X_ideep)

    def test_copy_from_ideep(self):
        op = core.CreateOperator(
            "CopyIDEEPToCPU",
            ["X_ideep"],
            ["X"],
        )
        op.device_option.CopyFrom(self._get_deep_device())
        n = randint(1, 128)
        c = randint(1, 64)
        h = randint(1, 128)
        w = randint(1, 128)
        X = np.random.rand(n, c, h, w).astype(np.float32)
        workspace.FeedBlob("X_ideep", X, self._get_deep_device())
        workspace.RunOperatorOnce(op)
        X_ideep = workspace.FetchBlob("X")
        np.testing.assert_allclose(X, X_ideep)

    def test_copy_from_ideep_zero_dim(self):
        op = core.CreateOperator(
            "CopyIDEEPToCPU",
            ["X_ideep"],
            ["X"],
        )
        op.device_option.CopyFrom(self._get_deep_device())
        n = 0
        c = randint(1, 64)
        X = np.random.rand(n, c).astype(np.float32)
        workspace.FeedBlob("X_ideep", X, self._get_deep_device())
        workspace.RunOperatorOnce(op)
        X_ideep = workspace.FetchBlob("X")
        np.testing.assert_allclose(X, X_ideep)

    def test_copy_from_ideep_fallthrough(self):
        op = core.CreateOperator(
            "CopyIDEEPToCPU",
            ["X_ideep"],
            ["X"],)
        op.device_option.CopyFrom(self._get_deep_device())
        n = randint(1, 128)
        c = randint(1, 64)
        h = randint(1, 128)
        w = randint(1, 128)
        X = np.random.rand(n, c, h, w).astype(np.float32)
        workspace.FeedBlob("X_ideep", X)
        workspace.RunOperatorOnce(op)
        X_ideep = workspace.FetchBlob("X")
        np.testing.assert_allclose(X, X_ideep)

if __name__ == "__main__":
    unittest.main()