split tile() also for MosaicFactory

[vincentsarago]

ref #141
🤦

titiler/titiler/endpoints/factory.py

Lines 784 to 1049 in 87ddb17

	############################################################################
	# /tiles
	############################################################################
	def tile(self): # noqa: C901
	"""Register /tiles endpoints."""
	tile_endpoint_params = img_endpoint_params.copy()
	@self.router.get(r"/tiles/{z}/{x}/{y}", **tile_endpoint_params)
	@self.router.get(r"/tiles/{z}/{x}/{y}.{format}", **tile_endpoint_params)
	@self.router.get(r"/tiles/{z}/{x}/{y}@{scale}x", **tile_endpoint_params)
	@self.router.get(
	r"/tiles/{z}/{x}/{y}@{scale}x.{format}", **tile_endpoint_params
	)
	@self.router.get(
	r"/tiles/{TileMatrixSetId}/{z}/{x}/{y}", **tile_endpoint_params
	)
	@self.router.get(
	r"/tiles/{TileMatrixSetId}/{z}/{x}/{y}.{format}", **tile_endpoint_params
	)
	@self.router.get(
	r"/tiles/{TileMatrixSetId}/{z}/{x}/{y}@{scale}x", **tile_endpoint_params
	)
	@self.router.get(
	r"/tiles/{TileMatrixSetId}/{z}/{x}/{y}@{scale}x.{format}",
	**tile_endpoint_params,
	)
	def tile(
	z: int = Path(..., ge=0, le=30, description="Mercator tiles's zoom level"),
	x: int = Path(..., description="Mercator tiles's column"),
	y: int = Path(..., description="Mercator tiles's row"),
	tms: TileMatrixSet = Depends(self.tms_dependency),
	scale: int = Query(
	1, gt=0, lt=4, description="Tile size scale. 1=256x256, 2=512x512..."
	),
	format: ImageType = Query(
	None, description="Output image type. Default is auto."
	),
	src_path=Depends(self.path_dependency),
	layer_params=Depends(self.layer_dependency),
	dataset_params=Depends(self.dataset_dependency),
	render_params=Depends(self.render_dependency),
	pixel_selection: PixelSelectionMethod = Query(
	PixelSelectionMethod.first, description="Pixel selection method."
	),
	kwargs: Dict = Depends(self.additional_dependency),
	):
	"""Create map tile from a COG."""
	timings = []
	headers: Dict[str, str] = {}
	tilesize = scale * 256
	threads = int(os.getenv("MOSAIC_CONCURRENCY", MAX_THREADS))
	with utils.Timer() as t:
	with self.reader(
	src_path.url,
	reader=self.dataset_reader,
	reader_options=self.reader_options,
	) as src_dst:
	mosaic_read = t.from_start
	timings.append(("mosaicread", round(mosaic_read * 1000, 2)))
	data, _ = src_dst.tile(
	x,
	y,
	z,
	pixel_selection=pixel_selection.method(),
	threads=threads,
	tilesize=tilesize,
	**layer_params.kwargs,
	**dataset_params.kwargs,
	**kwargs,
	)
	timings.append(("dataread", round((t.elapsed - mosaic_read) * 1000, 2)))
	if not data.assets:
	raise TileNotFoundError(f"Tile {z}/{x}/{y} was not found")
	if not format:
	format = ImageType.jpg if data.mask.all() else ImageType.png
	with utils.Timer() as t:
	image = data.post_process(
	in_range=render_params.rescale_range,
	color_formula=render_params.color_formula,
	)
	timings.append(("postprocess", round(t.elapsed * 1000, 2)))
	with utils.Timer() as t:
	content = image.render(
	add_mask=render_params.return_mask,
	img_format=format.driver,
	colormap=render_params.colormap,
	**format.profile,
	)
	timings.append(("format", round(t.elapsed * 1000, 2)))
	if timings:
	headers["Server-Timing"] = ", ".join(
	[f"{name};dur={time}" for (name, time) in timings]
	)
	headers["X-Assets"] = ",".join(data.assets)
	return Response(
	content, media_type=ImageMimeTypes[format.value].value, headers=headers,
	)
	@self.router.get(
	"/tilejson.json",
	response_model=TileJSON,
	responses={200: {"description": "Return a tilejson"}},
	response_model_exclude_none=True,
	)
	@self.router.get(
	"/{TileMatrixSetId}/tilejson.json",
	response_model=TileJSON,
	responses={200: {"description": "Return a tilejson"}},
	response_model_exclude_none=True,
	)
	def tilejson(
	request: Request,
	tms: TileMatrixSet = Depends(self.tms_dependency),
	src_path=Depends(self.path_dependency),
	tile_format: Optional[ImageType] = Query(
	None, description="Output image type. Default is auto."
	),
	tile_scale: int = Query(
	1, gt=0, lt=4, description="Tile size scale. 1=256x256, 2=512x512..."
	),
	minzoom: Optional[int] = Query(
	None, description="Overwrite default minzoom."
	),
	maxzoom: Optional[int] = Query(
	None, description="Overwrite default maxzoom."
	),
	layer_params=Depends(self.layer_dependency), # noqa
	dataset_params=Depends(self.dataset_dependency), # noqa
	render_params=Depends(self.render_dependency), # noqa
	pixel_selection: PixelSelectionMethod = Query(
	PixelSelectionMethod.first, description="Pixel selection method."
	), # noqa
	kwargs: Dict = Depends(self.additional_dependency), # noqa
	):
	"""Return TileJSON document for a COG."""
	route_params = {
	"z": "{z}",
	"x": "{x}",
	"y": "{y}",
	"scale": tile_scale,
	"TileMatrixSetId": tms.identifier,
	}
	if tile_format:
	route_params["format"] = tile_format.value
	tiles_url = self.url_for(request, "tile", **route_params)
	q = dict(request.query_params)
	q.pop("TileMatrixSetId", None)
	q.pop("tile_format", None)
	q.pop("tile_scale", None)
	q.pop("minzoom", None)
	q.pop("maxzoom", None)
	qs = urlencode(list(q.items()))
	tiles_url += f"?{qs}"
	with self.reader(src_path.url) as src_dst:
	center = list(src_dst.center)
	if minzoom:
	center[-1] = minzoom
	tjson = {
	"bounds": src_dst.bounds,
	"center": tuple(center),
	"minzoom": minzoom if minzoom is not None else src_dst.minzoom,
	"maxzoom": maxzoom if maxzoom is not None else src_dst.maxzoom,
	"name": os.path.basename(src_path.url),
	"tiles": [tiles_url],
	}
	return tjson
	@self.router.get("/WMTSCapabilities.xml", response_class=XMLResponse)
	@self.router.get(
	"/{TileMatrixSetId}/WMTSCapabilities.xml", response_class=XMLResponse
	)
	def wmts(
	request: Request,
	tms: TileMatrixSet = Depends(self.tms_dependency),
	src_path=Depends(self.path_dependency),
	tile_format: ImageType = Query(
	ImageType.png, description="Output image type. Default is png."
	),
	tile_scale: int = Query(
	1, gt=0, lt=4, description="Tile size scale. 1=256x256, 2=512x512..."
	),
	minzoom: Optional[int] = Query(
	None, description="Overwrite default minzoom."
	),
	maxzoom: Optional[int] = Query(
	None, description="Overwrite default maxzoom."
	),
	layer_params=Depends(self.layer_dependency), # noqa
	dataset_params=Depends(self.dataset_dependency), # noqa
	render_params=Depends(self.render_dependency), # noqa
	pixel_selection: PixelSelectionMethod = Query(
	PixelSelectionMethod.first, description="Pixel selection method."
	), # noqa
	kwargs: Dict = Depends(self.additional_dependency), # noqa
	):
	"""OGC WMTS endpoint."""
	route_params = {
	"z": "{TileMatrix}",
	"x": "{TileCol}",
	"y": "{TileRow}",
	"scale": tile_scale,
	"format": tile_format.value,
	"TileMatrixSetId": tms.identifier,
	}
	tiles_url = self.url_for(request, "tile", **route_params)
	q = dict(request.query_params)
	q.pop("TileMatrixSetId", None)
	q.pop("tile_format", None)
	q.pop("tile_scale", None)
	q.pop("minzoom", None)
	q.pop("maxzoom", None)
	q.pop("SERVICE", None)
	q.pop("REQUEST", None)
	qs = urlencode(list(q.items()))
	tiles_url += f"?{qs}"
	with self.reader(src_path.url) as src_dst:
	bounds = src_dst.bounds
	minzoom = minzoom if minzoom is not None else src_dst.minzoom
	maxzoom = maxzoom if maxzoom is not None else src_dst.maxzoom
	media_type = ImageMimeTypes[tile_format.value].value
	tileMatrix = []
	for zoom in range(minzoom, maxzoom + 1):
	matrix = tms.matrix(zoom)
	tm = f"""
	<TileMatrix>
	<ows:Identifier>{matrix.identifier}</ows:Identifier>
	<ScaleDenominator>{matrix.scaleDenominator}</ScaleDenominator>
	<TopLeftCorner>{matrix.topLeftCorner[0]} {matrix.topLeftCorner[1]}</TopLeftCorner>
	<TileWidth>{matrix.tileWidth}</TileWidth>
	<TileHeight>{matrix.tileHeight}</TileHeight>
	<MatrixWidth>{matrix.matrixWidth}</MatrixWidth>
	<MatrixHeight>{matrix.matrixHeight}</MatrixHeight>
	</TileMatrix>"""
	tileMatrix.append(tm)
	return templates.TemplateResponse(
	"wmts.xml",
	{
	"request": request,
	"tiles_endpoint": tiles_url,
	"bounds": bounds,
	"tileMatrix": tileMatrix,
	"tms": tms,
	"title": "Cloud Optimized GeoTIFF",
	"layer_name": "cogeo",
	"media_type": media_type,
	},
	media_type=MimeTypes.xml.value,
	)