<commit_before>import unittest

import numpy
import six

import chainer
from chainer.backends import cuda
from chainer import functions
from chainer import gradient_check
from chainer import testing
from chainer.testing import attr
from chainer.testing import backend
import chainerx


def _to_fcontiguous(arrays):
    xp = chainer.backend.get_array_module(*arrays)
    return [xp.asfortranarray(a) for a in arrays]


@testing.parameterize(*testing.product({
    'cover_all': [True, False],
    'dtype': [numpy.float16, numpy.float32, numpy.float64],
    'c_contiguous': [True, False],
}))
@backend.inject_backend_tests(
    ['test_forward',
     'test_forward_output_size_zero',
     'test_backward',
     'test_double_backward'],
    # CPU tests
    testing.product({
        'use_cuda': [False],
        'use_ideep': ['never', 'always'],
    })
    # GPU tests
    + testing.product({
        'use_cuda': [True],
        'use_cudnn': ['never', 'always'],
    })
    # ChainerX tests
    + testing.product({
        'use_chainerx': [True],
        'chainerx_device': ['native:0', 'cuda:0'],
    }))
class TestMaxPooling2D(unittest.TestCase):

    def setUp(self):
        dtype = self.dtype

        # Avoid unstability of numerical gradient
        x = numpy.arange(2 * 3 * 4 * 3, dtype=dtype).reshape(2, 3, 4, 3)
        numpy.random.shuffle(x)
        x = 2 * x / x.size - 1
        if self.cover_all:
            gy = numpy.random.uniform(-1, 1, (2, 3, 3, 2)).astype(dtype)
        else:
            gy = numpy.random.uniform(-1, 1, (2, 3, 2, 2)).astype(dtype)
        ggx = numpy.random.uniform(-1, 1, (2, 3, 4, 3)).astype(dtype)

        self.output_shape = gy.shape

        self.inputs = [x]
        self.grad_outputs = [gy]
        self.grad_grad_inputs = [ggx]

        if self.dtype == numpy.float16:
            self.check_backward_options = {
                'atol': 1e-3, 'rtol': 1e-2}
            self.check_double_backward_options = {
                'atol': 1e-3, 'rtol': 1e-2}
        else:
            self.check_backward_options = {
                'atol': 1e-4, 'rtol': 1e-3}
            self.check_double_backward_options = {
                'atol': 1e-4, 'rtol': 1e-3}

    def forward_cpu(self, inputs):
        x, = inputs
        expect = numpy.empty(self.output_shape, dtype=self.dtype)
        for i in six.moves.range(2):
            for c in six.moves.range(3):
                xx = x[i, c]
                if self.cover_all:
                    expect[i, c] = numpy.array([
                        [xx[0:2, 0:2].max(), xx[0:2, 1:3].max()],
                        [xx[1:4, 0:2].max(), xx[1:4, 1:3].max()],
                        [xx[3:4, 0:2].max(), xx[3:4, 1:3].max()]])
                else:
                    expect[i, c] = numpy.array([
                        [xx[0:2, 0:2].max(), xx[0:2, 1:3].max()],
                        [xx[1:4, 0:2].max(), xx[1:4, 1:3].max()]])
        return expect,

    def check_forward(self, inputs, backend_config):
        # TODO(sonots): Support it
        if backend_config.use_chainerx and self.dtype == numpy.float16:
            raise unittest.SkipTest('ChainerX does not support float16')

        y_expect, = self.forward_cpu(inputs)

        # TODO(sonots): Cleanup to use testing.backend.get_array after
        # chainerx.asfortranarray is implemented.
        if (backend_config.use_cuda
            or (backend_config.use_chainerx
                and backend_config.chainerx_device.startswith('cuda:'))):
            inputs = cuda.to_gpu(inputs)
        if not self.c_contiguous:
            inputs = _to_fcontiguous(inputs)
        if backend_config.use_chainerx:
            inputs = chainer.backend.to_chainerx(inputs)

        with backend_config:
            x, = inputs
            y = functions.max_pooling_2d(x, 3, stride=2, pad=1,
                                         cover_all=self.cover_all)
        assert self.dtype == y.data.dtype
        assert self.output_shape == y.data.shape
        testing.assert_allclose(y_expect, y.data)

    def test_forward(self, backend_config):
        self.check_forward(self.inputs, backend_config)

    def test_forward_cpu_wide(self):  # see #120
        x_data = numpy.random.rand(2, 3, 15, 15).astype(self.dtype)
        x = chainer.Variable(x_data)
        functions.max_pooling_2d(x, 6, stride=6, pad=0)

    def test_forward_output_size_zero(self, backend_config):
        if backend_config.use_chainerx:
            # TODO(sonots): Support it
            if self.dtype == numpy.float16:
                raise unittest.SkipTest('ChainerX does not support float16')

        try:
            x = numpy.random.rand(4, 4, 1, 4).astype(self.dtype)
            # TODO(sonots): Cleanup to use testing.backend.get_array after
            # chainerx.asfortranarray is implemented.
            if (backend_config.use_cuda
                or (backend_config.use_chainerx
                    and backend_config.chainerx_device.startswith('cuda:'))):
                x = cuda.to_gpu(x)
            if backend_config.use_chainerx:
                x = chainer.backend.to_chainerx(x)
            x = chainer.Variable(x)
            with backend_config:
                functions.max_pooling_2d(x, 3, stride=2)

            unittest.fail('should raise AssertionError or chainerx.DimensionError')
        except Exception as e:
            assert isinstance(e, (AssertionError, chainerx.DimensionError))
            if isinstance(e, AssertionError):
                assert str(e) == 'Height in the output should be positive.'
            else:
                assert str(e) == 'Output size should be positive.'

        try:
            x = numpy.random.rand(4, 4, 4, 1).astype(self.dtype)
            # TODO(sonots): Cleanup to use testing.backend.get_array after
            # chainerx.asfortranarray is implemented.
            if (backend_config.use_cuda
                or (backend_config.use_chainerx
                    and backend_config.chainerx_device.startswith('cuda:'))):
                x = cuda.to_gpu(x)
            if backend_config.use_chainerx:
                x = chainer.backend.to_chainerx(x)
            x = chainer.Variable(x)
            with backend_config:
                functions.max_pooling_2d(x, 3, stride=2)

            unittest.fail('should raise AssertionError or chainerx.DimensionError')
        except Exception as e:
            assert isinstance(e, (AssertionError, chainerx.DimensionError))
            if isinstance(e, AssertionError):
                assert str(e) == 'Width in the output should be positive.'
            else:
                assert str(e) == 'Output size should be positive.'

    def check_backward(self, inputs, grad_outputs, backend_config):
        # TODO(sonots): Support it
        if backend_config.use_chainerx and self.dtype == numpy.float16:
            raise unittest.SkipTest('ChainerX does not support float16')

        # TODO(sonots): Cleanup to use testing.backend.get_array after
        # chainerx.asfortranarray is implemented.
        if (backend_config.use_cuda
            or (backend_config.use_chainerx
                and backend_config.chainerx_device.startswith('cuda:'))):
            inputs = cuda.to_gpu(inputs)
            grad_outputs = cuda.to_gpu(grad_outputs)
        if not self.c_contiguous:
            inputs = _to_fcontiguous(inputs)
            grad_outputs = _to_fcontiguous(grad_outputs)
        if backend_config.use_chainerx:
            inputs = chainer.backend.to_chainerx(inputs)
            grad_outputs = chainer.backend.to_chainerx(grad_outputs)

        def f(x):
            return functions.max_pooling_2d(
                x, 3, stride=2, pad=1, cover_all=self.cover_all)

        with backend_config:
            gradient_check.check_backward(
                f, inputs, grad_outputs, dtype='d',
                **self.check_backward_options)

    def test_backward(self, backend_config):<commit_msg>Simplified test_forward_output_size_zero<commit_after>import unittest

