""" Reach-based river network graph construction. Builds the graph from BD TOPO's own tronçon-to-tronçon connectivity (`lien_vers_noeud_hydrographique_ini` / `_fin`), not from inferring station order along a single extracted line. This is the NEXT_DOWN-equivalent approach: each tronçon is a directed edge between two real hydrographic node IDs, so confluences (multiple tronçons converging on one node) and branching fall directly out of the data rather than needing to be modeled separately. Flow direction, per tronçon, comes from `sens_de_l_ecoulement` where that field is usably populated, falling back to comparing the Z (altitude) coordinate at each tronçon's endpoints — higher Z is upstream — since BD TOPO geometry already carries real altitude at every vertex (see BDTopoHydroLoader.get_elevation_profile). `check_flow_direction_coverage` reports how often each source actually applies before you trust either one blindly; the `sens_de_l_ecoulement` value vocabulary hasn't been verified against a real full export, so this treats it as unconfirmed until checked. Node types in the final graph: - real gauges (snapped onto the nearest reach) - real confluences (any hydrographic node with in-degree >= 2) - virtual nodes, inserted along confluence-free stretches longer than `virtual_node_spacing_km` so "predict at any point" has reasonable resolution without a node at every tronçon vertex Targets (discharge, water level) are never assigned to confluence or virtual nodes — see node_features.py's inputs-vs-targets distinction. That principle doesn't change here; it matters more here, since most nodes in this graph will be ungauged by construction. """ from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Set, Tuple import math import re import numpy as np import pandas as pd import networkx as nx def _haversine_km(lat1, lon1, lat2, lon2) -> float: R = 6371.0 lat1, lon1, lat2, lon2 = map(math.radians, [lat1, lon1, lat2, lon2]) dlat, dlon = lat2 - lat1, lon2 - lon1 a = math.sin(dlat / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2 return R * 2 * math.asin(math.sqrt(a)) # River name filter per basin, gathered from BD TOPO catchment-polygon # toponyms surfaced earlier in this project (cross_check_catchments.py's # "bdtopo_catchment_name" output) — not a guess, these are real tributary # names confirmed to exist in this watershed's data. Extend if others turn # up once the broadened pull actually runs. TRIBUTARY_NAMES = { 0: [ # La Eure "Eure", "Livier", "Coinon", "Houdouenne", "Blaise", "Avre", "Voise", "Drouette", "Maltorne", "Vesgre", "Etang", "Autheuil", "Iton", ], 1: [ # La Risle "Risle", "Corru", "Finard", "Sommaire", "Bave", "Charentonne", "Croix Blanche", "Bec", "Echauds", "Vieux Bourg", ], } @dataclass class ReachGraphReport: n_troncons_matched: int n_hydro_nodes: int n_confluences: int n_raw_multi_inflow: int flow_direction_from_sens: int flow_direction_from_elevation: int flow_direction_ambiguous: int n_gauges_snapped: int n_virtual_nodes_inserted: int disconnected_components: int largest_component_troncons: int def __str__(self) -> str: total_dir = (self.flow_direction_from_sens + self.flow_direction_from_elevation + self.flow_direction_ambiguous) pct_sens = 100 * self.flow_direction_from_sens / total_dir if total_dir else 0 return ( f"Reach graph: {self.n_troncons_matched} tronçons matched, " f"{self.n_hydro_nodes} hydrographic nodes\n" f" confluences (different named river actually joining): {self.n_confluences}\n" f" ({self.n_raw_multi_inflow} nodes have in-degree >= 2 total -- the gap between " f"the two is same-river fine segmentation, not real branching)\n" f" flow direction: {self.flow_direction_from_sens} from sens_de_l_ecoulement " f"({pct_sens:.0f}%), {self.flow_direction_from_elevation} from elevation, " f"{self.flow_direction_ambiguous} ambiguous (dropped)\n" f" {self.n_gauges_snapped} real gauges snapped, " f"{self.n_virtual_nodes_inserted} virtual nodes inserted\n" f" topology: {self.disconnected_components} component(s), " f"largest has {self.largest_component_troncons} tronçons" + (" -- single connected network" if self.disconnected_components == 1 else " -- see disconnected_components; only the largest is used") ) def check_flow_direction_coverage(features: List[dict]) -> Dict[str, int]: """ Report how populated and how varied `sens_de_l_ecoulement` actually is in a real export, before any code decides to trust it. Run this first against the real data. """ from collections import Counter values = Counter() for f in features: v = f.get("properties", {}).get("sens_de_l_ecoulement") values[v if v not in (None, "") else ""] += 1 return dict(values) def load_troncons_for_basin( troncon_geojson: dict, basin_id: int, name_field: str = "cpx_toponyme_de_cours_d_eau", extra_names: Optional[List[str]] = None, anchor_stations: Optional[pd.DataFrame] = None, max_distance_from_anchor_km: float = 20.0, ) -> List[dict]: """ Filter troncon_hydrographique features to a basin's river + known tributaries, instead of a single named-river-only match. This is what lets branching topology exist at all -- matching only "Risle" or "Eure" by name discards every tributary tronçon before the graph is even built. Name matching alone is NOT sufficient at this bounding-box scale (~100x130 km): several tributary names are short/common enough ("Bec", "Avre", "Bave", "Corru", "Etang", "Iton") to genuinely match an unrelated stream elsewhere in the box that happens to share the word, not the actual tributary. Confirmed against real data: a name-only pull for these two basins produced 487/214 disconnected components and every real gauge snapping 1.5-50 km from the kept network -- a name collision pulling in geographically unrelated tronçons is the most likely explanation. If `anchor_stations` is given (this basin's real gauge coordinates), every matched tronçon is additionally required to have at least one vertex within `max_distance_from_anchor_km` of some anchor station. This can't fully separate "real tributary far from any gauge" from "name collision," but it does reject tronçons that are unambiguously nowhere near this basin's actual stations -- which the real run's 50 km snap distances show is happening. """ names = list(TRIBUTARY_NAMES.get(basin_id, [])) + (extra_names or []) names_lower = [n.lower() for n in names] name_matched = [] for f in troncon_geojson.get("features", []): toponym = str(f.get("properties", {}).get(name_field, "")).lower() if any(n in toponym for n in names_lower): name_matched.append(f) if anchor_stations is None or anchor_stations.empty: return name_matched anchors = list(zip(anchor_stations["latitude"], anchor_stations["longitude"])) matched = [] for f in name_matched: coords = f.get("geometry", {}).get("coordinates", []) if not coords: continue # Cheap pre-check on the tronçon's own endpoints before checking every vertex. endpoints = [coords[0], coords[-1]] if any(_haversine_km(alat, alon, pt[1], pt[0]) <= max_distance_from_anchor_km for pt in endpoints for alat, alon in anchors): matched.append(f) return matched def _endpoint_altitudes(coords: List[list]) -> Tuple[Optional[float], Optional[float]]: """(start_z, end_z) if the geometry has a Z coordinate, else (None, None).""" if not coords or len(coords[0]) < 3: return None, None return coords[0][2], coords[-1][2] def normalize_toponym(name: Optional[str]) -> Optional[str]: """ Normalize a river name for comparison: lowercase, strip whitespace, drop a leading article ("la "/"le "/"l'"/"les "), drop a "bras de/du/d'" prefix (a named secondary channel of a river -- "Bras de la Charentonne" is an arm of the Charentonne, not a different river; confirmed against real data as a genuine remaining false-confluence source even after the topological split-rejoin check, likely because the actual split point sits beyond that check's search radius or outside the selected component), and drop any parenthetical qualifier ("la Risle (Grand Bief)" -> "risle"). BD TOPO's toponym field is not guaranteed to be consistently formatted across tronçons for the same physical river -- comparing raw strings for uniqueness would count formatting variation (or a named arm of the same river) as a different river and keep inflating the confluence count. Returns None if there's nothing left after normalization. """ if not name: return None n = str(name).strip().lower() n = re.sub(r"\([^)]*\)", "", n).strip() # drop parenthetical qualifiers for prefix in ("bras de la ", "bras de l'", "bras du ", "bras de ", "bras d'"): if n.startswith(prefix): n = n[len(prefix):] break for article in ("l'", "la ", "le ", "les "): if n.startswith(article): n = n[len(article):] break n = n.strip() return n or None def _ancestors_within_km(G: nx.MultiDiGraph, start, max_km: float) -> Dict: """Bounded backward BFS (via predecessors) from `start`, tracking cumulative upstream distance. Returns {node: distance_km}. G is a MultiDiGraph (parallel edges between the same two nodes are real -- e.g. two distinct tronçons that both happen to run directly between the same pair of hydrographic nodes, exactly the shape a short braid takes). Where multiple parallel edges exist between a predecessor and n, the shortest is used -- the conservative choice for "is there any path within range," not an arbitrary one. """ visited = {start: 0.0} frontier = [start] while frontier: new_frontier = [] for n in frontier: for pred in G.predecessors(n): parallel = G.get_edge_data(pred, n) # {key: data} for every parallel edge edge_dist = min((d.get("distance_km", 0.0) or 0.0) for d in parallel.values()) cum = visited[n] + edge_dist if cum <= max_km and (pred not in visited or cum < visited[pred]): visited[pred] = cum new_frontier.append(pred) frontier = new_frontier return visited def is_split_rejoin(G: nx.DiGraph, node, max_upstream_km: float = 15.0) -> bool: """ True if `node`'s incoming branches trace back to a common upstream node within `max_upstream_km` -- i.e. this is the same water splitting into parallel channels and recombining (braiding, an anabranch, a named secondary "bras"/"bief"/"noue"), not a tributary confluence. No new drainage area or mass enters the system at a split-rejoin; treating one as a real confluence would misinform a mass-balance physics constraint. Doesn't depend on naming conventions at all -- catches a split-rejoin even when BD TOPO gives its two channels distinct names that `normalize_toponym` has no way to recognize as "the same river." """ in_edges = list(G.in_edges(node)) if len(in_edges) < 2: return False branch_sources = [u for u, _ in in_edges] ancestor_sets = [set(_ancestors_within_km(G, src, max_upstream_km).keys()) for src in branch_sources] common = set.intersection(*ancestor_sets) if ancestor_sets else set() common.discard(node) return len(common) > 0 def find_split_points(G: nx.DiGraph) -> Set: """ Nodes where a single upstream channel divides into 2+ downstream paths -- unambiguous by construction, unlike confluences/rejoins: a split has exactly one thing flowing in and multiple things flowing out, so it's always the same water partitioning, regardless of what the branches get named downstream. No name or topology check needed here the way `find_real_confluences` needs one -- out-degree >= 2 is a sufficient physical definition on its own. Useful as the other half of a mass-conservation pair with `find_rejoin_points`: for a genuine split-then-rejoin, flow at the split should equal flow at the corresponding rejoin, since no new catchment area is added anywhere in between. """ return {n for n in G.nodes if G.out_degree(n) >= 2} def find_rejoin_points(G: nx.DiGraph, max_upstream_km: float = 15.0) -> Set: """ Nodes where two branches recombine after a common upstream split -- the companion classification to `find_real_confluences`, made explicit and persistable rather than silently discarded. A rejoin and a real confluence look identical by raw in-degree; the distinguishing test is the same one `find_real_confluences` already uses (a common upstream ancestor within `max_upstream_km`), just keeping the nodes that test *positively* identifies instead of the ones it excludes. These matter for a different physics constraint than a real confluence: no new mass enters at a rejoin (or its paired split) -- what should hold is Q_upstream_of_split ~= Q_downstream_of_rejoin, not "new tributary inflow contributes here." """ rejoins = set() for n in G.nodes: if G.in_degree(n) >= 2 and is_split_rejoin(G, n, max_upstream_km): rejoins.add(n) return rejoins def pair_splits_and_rejoins(G: nx.DiGraph, max_upstream_km: float = 15.0) -> pd.DataFrame: """ For each rejoin point, identify which upstream split it pairs with -- the actual link a mass-conservation constraint needs (Q_at_this_split ~= Q_at_this_rejoin), not just "splits exist somewhere and rejoins exist somewhere" as two independent facts. Returns: DataFrame [rejoin_node, split_node, upstream_distance_km, branch_nodes] -- one row per rejoin, `branch_nodes` is the list of nodes along the shortest path of each branch between the split and the rejoin (so edge-level attributes for the braid are easy to pull later). A rejoin whose branches share more than one common ancestor within range keeps the CLOSEST one (shortest upstream distance), since that's the most specific matching split. """ rows = [] for rejoin in find_rejoin_points(G, max_upstream_km): in_edges = list(G.in_edges(rejoin)) branch_sources = [u for u, _ in in_edges] ancestor_sets = [_ancestors_within_km(G, src, max_upstream_km) for src in branch_sources] common = set(ancestor_sets[0].keys()) for s in ancestor_sets[1:]: common &= set(s.keys()) common.discard(rejoin) if not common: continue # closest common ancestor = the actual split point for this rejoin split = min(common, key=lambda n: max(a.get(n, float("inf")) for a in ancestor_sets)) dist = max(a.get(split, 0.0) for a in ancestor_sets) branch_nodes = [] for src in branch_sources: try: path = nx.shortest_path(G, split, src) branch_nodes.append(path) except nx.NetworkXNoPath: branch_nodes.append([]) rows.append({ "rejoin_node": rejoin, "split_node": split, "upstream_distance_km": dist, "branch_nodes": branch_nodes, }) return pd.DataFrame(rows) def find_real_confluences(G: nx.DiGraph, max_upstream_km: float = 15.0) -> Set: """ Nodes where a genuinely different, independently-sourced river joins -- not just where in-degree happens to be >= 2, and not a channel split that rejoins downstream (see `find_rejoin_points`, which keeps that classification rather than discarding it -- splits and rejoins carry real mass-conservation information of their own, just a different kind than a tributary confluence: no new catchment area is added at either point in a braid, whereas a real confluence genuinely does add one). Two filters, in order: 1. In-degree >= 2 with more than one distinct (normalized) river name -- see `normalize_toponym` for why raw string comparison isn't reliable. BD TOPO's fine tronçon segmentation produces many same-river multi-inflow nodes with no real branching involved (confirmed against real data -- distances as short as 4.6m at some flagged "confluences"). 2. NOT a rejoin: the branches must not trace back to a common upstream node within `max_upstream_km`. A braided or anabranched stretch can satisfy filter 1 (different channel names -- e.g. a named "bras"/arm) while still being the same water recombining, not a real tributary. """ confluences = set() for n in G.nodes: in_edges = list(G.in_edges(n, data=True)) if len(in_edges) < 2: continue toponyms = {normalize_toponym(e[2].get("toponym")) for e in in_edges} toponyms.discard(None) if len(toponyms) > 1 and not is_split_rejoin(G, n, max_upstream_km): confluences.add(n) return confluences def summarize_confluences_by_tributary(G: nx.DiGraph) -> pd.DataFrame: """ For each real confluence, which named tributary is actually joining there. This is the direct check against "the number of confluences should be explained by the number of tributaries" -- if this table has far more rows than TRIBUTARY_NAMES has entries for this basin, something is still over-counting (or a tributary itself has real internal sub-confluences worth knowing about, which this makes visible rather than hiding inside one aggregate number). Returns: DataFrame [node, main_river, joining_river, joining_river_raw] -- one row per real confluence. `joining_river` is normalized; `joining_river_raw` keeps the original string(s) seen, so you can tell formatting variation apart from a genuinely different tributary at a glance. """ rows = [] for n in find_real_confluences(G): in_edges = list(G.in_edges(n, data=True)) by_norm: Dict[str, List[str]] = {} for _, _, data in in_edges: raw = data.get("toponym") norm = normalize_toponym(raw) if norm: by_norm.setdefault(norm, []).append(raw) if len(by_norm) < 2: continue sorted_names = sorted(by_norm.items(), key=lambda kv: -len(kv[1])) main_river = sorted_names[0][0] for joining, raws in sorted_names[1:]: rows.append({ "node": n, "main_river": main_river, "joining_river": joining, "joining_river_raw_values_seen": sorted(set(raws)), }) return pd.DataFrame(rows) def build_node_link_digraph( features: List[dict], sens_field: str = "sens_de_l_ecoulement", sens_downstream_values: Tuple[str, ...] = ("Sens direct", "Direct", "01"), sens_upstream_values: Tuple[str, ...] = ("Sens indirect", "Indirect", "02"), name_field: str = "cpx_toponyme_de_cours_d_eau", ) -> Tuple[nx.DiGraph, ReachGraphReport]: """ Build a directed graph from tronçon-to-node linkage: one edge per tronçon, from its upstream hydrographic node to its downstream one. Direction is resolved per tronçon: `sens_de_l_ecoulement` is used if its value is recognized (the exact vocabulary is unverified against a real full export -- extend `sens_downstream_values`/`sens_upstream_values` once `check_flow_direction_coverage` shows what's actually there). Otherwise falls back to comparing the tronçon's own endpoint altitudes. A tronçon with neither a usable `sens` value nor a usable Z coordinate is dropped rather than guessed at, and counted in the report. CONFLUENCE DEFINITION: a node has in-degree >= 2 whenever two or more tronçons happen to end there -- which turns out NOT to reliably mean "a tributary joins here." BD TOPO segments a single continuous channel into very short tronçons (confirmed against real data: many under 20m), and dense node placement alone produces spurious in-degree >= 2 with no real branching involved. A "real" confluence here instead requires the incoming tronçon(s) to carry a DIFFERENT river name (`cpx_toponyme_de_cours_d_eau`) than the outgoing one -- "la Charentonne" joining "la Risle" is a real confluence; two same-named "la Risle" fragments both ending at the same node is fine-grained segmentation, not a tributary. `report.n_confluences` uses this definition; `n_raw_in_degree_ge2` is kept alongside it so the difference between the two is visible, not silently discarded. Returns: (digraph, report) -- digraph nodes are BD TOPO hydrographic node IDs; each edge carries the tronçon's geometry, distance_km, elevation_drop_m, toponym, basin_id, and cleabs as edge data. """ G = nx.MultiDiGraph() # parallel edges between the same node pair are real # data, not a collision -- two distinct tronçons directly connecting the # same two hydrographic nodes is exactly what a short braid looks like. # A plain DiGraph silently OVERWRITES the second such edge's data on # add_edge rather than keeping both (confirmed as a real, silent data # loss bug, not a hypothetical -- caught while testing a synthetic # adjacent-split-rejoin case). n_sens, n_elev, n_ambiguous = 0, 0, 0 for f in features: props = f.get("properties", {}) ini = props.get("lien_vers_noeud_hydrographique_ini") fin = props.get("lien_vers_noeud_hydrographique_fin") if not ini or not fin: n_ambiguous += 1 continue coords = f.get("geometry", {}).get("coordinates", []) if not coords: n_ambiguous += 1 continue sens = props.get(sens_field) z_start, z_end = _endpoint_altitudes(coords) if sens in sens_downstream_values: u, v = ini, fin n_sens += 1 elif sens in sens_upstream_values: u, v = fin, ini n_sens += 1 elif z_start is not None and z_end is not None and z_start != z_end: u, v = (ini, fin) if z_start > z_end else (fin, ini) n_elev += 1 else: n_ambiguous += 1 continue lon1, lat1 = coords[0][0], coords[0][1] lon2, lat2 = coords[-1][0], coords[-1][1] dist_km = _haversine_km(lat1, lon1, lat2, lon2) elev_drop = (z_start - z_end) if (z_start is not None and z_end is not None) else None if u != ini: # direction was flipped relative to raw geometry order elev_drop = -elev_drop if elev_drop is not None else None coords_ordered = list(reversed(coords)) else: coords_ordered = coords G.add_edge(u, v, distance_km=dist_km, elevation_drop_m=elev_drop, coords=coords_ordered, cleabs=props.get("cleabs"), toponym=props.get(name_field)) components = list(nx.weakly_connected_components(G)) largest = max((len(c) for c in components), default=0) real_confluences = find_real_confluences(G) raw_multi_inflow = [n for n in G.nodes if G.in_degree(n) >= 2] report = ReachGraphReport( n_troncons_matched=len(features), n_hydro_nodes=G.number_of_nodes(), n_confluences=len(real_confluences), n_raw_multi_inflow=len(raw_multi_inflow), flow_direction_from_sens=n_sens, flow_direction_from_elevation=n_elev, flow_direction_ambiguous=n_ambiguous, n_gauges_snapped=0, n_virtual_nodes_inserted=0, disconnected_components=len(components), largest_component_troncons=largest, ) return G, report def largest_component(G: nx.DiGraph) -> nx.DiGraph: """Keep only the largest weakly-connected component, by tronçon count. See `best_component_for_stations` for a criterion that's actually validated against real gauges instead -- "largest" has no guarantee of being anywhere near the stations we care about, and confirmed against real data it wasn't.""" components = list(nx.weakly_connected_components(G)) if not components: return G biggest = max(components, key=len) return G.subgraph(biggest).copy() def best_component_for_stations( G: nx.DiGraph, stations_df: pd.DataFrame, search_radius_km: float = 5.0 ) -> Tuple[nx.DiGraph, Dict[int, int]]: """ Keep the connected component containing the most real gauge stations, rather than the component with the most tronçons. A large-by-count component has no guarantee of being the network our actual gauges sit on; a component chosen by how many real, trusted station coordinates fall near it does. A station "belongs" to a component if any of that component's tronçon vertices falls within `search_radius_km` of the station. Returns: (best_component_subgraph, {component_index: n_stations_near_it}) -- the second value is diagnostic: if the winning component only narrowly beats the runner-up, or several components each capture a handful of stations, that's a sign the basin's real network is still fragmented and no single component is "the" river. """ components = list(nx.weakly_connected_components(G)) if not components: return G, {} counts = {} for i, comp in enumerate(components): subG = G.subgraph(comp) n_near = 0 for _, station in stations_df.iterrows(): found = False for _, _, data in subG.edges(data=True): for pt in data["coords"]: if _haversine_km(station["latitude"], station["longitude"], pt[1], pt[0]) <= search_radius_km: found = True break if found: break if found: n_near += 1 counts[i] = n_near best_idx = max(counts, key=counts.get) return G.subgraph(components[best_idx]).copy(), counts def nearest_edge_to_point(G: nx.DiGraph, lat: float, lon: float) -> Tuple[Tuple, Tuple, float, float]: """ Find the (u, v) edge in G whose geometry passes nearest to (lat, lon), and how far along that edge (0-1 fraction from u to v) the nearest point falls. Returns: (u, v, fraction, distance_km) """ best = None for u, v, data in G.edges(data=True): coords = data["coords"] for i in range(len(coords) - 1): ax, ay = coords[i][0], coords[i][1] bx, by = coords[i + 1][0], coords[i + 1][1] abx, aby = bx - ax, by - ay denom = abx ** 2 + aby ** 2 t = 0.0 if denom == 0 else np.clip(((lon - ax) * abx + (lat - ay) * aby) / denom, 0.0, 1.0) proj_lon, proj_lat = ax + t * abx, ay + t * aby dist = _haversine_km(lat, lon, proj_lat, proj_lon) if best is None or dist < best[3]: # fraction along the WHOLE edge, not just this segment seg_count = len(coords) - 1 frac = (i + t) / seg_count if seg_count else 0.0 best = (u, v, frac, dist) return best def snap_gauges_to_reach_graph(G: nx.DiGraph, stations_df: pd.DataFrame) -> pd.DataFrame: """ Snap each station onto its nearest edge in the reach graph. Args: G: from build_node_link_digraph (ideally largest_component'd first). stations_df: [station_code, latitude, longitude, elevation_m, ...] Returns: stations_df with added columns: reach_u, reach_v, reach_fraction, snap_distance_km """ rows = [] for _, row in stations_df.iterrows(): u, v, frac, dist = nearest_edge_to_point(G, row["latitude"], row["longitude"]) rows.append({"station_code": row["station_code"], "reach_u": u, "reach_v": v, "reach_fraction": frac, "snap_distance_km": dist}) snapped = pd.DataFrame(rows) return stations_df.merge(snapped, on="station_code", how="left") def build_reach_graph_tables( G: nx.DiGraph, gauges_snapped: pd.DataFrame, basin_id: int, ) -> Tuple[pd.DataFrame, pd.DataFrame]: """ Assemble the final nodes_df/edges_df for this basin, combining real gauges, real confluences, and virtual nodes into one table in the same shape build_pyg_graph/build_pyg_graphs_per_basin already expect -- so nothing downstream of graph construction needs to change. A gauge station keeps its own real coordinates and elevation (not the snapped point's) since that's the actual, correct location of the physical instrument; it's associated with the graph via its nearest edge, splitting that edge at the gauge's snapped position. Every other node in G (confluences, virtual nodes already inserted upstream by insert_virtual_nodes) becomes its own graph node using coordinates taken directly from the reach geometry. Args: G: reach digraph, ideally after largest_component() and insert_virtual_nodes(). gauges_snapped: output of snap_gauges_to_reach_graph, run against this same G. basin_id: 0 (Eure) or 1 (Risle). Returns: (nodes_df, edges_df) in the same schema as build_surface_edges's output: nodes_df has [station_code, basin_id, latitude, longitude, elevation_m, is_gauged]; edges_df has [source, target, basin_id, distance_km, elevation_drop_m, verified_continuous]. """ real_confluences = find_real_confluences(G) # single authoritative source of truth, # computed once here where the actual graph topology is available -- everything # downstream (both apps, both scripts) reads the saved is_confluence column # rather than each re-deriving it from the flat CSV, which is exactly how the # split-rejoin exclusion and the earlier normalization fix drifted out of sync # across files before. split_points = find_split_points(G) rejoin_points = find_rejoin_points(G) # Splits and rejoins are NOT discarded as noise -- they carry a real, # different mass-conservation constraint than a tributary confluence # (Q_at_split ~= Q_at_rejoin, no new catchment area involved, vs. a # confluence genuinely adding one). braid_id links a rejoin to its # paired split so that constraint is actually constructible later, # not just "some splits and rejoins exist somewhere, unpaired." braid_pairs = pair_splits_and_rejoins(G) rejoin_to_split = dict(zip(braid_pairs["rejoin_node"], braid_pairs["split_node"])) if not braid_pairs.empty else {} node_rows = {} for n, data in G.nodes(data=True): lat, lon, elev = data.get("latitude"), data.get("longitude"), data.get("elevation_m") if lat is None or lon is None: # Real BD TOPO hydrographic nodes carry no attributes of their # own -- only the edges know coordinates. Derive this node's # position from any adjacent edge's endpoint nearest to it # (an out-edge starts here; an in-edge ends here). out_edges = list(G.out_edges(n, data=True)) in_edges = list(G.in_edges(n, data=True)) if out_edges: p = out_edges[0][2]["coords"][0] elif in_edges: p = in_edges[0][2]["coords"][-1] else: continue # isolated node with no geometry anywhere -- can't place it lon, lat = p[0], p[1] elev = p[2] if len(p) > 2 else None node_rows[n] = { "station_code": str(n), "basin_id": basin_id, "latitude": lat, "longitude": lon, "elevation_m": elev, "is_gauged": False, "is_confluence": n in real_confluences, "is_split_point": n in split_points, "is_rejoin_point": n in rejoin_points, "braid_id": str(rejoin_to_split[n]) if n in rejoin_to_split else None, } edge_rows = [] for u, v, data in G.edges(data=True): edge_rows.append({ "source": str(u), "target": str(v), "basin_id": basin_id, "distance_km": data["distance_km"], "elevation_drop_m": data.get("elevation_drop_m"), "verified_continuous": True, "cleabs": data.get("cleabs"), "toponym": data.get("toponym"), }) # Splice each gauge into its nearest edge: gauge_u -> GAUGE -> gauge_v, # replacing the direct u->v edge, so the gauge becomes a real node on # the path rather than a disconnected point sitting near the line. for _, g in gauges_snapped.dropna(subset=["reach_u", "reach_v"]).iterrows(): u, v, frac = g["reach_u"], g["reach_v"], g["reach_fraction"] code = g["station_code"] node_rows[code] = { "station_code": code, "basin_id": basin_id, "latitude": g["latitude"], "longitude": g["longitude"], "elevation_m": g.get("elevation_m"), "is_gauged": True, "is_confluence": False, "is_split_point": False, "is_rejoin_point": False, "braid_id": None, "snap_distance_km": g.get("snap_distance_km"), } original = next((e for e in edge_rows if e["source"] == str(u) and e["target"] == str(v)), None) if original is None: continue edge_rows.remove(original) d, drop = original["distance_km"], original["elevation_drop_m"] edge_rows.append({"source": str(u), "target": code, "basin_id": basin_id, "distance_km": d * frac, "elevation_drop_m": drop * frac if drop is not None else None, "verified_continuous": original["verified_continuous"], "cleabs": original.get("cleabs"), "toponym": original.get("toponym")}) edge_rows.append({"source": code, "target": str(v), "basin_id": basin_id, "distance_km": d * (1 - frac), "elevation_drop_m": drop * (1 - frac) if drop is not None else None, "verified_continuous": original["verified_continuous"], "cleabs": original.get("cleabs"), "toponym": original.get("toponym")}) nodes_df = pd.DataFrame(node_rows.values()) edges_df = pd.DataFrame(edge_rows) return nodes_df, edges_df def insert_virtual_nodes(G: nx.MultiDiGraph, spacing_km: float = 5.0) -> Tuple[nx.MultiDiGraph, int]: """ Split any edge longer than `spacing_km` into evenly-spaced segments by inserting virtual node IDs along it, so long confluence-free stretches get intermediate nodes instead of one edge covering many kilometers. Virtual node IDs are strings like "VIRTUAL::{u}::{v}::{i}::{key}" -- distinct from real BD TOPO node IDs by construction, so they can never collide. The edge `key` is included specifically because two parallel tronçons between the same (u, v) pair are real, not a collision (see build_node_link_digraph) -- each needs its own distinctly-IDed virtual chain, and removing the correct specific parallel edge (not an arbitrary one) requires operating on its key, not just its endpoints. Returns: (new_digraph, n_virtual_nodes_inserted) """ G2 = G.copy() n_inserted = 0 for u, v, key, data in list(G.edges(keys=True, data=True)): dist = data["distance_km"] if dist <= spacing_km: continue n_segments = max(2, int(np.ceil(dist / spacing_km))) coords = data["coords"] n_points = len(coords) G2.remove_edge(u, v, key) prev_node = u for i in range(1, n_segments): frac = i / n_segments idx = min(int(frac * (n_points - 1)), n_points - 2) local_t = frac * (n_points - 1) - idx p0, p1 = coords[idx], coords[idx + 1] lon = p0[0] + local_t * (p1[0] - p0[0]) lat = p0[1] + local_t * (p1[1] - p0[1]) z = (p0[2] + local_t * (p1[2] - p0[2])) if len(p0) > 2 else None vnode = f"VIRTUAL::{u}::{v}::{key}::{i}" G2.add_node(vnode, latitude=lat, longitude=lon, elevation_m=z) seg_dist = dist / n_segments seg_drop = (data["elevation_drop_m"] / n_segments) if data["elevation_drop_m"] is not None else None G2.add_edge(prev_node, vnode, distance_km=seg_dist, elevation_drop_m=seg_drop, coords=[coords[idx], coords[idx + 1]], cleabs=data.get("cleabs"), toponym=data.get("toponym")) prev_node = vnode n_inserted += 1 seg_dist = dist / n_segments seg_drop = (data["elevation_drop_m"] / n_segments) if data["elevation_drop_m"] is not None else None G2.add_edge(prev_node, v, distance_km=seg_dist, elevation_drop_m=seg_drop, coords=coords[-2:], cleabs=data.get("cleabs"), toponym=data.get("toponym")) return G2, n_inserted