Merge pull request #1 from dwest77a/functional

Version 2024.8.16: Functional XarrayActive

Author: dwest77a

.github/workflows/ci.yml

@@ -0,0 +1,32 @@
+name: Automatic Test
+# Specify which GitHub events will trigger a CI build
+
+on: push
+# Define a single job, build
+
+jobs:
+  build:
+    # Specify an OS for the runner
+    runs-on: ubuntu-latest
+
+    #Define steps
+    steps:
+
+      # Firstly, checkout repo
+      - name: Checkout repository
+        uses: actions/checkout@v2
+      # Set up Python env
+      - name: Setup Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: 3.11
+      # Install dependencies
+      - name: Install Python dependencies
+        run: |
+          python3 -m pip install --upgrade pip
+          pip3 install -r requirements.txt
+          pip3 install -e .
+      # Test with pytest
+      - name: Run pytest
+        run: |
+          pytest

XarrayActive/active_chunk.py

@@ -1,95 +1,158 @@
 import numpy as np
+from itertools import product
 
 
-# Holds all CFA-specific Active routines.
+class ActiveOptionsContainer:
+    """
+    Container for ActiveOptions properties.
+    """
+    @property
+    def active_options(self):
+        """
+        Property of the datastore that relates private option variables to the standard 
+        ``active_options`` parameter.
+        """
+        return {
+            'chunks': self._active_chunks,
+            'chunk_limits': self._chunk_limits,
+        }
+    
+    @active_options.setter
+    def active_options(self, value):
+        self._set_active_options(**value)
+
+    def _set_active_options(self, chunks={}, chunk_limits=True):
+
+        if chunks == {}:
+            raise NotImplementedError(
+                'Default chunking is not implemented, please provide a chunk scheme '
+                ' - active_options = {"chunks": {}}'
+            )
+
+        self._active_chunks = chunks
+        self._chunk_limits = chunk_limits
+
+# Holds all Active routines.
 class ActiveChunk:
 
     description = "Container class for Active routines performed on each chunk. All active-per-chunk content can be found here."
-
-    def __init__(self, *args, **kwargs):
-        raise NotImplementedError
 
     def _post_process_data(self, data):
         # Perform any post-processing steps on the data here
         return data
 
-    def _standard_mean(self, axis=None, skipna=None, **kwargs):
+    def _standard_sum(self, axes=None, skipna=None, **kwargs):
         """
         Standard Mean routine matches the normal routine for dask, required at this
         stage if Active mean not available.
         """
-        size = 1
-        for i in axis:
-            size *= self.shape[i]
 
         arr = np.array(self)
         if skipna:
-            total = np.nanmean(arr, axis=axis, **kwargs) *size
+            total = np.nansum(arr, axis=axes, **kwargs)
         else:
-            total = np.mean(arr, axis=axis, **kwargs) *size
-        return {'n': self._numel(arr, axis=axis), 'total': total}
+            total = np.sum(arr, axis=axes, **kwargs)
+        return total
+    
+    def _standard_max(self, axes=None, skipna=None, **kwargs):
+        return np.max(self, axis=axes)
+    
+    def _standard_min(self, axes=None, skipna=None, **kwargs):
+        return np.min(self, axis=axes)
 
-    def _numel(self, axis=None):
-        if not axis:
+    def _numel(self, method, axes=None):
+        if not axes:
             return self.size
 
         size = 1
-        for i in axis:
+        for i in axes:
             size *= self.shape[i]
         newshape = list(self.shape)
-        newshape[axis] = 1
+        for ax in axes:
+            newshape[ax] = 1
 
         return np.full(newshape, size)
 
-    def active_mean(self, axis=None, skipna=None, **kwargs):
+    def active_method(self, method, axis=None, skipna=None, **kwargs):
         """
         Use PyActiveStorage package functionality to perform mean of this Fragment.
 
-        :param axis:        (int) The axis over which to perform the active_mean operation.
+        :param axis:        (int) The axes over which to perform the active_mean operation.
 
         :param skipna:      (bool) Skip NaN values when calculating the mean.
 
         :returns:       A ``duck array`` (numpy-like) with the reduced array or scalar value, 
-                        as specified by the axis parameter.
+                        as specified by the axes parameter.
         """
