migrate functions

.travis.yml

@@ -10,7 +10,6 @@ install:
   - pip install coveralls
 
 script:
-  - ls
   - nosetests test/functions -a '!gpu,!local' --with-xunit --with-coverage --cover-erase --cover-package=src.functions --verbose
 
 after_success:

test/functions/__init__.py

@@ -0,0 +1 @@
+__author__ = 'karita'

test/functions/test_gaussian_mixture_2d.py

@@ -0,0 +1,95 @@
+import unittest
+
+import numpy
+from numpy.random import uniform, binomial
+
+import chainer
+from chainer import cuda
+from chainer import gradient_check
+from chainer import testing
+from chainer.testing import attr, condition
+
+from src.functions.gaussian_mixture_2d import gaussian_mixture_2d
+from src.functions.gaussian_mixture_2d_ref import gaussian_mixture_2d_ref
+from src import gravesnet
+
+
+if cuda.available:
+    cuda.init()
+
+
+class TestGaussianMixture2d(unittest.TestCase):
+
+    def setUp(self):
+        # each 2D-Gaussian contains 6 params: weight, mean(2), stddev(2), corr
+        self.ngauss = numpy.random.randint(1, 5)
+        input_size = 6 * self.ngauss + 1
+        mini_batch = numpy.random.randint(1, 5)
+        self.x   = uniform(-1, 1, (mini_batch, input_size)).astype(numpy.float32)
+        self.t_x = uniform(-1, 1, (mini_batch, 2)).astype(numpy.float32)
+        b_rand   = [[binomial(1, 0.9) for _ in range(mini_batch)]]
+        self.t_e = numpy.array(b_rand).astype(numpy.int32).reshape((mini_batch, 1))
+        self.g = uniform(-1, 1, (mini_batch, self.ngauss)).astype(numpy.float32)
+
+    def check(self):
+        t_x = chainer.Variable(self.context(self.t_x))
+        t_e = chainer.Variable(self.context(self.t_e))
+        x = chainer.Variable(self.context(self.x))
+
+        loss = gravesnet.loss_func(self.ngauss, x, t_x, t_e)
+        loss.backward()
+        self.assertEqual(None, t_x.grad)
+        self.assertEqual(None, t_e.grad)
+
+        func = loss.creator
+        f = lambda: func.forward((self.ngauss, x.data, t_x.data, t_e.data))
+        loss.grad *= 10.0
+        # gx, = gradient_check.numerical_grad(f, (x.data,), (loss.grad,), eps=1e-2)
+        return x.grad, loss.data
+
+    def check_ref(self):
+        t_x = chainer.Variable(self.context(self.t_x))
+        t_e = chainer.Variable(self.context(self.t_e))
+        x = chainer.Variable(self.context(self.x))
+        y, e = gravesnet.split_args(self.ngauss, x, t_x, t_e)
+        p = gaussian_mixture_2d(*y)
+        q = gaussian_mixture_2d_ref(*y)
+        gradient_check.assert_allclose(p.data, q.data)
+
+
+        # TODO: Check and pass backward
+        # x.grad = None
+        # p_loss = gravesnet.concat_losses(p, e, t_e)
+        # q_loss = gravesnet.concat_losses(q, e, t_e)
+        # p_loss.backward()
+        # p_xg = x.grad.copy()
+        # x.grad = None
+        # q_loss.backward()
+        # q_xg = x.grad.copy()
+        # print(p_xg, q_xg)
+        # gradient_check.assert_allclose(p_loss.data, q_loss.data)
+        # gradient_check.assert_allclose(p_xg, q_xg)
+
+    @condition.retry(3)
+    def test_original_versus_chainer_cpu(self):
+        self.context = lambda x: x
+        self.check_ref()
+
+    @condition.retry(3)
+    @attr.gpu
+    def test_original_versus_chainer_gpu(self):
+        self.context = cuda.to_gpu
+        self.check_ref()
+
+    @condition.retry(3)
+    @attr.gpu
+    def test_cpu_versus_gpu(self):
+        self.context = lambda x: x
+        cpu, closs = self.check()
+        self.context = cuda.to_gpu
+        gpu, gloss = self.check()
+        numpy.testing.assert_almost_equal(closs, cuda.to_cpu(gloss))
+        gradient_check.assert_allclose(gpu, cpu)
+
+
+testing.run_module(__name__, __file__)

test/functions/test_gradient_clip.py

