""" Station utilities for basin assignment and metadata handling. Single Responsibility: Station-related helper functions. """ import pandas as pd from pathlib import Path from typing import Dict, List, Optional from ..config.settings import STATION_LIST, BASIN_IDS def assign_basin_id(station_name: str) -> int: """ Assign basin ID based on station name. Args: station_name: Station name (e.g., "La Risle à Rai", "L'Eure à Chartres") Returns: Basin ID: 0 for La Risle, 1 for La Eure """ station_name_lower = station_name.lower() if "risle" in station_name_lower: return BASIN_IDS["LA_RISLE"] elif "eure" in station_name_lower or "bras de l'eure" in station_name_lower: return BASIN_IDS["LA_EURE"] else: raise ValueError(f"Cannot determine basin for station: {station_name}") def load_station_metadata( filepath: Optional[Path] = None, add_basin_id: bool = True ) -> pd.DataFrame: """ Load station metadata from CSV. Args: filepath: Path to station_list.csv (default: from settings) add_basin_id: Whether to add basin_id column Returns: DataFrame with station metadata """ if filepath is None: filepath = STATION_LIST df = pd.read_csv(filepath) # Add basin_id column if requested if add_basin_id and "basin_id" not in df.columns: df["basin_id"] = df["station_name"].apply(assign_basin_id) return df def get_stations_by_basin(basin_id: int, filepath: Optional[Path] = None) -> List[str]: """ Get list of station codes for a specific basin. Args: basin_id: Basin identifier (0=Risle, 1=Eure) filepath: Path to station_list.csv (optional) Returns: List of station codes """ df = load_station_metadata(filepath, add_basin_id=True) return df[df["basin_id"] == basin_id]["station_code"].tolist() def get_basin_name(basin_id: int) -> str: """ Get basin name from basin ID. Args: basin_id: Basin identifier Returns: Basin name """ basin_map = {v: k for k, v in BASIN_IDS.items()} return basin_map.get(basin_id, "UNKNOWN") def validate_station_code(station_code: str, filepath: Optional[Path] = None) -> bool: """ Check if station code exists in station list. Args: station_code: Station code to validate filepath: Path to station_list.csv (optional) Returns: True if valid """ df = load_station_metadata(filepath, add_basin_id=False) return station_code in df["station_code"].values def get_station_coordinates( station_code: str, filepath: Optional[Path] = None, crs: str = "wgs84" ) -> tuple: """ Get coordinates for a station. Args: station_code: Station code filepath: Path to station_list.csv (optional) crs: Coordinate system ("wgs84" for lat/lon, "lambert93" for X/Y) Returns: Tuple of (x, y) or (lon, lat) coordinates """ df = load_station_metadata(filepath, add_basin_id=False) station = df[df["station_code"] == station_code] if len(station) == 0: raise ValueError(f"Station code not found: {station_code}") station = station.iloc[0] if crs.lower() == "wgs84": return (station["lon"], station["lat"]) elif crs.lower() == "lambert93": return (station["X"], station["Y"]) else: raise ValueError(f"Unknown CRS: {crs}. Use 'wgs84' or 'lambert93'") def get_station_summary() -> Dict: """ Get summary statistics of stations. Returns: Dictionary with station summary """ df = load_station_metadata(add_basin_id=True) risle_stations = df[df["basin_id"] == BASIN_IDS["LA_RISLE"]] eure_stations = df[df["basin_id"] == BASIN_IDS["LA_EURE"]] return { "total_stations": len(df), "risle_stations": len(risle_stations), "eure_stations": len(eure_stations), "risle_codes": risle_stations["station_code"].tolist(), "eure_codes": eure_stations["station_code"].tolist(), "bbox": { "lon_min": df["lon"].min(), "lon_max": df["lon"].max(), "lat_min": df["lat"].min(), "lat_max": df["lat"].max() } }