"""Download Copernicus DEM GLO-30 COG tiles for an AOI.""" from __future__ import annotations import argparse import math from pathlib import Path from urllib.error import HTTPError, URLError from urllib.request import urlretrieve COPERNICUS_DEM_30M_BASE_URL = "https://copernicus-dem-30m.s3.amazonaws.com" def parse_bbox(raw: str) -> tuple[float, float, float, float]: parts = [float(part.strip()) for part in raw.split(",")] if len(parts) != 4: raise argparse.ArgumentTypeError("bbox must be minlon,minlat,maxlon,maxlat") minlon, minlat, maxlon, maxlat = parts if minlon >= maxlon or minlat >= maxlat: raise argparse.ArgumentTypeError("bbox min values must be lower than max values") return minlon, minlat, maxlon, maxlat def _hemisphere(value: int, positive: str, negative: str) -> str: return positive if value >= 0 else negative def tile_name(lat: int, lon: int) -> str: lat_prefix = _hemisphere(lat, "N", "S") lon_prefix = _hemisphere(lon, "E", "W") return ( f"Copernicus_DSM_COG_10_{lat_prefix}{abs(lat):02d}_00_" f"{lon_prefix}{abs(lon):03d}_00_DEM" ) def tile_url(lat: int, lon: int) -> str: name = tile_name(lat, lon) return f"{COPERNICUS_DEM_30M_BASE_URL}/{name}/{name}.tif" def tiles_for_bbox(bbox: tuple[float, float, float, float]) -> list[tuple[int, int]]: minlon, minlat, maxlon, maxlat = bbox lats = range(math.floor(minlat), math.ceil(maxlat)) lons = range(math.floor(minlon), math.ceil(maxlon)) return [(lat, lon) for lat in lats for lon in lons] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Bootstrap Copernicus DEM for an AOI") parser.add_argument("--bbox", required=True, type=parse_bbox, help="minlon,minlat,maxlon,maxlat") parser.add_argument("--output-dir", default=Path("data/dem"), type=Path) parser.add_argument("--overwrite", action="store_true", help="Re-download existing tiles") return parser.parse_args() def main() -> None: args = parse_args() args.output_dir.mkdir(parents=True, exist_ok=True) downloaded = 0 skipped = 0 failed: list[str] = [] for lat, lon in tiles_for_bbox(args.bbox): name = tile_name(lat, lon) destination = args.output_dir / f"{name}.tif" if destination.exists() and not args.overwrite: skipped += 1 print(f"skip existing {destination}") continue url = tile_url(lat, lon) print(f"download {url}") try: urlretrieve(url, destination) except (HTTPError, URLError) as exc: failed.append(f"{name}: {exc}") if destination.exists(): destination.unlink() continue downloaded += 1 print(f"downloaded={downloaded} skipped={skipped} failed={len(failed)}") if failed: for item in failed: print(f"failed {item}") raise SystemExit(1) if __name__ == "__main__": main()