""" Load and process Cyclone Mocha track data from IBTrACS. Falls back to manually curated track points if parsing fails. """ import numpy as np import pandas as pd import logging from src.config import CYCLONE_TRACK_FILE, PROCESSED_CYCLONE_TRACK, PROCESSED_DIR logger = logging.getLogger(__name__) # Curated Mocha track from Wikipedia / IMD bulletins — used as fallback MANUAL_MOCHA_TRACK = [ {"timestamp": "2023-05-08T06:00:00Z", "lat": 10.0, "lon": 87.0, "wind_knots": 30, "pressure_hpa": 998, "category": "Depression", "rmw_nm": 80}, {"timestamp": "2023-05-09T06:00:00Z", "lat": 10.5, "lon": 87.5, "wind_knots": 45, "pressure_hpa": 992, "category": "Cyclonic Storm", "rmw_nm": 60}, {"timestamp": "2023-05-10T06:00:00Z", "lat": 11.0, "lon": 87.8, "wind_knots": 65, "pressure_hpa": 980, "category": "Severe CS", "rmw_nm": 45}, {"timestamp": "2023-05-11T06:00:00Z", "lat": 12.0, "lon": 88.0, "wind_knots": 85, "pressure_hpa": 968, "category": "Very Severe CS", "rmw_nm": 30}, {"timestamp": "2023-05-12T00:00:00Z", "lat": 13.5, "lon": 89.0, "wind_knots": 120, "pressure_hpa": 940, "category": "Extremely Severe CS", "rmw_nm": 15}, {"timestamp": "2023-05-12T12:00:00Z", "lat": 15.0, "lon": 90.0, "wind_knots": 140, "pressure_hpa": 930, "category": "Super Cyclonic Storm", "rmw_nm": 12}, {"timestamp": "2023-05-13T06:00:00Z", "lat": 17.0, "lon": 91.5, "wind_knots": 150, "pressure_hpa": 920, "category": "Super Cyclonic Storm", "rmw_nm": 10}, {"timestamp": "2023-05-13T18:00:00Z", "lat": 18.5, "lon": 92.5, "wind_knots": 155, "pressure_hpa": 916, "category": "Super Cyclonic Storm", "rmw_nm": 10}, {"timestamp": "2023-05-14T06:00:00Z", "lat": 20.0, "lon": 93.0, "wind_knots": 140, "pressure_hpa": 928, "category": "Super Cyclonic Storm", "rmw_nm": 12}, {"timestamp": "2023-05-14T12:00:00Z", "lat": 21.0, "lon": 93.5, "wind_knots": 80, "pressure_hpa": 970, "category": "Severe CS", "rmw_nm": 20}, ] def load_cyclone_track(): """ Load Cyclone Mocha track from IBTrACS CSV (or manual fallback). Returns: DataFrame with columns: timestamp, lat, lon, wind_knots, pressure_hpa, category. """ logger.info("Loading cyclone track data...") if CYCLONE_TRACK_FILE.exists(): try: df = _load_from_ibtracs() if df is not None and len(df) > 0: logger.info(f"Loaded {len(df)} track points from IBTrACS") _save_processed(df) return df except Exception as e: logger.warning(f"Error reading IBTrACS file: {e}") logger.info("Falling back to manual track data...") else: logger.info("IBTrACS file not found, using manual track data...") df = pd.DataFrame(MANUAL_MOCHA_TRACK) df["timestamp"] = pd.to_datetime(df["timestamp"]) logger.info(f"Using manual track data: {len(df)} points") _save_processed(df) return df def _load_from_ibtracs(): """Parse IBTrACS CSV and extract Cyclone Mocha records.""" # IBTrACS has a units row after the header — skip it try: df = pd.read_csv(CYCLONE_TRACK_FILE, skiprows=[1], low_memory=False) except Exception: df = pd.read_csv(CYCLONE_TRACK_FILE, low_memory=False) logger.info(f"IBTrACS columns (first 15): {list(df.columns[:15])}") # Find Mocha by name mocha = None for name_col in ["NAME", "name"]: if name_col in df.columns: mocha = df[df[name_col].astype(str).str.upper().str.strip() == "MOCHA"] if len(mocha) > 0: break if mocha is None or len(mocha) == 0: # Try SID pattern if "SID" in df.columns: mocha = df[ df["SID"].astype(str).str.contains("2023", na=False) & df["SID"].astype(str).str.contains("NI", na=False) ] if mocha is None or len(mocha) == 0: logger.warning("Could not find Cyclone Mocha in IBTrACS file") return None logger.info(f"Found {len(mocha)} Mocha records in IBTrACS") # Map to standard columns result = pd.DataFrame() for col in ["ISO_TIME", "iso_time"]: if col in mocha.columns: result["timestamp"] = pd.to_datetime(mocha[col].values) break for col in ["LAT", "lat"]: if col in mocha.columns: result["lat"] = pd.to_numeric(mocha[col].values, errors="coerce") break for col in ["LON", "lon"]: if col in mocha.columns: result["lon"] = pd.to_numeric(mocha[col].values, errors="coerce") break for col in ["WMO_WIND", "USA_WIND", "wmo_wind"]: if col in mocha.columns: result["wind_knots"] = pd.to_numeric(mocha[col].values, errors="coerce") break for col in ["WMO_PRES", "USA_PRES", "wmo_pres"]: if col in mocha.columns: result["pressure_hpa"] = pd.to_numeric(mocha[col].values, errors="coerce") break # Extract radius of maximum wind (nautical miles) for Holland model for col in ["USA_RMW", "usa_rmw", "BOM_RMW", "REUNION_RMW"]: if col in mocha.columns: rmw_series = pd.to_numeric(mocha[col], errors="coerce").reset_index(drop=True) if rmw_series.notna().any(): result["rmw_nm"] = rmw_series.values logger.info(f"Extracted Rmax from {col}: {rmw_series.notna().sum()} valid values") break # Empirical fallback: Knaff & Zehr (2007) if no agency Rmax available if "rmw_nm" not in result.columns or result["rmw_nm"].isna().all(): logger.warning("No agency Rmax found — estimating via Knaff & Zehr (2007)") wind = result.get("wind_knots", pd.Series(dtype=float)) lat = result.get("lat", pd.Series(dtype=float)) result["rmw_nm"] = 66.785 - 0.09102 * wind + 1.0619 * (lat - 25.0) result["rmw_nm"] = result["rmw_nm"].clip(lower=5, upper=120) else: # Interpolate isolated NaN gaps in agency data result["rmw_nm"] = result["rmw_nm"].interpolate(method="linear", limit_direction="both") result = result.dropna(subset=["lat", "lon"]) result["category"] = result["wind_knots"].apply(_classify_cyclone) result = result.sort_values("timestamp").reset_index(drop=True) return result def _classify_cyclone(wind_knots): """Classify cyclone intensity per IMD scale.""" if pd.isna(wind_knots): return "Unknown" if wind_knots >= 120: return "Super Cyclonic Storm" if wind_knots >= 90: return "Extremely Severe CS" if wind_knots >= 64: return "Very Severe CS" if wind_knots >= 48: return "Severe CS" if wind_knots >= 34: return "Cyclonic Storm" if wind_knots >= 28: return "Deep Depression" return "Depression" def _save_processed(df): """Save processed track data and print summary.""" PROCESSED_DIR.mkdir(parents=True, exist_ok=True) df.to_csv(PROCESSED_CYCLONE_TRACK, index=False) logger.info(f"Saved processed track → {PROCESSED_CYCLONE_TRACK}") if "wind_knots" in df.columns: mx = df["wind_knots"].max() logger.info(f"Peak intensity: {mx} knots ({_classify_cyclone(mx)})") if "pressure_hpa" in df.columns: mn = df["pressure_hpa"].min() logger.info(f"Minimum pressure: {mn} hPa") if "rmw_nm" in df.columns: rmw_valid = df["rmw_nm"].dropna() logger.info( f"Rmax (RMW): {len(rmw_valid)} values, " f"range {rmw_valid.min():.0f}–{rmw_valid.max():.0f} nm" ) logger.info(f"Track span: {df['timestamp'].min()} → {df['timestamp'].max()}") if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") load_cyclone_track()