Merge pull request #2 from miaotianyi/feature/pooling

Feature/pooling

requirements.txt

@@ -1,2 +1,6 @@
 torch>=1.7.1
-numpy
+numpy>=1.20.1
+scipy>=1.6.2
+
+scikit-image>=0.18.1
+matplotlib>=3.3.4

requirements_ci.txt

@@ -1 +1,6 @@
 PyWavelets
+scipy~=1.6.2
+
+numpy~=1.20.1
+scikit-image~=0.18.1
+matplotlib~=3.3.4

tests/test_ndspec.py

@@ -0,0 +1,145 @@
+import unittest
+import numpy as np
+from torchimage.utils import NdSpec
+import torch
+
+
+class MyTestCase(unittest.TestCase):
+    def test_kernels(self):
+        nds = NdSpec([1, 2, 3], item_shape=(-1, ))
+        self.assertTrue(nds.is_item)
+        for i in range(-10, 10):
+            self.assertEqual(nds[i], [1, 2, 3])
+
+    def test_ragged(self):
+        for filter_list in [
+            [[1, 2, 3], [4, 5], [6, 7, 8], [9]],
+            ([1, 2, 3], [4, 5], [6, 7, 8], [9]),
+            ([1, 2, 3], (4, 5), [6, 7, 8], (9,)),
+            ([1, 2, 3], (4, 5), [6, 7, 8], 9),
+            (np.array([1, 2, 3]), (4, 5), [6, 7, 8], 9),
+            (torch.tensor([1, 2, 3]), (4, 5), [6, 7, 8], 9),
+            (torch.tensor([[1, 2], [2, 3]]), (4, 5), [6, 7, 8], 9),
+        ]:
+            with self.subTest(data=filter_list):
+                nds = NdSpec(filter_list, item_shape=())
+                self.assertEqual(len(nds), 4)
+                self.assertEqual(len([x for x in nds]), 4)
+                self.assertFalse(nds.is_item)
+                self.assertEqual(len(nds), len(filter_list))
+                for i in range(-len(filter_list), len(filter_list)):
+                    # these objects shouldn't change at all
+                    self.assertIs(filter_list[i], nds[i])
+
+    def test_empty(self):
+        # each item is a list of unknown length; can be empty (is empty in this case)
+        data = []
+        a = NdSpec(data, item_shape=(-1,))
+        self.assertIs(a[0], data)
+        # with self.assertRaises(ValueError):
+        with self.assertRaises(ValueError):
+            NdSpec([], item_shape=(1, 2, 3))  # cannot accept empty input
+
+    def test_list_singleton(self):
+        nds = NdSpec([[1, 2]], item_shape=[2])
+        self.assertFalse(nds.is_item)
+        self.assertEqual(len(nds), 1)
+        self.assertEqual(nds[0], [1, 2])
+        self.assertEqual(nds[-1], [1, 2])
+        self.assertEqual([x for x in nds], [[1, 2]])
+        for i in list(range(-20, -1)) + list(range(1, 15)):
+            with self.assertRaises(IndexError):
+                nds[i]
+
+    def test_item_list(self):
+        nds = NdSpec([1, 2], item_shape=[2])
+        self.assertEqual(len(nds), 0)
+        self.assertEqual([x for x in nds], [[1, 2]])
+        self.assertTrue(nds.is_item)
+        for i in np.random.randint(-100, 100, 50):
+            self.assertEqual(nds[i], [1, 2])
+
+    def test_item_scalar(self):
+        nds = NdSpec(423, item_shape=[])
+        self.assertEqual(len(nds), 0)
+        self.assertEqual([x for x in nds], [423])
+        self.assertTrue(nds.is_item)
+        for i in np.random.randint(-100, 100, 50):
+            self.assertEqual(nds[i], 423)
+
+    def test_broadcast_1(self):
+        nds = NdSpec(423, item_shape=[2])
+        self.assertTrue(nds.is_item)
+        self.assertEqual(len(nds), 0)
+        self.assertEqual([x for x in nds], [(423, 423)])
+        for i in range(-10, 10):
+            self.assertEqual(nds[-i], (423, 423))
+
+    def test_broadcast_2(self):
+        nds = NdSpec([1, 2, 4], item_shape=[2, 3])
+        self.assertTrue(nds.is_item)
+        self.assertEqual(len(nds), 0)
+        self.assertEqual(list(nds), [([1, 2, 4], [1, 2, 4])])
+        for i in range(-10, 10):
+            self.assertEqual(nds[-i], ([1, 2, 4], [1, 2, 4]))
+
+    def test_broadcast_3(self):
+        nds = NdSpec("hello", item_shape=[1, 2, 3])
+        self.assertTrue(nds.is_item)
+        self.assertEqual(len(nds), 0)
+        self.assertEqual([((("hello",) * 3,) * 2,)], list(nds))
+        for i in range(-10, 10):
+            self.assertEqual(nds[i], ((("hello",) * 3,) * 2,))
+
+    def test_filter_list(self):
+        data_1 = [[1, 2], [3, 4, 5], [6, 7, 8, 9]]
+        data_2 = [[1, 2], (3, 4), np.array([5, 6]), torch.tensor([7., 8.], requires_grad=True)]
+        for expected, actual in zip(data_1, NdSpec(data_1, item_shape=[-1])):
+            self.assertEqual(expected, actual)
+
+        for expected, actual in zip(data_2, NdSpec(data_2, item_shape=[-1])):
+            self.assertIs(expected, actual)
+
+    def test_filter_list_2(self):
+        a = NdSpec(2, item_shape=[-1])
+        self.assertEqual((2,), a[0])
+        self.assertEqual("NdSpec(data=(2,), item_shape=(-1,))", str(a))
+
+    def test_index_tuple_1(self):
+        d1 = [1, 2]
+        d2 = [3, 4, 5]
+        a = NdSpec([d1, d2], item_shape=())
+        self.assertIs(a[0], d1)
+        self.assertIs(a[1], d2)
+        for i, val in enumerate(d1):
+            self.assertEqual(a[0, i], val)
+        for i, val in enumerate(d2):
+            self.assertEqual(a[1, i], val)
+
+    def test_map_1(self):
+        for _ in range(10):
+            a = np.random.rand(10)
+            nds = NdSpec(a, item_shape=[])
+            self.assertTrue(np.array_equal(a * 2, nds.map(lambda x: x * 2)))
+
+    def test_zip_1(self):
+        a1 = NdSpec([1, 2, 3])
+        a2 = NdSpec([4, 5, 6], item_shape=[-1])
+        a3 = NdSpec([[7, 8], [9, 10], [11]], item_shape=[-1])
+        actual = NdSpec.zip(a1, a2, a3)
+        expected = NdSpec(data=[(1, [4, 5, 6], [7, 8]),
+                                (2, [4, 5, 6], [9, 10]),
+                                (3, [4, 5, 6], [11])], item_shape=(3,))
+
+        self.assertEqual(expected.data, actual.data)
+        self.assertEqual(expected.item_shape, actual.item_shape)
+
+    def test_zip_2(self):
+        a1 = NdSpec([1, 2, 3])
+        a2 = NdSpec([4, 5, 6, 7])
+        with self.assertRaises(ValueError):
+            NdSpec.zip(a1, a2)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_padder.py

