dev

Author: davidhassell

cf/cfimplementation.py

@@ -48,6 +48,33 @@ class CFImplementation(cfdm.CFDMImplementation):
 
     """
 
+    def nc_set_hdf5_chunksizes(self, data, sizes, override=False):
+        """Set the data HDF5 chunksizes.
+
+        .. versionadded:: NEXTVERSION
+
+        :Parameters:
+
+            data: `Data`
+                The data.
+
+            sizes: sequence of `int`
+                The new HDF5 chunk sizes.
+
+            override: `bool`, optional
+                If True then set the HDF5 chunks sizes even if some
+                have already been specified. If False, the default,
+                then only set the HDF5 chunks sizes if some none have
+                already been specified.
+
+        :Returns:
+
+            `None`
+
+        """
+        if override or not data.nc_hdf5_chunksizes():
+            data.nc_set_hdf5_chunksizes(sizes)
+
     def set_construct(self, parent, construct, axes=None, copy=True, **kwargs):
         """Insert a construct into a field or domain.
 

cf/read_write/netcdf/netcdfwrite.py

@@ -431,10 +431,12 @@ def _create_cfa_data(self, ncvar, ncdimensions, data, cfvar):
             `None`
 
         """
+        import time # TODO
+        print (f"\n{ncvar}") # TODO
+        
         g = self.write_vars
 
         ndim = data.ndim
-
         cfa = self._cfa_aggregation_instructions(data, cfvar)
 
         # ------------------------------------------------------------
@@ -482,16 +484,21 @@ def _create_cfa_data(self, ncvar, ncdimensions, data, cfvar):
         aggregated_data_attr = []
 
         # Location
+        start = time.time() # TODO
         term = "location"
+        data = cfa[term]
+        self.implementation.nc_set_hdf5_chunksizes(data, data.shape)
         term_ncvar = self._cfa_write_term_variable(
-            cfa[term],
+            data ,#cfa[term],
             aggregated_data.get(term, f"cfa_{term}"),
             location_ncdimensions,
         )
         aggregated_data_attr.append(f"{term}: {term_ncvar}")
-
+        print (f"{term:<10}: {time.time() - start:.3}")
+        
         # File
         term = "file"
+        start = time.time() # TODO
         if substitutions:
             # Create the "substitutions" netCDF attribute
             subs = []
@@ -502,16 +509,20 @@ def _create_cfa_data(self, ncvar, ncdimensions, data, cfvar):
         else:
             attributes = None
 
+        data = cfa[term]
+        self.implementation.nc_set_hdf5_chunksizes(data, data.shape)
         term_ncvar = self._cfa_write_term_variable(
-            cfa[term],
+            data, #cfa[term],
             aggregated_data.get(term, f"cfa_{term}"),
             fragment_ncdimensions,
             attributes=attributes,
         )
         aggregated_data_attr.append(f"{term}: {term_ncvar}")
+        print (f"{term:<10}: {time.time() - start:.3}")
 
         # Address
         term = "address"
+        start = time.time() # TODO
 
         # Attempt to reduce addresses to a common scalar value
         u = cfa[term].unique().compressed().persist()
@@ -521,15 +532,19 @@ def _create_cfa_data(self, ncvar, ncdimensions, data, cfvar):
         else:
             dimensions = fragment_ncdimensions
 
+        data = cfa[term]
+        self.implementation.nc_set_hdf5_chunksizes(data, data.shape)
         term_ncvar = self._cfa_write_term_variable(
-            cfa[term],
+            data, #         cfa[term],
             aggregated_data.get(term, f"cfa_{term}"),
             dimensions,
         )
         aggregated_data_attr.append(f"{term}: {term_ncvar}")
+        print (f"{term:<10}: {time.time() - start:.3}")
 
         # Format
         term = "format"
+        start = time.time() # TODO
 
         # Attempt to reduce addresses to a common scalar value
         u = cfa[term].unique().compressed().persist()
@@ -539,13 +554,16 @@ def _create_cfa_data(self, ncvar, ncdimensions, data, cfvar):
         else:
             dimensions = fragment_ncdimensions
 
+        data = cfa[term]
+        self.implementation.nc_set_hdf5_chunksizes(data, data.shape)
         term_ncvar = self._cfa_write_term_variable(
-            cfa[term],
+            data, #cfa[term],
             aggregated_data.get(term, f"cfa_{term}"),
             dimensions,
         )
         aggregated_data_attr.append(f"{term}: {term_ncvar}")
 
+        print (f"{term:<10}: {time.time() - start:.3}")
         # ------------------------------------------------------------
         # Look for non-standard CFA terms stored as field ancillaries
         # on a field and write them to the CFA-netCDF file
@@ -809,8 +827,10 @@ def _cfa_write_non_standard_terms(
             terms.append(term)
 
             # Create the new CFA term variable
+            data = type(data)(dx)
+            self.implementation.nc_set_hdf5_chunksizes(data, data.shape)
             term_ncvar = self._cfa_write_term_variable(
-                data=type(data)(dx),
+                data=data, #type(data)(dx),
                 ncvar=aggregated_data.get(term, f"cfa_{term}"),
                 ncdimensions=fragment_ncdimensions,
             )
@@ -884,7 +904,9 @@ def _cfa_aggregation_instructions(self, data, cfvar):
         from os.path import abspath, join, relpath
         from pathlib import PurePath
         from urllib.parse import urlparse
-
+        import time # TODO
+        start = time.time() # TODO
+        
         g = self.write_vars
 
         # Define the CFA file susbstitutions, giving precedence over
@@ -899,11 +921,15 @@ def _cfa_aggregation_instructions(self, data, cfvar):
         # Size of the trailing dimension
         n_trailing = 0
 
+        start1 = time.time() # TODO
         aggregation_file = []
         aggregation_address = []
         aggregation_format = []
+        nnn = 0
         for indices in data.chunk_indices():
+            nnn += 1
             file_details = self._cfa_get_file_details(data[indices])
+       
             if len(file_details) != 1:
                 if file_details:
                     raise ValueError(
@@ -948,7 +974,9 @@ def _cfa_aggregation_instructions(self, data, cfvar):
             aggregation_file.append(tuple(filenames2))
             aggregation_address.append(addresses)
             aggregation_format.append(formats)
-
+        print ('len(data.chunk_indices()) =',nnn)
+        print (f"loop 1: {time.time() - start1:.3}")
+       
         # Pad each value of the aggregation instruction arrays so that
         # it has 'n_trailing' elements
         a_shape = data.numblocks
@@ -1005,6 +1033,7 @@ def _cfa_aggregation_instructions(self, data, cfvar):
         # Return Data objects
         # ------------------------------------------------------------
         data = type(data)
+        print (f"_cfa_aggregation_instructions: {time.time() - start:.3}")
         return {
             "location": data(aggregation_location),
             "file": data(aggregation_file),