<commit_before>import unittest

import numpy

import chainer
from chainer.backends import cuda
import chainer.functions as F
from chainer import gradient_check
from chainer import testing
from chainer.testing import attr
from chainer.testing import backend
from chainer.testing import condition


@testing.parameterize(*(testing.product({
    'c_contiguous': [True, False],
    'cover_all': [True, False],
    'x_dtype': [numpy.float32],
    'W_dtype': [numpy.float32],
    'dilate': [1],
    'groups': [1, 2],
    'nobias': [True, False],
}) + testing.product({
    'c_contiguous': [False],
    'cover_all': [False],
    'x_dtype': [numpy.float16, numpy.float32, numpy.float64],
    'W_dtype': [numpy.float16, numpy.float32, numpy.float64],
    'dilate': [1],
    'groups': [1, 2],
    'nobias': [True, False],
})))
@backend.inject_backend_tests(
    # TODO(hvy): test 'cuda:0'
    #['test_forward', 'test_backward', 'test_double_backward'],
    ['test_forward', 'test_backward'],
    # ChainerX tests
    testing.product({
        'use_chainerx': [True],
        'chainerx_device': ['native:0', 'cuda:0'],
    })
    # CPU tests
    + testing.product({
        'use_cuda': [False],
        'use_ideep': ['never', 'always'],
    })
    # GPU tests
    + testing.product([
        [{'use_cuda': True}],

        # Without cuDNN
        testing.product({
            'use_cudnn': ['never'],
        })
        # With cuDNN
        + testing.product({
            'use_cudnn': ['always'],
            'cudnn_deterministic': [True, False],
            'autotune': [True, False],
        })]))
