_MANDATORY_PARAMETERS = [
    ("process", (str, list, type(None))),  # path to .py file or module path
    ("pyramid", dict),  # process pyramid definition
    ("input", (dict, type(None))),  # files & other types
    ("output", dict),  # process output parameters
    ("zoom_levels", (int, dict, list)),  # process zoom levels
]
_RESERVED_PARAMETERS = [
    "area",  # geometry limiting process area
    "area_crs",  # optional CRS of area (default: process CRS)
    "baselevels",  # enable interpolation from other zoom levels
    "bounds",  # process bounds
    "bounds_crs",  # optional CRS of bounds (default: process CRS)
    "config_dir",  # configuration base directory
    "metatiling",  # process metatile size (deprecated)
    "pixelbuffer",  # buffer around each tile in pixels (deprecated)
    "process",  # path to .py file or module path
    "process_minzoom",  # minimum zoom where process is valid (deprecated)
    "process_maxzoom",  # maximum zoom where process is valid (deprecated)
    "process_zoom",  # single zoom where process is valid (deprecated)
    "process_bounds",  # process boundaries (deprecated)
]
_OUTPUT_PARAMETERS = [
    "format",
    "path",
    "grid",
    "pixelbuffer",
    "metatiling",
    "delimiters",
    "mode",
    "stac",
]
class MapcheteConfig(object):
    def __init__(
        self,
        input_config,
        zoom=None,
        area=None,
        area_crs=None,
        bounds=None,
        bounds_crs=None,
        single_input_file=None,
        mode="continue",
        debug=False,
        **kwargs,
    ):
    def input_at_zoom(self, key=None, zoom=None):
    def preprocessing_tasks_per_input(self):
    def preprocessing_tasks(self):
    def preprocessing_tasks_count(self):
    def preprocessing_task_finished(self, task_key):
    def set_preprocessing_task_result(self, task_key, result):
    def zoom_levels(self):
    def init_zoom_levels(self):
    def effective_bounds(self):
    def _output_params(self):
    def output(self):
    def output_reader(self):
    def input(self):
    def baselevels(self):
    def process_func(self):
    def get_process_func_params(self, zoom):
    def get_inputs_for_tile(self, tile):
    def params_at_zoom(self, zoom):
    def area_at_zoom(self, zoom=None):
    def _area_at_zoom(self, zoom):
    def bounds_at_zoom(self, zoom=None):
    def _get_process_area(
        self,
        area=None,
        bounds=None,
        area_fallback=None,
        bounds_fallback=None,
        area_crs=None,
        bounds_crs=None,
    ):
    def crs(self):
    def metatiling(self):
    def pixelbuffer(self):
    def inputs(self):
    def process_file(self):
    def at_zoom(self, zoom):
    def process_area(self, zoom=None):
    def process_bounds(self, zoom=None):
def get_hash(x, length=16):
def get_zoom_levels(process_zoom_levels=None, init_zoom_levels=None):
def snap_bounds(bounds=None, pyramid=None, zoom=None):
def clip_bounds(bounds=None, clip=None):
def raw_conf(mapchete_file):
def raw_conf_process_pyramid(raw_conf, reset_pixelbuffer=False):
def raw_conf_output_pyramid(raw_conf):
def bounds_from_opts(
    wkt_geometry=None,
    point=None,
    point_crs=None,
    zoom=None,
    bounds=None,
    bounds_crs=None,
    raw_conf=None,
):
def get_process_func(process=None, config_dir=None, run_compile=False):
def initialize_inputs(
    raw_inputs,
    config_dir=None,
    pyramid=None,
    delimiters=None,
    readonly=False,
):
def _load_process_module(process=None, config_dir=None, run_compile=False):
def open_inputs(inputs, tile):
def _config_to_dict(input_config):
def _raw_at_zoom(config, zooms):
def _element_at_zoom(name, element, zoom):
def _filter_by_zoom(element=None, conf_string=None, zoom=None):
def _strip_zoom(input_string, strip_string):
def _flatten_tree(tree, old_path=None):
def _unflatten_tree(flat):
def _map_to_new_config(config):
def _guess_geometry(i, base_dir=None): class MapcheteConfigError(ValueError):
    """Raised when a mapchete process configuration is invalid.""" class InputData(object):
