Upload folder using huggingface_hub (part 2)

scripts/build_dynamic_tensors.py

@@ -0,0 +1,117 @@
+"""
+Builds genuine [n_nodes, T] dynamic tensors (discharge, groundwater,
+climate) for a basin's reach graph and feeds them directly into
+physics_losses.py's routing_consistency_loss -- the actual integration
+point dynamic_features.py exists for. Without this script, that
+function has real inputs it could consume but nothing actually
+producing them.
+
+Usage:
+    python -m scripts.build_dynamic_tensors --data-root datasets --basin risle
+"""
+import argparse
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+try:
+    from src.graph.dynamic_features import (
+        build_discharge_timeseries, build_groundwater_timeseries,
+        build_climate_timeseries, assemble_dynamic_tensor,
+    )
+    from src.graph.physics_losses import build_routing_index, routing_consistency_loss
+except ImportError:
+    import sys
+    sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+    from src.graph.dynamic_features import (
+        build_discharge_timeseries, build_groundwater_timeseries,
+        build_climate_timeseries, assemble_dynamic_tensor,
+    )
+    from src.graph.physics_losses import build_routing_index, routing_consistency_loss
+
+BASIN_FILE_NAMES = {0: "eure", 1: "risle"}
+
+
+def run_for_basin(data_root: Path, basin_id: int, file_key: str, date_range, skip_climate: bool) -> None:
+    graph_dir = data_root / "reach_graph"
+    nodes_path = graph_dir / f"{file_key}_nodes_enriched.csv"
+    edges_path = graph_dir / f"{file_key}_edges.csv"
+    if not nodes_path.exists() or not edges_path.exists():
+        print(f"{file_key}: missing enriched nodes/edges -- run build_reach_graphs.py + "
+              f"enrich_reach_graph.py first. Skipping.")
+        return
+
+    nodes_df = pd.read_csv(nodes_path)
+    edges_df = pd.read_csv(edges_path)
+    print(f"--- {file_key}: {len(nodes_df)} nodes, {len(edges_df)} edges ---")
+
+    discharge_wide = build_discharge_timeseries(nodes_df, data_root / "hydrometric", date_range)
+    Q, dates = assemble_dynamic_tensor(nodes_df, discharge_wide)
+    n_real = int((~np.isnan(Q)).sum())
+    print(f"discharge tensor: {Q.shape}, {n_real}/{Q.size} real (non-NaN) values "
+          f"({100*n_real/Q.size:.3f}% coverage -- expect this to be tiny, only real "
+          f"gauges with real observations ever have a value here)")
+
+    level_wide, depth_wide = build_groundwater_timeseries(nodes_df, data_root / "ades", date_range)
+    level_tensor, _ = assemble_dynamic_tensor(nodes_df, level_wide)
+    depth_tensor, _ = assemble_dynamic_tensor(nodes_df, depth_wide)
+    print(f"groundwater level tensor: {level_tensor.shape}, "
+          f"{int((~np.isnan(level_tensor)).sum())} real values")
+
+    climate_tensors = {}
+    if not skip_climate and (data_root / "safran").exists():
+        try:
+            climate_dict = build_climate_timeseries(nodes_df, data_root / "safran", date_range)
+            for var, wide in climate_dict.items():
+                tensor, _ = assemble_dynamic_tensor(nodes_df, wide)
+                climate_tensors[var] = tensor
+            print(f"climate variables: {list(climate_tensors.keys())}")
+        except Exception as e:
+            print(f"climate skipped (error: {e})")
+
+    out_dir = graph_dir / "dynamic"
+    out_dir.mkdir(exist_ok=True)
+    save_kwargs = {
+        "discharge": Q, "groundwater_level": level_tensor, "groundwater_depth": depth_tensor,
+        "dates": np.array([str(d) for d in dates]), "station_codes": nodes_df["station_code"].values,
+    }
+    save_kwargs.update({f"climate_{k}": v for k, v in climate_tensors.items()})
+    out_path = out_dir / f"{file_key}_dynamic.npz"
+    np.savez(out_path, **save_kwargs)
+    print(f"saved to {out_path}")
+
+    # The actual integration: routing_consistency_loss needs Q + routing_index together.
+    routing_index = build_routing_index(nodes_df, edges_df, timestep_hours=24.0)
+    print(f"routing_index: {len(routing_index)} edge(s) with a usable lag at this timestep")
+
+    loss = routing_consistency_loss(Q, routing_index)
+    print(f"routing_consistency_loss on the REAL discharge tensor: {loss}")
+    if np.isnan(loss):
+        print("  ^ NaN. This is exactly the thing worth checking before assuming this loss "
+              "is usable: real Q is almost entirely NaN (only real gauges with real "
+              "observations have values), and if routing_consistency_loss doesn't mask "
+              "NaN out of its residuals before averaging, one missing value anywhere "
+              "poisons the entire loss to NaN. See whether this fired.")
+    print()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Build dynamic tensors and wire into physics_losses.py")
+    parser.add_argument("--data-root", type=Path, default=Path("datasets"))
+    parser.add_argument("--basin", choices=["eure", "risle", "both"], default="both")
+    parser.add_argument("--start-date", type=str, default="2013-01-01")
+    parser.add_argument("--end-date", type=str, default="2026-12-31")
+    parser.add_argument("--skip-climate", action="store_true")
+    args = parser.parse_args()
+
+    date_range = (args.start_date, args.end_date)
+    basins = BASIN_FILE_NAMES.items() if args.basin == "both" else \
+        [(k, v) for k, v in BASIN_FILE_NAMES.items() if v == args.basin]
+
+    for basin_id, file_key in basins:
+        run_for_basin(args.data_root, basin_id, file_key, date_range, args.skip_climate)
+
+
+if __name__ == "__main__":
+    main()

scripts/download_bdcavites.py

