Source code for podpac.core.data.interpolation

from __future__ import division, unicode_literals, print_function, absolute_import
from copy import deepcopy
from collections import OrderedDict
from six import string_types

import traitlets as tl
import numpy as np

# podpac imports
from podpac.core.data.interpolator import Interpolator
from podpac.core.data.interpolators import NearestNeighbor, NearestPreview, Rasterio, ScipyPoint, ScipyGrid

INTERPOLATION_DEFAULT = "nearest"
"""str : Default interpolation method used when creating a new :class:`Interpolation` class """

INTERPOLATORS = [NearestNeighbor, NearestPreview, Rasterio, ScipyPoint, ScipyGrid]
"""list : list of available interpolator classes"""

INTERPOLATORS_DICT = {}
"""dict : Dictionary of a string interpolator name and associated interpolator class"""

INTERPOLATION_METHODS = [
    "nearest_preview",
    "nearest",
    "bilinear",
    "cubic",
    "cubic_spline",
    "lanczos",
    "average",
    "mode",
    "gauss",
    "max",
    "min",
    "med",
    "q1",
    "q3",
    "spline_2",
    "spline_3",
    "spline_4",
]

INTERPOLATION_METHODS_DICT = {}
"""dict: Dictionary of string interpolation methods and associated interpolator classes
   (i.e. ``'nearest': [NearestNeighbor, Rasterio, Scipy]``) """


def load_interpolators():
    """Load interpolators from :list:`INTERPOLATORS`

    Defines :dict:`INTERPOLATORS_DICT`, and :dict:`INTERPOLATION_METHODS_DICT`
    """

    # create empty arrays in INTEPROLATOR_METHODS
    for method in INTERPOLATION_METHODS:
        INTERPOLATION_METHODS_DICT[method] = []

    # fill dictionaries with interpolator properties
    for interpolator_class in INTERPOLATORS:
        interpolator = interpolator_class()
        INTERPOLATORS_DICT[interpolator.name] = interpolator_class

        for method in INTERPOLATION_METHODS:
            if method in interpolator.methods_supported:
                INTERPOLATION_METHODS_DICT[method].append(interpolator_class)


# load interpolators when module is first loaded
load_interpolators()


