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 DaskExecutor(_ExecutorBase):
    """Execute tasks using dask cluster."""

    def __init__(
        self,
        *args,
        address=None,
        dask_scheduler=None,
        dask_client=None,
        max_workers=None,
        **kwargs,
    ):
        from dask.distributed import as_completed, Client, LocalCluster

        self._executor_client = dask_client
        self._local_cluster = None
        if self._executor_client:  # pragma: no cover
            logger.info("using existing dask client: %s", dask_client)
        else:
            local_cluster_kwargs = dict(
                n_workers=max_workers or os.cpu_count(), threads_per_worker=1
            )
            self._executor_cls = Client
            if dask_scheduler is None:
                self._local_cluster = LocalCluster(**local_cluster_kwargs)
            self._executor_kwargs = dict(address=dask_scheduler or self._local_cluster)
            logger.info(
                "starting dask.distributed.Client with kwargs %s", self._executor_kwargs
            )
        self._ac_iterator = as_completed(
            loop=self._executor.loop, with_results=True, raise_errors=True
        )
        self._submitted = 0
        super().__init__(*args, **kwargs)

    def _map(self, func, iterable, fargs=None, fkwargs=None):
        fargs = fargs or []
        fkwargs = fkwargs or {}
        return [
            f.result()
            for f in self._executor.map(partial(func, *fargs, **fkwargs), iterable)
        ]

    def _wait(self):
        from dask.distributed import wait

        wait(self.running_futures)

    def as_completed(
        self,
        func,
        iterable,
        fargs=None,
        fkwargs=None,
        max_submitted_tasks=500,
        item_skip_bool=False,
        chunksize=100,
        **kwargs,
    ):
        """
        Submit tasks to executor and start yielding finished futures.

        Sometimes dask catches an exception (e.g. KilledWorker) and sends a cancel signal to all
        remaining futures. In these cases it can happen that we get a cancelled future before the
        future which contains the KilledWorker (or whatever the source exception was).
        Here we try to iterate through the remaining futures until we can raise the original
        exception as it may contain more relevant information

        Parameters
        ----------
        func : function
            Function to paralellize.
        iterable : iterable
            Iterable with items to parallelize. Each item should be the first argument of
            function.
        fargs : tuple
            Further function arguments.
        fkwargs : dict
            Further function keyword arguments.
        max_submitted_tasks : int
            Make sure that not more tasks are submitted to dask scheduler at once. (default: 500)
        chunksize : int
            Submit tasks in chunks to scheduler.

        Yields
        ------
        finished futures

        """
        from dask.distributed import TimeoutError

        max_submitted_tasks = max_submitted_tasks or 1
        chunksize = chunksize or 1

        try:
            fargs = fargs or ()
            fkwargs = fkwargs or {}
            chunk = []
            for item in iterable:

                # abort if execution is cancelled
                if self.cancelled:  # pragma: no cover
                    logger.debug("executor cancelled")
                    return

                # skip task submission if option is activated
                if item_skip_bool:
                    item, skip, skip_info = item
                    if skip:
                        yield SkippedFuture(item, skip_info=skip_info)
                        continue

                # add processing item to chunk
                chunk.append(item)

                # submit chunk of tasks, if
                # (1) chunksize is reached, or
                # (2) remaining free task spots are less than tasks in chunk
                remaining_spots = max_submitted_tasks - self._submitted
                if len(chunk) % chunksize == 0 or remaining_spots == len(chunk):
                    logger.debug("submitted futures: %s", self._submitted)
                    logger.debug("remaining spots for futures: %s", remaining_spots)
                    logger.debug("current chunk size: %s", len(chunk))
                    self._submit_chunk(
                        chunk=chunk,
                        func=func,
                        fargs=fargs,
                        fkwargs=fkwargs,
                    )
                    chunk = []

                # yield finished tasks, if
                # (1) there are finished tasks available, or
                # (2) maximum allowed number of running tasks is reached
                max_submitted_tasks_reached = self._submitted >= max_submitted_tasks
                if self._ac_iterator.has_ready() or max_submitted_tasks_reached:
                    # yield batch of finished futures
                    # if maximum submitted tasks limit is reached, block call and wait for finished futures
                    logger.debug(
                        "wait for finished tasks: %s", max_submitted_tasks_reached
                    )
                    batch = self._ac_iterator.next_batch(
                        block=max_submitted_tasks_reached
                    )
                    try:
                        yield from self._yield_from_batch(batch)
                    except JobCancelledError:  # pragma: no cover
                        return
                    logger.debug("%s futures still on cluster", self._submitted)

            # submit last chunk of items
            self._submit_chunk(
                chunk=chunk,
                func=func,
                fargs=fargs,
                fkwargs=fkwargs,
            )
            chunk = []
            # yield remaining futures as they finish
            if self._ac_iterator is not None:
                logger.debug("yield %s remaining futures", self._submitted)
                for batch in self._ac_iterator.batches():
                    try:
                        yield from self._yield_from_batch(batch)
                    except JobCancelledError:  # pragma: no cover
                        return

        finally:
            # reset so futures won't linger here for next call
            self.running_futures = set()
            self._ac_iterator.clear()
            self._submitted = 0

    def _submit_chunk(self, chunk=None, func=None, fargs=None, fkwargs=None):
        logger.debug("submit chunk of %s items to cluster", len(chunk))
        futures = self._executor.map(partial(func, *fargs, **fkwargs), chunk)
        self._ac_iterator.update(futures)
        self._submitted += len(futures)

    def _yield_from_batch(self, batch):
        from dask.distributed import TimeoutError
        from distributed.comm.core import CommClosedError

        cancelled_futures = []

        for future, result in batch:
            self._submitted -= 1
            if self.cancelled:  # pragma: no cover
                logger.debug("executor cancelled")
                raise JobCancelledError()
            try:
                yield self._finished_future(future, result, _dask=True)
            except TimeoutError:  # pragma: no cover
                logger.error(
                    "%s: couldn't fetch future result() or exception() in %ss",
                    future,
                    FUTURE_TIMEOUT,
                )
                self._retry(future)
            except CancelledError as exc:  # pragma: no cover
                logger.error("%s got cancelled: %s", future, exc)
                cancelled_futures.append(future)

        if cancelled_futures:  # pragma: no cover
            logger.error("caught %s cancelled_futures", len(cancelled_futures))
            try:
                logger.debug("try to get scheduler logs...")
                logger.debug(
                    "scheduler logs: %s", self._executor.get_scheduler_logs(n=1000)
                )
            except Exception as e:
                logger.exception(e)
            status = self._executor.status
            if status in ("running", "connecting"):
                try:
                    logger.debug("retry %s futures...", len(cancelled_futures))
                    for future in cancelled_futures:
                        self._retry(future)
                except KeyError:
                    raise RuntimeError(
                        f"unable to retry {len(cancelled_futures)} cancelled futures {self._executor} ({status})"
                    )
            else:
                raise RuntimeError(
                    f"client lost connection to scheduler {self._executor} ({status})"
                )

    def _retry(self, future):  # pragma: no cover
        logger.debug("retry future %s", future)
        future.retry()
        self._ac_iterator.add(future)
        self._submitted += 1

    @cached_property
    def _executor(self):
        return self._executor_client or self._executor_cls(
            *self._executor_args, **self._executor_kwargs
        )

    def __exit__(self, *args):
        """Exit context manager."""
        if self._executor_client:  # pragma: no cover
            logger.info("client not closing as it was passed on as kwarg")
        else:
            logger.info("closing executor %s...", self._executor)
            try:
                self._executor.close()
            except Exception:  # pragma: no cover
                self._executor.__exit__(*args)
            logger.info("closed executor %s", self._executor)
        if self._local_cluster:
            logger.info("closing %s", self._local_cluster)
            self._local_cluster.close() class Executor:
    """
    Executor factory for dask and concurrent.futures executor

    Will move into the mapchete core package.
    """

    def __new__(cls, *args, concurrency=None, **kwargs):
        if concurrency == "dask":
            try:
                return DaskExecutor(*args, **kwargs)
            except ImportError as exc:  # pragma: no cover
                raise ImportError(
                    "this feature requires the mapchete[dask] extra"
                ) from exc

        elif concurrency is None or kwargs.get("max_workers") == 1:
            return SequentialExecutor(*args, **kwargs)

        elif concurrency in ["processes", "threads"]:
            return ConcurrentFuturesExecutor(*args, concurrency=concurrency, **kwargs)

        else:  # pragma: no cover
            raise ValueError(
                f"concurrency must be one of None, 'processes', 'threads' or 'dask', not {concurrency}"
            ) class SkippedFuture:
    """Wrapper class to mimick future interface for empty tasks."""

    def __init__(self, result=None, skip_info=None, **kwargs):
        self._result = result
        self.skip_info = skip_info

    def result(self):
        """Only return initial result value."""
        return self._result

    def exception(self):  # pragma: no cover
        """Nothing to raise here."""
        return

    def cancelled(self):  # pragma: no cover
        """Nothing to cancel here."""
        return False

    def __repr__(self):  # pragma: no cover
        """Return string representation."""
        return f"<SkippedFuture: type: {type(self._result)}, exception: {type(self._exception)})" class FinishedFuture:
    """Wrapper class to mimick future interface."""

    def __init__(self, future=None, result=None):
        """Set attributes."""
        try:
            self._result, self._exception = (
                result or future.result(timeout=FUTURE_TIMEOUT),
                None,
            )
        except Exception as e:  # pragma: no cover
            self._result, self._exception = None, e

    def result(self):
        """Return task result."""
        if self._exception:  # pragma: no cover
            logger.exception(self._exception)
            raise self._exception
        return self._result

    def exception(self):  # pragma: no cover
        """Raise task exception if any."""
        return self._exception

    def cancelled(self):  # pragma: no cover
        """Sequential futures cannot be cancelled."""
        return False

    def __repr__(self):  # pragma: no cover
        """Return string representation."""
        return f"<FinishedFuture: type: {type(self._result)}, exception: {type(self._exception)})" class MapcheteNodataTile(Exception):
    """Indicates an empty tile.""" def to_dask_collection(batches):
    from dask.delayed import delayed

    tasks = {}
    previous_batch = None
    for batch in batches:
        logger.debug("converting batch %s", batch)
        if previous_batch:
            logger.debug("previous batch had %s tasks", len(previous_batch))
        for task in batch.values():
            if previous_batch:
                dependencies = {
                    child.id: tasks[child]
                    for child in previous_batch.intersection(task)
                }
                logger.debug(
                    "found %s dependencies from last batch for task %s",
                    len(dependencies),
                    task,
                )
            else:
                dependencies = {}
            tasks[task] = delayed(batch.func)(
                task, dependencies=dependencies, **batch.fkwargs
            )
        previous_batch = batch
    return list(tasks.values()) class TileTaskBatch(TaskBatch):
    """Combines TileTask instances of same pyramid and zoom level into one batch."""

    def __init__(self, tasks=None, id=None, func=None, fkwargs=None):
        self.id = id or uuid4().hex
        self.bounds = None, None, None, None
        self.tasks = {tile: item for tile, item in self._validate(tasks)}
        self._update_bounds()
        self.func = func or _execute_task_wrapper
        self.fkwargs = fkwargs or {}

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

    def intersection(self, other):
        if isinstance(other, TileTask):
            if other.tile.zoom + 1 != self._zoom:  # pragma: no cover
                raise ValueError("intersecting tile has to be from zoom level above")
            return [
                self.tasks[child]
                for child in other.tile.get_children()
                if child in self.tasks
            ]

        if isinstance(other, Task):
            return self.intersection(other.bounds)

        elif isinstance(other, (tuple, list)):
            return [
                self.tasks[tile]
                for tile in self._tp.tiles_from_bounds(bounds=other, zoom=self._zoom)
                if tile in self.tasks
            ]

        else:
            raise TypeError(
                "intersections only works with other Task instances or bounds"
            )

    def _validate(self, items):
        self._tp = None
        self._zoom = None
        for item in items:
            if not isinstance(item, TileTask):  # pragma: no cover
                raise TypeError(
                    "TileTaskBatch items must be TileTasks, not %s", type(item)
                )
            if self._tp is None:
                self._tp = item.tile.buffered_tp
            elif item.tile.buffered_tp != self._tp:  # pragma: no cover
                raise TypeError("all TileTasks must derive from the same pyramid.")
            if self._zoom is None:
                self._zoom = item.tile.zoom
            elif item.tile.zoom != self._zoom:  # pragma: no cover
                raise TypeError("all TileTasks must lie on the same zoom level")
            yield item.tile, 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())
            new.old = e
            raise new

        return process_data

    def _interpolate_from_baselevel(self, baselevel=None, dependencies=None):
        # This is a special tile derived from a pyramid which has the pixelbuffer setting
        # from the output pyramid but metatiling from the process pyramid. This is due to
        # performance reasons as for the usual case overview tiles do not need the
        # process pyramid pixelbuffers.
        tile = self.config_baselevels["tile_pyramid"].tile(*self.tile.id)

        # get output_tiles that intersect with process tile
        output_tiles = (
            list(self.output_reader.pyramid.tiles_from_bounds(tile.bounds, tile.zoom))
            if tile.pixelbuffer > self.output_reader.pyramid.pixelbuffer
            else self.output_reader.pyramid.intersecting(tile)
        )

        with Timer() as duration:
            # resample from parent tile
            if baselevel == "higher":
                parent_tile = self.tile.get_parent()
                process_data = raster.resample_from_array(
                    self.output_reader.read(parent_tile),
                    in_affine=parent_tile.affine,
                    out_tile=self.tile,
                    resampling=self.config_baselevels["higher"],
                    nodata=self.output_reader.output_params["nodata"],
                )
            # resample from children tiles
            elif baselevel == "lower":
                src_tiles = {}
                for process_info in dependencies.values():
                    logger.debug("reading output from dependend tasks")
                    process_tile = process_info.tile
                    for output_tile in self.output_reader.pyramid.intersecting(
                        process_tile
                    ):
                        if process_info.data is not None:
                            src_tiles[output_tile] = raster.extract_from_array(
                                in_raster=process_info.data,
                                in_affine=process_tile.affine,
                                out_tile=output_tile,
                            )
                if self.output_reader.pyramid.pixelbuffer:  # pragma: no cover
                    # if there is a pixelbuffer around the output tiles, we need to read more child tiles
                    child_tiles = [
                        child_tile
                        for output_tile in output_tiles
                        for child_tile in self.output_reader.pyramid.tiles_from_bounds(
                            output_tile.bounds, output_tile.zoom + 1
                        )
                    ]
                else:
                    child_tiles = [
                        child_tile
                        for output_tile in output_tiles
                        for child_tile in output_tile.get_children()
                    ]
                for child_tile in child_tiles:
                    if child_tile not in src_tiles:
                        src_tiles[child_tile] = self.output_reader.read(child_tile)

                process_data = raster.resample_from_array(
                    in_raster=raster.create_mosaic(
                        [(src_tile, data) for src_tile, data in src_tiles.items()],
                        nodata=self.output_reader.output_params["nodata"],
                    ),
                    out_tile=self.tile,
                    resampling=self.config_baselevels["lower"],
                    nodata=self.output_reader.output_params["nodata"],
                )
        logger.debug((self.tile.id, "generated from baselevel", str(duration)))
        return process_data class TaskBatch:
    def __init__(self, tasks=None, id=None, func=None, fkwargs=None):
        if tasks is None:
            raise TypeError("TaskBatch requires at least one Task")
        self.id = id or uuid4().hex
        self.tasks = IndexedFeatures(
            (self._validate(t) for t in tasks), allow_non_geo_objects=True
        )
        self.bounds = self.tasks.bounds
        self.func = func or _execute_task_wrapper
        self.fkwargs = fkwargs or {}

    def __repr__(self):  # pragma: no cover
        return f"TaskBatch(id={self.id}, bounds={self.bounds})"

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

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

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

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

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

    def intersection(self, other):
        if isinstance(other, Task):
            return self.tasks.filter(other.bounds)
        elif isinstance(other, (tuple, list)):
            return self.tasks.filter(validate_bounds(other))
        else:
            raise TypeError(
                f"intersection only works with other Task instances or bounds, not {other}"
            )

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

    Can optionally have spatial properties attached which helps building up dependencies
    between tasks.
    """

    def __init__(
        self,
        id=None,
        func=None,
        fargs=None,
        fkwargs=None,
        geometry=None,
        bounds=None,
        dependencies=None,
    ):
        self.id = id or uuid4().hex
        self.func = func
        self.fargs = fargs or ()
        self.fkwargs = fkwargs or {}
        self.dependencies = dependencies or {}
        if geometry and bounds:
            raise ValueError("only provide one of either 'geometry' or 'bounds'")
        elif geometry:
            self.geometry = to_shape(geometry)
            self.bounds = self.geometry.bounds
        elif bounds:
            self.bounds = validate_bounds(bounds)
            self.geometry = box(*self.bounds)
        else:
            self.bounds, self.geometry = None, None

    def __repr__(self):  # pragma: no cover
        return f"Task(id={self.id}, bounds={self.bounds})"

    def to_dict(self):
        return {
            "id": self.id,
            "bounds": self.bounds,
            "geometry": mapping(self.geometry) if self.geometry else None,
            "properties": {},
        }

    def add_dependencies(self, dependencies):
        dependencies = dependencies or {}
        if not isinstance(dependencies, dict):
            raise TypeError(
                f"dependencies must be a dictionary, not {type(dependencies)}"
            )
        self.dependencies.update(dependencies)

    def execute(self, dependencies=None):
        return self.func(*self.fargs, **self.fkwargs)

    def has_geometry(self):
        return self.geometry is not None

    @property
    def __geo_interface__(self):
        if self.has_geometry():
            return mapping(self.geometry)
        else:
            raise NoTaskGeometry(f"{self} has no geo information assigned") 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 validate_zooms(zooms, expand=True):
    """
    Return a list of zoom levels.

    Following inputs are converted:
    - int --> [int]
    - dict{min, max} --> [min ... max + 1]
    - [int] --> [int]
    - [int, int] --> [smaller int, bigger int + 1]

    Parameters
    ----------
    zoom : dict, int or list
    expand : bool
        Return full list of zoom levels instead of [min, max]

    Returns
    -------
    List of zoom levels.
    """
    if isinstance(zooms, dict):
        if any([a not in zooms for a in ["min", "max"]]):
            raise TypeError("min and max zoom required: %s" % str(zooms))
        zmin = validate_zoom(zooms["min"])
        zmax = validate_zoom(zooms["max"])
        if zmin > zmax:
            raise TypeError(
                "max zoom must not be smaller than min zoom: %s" % str(zooms)
            )
        return list(range(zmin, zmax + 1)) if expand else zooms
    elif isinstance(zooms, list):
        if len(zooms) == 1:
            return zooms
        elif len(zooms) == 2:
            zmin, zmax = sorted([validate_zoom(z) for z in zooms])
            return list(range(zmin, zmax + 1)) if expand else zooms
        else:
            return zooms
    else:
        return [validate_zoom(zooms)] if expand else zooms
from collections import namedtuple
from contextlib import ExitStack
from itertools import chain
from shapely.geometry import mapping
from tilematrix._funcs import Bounds
from traceback import format_exc
from typing import Generator
from mapchete.config import get_process_func
from mapchete._executor import DaskExecutor, Executor, SkippedFuture, FinishedFuture
from mapchete.errors import MapcheteNodataTile
from mapchete._tasks import to_dask_collection, TileTaskBatch, TileTask, TaskBatch, Task
from mapchete._timer import Timer
from mapchete.validate import validate_zooms
    from dask.delayed import delayed
    from distributed import as_completed
import logging
import multiprocessing
            yield future


def _filter_skipable(
    process=None, tiles_batches=None, target_set=None, skip_output_check=False
):
    if skip_output_check:  # pragma: no cover
        for batch in tiles_batches:
            for tile in batch:
                yield (tile, False, None)
    else:
        target_set = target_set or set()
        for tile, skip in process.skip_tiles(tiles_batches=tiles_batches):
            if skip and tile not in target_set:
                yield (tile, True, "output already exists")
            else:
                yield (tile, False, None)


def _run_multi(
    executor=None,
    func=None,
    zoom_levels=None,
    process=None,
    dask_scheduler=None,
    dask_max_submitted_tasks=500,
    dask_chunksize=100,
    workers=None,
    multiprocessing_start_method=None,
    multiprocessing_module=None,
    write_in_parent_process=False,
    fkwargs=None,
    skip_output_check=False,
):
    total_tiles = process.count_tiles(min(zoom_levels), max(zoom_levels))
    workers = min([workers, total_tiles])
    num_processed = 0

    # If an Executor is passed on, don't close after processing. If no Executor is passed on,
    # create one and properly close it afterwards.
    with ExitStack() as exit_stack:
        # create fresh Executor if it was not passed on
        if executor is None:
            executor = exit_stack.enter_context(
                Executor(
                    max_workers=workers,
                    concurrency="dask" if dask_scheduler else "processes",
                    start_method=multiprocessing_start_method,
                    multiprocessing_module=multiprocessing_module,
                    dask_scheduler=dask_scheduler,
                )
            )
        with Timer() as duration:
            logger.info(
                "run process on %s tiles using %s workers on executor %s",
                total_tiles,
                workers,
                executor,
            )
            if process.config.baselevels:
                f = _run_multi_overviews
            else:
                f = _run_multi_no_overviews
            for num_processed, future in enumerate(
                f(
                    zoom_levels=zoom_levels,
                    executor=executor,
                    func=func,
                    process=process,
                    skip_output_check=skip_output_check,
                    fkwargs=fkwargs,
                    dask_chunksize=dask_chunksize,
                    dask_max_submitted_tasks=dask_max_submitted_tasks,
                    write_in_parent_process=write_in_parent_process,
                ),
                1,
            ):
                logger.debug("task %s finished: %s", num_processed, future)
                yield future

        logger.info("%s tile(s) iterated in %s", str(num_processed), duration)


def _compute_task_graph(
    executor=None,
    process=None,
    skip_output_check=False,
    zoom_levels=None,
    tile=None,
    with_results=False,
    raise_errors=False,
    **kwargs,
):
    # TODO optimize memory management, e.g. delete preprocessing tasks from input
    # once the dask graph is ready.

    # materialize all tasks including dependencies
    with Timer() as t:
        coll = to_dask_collection(
            process.task_batches(
                zoom=zoom_levels, tile=tile, skip_output_check=skip_output_check
            )
        )
    logger.debug("dask collection with %s tasks generated in %s", len(coll), t)

    # send to scheduler
    with Timer() as t:
        futures = executor._executor.compute(coll, optimize_graph=True, traverse=True)
    logger.debug("%s tasks sent to scheduler in %s", len(futures), t)

    logger.debug("wait for tasks to finish...")
    for batch in as_completed(
        futures,
        with_results=with_results,
        raise_errors=raise_errors,
        loop=executor._executor.loop,
    ).batches():
        for future in batch:
            futures.remove(future)
            if process.config.output.write_in_parent_process:

                yield FinishedFuture(