class Timer:
    """
    Context manager to facilitate timing code.

    Examples
    --------
    >>> with Timer() as t:
            ...  # some longer running code
    >>> print(t)  # prints elapsed time

    based on http://preshing.com/20110924/timing-your-code-using-pythons-with-statement/
    """

    def __init__(self, elapsed=0.0, str_round=3):
        self._elapsed = elapsed
        self._str_round = str_round
        self.start = None
        self.end = None

    def __enter__(self):
        self.start = time.time()
        return self

    def __exit__(self, *args):
        self.end = time.time()
        self._elapsed = self.end - self.start

    def __lt__(self, other):
        return self._elapsed < other._elapsed

    def __le__(self, other):
        return self._elapsed <= other._elapsed

    def __eq__(self, other):
        return self._elapsed == other._elapsed

    def __ne__(self, other):
        return self._elapsed != other._elapsed

    def __ge__(self, other):
        return self._elapsed >= other._elapsed

    def __gt__(self, other):
        return self._elapsed > other._elapsed

    def __add__(self, other):
        return Timer(elapsed=self._elapsed + other._elapsed)

    def __sub__(self, other):
        return Timer(elapsed=self._elapsed - other._elapsed)

    def __repr__(self):
        return "Timer(start=%s, end=%s, elapsed=%s)" % (
            self.start,
            self.end,
            self.__str__(),
        )

    def __str__(self):
        minutes, seconds = divmod(self.elapsed, 60)
        hours, minutes = divmod(minutes, 60)
        if hours:
            return "%sh %sm %ss" % (int(hours), int(minutes), int(seconds))
        elif minutes:
            return "%sm %ss" % (int(minutes), int(seconds))
        else:
            return "%ss" % round(seconds, self._str_round)

    @property
    def elapsed(self):
        return (
            time.time() - self.start if self.start and not self.end else self._elapsed
        ) def get_process_func(process=None, config_dir=None, run_compile=False):
    """Import and return process function."""
    logger.debug(f"get process function from {process}")
    process_module = _load_process_module(
        process=process, config_dir=config_dir, run_compile=run_compile
    )
    try:
        if hasattr(process_module, "Process"):
            logger.error(
                """instanciating MapcheteProcess is deprecated, """
                """provide execute() function instead"""
            )
        if hasattr(process_module, "execute"):
            return process_module.execute
        else:
            raise ImportError("No execute() function found in %s" % process)
    except ImportError as e:
        raise MapcheteProcessImportError(e) class MapcheteNodataTile(Exception):
    """Indicates an empty tile.""" class NoTaskGeometry(TypeError):
    """Raised when Task has no assigned geo information.""" class MapcheteProcessException(Exception):
    """Raised when a mapchete process execution fails.""" class MapcheteProcessOutputError(ValueError):
    """Raised when a mapchete process output is invalid.""" def read_raster_window(
    input_files,
    tile,
    indexes=None,
    resampling="nearest",
    src_nodata=None,
    dst_nodata=None,
    gdal_opts=None,
):
def _read_raster_window(
    input_files,
    tile,
    indexes=None,
    resampling="nearest",
    src_nodata=None,
    dst_nodata=None,
):
def _get_warped_edge_array(
    tile=None,
    input_file=None,
    indexes=None,
    dst_shape=None,
    resampling=None,
    src_nodata=None,
    dst_nodata=None,
):
def _get_warped_array(
    input_file=None,
    indexes=None,
    dst_bounds=None,
    dst_shape=None,
    dst_crs=None,
    resampling=None,
    src_nodata=None,
    dst_nodata=None,
):
def _rasterio_read(
    input_file=None,
    indexes=None,
    dst_bounds=None,
    dst_shape=None,
    dst_crs=None,
    resampling=None,
    src_nodata=None,
    dst_nodata=None,
):
    def _read(
        src, indexes, dst_bounds, dst_shape, dst_crs, resampling, src_nodata, dst_nodata
    ):
def read_raster_no_crs(input_file, indexes=None, gdal_opts=None):
    def __init__(
        self, in_tile=None, in_data=None, out_profile=None, out_tile=None, tags=None
    ):
    def __enter__(self):
    def __exit__(self, *args):
def write_raster_window(
    in_tile=None,
    in_data=None,
    out_profile=None,
    out_tile=None,
    out_path=None,
    tags=None,
    fs=None,
    **kwargs,
):
def _write_tags(dst, tags):
def rasterio_write(path, mode=None, fs=None, in_memory=True, *args, **kwargs):
    def __init__(self, path, *args, fs=None, in_memory=True, **kwargs):
    def __enter__(self):
    def __exit__(self, *args):
