added bootstrap

scripts/bootstrap_dem.py

@@ -1,29 +1,90 @@
-"""Bootstrap Copernicus DEM tiles for an AOI.
-
-The production implementation should download Copernicus DEM GLO-30 tiles and
-write Cloud-Optimized GeoTIFFs into `data/dem`.
-"""
+"""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, help="AOI bbox: minlon,minlat,maxlon,maxlat")
-    parser.add_argument("--output-dir", default="/data/dem", type=Path)
+    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)
-    raise NotImplementedError(
-        "Copernicus DEM download/mosaic/COG conversion is implemented in a later stage. "
-        f"Requested bbox={args.bbox}, output_dir={args.output_dir}"
-    )
+    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__":