Spaces:
Running on Zero
Running on Zero
File size: 4,458 Bytes
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 | """
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()
}
}
|