From ae46b075c7b7809c9adfc04ea79aaa1887ca89ae Mon Sep 17 00:00:00 2001 From: dwest77 Date: Tue, 2 Dec 2025 17:02:50 +0000 Subject: [PATCH 1/7] New changes for combine functionality --- cfapyx/creator.py | 224 +++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 220 insertions(+), 4 deletions(-) diff --git a/cfapyx/creator.py b/cfapyx/creator.py index 03269f7..d5d180e 100644 --- a/cfapyx/creator.py +++ b/cfapyx/creator.py @@ -18,6 +18,164 @@ CONCAT_MSG = 'See individual datasets for more information.' +class CFACombineMixin: + """ + Methods applying specifically to the combining of aggregated files. + + These methods override the normal create functions for an aggregation file, + as much of the work has been done for us at this point. + + NOTE: Aggregation will be limited to dimensions that are already aggregated. i.e This + feature will only work on files where aggregation dimensions are consistent (i.e extending dimensions). + + These functions must: + - Identify the dimension(s) that are being extended and identify the affected variables. + - Combine scalar dimensions `f_` if those dimensions are already greater than 1. + - Identify the affected fragment constructors (map, uris) and combine along the extending dimensions. + """ + + def combine(self, arranged_files: tuple, global_attrs: dict, var_info: dict, dim_info: dict): + """ + Combine arranged files according to their coordinates. + """ + st_dim_info = {} + + # Identify Extending dimensions + ext_dims = [] + for d, info in dim_info.items(): + if not isinstance(info, list): + continue + if 'f_' not in d: + continue + + for fileinst in info: + if fileinst['size'] != 1: + ext_dims.append(d) + break + + if d not in ext_dims: + # Non extending aggregation dimension + rd = d.split('_')[-1] + + # Skip coordinate dimensions here + if not isinstance(dim_info[rd], dict): + if len(set([i['size'] for i in dim_info[rd]])) != 1: + raise ValueError( + f'Non-extending dimension {rd} differs in size between files' + ) + st_dim_info[rd] = dim_info[rd][0] + else: + st_dim_info[rd] = dim_info[rd] + if len(set(dim_info[rd]['sizes'])) != 1: + raise ValueError( + f'Non-extending dimension {rd} differs in size between files' + ) + st_dim_info[rd]['size'] = dim_info[rd]['sizes'][0] + st_dim_info[rd]['f_size'] = dim_info[d][0]['size'] + + new_dim_sizes = {} + for d in ext_dims: + real_dim = d.split('_')[-1] + new_dim_sizes[d] = sum([i['size'] for i in dim_info.pop(d)]) + + # Dim info transformations for writing + st_dim_info[real_dim] = dim_info[real_dim] + st_dim_info[real_dim]['f_size'] = new_dim_sizes[d] + st_dim_info[real_dim]['size'] = sum(dim_info[real_dim]['sizes']) + + # Sort array components using index list + sorted_a = [[] for a in dim_info[real_dim]['arrays']] + sorted_starts = np.argsort(dim_info[real_dim]['starts']) + for x, a in enumerate(dim_info[real_dim]['arrays']): + + sorted_a[sorted_starts[x]] = a + + # Assumes they have been sorted - not necessarily the case + array = np.concatenate( + sorted_a, + axis=0 + ) + + st_dim_info[real_dim]['arrays'] = array + + sorted_dim_sizes = sorted([ + (v, d) for d, v in new_dim_sizes.items()], + key = lambda x: x[0] + ) + + orders = [np.argsort(dim_info[d.split('_')[-1]]['starts']) for d in ext_dims] + + st_var_info = { + v: info for v, info in var_info.items() if 'fragment_' not in v + } + + # Identify the affected fragment components + extension_vars = [] + for v, info in var_info.items(): + for d in ext_dims: + if d in info['dims']: + extension_vars.append(v) + + # Perform ordering/concatenation for fragment components + for ev in extension_vars: + + st_var_info[ev] = var_info[ev] + # Sort array components using index list + sorted_a = [[] for a in var_info[ev]['arr']] + for x, a in enumerate(var_info[ev]['arr']): + + for nd in sorted_dim_sizes: + if nd[1] in var_info[ev]['dims']: + dominant_dim = nd[1] + break + + sorted_a[ + orders[ext_dims.index(dominant_dim)][x] + ] = a + + # Assumes they have been sorted - not necessarily the case + array = np.concatenate( + sorted_a, + axis=var_info[ev]['dims'].index(dominant_dim) + ) + + st_var_info[ev]['arr'] = array + + # Writing part done as a second step as before. + # Need to perform any transformations to get to that stage. + + # Define the fragment space + self.fragment_space = [v['f_size'] for v in dim_info.values() if 'f_size' in v] + + # Assemble the location with correct dimensions + location = self._assemble_expanded_location(arranged_files, st_dim_info) + for coord, file in arranged_files.items(): + newcoord = [] + for x, c in enumerate(coord): + dx = orders[x].index(c) + newcoord.append(coord) + location[tuple(newcoord)] = file + + self.global_attrs = global_attrs + self.dim_info = st_dim_info + self.var_info = st_var_info + self.location = location + + def _assemble_expanded_location(arranged_files, orders): + """ + Custom function for assembling location for these new file types. + + Won't work like this because we need to reformat location + """ + location = {} + for coord, file in arranged_files.items(): + newcoord = [] + for x, c in enumerate(coord): + dx = orders[x].index(c) + newcoord.append(coord) + location[tuple(newcoord)] = file + return location + class CFACreateMixin: """ Mixin class for ``Create`` methods for a CFA-netCDF dataset. @@ -41,6 +199,7 @@ def _first_pass(self, agg_dims: list = None) -> tuple: ## First Pass - Determine dimensions for x, file in enumerate(self.files): + is_aggregated = False logger.info(f'First pass: File {x+1}/{len(self.files)}') ds = self._call_file(file) @@ -51,6 +210,18 @@ def _first_pass(self, agg_dims: list = None) -> tuple: all_dims = ds.dimensions.keys() all_vars = ds.variables.keys() + # Determine aggregation combine or fragment combine + for v in all_vars: + if hasattr(ds[v],'aggregated_dimensions'): + is_aggregated = True + if self.agg_combine is None: + self.agg_combine = True + elif not self.agg_combine: + raise ValueError( + 'Mixed file types not allowed. Can only combine' + ' fragment sets or aggregation sets.' + ) + coord_variables = [] pure_dimensions = [] variables = [] @@ -71,6 +242,7 @@ def _first_pass(self, agg_dims: list = None) -> tuple: if not var_info: var_info = {v: {} for v in variables} + # Should be consistent across fragments or aggregation files logger.info(f'Coordinate variables: {coord_variables}') logger.info(f'Pure dimensions: {pure_dimensions}') logger.info(f'Variables: {variables}') @@ -105,8 +277,17 @@ def _first_pass(self, agg_dims: list = None) -> tuple: if new_info['type'] == 'coord': # Only coordinate dimensions can have attributes dim_info = self._update_info(ds[d], dim_info, new_info) + if is_aggregated: + if 'sizes' not in dim_info[d]: + dim_info[d]['sizes'] = [] + dim_info[d]['sizes'].append(arr_components['sizes']) else: - dim_info[d] = new_info + if is_aggregated and (dim_info[d] != {} or isinstance(dim_info[d], list)): + if not isinstance(dim_info[d], list): + dim_info[d] = [dim_info[d]] + dim_info[d].append(new_info) + else: + dim_info[d] = new_info if arr_components is not None: if first_time: @@ -141,8 +322,28 @@ def _first_pass(self, agg_dims: list = None) -> tuple: '_FillValue': fill, } + if is_aggregated: + # Special handling for extraction of string variables + # Easier to keep numpy arrays separate if they are wrapped in lists + if ds[v].size == 1: + new_info['arr'] = [np.array(ds[v][0], dtype=ds[v].dtype)] + else: + new_info['arr'] = [np.array(list(ds[v]), dtype=ds[v].dtype)] + var_info = self._update_info(ds[v], var_info, new_info) + # No variables in current file are aggregations + if self.agg_combine is None and not is_aggregated: + self.agg_combine = is_aggregated + + # Any variables in current file are aggregated, while previous + # files were not. + if self.agg_combine != is_aggregated: + raise ValueError( + 'Mixed file types not allowed. Can only combine' + ' fragment sets or aggregation sets.' + ) + arranged_files[tuple(fcoord)] = file return arranged_files, global_attrs, var_info, dim_info @@ -164,7 +365,7 @@ def _second_pass( for x, file in enumerate(second_set): logger.info(f'Second pass: File {x+1}/{len(self.files)}') - ds = self._call_file(file) # Ideally don't want to do this twice. + ds = self._call_file(file) # Typically have to do this twice. for v in non_aggregated: new_values = np.array(ds.variables[v]) @@ -261,7 +462,11 @@ def _update_info( info[id]['attrs'] = self._accumulate_attrs(info[id]['attrs'], attrs) for attr, value in new_info.items(): - if value != info[id][attr]: + if attr == 'arr': + if 'arr' not in info[id]: + info[id]['arr'] = [] + info[id]['arr'] += value + elif value != info[id][attr]: if np.isnan(value) and np.isnan(info[id][attr]): pass else: @@ -750,7 +955,7 @@ def _write_nonagg_variable( if 'data' in meta: var_arr[:] = meta['data'] -class CFANetCDF(CFACreateMixin, CFAWriteMixin): +class CFANetCDF(CFACreateMixin, CFAWriteMixin, CFACombineMixin): """ CFA-netCDF file constructor class, enables creation and @@ -768,6 +973,8 @@ def __init__(self, files: list, concat_msg : str = CONCAT_MSG): the provided set of files. A custom concat message can also be set here if needed.""" + self.agg_combine = None + if isinstance(files, str): fileset = glob.glob(files) self.files = self._filter_files(fileset) @@ -816,6 +1023,15 @@ def create( # First pass collect info arranged_files, global_attrs, var_info, dim_info = self._first_pass(agg_dims=agg_dims) + if self.agg_combine: + self.combine( + arranged_files, + global_attrs, + var_info, + dim_info + ) + return + global_attrs, var_info, dim_info = self._apply_filters(updates, removals, global_attrs, var_info, dim_info) # Arrange aggregation dimensions From 205f4e3b0c23d7fddc6a67fe1e47c5de7bf8a7bf Mon Sep 17 00:00:00 2001 From: dwest77 Date: Wed, 3 Dec 2025 14:42:09 +0000 Subject: [PATCH 2/7] Finalised extension functionality features --- cfapyx/creator.py | 366 +++++++++++++++++++++++++--------------------- 1 file changed, 198 insertions(+), 168 deletions(-) diff --git a/cfapyx/creator.py b/cfapyx/creator.py index d5d180e..5d2d704 100644 --- a/cfapyx/creator.py +++ b/cfapyx/creator.py @@ -18,164 +18,6 @@ CONCAT_MSG = 'See individual datasets for more information.' -class CFACombineMixin: - """ - Methods applying specifically to the combining of aggregated files. - - These methods override the normal create functions for an aggregation file, - as much of the work has been done for us at this point. - - NOTE: Aggregation will be limited to dimensions that are already aggregated. i.e This - feature will only work on files where aggregation dimensions are consistent (i.e extending dimensions). - - These functions must: - - Identify the dimension(s) that are being extended and identify the affected variables. - - Combine scalar dimensions `f_` if those dimensions are already greater than 1. - - Identify the affected fragment constructors (map, uris) and combine along the extending dimensions. - """ - - def combine(self, arranged_files: tuple, global_attrs: dict, var_info: dict, dim_info: dict): - """ - Combine arranged files according to their coordinates. - """ - st_dim_info = {} - - # Identify Extending dimensions - ext_dims = [] - for d, info in dim_info.items(): - if not isinstance(info, list): - continue - if 'f_' not in d: - continue - - for fileinst in info: - if fileinst['size'] != 1: - ext_dims.append(d) - break - - if d not in ext_dims: - # Non extending aggregation dimension - rd = d.split('_')[-1] - - # Skip coordinate dimensions here - if not isinstance(dim_info[rd], dict): - if len(set([i['size'] for i in dim_info[rd]])) != 1: - raise ValueError( - f'Non-extending dimension {rd} differs in size between files' - ) - st_dim_info[rd] = dim_info[rd][0] - else: - st_dim_info[rd] = dim_info[rd] - if len(set(dim_info[rd]['sizes'])) != 1: - raise ValueError( - f'Non-extending dimension {rd} differs in size between files' - ) - st_dim_info[rd]['size'] = dim_info[rd]['sizes'][0] - st_dim_info[rd]['f_size'] = dim_info[d][0]['size'] - - new_dim_sizes = {} - for d in ext_dims: - real_dim = d.split('_')[-1] - new_dim_sizes[d] = sum([i['size'] for i in dim_info.pop(d)]) - - # Dim info transformations for writing - st_dim_info[real_dim] = dim_info[real_dim] - st_dim_info[real_dim]['f_size'] = new_dim_sizes[d] - st_dim_info[real_dim]['size'] = sum(dim_info[real_dim]['sizes']) - - # Sort array components using index list - sorted_a = [[] for a in dim_info[real_dim]['arrays']] - sorted_starts = np.argsort(dim_info[real_dim]['starts']) - for x, a in enumerate(dim_info[real_dim]['arrays']): - - sorted_a[sorted_starts[x]] = a - - # Assumes they have been sorted - not necessarily the case - array = np.concatenate( - sorted_a, - axis=0 - ) - - st_dim_info[real_dim]['arrays'] = array - - sorted_dim_sizes = sorted([ - (v, d) for d, v in new_dim_sizes.items()], - key = lambda x: x[0] - ) - - orders = [np.argsort(dim_info[d.split('_')[-1]]['starts']) for d in ext_dims] - - st_var_info = { - v: info for v, info in var_info.items() if 'fragment_' not in v - } - - # Identify the affected fragment components - extension_vars = [] - for v, info in var_info.items(): - for d in ext_dims: - if d in info['dims']: - extension_vars.append(v) - - # Perform ordering/concatenation for fragment components - for ev in extension_vars: - - st_var_info[ev] = var_info[ev] - # Sort array components using index list - sorted_a = [[] for a in var_info[ev]['arr']] - for x, a in enumerate(var_info[ev]['arr']): - - for nd in sorted_dim_sizes: - if nd[1] in var_info[ev]['dims']: - dominant_dim = nd[1] - break - - sorted_a[ - orders[ext_dims.index(dominant_dim)][x] - ] = a - - # Assumes they have been sorted - not necessarily the case - array = np.concatenate( - sorted_a, - axis=var_info[ev]['dims'].index(dominant_dim) - ) - - st_var_info[ev]['arr'] = array - - # Writing part done as a second step as before. - # Need to perform any transformations to get to that stage. - - # Define the fragment space - self.fragment_space = [v['f_size'] for v in dim_info.values() if 'f_size' in v] - - # Assemble the location with correct dimensions - location = self._assemble_expanded_location(arranged_files, st_dim_info) - for coord, file in arranged_files.items(): - newcoord = [] - for x, c in enumerate(coord): - dx = orders[x].index(c) - newcoord.append(coord) - location[tuple(newcoord)] = file - - self.global_attrs = global_attrs - self.dim_info = st_dim_info - self.var_info = st_var_info - self.location = location - - def _assemble_expanded_location(arranged_files, orders): - """ - Custom function for assembling location for these new file types. - - Won't work like this because we need to reformat location - """ - location = {} - for coord, file in arranged_files.items(): - newcoord = [] - for x, c in enumerate(coord): - dx = orders[x].index(c) - newcoord.append(coord) - location[tuple(newcoord)] = file - return location - class CFACreateMixin: """ Mixin class for ``Create`` methods for a CFA-netCDF dataset. @@ -280,7 +122,11 @@ def _first_pass(self, agg_dims: list = None) -> tuple: if is_aggregated: if 'sizes' not in dim_info[d]: dim_info[d]['sizes'] = [] - dim_info[d]['sizes'].append(arr_components['sizes']) + if 'starts' not in dim_info[d]: + dim_info[d]['starts'] = [] + #dim_info[d]['sizes'].append(arr_components['sizes']) + # This is to prevent the line 141 below triggering again. + #dim_info[d]['starts'].append(arr_components['starts']) else: if is_aggregated and (dim_info[d] != {} or isinstance(dim_info[d], list)): if not isinstance(dim_info[d], list): @@ -305,7 +151,7 @@ def _first_pass(self, agg_dims: list = None) -> tuple: for v in variables: try: - fill = ds[v].getncattr('_FillValue') + fill = ds[v].get_fill_value() except: fill = None @@ -715,7 +561,7 @@ def _write_dimensions(self): for dim, di in self.dim_info.items(): - f_size = di['f_size'] + f_size = di.get('f_size', None) dim_size = di['size'] real_part = self.ds.createDimension( @@ -723,10 +569,11 @@ def _write_dimensions(self): dim_size ) - frag_part = self.ds.createDimension( - f'f_{dim}', - f_size, - ) + if f_size is not None: + frag_part = self.ds.createDimension( + f'f_{dim}', + f_size, + ) f_dims[f'f_{dim}'] = f_size @@ -953,9 +800,12 @@ def _write_nonagg_variable( logger.warning(err) if 'data' in meta: - var_arr[:] = meta['data'] + if meta['dtype'] == str: + var_arr[:] = np.array(meta['data'], dtype=meta['dtype']) + else: + var_arr[:] = meta['data'] -class CFANetCDF(CFACreateMixin, CFAWriteMixin, CFACombineMixin): +class CFANetCDF(CFACreateMixin, CFAWriteMixin): """ CFA-netCDF file constructor class, enables creation and @@ -1024,7 +874,7 @@ def create( arranged_files, global_attrs, var_info, dim_info = self._first_pass(agg_dims=agg_dims) if self.agg_combine: - self.combine( + self._combine( arranged_files, global_attrs, var_info, @@ -1089,6 +939,14 @@ def write( f_dims['versions'] = self.max_files + if self.agg_combine: + + # Skip writing shapes + + self._write_variables() + self.ds.close() + return + self._write_shape_dims(f_dims) self._write_fragment_shapes() self._write_fragment_addresses() @@ -1128,6 +986,178 @@ def handle_conventions(self, value) -> str: ) return delim.join(updated_conventions) + def _combine(self, arranged_files: tuple, global_attrs: dict, var_info: dict, dim_info: dict): + """ + Combine arranged files according to their coordinates. + + NOTE: Aggregation will be limited to dimensions that are already aggregated. i.e This + feature will only work on files where aggregation dimensions are consistent (i.e extending dimensions). + + These functions must: + - Identify the dimension(s) that are being extended and identify the affected variables. + - Combine scalar dimensions `f_` if those dimensions are already greater than 1. + - Identify the affected fragment constructors (map, uris) and combine along the extending dimensions. + """ + st_dim_info = {} + + # Identify Extending dimensions + ext_dims = [] + for d, info in dim_info.items(): + + if not isinstance(info, list): + # Coordinate variables are not collected into a list + # Default value may be overridden by future construction + + #st_dim_info[d] = info + #st_dim_info[d]['arrays'] = info['arrays'][0] + continue + + if 'f_' not in d: + if d not in st_dim_info: + + # Default value - reset using f_ numbers if needed + if isinstance(info, list): + if info[0]['size'] != info[-1]['size']: + raise ValueError( + 'Aggregation not possible for differing non-aggregated values.' + ) + st_dim_info[d] = info[0] + else: + st_dim_info[d] = info + st_dim_info[d]['array'] = st_dim_info[d].get('arrays',None) + + if 'map_' in d: + # Constructor dimension - don't add more f_dims + st_dim_info[d].pop('f_size') + + continue + + for fileinst in info: + if fileinst['size'] != 1: + ext_dims.append(d) + break + + if d not in ext_dims: + # Non extending aggregation dimension + rd = d.split('_')[-1] + + # Skip coordinate dimensions here + if not isinstance(dim_info[rd], dict): + if len(set([i['size'] for i in dim_info[rd]])) != 1: + raise ValueError( + f'Non-extending dimension {rd} differs in size between files' + ) + st_dim_info[rd] = dim_info[rd][0] + else: + st_dim_info[rd] = dim_info[rd] + st_dim_info[rd]['array'] = st_dim_info[rd]['arrays'][0] + if len(set(dim_info[rd]['sizes'])) != 1: + raise ValueError( + f'Non-extending dimension {rd} differs in size between files' + ) + st_dim_info[rd]['size'] = dim_info[rd]['sizes'][0] + st_dim_info[rd]['f_size'] = info[0]['size'] + + new_dim_sizes = {} + for d in ext_dims: + real_dim = d.split('_')[-1] + new_dim_sizes[d] = sum([i['size'] for i in dim_info[d]]) + + # Dim info transformations for writing + st_dim_info[real_dim] = dim_info[real_dim] + st_dim_info[real_dim]['f_size'] = new_dim_sizes[d] + st_dim_info[real_dim]['size'] = sum(dim_info[real_dim]['sizes']) + + # Sort array components using index list + sorted_a = [[] for a in dim_info[real_dim]['arrays']] + sorted_starts = np.argsort(dim_info[real_dim]['starts']) + for x, a in enumerate(dim_info[real_dim]['arrays']): + + sorted_a[sorted_starts[x]] = a + + array = np.concatenate( + sorted_a, + axis=0 + ) + + st_dim_info[real_dim]['array'] = array + + + sorted_dim_sizes = sorted([ + (v, d) for d, v in new_dim_sizes.items()], + key = lambda x: x[0] + ) + + orders = [np.argsort(dim_info[d.split('_')[-1]]['starts']) for d in ext_dims] + + # Identify the affected fragment components + extension_vars = [] + for v, info in var_info.items(): + extended = False + for d in ext_dims: + if d in info['dims']: + extension_vars.append(v) + + if not extended: + var_info[v]['data'] = var_info[v]['arr'][0] + + # Perform ordering/concatenation for fragment components + for ev in extension_vars: + + # Sort array components using index list + sorted_a = [[] for a in var_info[ev]['arr']] + for x, a in enumerate(var_info[ev]['arr']): + + for nd in sorted_dim_sizes: + if nd[1] in var_info[ev]['dims']: + dominant_dim = nd[1] + break + + sorted_a[ + orders[ext_dims.index(dominant_dim)][x] + ] = a + + # Assumes they have been sorted - not necessarily the case + array = np.concatenate( + sorted_a, + axis=var_info[ev]['dims'].index(dominant_dim) + ) + + if 'fragment_map_' in ev: + + array = np.ma.masked_values(array, var_info[ev]['_FillValue']) + # Smooth paddings + for i, row in enumerate(array): + premask = None + gap = False + for j, value in enumerate(row): + if premask is None and not np.ma.is_masked(value): + premask = value + elif np.ma.is_masked(value): + gap = True + elif not np.ma.is_masked(value) and gap: + if value == premask: + array[i][j] = np.ma.masked + + var_info[ev]['data'] = array + + # Writing part done as a second step as before. + # Need to perform any transformations to get to that stage. + + # Define the fragment space + self.fragment_space = [v['f_size'] for v in dim_info.values() if 'f_size' in v] + + self.global_attrs = global_attrs + self.dim_info = st_dim_info + self.var_info = var_info + + def _write_extension(self): + """ + Stremlines writing extended `combined` file where certain arrangements are not needed. + """ + + pass + def display_attrs(self): """ Display the global attributes consolidated in the From 2da68d3c80d240190cabf5611d11275c026d98b6 Mon Sep 17 00:00:00 2001 From: dwest77 Date: Wed, 3 Dec 2025 14:44:14 +0000 Subject: [PATCH 3/7] Reworked to use terminology --- cfapyx/creator.py | 41 +++++++++++++++++------------------------ 1 file changed, 17 insertions(+), 24 deletions(-) diff --git a/cfapyx/creator.py b/cfapyx/creator.py index 5d2d704..ad10e3a 100644 --- a/cfapyx/creator.py +++ b/cfapyx/creator.py @@ -52,15 +52,15 @@ def _first_pass(self, agg_dims: list = None) -> tuple: all_dims = ds.dimensions.keys() all_vars = ds.variables.keys() - # Determine aggregation combine or fragment combine + # Determine if standard aggregation or fragment extension for v in all_vars: if hasattr(ds[v],'aggregated_dimensions'): is_aggregated = True - if self.agg_combine is None: - self.agg_combine = True - elif not self.agg_combine: + if self.agg_extend is None: + self.agg_extend = True + elif not self.agg_extend: raise ValueError( - 'Mixed file types not allowed. Can only combine' + 'Mixed file types not allowed. Can only extend' ' fragment sets or aggregation sets.' ) @@ -179,14 +179,14 @@ def _first_pass(self, agg_dims: list = None) -> tuple: var_info = self._update_info(ds[v], var_info, new_info) # No variables in current file are aggregations - if self.agg_combine is None and not is_aggregated: - self.agg_combine = is_aggregated + if self.agg_extend is None and not is_aggregated: + self.agg_extend = is_aggregated # Any variables in current file are aggregated, while previous # files were not. - if self.agg_combine != is_aggregated: + if self.agg_extend != is_aggregated: raise ValueError( - 'Mixed file types not allowed. Can only combine' + 'Mixed file types not allowed. Can only extend' ' fragment sets or aggregation sets.' ) @@ -823,7 +823,7 @@ def __init__(self, files: list, concat_msg : str = CONCAT_MSG): the provided set of files. A custom concat message can also be set here if needed.""" - self.agg_combine = None + self.agg_extend = None if isinstance(files, str): fileset = glob.glob(files) @@ -873,8 +873,8 @@ def create( # First pass collect info arranged_files, global_attrs, var_info, dim_info = self._first_pass(agg_dims=agg_dims) - if self.agg_combine: - self._combine( + if self.agg_extend: + self._extend( arranged_files, global_attrs, var_info, @@ -939,7 +939,7 @@ def write( f_dims['versions'] = self.max_files - if self.agg_combine: + if self.agg_extend: # Skip writing shapes @@ -986,17 +986,17 @@ def handle_conventions(self, value) -> str: ) return delim.join(updated_conventions) - def _combine(self, arranged_files: tuple, global_attrs: dict, var_info: dict, dim_info: dict): + def _extend(self, arranged_files: tuple, global_attrs: dict, var_info: dict, dim_info: dict): """ - Combine arranged files according to their coordinates. + Extend arranged files according to their coordinates. NOTE: Aggregation will be limited to dimensions that are already aggregated. i.e This feature will only work on files where aggregation dimensions are consistent (i.e extending dimensions). These functions must: - Identify the dimension(s) that are being extended and identify the affected variables. - - Combine scalar dimensions `f_` if those dimensions are already greater than 1. - - Identify the affected fragment constructors (map, uris) and combine along the extending dimensions. + - Extend scalar dimensions `f_` if those dimensions are already greater than 1. + - Identify the affected fragment constructors (map, uris) and extend along the extending dimensions. """ st_dim_info = {} @@ -1151,13 +1151,6 @@ def _combine(self, arranged_files: tuple, global_attrs: dict, var_info: dict, di self.dim_info = st_dim_info self.var_info = var_info - def _write_extension(self): - """ - Stremlines writing extended `combined` file where certain arrangements are not needed. - """ - - pass - def display_attrs(self): """ Display the global attributes consolidated in the From b04b0c42ea4640e289f54a952a638049ff2a2eb2 Mon Sep 17 00:00:00 2001 From: dwest77 Date: Wed, 3 Dec 2025 14:44:34 +0000 Subject: [PATCH 4/7] Updated for pre-release --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 0c42b90..67434cc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "cfapyx" -version = "2025.12.2" +version = "2025.12.3a" description = "The pure-Python implementation of the CF Aggregation conventions, including the Xarray engine to enable reading CFA-netCDF files." authors = [ {name = "Daniel Westwood",email = "daniel.westwood@stfc.ac.uk"} From 1a9dab9705586b9fcd03041d0e56913a8c73f681 Mon Sep 17 00:00:00 2001 From: dwest77 Date: Wed, 3 Dec 2025 16:00:32 +0000 Subject: [PATCH 5/7] Fixed syntax issue with attribute collection --- cfapyx/creator.py | 1 + 1 file changed, 1 insertion(+) diff --git a/cfapyx/creator.py b/cfapyx/creator.py index ad10e3a..8f5a1b6 100644 --- a/cfapyx/creator.py +++ b/cfapyx/creator.py @@ -512,6 +512,7 @@ def _accumulate_attrs(self, attrs: dict, ncattrs: dict) -> dict: for correct values. """ + first_time = False if attrs is None: first_time = True attrs = {} From 393538b425e1f346662b023bd5959c6e528ac2e0 Mon Sep 17 00:00:00 2001 From: dwest77 Date: Wed, 3 Dec 2025 16:00:43 +0000 Subject: [PATCH 6/7] Updated documentation on extensions --- docs/source/creator_use.rst | 30 ++++++++++++++++++++++++++++-- 1 file changed, 28 insertions(+), 2 deletions(-) diff --git a/docs/source/creator_use.rst b/docs/source/creator_use.rst index 0d36848..5e85265 100644 --- a/docs/source/creator_use.rst +++ b/docs/source/creator_use.rst @@ -2,7 +2,11 @@ CFA Creator =========== -The cfapyx engine also contains a class for use when creating ``CFA-netCDF`` files. These automatically follow the Aggregation conventions for CF-1.12. +The cfapyx engine also contains a class for use when creating ``CFA-netCDF`` files. These automatically follow the Aggregation conventions for CF-1.12, and the aggregation files produced will have this as the CF convention value in conventions. + +.. note:: + + CF Convention 1.12 is added to the aggregation file, but only applies to the aggregation file and not the fragment files. For fragment files compliant at a higher convention than 1.12, this convention will be carried over to the aggregation file, but as a minimum all files produced by the Creator tool will be 1.12 - this also enables interpreting of these files by the CFDM at that version and also cf-python. Logging ------- @@ -77,4 +81,26 @@ This file may be read into Xarray as a familiar xarray dataset with: xarray_ds = xarray.open_dataset(output_file, engine='CFA') Where the engine is required to decode the aggregation instructions contained in the ``CFA-netCDF`` file. Note that -without this engine the aggregation instructions will be displayed but not decoded. \ No newline at end of file +without this engine the aggregation instructions will be displayed but not decoded. + +Extension/Parallelisation +------------------------- + +The ``CFANetCDF`` object now supports the extension/concatenation of multiple aggregation files, enabling the use of parallelisation. Here the syntax ``extension`` applies to dimensions that are already aggregated and will be ``extended`` by combining multiple aggregations. All aggregation dimensions must be already established in each partial aggregation file (i.e all files must be time-aggregated or otherwise), as the functionality to aggregate per-dimension in parallel is not supported. + +To make use of this functionality, simply apply the create/write process above to batches of fragment files at any size required, then feed the resulting aggregation files back into a new ``CFANetCDF`` creator object to merge the aggregations. + +:: + + from cfapyx import CFANetCDF + + ds = CFANetCDF( + set_of_agg_files, # The set of aggregation files to be combined/extended + concat_msg = 'See individual files for more details', # Replaces attributes that differ + ) + +Given a sufficient setup with use of parallel jobs (and a dependent final job) it is now possible to parallelise the aggregation process to an arbitrary arrangement of jobs. + +.. note:: + + For the most time-efficient parallel arrangement, use file batches of size ``root(n)`` where n is the total number of files. This creates a set of ``root(n)`` partial aggregation files, which then minimises the effective total aggregation time to just ``2*root(n)``. \ No newline at end of file From 01b0c34cc1fed3b430ee42d2abe6cc5d42674137 Mon Sep 17 00:00:00 2001 From: dwest77 Date: Wed, 3 Dec 2025 16:01:01 +0000 Subject: [PATCH 7/7] Updated to full release version --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 67434cc..312f8b3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "cfapyx" -version = "2025.12.3a" +version = "2025.12.3" description = "The pure-Python implementation of the CF Aggregation conventions, including the Xarray engine to enable reading CFA-netCDF files." authors = [ {name = "Daniel Westwood",email = "daniel.westwood@stfc.ac.uk"}