class TestConvolution2DFunction(unittest.TestCase):

    def setUp(self):
        batches = 2
        in_channels_a_group = 3
        out_channels_a_group = 2
        in_channels = in_channels_a_group * self.groups
        out_channels = out_channels_a_group * self.groups
        kh, kw = (3, 3)
        self.stride = 2
        self.pad = (int(kh / 2) * self.dilate, int(kw / 2) * self.dilate)
        W = numpy.random.normal(
            0, numpy.sqrt(1. / (kh * kw * in_channels_a_group)),
            (out_channels, in_channels_a_group, kh, kw)).astype(self.W_dtype)

        if self.nobias:
            b = None
        else:
            b = numpy.random.uniform(
                -1, 1, out_channels).astype(self.x_dtype)

        x = numpy.random.uniform(
            -1, 1, (batches, in_channels, 4, 3)).astype(self.x_dtype)
        if self.cover_all:
            gy = numpy.random.uniform(
                -1, 1, (batches, out_channels, 3, 2)).astype(self.x_dtype)
        else:
            gy = numpy.random.uniform(
                -1, 1, (batches, out_channels, 2, 2)).astype(self.x_dtype)
        ggx = numpy.random.uniform(-1, 1, x.shape).astype(
            self.x_dtype)
        ggW = numpy.random.uniform(-1, 1, W.shape).astype(
            self.W_dtype)
        ggb = None if b is None else numpy.random.uniform(
            -1, 1, b.shape).astype(self.x_dtype)

        self.inputs = [x, W, b]
        self.grad_outputs = [gy]
        self.grad_grad_inputs = [ggx, ggW, ggb]

    def forward(self, inputs):
        x, W, b = inputs
        x = chainer.Variable(x)
        W = chainer.Variable(W)
        b = None if b is None else chainer.Variable(b)
        return F.convolution_2d(
            x, W, b, stride=self.stride, pad=self.pad,
            cover_all=self.cover_all, dilate=self.dilate,
            groups=self.groups)

    def check_forward(self, inputs, backend_config):
        with chainer.using_config('use_ideep', 'never'):
            y_expected = self.forward(inputs)

        inputs = backend_config.get_array(inputs)

        with backend_config:
            y_actual = self.forward(inputs)

        testing.assert_allclose(
            y_expected.data, y_actual.data, atol=5e-4, rtol=5e-3)

    def test_forward(self, backend_config):
        # TODO(hvy): chainerx does not support fp16 yet
        if backend_config.use_chainerx:
            if (any(x.dtype == numpy.float16
                    for x in self.inputs if x is not None)
                    or self.x_dtype != self.W_dtype):
                raise unittest.SkipTest('Not yet supported')

        self.check_forward(self.inputs, backend_config)

    def check_backward(self, inputs, grad_outputs, backend_config):
        if backend_config.use_cuda:
            inputs = cuda.to_gpu(inputs)
            grad_outputs = cuda.to_gpu(grad_outputs)

        x_data, W_data, b_data = inputs
        gy_data, = grad_outputs

        if not self.c_contiguous:
            xp = chainer.backend.get_array_module(*inputs)
            x_data = xp.asfortranarray(x_data)
            W_data = xp.asfortranarray(W_data)
            gy_data = xp.asfortranarray(gy_data)
            if b_data is not None:
                b = xp.empty((len(b_data) * 2,), dtype=b_data.dtype)
                b[::2] = b_data
                b_data = b[::2]
                assert not b_data.flags.c_contiguous

        args = (x_data, W_data)
        if b_data is not None:
            args += (b_data,)

        if backend_config.use_chainerx:
            args = chainer.backend.to_chainerx(args)
            gy_data = chainer.backend.to_chainerx(gy_data)

        if not self.c_contiguous:
            for x in args + type(args)((gy_data,)):
                if backend_config.use_chainerx:
                    assert not x.is_contiguous
                else:
                    assert not x.flags.c_contiguous

        def f(*args):
            return F.convolution_2d(*args, stride=self.stride, pad=self.pad,
                                    cover_all=self.cover_all,
                                    dilate=self.dilate, groups=self.groups)

        with backend_config:
            gradient_check.check_backward(
                f, args, gy_data, dtype='d', atol=5e-4, rtol=5e-3)

    @condition.retry(3)
    def test_backward(self, backend_config):
        # TODO(hvy): chainerx does not support fp16 yet
        if backend_config.use_chainerx:
            # TODO(hvy): Remove dilation and groups checks once retention
            # with ChainerX is implemented
            if (any(x.dtype == numpy.float16
                    for x in self.inputs if x is not None)
                    or self.x_dtype != self.W_dtype
                    or self.dilate > 1
                    or self.groups > 1):
                raise unittest.SkipTest('Not yet supported')

        self.check_backward(self.inputs, self.grad_outputs, backend_config)

    '''
    def check_double_backward(
            self, inputs, grad_outputs, grad_grad_inputs, backend_config):
        xp = backend_config.xp

        if backend_config.use_cuda:<commit_msg>Test cuda and backward<commit_after>import unittest

import numpy

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


# Supports numpy, cupy and chainerx ndarrays.
def _arrays_as_non_contiguous(arrays):
    assert isinstance(arrays, (list, tuple))

    xp = chainer.backend.get_array_module(*arrays)
    non_cont_arrays = []
    for a in arrays:
        if a is None:
            non_cont_a = None
        elif a.ndim == 1:
            if xp is chainerx:
                # chainerx.ndarray does not support item assignment.
                non_cont_a = xp.diag(
                    xp.diag(a, device=a.device), device=a.device)
            else:  # numpy or cupy
                non_cont_a = xp.empty((len(a) * 2,), dtype=a.dtype)
                non_cont_a[::2] = a
                a = non_cont_a[::2]
                non_cont_a = a
        else:
            non_cont_a = a.T.copy().T
        non_cont_arrays.append(non_cont_a)

    # Check non contiguousness.
    for a in non_cont_arrays:
        if a is None:
            continue
        elif xp is chainerx:
            assert not a.is_contiguous
        else:  # numpy, cupy
            assert not a.flags.c_contiguous

    return type(arrays)(non_cont_arrays)


@testing.parameterize(*(testing.product({
    'c_contiguous': [True, False],
    'cover_all': [True, False],
    'x_dtype': [numpy.float32],
    'W_dtype': [numpy.float32],
    'dilate': [1],
    'groups': [1, 2],
    'nobias': [True, False],
}) + testing.product({
    'c_contiguous': [False],
    'cover_all': [False],
    'x_dtype': [numpy.float16, numpy.float32, numpy.float64],
    'W_dtype': [numpy.float16, numpy.float32, numpy.float64],
    'dilate': [1],
    'groups': [1, 2],
    'nobias': [True, False],
})))
@backend.inject_backend_tests(
    ['test_forward', 'test_backward'],
    # ChainerX tests
    testing.product({
        'use_chainerx': [True],
        'chainerx_device': ['native:0', 'cuda:0'],
    })
    # CPU tests
    + testing.product({
        'use_cuda': [False],
        'use_ideep': ['never', 'always'],
    })
    # GPU tests
    + testing.product([
        [{'use_cuda': True}],

        # Without cuDNN
        testing.product({
            'use_cudnn': ['never'],
        })
        # With cuDNN
        + testing.product({
            'use_cudnn': ['always'],
            'cudnn_deterministic': [True, False],
            'autotune': [True, False],
        })]))
class TestConvolution2DFunction(unittest.TestCase):

    def setUp(self):
        batches = 2
        in_channels_a_group = 3
        out_channels_a_group = 2
        in_channels = in_channels_a_group * self.groups
        out_channels = out_channels_a_group * self.groups
        kh, kw = (3, 3)
        self.stride = 2
        self.pad = (int(kh / 2) * self.dilate, int(kw / 2) * self.dilate)
        W = numpy.random.normal(
            0, numpy.sqrt(1. / (kh * kw * in_channels_a_group)),
            (out_channels, in_channels_a_group, kh, kw)).astype(self.W_dtype)

        if self.nobias:
            b = None
        else:
            b = numpy.random.uniform(
                -1, 1, out_channels).astype(self.x_dtype)

        x = numpy.random.uniform(
            -1, 1, (batches, in_channels, 4, 3)).astype(self.x_dtype)
        if self.cover_all:
            gy = numpy.random.uniform(
                -1, 1, (batches, out_channels, 3, 2)).astype(self.x_dtype)
        else:
            gy = numpy.random.uniform(
                -1, 1, (batches, out_channels, 2, 2)).astype(self.x_dtype)
        ggx = numpy.random.uniform(-1, 1, x.shape).astype(
            self.x_dtype)
        ggW = numpy.random.uniform(-1, 1, W.shape).astype(
            self.W_dtype)
        ggb = None if b is None else numpy.random.uniform(
            -1, 1, b.shape).astype(self.x_dtype)

        self.inputs = [x, W, b]
        self.grad_outputs = [gy]
        self.grad_grad_inputs = [ggx, ggW, ggb]

    def forward(self, inputs):
        x, W, b = inputs
        x = chainer.Variable(x)
        W = chainer.Variable(W)
        b = None if b is None else chainer.Variable(b)
        return F.convolution_2d(
            x, W, b, stride=self.stride, pad=self.pad,
            cover_all=self.cover_all, dilate=self.dilate,
            groups=self.groups)

    def check_forward(self, inputs, backend_config):
        with chainer.using_config('use_ideep', 'never'):
            y_expected = self.forward(inputs)

        inputs = backend_config.get_array(inputs)

        with backend_config:
            y_actual = self.forward(inputs)

        testing.assert_allclose(
            y_expected.data, y_actual.data, atol=5e-4, rtol=5e-3)

    def test_forward(self, backend_config):
        # TODO(hvy): chainerx does not support fp16 yet
        if backend_config.use_chainerx:
            if (any(x.dtype == numpy.float16
                    for x in self.inputs if x is not None)
                    or self.x_dtype != self.W_dtype):
                raise unittest.SkipTest('Not yet supported')

        self.check_forward(self.inputs, backend_config)

    def check_backward(self, inputs, grad_outputs, backend_config):
        inputs = backend_config.get_array(inputs)
        grad_outputs = backend_config.get_array(grad_outputs)

        if not self.c_contiguous:
            inputs = _arrays_as_non_contiguous(inputs)
            grad_outputs = _arrays_as_non_contiguous(grad_outputs)

        # Exclude bias if None.
        if inputs[-1] is None:
            inputs = inputs[:-1]

        def f(*args):
            return F.convolution_2d(*args, stride=self.stride, pad=self.pad,
                                    cover_all=self.cover_all,
                                    dilate=self.dilate, groups=self.groups)

        with backend_config:
            gradient_check.check_backward(
                f, inputs, grad_outputs, dtype='d', atol=5e-4, rtol=5e-3)

    @condition.retry(3)
    def test_backward(self, backend_config):
        # TODO(hvy): chainerx does not support fp16 yet
        if backend_config.use_chainerx:
            # TODO(hvy): Remove dilation and groups checks once retention
            # with ChainerX is implemented
            if (any(x.dtype == numpy.float16
                    for x in self.inputs if x is not None)
                    or self.x_dtype != self.W_dtype
                    or self.dilate > 1
                    or self.groups > 1
                    or (backend_config.chainerx_device == 'cuda:0'
                        and self.cover_all)):
                raise unittest.SkipTest('Not yet supported')

        self.check_backward(self.inputs, self.grad_outputs, backend_config)

    '''
    def check_double_backward(
            self, inputs, grad_outputs, grad_grad_inputs, backend_config):
        xp = backend_config.xp

        if backend_config.use_cuda:
{'fraction_non_alphanumeric': 0.11354378818737271, 'fraction_numerical': 0.011856270002909513, 'mean_word_length': 2.0250825082508253, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '1375763', 'n_tokens_mistral': 4575, 'n_tokens_neox': 4390, 'n_words': 978}