@@ -16,7 +16,7 @@ def assertArrayEqual(self, a, b, tol=None, msg=None):
         if tol is None:
             tol = self.tol
         lib = torch if (torch.is_tensor(a) and torch.is_tensor(b)) else np
-        self.assertLess(lib.abs(a - b).sum(), tol, msg=f"{a} is not equal to {b}")
+        self.assertLess(lib.abs(a - b).sum(), tol, msg=f"{a} is not equal to {b}, {msg}")
 
     def test_const_padding(self):
         mode = "constant"

tests/test_pooling.py

@@ -0,0 +1,123 @@
+import unittest
+
+from torchimage.linalg import outer
+from torchimage.padding import GenericPadNd
+from torchimage.pooling.base import SeparablePoolNd
+from torchimage.pooling.gaussian import GaussianPoolNd
+from torchimage.pooling.uniform import AveragePoolNd
+from torchimage.filtering.utils import _same_padding_pair
+from torchimage.filtering.decorator import pool_to_filter
+
+
+import numpy as np
+import torch
+from functools import reduce
+from scipy import ndimage
+
+NDIMAGE_PAD_MODES = [("symmetric", "reflect"),
+                     ("replicate", "nearest"),
+                     ("constant", "constant"),
+                     ("reflect", "mirror"),
+                     ("circular", "wrap")]
+
+
+class SeparableFilterNd(SeparablePoolNd):
+    # hand-written for testing only
+    def __init__(self, kernel: object):
+        super(SeparableFilterNd, self).__init__(kernel=kernel, stride=1)
+
+    def forward(self, x: torch.Tensor, axes=None, same=True, padder: GenericPadNd = None):
+        if same and padder is not None:
+            # same padding
+            same_pad_width = self.kernel_size.map(_same_padding_pair)
+            padder = GenericPadNd(pad_width=same_pad_width,
+                                  mode=padder.mode.data,
+                                  constant_values=padder.constant_values.data,
+                                  end_values=padder.end_values.data,
+                                  stat_length=padder.stat_length.data)
+
+        return super(SeparableFilterNd, self).forward(x, axes=axes, padder=padder)
+
+
+class MyTestCase(unittest.TestCase):
+    def test_uniform(self):
+        for n in range(1, 10):
+            x = torch.rand(100, 41) * 100 - 50
+            x = torch.round(x)
+            for ti_mode, ndimage_mode in NDIMAGE_PAD_MODES:
+                y_ti = SeparableFilterNd(np.ones(n) / n)(x, same=True, padder=GenericPadNd(mode=ti_mode))
+                y_ndimage = ndimage.uniform_filter(x.numpy(), size=n, mode=ndimage_mode)
+                result = np.allclose(y_ti.numpy(), y_ndimage, rtol=1e-5, atol=1e-5, equal_nan=False)
+                with self.subTest(ti_mode=ti_mode, ndimage_mode=ndimage_mode, n=n):
+                    self.assertTrue(result)
+
+    def test_conv(self):
+        for ti_mode, ndimage_mode in NDIMAGE_PAD_MODES:
+            for ndim in range(1, 5):
+                kernel_size = np.random.randint(1, 10, size=ndim)
+                kernels = [np.random.rand(ks) for ks in kernel_size]
+                shape = tuple(np.random.randint(20, 50, size=ndim))
+                x = torch.rand(*shape)
+                full_conv_tensor = reduce(outer, kernels)
+                # note that convolve in neural network is correlate in signal processing
+                y_ndimage = ndimage.correlate(x.numpy(), weights=full_conv_tensor, mode=ndimage_mode)
+                y_ti = SeparableFilterNd(kernels)(x, same=True, padder=GenericPadNd(mode=ti_mode))
+                result = np.allclose(y_ti.numpy(), y_ndimage, rtol=1e-7, atol=1e-5, equal_nan=False)
+                with self.subTest(ti_mode=ti_mode, ndimage_mode=ndimage_mode, ndim=ndim, kernel_size=kernel_size, shape=shape):
+                    self.assertTrue(result)
+
+    def test_wrapper_1(self):
+        # wrapped image filter should behave the same way as its base pooling class
+        GaussianFilterNd = pool_to_filter(GaussianPoolNd, same=True)
+
+        x = torch.rand(17, 100, 5)
+
+        # gaussian filter type
+        gf_1 = GaussianFilterNd(9, sigma=1.5, order=0)
+        gf_2 = GaussianFilterNd(9, 1.5, 0)
+        gp = GaussianPoolNd(9, sigma=1.5, order=0, stride=1)
+
+        y1 = gf_1(x, padder=GenericPadNd(mode="reflect"))
+        y2 = gf_2(x, padder=GenericPadNd(mode="reflect"))
+        y = gp(x, padder=GenericPadNd(pad_width=4, mode="reflect"))
+        self.assertEqual(torch.abs(y1 - y).max().item(), 0)
+        self.assertEqual(torch.abs(y2 - y).max().item(), 0)
+
+    def test_gaussian_1(self):
+        sigma = 1.5
+
+        GaussianFilterClass = pool_to_filter(GaussianPoolNd, same=True)
+        for truncate in range(2, 10, 2):
+            for order in range(6):
+                for ti_mode, ndimage_mode in NDIMAGE_PAD_MODES:
+                    x = torch.rand(10, 37, 21, dtype=torch.float64)
+                    y_sp = ndimage.gaussian_filter(x.numpy(), sigma=sigma, order=order, mode=ndimage_mode, truncate=truncate)
+                    gf1 = GaussianFilterClass(kernel_size=2 * truncate * sigma + 1, sigma=sigma, order=order)
+                    y_ti = gf1(x, axes=None, padder=GenericPadNd(mode=ti_mode))
+                    y_ti = y_ti.numpy()
+                    self.assertLess(np.abs(y_sp - y_ti).max(), 1e-10)
+
+    def test_precision_1(self):
+        # 1d convolution precision testing
+        for ti_mode, ndimage_mode in NDIMAGE_PAD_MODES:
+            x = torch.rand(10, dtype=torch.float64)
+            w = torch.rand(5, dtype=torch.float64)
+            y1 = ndimage.correlate1d(x.numpy(), w.numpy(), axis=-1, mode=ndimage_mode, origin=0)
+            y2 = pool_to_filter(SeparablePoolNd, same=True)(w)(x, padder=GenericPadNd(mode=ti_mode)).numpy()
+            result = np.allclose(y1, y2, rtol=1e-9, atol=1e-9)
+            with self.subTest(ti_mode=ti_mode, ndimage_mode=ndimage_mode):
+                self.assertTrue(result)
+
+    def test_average_1(self):
+        UniformFilterNd = pool_to_filter(AveragePoolNd, same=True)
+        for kernel_size in range(3, 15, 2):
+            x = torch.rand(13, 25, 18, dtype=torch.float64)
+            for ti_mode, ndimage_mode in NDIMAGE_PAD_MODES:
+                y_ti = UniformFilterNd(kernel_size=kernel_size)(x, padder=GenericPadNd(mode=ti_mode)).numpy()
+                y_ndi = ndimage.uniform_filter(x.numpy(), size=kernel_size, mode=ndimage_mode)
+                with self.subTest(kernel_size=kernel_size, ti_mode=ti_mode, ndimage_mode=ndimage_mode):
+                    self.assertLess(np.abs(y_ti - y_ndi).max(), 1e-10)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_ragged.py

