Merge pull request #4 from dwest77a/functional

2024.9.0 Final Update

Author: dwest77a

README.md

@@ -1,19 +1,19 @@
 # XarrayActive
-For use with the Xarray module as an additional backend.
+For use with the Xarray module as an additional backend. See the module[PyActiveStorage](https://github.com/NCAS-CMS/PyActiveStorage) for more details.
 
 ## Installation
 
 ```
 pip install xarray==2024.6.0
-pip install -e .
+pip install XarrayActive==2024.9.0
 ```
 
 ## Usage
 
 ```
 import xarray as xr
 
-ds = xr.open_dataset('cfa_file.nc', engine='Active')
+ds = xr.open_dataset('any_file.nc', engine='Active')
 # Plot data
 
 ```

XarrayActive/__init__.py

@@ -1,4 +1,4 @@
-from .active_xarray import ActiveDataset    # Used by CFAPyX 
-from .active_dask import DaskActiveArray    # Used by CFAPyX
-from .active_chunk import ActiveChunk       # Used by CFAPyX
-from .backend import ActiveBackendEntrypoint
+from XarrayActive.active_xarray import ActiveDataset    # Used by CFAPyX 
+from XarrayActive.active_dask import DaskActiveArray    # Used by CFAPyX
+from XarrayActive.active_chunk import ActiveChunk       # Used by CFAPyX
+from XarrayActive.backend import ActiveBackendEntrypoint

XarrayActive/__pycache__/__init__.cpython-312.pyc

@@ -4,7 +4,7 @@
 
 class ActiveOptionsContainer:
     """
-    Container for ActiveOptions properties.
+    Container for ActiveOptions properties. Only for use within XarrayActive.
     """
     @property
     def active_options(self):
@@ -23,28 +23,40 @@ def active_options(self, value):
 
     def _set_active_options(self, chunks={}, chunk_limits=True):
 
+        # Auto chunking is not currently supported - 23/08/24
         if chunks == {}:
             raise NotImplementedError(
                 'Default chunking is not implemented, please provide a chunk scheme '
                 ' - active_options = {"chunks": {}}'
             )
+        
+        if 'auto' in chunks.items():
+            raise NotImplementedError(
+                'Auto 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:
+    """
+    Container class for all Active-required methods to perform on each chunk. 
+    All active-per-chunk content should be found here.
+    """
 
-    description = "Container class for Active routines performed on each chunk. All active-per-chunk content can be found here."
+    description = "Container class for Active routines performed on each chunk."
 
     def _post_process_data(self, data):
-        # Perform any post-processing steps on the data here
+        """
+        Perform any post-processing steps on the data here.
+        """
         return data
 
     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.
+        Standard sum routine matches the normal routine for dask, required at this
+        stage if Active mean/sum not available.
         """
 
         arr = np.array(self)
@@ -55,12 +67,31 @@ def _standard_sum(self, axes=None, skipna=None, **kwargs):
         return total
 
     def _standard_max(self, axes=None, skipna=None, **kwargs):
+        """
+        Standard max routine if Active not available, warning will be given.
+        Kwargs may be necessary to add here.
+        """
         return np.max(self, axis=axes)
 
     def _standard_min(self, axes=None, skipna=None, **kwargs):
+        """
+        Standard min routine if Active not available, warning will be given.
+        Kwargs may be necessary to add here.
+        """
         return np.min(self, axis=axes)
 
     def _numel(self, method, axes=None):
+        """
+        Number of elements remaining after a reduction, to allow
+        dask to combine reductions from all different chunks.
+        Example:
+            (2,3,4) chunk reduced along second dimension. Will
+            give a (2,3) array where each value is 4 - for the 
+            length of the dimension along which a reduction
+            took place.
+
+        """
+        # Applied reduction across all axes
         if not axes:
             return self.size
 
@@ -91,41 +122,47 @@ def active_method(self, method, axis=None, skipna=None, **kwargs):
             'max' : self._standard_max,
             'min' : self._standard_min
         }
-        ret = None
+        partial = 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.")
-            ret = {
+            partial = {
                 'n': n,
                 'total': standard_methods[method](axes=axis, skipna=skipna, **kwargs)
             }
 
-        if not ret:
+        if not partial:
 
+            # Create Active client
             active = Active(self.filename, self.address)
             active.method = method
+
+            # Fetch extent for this chunk instance.
             extent = tuple(self.get_extent())
 
+            # Properly format the 'axis' kwarg.
             if axis == None:
                 axis = tuple([i for i in range(self.ndim)])
 
+            # Determine reduction parameter for combining chunk results for dask.
             n = self._numel(method, axes=axis)
 
             if len(axis) == self.ndim:
                 data   = active[extent]
                 t = self._post_process_data(data) * n
 
-                ret = {
+                partial = {
                     'n': n,
                     'total': t
                 }
 
-        if not ret:
+        if not partial:
             # Experimental Recursive requesting to get each 1D column along the axes being requested.
+            # - May be very bad performance due to many requests for (1,1,X) shapes
             range_recursives = []
             for dim in range(self.ndim):
                 if dim not in axis:
@@ -135,17 +172,21 @@ def active_method(self, method, axis=None, skipna=None, **kwargs):
             results = np.array(self._get_elements(active, range_recursives, hyperslab=[]))
 
             t = self._post_process_data(results) * n
-            ret = {
+            partial = {
                 'n': n,
                 'total': t
             }
 
         if method == 'mean':
-            return ret
+            return partial
         else:
-            return ret['total']/ret['n']
+            return partial['total']/partial['n']
 
     def _get_elements(self, active, recursives, hyperslab=[]):
+        """
+        Recursive function to fetch and arrange the appropriate column slices
+        from Active.
+        """
         dimarray = []
         if not len(recursives) > 0:
 

XarrayActive/__pycache__/active_chunk.cpython-312.pyc

@@ -4,6 +4,10 @@
 from dask.array.core import _concatenate2
 import numpy as np
 
+from .active_chunk import ActiveChunk
+
+## Partition Methods are the first step in the Dask Reductions.
+
 def partition_mean(arr, *args, **kwargs):
     return partition_method(arr, 'mean', *args, **kwargs)
 
@@ -18,14 +22,19 @@ def partition_sum(arr, *args, **kwargs):
 
 def partition_method(arr, method, *args, **kwargs):
     if hasattr(arr,'active_method'):
+        # Active method for each array partition
         return arr.active_method(method,*args, **kwargs)
+    elif arr.size != 0:
+        print('ActiveWarning: Using standard mean given non-active array partition')
+        return arr.mean(*args, **kwargs)
     else:
-        # Additional handling for 'meta' calculations in dask.
-        # Not currently implemented, bypassed using None
-        if arr.size == 0:
-            return None
+        # Computing meta - dask operation not fully utilised.
         return None
 
+## Combining results from Partition methods
+# - Dask built-in mean-agg and mean-combine for mean.
+# - Min/Max/Sum require simple functions for combine/aggregation.
+
 def general_combine(pairs, axis=None):
     if not isinstance(pairs, list):
         pairs = [pairs]
@@ -46,14 +55,15 @@ class DaskActiveArray(da.Array):
 
     @property
     def is_active(self):
+        # Quick way of distinguishing from Dask Array
         return True
 
-    #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 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):
         """
@@ -78,7 +88,6 @@ def active_mean(self, axis=None, skipna=None):
         :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_mean,
@@ -163,4 +172,17 @@ def active_sum(self, axis=None, skipna=None):
             dtype=self.dtype,
         )
 
-        return newarr
+        return newarr
+    
+    def active_method(self, method, axis=None, skipna=None, **kwargs):
+        """
+        Pointer to the active methods of the DaskActiveArray, for use
+        in the nested Dask setup with CFA Dask-AnySize Chunks.
+        """
+        methods = {
+            'mean':self.active_mean,
+            'max':self.active_max,
+            'min':self.active_min,
+            'sum':self.active_sum
+        }
+        return methods[method](axis=axis, skipna=skipna)