""" GHCN Daily data processor - converts raw .dly files to training format """ import os from pathlib import Path from typing import Dict, List, Optional, Tuple import numpy as np import pandas as pd from datetime import datetime, timedelta from loguru import logger class GHCNProcessor: """Process GHCN Daily files into training-ready format.""" # GHCN file format: fixed-width columns # ID (11) + Year (4) + Month (2) + Element (4) + 31 * (Value(5) + MFlag(1) + QFlag(1) + SFlag(1)) ELEMENTS = ['TMAX', 'TMIN', 'PRCP', 'SNOW', 'SNWD'] MISSING_VALUE = -9999 def __init__(self, raw_dir: Path, processed_dir: Path, stations_file: Optional[Path] = None): self.raw_dir = Path(raw_dir) self.processed_dir = Path(processed_dir) self.stations_file = stations_file self.stations_dir = self.raw_dir / "stations" self.processed_dir.mkdir(parents=True, exist_ok=True) # Load station metadata if available self.station_metadata = {} if stations_file and stations_file.exists(): self._load_station_metadata() def _load_station_metadata(self): """Load station lat/lon from stations file.""" with open(self.stations_file, 'r') as f: for line in f: # GHCN stations file format: # ID (11) + LAT (9) + LON (10) + ELEV (7) + STATE (3) + NAME (31) + ... station_id = line[0:11].strip() lat = float(line[12:20].strip()) lon = float(line[21:30].strip()) elev = float(line[31:37].strip()) if line[31:37].strip() else 0.0 name = line[41:71].strip() self.station_metadata[station_id] = { 'lat': lat, 'lon': lon, 'elevation': elev, 'name': name } def parse_dly_file(self, filepath: Path) -> pd.DataFrame: """Parse a single .dly file into a DataFrame.""" records = [] with open(filepath, 'r') as f: for line in f: if len(line) < 269: # Minimum valid line length continue station_id = line[0:11] year = int(line[11:15]) month = int(line[15:17]) element = line[17:21] if element not in self.ELEMENTS: continue # Parse 31 daily values for day in range(1, 32): try: start = 21 + (day - 1) * 8 value_str = line[start:start+5].strip() mflag = line[start+5:start+6] qflag = line[start+6:start+7] if not value_str: continue value = int(value_str) # Skip missing values and flagged quality issues if value == self.MISSING_VALUE: continue if qflag.strip() not in ['', ' ']: # Has quality flag continue # Create date try: date = datetime(year, month, day) except ValueError: continue # Invalid date (e.g., Feb 30) records.append({ 'station_id': station_id, 'date': date, 'element': element, 'value': value }) except (ValueError, IndexError): continue if not records: return pd.DataFrame() df = pd.DataFrame(records) # Pivot to get elements as columns df = df.pivot_table( index=['station_id', 'date'], columns='element', values='value', aggfunc='first' ).reset_index() # Convert units: temps from tenths of °C, precip from tenths of mm if 'TMAX' in df.columns: df['TMAX'] = df['TMAX'] / 10.0 if 'TMIN' in df.columns: df['TMIN'] = df['TMIN'] / 10.0 if 'PRCP' in df.columns: df['PRCP'] = df['PRCP'] / 10.0 if 'SNOW' in df.columns: df['SNOW'] = df['SNOW'] / 10.0 if 'SNWD' in df.columns: df['SNWD'] = df['SNWD'] / 10.0 return df def process_all_stations(self, min_years: int = 10) -> pd.DataFrame: """Process all station files and combine.""" all_data = [] station_files = list(self.stations_dir.glob("*.dly")) logger.info(f"Processing {len(station_files)} station files...") for i, filepath in enumerate(station_files): if (i + 1) % 50 == 0: logger.info(f"Processed {i + 1}/{len(station_files)} stations") df = self.parse_dly_file(filepath) if df.empty: continue # Check if station has enough data years_of_data = (df['date'].max() - df['date'].min()).days / 365 if years_of_data < min_years: continue # Add station metadata station_id = filepath.stem if station_id in self.station_metadata: meta = self.station_metadata[station_id] df['lat'] = meta['lat'] df['lon'] = meta['lon'] df['elevation'] = meta['elevation'] all_data.append(df) if not all_data: logger.error("No valid station data found!") return pd.DataFrame() combined = pd.concat(all_data, ignore_index=True) logger.success(f"Combined {len(combined)} records from {len(all_data)} stations") return combined def create_training_sequences( self, df: pd.DataFrame, input_days: int = 30, target_days: int = 14, stride: int = 7 ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """ Create training sequences for the model. Args: df: DataFrame with processed weather data input_days: Number of days of history to use as input target_days: Number of days to predict stride: Step size between sequences Returns: X: Input sequences [N, input_days, features] Y: Target sequences [N, target_days, features] meta: Station metadata [N, 4] (lat, lon, elev, day_of_year) """ sequences_X = [] sequences_Y = [] sequences_meta = [] # Features we'll use features = ['TMAX', 'TMIN', 'PRCP'] # Process each station separately stations = df['station_id'].unique() logger.info(f"Creating sequences from {len(stations)} stations...") for station_id in stations: station_df = df[df['station_id'] == station_id].copy() station_df = station_df.sort_values('date') # Ensure we have required features for feat in features: if feat not in station_df.columns: station_df[feat] = np.nan # Fill missing values with interpolation station_df[features] = station_df[features].interpolate(method='linear', limit=7) # Drop rows with too many NaN station_df = station_df.dropna(subset=['TMAX', 'TMIN']) if len(station_df) < input_days + target_days: continue # Get metadata lat = station_df['lat'].iloc[0] if 'lat' in station_df.columns else 0 lon = station_df['lon'].iloc[0] if 'lon' in station_df.columns else 0 elev = station_df['elevation'].iloc[0] if 'elevation' in station_df.columns else 0 # Create sequences values = station_df[features].values dates = station_df['date'].values for i in range(0, len(values) - input_days - target_days, stride): X = values[i:i + input_days] Y = values[i + input_days:i + input_days + target_days] # Skip if too many NaN if np.isnan(X).sum() > input_days * len(features) * 0.3: continue if np.isnan(Y).sum() > target_days * len(features) * 0.3: continue # Fill remaining NaN with mean X = np.nan_to_num(X, nan=np.nanmean(X)) Y = np.nan_to_num(Y, nan=np.nanmean(Y)) # Get day of year for the first target day target_date = pd.Timestamp(dates[i + input_days]) day_of_year = target_date.dayofyear / 365.0 # Normalize sequences_X.append(X) sequences_Y.append(Y) sequences_meta.append([lat, lon, elev, day_of_year]) if not sequences_X: logger.error("No valid sequences created!") return np.array([]), np.array([]), np.array([]) X = np.array(sequences_X, dtype=np.float32) Y = np.array(sequences_Y, dtype=np.float32) meta = np.array(sequences_meta, dtype=np.float32) logger.success(f"Created {len(X)} training sequences") logger.info(f"X shape: {X.shape}, Y shape: {Y.shape}, meta shape: {meta.shape}") return X, Y, meta def save_training_data(self, X: np.ndarray, Y: np.ndarray, meta: np.ndarray): """Save processed training data.""" output_dir = self.processed_dir / "training" output_dir.mkdir(parents=True, exist_ok=True) np.save(output_dir / "X.npy", X) np.save(output_dir / "Y.npy", Y) np.save(output_dir / "meta.npy", meta) logger.success(f"Saved training data to {output_dir}") # Save normalization stats stats = { 'X_mean': X.mean(axis=(0, 1)), 'X_std': X.std(axis=(0, 1)), 'Y_mean': Y.mean(axis=(0, 1)), 'Y_std': Y.std(axis=(0, 1)), } np.savez(output_dir / "stats.npz", **stats) def main(): """Process GHCN data for training.""" from pathlib import Path base_dir = Path(__file__).parent.parent.parent raw_dir = base_dir / "data" / "raw" / "ghcn_daily" processed_dir = base_dir / "data" / "processed" stations_file = raw_dir / "ghcnd-stations.txt" processor = GHCNProcessor(raw_dir, processed_dir, stations_file) # Process all stations df = processor.process_all_stations(min_years=10) if df.empty: logger.error("No data to process!") return # Save intermediate CSV for inspection df.to_parquet(processed_dir / "ghcn_combined.parquet") logger.info(f"Saved combined data to {processed_dir / 'ghcn_combined.parquet'}") # Create training sequences X, Y, meta = processor.create_training_sequences( df, input_days=30, target_days=14, stride=7 ) if len(X) > 0: processor.save_training_data(X, Y, meta) if __name__ == "__main__": main()