@@ -0,0 +1,146 @@
+"""
+Downloads BDCavités (BRGM's national underground cavity inventory --
+karst sinkholes, quarries, marl pits, natural cavities) for the Eure/
+Risle area, via Géorisques' WFS.
+
+Directly relevant to the bétoire investigation: this is an independent,
+purpose-built dataset for exactly this phenomenon, unlike inferring it
+from Hub'Eau station naming or BD TOPO's provisional karst attribute.
+
+CONFIRMED (fetched GetCapabilities directly): endpoint
+https://georisques.gouv.fr/services, WFS 1.1.0, typeName
+CAVITE_LOCALISEE ("Cavités souterraines abandonnées d'origine non
+minière"), GeoJSON output supported directly. Axis order for this
+specific server is NOT separately confirmed -- reusing the same
+lon,lat-then-lat,lon retry that scripts/download_bdtopo_hydro.py needed
+for a different WFS server, since different servers have behaved
+differently on this before and there's no reason to assume this one
+won't too.
+
+Usage:
+    python -m scripts.download_bdcavites --check
+    python -m scripts.download_bdcavites
+"""
+import argparse
+import json
+import sys
+from pathlib import Path
+
+import requests
+
+WFS_URL = "https://georisques.gouv.fr/services"
+TYPE_NAME = "CAVITE_LOCALISEE"
+
+# Same widened bbox as scripts/download_bdtopo_hydro.py's BBOX (updated
+# after confirmed real evidence of truncation on the original, tighter
+# box) -- keeping both scripts scoped to the same area.
+BBOX = (-0.1, 47.7, 2.1, 49.9)  # (min_lon, min_lat, max_lon, max_lat)
+
+
+def check_typename() -> bool:
+    params = {
+        "SERVICE": "WFS", "VERSION": "1.1.0", "REQUEST": "GetFeature",
+        "TYPENAME": TYPE_NAME, "MAXFEATURES": 1, "OUTPUTFORMAT": "application/json; subtype=geojson; charset=utf-8",
+    }
+    resp = requests.get(WFS_URL, params=params, timeout=30)
+    ok, detail = _parse_response(resp)
+    print(f"  {TYPE_NAME}: {'OK' if ok else 'FAILED'}")
+    if not ok:
+        print(f"    {detail}")
+    return ok
+
+
+def _parse_response(resp: "requests.Response"):
+    if resp.status_code != 200:
+        return False, f"HTTP {resp.status_code}: {resp.text[:400]}"
+    try:
+        data = resp.json()
+    except ValueError:
+        return False, f"Response was not JSON (likely a WFS ExceptionReport): {resp.text[:400]}"
+    if isinstance(data, dict) and data.get("type") == "FeatureCollection":
+        return True, ""
+    return False, f"Response was JSON but not a FeatureCollection: {str(data)[:400]}"
+
+
+def fetch_all_pages(bbox_param: str, page_size: int = 1000) -> list:
+    all_features = []
+    start_index = 0
+    while True:
+        params = {
+            "SERVICE": "WFS", "VERSION": "1.1.0", "REQUEST": "GetFeature",
+            "TYPENAME": TYPE_NAME, "BBOX": bbox_param,
+            "OUTPUTFORMAT": "application/json; subtype=geojson; charset=utf-8",
+            "MAXFEATURES": page_size, "STARTINDEX": start_index,
+        }
+        resp = requests.get(WFS_URL, params=params, timeout=60)
+        ok, detail = _parse_response(resp)
+        if not ok:
+            print(f"    ERROR at startIndex={start_index}: {detail}")
+            break
+        data = resp.json()
+        features = data.get("features", [])
+        if not features:
+            break
+        all_features.extend(features)
+        if len(features) < page_size:
+            break
+        start_index += page_size
+    return all_features
+
+
+def fetch_with_axis_retry(bbox: tuple, out_path: Path) -> None:
+    min_lon, min_lat, max_lon, max_lat = bbox
+    orderings = [
+        ("lon,lat", f"{min_lon},{min_lat},{max_lon},{max_lat},urn:ogc:def:crs:EPSG::4326"),
+        ("lat,lon", f"{min_lat},{min_lon},{max_lat},{max_lon},urn:ogc:def:crs:EPSG::4326"),
+    ]
+    features = []
+    for label, bbox_param in orderings:
+        print(f"  trying axis order {label}...")
+        features = fetch_all_pages(bbox_param)
+        if features:
+            print(f"  -> {label} worked ({len(features)} feature(s))")
+            break
+        print(f"  -> {label} returned 0 features")
+
+    geojson = {"type": "FeatureCollection", "features": features}
+    out_path.write_text(json.dumps(geojson))
+    if features:
+        print(f"Saved {len(features)} feature(s) to {out_path}")
+    else:
+        print(f"Saved an EMPTY file to {out_path} -- both axis orders returned nothing. "
+              f"Run --check first, or this area may genuinely have zero recorded cavities "
+              f"(plausible -- BDCavités coverage is built department-by-department and "
+              f"isn't uniformly complete everywhere).")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Download BDCavités for the Eure/Risle area")
+    parser.add_argument("--check", action="store_true")
+    parser.add_argument("--output-dir", type=Path, default=Path("datasets/bdcavites"))
+    args = parser.parse_args()
+
+    if args.check:
+        print("Testing typeName against the live service...")
+        ok = check_typename()
+        if not ok:
+            print("\nFailed -- the typeName or service details may have changed since this "
+                  "script was written. Fetch GetCapabilities directly to check:")
+            print(f"  {WFS_URL}?SERVICE=WFS&VERSION=1.1.0&REQUEST=GetCapabilities")
+        else:
+            print("\nOK. Re-run without --check to download.")
+        return
+
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    print("Fetching CAVITE_LOCALISEE (BDCavités)...")
+    fetch_with_axis_retry(BBOX, args.output_dir / "cavite_localisee.geojson")
+    print()
+    print("Once downloaded, cross-reference against the amont/aval bétoire stations "
+          "(H605641101 at 48.98492,0.78902 and H605641201 at 49.04707,0.79984) the same "
+          "way scripts/analyze_bdtopo_hydro.py's check_karst_near_betoire did for the "
+          "BD TOPO karst attribute -- this is a genuinely independent second check, "
+          "not a re-run of the same one.")
+
+
+if __name__ == "__main__":
+    main()

scripts/download_bdcharm.py

@@ -0,0 +1,107 @@
+"""
+Downloads BD Charm-50 (BRGM's harmonized 1:50,000 geological maps) for
+the departments covering the Eure/Risle study area.
+
+CONFIRMED (via data.gouv.fr / InfoTerre): free, open (Licence Ouverte),
+no authentication -- direct per-department ZIP download from InfoTerre.
+This is a genuinely different access pattern than the WFS sources
+elsewhere in this project (BD TOPO, BDCavités): no bbox query, no axis-
+order ambiguity, just a fixed URL per department. Real, working URL
+pattern (found directly, not guessed):
+    http://infoterre.brgm.fr/telechargements/BDCharm50/GEO050K_HARM_XXX.zip
+where XXX is the 3-digit department code (e.g. 027 for Eure).
+
+NOTE: a separate, DIFFERENT distribution of this same underlying dataset
+exists that requires CIGAL network membership (seen for at least Alsace
+regional data) -- that is NOT what this script uses. This script only
+uses the free InfoTerre download confirmed via data.gouv.fr.
+
+DEPARTMENTS: 27 (Eure), 61 (Orne), 28 (Eure-et-Loir) -- 27 and 61
+confirmed directly from real INSEE codes seen in this project's own
+station_list.csv (61342, 27040, 27116, 27468); 28 included because the
+Eure's own southern tributaries (Voise, Drouette) run through Chartres/
+Dreux, unambiguously in Eure-et-Loir. Add more department codes via
+--departments if the real geographic extent turns out to need them.
+
+Usage:
+    python -m scripts.download_bdcharm50
+"""
+import argparse
+import zipfile
+from pathlib import Path
+
+import requests
+
+DEFAULT_DEPARTMENTS = ["027", "028", "061"]
+BASE_URL = "http://infoterre.brgm.fr/telechargements/BDCharm50"
+
+
+def download_department(dept: str, output_dir: Path) -> bool:
+    url = f"{BASE_URL}/GEO050K_HARM_{dept}.zip"
+    zip_path = output_dir / f"GEO050K_HARM_{dept}.zip"
+
+    print(f"  department {dept}: {url}")
+    try:
+        resp = requests.get(url, timeout=120, stream=True)
+    except requests.RequestException as e:
+        print(f"    FAILED: {e}")
+        return False
+
+    if resp.status_code != 200:
+        print(f"    FAILED: HTTP {resp.status_code}")
+        return False
+
+    content_type = resp.headers.get("Content-Type", "")
+    if "zip" not in content_type and "octet-stream" not in content_type:
+        # A 200 status with an HTML content-type here usually means an
+        # error page or a "department not available" page was returned
+        # instead of the real ZIP -- catching this explicitly rather
+        # than silently saving an HTML file with a .zip extension.
+        print(f"    WARNING: Content-Type is {content_type!r}, not zip -- "
+              f"this department's file may not exist at this URL. Saving "
+              f"anyway for inspection, but verify before trusting it.")
+
+    zip_path.write_bytes(resp.content)
+    size_kb = zip_path.stat().st_size / 1024
+    print(f"    saved {zip_path} ({size_kb:.0f} KB)")
+
+    extract_dir = output_dir / f"dept_{dept}"
+    try:
+        with zipfile.ZipFile(zip_path) as zf:
+            zf.extractall(extract_dir)
+        shp_files = list(extract_dir.rglob("*.shp"))
+        print(f"    extracted to {extract_dir} ({len(shp_files)} .shp file(s) found)")
+        return True
+    except zipfile.BadZipFile:
+        print(f"    FAILED: downloaded file is not a valid zip -- likely an error page, "
+              f"not real data. Check {zip_path} directly.")
+        return False
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Download BD Charm-50 geological maps")
+    parser.add_argument("--departments", nargs="+", default=DEFAULT_DEPARTMENTS,
+                         help="3-digit department codes, e.g. 027 028 061")
+    parser.add_argument("--output-dir", type=Path, default=Path("datasets/bdcharm50"))
+    args = parser.parse_args()
+
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    print(f"Downloading BD Charm-50 for departments: {args.departments}")
+    print()
+
+    results = {}
+    for dept in args.departments:
+        results[dept] = download_department(dept, args.output_dir)
+        print()
+
+    print("=" * 60)
+    ok = [d for d, r in results.items() if r]
+    failed = [d for d, r in results.items() if not r]
+    print(f"Succeeded: {ok}")
+    if failed:
+        print(f"Failed: {failed} -- check the URL pattern still matches by visiting "
+              f"https://infoterre.brgm.fr/page/telechargement-cartes-geologiques directly")
+
+
+if __name__ == "__main__":
+    main()