def extract_from_array(in_raster=None, in_affine=None, out_tile=None):
def resample_from_array(
    in_raster=None,
    in_affine=None,
    out_tile=None,
    in_crs=None,
    resampling="nearest",
    nodataval=None,
    nodata=0,
):
def create_mosaic(tiles, nodata=0):
def bounds_to_ranges(out_bounds=None, in_affine=None, in_shape=None):
def tiles_to_affine_shape(tiles):
def _get_tiles_properties(tiles):
def _shift_required(tiles):
            def gen_groups(items):
def memory_file(data=None, profile=None):
def prepare_array(data, masked=True, nodata=0, dtype="int16"):
def _prepare_iterable(data, masked, nodata, dtype):
def _prepare_masked(data, masked=True, nodata=0, dtype=None):
def convert_raster(inp, out, overwrite=False, exists_ok=True, **kwargs):
def read_raster(inp, **kwargs):
class RasterWindowMemoryFile:
class RasterioRemoteWriter: def to_shape(geom):
    """
    Convert geometry to shapely geometry if necessary.

    Parameters
    ----------
    geom : shapely geometry or GeoJSON mapping

    Returns
    -------
    shapely geometry
    """
    return shape(geom) if isinstance(geom, dict) else geom class IndexedFeatures:
    """
    Behaves like a mapping of GeoJSON-like objects but has a filter() method.

    Parameters
    ----------
    features : iterable
        Features to be indexed
    index : string
        Spatial index to use. Can either be "rtree" (if installed) or None.
    """

    def __init__(self, features, index="rtree", allow_non_geo_objects=False, crs=None):
        if index == "rtree":
            try:
                from rtree import index

                self._index = index.Index()
            except ImportError:  # pragma: no cover
                warnings.warn(
                    "It is recommended to install rtree in order to significantly speed up spatial indexes."
                )
                self._index = FakeIndex()
        else:
            self._index = FakeIndex()

        self.crs = features.crs if hasattr(features, "crs") else crs
        self._items = {}
        self._non_geo_items = set()
        self.bounds = (None, None, None, None)
        for feature in features:
            if isinstance(feature, tuple):
                id_, feature = feature
            else:
                id_ = self._get_feature_id(feature)
            self._items[id_] = feature
            bounds = self._get_feature_bounds(
                feature, allow_non_geo_objects=allow_non_geo_objects
            )
            if bounds is None:
                self._non_geo_items.add(id_)
            else:
                self._update_bounds(bounds)
                self._index.insert(id_, bounds)

    def __repr__(self):  # pragma: no cover
        return f"IndexedFeatures(features={len(self)}, index={self._index.__repr__()})"

    def __len__(self):
        return len(self._items)

    def __str__(self):  # pragma: no cover
        return "IndexedFeatures([%s])" % (", ".join([str(f) for f in self]))

    def __getitem__(self, key):
        try:
            return self._items[hash(key)]
        except KeyError:
            raise KeyError(f"no feature with id {key} exists")

    def __iter__(self):
        return iter(self._items.values())

    def items(self):
        return self._items.items()

    def keys(self):
        return self._items.keys()

    def values(self):
        return self._items.values()

    def filter(self, bounds=None, bbox=None):
        """
        Return features intersecting with bounds.

        Parameters
        ----------
        bounds : list or tuple
            Bounding coordinates (left, bottom, right, top).

        Returns
        -------
        features : list
            List of features.
        """
        bounds = bounds or bbox
        return [
            self._items[id_]
            for id_ in chain(self._index.intersection(bounds), self._non_geo_items)
        ]

    def _update_bounds(self, bounds):
        left, bottom, right, top = self.bounds
        self.bounds = (
            bounds.left if left is None else min(left, bounds.left),
            bounds.bottom if bottom is None else min(bottom, bounds.bottom),
            bounds.right if right is None else max(right, bounds.right),
            bounds.top if top is None else max(top, bounds.top),
        )

    def _get_feature_id(self, feature):
        if hasattr(feature, "id"):
            return hash(feature.id)
        elif isinstance(feature, dict) and "id" in feature:
            return hash(feature["id"])
        else:
            try:
                return hash(feature)
            except TypeError:
                raise TypeError("features need to have an id or have to be hashable")

    def _get_feature_bounds(self, feature, allow_non_geo_objects=False):
        try:
            if hasattr(feature, "bounds"):
                return validate_bounds(feature.bounds)
            elif hasattr(feature, "__geo_interface__"):
                return validate_bounds(shape(feature).bounds)
            elif feature.get("bounds"):
                return validate_bounds(feature["bounds"])
            elif feature.get("geometry"):
                return validate_bounds(to_shape(feature["geometry"]).bounds)
            else:
                raise TypeError("no bounds")
        except Exception as exc:
            if allow_non_geo_objects:
                return None
            else:
                logger.exception(exc)
                raise TypeError(f"cannot determine bounds from feature: {feature}") def validate_bounds(bounds):
    """
    Return validated bounds.

    Bounds must be a list or tuple with exactly four elements.

    Parameters
    ----------
    bounds : list or tuple

    Returns
    -------
    Bounds

    Raises
    ------
    TypeError if type is invalid.
    """
    if not isinstance(bounds, (tuple, list)):
        raise TypeError("bounds must be either a tuple or a list: %s" % str(bounds))
    if len(bounds) != 4:
        raise ValueError("bounds has to have exactly four values: %s" % str(bounds))
    return Bounds(*bounds) class MapcheteProcess(object):
    """
    Process class inherited by user process script.

    Its attributes and methods can be accessed via "self" from within a
    Mapchete process Python file.

    Parameters
    ----------
    tile : BufferedTile
        Tile process should be run on
    config : MapcheteConfig
        process configuration
    params : dictionary
        process parameters

    Attributes
    ----------
    identifier : string
        process identifier
    title : string
        process title
    version : string
        process version string
    abstract : string
        short text describing process purpose
    tile : BufferedTile
        Tile process should be run on
    tile_pyramid : BufferedTilePyramid
        process tile pyramid
    output_pyramid : BufferedTilePyramid
        output tile pyramid
    params : dictionary
        process parameters
    """

    def __init__(
        self, tile=None, params=None, input=None, output_params=None, config=None
    ):
        """Initialize Mapchete process."""
        self.identifier = ""
        self.title = ""
        self.version = ""
        self.abstract = ""

        self.tile = tile
        self.tile_pyramid = tile.tile_pyramid
        if config is not None:
            input = config.get_inputs_for_tile(tile)
            params = config.params_at_zoom(tile.zoom)
        self.params = dict(params, input=input)
        self.input = input
        self.output_params = output_params

    def write(self, data, **kwargs):
        """Deprecated."""
        raise DeprecationWarning(
            "Please return process output data instead of using self.write()."
        )

    def read(self, **kwargs):
        """
        Read existing output data from a previous run.

        Returns
        -------
        process output : NumPy array (raster) or feature iterator (vector)
        """
        raise DeprecationWarning(
            "Read existing output from within a process is deprecated"
        )

    @deprecated_kwargs
    def open(self, input_id, **kwargs):
        """
        Open input data.

        Parameters
        ----------
        input_id : string
            input identifier from configuration file or file path
        kwargs : driver specific parameters (e.g. resampling)

        Returns
        -------
        tiled input data : InputTile
            reprojected input data within tile
        """
        if input_id not in self.input:
            raise ValueError("%s not found in config as input" % input_id)
        return self.input[input_id]

    def hillshade(self, elevation, azimuth=315.0, altitude=45.0, z=1.0, scale=1.0):
        """
        Calculate hillshading from elevation data.

        Parameters
        ----------
        elevation : array
            input elevation data
        azimuth : float
            horizontal angle of light source (315: North-West)
        altitude : float
            vertical angle of light source (90 would result in slope shading)
        z : float
            vertical exaggeration factor
        scale : float
            scale factor of pixel size units versus height units (insert 112000
            when having elevation values in meters in a geodetic projection)

        Returns
        -------
        hillshade : array
        """
        return commons_hillshade.hillshade(
            elevation, self.tile, azimuth, altitude, z, scale
        )

    def contours(self, elevation, interval=100, field="elev", base=0):
        """
        Extract contour lines from elevation data.

        Parameters
        ----------
        elevation : array
            input elevation data
        interval : integer
            elevation value interval when drawing contour lines
        field : string
            output field name containing elevation value
        base : integer
            elevation base value the intervals are computed from

        Returns
        -------
        contours : iterable
            contours as GeoJSON-like pairs of properties and geometry
        """
        return commons_contours.extract_contours(
            elevation, self.tile, interval=interval, field=field, base=base
        )

    def clip(self, array, geometries, inverted=False, clip_buffer=0):
        """
        Clip array by geometry.

        Parameters
        ----------
        array : array
            raster data to be clipped
        geometries : iterable
            geometries used to clip source array
        inverted : bool
            invert clipping (default: False)
        clip_buffer : int
            buffer (in pixels) geometries before applying clip

        Returns
        -------
        clipped array : array
        """
        return commons_clip.clip_array_with_vector(
            array,
            self.tile.affine,
            geometries,
            inverted=inverted,
            clip_buffer=clip_buffer * self.tile.pixel_x_size,
        )
