94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
"""Bootstrap landcover and canopy rasters 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
|
|
|
|
WORLDCOVER_BASE_URL = "https://esa-worldcover.s3.eu-central-1.amazonaws.com/v200/2021/map"
|
|
WORLDCOVER_GRID_URL = (
|
|
"https://esa-worldcover.s3.eu-central-1.amazonaws.com/v200/2021/esa_worldcover_2021_grid.geojson"
|
|
)
|
|
|
|
|
|
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 _tile_id(lat: int, lon: int) -> str:
|
|
lat_prefix = "N" if lat >= 0 else "S"
|
|
lon_prefix = "E" if lon >= 0 else "W"
|
|
return f"{lat_prefix}{abs(lat):02d}{lon_prefix}{abs(lon):03d}"
|
|
|
|
|
|
def tiles_for_bbox(bbox: tuple[float, float, float, float]) -> list[str]:
|
|
minlon, minlat, maxlon, maxlat = bbox
|
|
lats = range(math.floor(minlat / 3) * 3, math.ceil(maxlat), 3)
|
|
lons = range(math.floor(minlon / 3) * 3, math.ceil(maxlon), 3)
|
|
return [_tile_id(lat, lon) for lat in lats for lon in lons]
|
|
|
|
|
|
def tile_url(tile: str) -> str:
|
|
return f"{WORLDCOVER_BASE_URL}/ESA_WorldCover_10m_2021_v200_{tile}_Map.tif"
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Bootstrap WorldCover/CORINE/canopy data")
|
|
parser.add_argument("--bbox", required=True, type=parse_bbox, help="minlon,minlat,maxlon,maxlat")
|
|
parser.add_argument("--landcover-dir", default=Path("data/landcover"), type=Path)
|
|
parser.add_argument("--canopy-dir", default=Path("data/canopy"), type=Path)
|
|
parser.add_argument("--overwrite", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
args.landcover_dir.mkdir(parents=True, exist_ok=True)
|
|
args.canopy_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
downloaded = 0
|
|
skipped = 0
|
|
failed: list[str] = []
|
|
|
|
for tile in tiles_for_bbox(args.bbox):
|
|
destination = args.landcover_dir / f"ESA_WorldCover_10m_2021_v200_{tile}_Map.tif"
|
|
if destination.exists() and not args.overwrite:
|
|
skipped += 1
|
|
print(f"skip existing {destination}")
|
|
continue
|
|
|
|
url = tile_url(tile)
|
|
print(f"download {url}")
|
|
try:
|
|
urlretrieve(url, destination)
|
|
except (HTTPError, URLError) as exc:
|
|
failed.append(f"{tile}: {exc}")
|
|
if destination.exists():
|
|
destination.unlink()
|
|
continue
|
|
downloaded += 1
|
|
|
|
print(
|
|
"Canopy bootstrap is not bundled with WorldCover tiles. "
|
|
"Place canopy height GeoTIFF/COG files manually under "
|
|
f"{args.canopy_dir} if needed."
|
|
)
|
|
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()
|