""" Node feature construction for the surface-connectivity graph. Builds a per-station feature table by pulling from every available loader, using each loader's own spatial interpolation/aggregation so a station gets a value even when it doesn't sit exactly on a raw data point (between IDPR points, outside a well's radius, off an ERA5 grid cell, etc). Loaders used, each skipped gracefully with a clear message if its data files aren't present (same pattern as generate_plots.py) so one missing dataset never stops the others: - StationElevationsLoader: elevation_m, latitude, longitude (base table) - IDPRLoader: infiltration/runoff tendency, via nearest spatial point - ADESLoader: groundwater level/depth, via the loader's own aggregate_to_stations (radius-averaged, not a graph edge — see the earlier decision not to use wells for subsurface edges; this is the "use as a station covariate" path instead) - SAFRANLoader: ERA5 climate variables, via the loader's own nearest/linear interpolation to each station point - HydrometricLoader: discharge/water-level summary stats IMPORTANT — inputs vs. targets: elevation/IDPR/groundwater/climate are genuine INPUT covariates, and filling them in via spatial interpolation for any station is expected, ordinary feature engineering. Discharge and water-level are the model's TARGET variables: they're reported here as plain summary stats per gauge (prefixed `target_`) and are left NaN for any station without hydrometric data — never interpolated or guessed. Interpolating a target would be label leakage, not feature engineering; don't feed the `target_*` columns back in as model inputs. """ from dataclasses import dataclass, field from pathlib import Path from typing import List, Optional, Tuple import numpy as np import pandas as pd from ..data.river_graph import assign_basin_id from ..data.loaders.idpr import IDPRLoader from ..data.loaders.ades import ADESLoader from ..data.loaders.station_elevations import StationElevationsLoader from ..data.loaders.catchment import CatchmentAreaLoader @dataclass class NodeFeatureReport: """What actually got included, so you can tell at a glance whether a feature block is real or silently missing before trusting the table.""" n_stations: int included: List[str] = field(default_factory=list) skipped: List[str] = field(default_factory=list) def __str__(self) -> str: lines = [f"Node features for {self.n_stations} stations:"] for name in self.included: lines.append(f" \u2713 {name}") for name in self.skipped: lines.append(f" \u2014 {name} (skipped)") return "\n".join(lines) def build_base_station_table(elevations_df: pd.DataFrame, prefix_map=None) -> pd.DataFrame: """Base node table: station_code, basin_id, latitude, longitude, elevation_m.""" df = elevations_df.dropna(subset=["latitude", "longitude"]).copy() df["basin_id"] = df["station_code"].apply(lambda c: assign_basin_id(c, prefix_map)) unmatched = df[df["basin_id"].isna()] if not unmatched.empty: print(f"build_base_station_table: {len(unmatched)} station(s) matched no basin " f"prefix and are excluded: {unmatched['station_code'].tolist()}") df = df.dropna(subset=["basin_id"]).copy() df["basin_id"] = df["basin_id"].astype(int) return df[["station_code", "basin_id", "latitude", "longitude", "elevation_m"]].reset_index(drop=True) def _haversine_km_vec(lat1: float, lon1: float, lat2, lon2): """Vectorized haversine: one (lat1, lon1) point against arrays lat2/lon2. Used for the exact narrow-phase distance check after a KD-tree coarse prefilter (see add_groundwater_features).""" R = 6371.0 lat1r, lon1r = np.radians(lat1), np.radians(lon1) lat2r, lon2r = np.radians(lat2), np.radians(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)) def add_idpr_features(nodes_df: pd.DataFrame, idpr_path: Path) -> pd.DataFrame: """ IDPR value per station. Prefers an exact join when the IDPR data already has a per-station identifier column (e.g. it was extracted at each gauge's coordinates ahead of time, as opposed to being a raw spatial point cloud/raster export) — this is both simpler and more accurate than nearest-neighbor search, since it isn't a real interpolation, it's the actual value at that exact station. Falls back to nearest-spatial-point search (KD-tree) for a generic IDPR point cloud with no station identifiers. NOTE on the fallback path: it does NOT reproject coordinates — if IDPR's coordinates are in a different CRS than the station lat/lon, results will be wrong. Reproject one side to match the other first if needed. """ from scipy.spatial import cKDTree loader = IDPRLoader(data_path=idpr_path) idpr_df = loader.load() value_col = loader._find_value_column(idpr_df) id_col = next((c for c in ["station_id", "station_code", "code_station"] if c in idpr_df.columns), None) if id_col is not None and set(nodes_df["station_code"]).issubset(set(idpr_df[id_col])): merged = nodes_df.merge( idpr_df[[id_col, value_col]].rename(columns={id_col: "station_code", value_col: "idpr_value"}), on="station_code", how="left", ) merged["idpr_nearest_point_distance"] = 0.0 # exact match, not interpolated return merged x_col = next((c for c in ["lon", "longitude", "x"] if c in idpr_df.columns), None) y_col = next((c for c in ["lat", "latitude", "y"] if c in idpr_df.columns), None) if not x_col or not y_col: raise ValueError("Could not detect coordinate columns in IDPR data.") tree = cKDTree(idpr_df[[x_col, y_col]].values) query_coords = nodes_df[["longitude", "latitude"]].values dist, idx = tree.query(query_coords) out = nodes_df.copy() out["idpr_value"] = idpr_df[value_col].values[idx] out["idpr_nearest_point_distance"] = dist # in the IDPR file's own coordinate units 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 CatchmentAreaLoader (see download_catchment_area.py for how this is sourced from Hub'Eau's own referentiel — no interpolation involved, it's the actual reported value or nothing). NOTE: this is genuinely missing (not degraded/approximated) for roughly 40% of stations in this project's data — Hub'Eau doesn't publish surface_bv for every site (observer stations, some partner- network sites, etc). That's real and expected; the missingness flag generated by build_pyg_graph's `add_missingness_flags` will capture it correctly as long as this column has NaNs, which it does by design — don't fill it in here. """ loader = CatchmentAreaLoader(data_path=catchment_path) catchment_df = loader.load()[["station_code", "catchment_area_km2"]] return nodes_df.merge(catchment_df, on="station_code", how="left") def add_groundwater_features( nodes_df: pd.DataFrame, ades_path: Path, max_distance_km: float = 20.0, date_range: Optional[Tuple[str, str]] = None, ) -> pd.DataFrame: """ Radius-averaged groundwater level/depth per station. Does NOT use ADESLoader.aggregate_to_stations -- that method loops per station and does a full haversine .apply() over the ENTIRE groundwater dataframe for each one (O(n_stations * n_readings)). At 27 stations against ~272k readings that's slow but tolerable; at the reach graph's ~4,500 nodes it's over a billion row-wise Python calls, not a proportionally-worse runtime but an unusable one. Instead: reduce to each well's most recent reading first (collapses ~272k readings to ~100 wells -- we only need each station's latest aggregate anyway, same as the old code kept via `.sort_values("date").drop_duplicates(...)` after the fact), then a KD-tree coarse prefilter narrows each station to a handful of nearby candidate wells, with exact haversine distance computed only on that small candidate set to get correct radius membership and correct averages -- fast without trading away correctness for it. date_range: if given, (start, end) date strings (inclusive) -- a well's readings are filtered to this window BEFORE picking its "most recent" one, so a well with no reading in the window is correctly excluded from that call rather than falling back to some older reading from outside the period the rest of the graph's features are being built for (real wells in this project report on wildly different schedules -- see the ground-truth investigation that originally found this "most recent" logic needed care). """ from scipy.spatial import cKDTree loader = ADESLoader(data_path=ades_path) gw_df = loader.load() if date_range is not None: start, end = date_range gw_df = gw_df[(gw_df["date"] >= start) & (gw_df["date"] <= end)] latest_per_well = gw_df.sort_values("date").drop_duplicates("code_bss", keep="last").reset_index(drop=True) if latest_per_well.empty: out = nodes_df.copy() out["avg_groundwater_level_m"] = np.nan out["avg_groundwater_depth_m"] = np.nan out["n_nearby_wells"] = 0 return out well_lat = latest_per_well["lat"].values well_lon = latest_per_well["lon"].values tree = cKDTree(np.column_stack([well_lon, well_lat])) # Coarse degree-radius, deliberately oversized (1.5x safety factor): # degrees-to-km varies with latitude and direction (lon degrees are # worth fewer km than lat degrees away from the equator), so this # only needs to be a safe upper bound, not an accurate radius -- # the exact haversine pass afterward is what determines real # membership and distances. coarse_radius_deg = (max_distance_km / 111.0) * 1.5 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) avg_levels, avg_depths, n_wells_list = [], [], [] for i, candidates in enumerate(candidate_lists): if not candidates: avg_levels.append(np.nan); avg_depths.append(np.nan); n_wells_list.append(0) continue cand_idx = np.array(candidates) dist = _haversine_km_vec(station_lat[i], station_lon[i], well_lat[cand_idx], well_lon[cand_idx]) within = dist <= max_distance_km if not within.any(): avg_levels.append(np.nan); avg_depths.append(np.nan); n_wells_list.append(0) continue sel = latest_per_well.iloc[cand_idx[within]] avg_levels.append(sel["groundwater_level_m"].mean()) avg_depths.append(sel["groundwater_depth_m"].mean() if "groundwater_depth_m" in sel.columns else np.nan) n_wells_list.append(int(within.sum())) out = nodes_df.copy() out["avg_groundwater_level_m"] = avg_levels out["avg_groundwater_depth_m"] = avg_depths out["n_nearby_wells"] = n_wells_list return out def add_safran_features( nodes_df: pd.DataFrame, safran_path: Path, date_range: Optional[Tuple[str, str]] = None, ) -> pd.DataFrame: """ Climate summary stats per station, interpolated from ERA5 via SAFRANLoader's own station-point interpolation. Requires `xarray` and actual era5_*.nc files under safran_path; returns nodes_df unchanged (with a clear message) if either is missing. date_range: if given, (start, end) date strings (inclusive) -- filters ERA5 timesteps to this window BEFORE aggregating (mean for temp/wind/solar, sum for precip/evap/snow/runoff), so e.g. climate_precip_mm becomes "total precip over the window" rather than "total precip over the full 1960-2026 record", which wouldn't be comparable to a discharge target computed over a specific training period. """ try: from ..data.loaders.safran import SAFRANLoader except ImportError as e: print(f"add_safran_features: skipped ({e})") return nodes_df try: import xarray # noqa: F401 except ImportError: print("add_safran_features: skipped (xarray not installed)") return nodes_df station_coords = nodes_df.rename(columns={"latitude": "lat", "longitude": "lon"})[ ["station_code", "lat", "lon"] ] loader = SAFRANLoader(data_path=safran_path, station_coords=station_coords) try: df = loader.load() except FileNotFoundError as e: print(f"add_safran_features: skipped ({e})") return nodes_df df = loader.convert_units(df) if date_range is not None: start, end = date_range df = df[(df["date"] >= start) & (df["date"] <= end)] if df.empty: print(f"add_safran_features: skipped (no ERA5 timesteps fall within {date_range})") return nodes_df agg_spec = {c: "mean" for c in ["temp_C", "wind_speed_ms", "solar_Wm2"] if c in df.columns} for c in ["precip_mm", "evap_mm", "snow_mm", "runoff_mm"]: if c in df.columns: agg_spec[c] = "sum" if not agg_spec: print("add_safran_features: skipped (no recognized variables after unit conversion)") return nodes_df summary = df.groupby("station_code").agg(agg_spec) summary = summary.add_prefix("climate_") return nodes_df.merge(summary, on="station_code", how="left") def add_hydrometric_target_stats( nodes_df: pd.DataFrame, hydrometric_path: Path, date_range: Optional[Tuple[str, str]] = None, ) -> pd.DataFrame: """ Discharge/water-level summary stats per gauge, prefixed `target_`. Left-joined so ungauged stations get NaN rather than being dropped — that NaN is meaningful (no observations), not a gap to fill in. date_range: if given, (start, end) date strings (inclusive) -- filters observations to this window before computing mean/std/count, so the target reflects the same training period the input features (climate, groundwater) were built for, not each source's own full and differently-shaped historical record. """ from ..data.loaders.hydrometric import HydrometricLoader loader = HydrometricLoader(data_path=hydrometric_path) try: df = loader.load() except FileNotFoundError as e: print(f"add_hydrometric_target_stats: skipped ({e})") return nodes_df if date_range is not None: start, end = date_range df = df[(df["date"] >= start) & (df["date"] <= end)] agg_spec = {} if "discharge_m3s" in df.columns: agg_spec["discharge_m3s"] = ["mean", "std", "count"] if "waterlevel_mm" in df.columns: agg_spec["waterlevel_mm"] = ["mean", "std", "count"] if not agg_spec: return nodes_df summary = df.groupby("station_code").agg(agg_spec) summary.columns = [f"target_{a}_{b}" for a, b in summary.columns] return nodes_df.merge(summary, on="station_code", how="left") def build_node_features( station_elevations_path: Optional[Path] = None, idpr_path: Optional[Path] = None, ades_path: Optional[Path] = None, 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, ) -> "tuple[pd.DataFrame, NodeFeatureReport]": """ Orchestrator: build the base station table, then layer on every available loader's features. Any path left as None (or pointing to missing files) is skipped with a message rather than raising. Two ways to get the base table: - `station_elevations_path` (original path): builds the 27-gauge table via StationElevationsLoader, as before. - `base_nodes_df` (for the reach graph): pass an already-built node table directly -- e.g. the ~4,500-node reach graph from scripts/build_reach_graphs.py (real gauges + real confluences + splits/rejoins + virtual infill nodes), which every add_*_features function already works against unchanged, since they only ever assumed [station_code, latitude, longitude, ...] columns, not a specific row count. Extra columns already on base_nodes_df (is_gauged, is_confluence, is_split_point, is_rejoin_point, braid_id, snap_distance_km) pass through every merge untouched. Exactly one of these two must be given. date_range: optional (start, end) date strings (inclusive), applied to every time-varying source (groundwater, climate, hydrometric targets) so all three describe the same period -- otherwise each source silently aggregates over its own full, differently-shaped historical record (ADES wells reporting from 1972-2026 on wildly different schedules, ERA5 spanning 1960-2026, hydrometric records with their own per-station date ranges), which makes "climate over the period" and "discharge over the period" describe different periods without anyone intending that. Static sources (IDPR, elevation, catchment area) are unaffected -- they don't vary in time. Returns: (nodes_df, report) """ if (station_elevations_path is None) == (base_nodes_df is None): raise ValueError("Provide exactly one of station_elevations_path or base_nodes_df.") if base_nodes_df is not None: nodes_df = base_nodes_df.copy() report = NodeFeatureReport(n_stations=len(nodes_df), included=["reach graph node table (base)"]) else: nodes_df = build_base_station_table( StationElevationsLoader(data_path=station_elevations_path).load(), prefix_map ) report = NodeFeatureReport(n_stations=len(nodes_df), included=["station_elevations (base table)"]) date_aware = {"ades_groundwater", "safran_climate", "hydrometric_targets"} 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), ]: if path is None or not Path(path).exists(): report.skipped.append(f"{label} (no path given or not found)") continue try: if date_range is not None and label in date_aware: nodes_df = fn(nodes_df, path, date_range=date_range) else: nodes_df = fn(nodes_df, path) report.included.append(label if date_range is None or label not in date_aware else f"{label} (filtered to {date_range[0]}..{date_range[1]})") except Exception as e: report.skipped.append(f"{label} (error: {e})") return nodes_df, report