import numpy
import six

import chainer
from chainer.backends import cuda
from chainer import functions
from chainer import gradient_check
from chainer import testing
from chainer.testing import attr
from chainer.testing import backend


def _to_fcontiguous(arrays):
    xp = chainer.backend.get_array_module(*arrays)
    return [xp.asfortranarray(a) for a in arrays]


@testing.parameterize(*testing.product({
    'cover_all': [True, False],
    'dtype': [numpy.float16, numpy.float32, numpy.float64],
    'c_contiguous': [True, False],
}))
@backend.inject_backend_tests(
    ['test_forward',
     'test_forward_output_size_zero',
     'test_backward',
     'test_double_backward'],
    # CPU tests
    testing.product({
        'use_cuda': [False],
        'use_ideep': ['never', 'always'],
    })
    # GPU tests
    + testing.product({
        'use_cuda': [True],
        'use_cudnn': ['never', 'always'],
    })
    # ChainerX tests
    + testing.product({
        'use_chainerx': [True],
        'chainerx_device': ['native:0', 'cuda:0'],
    }))
class TestMaxPooling2D(unittest.TestCase):

    def setUp(self):
        dtype = self.dtype

        # Avoid unstability of numerical gradient
        x = numpy.arange(2 * 3 * 4 * 3, dtype=dtype).reshape(2, 3, 4, 3)
        numpy.random.shuffle(x)
        x = 2 * x / x.size - 1
        if self.cover_all:
            gy = numpy.random.uniform(-1, 1, (2, 3, 3, 2)).astype(dtype)
        else:
            gy = numpy.random.uniform(-1, 1, (2, 3, 2, 2)).astype(dtype)
        ggx = numpy.random.uniform(-1, 1, (2, 3, 4, 3)).astype(dtype)

        self.output_shape = gy.shape

        self.inputs = [x]
        self.grad_outputs = [gy]
        self.grad_grad_inputs = [ggx]

        if self.dtype == numpy.float16:
            self.check_backward_options = {
                'atol': 1e-3, 'rtol': 1e-2}
            self.check_double_backward_options = {
                'atol': 1e-3, 'rtol': 1e-2}
        else:
            self.check_backward_options = {
                'atol': 1e-4, 'rtol': 1e-3}
            self.check_double_backward_options = {
                'atol': 1e-4, 'rtol': 1e-3}

    def forward_cpu(self, inputs):
        x, = inputs
        expect = numpy.empty(self.output_shape, dtype=self.dtype)
        for i in six.moves.range(2):
            for c in six.moves.range(3):
                xx = x[i, c]
                if self.cover_all:
                    expect[i, c] = numpy.array([
                        [xx[0:2, 0:2].max(), xx[0:2, 1:3].max()],
                        [xx[1:4, 0:2].max(), xx[1:4, 1:3].max()],
                        [xx[3:4, 0:2].max(), xx[3:4, 1:3].max()]])
                else:
                    expect[i, c] = numpy.array([
                        [xx[0:2, 0:2].max(), xx[0:2, 1:3].max()],
                        [xx[1:4, 0:2].max(), xx[1:4, 1:3].max()]])
        return expect,

    def check_forward(self, inputs, backend_config):
        # TODO(sonots): Support it
        if backend_config.use_chainerx and self.dtype == numpy.float16:
            raise unittest.SkipTest('ChainerX does not support float16')

        y_expect, = self.forward_cpu(inputs)

        # TODO(sonots): Cleanup to use testing.backend.get_array after
        # chainerx.asfortranarray is implemented.
        if (backend_config.use_cuda
            or (backend_config.use_chainerx
                and backend_config.chainerx_device.startswith('cuda:'))):
            inputs = cuda.to_gpu(inputs)
        if not self.c_contiguous:
            inputs = _to_fcontiguous(inputs)
        if backend_config.use_chainerx:
            inputs = chainer.backend.to_chainerx(inputs)

        with backend_config:
            x, = inputs
            y = functions.max_pooling_2d(x, 3, stride=2, pad=1,
                                         cover_all=self.cover_all)
        assert self.dtype == y.data.dtype
        assert self.output_shape == y.data.shape
        testing.assert_allclose(y_expect, y.data)

    def test_forward(self, backend_config):
        self.check_forward(self.inputs, backend_config)

    def test_forward_cpu_wide(self):  # see #120
        x_data = numpy.random.rand(2, 3, 15, 15).astype(self.dtype)
        x = chainer.Variable(x_data)
        functions.max_pooling_2d(x, 6, stride=6, pad=0)

    def test_forward_output_size_zero(self, backend_config):
        if backend_config.use_chainerx:
            # TODO(sonots): Support it
            if self.dtype == numpy.float16:
                raise unittest.SkipTest('ChainerX does not support float16')

        with self.assertRaises(Exception):
            x = numpy.random.rand(4, 4, 1, 4).astype(self.dtype)
            # TODO(sonots): Cleanup to use testing.backend.get_array after
            # chainerx.asfortranarray is implemented.
            if (backend_config.use_cuda
                or (backend_config.use_chainerx
                    and backend_config.chainerx_device.startswith('cuda:'))):
                x = cuda.to_gpu(x)
            if backend_config.use_chainerx:
                x = chainer.backend.to_chainerx(x)
            x = chainer.Variable(x)
            with backend_config:
                functions.max_pooling_2d(x, 3, stride=2)

        with self.assertRaises(Exception):
            x = numpy.random.rand(4, 4, 4, 1).astype(self.dtype)
            # TODO(sonots): Cleanup to use testing.backend.get_array after
            # chainerx.asfortranarray is implemented.
            if (backend_config.use_cuda
                or (backend_config.use_chainerx
                    and backend_config.chainerx_device.startswith('cuda:'))):
                x = cuda.to_gpu(x)
            if backend_config.use_chainerx:
                x = chainer.backend.to_chainerx(x)
            x = chainer.Variable(x)
            with backend_config:
                functions.max_pooling_2d(x, 3, stride=2)

    def check_backward(self, inputs, grad_outputs, backend_config):
        # TODO(sonots): Support it
        if backend_config.use_chainerx and self.dtype == numpy.float16:
            raise unittest.SkipTest('ChainerX does not support float16')

        # TODO(sonots): Cleanup to use testing.backend.get_array after
        # chainerx.asfortranarray is implemented.
        if (backend_config.use_cuda
            or (backend_config.use_chainerx
                and backend_config.chainerx_device.startswith('cuda:'))):
            inputs = cuda.to_gpu(inputs)
            grad_outputs = cuda.to_gpu(grad_outputs)
        if not self.c_contiguous:
            inputs = _to_fcontiguous(inputs)
            grad_outputs = _to_fcontiguous(grad_outputs)
        if backend_config.use_chainerx:
            inputs = chainer.backend.to_chainerx(inputs)
            grad_outputs = chainer.backend.to_chainerx(grad_outputs)

        def f(x):
            return functions.max_pooling_2d(
                x, 3, stride=2, pad=1, cover_all=self.cover_all)

        with backend_config:
            gradient_check.check_backward(
                f, inputs, grad_outputs, dtype='d',
                **self.check_backward_options)

    def test_backward(self, backend_config):
{'fraction_non_alphanumeric': 0.1134714342261512, 'fraction_numerical': 0.01899986805647183, 'mean_word_length': 2.0949367088607596, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1794704', 'n_tokens_mistral': 5015, 'n_tokens_neox': 4792, 'n_words': 978}