Immigrate reference operations to a separate module (keras-team#9948)

Author: taehoonlee

tests/keras/backend/backend_test.py

@@ -1,14 +1,15 @@
 import pytest
 from numpy.testing import assert_allclose
 import numpy as np
-import scipy.signal as signal
 import scipy.sparse as sparse
 import warnings
 from keras.utils.test_utils import keras_test
 
 from keras import backend as K
 from keras.backend import floatx, set_floatx, variable
 from keras.utils.conv_utils import convert_kernel
+import reference_operations
+
 
 BACKENDS = []  # Holds a list of all available back-ends
 
@@ -192,162 +193,6 @@ def check_composed_tensor_operations(first_function_name, first_function_args,
     assert_list_pairwise(z_list)
 
 
-def normalize_ref_conv(func):
-    def wrapper(*args):
-        x = args[0]
-        w = args[1]
-        if x.ndim == 3:
-            w = np.flipud(w)
-            w = np.transpose(w, (1, 2, 0))
-            if args[3] == 'channels_last':
-                x = np.transpose(x, (0, 2, 1))
-        elif x.ndim == 4:
-            w = np.fliplr(np.flipud(w))
-            w = np.transpose(w, (2, 3, 0, 1))
-            if args[3] == 'channels_last':
-                x = np.transpose(x, (0, 3, 1, 2))
-        else:
-            w = np.flip(np.fliplr(np.flipud(w)), axis=2)
-            w = np.transpose(w, (3, 4, 0, 1, 2))
-            if args[3] == 'channels_last':
-                x = np.transpose(x, (0, 4, 1, 2, 3))
-
-        y = func(x, w, args[2], args[3])
-
-        if args[3] == 'channels_last':
-            if y.ndim == 3:
-                y = np.transpose(y, (0, 2, 1))
-            elif y.ndim == 4:
-                y = np.transpose(y, (0, 2, 3, 1))
-            else:
-                y = np.transpose(y, (0, 2, 3, 4, 1))
-
-        return y
-
-    return wrapper
-
-
-@normalize_ref_conv
-def ref_conv(x, w, padding, data_format):
-    y = []
-    for i in range(x.shape[0]):
-        _y = []
-        for j in range(w.shape[1]):
-            __y = []
-            for k in range(w.shape[0]):
-                __y.append(signal.convolve(x[i, k], w[k, j], mode=padding))
-            _y.append(np.sum(np.stack(__y, axis=-1), axis=-1))
-        y.append(_y)
-    y = np.array(y)
-    return y
-
-
-@normalize_ref_conv
-def ref_depthwise_conv(x, w, padding, data_format):
-    y = []
-    for i in range(x.shape[0]):
-        _y = []
-        for j in range(w.shape[0]):
-            __y = []
-            for k in range(w.shape[1]):
-                __y.append(signal.convolve(x[i, j], w[j, k], mode=padding))
-            _y.append(np.stack(__y, axis=0))
-        y.append(np.concatenate(_y, axis=0))
-    y = np.array(y)
-    return y
-
-
-def ref_separable_conv(x, w1, w2, padding, data_format):
-    x2 = ref_depthwise_conv(x, w1, padding, data_format)
-    return ref_conv(x2, w2, padding, data_format)
-
-
-def ref_pool(x, pool_size, strides, padding, data_format, pool_mode):
-    if data_format == 'channels_last':
-        if x.ndim == 3:
-            x = np.transpose(x, (0, 2, 1))
-        elif x.ndim == 4:
-            x = np.transpose(x, (0, 3, 1, 2))
-        else:
-            x = np.transpose(x, (0, 4, 1, 2, 3))
-
-    if padding == 'same':
-        pad = [(0, 0), (0, 0)] + [(s // 2, s // 2) for s in pool_size]
-        x = np.pad(x, pad, 'constant', constant_values=-np.inf)
-
-    # indexing trick
-    x = np.pad(x, [(0, 0), (0, 0)] + [(0, 1) for _ in pool_size],
-               'constant', constant_values=0)
-
-    if x.ndim == 3:
-        y = [x[:, :, k:k1:strides[0]]
-             for (k, k1) in zip(range(pool_size[0]), range(-pool_size[0], 0))]
-    elif x.ndim == 4:
-        y = []
-        for (k, k1) in zip(range(pool_size[0]), range(-pool_size[0], 0)):
-            for (l, l1) in zip(range(pool_size[1]), range(-pool_size[1], 0)):
-                y.append(x[:, :, k:k1:strides[0], l:l1:strides[1]])
-    else:
-        y = []
-        for (k, k1) in zip(range(pool_size[0]), range(-pool_size[0], 0)):
-            for (l, l1) in zip(range(pool_size[1]), range(-pool_size[1], 0)):
-                for (m, m1) in zip(range(pool_size[2]), range(-pool_size[2], 0)):
-                    y.append(x[:, :, k:k1:strides[0], l:l1:strides[1], m:m1:strides[2]])
-    y = np.stack(y, axis=-1)
-    if pool_mode == 'avg':
-        y = np.mean(np.ma.masked_invalid(y), axis=-1).data
-    elif pool_mode == 'max':
-        y = np.max(y, axis=-1)
-
-    if data_format == 'channels_last':
-        if y.ndim == 3:
-            y = np.transpose(y, (0, 2, 1))
-        elif y.ndim == 4:
-            y = np.transpose(y, (0, 2, 3, 1))
-        else:
-            y = np.transpose(y, (0, 2, 3, 4, 1))
-
-    return y
-
-
-def ref_rnn(x, w, init, go_backwards=False, mask=None, unroll=False, input_length=None):
-    w_i, w_h, w_o = w
-    h = []
-    o = []
-
-    if go_backwards:
-        t_list = range(x.shape[1] - 1, -1, -1)
-    else:
-        t_list = range(x.shape[1])
-
-    if mask is not None:
-        np_mask = K.eval(mask)
-    else:
-        np_mask = None
-
-    for (i, t) in enumerate(t_list):
-        h_t = np.dot(x[:, t], w_i)
-
-        if w_h is not None:
-            prev = h[i - 1] if i > 0 else init
-            h_t1 = np.dot(prev, w_h)
-            if np_mask is not None:
-                h_t1[np_mask[:, t] == 0] = prev[np_mask[:, t] == 0]
-        else:
-            h_t1 = 0
-
-        o_t = h_t + h_t1
-        if w_o is not None:
-            o_t = np.dot(o_t, w_o)
-        o.append(o_t)
-
-        if np_mask is not None:
-            h_t = h_t * np_mask[:, t].reshape(-1, 1)
-        h.append(h_t + h_t1)
-
-    return o[-1], np.stack(o, axis=1), np.stack(h, axis=1)
-
-
 class TestBackend(object):
 
     def test_is_keras_tensor(self):
@@ -724,7 +569,7 @@ def rnn_fn(x_k, h_k):
         ]
 
         for (i, kwargs) in enumerate(kwargs_list):
-            last_y1, y1, h1 = ref_rnn(x, [wi, wh, None], h0, **kwargs)
+            last_y1, y1, h1 = reference_operations.rnn(x, [wi, wh, None], h0, **kwargs)
             last_y2, y2, h2 = K.rnn(rnn_fn, x_k, h0_k, **kwargs)
 
             assert len(h2) == 1
@@ -771,8 +616,8 @@ def rnn_fn(x_k, h_k):
             y_k = K.dot(x_k, wi_k)
             return y_k, []
 
-        last_y1, y1, h1 = ref_rnn(x, [wi, None, None], None,
-                                  go_backwards=False, mask=None)
+        last_y1, y1, h1 = reference_operations.rnn(x, [wi, None, None], None,
+                                                   go_backwards=False, mask=None)
         last_y2, y2, h2 = K.rnn(rnn_fn, x_k, [],
                                 go_backwards=False, mask=None)
 
@@ -1071,7 +916,7 @@ def test_conv(self, op, input_shape, kernel_shape, padding, data_format):
         k = K.backend()
         _, x = parse_shape_or_val(input_shape)
         _, w = parse_shape_or_val(kernel_shape)
-        y1 = ref_conv(x, w, padding, data_format)
+        y1 = reference_operations.conv(x, w, padding, data_format)
         y2 = check_two_tensor_operation(
             op, x, w, [KTH if k == 'theano' else KC if k == 'cntk' else KTF],
             padding=padding, data_format=data_format,
@@ -1088,7 +933,7 @@ def test_depthwise_conv(self, op, input_shape, kernel_shape, padding, data_forma
         k = K.backend()
         _, x = parse_shape_or_val(input_shape)
         _, w = parse_shape_or_val(kernel_shape)
-        y1 = ref_depthwise_conv(x, w, padding, data_format)
+        y1 = reference_operations.depthwise_conv(x, w, padding, data_format)
         y2 = check_two_tensor_operation(
             op, x, w, [KTH if k == 'theano' else KC if k == 'cntk' else KTF],
             padding=padding, data_format=data_format,
@@ -1108,7 +953,7 @@ def test_depthwise_conv(self, op, input_shape, kernel_shape, padding, data_forma
     def test_pool(self, op, input_shape, pool_size, strides, padding, data_format, pool_mode):
         k = K.backend()
         _, x = parse_shape_or_val(input_shape)
-        y1 = ref_pool(x, pool_size, strides, padding, data_format, pool_mode)
+        y1 = reference_operations.pool(x, pool_size, strides, padding, data_format, pool_mode)
         y2 = check_single_tensor_operation(
             op, x, [KTH if k == 'theano' else KC if k == 'cntk' else KTF],
             pool_size=pool_size, strides=strides,
@@ -1174,7 +1019,7 @@ def test_separable_conv2d(self, op, input_shape, kernel_shape, depth_multiplier,
         _, x = parse_shape_or_val(input_shape)
         _, depthwise = parse_shape_or_val(kernel_shape + (input_depth, depth_multiplier))
         _, pointwise = parse_shape_or_val((1, 1) + (input_depth * depth_multiplier, 7))
-        y1 = ref_separable_conv(x, depthwise, pointwise, padding, data_format)
+        y1 = reference_operations.separable_conv(x, depthwise, pointwise, padding, data_format)
         if K.backend() == 'cntk':
             y2 = cntk_func_three_tensor(
                 op, input_shape,