Al/math/minpooling

deeptrack/math.py

@@ -55,9 +55,9 @@
 
 - `AveragePooling`: Apply average pooling to the image.
 
-- `MaxPooling`: Apply max pooling to the image.
+- `MaxPooling`: Apply max-pooling to the image.
 
-- `MinPooling`: Apply min pooling to the image.
+- `MinPooling`: Apply min-pooling to the image.
 
 - `MedianPooling`: Apply median pooling to the image.
 
@@ -1251,7 +1251,7 @@ def __init__(
 
 #TODO ***AL*** revise MaxPooling - torch, typing, docstring, unit test
 class MaxPooling(Pool):
-    """Apply max pooling to images.
+    """Apply max-pooling to images.
 
     This class reduces the resolution of an image by dividing it into
     non-overlapping blocks of size `ksize` and applying the max function to
@@ -1277,7 +1277,7 @@ class MaxPooling(Pool):
     Create an input image:
     >>> input_image = np.random.rand(32, 32)
 
-    Define a max pooling feature:
+    Define a max-pooling feature:
     >>> max_pooling = dt.MaxPooling(ksize=8)
     >>> output_image = max_pooling(input_image)
     >>> print(output_image.shape)
@@ -1297,9 +1297,9 @@ def __init__(
         ksize: PropertyLike[int] = 3,
         **kwargs: Any,
     ):
-        """Initialize the parameters for max pooling.
+        """Initialize the parameters for max-pooling.
 
-        This constructor initializes the parameters for max pooling.
+        This constructor initializes the parameters for max-pooling.
 
         Parameters
         ----------
@@ -1313,43 +1313,42 @@ def __init__(
         super().__init__(np.max, ksize=ksize, **kwargs)
 
 
-#TODO ***AL*** revise MinPooling - torch, typing, docstring, unit test
 class MinPooling(Pool):
-    """Apply min pooling to images.
+    """Apply min-pooling to images.
 
-    This class reduces the resolution of an image by dividing it into
-    non-overlapping blocks of size `ksize` and applying the min function to
+    `MinPooling` reduces the resolution of an image by dividing it into
+    non-overlapping blocks of size `ksize` and applying the `min` function to
     each block. The result is a downsampled image where each pixel value
-    represents the minimum value within the corresponding block of the
-    original image.
+    represents the minimum value within the corresponding block of the original
+    image.
+
+    If the backend is NumPy, the downsampling is performed using 
+    `skimage.measure.block_reduce`.
+
+    If the backend is PyTorch, the downsampling is performed using the inverse
+    of `torch.nn.functional.max_pool2d` by changing the sign of the input.
 
     Parameters
     ----------
     ksize: int
         Size of the pooling kernel.
-    **kwargs: dict
+    **kwargs: Any
         Additional parameters sent to the pooling function.
 
     Examples
     --------
     >>> import deeptrack as dt
-    >>> import numpy as np
 
     Create an input image:
+    >>> import numpy as np
+    >>>
     >>> input_image = np.random.rand(32, 32)
 
-    Define a min pooling feature:
-    >>> min_pooling = dt.MinPooling(ksize=3)
+    Define and use a min-pooling feature:
+    >>> min_pooling = dt.MinPooling(ksize=4)
     >>> output_image = min_pooling(input_image)
-    >>> print(output_image.shape)
-    (32, 32)
-
-    Notes
-    -----
-    Calling this feature returns a `np.ndarray` by default. If
-    `store_properties` is set to `True`, the returned array will be
-    automatically wrapped in an `Image` object. This behavior is handled
-    internally and does not affect the return type of the `get()` method.
+    >>> output_image.shape
+    (8, 8)
 
     """
 
@@ -1358,9 +1357,10 @@ def __init__(
         ksize: PropertyLike[int] = 3,
         **kwargs: Any,
     ):
-        """Initialize the parameters for min pooling.
+        """Initialize the parameters for min-pooling.
 
-        This constructor initializes the parameters for min pooling.
+        This constructor initializes the parameters for min-pooling and checks
+        whether to use the NumPy or PyTorch implementation, defaults to NumPy.
 
         Parameters
         ----------
@@ -1373,6 +1373,118 @@ def __init__(
 
         super().__init__(np.min, ksize=ksize, **kwargs)
 
+    def get(
+        self: MinPooling,
+        image: NDArray[Any] | torch.Tensor,
+        ksize: int=3,
+        **kwargs: Any,
+    ) -> NDArray[Any] | torch.Tensor:
+        """Min pooling of input.
+
+        Checks the current backend and chooses the appropriate function to pool
+        the input image, either `._get_torch()` or `._get_numpy()`.
+
+        Parameters
+        ----------
+        image: array or tensor
+            Input array or tensor to be pooled.
+        ksize: int
+            Kernel size of the pooling operation.
+
+        Returns
+        -------
+        array or tensor
+            The pooled image as `NDArray` or `torch.Tensor` depending on the
+            backend.
+
+        """
+
+        if self.get_backend() == "numpy":
+            return self._get_numpy(image, ksize, **kwargs)
+
+        if self.get_backend() == "torch":
+            return self._get_torch(image, ksize, **kwargs)
+
+        raise NotImplementedError(f"Backend {self.backend} not supported")
+
+    def _get_numpy(
+        self: MinPooling,
+        image: NDArray[Any],
+        ksize: int=3,
+        **kwargs: Any,
+    ) -> NDArray[Any]:
+        """Min-pooling with the NumPy backend.
+
+        Returns the result of the input array passed to the scikit
+        `image block_reduce()` function with `np.min()` as the pooling
+        function.
+
+        Parameters
+        ----------
+        image: NDArray
+            Input image to be pooled.
+        ksize: int
+            Kernel size of the pooling operation.
+
+        Returns
+        -------
+        NDArray
+            The pooled image as a `NDArray`.
+
+        """
+
+        return utils.safe_call(
+            skimage.measure.block_reduce,
+            image=image,
+            func=np.min,
+            block_size=ksize,
+            **kwargs,
+        )
+
+    def _get_torch(
+        self: MinPooling,
+        image: torch.Tensor,
+        ksize: int=3,
+        **kwargs: Any,
+    ) -> torch.Tensor:
+        """Min-pooling with the PyTorch backend.
+
+        As PyTorch does not have a min-pooling layer, the equivalent operation
+        is to first multiply the input tensor with `-1`, then perform
+        max-pooling, and finally multiply the max pooled tensor with `-1`.
+
+        Parameters
+        ----------
+        image: torch.Tensor
+            Input tensor to be pooled.
+        ksize: int
+            Kernel size of the pooling operation.
+
+        Returns
+        -------
+        torch.Tensor
+            The pooled image as a `torch.Tensor`.
+
+        """
+
+        # If input tensor is 2D
+        if len(image.shape) == 2:
+            # Add batch dimension for min-pooling
+            expanded_image = image.unsqueeze(0)
+
+            pooled_image = - torch.nn.functional.max_pool2d(
+                expanded_image * (-1),
+                kernel_size=ksize,
+            )
+
+            # Remove the expanded dim
+            return pooled_image.squeeze(0)
+
+        return -torch.nn.functional.max_pool2d(
+            image * (-1),
+            kernel_size=ksize,
+        )
+
 
 #TODO ***AL*** revise MedianPooling - torch, typing, docstring, unit test
 class MedianPooling(Pool):

deeptrack/tests/test_math.py

@@ -82,6 +82,16 @@ def test_Blur(self):
         #blurred_image = feature.resolve(input_image)
         #self.assertTrue(xp.all(blurred_image == expected_output))
 
+    def test_MinPooling(self):
+        input_image = xp.asarray([[1, 2, 3, 4], [5, 6, 7, 8]], dtype=float)
+        feature = math.MinPooling(ksize=2)
+        pooled_image = feature.resolve(input_image)
+
+        expected = xp.asarray([[1.0, 3.0]], dtype=float)
+
+        self.assertEqual(pooled_image.shape, (1, 2))
+        self.assertTrue(xp.all(pooled_image == expected))
+
 
 # Extending the test and setting the backend to torch
 @unittest.skipUnless(TORCH_AVAILABLE, "PyTorch is not installed.")