from collections import namedtuple
from itertools import chain
from shapely.geometry import box, mapping
from traceback import format_exc
from uuid import uuid4
from mapchete._timer import Timer
from mapchete.config import get_process_func
from mapchete.errors import (
    MapcheteNodataTile,
    NoTaskGeometry,
    MapcheteProcessException,
    MapcheteProcessOutputError,
)
from mapchete.io import raster
from mapchete.io._geometry_operations import to_shape
from mapchete.io.vector import IndexedFeatures
from mapchete.validate import validate_bounds
from mapchete._user_process import MapcheteProcess
    from dask.delayed import delayed
import logging

    def _validate(self, item):
        if isinstance(item, Task):
            return item
        else:
            raise TypeError("TaskBatch items must be Taskss, not %s", type(item))


class TileTask(Task):
    """
    Class to process on a specific process tile.

    If skip is set to True, all attributes will be set to None.
    """

    def __init__(self, tile=None, id=None, config=None, skip=False, dependencies=None):
        """Set attributes depending on baselevels or not."""
        self.tile = (
            config.process_pyramid.tile(*tile) if isinstance(tile, tuple) else tile
        )
        self.id = id or f"tile_task_{self.tile.zoom}-{self.tile.row}-{self.tile.col}"
        self.skip = skip
        self.config_zoom_levels = None if skip else config.zoom_levels
        self.config_baselevels = None if skip else config.baselevels
        self.process = None if skip else config.process
        self.config_dir = None if skip else config.config_dir
        if (
            skip
            or self.tile.zoom not in self.config_zoom_levels
            or self.tile.zoom in self.config_baselevels
        ):
            self.input, self.process_func_params, self.output_params = {}, {}, {}
        else:
            self.input = config.get_inputs_for_tile(tile)
            self.process_func_params = config.get_process_func_params(tile.zoom)
            self.output_params = config.output_reader.output_params
        self.mode = None if skip else config.mode
        self.output_reader = (
            None if skip or not config.baselevels else config.output_reader
        )
        super().__init__(
            id=f"tile_task_{tile.zoom}-{tile.row}-{tile.col}", geometry=tile.bbox
        )

    def execute(self, dependencies=None):
        """
        Run the Mapchete process and return the result.

        If baselevels are defined it will generate the result from the other zoom levels
        accordingly.

        Parameters
        ----------
        process_tile : Tile or tile index tuple
            Member of the process tile pyramid (not necessarily the output
            pyramid, if output has a different metatiling setting)

        Returns
        -------
        data : NumPy array or features
            process output
        """
        if self.mode not in ["memory", "continue", "overwrite"]:  # pragma: no cover
            raise ValueError("process mode must be memory, continue or overwrite")

        if self.tile.zoom not in self.config_zoom_levels:
            raise MapcheteNodataTile

        dependencies = dependencies or {}
        process_output = self._execute(dependencies=dependencies)
        if isinstance(process_output, str) and process_output == "empty":
            raise MapcheteNodataTile
        elif process_output is None:
            raise MapcheteProcessOutputError("process output is empty")
        return process_output

    def _execute(self, dependencies=None):
        # If baselevel is active and zoom is outside of baselevel,
        # interpolate from other zoom levels.
        if self.config_baselevels:
            if self.tile.zoom < min(self.config_baselevels["zooms"]):
                return self._interpolate_from_baselevel("lower", dependencies)
            elif self.tile.zoom > max(self.config_baselevels["zooms"]):
                return self._interpolate_from_baselevel("higher", dependencies)
        # Otherwise, execute from process file.
        process_func = get_process_func(
            process=self.process, config_dir=self.config_dir
        )
        try:
            with Timer() as duration:
                # append dependent preprocessing task results to input objects
                if dependencies:
                    for task_key, task_result in dependencies.items():
                        if not task_key.startswith("tile_task"):
                            inp_key, task_key = task_key.split(":")[0], ":".join(
                                task_key.split(":")[1:]
                            )
                            if task_key in [None, ""]:  # pragma: no cover
                                raise KeyError(f"malformed task key: {inp_key}")
                            for inp in self.input.values():
                                if inp.input_key == inp_key:
                                    inp.set_preprocessing_task_result(
                                        task_key=task_key, result=task_result.data
                                    )
                # Actually run process.
                process_data = process_func(
                    MapcheteProcess(
                        tile=self.tile,
                        params=self.process_func_params,
                        input=self.input,
                        output_params=self.output_params,
                    ),
                    **self.process_func_params,
                )
        except MapcheteNodataTile:
            raise
        except Exception as e:
            # Log process time
            logger.exception(e)
            logger.error((self.tile.id, "exception in user process", e, str(duration)))

            new = MapcheteProcessException(format_exc())