import numpy as np from .. import util from ..boundingregion import BoundingBox from ..dimension import dimension_name from ..element import Element from ..ndmapping import NdMapping, item_check from ..sheetcoords import SheetCoordinateSystem, Slice from .grid import GridInterface from .interface import DataError, Interface from .util import finite_range class ImageInterface(GridInterface): """ Interface for 2 or 3D arrays representing images of raw luminance values, RGB values or HSV values. """ types = (np.ndarray,) datatype = 'image' named = False @classmethod def init(cls, eltype, data, kdims, vdims): if kdims is None: kdims = eltype.kdims if vdims is None: vdims = eltype.vdims kwargs = {} dimensions = [dimension_name(d) for d in kdims + vdims] if isinstance(data, tuple): data = dict(zip(dimensions, data)) if isinstance(data, dict): xs, ys = np.asarray(data[kdims[0].name]), np.asarray(data[kdims[1].name]) xvalid = util.validate_regular_sampling(xs, eltype.rtol or util.config.image_rtol) yvalid = util.validate_regular_sampling(ys, eltype.rtol or util.config.image_rtol) if not xvalid or not yvalid: raise ValueError('ImageInterface only supports regularly sampled coordinates') l, r, xdensity, invertx = util.bound_range(xs, None, eltype._time_unit) b, t, ydensity, inverty = util.bound_range(ys, None, eltype._time_unit) kwargs['bounds'] = BoundingBox(points=((l, b), (r, t))) if len(vdims) == 1: data = np.flipud(np.asarray(data[vdims[0].name])) else: data = np.dstack([np.flipud(data[vd.name]) for vd in vdims]) if invertx: data = data[:, ::-1] if inverty: data = data[::-1, :] expected = (len(ys), len(xs)) shape = data.shape[:2] error = DataError if len(shape) > 1 and not eltype._binned else ValueError if shape != expected and not (not expected and shape == (1,)): raise error(f'Key dimension values and value array {vdims[0]} ' f'shapes do not match. Expected shape {expected}, ' f'actual shape: {shape}', cls) if not isinstance(data, np.ndarray) or data.ndim not in [2, 3]: raise ValueError('ImageInterface expects a 2D array.') elif not issubclass(eltype, SheetCoordinateSystem): raise ValueError('ImageInterface may only be used on elements ' 'that subclass SheetCoordinateSystem.') return data, {'kdims':kdims, 'vdims':vdims}, kwargs @classmethod def irregular(cls, dataset, dim): "ImageInterface does not support irregular data" return False @classmethod def shape(cls, dataset, gridded=False): if gridded: return dataset.data.shape[:2] else: return cls.length(dataset), len(dataset.dimensions()) @classmethod def dtype(cls, dataset, dimension): idx = dataset.get_dimension_index(dimension) if idx in [0, 1]: return np.dtype('float') else: return dataset.data.dtype @classmethod def length(cls, dataset): return np.prod(dataset.data.shape[:2], dtype=np.intp) @classmethod def validate(cls, dataset, vdims=True): pass @classmethod def redim(cls, dataset, dimensions): return dataset.data @classmethod def reindex(cls, dataset, kdims=None, vdims=None): data = dataset.data dropped_kdims = [kd for kd in dataset.kdims if kd not in kdims] constant = {} for kd in dropped_kdims: vals = cls.values(dataset, kd.name, expanded=False) if len(vals) == 1: constant[kd.name] = vals[0] if dropped_kdims or constant: return tuple(dataset.columns(kdims+vdims).values()) if vdims is not None and vdims != dataset.vdims and len(dataset.vdims) > 1: inds = [dataset.get_dimension_index(vd)-dataset.ndims for vd in vdims] return data[..., inds] if len(inds) > 1 else data[..., inds[0]] return data @classmethod def coords(cls, dataset, dim, ordered=False, expanded=False, edges=False): dim = dataset.get_dimension(dim, strict=True) if expanded: data = util.expand_grid_coords(dataset, dim) if edges and data.shape == dataset.data.shape: data = cls._infer_interval_breaks(data, axis=1) data = cls._infer_interval_breaks(data, axis=0) return data values = cls.values(dataset, dim, expanded=False) if edges: return cls._infer_interval_breaks(values) else: return values @classmethod def range(cls, obj, dim): dim = obj.get_dimension(dim, strict=True) dim_idx = obj.get_dimension_index(dim) if dim_idx in [0, 1] and obj.bounds: l, b, r, t = obj.bounds.lbrt() if dim_idx: (low, high) = (b, t) density = obj.ydensity else: low, high = (l, r) density = obj.xdensity halfd = (1./density)/2. if isinstance(low, util.datetime_types): halfd = np.timedelta64(int(round(halfd)), obj._time_unit) drange = (low+halfd, high-halfd) elif 1 < dim_idx < len(obj.vdims) + 2: dim_idx -= 2 data = np.atleast_3d(obj.data)[:, :, dim_idx] if dim.nodata is not None: data = cls.replace_value(data, dim.nodata) drange = finite_range(data, np.nanmin(data), np.nanmax(data)) else: drange = (None, None) return drange @classmethod def values( cls, dataset, dim, expanded=True, flat=True, compute=True, keep_index=False ): """ The set of samples available along a particular dimension. """ dim_idx = dataset.get_dimension_index(dim) if dim_idx in [0, 1]: l, b, r, t = dataset.bounds.lbrt() dim2, dim1 = dataset.data.shape[:2] xdate, ydate = isinstance(l, util.datetime_types), isinstance(b, util.datetime_types) if l == r or dim1 == 0: xlin = np.full((dim1,), l, dtype=('datetime64[us]' if xdate else 'float')) elif xdate: xlin = util.date_range(l, r, dim1, dataset._time_unit) else: xstep = float(r - l)/dim1 xlin = np.linspace(l+(xstep/2.), r-(xstep/2.), dim1) if b == t or dim2 == 0: ylin = np.full((dim2,), b, dtype=('datetime64[us]' if ydate else 'float')) elif ydate: ylin = util.date_range(b, t, dim2, dataset._time_unit) else: ystep = float(t - b)/dim2 ylin = np.linspace(b+(ystep/2.), t-(ystep/2.), dim2) if expanded: values = np.meshgrid(ylin, xlin)[abs(dim_idx-1)] return values.flatten() if flat else values.T else: return ylin if dim_idx else xlin elif dataset.ndims <= dim_idx < len(dataset.dimensions()): # Raster arrays are stored with different orientation # than expanded column format, reorient before expanding if dataset.data.ndim > 2: data = dataset.data[:, :, dim_idx-dataset.ndims] else: data = dataset.data data = np.flipud(data) return data.T.flatten() if flat else data else: return None @classmethod def mask(cls, dataset, mask, mask_val=np.nan): masked = dataset.data.copy().astype('float') masked[np.flipud(mask)] = mask_val return masked @classmethod def select(cls, dataset, selection_mask=None, **selection): """ Slice the underlying numpy array in sheet coordinates. """ selection = {k: slice(*sel) if isinstance(sel, tuple) else sel for k, sel in selection.items()} coords = tuple(selection[kd.name] if kd.name in selection else slice(None) for kd in dataset.kdims) if not any([isinstance(el, slice) for el in coords]): return dataset.data[dataset.sheet2matrixidx(*coords)] # Apply slices xidx, yidx = coords l, b, r, t = dataset.bounds.lbrt() if isinstance(xidx, slice): l = l if xidx.start is None else max(l, xidx.start) r = r if xidx.stop is None else min(r, xidx.stop) if isinstance(yidx, slice): b = b if yidx.start is None else max(b, yidx.start) t = t if yidx.stop is None else min(t, yidx.stop) bounds = BoundingBox(points=((l, b), (r, t))) slc = Slice(bounds, dataset) return slc.submatrix(dataset.data) @classmethod def sample(cls, dataset, samples=None): """ Sample the Raster along one or both of its dimensions, returning a reduced dimensionality type, which is either a ItemTable, Curve or Scatter. If two dimension samples and a new_xaxis is provided the sample will be the value of the sampled unit indexed by the value in the new_xaxis tuple. """ if samples is None: samples = [] if len(samples[0]) == 1: select = {dataset.kdims[0].name: [s[0] for s in samples]} return tuple(dataset.select(**select).columns().values()) return [c+(dataset.data[dataset._coord2matrix(c)],) for c in samples] @classmethod def groupby(cls, dataset, dim_names, container_type, group_type, **kwargs): # Get dimensions information dimensions = [dataset.get_dimension(d) for d in dim_names] kdims = [kdim for kdim in dataset.kdims if kdim not in dimensions] # Update the kwargs appropriately for Element group types group_kwargs = {} group_type = dict if group_type == 'raw' else group_type if issubclass(group_type, Element): group_kwargs.update(util.get_param_values(dataset)) group_kwargs['kdims'] = kdims group_kwargs.update(kwargs) if len(dimensions) == 1: didx = dataset.get_dimension_index(dimensions[0]) coords = dataset.dimension_values(dimensions[0], expanded=False) xvals = dataset.dimension_values(abs(didx-1), expanded=False) samples = [(i, slice(None)) if didx else (slice(None), i) for i in range(dataset.data.shape[abs(didx-1)])] data = np.flipud(dataset.data) groups = [(c, group_type((xvals, data[s]), **group_kwargs)) for s, c in zip(samples, coords)] else: data = zip(*[dataset.dimension_values(i) for i in range(len(dataset.dimensions()))]) groups = [(g[:dataset.ndims], group_type([g[dataset.ndims:]], **group_kwargs)) for g in data] if issubclass(container_type, NdMapping): with item_check(False): return container_type(groups, kdims=dimensions) else: return container_type(groups) @classmethod def unpack_scalar(cls, dataset, data): """ Given a dataset object and data in the appropriate format for the interface, return a simple scalar. """ if np.isscalar(data) or len(data) != 1: return data key = next(iter(data.keys())) if len(data[key]) == 1 and key in dataset.vdims: return data[key][0] @classmethod def aggregate(cls, dataset, kdims, function, **kwargs): kdims = [dimension_name(kd) for kd in kdims] axes = tuple(dataset.ndims-dataset.get_dimension_index(kdim)-1 for kdim in dataset.kdims if kdim not in kdims) data = np.atleast_1d(function(dataset.data, axis=axes, **kwargs)) if not kdims: if len(dataset.vdims) == 1: return (data if np.isscalar(data) else data[0], []) else: return ({vd.name: np.array([v]) for vd, v in zip(dataset.vdims, data)}, []) elif len(axes) == 1: return ({kdims[0]: cls.values(dataset, axes[0], expanded=False), dataset.vdims[0].name: data[::-1] if axes[0] else data}, []) Interface.register(ImageInterface)