Spaces:
Runtime error
Runtime error
| """ | |
| 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() | |