@@ -0,0 +1,103 @@
+import unittest
+
+import numpy as np
+import torch
+from torchimage.utils.ragged import get_ragged_ndarray, expand_ragged_ndarray
+
+
+class MyTestCase(unittest.TestCase):
+    def setUp(self) -> None:
+        self.n_trials = 20
+
+    def test_get_shape_1(self):
+        f = get_ragged_ndarray
+        # test regular arrays
+        for _ in range(self.n_trials):
+            ndim = np.random.randint(0, 10)
+            shape = np.random.randint(0, 5, ndim)
+            with self.subTest(shape=shape):
+                self.assertEqual(f(np.empty(shape))[1], tuple(shape))
+
+    def test_get_shape_2(self):
+        f = get_ragged_ndarray
+        data, shape = f([1, [2, 3]], strict=True)
+        self.assertEqual(data[0], 1)
+        self.assertEqual(data[1], [2, 3])
+        self.assertEqual(shape, (2,))
+
+        original = [np.random.rand(np.random.randint(0, 8)) for _ in range(100)]
+        data, shape = f(original, strict=False)
+        self.assertEqual(shape, (100, -1))
+        for a, b in zip(data, original):
+            self.assertTrue(np.array_equal(a, b))
+
+        data, shape = f((1, (2, 3)), strict=False)
+        self.assertEqual(data, ((1,), (2, 3)))
+        self.assertEqual(shape, (2, -1))
+
+        data, shape = f((1, (2, (3, 4), 5)), strict=False)
+        self.assertEqual(data, (((1,),), ((2, ), (3, 4), (5,))))
+        self.assertEqual(shape, (2, -1, -1))
+
+        data, shape = f(((1, 2), ((3, 4), (5, 6))), strict=False)
+        self.assertEqual(data, (((1, 2),), ((3, 4), (5, 6))))
+        self.assertEqual(shape, (2, -1, 2))
+
+        source = ([1, 2], [3, 4], [5, 6])
+        data, shape = f(source, strict=False)
+        self.assertEqual(data, source)
+        self.assertEqual(shape, (3, 2))
+
+    def test_expand(self):
+        for new_shape in [
+            [1, 2, -1],
+            [1, -1, -1]
+        ]:
+            arr, shape = expand_ragged_ndarray([[1], [2, 3]], old_shape=[2, -1], new_shape=new_shape)
+            self.assertEqual(arr, ([[1], [2, 3]],))
+            self.assertEqual(shape, (1, 2, -1))
+
+        arr, shape = expand_ragged_ndarray([[1], [2], [3]], old_shape=[-1, -1], new_shape=[1, 3, 6])
+        self.assertEqual(arr, (((1,) * 6,) + ((2,) * 6,) + ((3,) * 6,),))
+        self.assertEqual(shape, (1, 3, 6))
+
+        arr, shape = get_ragged_ndarray("hello")
+        self.assertEqual(arr, "hello")
+        self.assertEqual(shape, ())
+
+        arr, shape = expand_ragged_ndarray("hello", old_shape=(), new_shape=[1, 3, 6])
+        self.assertEqual(arr, ((("hello",) * 6,) * 3,))
+        self.assertEqual(shape, (1, 3, 6))
+
+    def test_filter_list_1(self):
+        data = [[1, 2], [3, 4, 5], [6, 7, 8, 9]]
+        arr, shape = get_ragged_ndarray(data, strict=True)
+        actual_arr, actual_shape = expand_ragged_ndarray(arr, shape, new_shape=[-1])
+        self.assertEqual(actual_arr, data)  # should remain unchanged
+        self.assertEqual(actual_shape, (3, -1))
+
+    def test_filter_list_2(self):
+        data = [[1, 2], (3, 4), np.array([5, 6]), torch.tensor([7., 8.], requires_grad=True)]
+        arr, shape = get_ragged_ndarray(data, strict=True)
+        actual_arr, actual_shape = expand_ragged_ndarray(arr, shape, new_shape=[-1])
+        self.assertEqual(data, actual_arr)
+        self.assertEqual((4, 2), actual_shape)
+
+    def test_get_shape_3(self):
+        data = [np.array(1.), 2, 3]
+        arr, shape = get_ragged_ndarray(data, strict=False)
+        self.assertIs(data, arr)
+        self.assertEqual(shape, (3,))
+
+    def test_empty(self):
+        data = []
+        arr, shape = get_ragged_ndarray(data=data, strict=True)
+        self.assertIs(arr, data)
+        self.assertEqual(shape, (0,))
+        with self.assertRaises(ValueError):
+            arr_2, shape_2 = expand_ragged_ndarray(data=arr, old_shape=shape, new_shape=(1, 2, 3))
+
+
+if __name__ == '__main__':
+    unittest.main()
+