Source code for iris.fileformats.rules

# Copyright Iris contributors
# This file is part of Iris and is released under the BSD license.
# See LICENSE in the root of the repository for full licensing details.
"""Generalised mechanisms for metadata translation and cube construction."""

import collections
import warnings

import cf_units

from iris.analysis import Linear
import iris.cube
import iris.exceptions
import iris.fileformats.um_cf_map
import iris.warnings

Factory = collections.namedtuple("Factory", ["factory_class", "args"])
ReferenceTarget = collections.namedtuple("ReferenceTarget", ("name", "transform"))

[docs] class ConcreteReferenceTarget: """Everything you need to make a real Cube for a named reference.""" def __init__(self, name, transform=None): #: The name used to connect references with references. = name #: An optional transformation to apply to the cubes. self.transform = transform self._src_cubes = iris.cube.CubeList() self._final_cube = None
[docs] def add_cube(self, cube): self._src_cubes.append(cube)
[docs] def as_cube(self): if self._final_cube is None: src_cubes = self._src_cubes if len(src_cubes) > 1: # Merge the reference cubes to allow for # time-varying surface pressure in hybrid-presure. src_cubes = src_cubes.merge(unique=False) if len(src_cubes) > 1: warnings.warn( "Multiple reference cubes for {}".format(, category=iris.warnings.IrisUserWarning, ) src_cube = src_cubes[-1] if self.transform is None: self._final_cube = src_cube else: final_cube = src_cube.copy() attributes = self.transform(final_cube) for name, value in attributes.items(): setattr(final_cube, name, value) self._final_cube = final_cube return self._final_cube
[docs] class Reference(iris.util._OrderedHashable): _names = ("name",) """ A named placeholder for inter-field references. """
[docs] def scalar_coord(cube, coord_name): """Try to find a single-valued coord with the given name.""" found_coord = None for coord in cube.coords(coord_name): if coord.shape == (1,): found_coord = coord break return found_coord
[docs] def vector_coord(cube, coord_name): """Try to find a one-dimensional, multi-valued coord with the given name.""" found_coord = None for coord in cube.coords(coord_name): if len(coord.shape) == 1 and coord.shape[0] > 1: found_coord = coord break return found_coord
[docs] def scalar_cell_method(cube, method, coord_name): """Try to find the given type of cell method over a single coord with the given name.""" found_cell_method = None for cell_method in cube.cell_methods: if cell_method.method == method and len(cell_method.coord_names) == 1: name = cell_method.coord_names[0] if name == coord_name: coords = cube.coords(name) if len(coords) == 1: found_cell_method = cell_method return found_cell_method
[docs] def has_aux_factory(cube, aux_factory_class): """Determine :class:`~iris.aux_factory.AuxCoordFactory` availability within cube. Try to find an :class:`~iris.aux_factory.AuxCoordFactory` instance of the specified type on the cube. """ for factory in cube.aux_factories: if isinstance(factory, aux_factory_class): return True return False
[docs] def aux_factory(cube, aux_factory_class): """Retrieve :class:`~iris.aux_factory.AuxCoordFactory` instance from cube. Return the :class:`~iris.aux_factory.AuxCoordFactory` instance of the specified type from a cube. """ aux_factories = [ aux_factory for aux_factory in cube.aux_factories if isinstance(aux_factory, aux_factory_class) ] if not aux_factories: raise ValueError( "Cube does not have an aux factory of type {!r}.".format(aux_factory_class) ) elif len(aux_factories) > 1: raise ValueError( "Cube has more than one aux factory of type {!r}.".format(aux_factory_class) ) return aux_factories[0]
class _ReferenceError(Exception): """Signals an invalid/missing reference field.""" pass def _dereference_args(factory, reference_targets, regrid_cache, cube): """Convert all the arguments for a factory into concrete coordinates.""" args = [] for arg in factory.args: if isinstance(arg, Reference): if in reference_targets: src = reference_targets[].as_cube() # If necessary, regrid the reference cube to # match the grid of this cube. src = _ensure_aligned(regrid_cache, src, cube) if src is not None: new_coord = iris.coords.AuxCoord(, src.standard_name, src.long_name, src.var_name, src.units, attributes=src.attributes, ) dims = [ cube.coord_dims(src_coord)[0] for src_coord in src.dim_coords ] cube.add_aux_coord(new_coord, dims) args.append(new_coord) else: raise _ReferenceError( "Unable to regrid reference for {!r}".format( ) else: raise _ReferenceError( "The source data contains no field(s) for {!r}.".format( ) else: # If it wasn't a Reference, then arg is a dictionary # of keyword arguments for cube.coord(...). args.append(cube.coord(**arg)) return args def _regrid_to_target(src_cube, target_coords, target_cube): # Interpolate onto the target grid. sample_points = [(coord, coord.points) for coord in target_coords] result_cube = src_cube.interpolate(sample_points, Linear()) # Any scalar coords on the target_cube will have become vector # coords on the resample src_cube (i.e. result_cube). # These unwanted vector coords need to be pushed back to scalars. index = [slice(None, None)] * result_cube.ndim for target_coord in target_coords: if not target_cube.coord_dims(target_coord): result_dim = result_cube.coord_dims(target_coord)[0] index[result_dim] = 0 if not all(key == slice(None, None) for key in index): result_cube = result_cube[tuple(index)] return result_cube def _ensure_aligned(regrid_cache, src_cube, target_cube): """Ensure dimension compatible cubes are spatially aligned. Returns a version of `src_cube` suitable for use as an AuxCoord on `target_cube`, or None if no version can be made. """ result_cube = None # Check that each of src_cube's dim_coords matches up with a single # coord on target_cube. try: target_coords = [] for dim_coord in src_cube.dim_coords: target_coords.append(target_cube.coord(dim_coord)) except iris.exceptions.CoordinateNotFoundError: # One of the src_cube's dim_coords didn't exist on the # target_cube... so we can't regrid (i.e. just return None). pass else: # So we can use `iris.analysis.interpolate.linear()` later, # ensure each target coord is either a scalar or maps to a # single, distinct dimension. target_dims = [target_cube.coord_dims(coord) for coord in target_coords] target_dims = list(filter(None, target_dims)) unique_dims = set() for dims in target_dims: unique_dims.update(dims) compatible = len(target_dims) == len(unique_dims) if compatible: cache_key = id(src_cube) if cache_key not in regrid_cache: regrid_cache[cache_key] = ([src_cube.dim_coords], [src_cube]) grids, cubes = regrid_cache[cache_key] # 'grids' is a list of tuples of coordinates, so convert # the 'target_coords' list into a tuple to be consistent. target_coords = tuple(target_coords) try: # Look for this set of target coordinates in the cache. i = grids.index(target_coords) result_cube = cubes[i] except ValueError: # Not already cached, so do the hard work of interpolating. result_cube = _regrid_to_target(src_cube, target_coords, target_cube) # Add it to the cache. grids.append(target_coords) cubes.append(result_cube) return result_cube _loader_attrs = ("field_generator", "field_generator_kwargs", "converter")
[docs] class Loader(collections.namedtuple("Loader", _loader_attrs)):
[docs] def __new__(cls, field_generator, field_generator_kwargs, converter): """Create a definition of a field-based Cube loader. Parameters ---------- field_generator : A callable that accepts a filename as its first argument and returns an iterable of field objects. field_generator_kwargs : Additional arguments to be passed to the field_generator. converter : A callable that converts a field object into a Cube. """ return tuple.__new__(cls, (field_generator, field_generator_kwargs, converter))
ConversionMetadata = collections.namedtuple( "ConversionMetadata", ( "factories", "references", "standard_name", "long_name", "units", "attributes", "cell_methods", "dim_coords_and_dims", "aux_coords_and_dims", ), ) def _make_cube(field, converter): # Convert the field to a Cube. metadata = converter(field) cube_data = field.core_data() cube = iris.cube.Cube( cube_data, attributes=metadata.attributes, cell_methods=metadata.cell_methods, dim_coords_and_dims=metadata.dim_coords_and_dims, aux_coords_and_dims=metadata.aux_coords_and_dims, ) # Temporary code to deal with invalid standard names in the # translation table. if metadata.standard_name is not None: cube.rename(metadata.standard_name) if metadata.long_name is not None: cube.long_name = metadata.long_name if metadata.units is not None: # Temporary code to deal with invalid units in the translation # table. try: cube.units = metadata.units except ValueError: msg = "Ignoring PP invalid units {!r}".format(metadata.units) warnings.warn(msg, category=iris.warnings.IrisIgnoringWarning) cube.attributes["invalid_units"] = metadata.units cube.units = cf_units._UNKNOWN_UNIT_STRING return cube, metadata.factories, metadata.references def _resolve_factory_references( cube, factories, concrete_reference_targets, regrid_cache={} ): # Attach the factories for a cube, building them from references. # Note: the regrid_cache argument lets us share and reuse regridded data # across multiple result cubes. for factory in factories: try: args = _dereference_args( factory, concrete_reference_targets, regrid_cache, cube ) except _ReferenceError as e: msg = "Unable to create instance of {factory}. " + str(e) factory_name = factory.factory_class.__name__ warnings.warn( msg.format(factory=factory_name), category=iris.warnings.IrisUserWarning, ) else: aux_factory = factory.factory_class(*args) cube.add_aux_factory(aux_factory) def _load_pairs_from_fields_and_filenames( fields_and_filenames, converter, user_callback_wrapper=None ): # The underlying mechanism for the public 'load_pairs_from_fields' and # 'load_cubes'. # Slightly more complicated than 'load_pairs_from_fields', only because it # needs a filename associated with each field to support the load callback. concrete_reference_targets = {} results_needing_reference = [] for field, filename in fields_and_filenames: # Convert the field to a Cube, passing down the 'converter' function. cube, factories, references = _make_cube(field, converter) # Post modify the new cube with a user-callback. # This is an ordinary Iris load callback, so it takes the filename. cube =, cube, field, filename) # Callback mechanism may return None, which must not be yielded. if cube is None: continue # Cross referencing. for reference in references: name = # Register this cube as a source cube for the named reference. target = concrete_reference_targets.get(name) if target is None: target = ConcreteReferenceTarget(name, reference.transform) concrete_reference_targets[name] = target target.add_cube(cube) if factories: results_needing_reference.append((cube, factories, field)) else: yield (cube, field) regrid_cache = {} for cube, factories, field in results_needing_reference: _resolve_factory_references( cube, factories, concrete_reference_targets, regrid_cache ) yield (cube, field)
[docs] def load_pairs_from_fields(fields, converter): """Convert iterable of fields into iterable of Cubes using the provided converter. Parameters ---------- fields : An iterable of fields. converter : An Iris converter function, suitable for use with the supplied fields. See the description in :class:`iris.fileformats.rules.Loader`. Returns ------- An iterable of (:class:`iris.cube.Cube`, field) pairs. """ return _load_pairs_from_fields_and_filenames( ((field, None) for field in fields), converter )
[docs] def load_cubes(filenames, user_callback, loader, filter_function=None): if isinstance(filenames, str): filenames = [filenames] def _generate_all_fields_and_filenames(): for filename in filenames: for field in loader.field_generator( filename, **loader.field_generator_kwargs ): # evaluate field against format specific desired attributes # load if no format specific desired attributes are violated if filter_function is None or filter_function(field): yield (field, filename) def loadcubes_user_callback_wrapper(cube, field, filename): # Run user-provided original callback function. result = cube if user_callback is not None: result = user_callback(cube, field, filename) return result all_fields_and_filenames = _generate_all_fields_and_filenames() for cube, field in _load_pairs_from_fields_and_filenames( all_fields_and_filenames, converter=loader.converter, user_callback_wrapper=loadcubes_user_callback_wrapper, ): yield cube