"""NOAA HURDAT2 - historial de huracanes Atlantico. Dataset: https://www.nhc.noaa.gov/data/#hurdat Format spec: https://www.nhc.noaa.gov/data/hurdat/hurdat2-format.pdf Descarga el archivo en data/hurdat2.txt en el primer uso (lazy). Re-descarga si tiene mas de 365 dias. Para una direccion dada (lat/lng), devuelve huracanes que pasaron a menos de N millas (default 150) en los ultimos K anos (default 20). Returns: { "lookback_years": 20, "max_distance_mi": 150, "total_hurricanes_nearby": , "hurricanes": [ {"name": "Ian", "year": 2022, "category": 4, "max_wind_mph": 155, "closest_pass_miles": 12}, ... ] } """ from __future__ import annotations import math import os import time from datetime import datetime from pathlib import Path import requests from .base import FetcherError, USER_AGENT, DEFAULT_TIMEOUT # URLs candidatas (NOAA renombra el archivo cada ano). # Si todas fallan, FetcherError. HURDAT2_URL_CANDIDATES = [ "https://www.nhc.noaa.gov/data/hurdat/hurdat2-1851-2024-040425.txt", "https://www.nhc.noaa.gov/data/hurdat/hurdat2-1851-2023-051124.txt", "https://www.nhc.noaa.gov/data/hurdat/hurdat2-1851-2022-050423.txt", ] HURDAT2_MAX_AGE_DAYS = 365 def _saffir_simpson(max_wind_mph: float) -> int: """Categoria Saffir-Simpson basada en max sustained wind (mph). 0 = TS (no huracan).""" if max_wind_mph >= 157: return 5 if max_wind_mph >= 130: return 4 if max_wind_mph >= 111: return 3 if max_wind_mph >= 96: return 2 if max_wind_mph >= 74: return 1 return 0 # tropical storm or less def _haversine_mi(lat1: float, lng1: float, lat2: float, lng2: float) -> float: """Distancia great-circle entre dos puntos en millas.""" R_MI = 3958.8 p1 = math.radians(lat1) p2 = math.radians(lat2) dp = math.radians(lat2 - lat1) dl = math.radians(lng2 - lng1) a = math.sin(dp/2)**2 + math.cos(p1) * math.cos(p2) * math.sin(dl/2)**2 return 2 * R_MI * math.asin(math.sqrt(a)) def _parse_coord(s: str) -> float | None: """Parse '26.5N' o '80.3W' a float (West/South negativos).""" s = s.strip() if not s or len(s) < 2: return None try: val = float(s[:-1]) d = s[-1].upper() if d in ("S", "W"): val = -val return val except ValueError: return None def _download_hurdat2(dest_path: Path) -> None: """Intenta descargar HURDAT2 desde varias URLs candidatas.""" dest_path.parent.mkdir(parents=True, exist_ok=True) last_err = None for url in HURDAT2_URL_CANDIDATES: try: r = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=30) if r.status_code == 200 and len(r.text) > 10000: dest_path.write_text(r.text, encoding="utf-8") return last_err = f"HTTP {r.status_code} from {url}" except requests.RequestException as e: last_err = f"{url}: {e}" continue raise FetcherError(f"No pude descargar HURDAT2 desde ninguna URL. Ultimo error: {last_err}") def _ensure_hurdat2_local(local_path: str | Path) -> Path: """Garantiza que el archivo local exista y este fresco. Descarga si hace falta.""" p = Path(local_path) needs_download = ( not p.exists() or (time.time() - p.stat().st_mtime) / 86400 > HURDAT2_MAX_AGE_DAYS ) if needs_download: _download_hurdat2(p) return p def fetch_hurricanes( lat: float, lng: float, years_back: int = 20, max_distance_mi: float = 150.0, hurdat2_path: str | Path = "data/hurdat2.txt", ) -> dict: """Busca huracanes que pasaron cerca de (lat, lng) en los ultimos N anos. 'Cerca' = al menos un track-point del huracan estuvo a <= max_distance_mi. Solo cuenta sistemas que alcanzaron category 1+ (max_wind >= 74 mph) en algun momento de su track. """ if lat is None or lng is None: raise FetcherError("lat/lng requeridos") p = _ensure_hurdat2_local(hurdat2_path) current_year = datetime.now().year min_year = current_year - years_back text = p.read_text(encoding="utf-8", errors="replace") lines = text.splitlines() hurricanes_nearby = [] i = 0 n = len(lines) while i < n: line = lines[i].strip() if not line: i += 1 continue parts = [x.strip() for x in line.split(",")] # Header: AL112022, IAN, 65, if len(parts) >= 3 and parts[0].startswith("AL") and len(parts[0]) >= 8: atcf_id = parts[0] name = parts[1] try: num_records = int(parts[2]) year = int(atcf_id[4:8]) except (ValueError, IndexError): i += 1 continue if year < min_year: # Skip todos los track lines de este huracan i += 1 + num_records continue max_wind_kt = 0 min_dist_mi = float("inf") for j in range(num_records): tl_idx = i + 1 + j if tl_idx >= n: break track_parts = [x.strip() for x in lines[tl_idx].split(",")] if len(track_parts) < 7: continue # 0:date 1:time 2:record_id 3:status 4:lat 5:lng 6:wind tlat = _parse_coord(track_parts[4]) tlng = _parse_coord(track_parts[5]) try: wind = int(track_parts[6]) except ValueError: wind = 0 if tlat is None or tlng is None: continue dist = _haversine_mi(lat, lng, tlat, tlng) if dist < min_dist_mi: min_dist_mi = dist if wind > max_wind_kt: max_wind_kt = wind max_wind_mph = max_wind_kt * 1.15078 # kt -> mph category = _saffir_simpson(max_wind_mph) # Solo contamos huracanes (cat 1+) que pasaron cerca if category >= 1 and min_dist_mi <= max_distance_mi: hurricanes_nearby.append({ "name": name if name else "UNNAMED", "year": year, "category": category, "max_wind_mph": int(round(max_wind_mph)), "closest_pass_miles": int(round(min_dist_mi)), }) i += 1 + num_records else: i += 1 # Ordenar: mas reciente y mas fuerte primero hurricanes_nearby.sort(key=lambda h: (-h["year"], -h["category"])) return { "lookback_years": years_back, "max_distance_mi": max_distance_mi, "total_hurricanes_nearby": len(hurricanes_nearby), "hurricanes": hurricanes_nearby, "source": "NOAA HURDAT2", }