+
+        standard_methods = {
+            'mean': self._standard_sum,
+            'sum' : self._standard_sum,
+            'max' : self._standard_max,
+            'min' : self._standard_min
+        }
+        ret = None
+        n = self._numel(method, axes=axis)
+
         try:
             from activestorage.active import Active
         except ImportError:
             # Unable to import Active package. Default to using normal mean.
             print("ActiveWarning: Unable to import active module - defaulting to standard method.")
-            return self._standard_mean(axis=axis, skipna=skipna, **kwargs)
-            
-        active = Active(self.filename, self.address)
-        active.method = "mean"
-        extent = self.get_extent()
-
-        if not axis is None:
-            return {
-                'n': self._numel(axis=axis),
-                'total': self._post_process_data(active[extent])
+            ret = {
+                'n': n,
+                'total': standard_methods[method](axes=axis, skipna=skipna, **kwargs)
             }
 
-        # Experimental Recursive requesting to get each 1D column along the axis being requested.
-        range_recursives = []
-        for dim in range(self.ndim):
-            if dim != axis:
-                range_recursives.append(range(extent[dim].start, extent[dim].stop+1))
-            else:
-                range_recursives.append(extent[dim])
-        results = np.array(self._get_elements(active, range_recursives, hyperslab=[]))
+        if not ret:
+            
+            active = Active(self.filename, self.address)
+            active.method = method
+            extent = tuple(self.get_extent())
+
+            if axis == None:
+                axis = tuple([i for i in range(self.ndim)])
+
+            n = self._numel(method, axes=axis)
+
+            if len(axis) == self.ndim:
+                data   = active[extent]
+                t = self._post_process_data(data) * n
+
+                ret = {
+                    'n': n,
+                    'total': t
+                }
+
+        if not ret:
+            # Experimental Recursive requesting to get each 1D column along the axes being requested.
+            range_recursives = []
+            for dim in range(self.ndim):
+                if dim not in axis:
+                    range_recursives.append(range(extent[dim].start, extent[dim].stop))
+                else:
+                    range_recursives.append(extent[dim])
+            results = np.array(self._get_elements(active, range_recursives, hyperslab=[]))
+
+            t = self._post_process_data(results) * n
+            ret = {
+                'n': n,
+                'total': t
+            }
 
-        return {
-            'n': self._numel(axis=axis),
-            'total': self._post_process_data(results)
-        }
+        if method == 'mean':
+            return ret
+        else:
+            return ret['total']/ret['n']
 
     def _get_elements(self, active, recursives, hyperslab=[]):
         dimarray = []
-        current = recursives[0]
-        if not len(recursives) > 1:
+        if not len(recursives) > 0:
 
             # Perform active slicing and meaning here.
-            return active[hyperslab]
+            return active[tuple(hyperslab)].flatten()[0]
+        
+        current = recursives[0]
 
         if type(current) == slice:
             newslab = hyperslab + [current]

XarrayActive/active_dask.py

@@ -1,13 +1,44 @@
 import dask.array as da
-from dask.array.reductions import mean_agg
+from dask.array.reductions import mean_agg, mean_combine, nanmax, nanmin
+from dask.utils import deepmap
+from dask.array.core import _concatenate2
+import numpy as np
 
-
-def block_active_mean(arr, *args, **kwargs):
-    if hasattr(arr,'active_mean'):
-        return arr.active_mean(*args, **kwargs)
+def partition_mean(arr, *args, **kwargs):
+    return partition_method(arr, 'mean', *args, **kwargs)
+    
+def partition_max(arr, *args, **kwargs):
+    return partition_method(arr, 'max', *args, **kwargs)
+    
+def partition_min(arr, *args, **kwargs):
+    return partition_method(arr, 'min', *args, **kwargs)
+    
+def partition_sum(arr, *args, **kwargs):
+    return partition_method(arr, 'sum', *args, **kwargs)
+    
+def partition_method(arr, method, *args, **kwargs):
+    if hasattr(arr,'active_method'):
+        return arr.active_method(method,*args, **kwargs)
     else:
-        # Here's where barebones Xarray might fall over - may need a non-CFA custom class.
-        raise NotImplementedError
+        # Additional handling for 'meta' calculations in dask.
+        # Not currently implemented, bypassed using None
+        if arr.size == 0:
+            return None
+        return None
+
+def general_combine(pairs, axis=None):
+    if not isinstance(pairs, list):
+        pairs = [pairs]
+    return _concatenate2(pairs, axes=axis)
+
+def max_agg(pairs, axis=None, **kwargs):
+    return general_combine(pairs, axis=axis).max(axis=axis, **kwargs)
+
+def min_agg(pairs, axis=None, **kwargs):
+    return general_combine(pairs, axis=axis).min(axis=axis, **kwargs)
+
+def sum_agg(pairs, axis=None, **kwargs):
+    return general_combine(pairs, axis=axis).sum(axis=axis, **kwargs)
 
 class DaskActiveArray(da.Array):
 
@@ -17,11 +48,12 @@ class DaskActiveArray(da.Array):
     def is_active(self):
         return True
 
-    def copy(self):
-        """
-        Create a new DaskActiveArray instance with all the same parameters as the current instance.
-        """
-        return DaskActiveArray(self.dask, self.name, self.chunks, meta=self)
+    #def copy(self):
+    #    """
+    #    Create a new DaskActiveArray instance with all the same parameters as the current instance.
+    #    """
+     #   copy_arr = DaskActiveArray(self.dask, self.name, self.chunks, meta=self)
+    #    return copy_arr
 
     def __getitem__(self, index):
         """
@@ -49,10 +81,86 @@ def active_mean(self, axis=None, skipna=None):
 
         newarr = da.reduction(
             self,
-            block_active_mean,
+            partition_mean,
             mean_agg,
+            combine=mean_combine,
+            axis=axis,
+            dtype=self.dtype,
+        )
+
+        return newarr
+
+    def active_max(self, axis=None, skipna=None):
+        """
+        Perform ``dask delayed`` active mean for each ``dask block`` which corresponds to a single ``chunk``.
+        Combines the results of the dask delayed ``active_max`` operations on each block into a single dask Array,
+        which is then mapped to a new DaskActiveArray object.
+
+        :param axis:        (int) The index of the axis on which to perform the active max.
+
+        :param skipna:      (bool) Skip NaN values when calculating the max.
+
+        :returns:       A new ``DaskActiveArray`` object which has been reduced along the specified axes using
+                        the concatenations of active_means from each chunk.
+        """
+
+        newarr = da.reduction(
+            self,
+            partition_max,
+            max_agg,
+            combine=max_agg,
+            axis=axis,
+            dtype=self.dtype,
+        )
+
+        return newarr
+    
+    def active_min(self, axis=None, skipna=None):
+        """
+        Perform ``dask delayed`` active mean for each ``dask block`` which corresponds to a single ``chunk``.
+        Combines the results of the dask delayed ``active_min`` operations on each block into a single dask Array,
+        which is then mapped to a new DaskActiveArray object.
+
+        :param axis:        (int) The index of the axis on which to perform the active min.
+
+        :param skipna:      (bool) Skip NaN values when calculating the min.
+
+        :returns:       A new ``DaskActiveArray`` object which has been reduced along the specified axes using
+                        the concatenations of active_means from each chunk.
+        """
+
+        newarr = da.reduction(
+            self,
+            partition_min,
+            min_agg,
+            combine=min_agg,
+            axis=axis,
+            dtype=self.dtype,
+        )
+
+        return newarr
+    
+    def active_sum(self, axis=None, skipna=None):
+        """
+        Perform ``dask delayed`` active mean for each ``dask block`` which corresponds to a single ``chunk``.
+        Combines the results of the dask delayed ``active_sum`` operations on each block into a single dask Array,
+        which is then mapped to a new DaskActiveArray object.
+
+        :param axis:        (int) The index of the axis on which to perform the active sum.
+
+        :param skipna:      (bool) Skip NaN values when calculating the sum.
+
+        :returns:       A new ``DaskActiveArray`` object which has been reduced along the specified axes using
+                        the concatenations of active_means from each chunk.
+        """
+
+        newarr = da.reduction(
+            self,
+            partition_sum,
+            sum_agg,
+            combine=sum_agg,
             axis=axis,
-            dtype=self.dtype
+            dtype=self.dtype,
         )
 
-        return DaskActiveArray(newarr.dask, newarr.name, newarr.chunks, meta=newarr)
+        return newarr