CirealKiller/Source_Identification_WAQI
0
1"""2Preprocessing module for air pollution source identification.3 4Provides feature engineering: cyclical time encoding, wind vector5decomposition, pollution ratios (PM2.5 / PM10), nitrogen ratio (NO2/NO),6and normalization of gaseous pollutants.7"""8 9import pandas as pd10import numpy as np11from sklearn.preprocessing import StandardScaler12 13# Gaseous pollutant columns for normalization14GASEOUS_COLS = ["co", "no", "no2", "o3", "so2", "nh3"]15 16# Canonical column name mapping (delhi_aqi uses pm2_5, pm10)17COLUMN_ALIASES = {18 "pm2_5": "PM25",19 "pm25": "PM25",20 "PM2.5": "PM25",21 "pm10": "PM10",22 "PM10": "PM10",23}24 25 26def _normalize_column_names(df: pd.DataFrame) -> pd.DataFrame:27 """28 Rename columns to canonical names (e.g., pm2_5 -> PM25).29 30 Args:31 df: Input DataFrame.32 33 Returns:34 DataFrame with standardized column names.35 """36 df = df.copy()37 for alias, canonical in COLUMN_ALIASES.items():38 if alias in df.columns and canonical not in df.columns:39 df[canonical] = df[alias]40 elif alias in df.columns and canonical in df.columns:41 df[canonical] = df[canonical].fillna(df[alias])42 return df43 44 45def _ensure_optional_columns(df: pd.DataFrame) -> pd.DataFrame:46 """47 Add wind and fire_count with default values if missing.48 49 Args:50 df: Input DataFrame.51 52 Returns:53 DataFrame with wind_dir, wind_speed, fire_count present.54 """55 df = df.copy()56 if "wind_dir" not in df.columns:57 df["wind_dir"] = 0.058 if "wind_speed" not in df.columns:59 df["wind_speed"] = 0.060 if "fire_count" not in df.columns:61 df["fire_count"] = 062 return df63 64 65def add_cyclical_month(df: pd.DataFrame, date_col: str = "date") -> pd.DataFrame:66 """67 Encode month as cyclical features using sin/cos to preserve periodicity.68 69 Args:70 df: DataFrame with a date column.71 date_col: Name of the date column (parsed if string).72 73 Returns:74 DataFrame with added columns month_sin and month_cos.75 """76 df = df.copy()77 if not pd.api.types.is_datetime64_any_dtype(df[date_col]):78 df[date_col] = pd.to_datetime(df[date_col], errors="coerce")79 months = df[date_col].dt.month80 df["month_sin"] = np.sin(2 * np.pi * months / 12)81 df["month_cos"] = np.cos(2 * np.pi * months / 12)82 return df83 84 85def add_wind_components(86 df: pd.DataFrame,87 wind_speed_col: str = "wind_speed",88 wind_dir_col: str = "wind_dir",89) -> pd.DataFrame:90 """91 Decompose wind into u (east-west) and v (north-south) components.92 93 Convention: wind_dir is direction wind comes FROM (meteorological).94 u positive = wind from west; v positive = wind from south.95 96 Args:97 df: DataFrame with wind_speed and wind_dir (degrees).98 wind_speed_col: Name of wind speed column.99 wind_dir_col: Name of wind direction column (degrees, 0=N).100 101 Returns:102 DataFrame with added columns wind_u and wind_v.103 """104 df = df.copy()105 speed = df[wind_speed_col].astype(float)106 deg = np.deg2rad(df[wind_dir_col].astype(float))107 df["wind_u"] = -speed * np.sin(deg)108 df["wind_v"] = -speed * np.cos(deg)109 return df110 111 112def add_pm_ratio(113 df: pd.DataFrame,114 pm25_col: str = "PM25",115 pm10_col: str = "PM10",116 ratio_name: str = "pm_ratio",117) -> pd.DataFrame:118 """119 Add pollution ratio PM2.5 / PM10, with safe division (avoid inf/NaN).120 121 Args:122 df: DataFrame with PM25 and PM10 columns.123 pm25_col: Name of PM2.5 column.124 pm10_col: Name of PM10 column.125 ratio_name: Name of the output ratio column.126 127 Returns:128 DataFrame with added ratio column (NaN where PM10 is 0).129 """130 df = df.copy()131 pm25 = df[pm25_col].astype(float)132 pm10 = df[pm10_col].astype(float)133 df[ratio_name] = np.where(pm10 > 0, pm25 / pm10, np.nan)134 return df135 136 137def add_nitrogen_ratio(138 df: pd.DataFrame,139 no_col: str = "no",140 no2_col: str = "no2",141 ratio_name: str = "nitrogen_ratio",142) -> pd.DataFrame:143 """144 Add nitrogen ratio NO2 / NO (tracer for combustion/oxidation state).145 146 Args:147 df: DataFrame with NO and NO2 columns.148 no_col: Name of NO column.149 no2_col: Name of NO2 column.150 ratio_name: Name of the output ratio column.151 152 Returns:153 DataFrame with added ratio column (NaN where NO is 0).154 """155 df = df.copy()156 no = df[no_col].astype(float)157 no2 = df[no2_col].astype(float)158 df[ratio_name] = np.where(no > 0, no2 / no, np.nan)159 return df160 161 162def normalize_gaseous_pollutants(163 df: pd.DataFrame,164 columns: list = None,165 scaler: StandardScaler = None,166) -> tuple:167 """168 Normalize gaseous pollutants (co, no, no2, o3, so2, nh3) to account for169 different scales. Uses StandardScaler (z-score normalization).170 171 Args:172 df: DataFrame containing gaseous pollutant columns.173 columns: List of column names to normalize; default GASEOUS_COLS.174 scaler: Fitted StandardScaler; if None, fit on data and return.175 176 Returns:177 Tuple of (transformed DataFrame, fitted StandardScaler).178 """179 df = df.copy()180 cols = columns or GASEOUS_COLS181 available = [c for c in cols if c in df.columns]182 if not available:183 return df, scaler184 X = df[available].astype(float)185 X = X.fillna(X.median())186 if scaler is None:187 scaler = StandardScaler()188 X_scaled = scaler.fit_transform(X)189 else:190 fit_cols = getattr(scaler, "feature_names_in_", cols)191 available = [c for c in fit_cols if c in df.columns]192 if not available:193 return df, scaler194 X = df[available].astype(float).fillna(0)195 X_scaled = scaler.transform(X)196 df[available] = X_scaled197 return df, scaler198 199 200def preprocess(201 df: pd.DataFrame,202 date_col: str = "date",203 wind_speed_col: str = "wind_speed",204 wind_dir_col: str = "wind_dir",205 pm25_col: str = "PM25",206 pm10_col: str = "PM10",207 scaler: StandardScaler = None,208) -> tuple:209 """210 Apply full preprocessing pipeline: cyclical month, wind u/v, PM ratio,211 nitrogen ratio, and gaseous normalization.212 213 Args:214 df: Raw DataFrame with date, PM, gaseous pollutants, optionally wind215 and fire_count.216 date_col: Name of date column.217 wind_speed_col: Name of wind speed column.218 wind_dir_col: Name of wind direction column.219 pm25_col: Name of PM2.5 column (or pm2_5).220 pm10_col: Name of PM10 column.221 scaler: Fitted StandardScaler for gaseous norm; None = fit on data.222 223 Returns:224 Tuple of (preprocessed DataFrame, fitted StandardScaler).225 Scaler is returned for persistence; use for prediction.226 """227 df = _normalize_column_names(df)228 df = _ensure_optional_columns(df)229 df = add_cyclical_month(df, date_col=date_col)230 df = add_wind_components(231 df, wind_speed_col=wind_speed_col, wind_dir_col=wind_dir_col232 )233 df = add_pm_ratio(df, pm25_col="PM25", pm10_col="PM10")234 # Add nitrogen ratio if no/no2 present235 if "no" in df.columns and "no2" in df.columns:236 df = add_nitrogen_ratio(df)237 else:238 df["nitrogen_ratio"] = np.nan239 # Normalize gaseous pollutants (add missing with 0 for consistency)240 for c in GASEOUS_COLS:241 if c not in df.columns:242 df[c] = 0.0243 df, scaler = normalize_gaseous_pollutants(df, columns=GASEOUS_COLS, scaler=scaler)244 return df, scaler245 246 247def get_feature_columns(exclude: list = None) -> list:248 """249 Return the list of feature column names used by the model.250 251 Includes cyclical time, wind, ratios, gaseous pollutants (normalized),252 and PM metrics.253 254 Args:255 exclude: Optional list of column names to exclude (e.g. for ablation).256 257 Returns:258 List of feature names.259 """260 cols = [261 "month_sin",262 "month_cos",263 "wind_u",264 "wind_v",265 "pm_ratio",266 "nitrogen_ratio",267 "co",268 "no",269 "no2",270 "o3",271 "so2",272 "nh3",273 "PM25",274 "PM10",275 "wind_speed",276 "wind_dir",277 "fire_count",278 ]279 if exclude:280 cols = [c for c in cols if c not in exclude]281 return cols282