Source code for src.data_loader

"""Loaders for observed and training datasets used by the forecast tool.

This module reads the various on-disk datasets the model is trained and
validated against, returning each as a tidy, date-indexed
:class:`pandas.DataFrame`:

- **GLCC** observed Net Basin Supply (:meth:`DataLoader.glcc`)
- **L2SWBM** precipitation, evaporation, and runoff (:meth:`DataLoader.l2swbm`)
- **GLSEA** lake surface temperatures (:meth:`DataLoader.glsea`)
- **SNODAS** snow water equivalent (:meth:`DataLoader.snodas`)
- USACE probability-of-exceedance tables (:meth:`DataLoader.lake_probabilities`)

Several loaders accept a ``units`` argument and convert between cubic metres
per second (``cms``) and millimetres per month (``mm``) using the per-lake
surface areas in :data:`lake_areas`.
"""

import pandas as pd
import numpy as np
import calendar
import os

# Surface area [m²] for each lake
lake_areas = {
    'superior': 78288645587.81192,
    'michigan-huron': 123626283030.46616,
    'erie': 18596386416.712486,
    'ontario': 15569248531.837788
}

[docs] class DataLoader: """Load the observed and training datasets used to fit and validate models. Each method reads one dataset family from a local directory and returns a cleaned, date-indexed :class:`pandas.DataFrame`. The loader is stateless; all paths and options are passed per call. """
[docs] def __init__(self): """Initialize the DataLoader. No configuration is required.""" pass
[docs] def glcc(self, directory, units='cms'): """ Load observed GLCC NBS data for all four lakes in either 'cms' or 'mm' units. """ filenames = { 'superior': 'LakeSuperior_MonthlyNetBasinSupply_1900to2025.csv', 'michigan-huron': 'LakeMichiganHuron_MonthlyNetBasinSupply_1900to2025.csv', 'erie': 'LakeErie_MonthlyNetBasinSupply_1900to2025.csv', 'ontario': 'LakeOntario_MonthlyNetBasinSupply_1900to2025.csv', } def format_data(df): df_long = df.melt(id_vars=['Year'], var_name='Month', value_name='Observed') df_long['date'] = pd.to_datetime( df_long['Year'].astype(str) + '-' + df_long['Month'].str[1:4] + '-01', format='%Y-%b-%d' ) return df_long[['date', 'Observed']].sort_values('date').reset_index(drop=True) dfs = [] for lake, filename in filenames.items(): path = os.path.join(directory, filename) df = pd.read_csv(path, skiprows=11) formatted = format_data(df).rename(columns={'Observed': f'{lake}_nbs_obs'}) dfs.append(formatted) # Merge all dataframes on 'date' df_merged = dfs[0] for df in dfs[1:]: df_merged = pd.merge(df_merged, df, on='date', how='outer') df_merged.replace(-99990.0, np.nan, inplace=True) lake_cols = [col for col in df_merged.columns if col.endswith('_nbs_obs')] df_merged.dropna(subset=lake_cols, how='all', inplace=True) if units == 'mm': df_merged['seconds_in_month'] = df_merged['date'].apply( lambda x: calendar.monthrange(x.year, x.month)[1] * 24 * 60 * 60 ) for col in lake_cols: lake = col.replace('_nbs_obs', '') area = lake_areas.get(lake) if area: df_merged[col] = ( df_merged[col] * df_merged['seconds_in_month'] / area * 1000 ) df_merged.drop(columns='seconds_in_month', inplace=True) df_merged.set_index('date', inplace=True) return df_merged
[docs] def l2swbm(self, directory): """ Load L2SWBM runoff, evaporation, and precipitation for all lakes. """ lakes = { 'superior': 'sup', 'michigan-huron': 'mih', 'erie': 'eri', 'ontario': 'ont' } variables = ['Evap', 'Runoff', 'Precip'] dfs = [] for lake_name in lakes.keys(): lake_data = {} for var in variables: lake_file_name = lake_name.replace('michigan-huron', 'miHuron') file_path = os.path.join(directory, f'{lake_file_name}{var}_MonthlyRun.csv') df = pd.read_csv(file_path) df['date'] = pd.to_datetime( df['Year'].astype(int).astype(str) + '-' + df['Month'].astype(int).astype(str).str.zfill(2) ) col_name = f'{lake_name.lower()}_{var.lower()}_obs' lake_data[col_name] = df['Median'] if 'date' not in lake_data: lake_data['date'] = df['date'] lake_df = pd.DataFrame(lake_data) dfs.append(lake_df) # Merge all lake dataframes on 'date' final_df = dfs[0] for df in dfs[1:]: final_df = pd.merge(final_df, df, on='date') final_df.set_index('date', inplace=True) final_df.rename(columns=lambda col: col.replace('precip_', 'precipitation_').replace('evap_', 'evaporation_'), inplace=True) return final_df
[docs] def glsea(self, file_path, units='K'): """ Reads GLSEA SST data from a file and returns a cleaned DataFrame. """ df = pd.read_csv(file_path, sep=r'\s+', skiprows=6, comment='-') df['date'] = pd.to_datetime(df['Year'].astype(str) + df['Day'].astype(str), format='%Y%j') df.set_index('date', inplace=True) df.drop(columns=['Year', 'Day'], inplace=True) df.columns = ['superior_sst', 'michigan_sst', 'huron_sst', 'erie_sst', 'ontario_sst', 'stclair_sst'] df['michigan-huron_sst'] = (df['michigan_sst'] + df['huron_sst']) / 2 df = df[['superior_sst', 'michigan-huron_sst', 'erie_sst', 'ontario_sst']] if units.upper() == 'K': df += 273.15 elif units.upper() != 'C': raise ValueError("Unsupported units. Use 'C' for Celsius or 'K' for Kelvin.") return df
[docs] def snodas(self,file): """ Load and process SNODAS SWE data. Parameters ---------- local_path : str Base path to local data directory. X_df : pd.DataFrame, optional DataFrame to align SNODAS data with by index. Returns ------- snodas : pd.DataFrame Processed SNODAS dataframe indexed by date. aligned_snodas : pd.DataFrame, optional SNODAS aligned to X_df index. aligned_X_df : pd.DataFrame, optional X_df aligned to SNODAS index. """ # ----------------------------- # Area weights # ----------------------------- eri_lake = 18596.39 eri_land = 81902.18 ont_lake = 15569.25 ont_land = 64465.19 mih_lake = 123626.28 mih_land = 236478.94 sup_lake = 78288.65 sup_land = 123736.05 mic_land = 118000.00 hur_land = 134100.00 # ----------------------------- # Load data # ----------------------------- snodas_data = pd.read_csv( file, sep=',', skiprows=5 ) # ----------------------------- # Create datetime column # ----------------------------- snodas_data['date'] = pd.to_datetime( snodas_data[['year', 'month', 'day']] ) # ----------------------------- # Michigan-Huron weighted merge # ----------------------------- snodas_data['michigan-huron_swe'] = ( (snodas_data['MIC'] * mic_land) + (snodas_data['HUR'] * hur_land) ) / mih_land # ----------------------------- # Rename columns # ----------------------------- snodas_data = snodas_data.rename(columns={ 'SUP': 'superior_swe', 'ERI': 'erie_swe', 'ONT': 'ontario_swe' }) # ----------------------------- # Keep desired columns # ----------------------------- snodas = snodas_data[[ 'date', 'superior_swe', 'michigan-huron_swe', 'erie_swe', 'ontario_swe' ]] # ----------------------------- # Set index # ----------------------------- snodas.set_index('date', inplace=True) return snodas
[docs] def lake_probabilities(self, file_dir, units="cms"): """ Reads probability of exceedance data for multiple Great Lakes from CSV files, converts units if requested, and combines them into a single tidy (long-format) DataFrame. Each input CSV contains probability of exceedance values (0.99–0.01) for 12 monthly columns (Jan–Dec). The function reshapes and merges all lakes into one DataFrame suitable for analysis or plotting. Parameters ---------- file_dir : str Path to the directory containing the lake probability CSV files. Expected files are: - SUP.probs.csv (Lake Superior) - MIH.probs.csv (Lakes Michigan-Huron) - ERI.probs.csv (Lake Erie) - ONT.probs.csv (Lake Ontario) units : str, optional, default="cms" Desired output units for flow values. - "cms" : cubic meters per second (original units from source files) - "mm" : millimeters per month, computed using lake surface area and number of days in each month. Returns ------- pandas.DataFrame A tidy DataFrame containing columns: - "month" : str, calendar month (Jan–Dec) - "lake" : str, lake name ("superior", "michigan-huron", "erie", "ontario") - "prob_exceedance" : float, probability of exceedance (0.99–0.01) - "value" : float, flow or equivalent value in chosen units Notes ----- - When units="mm", values are converted from m³/s to mm/month using ``value_mm = value_cms * 1000 * 86400 * days_in_month / lake_area``, where ``lake_area`` is the surface area of each lake in m². - Input CSVs are assumed to have 7 header rows before the data block, and contain a column named "Probability Of Exceedance". Example ------- >>> df = combine_lake_probabilities("/path/to/files/", units="mm") >>> df.query("lake == 'erie' and month == 'Jan'") """ # --- Lake file mapping and surface areas (m²) --- lake_files = { "superior": "SUP.probs.csv", "michigan-huron": "MIH.probs.csv", "erie": "ERI.probs.csv", "ontario": "ONT.probs.csv", } # Days in each month (non-leap year) days_in_month = { "Jan": 31, "Feb": 28, "Mar": 31, "Apr": 30, "May": 31, "Jun": 30, "Jul": 31, "Aug": 31, "Sep": 30, "Oct": 31, "Nov": 30, "Dec": 31 } # --- Internal helper to process each lake file --- def process_lake(filepath, lake_name): """ Reads and reshapes one lake's probability of exceedance data. """ # Read file, skipping header lines df = pd.read_csv(filepath, skiprows=7) # Extract probability levels and remove the column prob_values = df["Probability Of Exceedance"] df = df.drop(columns=["Probability Of Exceedance"]) # Transpose: make months rows, probabilities columns df_t = df.T df_t.columns = prob_values df_t = df_t.reset_index().rename(columns={"index": "month"}) # Optional conversion to mm/month if units.lower() == "mm": area = lake_areas[lake_name] for col in prob_values: df_t[col] = df_t.apply( lambda row: row[col] * 1000 * 86400 * days_in_month.get(row["month"], 30) / area, axis=1 ) # Reshape to long (tidy) format df_long = df_t.melt(id_vars=["month"], var_name="prob_exceedance", value_name="value") df_long["lake"] = lake_name return df_long # --- Process and combine all lakes --- all_lakes = [] for lake, filename in lake_files.items(): lake_df = process_lake(file_dir + filename, lake) all_lakes.append(lake_df) merged_df = pd.concat(all_lakes, ignore_index=True) # Clean and reorder merged_df["prob_exceedance"] = merged_df["prob_exceedance"].astype(float) merged_df = merged_df.sort_values(["lake", "prob_exceedance"]).reset_index(drop=True) merged_df = merged_df[["month", "lake", "prob_exceedance", "value"]] return merged_df