class InputTile(object):
class OutputDataBaseFunctions:
class OutputDataReader(OutputDataBaseFunctions):
class OutputDataWriter(OutputDataReader):
class TileDirectoryOutputReader(OutputDataReader):
class TileDirectoryOutputWriter(OutputDataWriter, TileDirectoryOutputReader):
class SingleFileOutputReader(OutputDataReader):
class SingleFileOutputWriter(OutputDataWriter, SingleFileOutputReader):
    METADATA = {"driver_name": None, "data_type": None, "mode": "r"}
    METADATA = {"driver_name": None, "data_type": None, "mode": "w"}
    def __init__(self, input_params, input_key=None, **kwargs):
    def open(self, tile, **kwargs):
    def bbox(self, out_crs=None):
    def exists(self):
    def cleanup(self):
    def add_preprocessing_task(
        self, func, fargs=None, fkwargs=None, key=None, geometry=None, bounds=None
    ):
    def get_preprocessing_task_result(self, task_key):
    def set_preprocessing_task_result(self, task_key, result):
    def preprocessing_task_finished(self, task_key):
    def __init__(self, tile, **kwargs):
    def read(self, **kwargs):
    def is_empty(self):
    def set_preprocessing_task_result(self, task_key=None, result=None):
    def __enter__(self):
    def __exit__(self, t, v, tb):
    def __init__(self, output_params, readonly=False, **kwargs):
    def is_valid_with_config(self, config):
    def get_path(self, tile):
    def extract_subset(self, input_data_tiles=None, out_tile=None):
    def prepare(self, **kwargs):
    def read(self, output_tile):
    def empty(self, process_tile):
    def open(self, tile, process):
    def for_web(self, data):
    def write(self, process_tile, data):
    def prepare_path(self, tile):
    def output_is_valid(self, process_data):
    def output_cleaned(self, process_data):
    def streamline_output(self, process_data):
    def close(self, exc_type=None, exc_value=None, exc_traceback=None):
    def __init__(self, output_params, readonly=False):
    def tiles_exist(self, process_tile=None, output_tile=None):
    def _read_as_tiledir(
        self,
        out_tile=None,
        td_crs=None,
        tiles_paths=None,
        profile=None,
        validity_check=False,
        indexes=None,
        resampling=None,
        dst_nodata=None,
        gdal_opts=None,
        **kwargs,
    ):
    def __init__(self, output_params, readonly=False):
    def tiles_exist(self, process_tile=None, output_tile=None):