@@ -0,0 +1,56 @@
+import unittest
+
+import chainer
+from chainer import cuda
+from chainer.gradient_check import assert_allclose
+from chainer.testing import attr, condition, run_module
+import numpy
+
+from src.functions.gradient_clip import gradient_clip
+
+
+if cuda.available:
+    cuda.init()
+
+
+class TestGradientClip(unittest.TestCase):
+
+    def setUp(self):
+        self.a = numpy.random.random()
+        self.b = numpy.random.random()
+        self.x = numpy.random.randn(4, 3).astype(numpy.float32)
+        self.y_grad = numpy.random.randn(4, 3).astype(numpy.float32)
+        lower = min(self.a, self.b)
+        upper = max(self.a, self.b)
+        self.x_grad_ab = numpy.clip(self.y_grad, lower, upper)
+        self.x_grad_a = numpy.clip(self.y_grad, -abs(self.a), abs(self.a))
+
+    def check_backward(self, f):
+        x = chainer.Variable(f(self.x))
+
+        y = gradient_clip(x, self.a, self.b)
+        y.creator.forward((x.data,))
+        y.grad = f(self.y_grad)
+        y.backward()
+        assert_allclose(y.data, x.data)
+        assert_allclose(x.grad, f(self.x_grad_ab))
+
+        y = gradient_clip(x, self.a)
+        y.creator.forward((x.data,))
+        y.grad = f(self.y_grad)
+        y.backward()
+        assert_allclose(y.data, x.data)
+        assert_allclose(x.grad, f(self.x_grad_a))
+
+
+    @condition.retry(100)
+    def test_backward_cpu(self):
+        self.check_backward(lambda x: x)
+
+    @attr.gpu
+    @condition.retry(100)
+    def test_backward_gpu(self):
+        self.check_backward(cuda.to_gpu)
+
+
+run_module(__name__, __file__)

test/functions/test_peephole_lstm.py

@@ -0,0 +1,166 @@
+import unittest
+
+import numpy
+
+import chainer
+from chainer import cuda
+from chainer import gradient_check
+from chainer import testing
+from chainer.testing import attr
+from chainer.testing import condition
+
+from src.functions.peephole_lstm import peephole_lstm
+
+if cuda.available:
+    cuda.init()
+
+
+def _sigmoid(x):
+    return 1 / (1 + numpy.exp(-x))
+
+
+class TestPeepholeLSTM(unittest.TestCase):
+
+    def setUp(self):
+        self.c_prev = numpy.random.uniform(-1,
+                                           1, (3, 2, 4)).astype(numpy.float32)
+        self.a = numpy.random.uniform(-1, 1, (3, 2, 4)).astype(numpy.float32)
+        self.x = numpy.random.uniform(-1, 1, (3, 6, 4)).astype(numpy.float32)
+
+        self.gc = numpy.random.uniform(-1, 1, (3, 2, 4)).astype(numpy.float32)
+        self.gh = numpy.random.uniform(-1, 1, (3, 2, 4)).astype(numpy.float32)
+
+    def flat(self):
+        self.c_prev = self.c_prev[:, :, 0].copy()
+        self.a = self.a[:, :, 0].copy()
+        self.x = self.x[:, :, 0].copy()
+        self.gc = self.gc[:, :, 0].copy()
+        self.gh = self.gh[:, :, 0].copy()
+
+    def check_forward(self, c_prev_data, a_data, x_data):
+        c_prev = chainer.Variable(c_prev_data)
+        a = chainer.Variable(a_data)
+        x = chainer.Variable(x_data)
+        c, h = peephole_lstm(c_prev, a, x)
+        self.assertEqual(c.data.dtype, numpy.float32)
+        self.assertEqual(h.data.dtype, numpy.float32)
+
+        # Compute expected out
+        a_in = self.a
+        i_in = self.x[:, [0, 3]]
+        f_in = self.x[:, [1, 4]]
+        o_in = self.x[:, [2, 5]]
+
+        c_expect = _sigmoid(i_in) * numpy.tanh(a_in) + \
+            _sigmoid(f_in) * self.c_prev
+        h_expect = _sigmoid(o_in) * numpy.tanh(c_expect)
+
+        gradient_check.assert_allclose(c_expect, c.data)
+        gradient_check.assert_allclose(h_expect, h.data)
+
+    @condition.retry(3)
+    def test_forward_cpu(self):
+        self.check_forward(self.c_prev, self.a, self.x)
+
+    @condition.retry(3)
+    def test_flat_forward_cpu(self):
+        self.flat()
+        self.test_forward_cpu()
+
+    @attr.gpu
+    @condition.retry(3)
+    def test_forward_gpu(self):
+        self.check_forward(cuda.to_gpu(self.c_prev), cuda.to_gpu(self.a), cuda.to_gpu(self.x))
+
+    @attr.gpu
+    @condition.retry(3)
+    def test_flat_forward_gpu(self):
+        self.flat()
+        self.test_forward_gpu()
+
+    def check_backward(self, c_prev_data, a_data, x_data, c_grad, h_grad):
+        c_prev = chainer.Variable(c_prev_data)
+        a = chainer.Variable(a_data)
+        x = chainer.Variable(x_data)
+        c, h = peephole_lstm(c_prev, a, x)
+        c.grad = c_grad
+        h.grad = h_grad
+        c.backward()
+
+        func = c.creator
+        f = lambda: func.forward((c_prev.data, a.data, x.data))
+        gc_prev, ga, gx = gradient_check.numerical_grad(
+            f, (c_prev.data, a.data, x.data), (c_grad, h_grad), eps=1e-2)
+
+        gradient_check.assert_allclose(gc_prev, c_prev.grad)
+        gradient_check.assert_allclose(ga, a.grad)
+        gradient_check.assert_allclose(gx, x.grad)
+
+    @condition.retry(3)
+    def test_full_backward_cpu(self):
+        self.check_backward(self.c_prev, self.a, self.x, self.gc, self.gh)
+
+    @condition.retry(3)
+    def test_flat_full_backward_cpu(self):
+        self.flat()
+        self.test_full_backward_cpu()
+
+    @condition.retry(3)
+    def test_no_gc_backward_cpu(self):
+        self.check_backward(self.c_prev, self.a, self.x, None, self.gh)
+
+    @condition.retry(3)
+    def test_flat_no_gc_backward_cpu(self):
+        self.flat()
+        self.test_no_gc_backward_cpu()
+
+    @condition.retry(3)
+    def test_no_gh_backward_cpu(self):
+        self.check_backward(self.c_prev, self.a, self.x, self.gc, None)
+
+    @condition.retry(3)
+    def test_flat_no_gh_backward_cpu(self):
+        self.flat()
+        self.test_no_gh_backward_cpu()
+
+    @attr.gpu
+    @condition.retry(3)
+    def test_full_backward_gpu(self):
+        self.check_backward(
+            cuda.to_gpu(self.c_prev), cuda.to_gpu(self.a), cuda.to_gpu(self.x),
+            cuda.to_gpu(self.gc), cuda.to_gpu(self.gh))
+
+    @attr.gpu
+    @condition.retry(3)
+    def test_flat_full_backward_gpu(self):
+        self.flat()
+        self.test_full_backward_gpu()
+
+    @attr.gpu
+    @condition.retry(3)
+    def test_no_gc_backward_gpu(self):
+        self.check_backward(
+            cuda.to_gpu(self.c_prev), cuda.to_gpu(self.a), cuda.to_gpu(self.x),
+            None, cuda.to_gpu(self.gh))
+
+    @attr.gpu
+    @condition.retry(3)
+    def test_flat_no_gc_backward_gpu(self):
+        self.flat()
+        self.test_no_gc_backward_gpu()
+
+    @attr.gpu
+    @condition.retry(3)
+    def test_no_gh_backward_gpu(self):
+        self.check_backward(
+            cuda.to_gpu(self.c_prev), cuda.to_gpu(self.a), cuda.to_gpu(self.x),
+            cuda.to_gpu(self.gc), None)
+
+    @attr.gpu
+    @condition.retry(3)
+    def test_flat_no_gh_backward_gpu(self):
+        self.flat()
+        self.test_no_gh_backward_gpu()
+
+
+testing.run_module(__name__, __file__)

