""" Download BD TOPO hydrography (troncon_hydrographique, surface_hydrographique) for the Eure/Risle area via IGN's Geoplateforme WFS, instead of pulling the whole national dataset. VERIFICATION NOTE: I confirmed the WFS endpoint itself is real and live (https://data.geopf.fr/wfs, WFS 2.0.0) and that a "BDTOPO_V3" namespace exists on it, by fetching GetCapabilities directly. I could NOT confirm the exact typeName for the hydrography layers within it — the full capabilities document is huge (thousands of layers across all IGN products) and got truncated before reaching them. DEFAULT_TYPENAMES below is my best guess based on IGN's usual naming convention, NOT a verified value. This script's first step (check_typenames) tests each name with a tiny query before attempting the real extraction, and tells you exactly how to find the right name yourself if my guess is wrong. Usage: python download_bdtopo_hydro.py --check # verify typeNames first python download_bdtopo_hydro.py # then run the real extraction """ import argparse import sys from pathlib import Path import requests import pandas as pd WFS_URL = "https://data.geopf.fr/wfs" # Best-guess typeNames -- see VERIFICATION NOTE above. Override with # --troncon-typename / --surface-typename / --catchment-typename if wrong. # troncon/surface confirmed correct against the real service (see # CONFIRMED note below); catchment is still an unverified guess. DEFAULT_TRONCON_TYPENAME = "BDTOPO_V3:troncon_hydrographique" DEFAULT_SURFACE_TYPENAME = "BDTOPO_V3:surface_hydrographique" DEFAULT_CATCHMENT_TYPENAME = "BDTOPO_V3:bassin_versant_topographique" # CONFIRMED: both troncon_hydrographique and surface_hydrographique # typeNames work against the live service (verified in this project's # actual run: 30045 tronçons fetched successfully). bassin_versant_ # topographique has NOT been verified the same way -- run --check # before trusting it. # Bounding box around the Eure/Risle stations, in WGS84 (matches the # lon/lat range in your real station_list.csv, with a small margin). BBOX = (0.3, 48.3, 1.7, 49.5) # (min_lon, min_lat, max_lon, max_lat) def find_hydro_typenames(keyword: str = "hydrographique") -> list: """ Fetch GetCapabilities and search for FeatureType entries containing `keyword`. This is the fallback if DEFAULT_*_TYPENAME turns out to be wrong -- run this to get the real list. """ print(f"Fetching WFS GetCapabilities and searching for '{keyword}'...") resp = requests.get(WFS_URL, params={ "SERVICE": "WFS", "VERSION": "2.0.0", "REQUEST": "GetCapabilities", }, timeout=60) resp.raise_for_status() import re names = re.findall(r"([^<]*)", resp.text) matches = [n for n in names if keyword.lower() in n.lower()] print(f"Found {len(matches)} matching layer(s):") for m in matches: print(f" {m}") return matches def check_typename(type_name: str) -> bool: """Tiny test query (count=1, no bbox) to confirm a typeName is valid before running the real extraction against it.""" params = { "SERVICE": "WFS", "VERSION": "2.0.0", "REQUEST": "GetFeature", "TYPENAMES": type_name, "COUNT": 1, "OUTPUTFORMAT": "application/json", } resp = requests.get(WFS_URL, params=params, timeout=30) ok, error_detail = _parse_wfs_response(resp) status = "OK" if ok else f"FAILED" print(f" {type_name}: {status}") if not ok: print(f" {error_detail}") return ok def _parse_wfs_response(resp: "requests.Response") -> "tuple[bool, str]": """ Check whether a WFS response is a real FeatureCollection or an error (WFS ExceptionReport, or a non-JSON/HTML error page). Returns (is_valid, detail) -- detail is empty on success, otherwise a human-readable explanation of what actually came back, so a failure is never silently swallowed into "0 features". """ if resp.status_code != 200: return False, f"HTTP {resp.status_code}: {resp.text[:400]}" try: data = resp.json() except ValueError: # Not JSON at all -- almost always means an XML ExceptionReport # or an HTML error page came back instead. 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, "" # Some WFS error responses come back as JSON but without a # FeatureCollection shape (e.g. {"exceptionText": [...]}) return False, f"Response was JSON but not a FeatureCollection: {str(data)[:400]}" def fetch_features(type_name: str, bbox: tuple, out_path: Path, srs: str = "EPSG:4326") -> None: """ GetFeature request over the given bbox, paginated via startIndex since a national layer clipped to a small bbox can still exceed the server's per-request feature limit. AXIS ORDER: when BBOX specifies its CRS via the URN form (urn:ogc:def:crs:EPSG::4326), the OGC spec requires lat,lon axis order -- NOT the traditional lon,lat most tools use. Getting this backwards doesn't error, it just silently matches zero real-world features (a well-known WFS 2.0 gotcha). Rather than guess, this tries lon,lat first and automatically retries with lat,lon swapped if the first attempt comes back empty, so a silent axis-order bug can't hide as "no data in this area" again. """ 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"), ] all_features = [] for label, bbox_param in orderings: print(f" trying axis order {label}...") all_features = _fetch_paginated(type_name, bbox_param) if all_features: print(f" -> {label} worked ({len(all_features)} feature(s)). " f"Use this axis order for future queries against this endpoint.") break print(f" -> {label} returned 0 features.") import json geojson = {"type": "FeatureCollection", "features": all_features} out_path.write_text(json.dumps(geojson)) if all_features: print(f"Saved {len(all_features)} feature(s) to {out_path}") else: print(f"Saved an EMPTY file to {out_path} -- both axis orders returned " f"nothing real. The typeName is valid (confirmed by --check), so this " f"suggests the bbox itself doesn't overlap any features in this layer, " f"or another parameter is off. Try a much larger bbox (e.g. all of " f"France: -5,41,10,51) as a sanity check.") def _fetch_paginated(type_name: str, bbox_param: str) -> list: """One bbox ordering's worth of paginated GetFeature calls.""" all_features = [] start_index = 0 page_size = 1000 while True: params = { "SERVICE": "WFS", "VERSION": "2.0.0", "REQUEST": "GetFeature", "TYPENAMES": type_name, "BBOX": bbox_param, "OUTPUTFORMAT": "application/json", "COUNT": page_size, "STARTINDEX": start_index, } resp = requests.get(WFS_URL, params=params, timeout=60) is_valid, error_detail = _parse_wfs_response(resp) if not is_valid: print(f" ERROR at startIndex={start_index}: {error_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 main() -> None: parser = argparse.ArgumentParser(description="Download BD TOPO hydrography for Eure/Risle") parser.add_argument("--check", action="store_true", help="Only verify typeNames, don't download anything") parser.add_argument("--troncon-typename", default=DEFAULT_TRONCON_TYPENAME) parser.add_argument("--surface-typename", default=DEFAULT_SURFACE_TYPENAME) parser.add_argument("--catchment-typename", default=DEFAULT_CATCHMENT_TYPENAME) parser.add_argument("--skip-catchments", action="store_true", help="Skip bassin_versant_topographique (untested typeName)") parser.add_argument("--output-dir", type=Path, default=Path("datasets/bdtopo_hydro")) args = parser.parse_args() if args.check: print("=" * 70) print("Testing default typeNames...") print("=" * 70) ok_troncon = check_typename(args.troncon_typename) ok_surface = check_typename(args.surface_typename) ok_catchment = args.skip_catchments or check_typename(args.catchment_typename) if not (ok_troncon and ok_surface and ok_catchment): print() print("At least one default typeName failed. Searching capabilities " "for the real name(s) instead:") find_hydro_typenames("hydrographique") find_hydro_typenames("bassin_versant") print() print("Re-run with --troncon-typename / --surface-typename / " "--catchment-typename set to whatever the search above found, " "then --check again to confirm.") else: print() print("All typeNames OK. Re-run without --check to download the real data.") return args.output_dir.mkdir(parents=True, exist_ok=True) print("Fetching troncon_hydrographique (river centerline geometry + attributes)...") fetch_features(args.troncon_typename, BBOX, args.output_dir / "troncon_hydrographique.geojson") print() print("Fetching surface_hydrographique (includes karst/Nature classification)...") fetch_features(args.surface_typename, BBOX, args.output_dir / "surface_hydrographique.geojson") if not args.skip_catchments: print() print("Fetching bassin_versant_topographique (catchment polygons)...") print("NOTE: this typeName is unverified -- if it fails, run --check first " "to find the real name, or pass --skip-catchments to skip it.") fetch_features(args.catchment_typename, BBOX, args.output_dir / "bassin_versant_topographique.geojson") print() print("Done. Load with: BDTopoHydroLoader(data_path=...)") print("To check for the karst-flagged reach near Grosley-sur-Risle/Ajou, use " "loader.check_karst_near_point(lat, lon) -- that's the check that would " "confirm or deny the bétoire location.") if __name__ == "__main__": main()