normalize signal's shapes to #signal x #features x #nodes

Author: mdeff

pygsp/filters/approximations.py

@@ -213,6 +213,41 @@ def _evaluate(self, x, method):
 
     def _filter(self, s, method, _):
 
+        # TODO: signal normalization will move to Filter.filter()
+
+        # Dimension 3: number of nodes.
+        if s.shape[-1] != self.G.N:
+            raise ValueError('The last dimension should be {}, '
+                             'the number of nodes. '
+                             'Got instead a signal of shape '
+                             '{}.'.format(self.G.N, s.shape))
+
+        # Dimension 2: number of input features.
+        if s.ndim == 1:
+            s = np.expand_dims(s, 0)
+        if s.shape[-2] != self.n_features_in:
+            if self.n_features_in == 1:
+                # Dimension can be omitted if there's 1 input feature.
+                s = np.expand_dims(s, -2)
+            else:
+                raise ValueError('The second to last dimension should be {}, '
+                                 'the number of input features. '
+                                 'Got instead a signal of shape '
+                                 '{}.'.format(self.n_features_in, s.shape))
+
+        # Dimension 1: number of independent signals.
+        if s.ndim < 3:
+            s = np.expand_dims(s, 0)
+
+        if s.ndim > 3:
+            raise ValueError('Signals should have at most 3 dimensions: '
+                             '#signals x #features x #nodes.')
+
+        assert s.ndim == 3
+        assert s.shape[2] == self.G.N  # Number of nodes.
+        assert s.shape[1] == self.n_features_in  # Number of input features.
+        # n_signals = s.shape[0]
+
         L = self.scale_operator(self.G.L, self.G.lmax)
 
         # Recursive and clenshaw are similarly fast.

pygsp/filters/filter.py

@@ -273,7 +273,11 @@ def filter(self, s, method=None, order=30):
         """
         if not sparse.issparse(s):
             s = np.asarray(s)  # For iterables.
-        return self._filter(s, method, order)
+
+        s = self._filter(s, method, order)
+
+        # Return a 1D signal if e.g. a 1D signal was filtered by one filter.
+        return s.squeeze()
 
     def _filter(self, s, method='chebyshev', order=30):
         r"""Default implementation for filters defined as kernel functions."""
@@ -344,8 +348,7 @@ def _filter(self, s, method='chebyshev', order=30):
         else:
             raise ValueError('Unknown method {}.'.format(method))
 
-        # Return a 1D signal if e.g. a 1D signal was filtered by one filter.
-        return s.squeeze()
+        return s
 
     def analyze(self, s, method='chebyshev', order=30):
         r"""Convenience alias to :meth:`filter`."""

pygsp/tests/test_filters.py

@@ -280,7 +280,7 @@ def test_evaluation_methods(self, K=30, F=5, N=100):
 
     def test_filter_identity(self, M=10, c=2.3):
         r"""Test that filtering with c0 only scales the signal."""
-        x = self._rs.uniform(size=(M, 1, self._G.N))
+        x = self._rs.uniform(size=(M, self._G.N))
         f = filters.Chebyshev(self._G, c)
         y = f.filter(x, method='recursive')
         np.testing.assert_equal(y, c * x)
@@ -331,3 +331,56 @@ def test_approximations(self, N=100, K=20):
         y1 = f1.filter(x.T).T
         y2 = f2.filter(x)
         np.testing.assert_allclose(y2.squeeze(), y1)
+
+    def test_shape_normalization(self):
+        """Test that signal's shapes are properly normalized."""
+        # TODO: should also test filters which are not approximations.
+
+        def test_normalization(M, Fin, Fout, K=7):
+
+            def test_shape(y, M, Fout, N=self._G.N):
+                """Test that filtered signals are squeezed."""
+                if Fout == 1 and M == 1:
+                    self.assertEqual(y.shape, (N,))
+                elif Fout == 1:
+                    self.assertEqual(y.shape, (M, N))
+                elif M == 1:
+                    self.assertEqual(y.shape, (Fout, N))
+                else:
+                    self.assertEqual(y.shape, (M, Fout, N))
+
+            coefficients = self._rs.uniform(size=(K, Fout, Fin))
+            f = filters.Chebyshev(self._G, coefficients)
+            assert f.shape == (Fin, Fout)
+            assert (f.n_features_in, f.n_features_out) == (Fin, Fout)
+
+            x = self._rs.uniform(size=(M, Fin, self._G.N))
+            y = f.filter(x)
+            test_shape(y, M, Fout)
+
+            if Fin == 1 or M == 1:
+                # It only makes sense to squeeze if one dimension is unitary.
+                x = x.squeeze()
+                y = f.filter(x)
+                test_shape(y, M, Fout)
+
+        # Test all possible correct combinations of input and output signals.
+        for M in [1, 9]:
+            for Fin in [1, 3]:
+                for Fout in [1, 5]:
+                    test_normalization(M, Fin, Fout)
+
+        # Test failure cases.
+        M, Fin, Fout, K = 9, 3, 5, 7
+        coefficients = self._rs.uniform(size=(K, Fout, Fin))
+        f = filters.Chebyshev(self._G, coefficients)
+        x = self._rs.uniform(size=(M, Fin, 2))
+        self.assertRaises(ValueError, f.filter, x)
+        x = self._rs.uniform(size=(M, 2, self._G.N))
+        self.assertRaises(ValueError, f.filter, x)
+        x = self._rs.uniform(size=(2, self._G.N))
+        self.assertRaises(ValueError, f.filter, x)
+        x = self._rs.uniform(size=(self._G.N))
+        self.assertRaises(ValueError, f.filter, x)
+        x = self._rs.uniform(size=(2, M, Fin, self._G.N))
+        self.assertRaises(ValueError, f.filter, x)