def is_numpy_or_masked_array(data):
def is_numpy_or_masked_array_with_tags(data):
def is_feature_list(data):
def _read_as_tiledir(
    data_type=None,
    out_tile=None,
    td_crs=None,
    tiles_paths=None,
    profile=None,
    validity_check=False,
    indexes=None,
    resampling=None,
    dst_nodata=None,
    gdal_opts=None,
    **kwargs,
): def path_is_remote(path, s3=True):
    """
    Determine whether file path is remote or local.

    Parameters
    ----------
    path : path to file

    Returns
    -------
    is_remote : bool
    """
    prefixes = ("http://", "https://", "/vsicurl/")
    if s3:
        prefixes += ("s3://", "/vsis3/")
    return path.startswith(prefixes) def path_exists(path, fs=None, **kwargs):
    """
    Check if file exists either remote or local.

    Parameters
    ----------
    path : path to file

    Returns
    -------
    exists : bool
    """
    fs = fs or fs_from_path(path, **kwargs)
    fs.invalidate_cache(path=path)
    return fs.exists(path) def makedirs(path):
    """
    Silently create all subdirectories of path if path is local.

    Parameters
    ----------
    path : path
    """
    fs = fs_from_path(path)
    # create parent directories on local filesystems
    if fs.protocol == "file":
        fs.makedirs(path, exist_ok=True) def write_raster_window(
    in_tile=None,
    in_data=None,
    out_profile=None,
    out_tile=None,
    out_path=None,
    tags=None,
    fs=None,
    **kwargs,
):
    """
    Write a window from a numpy array to an output file.

    Parameters
    ----------
    in_tile : ``BufferedTile``
        ``BufferedTile`` with a data attribute holding NumPy data
    in_data : array
    out_profile : dictionary
        metadata dictionary for rasterio
    out_tile : ``Tile``
        provides output boundaries; if None, in_tile is used
    out_path : string
        output path to write to
    tags : optional tags to be added to GeoTIFF file
    """
    if not isinstance(out_path, str):
        raise TypeError("out_path must be a string")
    logger.debug("write %s", out_path)
    if out_path == "memoryfile":
        raise DeprecationWarning(
            "Writing to memoryfile with write_raster_window() is deprecated. "
            "Please use RasterWindowMemoryFile."
        )
    out_tile = out_tile or in_tile
    validate_write_window_params(in_tile, out_tile, in_data, out_profile)

    # extract data
    window_data = (
        extract_from_array(
            in_raster=in_data, in_affine=in_tile.affine, out_tile=out_tile
        )
        if in_tile != out_tile
        else in_data
    )

    # use transform instead of affine
    if "affine" in out_profile:
        out_profile["transform"] = out_profile.pop("affine")

    # write if there is any band with non-masked data
    if window_data.all() is not ma.masked:

        try:
            with rasterio_write(out_path, "w", fs=fs, **out_profile) as dst:
                logger.debug((out_tile.id, "write tile", out_path))
                dst.write(window_data.astype(out_profile["dtype"], copy=False))
                _write_tags(dst, tags)
        except Exception as e:
            logger.exception("error while writing file %s: %s", out_path, e)
            raise
    else:
        logger.debug((out_tile.id, "array window empty", out_path)) def prepare_array(data, masked=True, nodata=0, dtype="int16"):
    """
    Turn input data into a proper array for further usage.

    Output array is always 3-dimensional with the given data type. If the output
    is masked, the fill_value corresponds to the given nodata value and the
    nodata value will be burned into the data array.

    Parameters
    ----------
    data : array or iterable
        array (masked or normal) or iterable containing arrays
    nodata : integer or float
        nodata value (default: 0) used if input is not a masked array and
        for output array
    masked : bool
        return a NumPy Array or a NumPy MaskedArray (default: True)
    dtype : string
        data type of output array (default: "int16")

    Returns
    -------
    array : array
    """
    # input is iterable
    if isinstance(data, (list, tuple)):
        return _prepare_iterable(data, masked, nodata, dtype)

    # special case if a 2D single band is provided
    elif isinstance(data, np.ndarray) and data.ndim == 2:
        data = ma.expand_dims(data, axis=0)

    # input is a masked array
    if isinstance(data, ma.MaskedArray):
        return _prepare_masked(data, masked, nodata, dtype)

    # input is a NumPy array
    elif isinstance(data, np.ndarray):
        if masked:
            return ma.masked_values(data.astype(dtype, copy=False), nodata, copy=False)
        else:
            return data.astype(dtype, copy=False)
    else:
        raise ValueError(
            "Data must be array, masked array or iterable containing arrays. "
            "Current data: %s (%s)" % (data, type(data))
        ) def memory_file(data=None, profile=None):
    """
    Return a rasterio.io.MemoryFile instance from input.

    Parameters
    ----------
    data : array
        array to be written
    profile : dict
        rasterio profile for MemoryFile
    """
    memfile = MemoryFile()
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        with memfile.open(
            **dict(profile, width=data.shape[-2], height=data.shape[-1])
        ) as dataset:
            dataset.write(data)
        return memfile @retry(logger=logger, exceptions=RasterioIOError, **MAPCHETE_IO_RETRY_SETTINGS)
