Spaces:
Running on Zero
Running on Zero
File size: 27,834 Bytes
a74054f d6b0b7a a74054f d6b0b7a a74054f 4bb7968 a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f 4bb7968 a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f d6b0b7a a74054f 4bb7968 a74054f d6b0b7a a74054f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 | """
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 |