test_pairwise_distance.py

# Copyright (c) 2020 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import print_function

import paddle
import paddle.fluid as fluid
import numpy as np
import unittest


def np_pairwise_distance(x, y, p=2.0, epsilon=1e-6, keepdim=False):
    distance = np.linalg.norm(x - y + epsilon, ord=p, axis=-1, keepdims=keepdim)
    # Paddle currently has not supported for 0-d Tensors, so even if keep_dim is False,
    # and neither x nor y is batched, a Tensor of shape (1, ) is returned
    if distance.ndim == 0:
        distance = np.expand_dims(distance, axis=0)
    return distance


def call_pairwise_distance_layer(x, y, p=2., epsilon=1e-6, keepdim='False'):
    pairwise_distance = paddle.nn.PairwiseDistance(p=p,
                                                   epsilon=epsilon,
                                                   keepdim=keepdim)
    distance = pairwise_distance(x=x, y=y)
    return distance


def call_pairwise_distance_functional(x,
                                      y,
                                      p=2.,
                                      epsilon=1e-6,
                                      keepdim='False'):
    distance = paddle.nn.functional.pairwise_distance(x=x,
                                                      y=y,
                                                      p=p,
                                                      epsilon=epsilon,
                                                      keepdim=keepdim)
    return distance


def test_static(place,
                x_np,
                y_np,
                p=2.0,
                epsilon=1e-6,
                keepdim=False,
                functional=False):
    prog = paddle.static.Program()
    startup_prog = paddle.static.Program()
    place = fluid.CUDAPlace(
        0) if paddle.fluid.core.is_compiled_with_cuda() else fluid.CPUPlace()
    paddle.enable_static()
    with paddle.static.program_guard(prog, startup_prog):
        x = paddle.fluid.data(name='x', shape=x_np.shape, dtype=x_np.dtype)
        y = paddle.fluid.data(name='y', shape=y_np.shape, dtype=x_np.dtype)

        if functional:
            distance = call_pairwise_distance_functional(x=x,
                                                         y=y,
                                                         p=p,
                                                         epsilon=epsilon,
                                                         keepdim=keepdim)
        else:
            distance = call_pairwise_distance_layer(x=x,
                                                    y=y,
                                                    p=p,
                                                    epsilon=epsilon,
                                                    keepdim=keepdim)
        exe = paddle.static.Executor(place)
        static_ret = exe.run(prog,
                             feed={
                                 'x': x_np,
                                 'y': y_np
                             },
                             fetch_list=[distance])
        static_ret = static_ret[0]
    paddle.disable_static()
    return static_ret


def test_dygraph(place,
                 x_np,
                 y_np,
                 p=2.0,
                 epsilon=1e-6,
                 keepdim=False,
                 functional=False):
    x = paddle.to_tensor(x_np)
    y = paddle.to_tensor(y_np)
    if functional:
        dy_distance = call_pairwise_distance_functional(x=x,
                                                        y=y,
                                                        p=p,
                                                        epsilon=epsilon,
                                                        keepdim=keepdim)
    else:
        dy_distance = call_pairwise_distance_layer(x=x,
                                                   y=y,
                                                   p=p,
                                                   epsilon=epsilon,
                                                   keepdim=keepdim)
    dygraph_ret = dy_distance.numpy()
    return dygraph_ret


def test_legacy_dygraph(place,
                        x_np,
                        y_np,
                        p=2.0,
                        epsilon=1e-6,
                        keepdim=False,
                        functional=False):
    paddle.fluid.framework._enable_legacy_dygraph()
    x = paddle.to_tensor(x_np)
    y = paddle.to_tensor(y_np)
    if functional:
        legacy_distance = call_pairwise_distance_functional(x=x,
                                                            y=y,
                                                            p=p,
                                                            epsilon=epsilon,
                                                            keepdim=keepdim)
    else:
        legacy_distance = call_pairwise_distance_layer(x=x,
                                                       y=y,
                                                       p=p,
                                                       epsilon=epsilon,
                                                       keepdim=keepdim)
    legacy_ret = legacy_distance.numpy()
    paddle.fluid.framework._disable_legacy_dygraph()
    return legacy_ret