def read_raster_no_crs(input_file, indexes=None, gdal_opts=None):
    """
    Wrapper function around rasterio.open().read().

    Parameters
    ----------
    input_file : str
        Path to file
    indexes : int or list
        Band index or list of band indexes to be read.
    gdal_opts : dict
        GDAL options passed on to rasterio.Env()

    Returns
    -------
    MaskedArray

    Raises
    ------
    FileNotFoundError if file cannot be found.
    """
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        try:
            with rasterio.Env(
                **get_gdal_options(
                    gdal_opts,
                    is_remote=path_is_remote(input_file, s3=True),
                    allowed_remote_extensions=os.path.splitext(input_file)[1],
                ),
            ) as env:
                logger.debug("reading %s with GDAL options %s", input_file, env.options)
                with rasterio.open(input_file, "r") as src:
                    return src.read(indexes=indexes, masked=True)
        except RasterioIOError as e:
            try:
                if path_exists(input_file):
                    raise MapcheteIOError(e)
            except Exception:
                raise MapcheteIOError(e)
            raise FileNotFoundError("%s not found" % input_file) def extract_from_array(in_raster=None, in_affine=None, out_tile=None):
    """
    Extract raster data window array.

    Parameters
    ----------
    in_raster : array or ReferencedRaster
    in_affine : ``Affine`` required if in_raster is an array
    out_tile : ``BufferedTile``

    Returns
    -------
    extracted array : array
    """
    if isinstance(in_raster, ReferencedRaster):  # pragma: no cover
        in_affine, in_raster = in_raster.affine, in_raster.data

    # get range within array
    minrow, maxrow, mincol, maxcol = bounds_to_ranges(
        out_bounds=out_tile.bounds, in_affine=in_affine, in_shape=in_raster.shape
    )
    # if output window is within input window
    if (
        minrow >= 0
        and mincol >= 0
        and maxrow <= in_raster.shape[-2]
        and maxcol <= in_raster.shape[-1]
    ):
        return in_raster[..., minrow:maxrow, mincol:maxcol]
    # raise error if output is not fully within input
    else:
        raise ValueError("extraction fails if output shape is not within input") def rasterio_write(path, mode=None, fs=None, in_memory=True, *args, **kwargs):
    """
    Wrap rasterio.open() but handle bucket upload if path is remote.

    Parameters
    ----------
    path : str
        Path to write to.
    mode : str
        One of the rasterio.open() modes.
    fs : fsspec.FileSystem
        Target filesystem.
    in_memory : bool
        On remote output store an in-memory file instead of writing to a tempfile.
    args : list
        Arguments to be passed on to rasterio.open()
    kwargs : dict
        Keyword arguments to be passed on to rasterio.open()

    Returns
    -------
    RasterioRemoteWriter if target is remote, otherwise return rasterio.open().
    """
    if path.startswith("s3://"):
        return RasterioRemoteWriter(path, fs=fs, in_memory=in_memory, *args, **kwargs)
    else:
        return rasterio.open(path, mode=mode, *args, **kwargs) class BufferedTile(Tile):
    """
    A special tile with fixed pixelbuffer.

    Parameters
    ----------
    tile : ``Tile``
    pixelbuffer : integer
        tile buffer in pixels

    Attributes
    ----------
    height : integer
        tile height in pixels
    width : integer
        tile width in pixels
    shape : tuple
        tile width and height in pixels
    affine : ``Affine``
        ``Affine`` object describing tile extent and pixel size
    bounds : tuple
        left, bottom, right, top values of tile boundaries
    bbox : ``shapely.geometry``
        tile bounding box as shapely geometry
    pixelbuffer : integer
        pixelbuffer used to create tile
    profile : dictionary
        rasterio metadata profile
    """

    def __init__(self, tile, pixelbuffer=0):
        """Initialize."""
        if isinstance(tile, BufferedTile):
            tile = TilePyramid(
                tile.tp.grid, tile_size=tile.tp.tile_size, metatiling=tile.tp.metatiling
            ).tile(*tile.id)
        Tile.__init__(self, tile.tile_pyramid, tile.zoom, tile.row, tile.col)
        self._tile = tile
        self.pixelbuffer = pixelbuffer
        self.buffered_tp = BufferedTilePyramid(
            tile.tp.to_dict(), pixelbuffer=pixelbuffer
        )

    @cached_property
    def left(self):
        return self.bounds.left

    @cached_property
    def bottom(self):
        return self.bounds.bottom

    @cached_property
    def right(self):
        return self.bounds.right

    @cached_property
    def top(self):
        return self.bounds.top

    @cached_property
    def height(self):
        """Return buffered height."""
        return self._tile.shape(pixelbuffer=self.pixelbuffer).height

    @cached_property
    def width(self):
        """Return buffered width."""
        return self._tile.shape(pixelbuffer=self.pixelbuffer).width

    @cached_property
    def shape(self):
        """Return buffered shape."""
        return self._tile.shape(pixelbuffer=self.pixelbuffer)

    @cached_property
    def affine(self):
        """Return buffered Affine."""
        return self._tile.affine(pixelbuffer=self.pixelbuffer)

    @cached_property
    def bounds(self):
        """Return buffered bounds."""
        return self._tile.bounds(pixelbuffer=self.pixelbuffer)

    @cached_property
    def bbox(self):
        """Return buffered bounding box."""
        return self._tile.bbox(pixelbuffer=self.pixelbuffer)

    def get_children(self):
        """
        Get tile children (intersecting tiles in next zoom level).

        Returns
        -------
        children : list
            a list of ``BufferedTiles``
        """
        return [BufferedTile(t, self.pixelbuffer) for t in self._tile.get_children()]

    def get_parent(self):
        """
        Get tile parent (intersecting tile in previous zoom level).

        Returns
        -------
        parent : ``BufferedTile``
        """
        return BufferedTile(self._tile.get_parent(), self.pixelbuffer)

    def get_neighbors(self, connectedness=8):
        """
        Return tile neighbors.

        Tile neighbors are unique, i.e. in some edge cases, where both the left
        and right neighbor wrapped around the antimeridian is the same. Also,
        neighbors ouside the northern and southern TilePyramid boundaries are
        excluded, because they are invalid.

        # -------------
        # | 8 | 1 | 5 |
        # -------------
        # | 4 | x | 2 |
        # -------------
        # | 7 | 3 | 6 |
        # -------------

        Parameters
        ----------
        connectedness : int
            [4 or 8] return four direct neighbors or all eight.

        Returns
        -------
        list of BufferedTiles
        """
        return [
            BufferedTile(t, self.pixelbuffer)
            for t in self._tile.get_neighbors(connectedness=connectedness)
        ]

    def is_on_edge(self):
        """Determine whether tile touches or goes over pyramid edge."""
        return (
            self.left <= self.tile_pyramid.left
            or self.bottom <= self.tile_pyramid.bottom  # touches_left
            or self.right >= self.tile_pyramid.right  # touches_bottom
            or self.top >= self.tile_pyramid.top  # touches_right  # touches_top
        )

    def __eq__(self, other):
        return (
            isinstance(other, self.__class__)
            and self.pixelbuffer == other.pixelbuffer
            and self.buffered_tp == other.buffered_tp
            and self.id == other.id
        )

    def __ne__(self, other):
        return not self.__eq__(other)

    def __repr__(self):
        return "BufferedTile(%s, tile_pyramid=%s, pixelbuffer=%s)" % (
            self.id,
            self.tp,
            self.pixelbuffer,
        )

    def __hash__(self):
        return hash(repr(self)) def deprecated_kwargs(func):
    """Decorator for open() functions warning of keyword argument usage."""

    def func_wrapper(*args, **kwargs):
        if kwargs:
            warnings.warn(
                "Using kwargs such in open() is deprecated and will have no effect."
            )
        if "resampling" in kwargs:
            raise DeprecationWarning(
                "'resampling' argument has no effect here and must be provided in read() "
                "function."
            )
        return func(*args, **kwargs)

    return func_wrapper
