"""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()