scripts/fetch_idpr_brgm.py

@@ -0,0 +1,144 @@
+"""
+Fetches fresh IDPR (Indice de Développement et de Persistance des
+Réseaux) values directly from BRGM's live geoservice, replacing/
+supplementing the manually-uploaded idpr.csv.
+
+CONFIRMED (fetched GetCapabilities directly): endpoint
+http://geoservices.brgm.fr/geologie, WMS 1.3.0, live and responding.
+IDPR is raster/grid data at five resolutions (IDPR_50M down to
+IDPR_5000M) -- not a vector feature type, so WFS GetFeature (the
+pattern used everywhere else in this project) doesn't apply. The
+correct WMS operation for extracting a value at a specific point from a
+raster layer is GetFeatureInfo: request a small map image centered on
+the point, then ask "what's the value at this pixel."
+
+This queries ONE point at a time (GetFeatureInfo has no bulk/vectorized
+equivalent the way BD TOPO's WFS BBOX queries did) -- scoped to the 27
+real gauge stations, matching what "redownload IDPR" means (fresh
+values at known station points), not the full ~4,500-node reach graph
+(which would mean thousands of individual HTTP requests -- a different,
+much larger undertaking than this).
+
+Usage:
+    python -m scripts.fetch_idpr_brgm --check
+    python -m scripts.fetch_idpr_brgm --stations datasets/station_elevations.csv
+"""
+import argparse
+import re
+from pathlib import Path
+
+import pandas as pd
+import requests
+
+WMS_URL = "http://geoservices.brgm.fr/geologie"
+LAYER = "IDPR_50M"  # highest resolution; use --layer to try a coarser one if this fails
+
+
+def get_feature_info_at_point(lat: float, lon: float, layer: str = LAYER,
+                               half_extent_deg: float = 0.001) -> "float | None":
+    """
+    Query IDPR value at a single point via WMS GetFeatureInfo: request a
+    tiny 3x3 pixel map centered on (lat, lon), then ask for the value at
+    the center pixel.
+    """
+    minx, miny = lon - half_extent_deg, lat - half_extent_deg
+    maxx, maxy = lon + half_extent_deg, lat + half_extent_deg
+    params = {
+        "SERVICE": "WMS", "VERSION": "1.1.1", "REQUEST": "GetFeatureInfo",
+        "LAYERS": layer, "QUERY_LAYERS": layer, "STYLES": "",
+        "BBOX": f"{minx},{miny},{maxx},{maxy}", "SRS": "EPSG:4326",
+        "WIDTH": 3, "HEIGHT": 3, "X": 1, "Y": 1,
+        "INFO_FORMAT": "text/plain",
+    }
+    try:
+        resp = requests.get(WMS_URL, params=params, timeout=30)
+    except requests.RequestException as e:
+        print(f"    request failed: {e}")
+        return None
+
+    if resp.status_code != 200:
+        print(f"    HTTP {resp.status_code}")
+        return None
+
+    return parse_getfeatureinfo_value(resp.text)
+
+
+def parse_getfeatureinfo_value(text: str) -> "float | None":
+    """
+    MapServer's text/plain GetFeatureInfo output is typically a small
+    block like:
+        Layer 'IDPR_50M'
+        Feature 0
+          value = '885'
+    Parsing defensively: find any line with 'value' (case-insensitive)
+    and extract the first number on it, rather than assuming an exact
+    format -- MapServer's plain-text output format has enough real
+    variation across deployments that a strict parser is more likely to
+    silently return nothing than a permissive one is to return a wrong
+    number.
+    """
+    for line in text.splitlines():
+        if "value" in line.lower():
+            match = re.search(r"[-+]?\d*\.?\d+", line)
+            if match:
+                return float(match.group())
+    return None
+
+
+def check_service() -> bool:
+    # A real point known to be in mainland France, well within IDPR coverage
+    test_lat, test_lon = 49.03, 0.79
+    print(f"Testing {LAYER} at ({test_lat}, {test_lon})...")
+    value = get_feature_info_at_point(test_lat, test_lon)
+    if value is not None:
+        print(f"  OK: value = {value}")
+        return True
+    print("  FAILED: no value returned. Try --layer IDPR_100M or IDPR_500M "
+          "(coarser resolutions), or check GetCapabilities directly:")
+    print(f"  {WMS_URL}?SERVICE=WMS&VERSION=1.3.0&REQUEST=GetCapabilities")
+    return False
+
+
+def fetch_for_stations(stations_path: Path, layer: str, output_path: Path) -> None:
+    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"
+
+    rows = []
+    for _, row in stations.iterrows():
+        code = row["station_code"]
+        value = get_feature_info_at_point(row[lat_col], row[lon_col], layer=layer)
+        print(f"  {code}: {value}")
+        rows.append({"station_code": code, "idpr_value_brgm_live": value,
+                      "latitude": row[lat_col], "longitude": row[lon_col]})
+
+    out = pd.DataFrame(rows)
+    n_ok = out["idpr_value_brgm_live"].notna().sum()
+    print()
+    print(f"{n_ok}/{len(out)} stations got a real value")
+    out.to_csv(output_path, index=False)
+    print(f"Saved to {output_path}")
+    if n_ok < len(out):
+        print("Some stations got no value -- this could mean they're just outside "
+              "IDPR's coverage, or the layer/resolution needs adjusting. Compare "
+              "against the existing idpr.csv for those specific stations before "
+              "assuming the live fetch is wrong.")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Fetch fresh IDPR values from BRGM's live geoservice")
+    parser.add_argument("--check", action="store_true")
+    parser.add_argument("--stations", type=Path, default=Path("datasets/station_elevations.csv"))
+    parser.add_argument("--layer", type=str, default=LAYER)
+    parser.add_argument("--output", type=Path, default=Path("datasets/idpr_brgm_live.csv"))
+    args = parser.parse_args()
+
+    if args.check:
+        check_service()
+        return
+
+    fetch_for_stations(args.stations, args.layer, args.output)
+
+
+if __name__ == "__main__":
+    main()

scripts/fetch_landcover.py

