scripts/fetch_landcover.py

"""
Samples ESA WorldCover landcover classification and NDVI at station
points, from the public AWS S3 Cloud-Optimized GeoTIFF tiles -- not the
Terrascope WMS.

WHY NOT WMS: a source dated July 2026 reports Terrascope's WMS actively
resets connections from non-browser HTTP clients (TLS fingerprinting,
confirmed across curl/wget/Node fetch with multiple User-Agents) --
this is not a coding problem to work around, it's the service
declining non-browser clients. Separately, that WMS was scheduled for
full phase-out in January 2026, already past. Using ESA's own
recommended alternative instead: direct S3 access, "avoids unzipping
steps" per ESA's own documentation, genuinely public, no auth/signing.

TILE: ESA WorldCover ships as 3x3 degree COG tiles named by their
southwest corner, e.g. "N48E000". All of this project's real station
coordinates (lon 0.43-1.54E, lat 48.39-49.34N) fall inside exactly one
tile: N48E000 -- computed directly (floor each coordinate to the
nearest 3-degree grid line), not guessed.

CAVEAT: the exact object key/path within the S3 bucket is reasoned from
documented ESA WorldCover file-naming conventions
(ESA_WorldCover_10m_2021_v200_{tile}_Map.tif for classification), NOT
independently verified against a live S3 listing -- no network access
in the environment this was written in to confirm it directly. Run
--check first.

NEEDS: rasterio (for reading COG tiles and sampling point values).

Usage:
    python -m scripts.fetch_worldcover_landcover --check
    python -m scripts.fetch_worldcover_landcover --stations datasets/station_elevations.csv
"""
import argparse
from pathlib import Path

import pandas as pd

BUCKET_BASE = "https://esa-worldcover.s3.eu-central-1.amazonaws.com"
TILE = "N48E000"
CLASSIFICATION_URL = f"{BUCKET_BASE}/v200/2021/map/ESA_WorldCover_10m_2021_v200_{TILE}_Map.tif"

# 11-class legend, from ESA's own product documentation -- needed to make
# the raw integer codes in the classification raster human-readable.
LANDCOVER_CLASSES = {
    10: "Tree cover", 20: "Shrubland", 30: "Grassland", 40: "Cropland",
    50: "Built-up", 60: "Bare/sparse vegetation", 70: "Snow and ice",
    80: "Permanent water bodies", 90: "Herbaceous wetland",
    95: "Mangrove", 100: "Moss and lichen",
}


def compute_tile_id(lat: float, lon: float) -> str:
    """3-degree grid tile ID containing (lat, lon), named by SW corner."""
    import math
    tile_lat = math.floor(lat / 3) * 3
    tile_lon = math.floor(lon / 3) * 3
    lat_str = f"N{tile_lat:02d}" if tile_lat >= 0 else f"S{-tile_lat:02d}"
    lon_str = f"E{tile_lon:03d}" if tile_lon >= 0 else f"W{-tile_lon:03d}"
    return f"{lat_str}{lon_str}"


def check_access() -> bool:
    try:
        import rasterio
    except ImportError:
        print("rasterio is not installed -- this is required. "
              "pip install rasterio --break-system-packages")
        return False

    print(f"Checking {CLASSIFICATION_URL} ...")
    try:
        with rasterio.open(CLASSIFICATION_URL) as src:
            print(f"  OK: opened tile, shape={src.shape}, crs={src.crs}, dtype={src.dtypes[0]}")
            # sample one known real point (a real station's coordinates)
            test_lat, test_lon = 49.03, 0.79
            vals = list(src.sample([(test_lon, test_lat)]))
            print(f"  Sample at ({test_lat}, {test_lon}): {vals[0][0]} "
                  f"({LANDCOVER_CLASSES.get(int(vals[0][0]), 'unknown class')})")
        return True
    except Exception as e:
        print(f"  FAILED: {e}")
        print(f"  The object key may not match the real S3 layout. Check the ESA WorldCover "
              f"AWS registry page directly: https://registry.opendata.aws/esa-worldcover/")
        return False


def fetch_for_stations(stations_path: Path, output_path: Path) -> None:
    import rasterio

    stations = pd.read_csv(stations_path)
    lat_col = "latitude" if "latitude" in stations.columns else "lat"
    lon_col = "longitude" if "longitude" in stations.columns else "lon"

    # Confirm every station actually falls in the hardcoded tile before
    # trusting a single-tile fetch -- if a future node set (e.g. the
    # full ~4,500-node reach graph, not just the 27 gauges) extends
    # beyond N48E000, this needs multiple tiles, not silently wrong data
    # from the one tile that happens to be loaded.
    tile_ids = stations.apply(lambda r: compute_tile_id(r[lat_col], r[lon_col]), axis=1)
    unexpected = tile_ids[tile_ids != TILE].unique()
    if len(unexpected) > 0:
        print(f"WARNING: some stations fall outside tile {TILE}: {list(unexpected)}. "
              f"This script only fetches {TILE} -- results for those stations will be wrong "
              f"or missing. Extend BUCKET fetching to cover {list(unexpected)} too.")

    with rasterio.open(CLASSIFICATION_URL) as src:
        coords = list(zip(stations[lon_col], stations[lat_col]))
        values = [v[0] for v in src.sample(coords)]

    out = stations[["station_code"]].copy()
    out["landcover_class_code"] = values
    out["landcover_class_name"] = [LANDCOVER_CLASSES.get(int(v), "unknown") for v in values]
    out.to_csv(output_path, index=False)
    print(f"Saved {len(out)} stations' landcover class to {output_path}")
    print(out["landcover_class_name"].value_counts())


def main() -> None:
    parser = argparse.ArgumentParser(description="Sample ESA WorldCover landcover at station points")
    parser.add_argument("--check", action="store_true")
    parser.add_argument("--stations", type=Path, default=Path("datasets/station_elevations.csv"))
    parser.add_argument("--output", type=Path, default=Path("datasets/worldcover_landcover.csv"))
    args = parser.parse_args()

    if args.check:
        check_access()
        return

    fetch_for_stations(args.stations, args.output)


if __name__ == "__main__":
    main()