from contextlib import ExitStack
from affine import Affine
from numpy import ma
from rasterio.enums import Resampling
from rasterio.profiles import Profile
from rasterio.rio.overview import get_maximum_overview_level
from rasterio.windows import from_bounds
from shapely.geometry import box
from tilematrix import Bounds
from mapchete.config import validate_values, snap_bounds, _OUTPUT_PARAMETERS
from mapchete.errors import MapcheteConfigError
from mapchete.formats import base
from mapchete.io import makedirs, path_exists, path_is_remote
from mapchete.io.raster import (
    write_raster_window,
    prepare_array,
    memory_file,
    read_raster_no_crs,
    extract_from_array,
    rasterio_write,
)
from mapchete.tile import BufferedTile
from mapchete.validate import deprecated_kwargs
import logging
import math
import os
import warnings
import numpy as np

    def read(self, output_tile, **kwargs):
        """
        Read existing process output.

        Parameters
        ----------
        output_tile : ``BufferedTile``
            must be member of output ``TilePyramid``

        Returns
        -------
        NumPy array
        """
        logger.debug("read %s", self.get_path(output_tile))
        try:
            return read_raster_no_crs(self.get_path(output_tile))
        except FileNotFoundError:
            return self.empty(output_tile)

    def empty(self, process_tile):
        """
        Return empty data.

        Parameters
        ----------
        process_tile : ``BufferedTile``
            must be member of process ``TilePyramid``

        Returns
        -------
        empty data : array
            empty array with data type provided in output profile
        """
        profile = self.profile(process_tile)
        return ma.masked_array(
            data=np.full(
                (profile["count"],) + process_tile.shape,
                profile["nodata"],
                dtype=profile["dtype"],
            ),
            mask=True,
            fill_value=profile["nodata"],
        )

    def profile(self, tile=None):
        """
        Create a metadata dictionary for rasterio.

        Parameters
        ----------
        tile : ``BufferedTile``

        Returns
        -------
        metadata : dictionary
            output profile dictionary used for rasterio.
        """
        dst_metadata = dict(
            GTIFF_DEFAULT_PROFILE,
            count=self.output_params["bands"],
            **{
                k: v
                for k, v in self.output_params.items()
                if k not in _OUTPUT_PARAMETERS
            },
        )
        dst_metadata.pop("transform", None)
        if tile is not None:
            dst_metadata.update(
                crs=tile.crs, width=tile.width, height=tile.height, affine=tile.affine
            )
        else:
            for k in ["crs", "width", "height", "affine"]:
                dst_metadata.pop(k, None)
        try:
            if "compression" in self.output_params:
                warnings.warn(
                    DeprecationWarning("use 'compress' instead of 'compression'")
                )
                dst_metadata.update(compress=self.output_params["compression"])
            else:
                dst_metadata.update(compress=self.output_params["compress"])
            dst_metadata.update(predictor=self.output_params["predictor"])
        except KeyError:
            pass
        return dst_metadata


class GTiffTileDirectoryOutputWriter(
    GTiffTileDirectoryOutputReader, base.TileDirectoryOutputWriter
):
    def write(self, process_tile, data):
        """
        Write data from process tiles into GeoTIFF file(s).

        Parameters
        ----------
        process_tile : ``BufferedTile``
            must be member of process ``TilePyramid``
        data : ``np.ndarray``
        """
        if isinstance(data, tuple) and len(data) == 2 and isinstance(data[1], dict):
            data, tags = data
        else:
            tags = {}
        data = prepare_array(
            data,
            masked=True,
            nodata=self.output_params["nodata"],
            dtype=self.profile(process_tile)["dtype"],
        )

        if data.mask.all():
            logger.debug("data empty, nothing to write")
        else:
            # Convert from process_tile to output_tiles and write
            for tile in self.pyramid.intersecting(process_tile):
                out_path = self.get_path(tile)
                self.prepare_path(tile)

                out_tile = BufferedTile(tile, self.pixelbuffer)