test/functions/test_split_axis.py

@@ -0,0 +1,51 @@
+from unittest import TestCase
+
+import chainer
+from chainer import cuda
+from chainer.functions import concat
+from chainer.gradient_check import assert_allclose
+from chainer import testing
+from chainer.testing import attr, condition
+import numpy
+
+from src.functions.spilit_axis import split_axis_by_widths
+
+
+if cuda.available:
+    cuda.init()
+
+
+class TestSplitAxis(TestCase):
+
+    def setUp(self):
+        # self.ws = [2, 4]
+        h = numpy.random.randint(1, 5)
+        l = h * numpy.random.randint(1, 5)
+        self.ws = [numpy.random.randint(1, 5)] * l
+        self.ws[0] = h
+        self.mini_batch = numpy.random.randint(1, 5)
+
+    def check(self, widths):
+        x_size = sum(self.ws)
+        x = numpy.arange(self.mini_batch * x_size,
+                         dtype=numpy.float32).reshape(self.mini_batch, x_size)
+        x = chainer.Variable(self.context(x))
+        y = split_axis_by_widths(x, widths)
+        z = concat(y)
+        assert_allclose(x.data, z.data)
+
+    @condition.retry(100)
+    def test_split_axis_cpu(self):
+        self.context = lambda x: x
+        self.check(self.ws)
+        self.check(self.ws[0])
+
+    @condition.retry(100)
+    @attr.gpu
+    def test_split_axis_gpu(self):
+        self.context = cuda.to_gpu
+        self.check(self.ws)
+        self.check(self.ws[0])
+
+
+testing.run_module(__name__, __file__)