Spaces:
Running on Zero
Running on Zero
| """ | |
| Hydrometric data loader for Hub'Eau API v2 discharge and water level data. | |
| Single Responsibility: Load and parse hydrometric observations. | |
| """ | |
| import pandas as pd | |
| import matplotlib.pyplot as plt | |
| import matplotlib.dates as mdates | |
| from pathlib import Path | |
| from typing import Optional, List | |
| from .base import BaseDataLoader | |
| class HydrometricLoader(BaseDataLoader): | |
| """ | |
| Loads Hub'Eau hydrometric data (discharge & water level). | |
| Handles actual Hub'Eau API v2 structure: | |
| - date_obs_elab: observation date | |
| - station_code: station ID | |
| - resultat_obs_elab: measured value | |
| - grandeur_hydro_elab: variable type (QmnJ, HIXnJ, etc.) | |
| - code_qualification: quality code (16=acceptable, 20=good) | |
| """ | |
| def __init__( | |
| self, | |
| data_path: Path, | |
| min_quality: bool = True, | |
| station_ids: Optional[List[str]] = None, | |
| discharge_grandeur: str = "QmnJ", # Fixed: Hub'Eau daily mean discharge | |
| waterlevel_grandeur: str = "HIXnJ" # Fixed: Hub'Eau daily max water level | |
| ): | |
| """ | |
| Initialize hydrometric loader. | |
| Args: | |
| data_path: Path to datasets/hydrometric directory | |
| min_quality: Only keep good quality data (quality_code >= 16) | |
| station_ids: Optional list of station IDs to filter | |
| discharge_grandeur: Hub'Eau code for discharge ('QmnJ', 'QIXnJ', etc.) | |
| waterlevel_grandeur: Hub'Eau code for water level ('HIXnJ', 'HIXM', etc.) | |
| """ | |
| super().__init__(data_path) | |
| self.min_quality = min_quality | |
| self.station_ids = station_ids | |
| self.discharge_grandeur = discharge_grandeur | |
| self.waterlevel_grandeur = waterlevel_grandeur | |
| def load(self) -> pd.DataFrame: | |
| """ | |
| Load hydrometric data from Hub'Eau CSV files. | |
| Returns: | |
| DataFrame with [date, station_code, discharge_m3s, waterlevel_mm] | |
| """ | |
| data_dir = Path(self.data_path) | |
| discharge_file = data_dir / "discharge_observations.csv" | |
| waterlevel_file = data_dir / "waterlevel_observations.csv" | |
| dfs = [] | |
| if discharge_file.exists(): | |
| discharge_df = self._load_single_variable( | |
| file_path=discharge_file, | |
| value_col='discharge_m3s', | |
| grandeur_code=self.discharge_grandeur | |
| ) | |
| dfs.append(discharge_df) | |
| if waterlevel_file.exists(): | |
| waterlevel_df = self._load_single_variable( | |
| file_path=waterlevel_file, | |
| value_col='waterlevel_mm', | |
| grandeur_code=self.waterlevel_grandeur | |
| ) | |
| dfs.append(waterlevel_df) | |
| if not dfs: | |
| raise FileNotFoundError(f"No hydrometric files found in {data_dir}") | |
| # Merge discharge and water level | |
| df = dfs[0] if len(dfs) == 1 else pd.merge( | |
| dfs[0], dfs[1], | |
| on=['date', 'station_code'], | |
| how='outer' | |
| ) | |
| # Filter by station IDs | |
| if self.station_ids: | |
| df = df[df["station_code"].isin(self.station_ids)] | |
| return df.sort_values(['date', 'station_code']).reset_index(drop=True) | |
| def _load_single_variable( | |
| self, | |
| file_path: Path, | |
| value_col: str, | |
| grandeur_code: Optional[str] = None | |
| ) -> pd.DataFrame: | |
| """Load and process a single Hub'Eau CSV file.""" | |
| df = pd.read_csv(file_path) | |
| # Rename to standard columns | |
| df = df.rename(columns={ | |
| 'date_obs_elab': 'date', | |
| 'resultat_obs_elab': 'value', | |
| 'code_qualification': 'quality_code', | |
| 'grandeur_hydro_elab': 'grandeur_code' | |
| }) | |
| # Explicitly filter by Hub'Eau variable code | |
| if grandeur_code and 'grandeur_code' in df.columns: | |
| df = df[df['grandeur_code'] == grandeur_code].copy() | |
| # Convert date | |
| df['date'] = pd.to_datetime(df['date']) | |
| # Filter by quality (16=acceptable, 20=good) | |
| if self.min_quality and 'quality_code' in df.columns: | |
| df = df[df['quality_code'] >= 16].copy() | |
| # Rename value column | |
| df[value_col] = df['value'] | |
| # Keep relevant columns | |
| df = df[['date', 'station_code', value_col]].copy() | |
| # Deduplicate safely | |
| df = df.sort_values('date').drop_duplicates( | |
| subset=['date', 'station_code'], | |
| keep='last' | |
| ) | |
| return df | |
| def get_metadata(self) -> dict: | |
| """Get hydrometric data metadata.""" | |
| meta = super().get_metadata() | |
| meta.update({ | |
| "data_type": "hydrometric", | |
| "source": "Hub'Eau API v2", | |
| "quality_filter": self.min_quality, | |
| "filtered_stations": self.station_ids, | |
| "discharge_grandeur": self.discharge_grandeur, | |
| "waterlevel_grandeur": self.waterlevel_grandeur | |
| }) | |
| return meta | |
| # ------------------------------------------------------------------ | |
| # Plotting | |
| # ------------------------------------------------------------------ | |
| def _plot_timeseries( | |
| self, | |
| df: pd.DataFrame, | |
| value_col: str, | |
| ylabel: str, | |
| title: str, | |
| stations: Optional[List[str]] = None, | |
| ax: Optional[plt.Axes] = None, | |
| figsize: tuple = (12, 5), | |
| save_path: Optional[Path] = None, | |
| ) -> plt.Axes: | |
| """Shared line-plot logic for a single variable, one line per station.""" | |
| if value_col not in df.columns: | |
| raise ValueError(f"Column '{value_col}' not found in data") | |
| plot_df = df.dropna(subset=[value_col]) | |
| if stations: | |
| plot_df = plot_df[plot_df["station_code"].isin(stations)] | |
| if plot_df.empty: | |
| raise ValueError("No data available to plot for the given stations/variable") | |
| standalone = ax is None | |
| if standalone: | |
| fig, ax = plt.subplots(figsize=figsize) | |
| for station_code, group in plot_df.groupby("station_code"): | |
| group = group.sort_values("date") | |
| ax.plot(group["date"], group[value_col], marker="o", markersize=2, | |
| linewidth=1, label=station_code) | |
| ax.set_title(title) | |
| ax.set_xlabel("Date") | |
| ax.set_ylabel(ylabel) | |
| ax.xaxis.set_major_locator(mdates.AutoDateLocator()) | |
| ax.xaxis.set_major_formatter(mdates.ConciseDateFormatter(ax.xaxis.get_major_locator())) | |
| ax.legend(title="Station", fontsize=8, loc="best") | |
| ax.grid(True, alpha=0.3) | |
| if standalone: | |
| plt.tight_layout() | |
| if save_path: | |
| plt.savefig(save_path, dpi=150, bbox_inches="tight") | |
| print(f"Plot saved to {save_path}") | |
| return ax | |
| def plot_waterlevel( | |
| self, | |
| df: Optional[pd.DataFrame] = None, | |
| stations: Optional[List[str]] = None, | |
| figsize: tuple = (12, 5), | |
| save_path: Optional[Path] = None, | |
| ) -> plt.Axes: | |
| """ | |
| Plot water level time series, one line per station. | |
| Args: | |
| df: Optional pre-loaded DataFrame. If not provided, data is loaded from disk. | |
| stations: Optional list of station codes to plot. Defaults to all stations present. | |
| figsize: Figure size in inches. | |
| save_path: If provided, saves the figure to this path. | |
| Returns: | |
| The matplotlib Axes object. | |
| """ | |
| if df is None: | |
| df = self.load() | |
| return self._plot_timeseries( | |
| df, | |
| value_col="waterlevel_mm", | |
| ylabel="Water level (mm)", | |
| title="Water Level Observations", | |
| stations=stations, | |
| figsize=figsize, | |
| save_path=save_path, | |
| ) | |
| def plot_discharge( | |
| self, | |
| df: Optional[pd.DataFrame] = None, | |
| stations: Optional[List[str]] = None, | |
| figsize: tuple = (12, 5), | |
| save_path: Optional[Path] = None, | |
| ) -> plt.Axes: | |
| """ | |
| Plot discharge time series, one line per station. | |
| Args: | |
| df: Optional pre-loaded DataFrame. If not provided, data is loaded from disk. | |
| stations: Optional list of station codes to plot. Defaults to all stations present. | |
| figsize: Figure size in inches. | |
| save_path: If provided, saves the figure to this path. | |
| Returns: | |
| The matplotlib Axes object. | |
| """ | |
| if df is None: | |
| df = self.load() | |
| return self._plot_timeseries( | |
| df, | |
| value_col="discharge_m3s", | |
| ylabel="Discharge (m³/s)", | |
| title="Discharge Observations", | |
| stations=stations, | |
| figsize=figsize, | |
| save_path=save_path, | |
| ) | |
| def plot_station( | |
| self, | |
| station_id: str, | |
| df: Optional[pd.DataFrame] = None, | |
| figsize: tuple = (12, 8), | |
| save_path: Optional[Path] = None, | |
| ) -> "plt.Figure": | |
| """ | |
| Plot discharge and water level for a single station, stacked on | |
| two subplots so their differing scales don't distort each other. | |
| Args: | |
| station_id: The station code to plot. | |
| df: Optional pre-loaded DataFrame. If not provided, data is loaded from disk. | |
| figsize: Figure size in inches. | |
| save_path: If provided, saves the figure to this path. | |
| Returns: | |
| The matplotlib Figure object. | |
| """ | |
| if df is None: | |
| df = self.load() | |
| station_df = df[df["station_code"] == station_id] | |
| if station_df.empty: | |
| raise ValueError(f"No data found for station '{station_id}'") | |
| fig, (ax1, ax2) = plt.subplots(2, 1, figsize=figsize, sharex=True) | |
| if "discharge_m3s" in station_df.columns and station_df["discharge_m3s"].notna().any(): | |
| self._plot_timeseries( | |
| station_df, "discharge_m3s", "Discharge (m³/s)", | |
| f"Discharge — {station_id}", ax=ax1, | |
| ) | |
| else: | |
| ax1.set_title(f"Discharge — {station_id} (no data)") | |
| if "waterlevel_mm" in station_df.columns and station_df["waterlevel_mm"].notna().any(): | |
| self._plot_timeseries( | |
| station_df, "waterlevel_mm", "Water level (mm)", | |
| f"Water Level — {station_id}", ax=ax2, | |
| ) | |
| else: | |
| ax2.set_title(f"Water Level — {station_id} (no data)") | |
| for ax in (ax1, ax2): | |
| ax.get_legend().remove() if ax.get_legend() else None | |
| plt.tight_layout() | |
| if save_path: | |
| plt.savefig(save_path, dpi=150, bbox_inches="tight") | |
| print(f"Plot saved to {save_path}") | |
| return fig | |
| def plot_data_availability( | |
| self, | |
| df: Optional[pd.DataFrame] = None, | |
| figsize: tuple = (10, 6), | |
| save_path: Optional[Path] = None, | |
| ) -> plt.Axes: | |
| """ | |
| Plot a bar chart of observation counts per station, split by variable. | |
| Useful for spotting stations with sparse or missing coverage. | |
| Args: | |
| df: Optional pre-loaded DataFrame. If not provided, data is loaded from disk. | |
| figsize: Figure size in inches. | |
| save_path: If provided, saves the figure to this path. | |
| Returns: | |
| The matplotlib Axes object. | |
| """ | |
| if df is None: | |
| df = self.load() | |
| counts = pd.DataFrame({ | |
| "discharge_m3s": df.groupby("station_code")["discharge_m3s"].count() | |
| if "discharge_m3s" in df.columns else 0, | |
| "waterlevel_mm": df.groupby("station_code")["waterlevel_mm"].count() | |
| if "waterlevel_mm" in df.columns else 0, | |
| }).fillna(0) | |
| fig, ax = plt.subplots(figsize=figsize) | |
| counts.plot(kind="bar", ax=ax, color=["steelblue", "darkorange"]) | |
| ax.set_title("Observation Count by Station") | |
| ax.set_xlabel("Station") | |
| ax.set_ylabel("Number of observations") | |
| ax.legend(title="Variable") | |
| ax.grid(True, alpha=0.3, axis="y") | |
| plt.xticks(rotation=45, ha="right") | |
| plt.tight_layout() | |
| if save_path: | |
| plt.savefig(save_path, dpi=150, bbox_inches="tight") | |
| print(f"Plot saved to {save_path}") | |
| return ax | |
| def plot_rating_curve( | |
| self, | |
| station_id: str, | |
| df: Optional[pd.DataFrame] = None, | |
| figsize: tuple = (8, 8), | |
| save_path: Optional[Path] = None, | |
| ) -> plt.Axes: | |
| """ | |
| Plot discharge vs. water level for a single station as a scatter | |
| (a simple rating-curve style view). Points are colored by year to | |
| help spot rating shifts (e.g. channel changes) over time. | |
| Args: | |
| station_id: The station code to plot. | |
| df: Optional pre-loaded DataFrame. If not provided, data is loaded from disk. | |
| figsize: Figure size in inches. | |
| save_path: If provided, saves the figure to this path. | |
| Returns: | |
| The matplotlib Axes object. | |
| """ | |
| if df is None: | |
| df = self.load() | |
| station_df = df[df["station_code"] == station_id].dropna( | |
| subset=["discharge_m3s", "waterlevel_mm"] | |
| ) | |
| if station_df.empty: | |
| raise ValueError( | |
| f"No overlapping discharge/water level data for station '{station_id}'" | |
| ) | |
| fig, ax = plt.subplots(figsize=figsize) | |
| years = station_df["date"].dt.year | |
| scatter = ax.scatter( | |
| station_df["waterlevel_mm"], station_df["discharge_m3s"], | |
| c=years, cmap="viridis", s=15, alpha=0.7, | |
| ) | |
| cbar = plt.colorbar(scatter, ax=ax) | |
| cbar.set_label("Year") | |
| ax.set_title(f"Rating Curve — {station_id}") | |
| ax.set_xlabel("Water level (mm)") | |
| ax.set_ylabel("Discharge (m³/s)") | |
| ax.grid(True, alpha=0.3) | |
| plt.tight_layout() | |
| if save_path: | |
| plt.savefig(save_path, dpi=150, bbox_inches="tight") | |
| print(f"Plot saved to {save_path}") | |
| return ax | |
| def plot_seasonal_climatology( | |
| self, | |
| variable: str = "discharge_m3s", | |
| df: Optional[pd.DataFrame] = None, | |
| stations: Optional[List[str]] = None, | |
| figsize: tuple = (10, 6), | |
| save_path: Optional[Path] = None, | |
| ) -> plt.Axes: | |
| """ | |
| Plot the monthly climatology (median with 25th-75th percentile band) | |
| of a variable, one line per station. Shows the typical seasonal | |
| cycle and its spread across all years of record. | |
| Args: | |
| variable: Either 'discharge_m3s' or 'waterlevel_mm'. | |
| df: Optional pre-loaded DataFrame. If not provided, data is loaded from disk. | |
| stations: Optional list of station codes to include. Defaults to all. | |
| figsize: Figure size in inches. | |
| save_path: If provided, saves the figure to this path. | |
| Returns: | |
| The matplotlib Axes object. | |
| """ | |
| if df is None: | |
| df = self.load() | |
| if variable not in df.columns: | |
| raise ValueError(f"Column '{variable}' not found in data") | |
| plot_df = df.dropna(subset=[variable]).copy() | |
| if stations: | |
| plot_df = plot_df[plot_df["station_code"].isin(stations)] | |
| if plot_df.empty: | |
| raise ValueError("No data available to plot for the given stations/variable") | |
| plot_df["month"] = plot_df["date"].dt.month | |
| fig, ax = plt.subplots(figsize=figsize) | |
| for station_code, group in plot_df.groupby("station_code"): | |
| stats = group.groupby("month")[variable].agg( | |
| median="median", q25=lambda x: x.quantile(0.25), q75=lambda x: x.quantile(0.75) | |
| ) | |
| line, = ax.plot(stats.index, stats["median"], marker="o", label=station_code) | |
| ax.fill_between(stats.index, stats["q25"], stats["q75"], | |
| color=line.get_color(), alpha=0.15) | |
| ylabel = "Discharge (m³/s)" if variable == "discharge_m3s" else "Water level (mm)" | |
| ax.set_title(f"Seasonal Climatology — {ylabel}") | |
| ax.set_xlabel("Month") | |
| ax.set_ylabel(ylabel) | |
| ax.set_xticks(range(1, 13)) | |
| ax.set_xticklabels(["Jan","Feb","Mar","Apr","May","Jun", | |
| "Jul","Aug","Sep","Oct","Nov","Dec"]) | |
| ax.legend(title="Station", fontsize=8) | |
| ax.grid(True, alpha=0.3) | |
| plt.tight_layout() | |
| if save_path: | |
| plt.savefig(save_path, dpi=150, bbox_inches="tight") | |
| print(f"Plot saved to {save_path}") | |
| return ax |