[docs]class InterpolationException(Exception): """ Custom label for interpolation exceptions """ pass
def interpolation_trait(default_value=INTERPOLATION_DEFAULT, allow_none=True, **kwargs): """Create a new interpolation trait Returns ------- tl.Union Union trait for an interpolation definition """ return tl.Union( [tl.Dict(), tl.Enum(INTERPOLATION_METHODS), tl.Instance(Interpolation)], allow_none=allow_none, default_value=default_value, **kwargs )
[docs]class Interpolation(object): """Create an interpolation class to handle one interpolation method per unstacked dimension. Used to interpolate data within a datasource. Parameters ---------- definition : str, tuple (str, list of podpac.core.data.interpolator.Interpolator), dict Interpolation definition used to define interpolation methods for each definiton. See :attr:`podpac.data.DataSource.interpolation` for more details. Raises ------ InterpolationException Raised when definition parameter is improperly formatted """ definition = None config = OrderedDict() # container for interpolation methods for each dimension _last_interpolator_queue = None # container for the last run interpolator queue - useful for debugging _last_select_queue = None # container for the last run select queue - useful for debugging
[docs] def __init__(self, definition=INTERPOLATION_DEFAULT): self.definition = definition self.config = OrderedDict() # if definition is None, set to default # TODO: do we want to always have a default for interpolation? # Or should there be an option to turn off interpolation? if self.definition is None: self.definition = INTERPOLATION_DEFAULT # set each dim to interpolator definition if isinstance(definition, dict): # covert input to an ordered dict to preserve order of dimensions definition = OrderedDict(definition) for key in iter(definition): # if dict is a default definition, skip the rest of the handling if not isinstance(key, tuple): if key in ["method", "params", "interpolators"]: method = self._parse_interpolation_method(definition) self._set_interpolation_method(("default",), method) break # if key is not a tuple, convert it to one and set it to the udims key if not isinstance(key, tuple): udims = (key,) else: udims = key # make sure udims are not already specified in config for config_dims in iter(self.config): if set(config_dims) & set(udims): raise InterpolationException( 'Dimensions "{}" cannot be defined '.format(udims) + "multiple times in interpolation definition {}".format(definition) ) # get interpolation method method = self._parse_interpolation_method(definition[key]) # add all udims to definition self._set_interpolation_method(udims, method) # set default if its not been specified in the dict if ("default",) not in self.config: default_method = self._parse_interpolation_method(INTERPOLATION_DEFAULT) self._set_interpolation_method(("default",), default_method) elif isinstance(definition, string_types): method = self._parse_interpolation_method(definition) self._set_interpolation_method(("default",), method) else: raise TypeError( '"{}" is not a valid interpolation definition type. '.format(definition) + "Interpolation definiton must be a string or dict" ) # make sure ('default',) is always the last entry in config dictionary default = self.config.pop(("default",)) self.config[("default",)] = default
def __repr__(self): rep = str(self.__class__.__name__) for udims in iter(self.config): # rep += '\n\t%s:\n\t\tmethod: %s\n\t\tinterpolators: %s\n\t\tparams: %s' % \ rep += "\n\t%s: %s, %s, %s" % ( udims, self.config[udims]["method"], [i.__class__.__name__ for i in self.config[udims]["interpolators"]], self.config[udims]["params"], ) return rep def _parse_interpolation_method(self, definition): """parse interpolation definitions into a tuple of (method, Interpolator) Parameters ---------- definition : str, dict interpolation definition See :attr:`podpac.data.DataSource.interpolation` for more details. Returns ------- dict dict with keys 'method', 'interpolators', and 'params' Raises ------ InterpolationException TypeError """ if isinstance(definition, string_types): if definition not in INTERPOLATION_METHODS: raise InterpolationException( '"{}" is not a valid interpolation shortcut. '.format(definition) + "Valid interpolation shortcuts: {}".format(INTERPOLATION_METHODS) ) return {"method": definition, "interpolators": INTERPOLATION_METHODS_DICT[definition], "params": {}} elif isinstance(definition, dict): # confirm method in dict if "method" not in definition: raise InterpolationException( "{} is not a valid interpolation definition. ".format(definition) + 'Interpolation definition dict must contain key "method" string value' ) else: method_string = definition["method"] # if specifying custom method, user must include interpolators if "interpolators" not in definition and method_string not in INTERPOLATION_METHODS: raise InterpolationException( '"{}" is not a valid interpolation shortcut. '.format(method_string) + 'Specify list "interpolators" or change "method" ' + "to a valid interpolation shortcut: {}".format(INTERPOLATION_METHODS) ) elif "interpolators" not in definition: interpolators = INTERPOLATION_METHODS_DICT[method_string] else: interpolators = definition["interpolators"] # default for params if "params" in definition: params = definition["params"] else: params = {} # confirm types if not isinstance(method_string, string_types): raise TypeError( "{} is not a valid interpolation method. ".format(method_string) + "Interpolation method must be a string" ) if not isinstance(interpolators, list): raise TypeError( "{} is not a valid interpolator definition. ".format(interpolators) + "Interpolator definition must be of type list containing Interpolator" ) if not isinstance(params, dict): raise TypeError( "{} is not a valid interpolation params definition. ".format(params) + "Interpolation params must be a dict" ) # handle when interpolator is a string (most commonly from a pipeline definition) for idx, interpolator_class in enumerate(interpolators): if isinstance(interpolator_class, string_types): if interpolator_class in INTERPOLATORS_DICT.keys(): interpolators[idx] = INTERPOLATORS_DICT[interpolator_class] else: raise TypeError( 'Interpolator "{}" is not in the dictionary of valid '.format(interpolator_class) + "interpolators: {}".format(INTERPOLATORS_DICT) ) # validate interpolator class for interpolator in interpolators: self._validate_interpolator(interpolator) # if all checks pass, return the definition return {"method": method_string, "interpolators": interpolators, "params": params} else: raise TypeError( '"{}" is not a valid Interpolator definition. '.format(definition) + "Interpolation definiton must be a string or dict." ) def _validate_interpolator(self, interpolator): """Make sure interpolator is a subclass of Interpolator Parameters ---------- interpolator : any input definition to validate Raises ------ TypeError Raises a type error if interpolator is not a subclass of Interpolator """ try: valid = issubclass(interpolator, Interpolator) if not valid: raise TypeError() except TypeError: raise TypeError( "{} is not a valid interpolator type. ".format(interpolator) + "Interpolator must be of type {}".format(Interpolator) ) def _set_interpolation_method(self, udims, definition): """Set the list of interpolation definitions to the input dimension Parameters ---------- udims : tuple tuple of dimensiosn to assign definition to definition : dict dict definition returned from _parse_interpolation_method """ method = deepcopy(definition["method"]) interpolators = deepcopy(definition["interpolators"]) params = deepcopy(definition["params"]) # instantiate interpolators for (idx, interpolator) in enumerate(interpolators): interpolators[idx] = interpolator(method=method, **params) definition["interpolators"] = interpolators # set to interpolation configuration for dims self.config[udims] = definition def _select_interpolator_queue(self, source_coordinates, eval_coordinates, select_method, strict=False): """Create interpolator queue based on interpolation configuration and requested/native source_coordinates Parameters ---------- source_coordinates : :class:`podpac.Coordinates` Description eval_coordinates : :class:`podpac.Coordinates` Description select_method : function method used to determine if interpolator can handle dimensions strict : bool, optional Raise an error if all dimensions can't be handled Returns ------- OrderedDict Dict of (udims: Interpolator) to run in order Raises ------ InterpolationException If `strict` is True, InterpolationException is raised when all dimensions cannot be handled """ source_dims = set(source_coordinates.udims) handled_dims = set() interpolator_queue = OrderedDict() # go through all dims in config for key in iter(self.config): # if the key is set to (default,), it represents all the remaining dimensions that have not been handled # __init__ makes sure that (default,) will always be the last key in on if key == ("default",): udims = tuple(source_dims - handled_dims) else: udims = key # get configured list of interpolators for dim definition interpolators = self.config[key]["interpolators"] # iterate through interpolators recording which dims they support for interpolator in interpolators: # if all dims have been handled already, skip the rest if not udims: break # see which dims the interpolator can handle can_handle = getattr(interpolator, select_method)(udims, source_coordinates, eval_coordinates) # if interpolator can handle all udims if not set(udims) - set(can_handle): # union of dims that can be handled by this interpolator and already supported dims handled_dims = handled_dims | set(can_handle) # set interpolator to work on that dimension in the interpolator_queue if dim has no interpolator if udims not in interpolator_queue: interpolator_queue[udims] = interpolator # throw error if the source_dims don't encompass all the supported dims # this should happen rarely because of default if len(source_dims) > len(handled_dims) and strict: missing_dims = list(source_dims - handled_dims) raise InterpolationException( "Dimensions {} ".format(missing_dims) + "can't be handled by interpolation definition:\n {}".format(self) ) # TODO: adjust by interpolation cost return interpolator_queue
[docs] def select_coordinates(self, source_coordinates, source_coordinates_index, eval_coordinates): """ Select a subset or coordinates if interpolator can downselect. At this point in the execution process, podpac has selected a subset of source_coordinates that intersects with the requested coordinates, dropped extra dimensions from requested coordinates, and confirmed source coordinates are not missing any dimensions. Parameters ---------- source_coordinates : :class:`podpac.Coordinates` Intersected source coordinates source_coordinates_index : list Index of intersected source coordinates. See :class:`podpac.data.DataSource` for more information about valid values for the source_coordinates_index eval_coordinates : :class:`podpac.Coordinates` Requested coordinates to evaluate Returns ------- (:class:`podpac.Coordinates`, list) Returns tuple with the first element subset of selected coordinates and the second element the indicies of the selected coordinates """ # TODO: short circuit if source_coordinates contains eval_coordinates # short circuit if source and eval coordinates are the same if source_coordinates == eval_coordinates: return source_coordinates, tuple(source_coordinates_index) interpolator_queue = self._select_interpolator_queue(source_coordinates, eval_coordinates, "can_select") self._last_select_queue = interpolator_queue selected_coords = deepcopy(source_coordinates) selected_coords_idx = deepcopy(source_coordinates_index) for udims in interpolator_queue: interpolator = interpolator_queue[udims] # run interpolation. mutates selected coordinates and selected coordinates index selected_coords, selected_coords_idx = interpolator.select_coordinates( udims, selected_coords, selected_coords_idx, eval_coordinates ) return selected_coords, tuple(selected_coords_idx)
[docs] def interpolate(self, source_coordinates, source_data, eval_coordinates, output_data): """Interpolate data from requested coordinates to source coordinates Parameters ---------- source_coordinates : :class:`podpac.Coordinates` Description source_data : podpac.core.units.UnitsDataArray Description eval_coordinates : :class:`podpac.Coordinates` Description output_data : podpac.core.units.UnitsDataArray Description Returns ------- podpac.core.units.UnitDataArray returns the new output UnitDataArray of interpolated data Raises ------ InterpolationException Raises InterpolationException when interpolator definition can't support all the dimensions of the requested coordinates """ # short circuit if the source data and requested coordinates are of shape == 1 if source_data.size == 1 and np.prod(eval_coordinates.shape) == 1: output_data[:] = source_data return output_data # TODO: short circuit if source_coordinates contains eval_coordinates # this has to be done better... # short circuit if source and eval coordinates are the same if not (set(source_coordinates.udims) - set(eval_coordinates.udims)): eq = True for udim in source_coordinates.udims: if not np.all(source_coordinates[udim].coordinates == eval_coordinates[udim].coordinates): eq = False if eq: output_data.data = source_data.transpose(*output_data.dims).data # transpose and insert return output_data interpolator_queue = self._select_interpolator_queue( source_coordinates, eval_coordinates, "can_interpolate", strict=True ) # for debugging purposes, save the last defined interpolator queue self._last_interpolator_queue = interpolator_queue # iterate through each dim tuple in the queue for udims in interpolator_queue: interpolator = interpolator_queue[udims] # run interpolation output_data = interpolator.interpolate( udims, source_coordinates, source_data, eval_coordinates, output_data ) return output_data