Implement get_abs_max on BucketResampler

pyresample/bucket/__init__.py

@@ -296,6 +296,38 @@ def get_max(self, data, fill_value=np.nan, skipna=True):
         LOG.info("Get max of values in each location")
         return self._call_pandas_groupby_statistics('max', data, fill_value, skipna)
 
+    def get_abs_max(self, data, fill_value=np.nan, skipna=True):
+        """Calculate absolute maximums for each bin with drop-in-a-bucket resampling.
+
+        Returns for each bin the original signed value which has the largest
+        absolute value.
+
+        .. warning::
+
+            The slow :meth:`pandas.DataFrame.groupby` method is temporarily used here,
+            as the `dask_groupby <https://github.com/dcherian/dask_groupby>`_ is still under development.
+
+        Parameters
+        ----------
+        data : Numpy or Dask array
+            Data to be binned.
+        skipna : boolean (optional)
+                If True, skips NaN values for the maximum calculation
+                (similarly to Numpy's `nanmax`). Buckets containing only NaN are set to zero.
+                If False, sets the bucket to NaN if one or more NaN values are present in the bucket
+                (similarly to Numpy's `max`).
+                In both cases, empty buckets are set to 0.
+                Default: True
+
+        Returns
+        -------
+        data : Numpy or Dask array
+            Bin-wise maximums in the target grid
+        """
+        max_ = self.get_max(data, fill_value=fill_value, skipna=skipna)
+        min_ = self.get_min(data, fill_value=fill_value, skipna=skipna)
+        return da.where(-min_ > max_, min_, max_)
+
     def get_count(self):
         """Count the number of occurrences for each bin using drop-in-a-bucket resampling.
 

pyresample/test/test_bucket.py

@@ -65,6 +65,7 @@ def test_init(self, get_indices, prj):
         self.assertTrue(hasattr(resampler, 'get_count'))
         self.assertTrue(hasattr(resampler, 'get_min'))
         self.assertTrue(hasattr(resampler, 'get_max'))
+        self.assertTrue(hasattr(resampler, 'get_abs_max'))
         self.assertTrue(hasattr(resampler, 'get_average'))
         self.assertTrue(hasattr(resampler, 'get_fractions'))
         self.assertIsNone(resampler.counts)
@@ -198,15 +199,15 @@ def test_get_min(self):
         data = da.from_array(np.array([[2, 11], [5, np.nan]]),
                              chunks=self.chunks)
         result = self._get_min_result(data)
-        # test multiple entries average
+        # test multiple entries minimum
         self.assertEqual(np.count_nonzero(result == 2), 1)
-        # test single entry average
+        # test single entry minimum
         self.assertEqual(np.count_nonzero(result == 5), 1)
         # test that minimum of bucket with only nan is nan, and empty buckets are nan
         self.assertEqual(np.count_nonzero(~np.isnan(result)), 2)
 
     def _get_max_result(self, data, **kwargs):
-        """Compute the bucket average with kwargs and check that no dask computation is performed."""
+        """Compute the bucket max with kwargs and check that no dask computation is performed."""
         with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
             result = self.resampler.get_max(data, **kwargs)
         return result.compute()
@@ -216,13 +217,31 @@ def test_get_max(self):
         data = da.from_array(np.array([[2, 11], [5, np.nan]]),
                              chunks=self.chunks)
         result = self._get_max_result(data)
-        # test multiple entries average
+        # test multiple entries maximum
         self.assertEqual(np.count_nonzero(result == 11), 1)
-        # test single entry average
+        # test single entry maximum
         self.assertEqual(np.count_nonzero(result == 5), 1)
         # test that minimum of bucket with only nan is nan, and empty buckets are nan
         self.assertEqual(np.count_nonzero(~np.isnan(result)), 2)
 
+    def test_get_abs_max(self):
+        """Test abs max bucket resampling."""
+        data = da.from_array(np.array([[2, -11], [5, np.nan]]),
+                             chunks=self.chunks)
+        result = self._get_abs_max_result(data)
+        # test multiple entries absolute maximum
+        self.assertEqual(np.count_nonzero(result == -11), 1)
+        # test single entry maximum
+        self.assertEqual(np.count_nonzero(result == 5), 1)
+        # test that minimum of bucket with only nan is nan, and empty buckets are nan
+        self.assertEqual(np.count_nonzero(~np.isnan(result)), 2)
+
+    def _get_abs_max_result(self, data, **kwargs):
+        """Compute the bucket abs max with kwargs and check that no dask computation is performed."""
+        with dask.config.set(scheduler=CustomScheduler(max_computes=0)):
+            result = self.resampler.get_abs_max(data, **kwargs)
+        return result.compute()
+
     def _get_average_result(self, data, **kwargs):
         """Compute the bucket average with kwargs and check that no dask computation is performed."""
         with dask.config.set(scheduler=CustomScheduler(max_computes=0)):