Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add rio-tiler-pds Custom tiler example #105

Closed
vincentsarago opened this issue Sep 9, 2020 · 1 comment
Closed

add rio-tiler-pds Custom tiler example #105

vincentsarago opened this issue Sep 9, 2020 · 1 comment
Labels
documentation Improvements or additions to documentation

Comments

@vincentsarago
Copy link
Member

"""PDS Tiler."""

from dataclasses import dataclass
from typing import List, Type, Union

from titiler.endpoints.factory import TMSTilerFactory
from titiler.dependencies import PathParams, BandsParams
from titiler.models.dataset import Info, Metadata

from rio_tiler_pds.landsat.aws import L8Reader
from rio_tiler_pds.sentinel.aws.sentinel2 import S2L1CReader, S2COGReader

from fastapi import FastAPI, Depends


@dataclass
class CustomPathParams(PathParams):
    """Create dataset path from args"""

    def __post_init__(self,):
        """Define dataset URL."""

        if self.url.startswith("landsat+"):
            self.reader = L8Reader
            self.url = self.url.replace("landsat+", "").strip()
        elif self.url.startswith("sentinel2l1c+"):
            self.reader = S2L1CReader
            self.url = self.url.replace("sentinel2l1c+", "").strip()
        elif self.url.startswith("sentinel2cog+"):
            self.reader = S2COGReader
            self.url = self.url.replace("sentinel2cog+", "").strip()


@dataclass
class CustomTiler(TMSTilerFactory):
    """Custom Tiler Class for STAC."""

    path_dependency: Type[CustomPathParams] = CustomPathParams
    additional_dependency: Type[BandsParams] = BandsParams

    # Overwrite _info method to return the list of assets when no assets is passed.
    def _info(self):
        """Register /info endpoint to router."""

        @self.router.get(
            "/info",
            response_model=Union[List[str], Info],
            response_model_exclude={"minzoom", "maxzoom", "center"},
            response_model_exclude_none=True,
            responses={200: {"description": "Return dataset's basic info."}},
        )
        def info(
            src_path=Depends(self.path_dependency),
            options=Depends(self.additional_dependency),
        ):
            """Return basic info."""
            reader = src_path.reader or self.reader
            with reader(src_path.url, **self.reader_options) as src_dst:
                if not options.kwargs.get("bands"):
                    return src_dst.bands
                info = src_dst.info(**options.kwargs)
            return info

    # Overwrite _metadata method because the STACTiler output model is different
    # cogMetadata -> Dict[str, cogMetadata]
    def _metadata(self):
        """Register /metadata endpoint to router."""

        @self.router.get(
            "/metadata",
            response_model=Metadata,
            response_model_exclude={"minzoom", "maxzoom", "center"},
            response_model_exclude_none=True,
            responses={200: {"description": "Return dataset's metadata."}},
        )
        def metadata(
            src_path=Depends(self.path_dependency),
            params=Depends(self.metadata_dependency),
            options=Depends(self.additional_dependency),
        ):
            """Return metadata."""
            reader = src_path.reader or self.reader
            with reader(src_path.url, **self.reader_options) as src_dst:
                kwargs = options.kwargs.copy()
                if params.nodata is not None:
                    kwargs["nodata"] = params.nodata
                info = src_dst.metadata(
                    params.pmin,
                    params.pmax,
                    indexes=params.indexes,
                    max_size=params.max_size,
                    hist_options=params.hist_options,
                    bounds=params.bounds,
                    resampling_method=params.resampling_method.name,
                    **kwargs,
                )
            return info


app = FastAPI()
tiler = CustomTiler()
app.include_router(tiler.router)
@vincentsarago vincentsarago added the documentation Improvements or additions to documentation label Sep 9, 2020
@vincentsarago
Copy link
Member Author

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
documentation Improvements or additions to documentation
Projects
None yet
Development

No branches or pull requests

1 participant