test_where_index.py

#   Copyright (c) 2019 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.

import unittest

import numpy as np
from op_test import OpTest

import paddle.fluid as fluid
import paddle.fluid.core as core
from paddle.fluid import Program, program_guard
from paddle.fluid.op import Operator


class TestWhereIndexOp(OpTest):
    def setUp(self):
        self.op_type = "where_index"
        self.init_config()

    def test_check_output(self):
        self.check_output()

    def init_config(self):
        self.inputs = {
            'Condition': np.array([True, False, True]),
        }

        self.outputs = {'Out': np.array([[0], [2]], dtype='int64')}


class TestAllFalse(unittest.TestCase):
    def setUp(self):
        self.op_type = "where_index"
        self.init_config()

    def check_with_place(self, place):
        scope = core.Scope()
        condition = scope.var('Condition').get_tensor()
        condition.set(self.cond_data, place)

        out = scope.var("Out").get_tensor()
        out.set(np.full(self.shape, 0).astype('int64'), place)

        op = Operator("where_index", Condition="Condition", Out="Out")
        op.run(scope, place)

        out_array = np.array(out)
        self.assertTrue((out_array == self.out_data).all())

    def init_config(self):
        self.cond_data = np.array([False, False, False])
        self.shape = (3, 1)
        self.out_data = np.array([], dtype='int64')

    def test_all_false(self):
        self.check_with_place(core.CPUPlace())

        if core.is_compiled_with_cuda():
            self.check_with_place(core.CUDAPlace(0))


class TestRank2(TestWhereIndexOp):
    def init_config(self):
        self.inputs = {
            'Condition': np.array([[True, False], [False, True]]),
        }

        self.outputs = {'Out': np.array([[0, 0], [1, 1]], dtype='int64')}


class TestRank3(TestWhereIndexOp):
    def init_config(self):
        self.inputs = {
            'Condition': np.array(
                [
                    [[True, False], [False, True]],
                    [[False, True], [True, False]],
                    [[False, False], [False, True]],
                ]
            ),
        }

        self.outputs = {
            'Out': np.array(
                [[0, 0, 0], [0, 1, 1], [1, 0, 1], [1, 1, 0], [2, 1, 1]],
                dtype='int64',
            )
        }


class TestWhereOpError(unittest.TestCase):
    def test_api(self):
        with program_guard(Program(), Program()):
            cond = fluid.layers.data(name='cond', shape=[4], dtype='bool')
            result = fluid.layers.where(cond)

            exe = fluid.Executor(fluid.CPUPlace())
            exe.run(fluid.default_startup_program())
            cond_i = np.array([True, False, False, False]).astype("bool")
            out = exe.run(fluid.default_main_program(), feed={'cond': cond_i})


class TestWhereRaiseError(unittest.TestCase):
    def test_errors(self):
        def test_type():
            fluid.layers.where([10])

        self.assertRaises(TypeError, test_type)


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