123 lines
4.3 KiB
Python
123 lines
4.3 KiB
Python
import base64
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
import zipfile
|
|
from pathlib import Path
|
|
from typing import Iterable, Optional
|
|
|
|
from fastapi.responses import FileResponse, Response
|
|
|
|
|
|
DEFAULT_PLACEHOLDER_PNG = base64.b64decode(
|
|
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMB/6Xq3lYAAAAASUVORK5CYII="
|
|
)
|
|
|
|
|
|
class TileCutError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class TileService:
|
|
def __init__(self, tiles_dir: Path, placeholder: Path, upload_dir: Path):
|
|
self.tiles_dir = Path(tiles_dir)
|
|
self.placeholder = Path(placeholder)
|
|
self.upload_dir = Path(upload_dir)
|
|
|
|
def ensure_tiles_dir(self) -> None:
|
|
self.tiles_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def ensure_upload_dir(self) -> None:
|
|
self.upload_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _find_gdal2tiles(self) -> str:
|
|
for name in ("gdal2tiles.py", "gdal2tiles"):
|
|
found = shutil.which(name)
|
|
if found:
|
|
return found
|
|
raise TileCutError("gdal2tiles not found in PATH. Install GDAL (e.g., apt install gdal-bin).")
|
|
|
|
def slice_to_tiles(
|
|
self,
|
|
source: Path,
|
|
destination: Path,
|
|
min_zoom: int,
|
|
max_zoom: int,
|
|
resampling: str = "bilinear",
|
|
tile_format: str = "png",
|
|
web_mercator: bool = True,
|
|
extra_args: Optional[Iterable[str]] = None,
|
|
scheme: str = "xyz",
|
|
) -> Path:
|
|
if not source.exists():
|
|
raise TileCutError(f"source does not exist: {source}")
|
|
if min_zoom < 0 or max_zoom < min_zoom:
|
|
raise TileCutError(f"invalid zoom range: {min_zoom}-{max_zoom}")
|
|
if tile_format.lower() != "png":
|
|
raise TileCutError("current gdal2tiles does not support custom format here; use png")
|
|
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
|
|
gdal2tiles = self._find_gdal2tiles()
|
|
cmd = [
|
|
gdal2tiles,
|
|
"-z",
|
|
f"{min_zoom}-{max_zoom}",
|
|
"-r",
|
|
resampling,
|
|
]
|
|
if web_mercator:
|
|
cmd += ["-p", "mercator"]
|
|
if scheme.lower() == "xyz":
|
|
cmd += ["--xyz"]
|
|
if extra_args:
|
|
cmd += list(extra_args)
|
|
cmd += [str(source), str(destination)]
|
|
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise TileCutError(f"gdal2tiles failed: {result.stderr or result.stdout}")
|
|
return destination
|
|
|
|
def get_tile_response(self, zoom: int, x: int, y: int):
|
|
tile_path = self.tiles_dir / str(zoom) / str(x) / f"{y}.png"
|
|
if tile_path.is_file():
|
|
return FileResponse(tile_path, media_type="image/png")
|
|
if self.placeholder.is_file():
|
|
return FileResponse(self.placeholder, media_type="image/png")
|
|
return Response(content=DEFAULT_PLACEHOLDER_PNG, media_type="image/png")
|
|
|
|
def get_tile_response_tms(self, zoom: int, x: int, y: int):
|
|
y_converted = (2**zoom - 1) - y
|
|
return self.get_tile_response(zoom, x, y_converted)
|
|
|
|
def unpack_if_archive(self, source: Path) -> Path:
|
|
lower = source.suffix.lower()
|
|
if lower == ".zip":
|
|
dest = self.upload_dir / source.stem
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
with zipfile.ZipFile(source, "r") as zf:
|
|
zf.extractall(dest)
|
|
return dest
|
|
if lower in {".tar", ".tgz", ".gz", ".bz2", ".xz"} or source.name.endswith((".tar.gz", ".tar.bz2", ".tar.xz")):
|
|
dest = self.upload_dir / source.stem
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
with tarfile.open(source, "r:*") as tf:
|
|
tf.extractall(dest)
|
|
return dest
|
|
return source
|
|
|
|
def copy_tile_tree(self, source_dir: Path, destination: Path) -> None:
|
|
if not source_dir.is_dir():
|
|
raise TileCutError(
|
|
f"tiles directory not found: {source_dir}. Expected folder with {Path('z/x/y.png')} structure or a zip/tar archive."
|
|
)
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
for item in source_dir.iterdir():
|
|
target = destination / item.name
|
|
if item.is_dir():
|
|
shutil.copytree(item, target, dirs_exist_ok=True)
|
|
else:
|
|
shutil.copy2(item, target)
|
|
|
|
|