Spaces:
Runtime error
Runtime error
File size: 7,854 Bytes
d64c823 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 | """
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()
|