# Path: mapchete/io/_path.py
# def fs_from_path(path, timeout=5, session=None, username=None, password=None, **kwargs):
#     """Guess fsspec FileSystem from path and initialize using the desired options."""
#     if path.startswith("s3://"):
#         return fsspec.filesystem(
#             "s3",
#             requester_pays=os.environ.get("AWS_REQUEST_PAYER") == "requester",
#             config_kwargs=dict(connect_timeout=timeout, read_timeout=timeout),
#             session=session,
#             client_kwargs=kwargs,
#         )
#     elif path.startswith(("http://", "https://")):
#         if username:  # pragma: no cover
#             from aiohttp import BasicAuth
# 
#             auth = BasicAuth(username, password)
#         else:
#             auth = None
#         return fsspec.filesystem("https", auth=auth, asynchronous=False, **kwargs)
#     else:
#         return fsspec.filesystem("file", asynchronous=False, **kwargs)
# 
# Path: mapchete/_tasks.py
# 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")


import mapchete
import pytest
import rasterio
from mapchete.io import fs_from_path
from mapchete._tasks import Task
    with mp.open("clip") as clip:
        assert clip.path.endswith("cache/aoi_br.fgb")
        assert clip.read()
    with mp.open("inp") as raster:
        assert raster.path.endswith("cache/cleantopo_br.tif")
        assert not raster.read().mask.all()


def test_preprocess_cache_raster_vector_tasks(preprocess_cache_raster_vector):
    with preprocess_cache_raster_vector.mp() as mp:
        for i in ["clip", "inp"]:
            input_data = mp.config.input_at_zoom(key=i, zoom=5)
            for task in input_data.preprocessing_tasks.values():
                assert isinstance(task, Task)
                assert task.has_geometry()


def test_preprocessing_tasks_dependencies(preprocess_cache_memory):
    with preprocess_cache_memory.mp() as mp:
        for i in ["clip", "inp"]:
            input_data = mp.config.input_at_zoom(key=i, zoom=5)
            for task in input_data.preprocessing_tasks.values():
                assert isinstance(task, Task)
                assert task.has_geometry()
        list(mp.compute())

        out_path = mp.config.output_reader.path
        total_tifs = len(
            [
                file

# Path: mapchete/io/_path.py
# def fs_from_path(path, timeout=5, session=None, username=None, password=None, **kwargs):
#     """Guess fsspec FileSystem from path and initialize using the desired options."""
#     if path.startswith("s3://"):
#         return fsspec.filesystem(
#             "s3",
#             requester_pays=os.environ.get("AWS_REQUEST_PAYER") == "requester",
#             config_kwargs=dict(connect_timeout=timeout, read_timeout=timeout),
#             session=session,
#             client_kwargs=kwargs,
#         )
#     elif path.startswith(("http://", "https://")):
#         if username:  # pragma: no cover
#             from aiohttp import BasicAuth
# 
#             auth = BasicAuth(username, password)
#         else:
#             auth = None
#         return fsspec.filesystem("https", auth=auth, asynchronous=False, **kwargs)
#     else:
#         return fsspec.filesystem("file", asynchronous=False, **kwargs)
# 
# Path: mapchete/_tasks.py
# 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")


# Path: test/test_preprocessing.py
import mapchete
import pytest
import rasterio
from mapchete.io import fs_from_path
from mapchete._tasks import Task

    with mp.open("clip") as clip:
        assert clip.path.endswith("cache/aoi_br.fgb")
        assert clip.read()
    with mp.open("inp") as raster:
        assert raster.path.endswith("cache/cleantopo_br.tif")
        assert not raster.read().mask.all()


def test_preprocess_cache_raster_vector_tasks(preprocess_cache_raster_vector):
    with preprocess_cache_raster_vector.mp() as mp:
        for i in ["clip", "inp"]:
            input_data = mp.config.input_at_zoom(key=i, zoom=5)
            for task in input_data.preprocessing_tasks.values():
                assert isinstance(task, Task)
                assert task.has_geometry()


def test_preprocessing_tasks_dependencies(preprocess_cache_memory):
    with preprocess_cache_memory.mp() as mp:
        for i in ["clip", "inp"]:
            input_data = mp.config.input_at_zoom(key=i, zoom=5)
            for task in input_data.preprocessing_tasks.values():
                assert isinstance(task, Task)
                assert task.has_geometry()
        list(mp.compute())

        out_path = mp.config.output_reader.path
        total_tifs = len(
            [
                file

                for directory in fs_from_path(out_path).walk(out_path)