@@ -0,0 +1,134 @@
+"""
+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()

scripts/fetch_worldcover_ndvi.py

@@ -0,0 +1,155 @@
+"""
+Samples ESA WorldCover's Sentinel-2 NDVI yearly percentile composite
+(p10/p50/p90) at station points, via the public S3 COGs.
+
+Unlike fetch_worldcover_landcover.py, this does NOT hand-compute the
+tile ID and guess the S3 key pattern -- VITO provides an authoritative
+grid file for exactly this purpose (found directly, not guessed):
+    https://esa-worldcover.s3.eu-central-1.amazonaws.com/esa_worldcover_grid_composites.fgb
+This is a FlatGeobuf containing one feature per 1x1 degree composite
+tile, with (per VITO's own documentation) the real S3 path/URL for
+each tile as an attribute -- reading it means never guessing a file
+naming convention, which is exactly the class of guess that failed for
+the IDPR WMS layer name earlier in this session.
+
+The composites use a 1x1 degree grid, NOT the 3x3 degree grid the
+landcover classification map uses -- so unlike fetch_worldcover_
+landcover.py's single hardcoded tile, station coordinates spanning
+more than 1 degree in either direction genuinely need more than one
+tile here. This script looks up each station's tile individually
+rather than assuming one tile covers everyone.
+
+NEEDS: geopandas (to read the grid FlatGeobuf) and rasterio (to sample
+the COGs). CAVEAT: the grid file's actual column names/schema are not
+independently verified here (no network access to inspect it directly)
+-- the script prints all columns and searches for a plausible URL/path
+column rather than assuming an exact name, and reports clearly if it
+can't find one.
+
+Usage:
+    python -m scripts.fetch_worldcover_ndvi --check
+    python -m scripts.fetch_worldcover_ndvi --stations datasets/station_elevations.csv
+"""
+import argparse
+from pathlib import Path
+
+import pandas as pd
+
+GRID_URL = "https://esa-worldcover.s3.eu-central-1.amazonaws.com/esa_worldcover_grid_composites.fgb"
+
+# The grid file's tile URLs use the s3:// scheme (confirmed against real
+# output: s3://esa-worldcover-s2/ndvi/2020/N48/...) -- GDAL's S3 driver
+# tries to SIGN requests with real AWS credentials by default even
+# though this bucket is fully public, unlike the HTTPS URL
+# fetch_worldcover_landcover.py used (a plain HTTPS GET needs no
+# signing at all, which is why that one worked without this). This
+# environment variable is the standard fix for reading a public S3
+# bucket without needing real credentials.
+import os
+os.environ.setdefault("AWS_NO_SIGN_REQUEST", "YES")
+
+
+def load_grid():
+    import geopandas as gpd
+    print(f"Loading tile grid from {GRID_URL} ...")
+    grid = gpd.read_file(GRID_URL)
+    print(f"  {len(grid)} tile(s), columns: {list(grid.columns)}")
+    return grid
+
+
+def find_url_column(grid) -> str:
+    """
+    The grid's real column name for the tile's S3 path isn't
+    independently confirmed -- search for a plausible one rather than
+    assume, and fail loudly (not silently) if nothing matches.
+    """
+    candidates = [c for c in grid.columns if any(
+        kw in c.lower() for kw in ("url", "href", "path", "s3", "ndvi", "product", "file")
+    )]
+    if not candidates:
+        raise ValueError(
+            f"No column looks like a tile URL/path among {list(grid.columns)}. "
+            f"Inspect the grid file's real schema directly (e.g. print(grid.head()) "
+            f"in a notebook) and adjust find_url_column accordingly."
+        )
+    print(f"  candidate URL/path column(s): {candidates} -- using {candidates[0]!r}")
+    return candidates[0]
+
+
+def check_access() -> bool:
+    try:
+        import geopandas  # noqa: F401
+        import rasterio  # noqa: F401
+    except ImportError as e:
+        print(f"Missing dependency: {e}. pip install geopandas rasterio --break-system-packages")
+        return False
+
+    try:
+        grid = load_grid()
+        url_col = find_url_column(grid)
+        print(f"  Sample values from {url_col!r}: {grid[url_col].head(3).tolist()}")
+        return True
+    except Exception as e:
+        print(f"  FAILED: {e}")
+        return False
+
+
+def fetch_for_stations(stations_path: Path, output_path: Path) -> None:
+    import geopandas as gpd
+    import rasterio
+    from shapely.geometry import Point
+
+    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"
+
+    grid = load_grid()
+    url_col = find_url_column(grid)
+
+    station_points = gpd.GeoDataFrame(
+        stations,
+        geometry=[Point(lon, lat) for lon, lat in zip(stations[lon_col], stations[lat_col])],
+        crs=grid.crs,
+    )
+    joined = gpd.sjoin(station_points, grid[[url_col, "geometry"]], how="left", predicate="within")
+
+    rows = []
+    for tile_url, group in joined.groupby(url_col):
+        if pd.isna(tile_url):
+            for _, row in group.iterrows():
+                rows.append({"station_code": row["station_code"], "ndvi_p10": None,
+                             "ndvi_p50": None, "ndvi_p90": None})
+            continue
+        print(f"  opening {tile_url} for {len(group)} station(s)...")
+        with rasterio.Env(AWS_NO_SIGN_REQUEST="YES"), rasterio.open(tile_url) as src:
+            coords = list(zip(group[lon_col], group[lat_col]))
+            values = list(src.sample(coords))
+        for (_, row), val in zip(group.iterrows(), values):
+            # NDVI percentile composite is documented as 3 bands: p90, p50, p10
+            rows.append({"station_code": row["station_code"],
+                         "ndvi_p90": float(val[0]) if len(val) > 0 else None,
+                         "ndvi_p50": float(val[1]) if len(val) > 1 else None,
+                         "ndvi_p10": float(val[2]) if len(val) > 2 else None})
+
+    out = pd.DataFrame(rows)
+    out.to_csv(output_path, index=False)
+    print(f"Saved {len(out)} stations' NDVI values to {output_path}")
+    print(out.describe())
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Sample ESA WorldCover NDVI 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_ndvi.csv"))
+    args = parser.parse_args()
+
+    if args.check:
+        check_access()
+        return
+
+    fetch_for_stations(args.stations, args.output)
+
+
+if __name__ == "__main__":
+    main()

src/graph/build_graph.py

@@ -302,6 +302,22 @@ def build_pyg_graph(
         "is_gauged", "is_confluence", "is_split_point", "is_rejoin_point",
         "snap_distance_km", "braid_id",
     }
+    # Feature columns that are genuinely time-varying in reality, even
+    # though today's pipeline only ever produces a period-aggregate
+    # value for them (see node_features.py's add_safran_features/
+    # add_groundwater_features/add_ndvi_features) -- distinct from
+    # STATIC_COLUMNS below, which are physically time-invariant (a
+    # location's elevation or IDPR index doesn't change on any
+    # timescale relevant here). n_nearby_wells is classified static
+    # despite living under the groundwater loader: it describes monitor
+    # coverage (which wells exist nearby), not a water-table quantity
+    # that itself varies day to day.
+    DYNAMIC_PREFIXES = ("climate_", "avg_groundwater_level", "avg_groundwater_depth", "ndvi_")
+
+    def _is_dynamic(col: str) -> bool:
+        base = col[:-len("__was_missing")] if col.endswith("__was_missing") else col
+        return any(base.startswith(p) for p in DYNAMIC_PREFIXES)
+
     target_cols = [c for c in nodes_df.columns if c.startswith("target_")]
     if feature_columns is None:
         feature_columns = [
@@ -368,6 +384,28 @@ def build_pyg_graph(
     data.feature_names = all_feature_cols
     data.basin_id = torch.tensor(nodes_df["basin_id"].values, dtype=torch.long)
 
+    # Static/dynamic split: `x` stays the full combined tensor (nothing
+    # existing that reads data.x breaks), but every feature is also
+    # tagged and split out separately -- x_static for physically
+    # time-invariant quantities (elevation, IDPR, catchment area,
+    # landcover), x_dynamic for quantities that vary in reality even
+    # though the current pipeline only ever hands them over as a single
+    # period-aggregate (climate, groundwater level/depth, NDVI). See
+    # DYNAMIC_PREFIXES above for exactly which columns land where, and
+    # node_features.py's add_safran_features/add_groundwater_features/
+    # add_ndvi_features docstrings for why "dynamic" here still means
+    # "one aggregated number," not a real time series yet -- a genuine
+    # multi-timestep pipeline is a separate, larger piece of future work
+    # this split does not attempt to solve on its own.
+    dynamic_mask = [_is_dynamic(c) for c in all_feature_cols]
+    static_idx = [i for i, d in enumerate(dynamic_mask) if not d]
+    dynamic_idx = [i for i, d in enumerate(dynamic_mask) if d]
+    data.static_feature_names = [all_feature_cols[i] for i in static_idx]
+    data.dynamic_feature_names = [all_feature_cols[i] for i in dynamic_idx]
+    x_np_for_split = feat[all_feature_cols].values
+    data.x_static = torch.tensor(x_np_for_split[:, static_idx], dtype=torch.float) if static_idx else None
+    data.x_dynamic = torch.tensor(x_np_for_split[:, dynamic_idx], dtype=torch.float) if dynamic_idx else None
+
     for bool_col in ("is_gauged", "is_confluence", "is_split_point", "is_rejoin_point"):
         if bool_col in nodes_df.columns:
             setattr(data, bool_col, torch.tensor(nodes_df[bool_col].values, dtype=torch.bool))

src/graph/dynamic_features.py

@@ -0,0 +1,205 @@
+"""
+Builds genuine [n_nodes, T] time series for the features that
+node_features.py's add_safran_features/add_groundwater_features/
+add_hydrometric_target_stats only ever hand over as a single period-
+aggregated number. That aggregation was always a real, named
+limitation (see build_graph.py's DYNAMIC_PREFIXES docstring and the
+README's §2.3) -- this module is what actually closes it.
+
+physics_losses.py's routing_consistency_loss specifically needs a real
+time dimension and had nothing to consume before this existed.
+
+Three sources, three different real challenges:
+  - discharge: already daily, already per-station -- just needs
+    pivoting to wide form, restricted to real gauges (no fabrication
+    for non-gauge nodes).
+  - groundwater: wells report on wildly irregular schedules (confirmed
+    against real data: 13 different "latest dates" among 18 wells
+    within 20km of one station). Needs resampling to a common daily
+    grid via forward-fill (a well's level changes slowly -- carrying
+    the last known reading forward is standard practice for sparse
+    level data, not an invented shortcut) BEFORE the spatial average,
+    not after -- averaging raw irregular readings per exact calendar
+    date is exactly the bug that made the old "aggregate_to_stations"
+    path undercount real well coverage by 5-10x (see node_features.py's
+    add_groundwater_features docstring).
+  - climate: SAFRANLoader.load() already returns per-date rows before
+    node_features.py's add_safran_features aggregates them -- this
+    just skips that aggregation step and pivots instead. UNTESTED in
+    this environment: no real ERA5/safran files were available here to
+    validate against (only real ADES and hydrometric data were).
+"""
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+import numpy as np
+import pandas as pd
+
+
+def build_discharge_timeseries(
+    nodes_df: pd.DataFrame, hydrometric_path: Path, date_range: Tuple[str, str],
+    freq: str = "D",
+) -> pd.DataFrame:
+    """
+    Real daily QmnJ discharge, pivoted to [date x station_code] wide
+    form, restricted to the date range and reindexed onto a regular
+    daily grid (missing days stay NaN -- no fabrication).
+
+    Deliberately does NOT forward-fill discharge the way groundwater
+    gets resampled below: discharge genuinely changes day to day and a
+    missing daily reading should stay missing, not be papered over with
+    the previous day's value the way a slowly-changing water table can
+    reasonably be.
+
+    Only real gauge station_codes appear as columns -- confluences and
+    virtual nodes never had discharge observations to begin with, same
+    principle as the period-aggregate target_* columns.
+    """
+    from ..data.loaders.hydrometric import HydrometricLoader
+
+    loader = HydrometricLoader(data_path=hydrometric_path)
+    df = loader.load()
+    if "discharge_m3s" not in df.columns:
+        raise ValueError("Loaded hydrometric data has no discharge_m3s column")
+
+    start, end = date_range
+    df = df[(df["date"] >= start) & (df["date"] <= end)]
+    wide = df.pivot_table(index="date", columns="station_code", values="discharge_m3s", aggfunc="mean")
+
+    full_index = pd.date_range(start, end, freq=freq)
+    wide = wide.reindex(full_index)
+    wide.index.name = "date"
+    return wide
+
+
+def build_groundwater_timeseries(
+    nodes_df: pd.DataFrame, ades_path: Path, date_range: Tuple[str, str],
+    max_distance_km: float = 20.0, freq: str = "D",
+) -> Tuple[pd.DataFrame, pd.DataFrame]:
+    """
+    Real groundwater level, resampled to a daily grid per well
+    (forward-fill) BEFORE spatial averaging, then averaged per node
+    using the SAME nearby-well set every day -- the spatial join
+    (which wells are near which node) is time-invariant, so it's
+    computed once, not repeated per date. That's what keeps this
+    tractable at real reach-graph scale: O(n_nodes) spatial lookups
+    total, not O(n_nodes x n_dates).
+
+    Returns:
+        (level_wide, depth_wide) -- both [date x station_code], node
+        ordering matching nodes_df. A node with zero wells within
+        max_distance_km gets NaN for every date, not zero -- "no
+        nearby monitoring" is meaningfully different from "measured
+        zero," same principle as node_features.py's static version.
+    """
+    from ..data.loaders.ades import ADESLoader
+    from scipy.spatial import cKDTree
+
+    loader = ADESLoader(data_path=ades_path)
+    gw_df = loader.load()
+    start, end = date_range
+    full_index = pd.date_range(start, end, freq=freq)
+
+    # Resample each well independently to the common daily grid, forward-fill.
+    wells_wide = gw_df.pivot_table(index="date", columns="code_bss", values="groundwater_level_m", aggfunc="mean")
+    wells_wide = wells_wide.reindex(wells_wide.index.union(full_index)).sort_index().ffill()
+    wells_wide = wells_wide.reindex(full_index)
+
+    depth_wide_raw = gw_df.pivot_table(index="date", columns="code_bss", values="groundwater_depth_m", aggfunc="mean") \
+        if "groundwater_depth_m" in gw_df.columns else None
+    if depth_wide_raw is not None:
+        depth_wide_raw = depth_wide_raw.reindex(depth_wide_raw.index.union(full_index)).sort_index().ffill()
+        depth_wide_raw = depth_wide_raw.reindex(full_index)
+
+    well_coords = gw_df.drop_duplicates("code_bss").set_index("code_bss")[["lat", "lon"]]
+    well_coords = well_coords.reindex(wells_wide.columns)  # align to the pivoted columns' order
+    tree = cKDTree(well_coords[["lon", "lat"]].values)
+    coarse_radius_deg = (max_distance_km / 111.0) * 1.5
+
+    def _haversine_km(lat1, lon1, lat2, lon2):
+        R = 6371.0
+        lat1r, lon1r, lat2r, lon2r = map(np.radians, [lat1, lon1, lat2, lon2])
+        dlat, dlon = lat2r - lat1r, lon2r - lon1r
+        a = np.sin(dlat / 2) ** 2 + np.cos(lat1r) * np.cos(lat2r) * np.sin(dlon / 2) ** 2
+        return R * 2 * np.arcsin(np.sqrt(a))
+
+    station_lon = nodes_df["longitude"].values
+    station_lat = nodes_df["latitude"].values
+    candidate_lists = tree.query_ball_point(np.column_stack([station_lon, station_lat]), r=coarse_radius_deg)
+
+    level_cols, depth_cols = {}, {}
+    for i, station_code in enumerate(nodes_df["station_code"]):
+        candidates = candidate_lists[i]
+        if not candidates:
+            level_cols[station_code] = pd.Series(np.nan, index=full_index)
+            depth_cols[station_code] = pd.Series(np.nan, index=full_index)
+            continue
+        cand_idx = np.array(candidates)
+        cand_lat = well_coords["lat"].values[cand_idx]
+        cand_lon = well_coords["lon"].values[cand_idx]
+        dist = _haversine_km(station_lat[i], station_lon[i], cand_lat, cand_lon)
+        within = cand_idx[dist <= max_distance_km]
+        if len(within) == 0:
+            level_cols[station_code] = pd.Series(np.nan, index=full_index)
+            depth_cols[station_code] = pd.Series(np.nan, index=full_index)
+            continue
+        well_names = wells_wide.columns[within]
+        level_cols[station_code] = wells_wide[well_names].mean(axis=1)
+        if depth_wide_raw is not None:
+            depth_cols[station_code] = depth_wide_raw[well_names].mean(axis=1)
+        else:
+            depth_cols[station_code] = pd.Series(np.nan, index=full_index)
+
+    level_wide = pd.DataFrame(level_cols)
+    depth_wide = pd.DataFrame(depth_cols)
+    return level_wide, depth_wide
+
+
+def build_climate_timeseries(
+    nodes_df: pd.DataFrame, safran_path: Path, date_range: Tuple[str, str],
+) -> Dict[str, pd.DataFrame]:
+    """
+    Real per-date ERA5 variables, pivoted per variable to [date x
+    station_code] wide form -- skips node_features.py's add_safran_
+    features aggregation step entirely rather than reversing it.
+
+    UNTESTED in this environment: no real era5_*.nc files were
+    available here (only ADES and hydrometric real data were). The
+    logic mirrors build_discharge_timeseries's pivot pattern directly,
+    but please verify the real output shape/values before trusting it
+    for training.
+    """
+    from ..data.loaders.safran import SAFRANLoader
+
+    station_coords = nodes_df.rename(columns={"latitude": "lat", "longitude": "lon"})[
+        ["station_code", "lat", "lon"]
+    ]
+    loader = SAFRANLoader(data_path=safran_path, station_coords=station_coords)
+    df = loader.load()
+    df = loader.convert_units(df)
+
+    start, end = date_range
+    df = df[(df["date"] >= start) & (df["date"] <= end)]
+
+    var_cols = [c for c in df.columns if c not in ("date", "station_code")]
+    result = {}
+    for var in var_cols:
+        wide = df.pivot_table(index="date", columns="station_code", values=var, aggfunc="mean")
+        result[var] = wide
+    return result
+
+
+def assemble_dynamic_tensor(nodes_df: pd.DataFrame, wide_df: pd.DataFrame) -> Tuple[np.ndarray, List[pd.Timestamp]]:
+    """
+    Aligns a [date x station_code] wide DataFrame onto nodes_df's own
+    row order, so the result matches data.x's node indexing exactly --
+    this is the piece that makes a build_*_timeseries output directly
+    usable as physics_losses.py's routing_consistency_loss's `Q`
+    argument (shape [n_nodes, T]).
+
+    A node whose station_code never appears as a column (every
+    confluence/virtual node, for discharge) gets an all-NaN row, not a
+    dropped row -- shape stays [n_nodes, T] regardless of coverage.
+    """
+    aligned = wide_df.reindex(columns=nodes_df["station_code"])
+    return aligned.values.T, list(wide_df.index)

src/graph/node_features.py

@@ -131,6 +131,181 @@ def add_idpr_features(nodes_df: pd.DataFrame, idpr_path: Path) -> pd.DataFrame:
     return out
 
 
+def add_geology_features(nodes_df: pd.DataFrame, bdcharm_path: Path) -> pd.DataFrame:
+    """
+    Geological formation class per node, one-hot encoded, from
+    scripts/download_bdcharm.py's per-department BD Charm-50 shapefiles
+    (datasets/bdcharm50/dept_{027,028,061}/).
+
+    One-hot, not a raw formation code -- same reasoning as
+    add_landcover_features: geological formation is a nominal category,
+    not an ordered quantity, so a raw code would let build_pyg_graph's
+    numeric auto-detection z-score it as if formation codes had a
+    meaningful numeric ordering, which they don't.
+    """
+    import geopandas as gpd
+    from shapely.geometry import Point
+
+    base = Path(bdcharm_path)
+    shp_files = sorted(base.glob("dept_*/**/*.shp"))
+    fgeol_files = [f for f in shp_files if "FGEOL" in f.name.upper()]
+    if not fgeol_files:
+        raise FileNotFoundError(
+            f"No *FGEOL* shapefile found under {base}/dept_*/. Found instead: "
+            f"{[f.name for f in shp_files]}. Inspect these directly to find the "
+            f"real geological-formations layer if the naming differs from what "
+            f"was expected."
+        )
+    print(f"add_geology_features: using formation layer(s): {[f.name for f in fgeol_files]}")
+
+    polygons = pd.concat([gpd.read_file(f) for f in fgeol_files], ignore_index=True)
+    polygons = gpd.GeoDataFrame(polygons, geometry="geometry")
+
+    label_candidates = [c for c in polygons.columns if any(
+        kw in c.upper() for kw in ("NOTATION", "CODE", "LEGENDE", "LEG", "LITHO", "FORMATION")
+    ) and c.upper() != "GEOMETRY"]
+    if not label_candidates:
+        raise ValueError(
+            f"No plausible lithology/formation column found among {list(polygons.columns)}. "
+            f"Inspect the real shapefile's attribute table directly and adjust the "
+            f"keyword list in add_geology_features."
+        )
+    label_col = label_candidates[0]
+    print(f"add_geology_features: using attribute column {label_col!r} "
+          f"(other candidates considered: {label_candidates[1:]})")
+
+    stations = gpd.GeoDataFrame(
+        nodes_df,
+        geometry=[Point(lon, lat) for lon, lat in zip(nodes_df["longitude"], nodes_df["latitude"])],
+        crs="EPSG:4326",
+    )
+    if polygons.crs is not None and polygons.crs != stations.crs:
+        polygons = polygons.to_crs(stations.crs)
+
+    joined = gpd.sjoin(stations, polygons[[label_col, "geometry"]], how="left", predicate="within")
+    joined = joined.drop_duplicates(subset="station_code" if "station_code" in joined.columns else joined.index.name)
+
+    n_matched = joined[label_col].notna().sum()
+    print(f"add_geology_features: {n_matched}/{len(nodes_df)} nodes matched to a real formation polygon")
+
+    dummies = pd.get_dummies(joined[label_col], prefix="geology")
+    result = pd.concat([nodes_df.reset_index(drop=True), dummies.reset_index(drop=True)], axis=1)
+    return result
+
+
+def add_cavites_features(
+    nodes_df: pd.DataFrame, bdcavites_path: Path, max_distance_km: float = 20.0
+) -> pd.DataFrame:
+    """
+    Distance to nearest known cavity/sinkhole + count within
+    max_distance_km, from scripts/download_bdcavites.py's
+    cavite_localisee.geojson (BRGM's national cavity inventory, via
+    Géorisques' WFS).
+
+    Same KD-tree coarse-prefilter + exact-haversine pattern already
+    validated for add_groundwater_features -- point data at unknown
+    real density, so avoid assuming either a fast-enough naive loop or
+    guessing at scale ahead of time.
+    """
+    import json
+    from scipy.spatial import cKDTree
+
+    path = Path(bdcavites_path)
+    if not path.exists():
+        raise FileNotFoundError(f"{path} not found")
+
+    geojson = json.loads(path.read_text())
+    features = geojson.get("features", [])
+    if not features:
+        print(f"add_cavites_features: {path} has 0 features -- "
+              f"every station will get distance=NaN, count=0")
+        out = nodes_df.copy()
+        out["distance_to_nearest_cavity_km"] = float("nan")
+        out["n_cavities_within_20km"] = 0
+        return out
+
+    cavity_coords = []
+    for f in features:
+        geom = f.get("geometry", {})
+        coords = geom.get("coordinates")
+        if geom.get("type") == "Point" and coords:
+            cavity_coords.append((coords[1], coords[0]))  # (lat, lon)
+    cavity_lat = np.array([c[0] for c in cavity_coords])
+    cavity_lon = np.array([c[1] for c in cavity_coords])
+    print(f"add_cavites_features: {len(cavity_coords)} point cavity feature(s) loaded from {path}")
+
+    tree = cKDTree(np.column_stack([cavity_lon, cavity_lat]))
+    coarse_radius_deg = (max_distance_km / 111.0) * 1.5  # see add_groundwater_features for rationale
+
+    station_lon = nodes_df["longitude"].values
+    station_lat = nodes_df["latitude"].values
+    candidate_lists = tree.query_ball_point(np.column_stack([station_lon, station_lat]), r=coarse_radius_deg)
+
+    min_dists, counts = [], []
+    for i, candidates in enumerate(candidate_lists):
+        if not candidates:
+            min_dists.append(float("nan"))
+            counts.append(0)
+            continue
+        cand_idx = np.array(candidates)
+        dist = _haversine_km_vec(station_lat[i], station_lon[i], cavity_lat[cand_idx], cavity_lon[cand_idx])
+        within = dist <= max_distance_km
+        min_dists.append(float(dist.min()) if within.any() else float("nan"))
+        counts.append(int(within.sum()))
+
+    out = nodes_df.copy()
+    out["distance_to_nearest_cavity_km"] = min_dists
+    out[f"n_cavities_within_{int(max_distance_km)}km"] = counts
+    return out
+
+
+def add_ndvi_features(nodes_df: pd.DataFrame, ndvi_path: Path) -> pd.DataFrame:
+    """
+    ESA WorldCover Sentinel-2 NDVI yearly percentile composite
+    (p10/p50/p90) per station, from scripts/fetch_worldcover_ndvi.py.
+
+    NOTE: raw values are VITO's scaled digital numbers, not independently
+    confirmed to be literal -1..1 NDVI (no access to their exact scale/
+    offset convention in this environment) -- internally consistent
+    (p90 > p50 > p10 holds for every real station checked), so the
+    relative signal is trustworthy even if the absolute units aren't
+    pinned down. Doesn't block use as a model feature: build_pyg_graph
+    z-scores every feature anyway, which is invariant to an unknown
+    linear scaling. Only matters if literal NDVI units are needed later
+    (e.g. for plotting against a textbook NDVI range) -- resolve the
+    scale/offset from VITO's product documentation before then.
+
+    Same coverage limitation as add_catchment_features/add_landcover_
+    features: exact station_code match, real gauges only for now.
+    """
+    ndvi_df = pd.read_csv(ndvi_path)[["station_code", "ndvi_p10", "ndvi_p50", "ndvi_p90"]]
+    return nodes_df.merge(ndvi_df, on="station_code", how="left")
+
+
+def add_landcover_features(nodes_df: pd.DataFrame, landcover_path: Path) -> pd.DataFrame:
+    """
+    ESA WorldCover landcover class per station, one-hot encoded --
+    NOT left as the raw integer class code (10=Tree cover, 50=Built-up,
+    etc.). Landcover is a nominal category, not an ordered quantity;
+    leaving it as a raw integer would let build_pyg_graph's numeric
+    auto-detection z-score it as if "Built-up" (50) were meaningfully
+    "more" than "Tree cover" (10) in some continuous sense, which isn't
+    physically true -- the same class of error as the structural-column
+    leakage bug found earlier in this project, just subtler since this
+    one IS meant to be a real model input, not excluded metadata.
+
+    From scripts/fetch_worldcover_landcover.py -- currently only
+    covers real gauge stations (exact station_code match, same
+    limitation as add_catchment_features): non-gauge reach graph nodes
+    get NaN/all-zero here until that script is extended to sample the
+    full node set, not just the 27 gauges.
+    """
+    landcover_df = pd.read_csv(landcover_path)[["station_code", "landcover_class_name"]]
+    dummies = pd.get_dummies(landcover_df["landcover_class_name"], prefix="landcover")
+    landcover_wide = pd.concat([landcover_df[["station_code"]], dummies], axis=1)
+    return nodes_df.merge(landcover_wide, on="station_code", how="left")
+
+
 def add_catchment_features(nodes_df: pd.DataFrame, catchment_path: Path) -> pd.DataFrame:
     """
     Catchment (drainage basin) area per station, in km², via
@@ -344,6 +519,10 @@ def build_node_features(
     safran_path: Optional[Path] = None,
     hydrometric_path: Optional[Path] = None,
     catchment_path: Optional[Path] = None,
+    landcover_path: Optional[Path] = None,
+    ndvi_path: Optional[Path] = None,
+    bdcavites_path: Optional[Path] = None,
+    bdcharm_path: Optional[Path] = None,
     prefix_map=None,
     base_nodes_df: Optional[pd.DataFrame] = None,
     date_range: Optional[Tuple[str, str]] = None,
@@ -397,6 +576,10 @@ def build_node_features(
     for label, path, fn in [
         ("idpr", idpr_path, add_idpr_features),
         ("catchment_area", catchment_path, add_catchment_features),
+        ("landcover", landcover_path, add_landcover_features),
+        ("ndvi", ndvi_path, add_ndvi_features),
+        ("cavites", bdcavites_path, add_cavites_features),
+        ("geology", bdcharm_path, add_geology_features),
         ("ades_groundwater", ades_path, add_groundwater_features),
         ("safran_climate", safran_path, add_safran_features),
         ("hydrometric_targets", hydrometric_path, add_hydrometric_target_stats),

src/graph/physics_losses.py

@@ -33,9 +33,28 @@ except ImportError:
     _HAS_TORCH = False
 
 
-def _mse(residual) -> float:
-    """Mean squared residual -- works identically on numpy or torch."""
-    return (residual ** 2).mean()
+def _mse(residual):
+    """
+    Mean squared residual, NaN-masked -- works identically on numpy or
+    torch. Real ground-truth Q is naturally, heavily NaN (only real
+    gauges with real observations ever have a value; confirmed against
+    real data: ~93.5% NaN for the reach graph's discharge tensor) --
+    without masking, ANY single NaN anywhere in the residual poisons
+    the entire mean to NaN, which isn't a rare edge case for this data,
+    it's the normal shape of it. Returns NaN only if truly nothing
+    usable exists (every entry NaN), which is a real "no data" signal
+    worth surfacing, not silently averaging to 0 and implying perfect
+    physics satisfaction when there was actually no evidence either way.
+    """
+    if _HAS_TORCH and isinstance(residual, torch.Tensor):
+        mask = ~torch.isnan(residual)
+        if not mask.any():
+            return residual.sum() * float("nan")
+        return (residual[mask] ** 2).mean()
+    mask = ~np.isnan(residual)
+    if not mask.any():
+        return np.nan
+    return (residual[mask] ** 2).mean()
 
 
 # ---------------------------------------------------------------------------

src/test_build_graph.py

@@ -30,11 +30,15 @@ try:
     from .graph.node_features import build_node_features
     from .graph.build_graph import build_surface_edges, build_pyg_graph, build_pyg_graphs_per_basin
     from .data.river_centerline import load_centerline, snap_gauges_to_centerline
+    from .graph.physics_losses import build_confluence_index, build_braid_index
 except ImportError:
     sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
     from src.graph.node_features import build_node_features
     from src.graph.build_graph import build_surface_edges, build_pyg_graph, build_pyg_graphs_per_basin
     from src.data.river_centerline import load_centerline, snap_gauges_to_centerline
+    from src.graph.physics_losses import build_confluence_index, build_braid_index
+
+import pandas as pd
 
 
 def _to_numpy(t):
@@ -265,13 +269,160 @@ def run_checks(data_root: Path, basin_file_names=None) -> bool:
     return all_ok
 
 
+def run_reach_graph_checks(data_root: Path, basin_file_names=None) -> bool:
+    """
+    Validates the reach graph pipeline (scripts/build_reach_graphs.py ->
+    scripts/enrich_reach_graph.py -> build_pyg_graph), separately from
+    run_checks' coverage of the original single-chain pipeline -- the
+    two build nodes_df/edges_df in genuinely different ways (real
+    branching topology + splits/rejoins vs. a single ordered chain), so
+    keeping their checks in separate functions keeps each readable
+    rather than threading conditionals through one long function.
+
+    Specifically regression-tests the three bugs found and fixed in the
+    previous session: structural columns (is_gauged/is_confluence/etc.)
+    leaking into model features, targets attaching to non-gauge nodes,
+    and edge_attr breaking on the real edges schema's extra string
+    columns (toponym, cleabs).
+
+    Gracefully returns True (not a failure) if the reach graph hasn't
+    been built/enriched yet -- this is meant to add coverage once that
+    pipeline is in use, not force it to exist.
+    """
+    basin_file_names = basin_file_names or {0: "eure", 1: "risle"}
+    all_ok = True
+    graph_dir = data_root / "reach_graph"
+
+    print()
+    print("=" * 60)
+    print("REACH GRAPH CHECKS")
+    print("=" * 60)
+
+    if not graph_dir.exists():
+        print(f"  No {graph_dir} found -- skipping (run scripts/build_reach_graphs.py "
+              f"and scripts/enrich_reach_graph.py first for this coverage). Not a failure.")
+        return True
+
+    any_basin_found = False
+    for basin_id, file_key in basin_file_names.items():
+        enriched_path = graph_dir / f"{file_key}_nodes_enriched.csv"
+        edges_path = graph_dir / f"{file_key}_edges.csv"
+        if not enriched_path.exists() or not edges_path.exists():
+            print(f"  basin {basin_id} ({file_key}): missing enriched nodes/edges CSV, skipped")
+            continue
+        any_basin_found = True
+
+        nodes_df = pd.read_csv(enriched_path)
+        edges_df = pd.read_csv(edges_path)
+        print()
+        print(f"--- basin {basin_id} ({file_key}): {len(nodes_df)} nodes, {len(edges_df)} edges ---")
+
+        structural_cols = ["is_gauged", "is_confluence", "is_split_point", "is_rejoin_point"]
+        for col in structural_cols:
+            all_ok &= _check(f"  '{col}' column present on nodes_df", col in nodes_df.columns)
+
+        n_gauged = int(nodes_df["is_gauged"].sum()) if "is_gauged" in nodes_df.columns else 0
+        n_confluence = int(nodes_df["is_confluence"].sum()) if "is_confluence" in nodes_df.columns else 0
+        n_split = int(nodes_df["is_split_point"].sum()) if "is_split_point" in nodes_df.columns else 0
+        n_rejoin = int(nodes_df["is_rejoin_point"].sum()) if "is_rejoin_point" in nodes_df.columns else 0
+        print(f"  {n_gauged} gauged, {n_confluence} confluences, {n_split} splits, {n_rejoin} rejoins")
+
+        data = build_pyg_graph(nodes_df, edges_df)
+        x_np = _to_numpy(data.x)
+        all_ok &= _check("  x has no NaN", not np.isnan(x_np).any())
+        all_ok &= _check("  x has no Inf", not np.isinf(x_np).any())
+        all_ok &= _check("  x row count matches node count", x_np.shape[0] == len(nodes_df),
+                          f"{x_np.shape[0]} vs {len(nodes_df)}")
+
+        # REGRESSION: structural columns must never leak into model features
+        leaked = [f for f in data.feature_names
+                  if f in ("is_gauged", "is_confluence", "is_split_point", "is_rejoin_point",
+                            "snap_distance_km", "braid_id")]
+        all_ok &= _check("  no structural columns leaked into feature_names", not leaked,
+                          f"leaked: {leaked}")
+
+        # REGRESSION: structural columns still accessible as their own Data attributes
+        for col in structural_cols:
+            all_ok &= _check(f"  data.{col} attribute present", hasattr(data, col))
+
+        # REGRESSION: targets only ever attach to gauged nodes, never confluences/virtual nodes
+        if hasattr(data, "y") and hasattr(data, "is_gauged"):
+            y_np = _to_numpy(data.y)
+            is_gauged_np = _to_numpy(data.is_gauged).astype(bool)
+            has_real_target = ~np.isnan(y_np).all(axis=1)
+            mislabeled = has_real_target & ~is_gauged_np
+            all_ok &= _check("  no target values on non-gauged nodes", not mislabeled.any(),
+                              f"{int(mislabeled.sum())} mislabeled node(s)")
+            all_ok &= _check("  target coverage matches gauge count or less",
+                              int(has_real_target.sum()) <= n_gauged,
+                              f"{int(has_real_target.sum())} with targets vs {n_gauged} gauged")
+
+        # REGRESSION: edge_attr stays 3 columns despite extra edge metadata (toponym, cleabs)
+        edge_attr_np = _to_numpy(data.edge_attr)
+        all_ok &= _check("  edge_attr has exactly 3 columns despite extra edge metadata",
+                          edge_attr_np.shape[1] == 3, f"got {edge_attr_np.shape[1]} columns")
+
+        # Physics loss index builders (physics_losses.py) sanity-checked against this same data
+        conf_idx = build_confluence_index(nodes_df, edges_df)
+        braid_idx = build_braid_index(nodes_df)
+        all_ok &= _check("  confluence_index count matches is_confluence sum",
+                          len(conf_idx) == n_confluence, f"{len(conf_idx)} vs {n_confluence}")
+        if conf_idx:
+            max_idx = max(max(upstream) for _, upstream in conf_idx)
+            all_ok &= _check("  confluence_index indices within node bounds",
+                              0 <= max_idx < len(nodes_df))
+            min_upstream = min(len(upstream) for _, upstream in conf_idx)
+            all_ok &= _check("  every confluence has >= 2 upstream branches", min_upstream >= 2)
+        if braid_idx:
+            max_braid_idx = max(max(pair) for pair in braid_idx)
+            all_ok &= _check("  braid_index indices within node bounds",
+                              0 <= max_braid_idx < len(nodes_df))
+        all_ok &= _check("  braid_index count matches is_rejoin_point sum",
+                          len(braid_idx) == n_rejoin, f"{len(braid_idx)} vs {n_rejoin}")
+
+        n_climate = int(nodes_df["climate_precip_mm"].notna().sum()) if "climate_precip_mm" in nodes_df.columns else 0
+        n_catchment = int(nodes_df["catchment_area_km2"].notna().sum()) if "catchment_area_km2" in nodes_df.columns else 0
+        n_idpr = int(nodes_df["idpr_value"].notna().sum()) if "idpr_value" in nodes_df.columns else 0
+        print(f"  coverage: climate {n_climate}/{len(nodes_df)}, idpr {n_idpr}/{len(nodes_df)}, "
+              f"catchment_area (Hub'Eau, gauges only) {n_catchment}/{len(nodes_df)}")
+        all_ok &= _check("  idpr present on the enriched table (was missing in a real run once -- "
+                          "check the idpr_path passed to enrich_reach_graph.py if this fails)",
+                          "idpr_value" in nodes_df.columns)
+
+        if "cumulative_catchment_area_km2" in nodes_df.columns:
+            n_cumulative = int(nodes_df["cumulative_catchment_area_km2"].notna().sum())
+            print(f"  cumulative_catchment_area_km2 (BD TOPO, graph-wide) coverage: "
+                  f"{n_cumulative}/{len(nodes_df)}")
+            all_ok &= _check("  cumulative_catchment_area_km2 covers meaningfully more than "
+                              "the Hub'Eau-only catchment_area_km2",
+                              n_cumulative > n_catchment,
+                              f"{n_cumulative} vs {n_catchment} -- if this fails, "
+                              f"scripts/compute_cumulative_catchment.py likely needs a re-run")
+        else:
+            print("  cumulative_catchment_area_km2 not present -- run "
+                  "scripts/compute_cumulative_catchment.py for graph-wide catchment coverage "
+                  "(known gap otherwise: only the ~8 real gauges Hub'Eau publishes it for).")
+
+    if not any_basin_found:
+        print("  No enriched reach graph files found for any basin -- run "
+              "scripts/build_reach_graphs.py then scripts/enrich_reach_graph.py first.")
+        return True
+
+    print()
+    print("=" * 60)
+    print("REACH GRAPH RESULT:", "ALL CHECKS PASSED" if all_ok else "SOME CHECKS FAILED — see above")
+    print("=" * 60)
+    return all_ok
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(description="Test build_graph.py / node_features.py against real data")
     parser.add_argument("--data-root", type=Path, default=Path("datasets"))
     args = parser.parse_args()
 
-    ok = run_checks(args.data_root)
-    sys.exit(0 if ok else 1)
+    ok_original = run_checks(args.data_root)
+    ok_reach_graph = run_reach_graph_checks(args.data_root)
+    sys.exit(0 if (ok_original and ok_reach_graph) else 1)
 
 
 if __name__ == "__main__":