class TestPairwiseDistance(unittest.TestCase):

    def test_pairwise_distance(self):
        epsilon = 1e-6
        all_shape = [[5], [100, 100]]
        dtypes = ['float32', 'float64']
        p_list = [-1, 0, 1, 2, np.inf, -np.inf]
        places = [paddle.CPUPlace()]
        if paddle.device.is_compiled_with_cuda():
            places.append(paddle.CUDAPlace(0))
        keeps = [False, True]
        for place in places:
            for shape in all_shape:
                for dtype in dtypes:
                    for p in p_list:
                        for keepdim in keeps:
                            x_np = np.random.random(shape).astype(dtype)
                            y_np = np.random.random(shape).astype(dtype)

                            static_ret = test_static(place,
                                                     x_np,
                                                     y_np,
                                                     p,
                                                     epsilon=epsilon,
                                                     keepdim=keepdim)
                            dygraph_ret = test_dygraph(place,
                                                       x_np,
                                                       y_np,
                                                       p,
                                                       epsilon=epsilon,
                                                       keepdim=keepdim)
                            legacy_ret = test_legacy_dygraph(place,
                                                             x_np,
                                                             y_np,
                                                             p,
                                                             epsilon=epsilon,
                                                             keepdim=keepdim)
                            excepted_value = np_pairwise_distance(
                                x_np, y_np, p, epsilon=epsilon, keepdim=keepdim)

                            self.assertEqual(static_ret.shape,
                                             excepted_value.shape)
                            self.assertEqual(dygraph_ret.shape,
                                             excepted_value.shape)
                            self.assertEqual(legacy_ret.shape,
                                             excepted_value.shape)

                            self.assertTrue(
                                np.allclose(static_ret, excepted_value))
                            self.assertTrue(
                                np.allclose(dygraph_ret, excepted_value))
                            self.assertTrue(
                                np.allclose(legacy_ret, excepted_value))

                            static_functional_ret = test_static(place,
                                                                x_np,
                                                                y_np,
                                                                p,
                                                                epsilon=epsilon,
                                                                keepdim=keepdim)
                            dygraph_functional_ret = test_dygraph(
                                place,
                                x_np,
                                y_np,
                                p,
                                epsilon=epsilon,
                                keepdim=keepdim)
                            legacy_functional_ret = test_legacy_dygraph(
                                place,
                                x_np,
                                y_np,
                                p,
                                epsilon=epsilon,
                                keepdim=keepdim)

                            self.assertEqual(static_functional_ret.shape,
                                             excepted_value.shape)
                            self.assertEqual(dygraph_functional_ret.shape,
                                             excepted_value.shape)
                            self.assertEqual(legacy_functional_ret.shape,
                                             excepted_value.shape)

                            self.assertTrue(
                                np.allclose(static_functional_ret,
                                            excepted_value))
                            self.assertTrue(
                                np.allclose(dygraph_functional_ret,
                                            excepted_value))
                            self.assertTrue(
                                np.allclose(legacy_functional_ret,
                                            excepted_value))

    def test_pairwise_distance_broadcast_1(self):
        shape_x = [100, 100]
        shape_y = [100, 1]
        epsilon = 1e-6
        keepdim = False
        place = paddle.CPUPlace()
        x_np = np.random.random(shape_x).astype('float32')
        y_np = np.random.random(shape_y).astype('float32')
        static_ret = test_static(place=place,
                                 x_np=x_np,
                                 y_np=y_np,
                                 epsilon=epsilon,
                                 keepdim=keepdim)
        dygraph_ret = test_dygraph(place=place,
                                   x_np=x_np,
                                   y_np=y_np,
                                   epsilon=epsilon,
                                   keepdim=keepdim)
        legacy_ret = test_legacy_dygraph(place=place,
                                         x_np=x_np,
                                         y_np=y_np,
                                         epsilon=epsilon,
                                         keepdim=keepdim)
        excepted_value = np_pairwise_distance(x_np,
                                              y_np,
                                              epsilon=epsilon,
                                              keepdim=keepdim)

        self.assertEqual(static_ret.shape, excepted_value.shape)
        self.assertEqual(dygraph_ret.shape, excepted_value.shape)
        self.assertEqual(legacy_ret.shape, excepted_value.shape)

        self.assertTrue(np.allclose(static_ret, excepted_value))
        self.assertTrue(np.allclose(dygraph_ret, excepted_value))
        self.assertTrue(np.allclose(legacy_ret, excepted_value))

        static_functional_ret = test_static(place=place,
                                            x_np=x_np,
                                            y_np=y_np,
                                            epsilon=epsilon,
                                            keepdim=keepdim,
                                            functional=True)
        dygraph_functional_ret = test_dygraph(place=place,
                                              x_np=x_np,
                                              y_np=y_np,
                                              epsilon=epsilon,
                                              keepdim=keepdim,
                                              functional=True)
        legacy_functional_ret = test_legacy_dygraph(place=place,
                                                    x_np=x_np,
                                                    y_np=y_np,
                                                    epsilon=epsilon,
                                                    keepdim=keepdim,
                                                    functional=True)

        self.assertEqual(static_functional_ret.shape, excepted_value.shape)
        self.assertEqual(dygraph_functional_ret.shape, excepted_value.shape)
        self.assertEqual(legacy_functional_ret.shape, excepted_value.shape)

        self.assertTrue(np.allclose(static_functional_ret, excepted_value))
        self.assertTrue(np.allclose(dygraph_functional_ret, excepted_value))
        self.assertTrue(np.allclose(legacy_functional_ret, excepted_value))

    def test_pairwise_distance_broadcast_2(self):
        shape_x = [100, 100]
        shape_y = [100]
        epsilon = 1e-6
        keepdim = False
        place = paddle.CPUPlace()
        x_np = np.random.random(shape_x).astype('float32')
        y_np = np.random.random(shape_y).astype('float32')
        static_ret = test_static(place=place,
                                 x_np=x_np,
                                 y_np=y_np,
                                 epsilon=epsilon,
                                 keepdim=keepdim)
        dygraph_ret = test_dygraph(place=place,
                                   x_np=x_np,
                                   y_np=y_np,
                                   epsilon=epsilon,
                                   keepdim=keepdim)
        legacy_ret = test_legacy_dygraph(place=place,
                                         x_np=x_np,
                                         y_np=y_np,
                                         epsilon=epsilon,
                                         keepdim=keepdim)
        excepted_value = np_pairwise_distance(x_np,
                                              y_np,
                                              epsilon=epsilon,
                                              keepdim=keepdim)

        self.assertEqual(static_ret.shape, excepted_value.shape)
        self.assertEqual(dygraph_ret.shape, excepted_value.shape)
        self.assertEqual(legacy_ret.shape, excepted_value.shape)

        self.assertTrue(np.allclose(static_ret, excepted_value))
        self.assertTrue(np.allclose(dygraph_ret, excepted_value))
        self.assertTrue(np.allclose(legacy_ret, excepted_value))

        static_functional_ret = test_static(place=place,
                                            x_np=x_np,
                                            y_np=y_np,
                                            epsilon=epsilon,
                                            keepdim=keepdim,
                                            functional=True)
        dygraph_functional_ret = test_dygraph(place=place,
                                              x_np=x_np,
                                              y_np=y_np,
                                              epsilon=epsilon,
                                              keepdim=keepdim,
                                              functional=True)
        legacy_functional_ret = test_legacy_dygraph(place=place,
                                                    x_np=x_np,
                                                    y_np=y_np,
                                                    epsilon=epsilon,
                                                    keepdim=keepdim,
                                                    functional=True)

        self.assertEqual(static_functional_ret.shape, excepted_value.shape)
        self.assertEqual(dygraph_functional_ret.shape, excepted_value.shape)
        self.assertEqual(legacy_functional_ret.shape, excepted_value.shape)

        self.assertTrue(np.allclose(static_functional_ret, excepted_value))
        self.assertTrue(np.allclose(dygraph_functional_ret, excepted_value))
        self.assertTrue(np.allclose(legacy_functional_ret, excepted_value))


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