Skip to content

Update to titiler #5

Open
Open
@kylebarron

Description

@kylebarron
from dataclasses import dataclass, field
from typing import List, Type, Union, Optional
from titiler.endpoints.factory import TMSTilerFactory
from titiler.dependencies import BandsParams
from titiler.models.dataset import Info, Metadata
from rio_tiler.io import BaseReader
from rio_tiler_pds.landsat.aws import L8Reader
from fastapi import FastAPI, Depends, Query
@dataclass
class CustomPathParams:
    """Create dataset path from args"""
    sceneid: str = Query(..., description="Landsat 8 Sceneid.")
    reader: Optional[Type[BaseReader]] = field(init=False, default=None)
    def __post_init__(self,):
        """Define dataset URL."""
        self.url = self.sceneid
@dataclass
class CustomTiler(TMSTilerFactory):
    """Custom Tiler Class for STAC."""
    reader: Type[L8Reader] = field(default=L8Reader)
    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
config = {
    "GDAL_DISABLE_READDIR_ON_OPEN": "FALSE",
    "CPL_VSIL_CURL_ALLOWED_EXTENSIONS": ".TIF,.ovr",
}
app = FastAPI()
tiler = CustomTiler(env=config)
app.include_router(tiler.router)

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions