"""GDACS Tropical Cyclone API client.
Fetches active and historical tropical cyclone events, advisory
timelines, and country-level impact data from the GDACS REST API.
No authentication required.
API docs: https://www.gdacs.org/gdacsapi/swagger/index.html
Endpoints
---------
- Event search (``/events/geteventlist/SEARCH``): paginated, filterable
by date range and alert level. One row per TC event.
- Event/episode detail (``/events/geteventdata`` and
``/events/getepisodedata``): full JSON for one event or episode,
including resource URLs (timeline, per-buffer impact).
- Timeline: per-advisory snapshots with position, wind speed, and
population exposure (``pop39``/``pop74``). Available 2015+.
- Impact buffers: country-level cumulative population exposure within
the cumulative wind footprint. Returns data ~2022+.
"""
import logging
from collections import Counter
from typing import Any, Dict, List, Literal, Optional, Tuple
import geopandas as gpd
import numpy as np
import pandas as pd
import pandera.pandas as pa
import requests
from ocha_lens.utils.storm import _to_gdf
logger = logging.getLogger(__name__)
BASE_URL = "https://www.gdacs.org/gdacsapi/api"
_TIMEOUT = 30
_PAGE_SIZE = 100
_EVENT_TYPE = "TC"
_BUFFER_KEYS = ("buffer39", "buffer74")
# Mapping of GDACS proprietary country codes to standard ISO 3166-1
# alpha-3. Most GDACS GMI_CNTRY values ARE valid ISO3 (USA, MEX, BHS,
# etc.); the entries here are the exceptions — territories and
# dependencies where GDACS uses an X-prefixed internal code. Add new
# entries here as new GDACS proprietary codes surface in real data.
GDACS_PROPRIETARY_TO_ISO3 = {
"XJE": "JEY", # Jersey (British Crown Dependency)
# "XGG": "GGY", # Guernsey — add if encountered
# "XIM": "IMN", # Isle of Man — add if encountered
}
[docs]
def to_iso3(gdacs_country_code: str) -> str:
"""Map a GDACS GMI_CNTRY code to standard ISO 3166-1 alpha-3.
Most GDACS codes are already valid ISO3 — pass-through. Only the
X-prefixed proprietary codes (see ``GDACS_PROPRIETARY_TO_ISO3``)
get remapped.
"""
return GDACS_PROPRIETARY_TO_ISO3.get(
gdacs_country_code, gdacs_country_code
)
AlertLevel = Literal["Green", "Orange", "Red"]
# Alert levels are queried one at a time and results deduped on
# eventid. The combined-alertlevel API filter is unreliable, and a
# storm's alert level can change during its lifetime — without
# per-level iteration + dedup, the same event would appear multiple
# times or be missed entirely.
_ALERT_LEVELS: Tuple[AlertLevel, ...] = ("Green", "Orange", "Red")
_TIMELINE_NUMERIC_COLS = [
"latitude",
"longitude",
"wind_speed",
"wind_gusts",
"pressure",
"pop",
"pop39",
"pop74",
"popstormsurge",
"alertscore",
"windrad_nm_34kt_ne",
"windrad_nm_34kt_se",
"windrad_nm_34kt_sw",
"windrad_nm_34kt_nw",
"windrad_nm_50kt_ne",
"windrad_nm_50kt_se",
"windrad_nm_50kt_sw",
"windrad_nm_50kt_nw",
"windrad_nm_64kt_ne",
"windrad_nm_64kt_se",
"windrad_nm_64kt_sw",
"windrad_nm_64kt_nw",
]
_TIMELINE_COLS = (
[
"advisory_number",
"name",
"advisory_datetime",
"actual",
"current",
"alertcolor",
"country",
]
+ _TIMELINE_NUMERIC_COLS
+ ["storm_status"]
)
_EVENT_ROW_KEYS = [
"eventid",
"name",
"alert_level",
"episode_alert",
"iso3",
"country",
"from_date",
"to_date",
"is_current",
"severity_kmh",
"severity_text",
"source",
"longitude",
"latitude",
]
# ---------------------------------------------------------------------------
# Schemas
# ---------------------------------------------------------------------------
#
# Types chosen to map cleanly to the SQL columns the pipeline writes
# (see ds-storms-pipeline/src/schemas/sql/gdacs_exposure.sql):
# pop_affected → INTEGER ⇒ pa.Column(int)
# distance_km → DOUBLE ⇒ pa.Column(float)
# iso3 → VARCHAR(3) ⇒ pa.Column(str)
# valid_time → TIMESTAMP ⇒ pa.Column(pd.Timestamp)
#
# Validation is called at the return point of each public function.
EVENT_SCHEMA = pa.DataFrameSchema(
{
"eventid": pa.Column(int, nullable=False),
"name": pa.Column(str, nullable=True),
"alert_level": pa.Column(
str,
pa.Check.isin(["Green", "Orange", "Red", ""]),
nullable=True,
),
"episode_alert": pa.Column(str, nullable=True),
"iso3": pa.Column(str, nullable=True),
"country": pa.Column(str, nullable=True),
"from_date": pa.Column(str, nullable=True),
"to_date": pa.Column(str, nullable=True),
"is_current": pa.Column(bool, nullable=False),
"severity_kmh": pa.Column(float, nullable=True),
"severity_text": pa.Column(str, nullable=True),
"source": pa.Column(
str,
pa.Check.isin(["NOAA", "JTWC", ""]),
nullable=True,
),
"geometry": pa.Column(gpd.array.GeometryDtype, nullable=True),
},
strict=True,
coerce=True,
)
TIMELINE_SCHEMA = pa.DataFrameSchema(
{
"advisory_number": pa.Column("Int64", nullable=True),
"name": pa.Column(str, nullable=True),
"advisory_datetime": pa.Column(pd.Timestamp, nullable=True),
"actual": pa.Column(str, nullable=True),
"current": pa.Column(str, nullable=True),
"alertcolor": pa.Column(str, nullable=True),
"country": pa.Column(str, nullable=True),
**{c: pa.Column(float, nullable=True) for c in _TIMELINE_NUMERIC_COLS},
"storm_status": pa.Column(str, nullable=True),
},
strict=True,
coerce=True,
)
ADM0_EXPOSURE_SCHEMA = pa.DataFrameSchema(
{
"iso3": pa.Column(str, nullable=False),
"country": pa.Column(str, nullable=True),
# pop_affected legitimately missing when no population is
# exposed to this wind threshold — preserved as null
"pop_affected": pa.Column("Int64", pa.Check.ge(0), nullable=True),
"distance_km": pa.Column(float, pa.Check.ge(0), nullable=True),
},
strict=True,
coerce=True,
)
ADM1_EXPOSURE_SCHEMA = pa.DataFrameSchema(
{
"iso3": pa.Column(str, nullable=False),
"country": pa.Column(str, nullable=True),
"fips_admin": pa.Column(str, nullable=True),
"gmi_admin": pa.Column(str, nullable=False),
"admin_name": pa.Column(str, nullable=True),
"admin_type": pa.Column(str, nullable=True),
"pop_admin": pa.Column("Int64", pa.Check.ge(0), nullable=True),
"pop_affected": pa.Column("Int64", pa.Check.ge(0), nullable=True),
"distance_km": pa.Column(float, pa.Check.ge(0), nullable=True),
},
strict=True,
coerce=True,
)
_session: Optional[requests.Session] = None
def _get_session() -> requests.Session:
global _session
if _session is None:
_session = requests.Session()
return _session
def _get_json(
url: str,
params: Optional[Dict[str, Any]] = None,
*,
allow_empty: bool = False,
) -> Any:
"""GET a URL and return parsed JSON, raising on HTTP error.
If ``allow_empty`` is True, returns None for an empty response
body instead of raising — used by paginated endpoints where an
empty body signals end-of-pages.
"""
r = _get_session().get(url, params=params, timeout=_TIMEOUT)
r.raise_for_status()
if allow_empty and not r.text.strip():
return None
return r.json()
[docs]
def get_events(
from_date: Optional[str] = None,
to_date: Optional[str] = None,
alert_levels: Optional[List[AlertLevel]] = None,
source: Optional[Literal["NOAA", "JTWC"]] = None,
page_size: int = _PAGE_SIZE,
) -> gpd.GeoDataFrame:
"""Fetch GDACS tropical cyclone events.
Auto-paginates the GDACS event search endpoint, iterating over
alert levels separately and deduplicating on ``eventid``.
Parameters
----------
from_date, to_date : str, optional
ISO date strings ``"YYYY-MM-DD"``. Default: GDACS API defaults.
alert_levels : list of {"Green", "Orange", "Red"}, optional
Alert levels to query. Default: all three.
source : {"NOAA", "JTWC"}, optional
Filter to events tagged with this source. Applied client-side
because the API source filter is unreliable. Default: no
filter.
page_size : int, default 100
GDACS caps page size at 100; do not exceed.
Returns
-------
geopandas.GeoDataFrame
One row per event. Empty GeoDataFrame with the expected
columns if no events match.
"""
levels = list(alert_levels) if alert_levels else list(_ALERT_LEVELS)
seen: Dict[int, Dict[str, Any]] = {}
for level in levels:
page = 1
while True:
params: Dict[str, Any] = {
"eventlist": _EVENT_TYPE,
"alertlevel": level,
"pageSize": page_size,
"pageNumber": page,
}
if from_date:
params["fromDate"] = from_date
if to_date:
params["toDate"] = to_date
data = _get_json(
f"{BASE_URL}/events/geteventlist/SEARCH",
params=params,
allow_empty=True,
)
if data is None:
break
features = data.get("features", [])
if not features:
break
for f in features:
row = _event_feature_to_row(f)
if source and row["source"] != source:
continue
if row["eventid"] in seen:
continue
seen[row["eventid"]] = row
if len(features) < page_size:
break
page += 1
if not seen:
return _to_gdf(pd.DataFrame(columns=_EVENT_ROW_KEYS))
return EVENT_SCHEMA.validate(_to_gdf(pd.DataFrame(list(seen.values()))))
[docs]
def get_event_detail(eventid: int) -> Dict[str, Any]:
"""Fetch full event detail JSON.
Includes the episode list and resource URLs for the timeline and
per-buffer impact endpoints.
"""
return _get_json(
f"{BASE_URL}/events/geteventdata",
params={"eventtype": _EVENT_TYPE, "eventid": eventid},
)
class NoEpisodesError(ValueError):
"""Raised when a GDACS event has no episodes in its detail."""
class EpisodeUrlFormatError(ValueError):
"""Raised when GDACS's episode `details` URL doesn't contain a
parseable `episodeid=` query parameter (API contract change)."""
[docs]
def latest_episode_id(event_detail: Dict[str, Any]) -> int:
"""Pull the latest episode id out of an event-detail JSON.
GDACS's ``properties.episodes`` is a list of ``{details, ...}``
dicts; the episode id is embedded in the ``details`` URL as a
query param. This helper hides that GDACS-internal quirk.
Raises
------
NoEpisodesError
Event has no episodes (legitimately new event with no
advisories yet — callers may want to handle this).
EpisodeUrlFormatError
Episode URL is malformed — likely a GDACS API contract
change; should surface loudly rather than be skipped.
"""
episodes = event_detail["properties"]["episodes"]
if not episodes:
raise NoEpisodesError("event has no episodes")
url = episodes[-1]["details"]
if "episodeid=" not in url:
raise EpisodeUrlFormatError(
f"`episodeid=` not found in details URL: {url!r}"
)
return int(url.split("episodeid=")[-1].split("&")[0])
[docs]
def get_episode_detail(eventid: int, episodeid: int) -> Dict[str, Any]:
"""Fetch full episode detail JSON.
Each episode is one model run (issuance), typically every 6 hours.
The structure mirrors :func:`get_event_detail`, but resource URLs
point to episode-specific data.
"""
return _get_json(
f"{BASE_URL}/events/getepisodedata",
params={
"eventtype": _EVENT_TYPE,
"eventid": eventid,
"episodeid": episodeid,
},
)
class NoTimelineError(ValueError):
"""Raised when a GDACS event has no timeline resource URL in its
impact entries. Distinguishable from "timeline exists but empty"
(which returns an empty DataFrame, validated by TIMELINE_SCHEMA).
"""
[docs]
def get_timeline(
eventid: int, detail: Optional[Dict[str, Any]] = None
) -> pd.DataFrame:
"""Fetch the advisory timeline for one TC event.
Each row is one advisory with position, wind speed, population
exposure (``pop39``/``pop74``), and quadrant wind radii.
``pop39``/``pop74`` are instantaneous snapshots — not cumulative.
Parameters
----------
eventid : int
detail : dict, optional
Pre-fetched output of :func:`get_event_detail` for this
eventid. Pass it to skip the internal detail fetch when the
caller already has it (saves one HTTP round-trip).
Raises
------
NoTimelineError
Event has no timeline resource URL in its impacts list
(legitimately rare — most TC events have one).
Returns
-------
pandas.DataFrame
Sorted by ``advisory_number`` ascending. Empty DataFrame
(with expected columns) if the timeline endpoint returns
no items — pandera validates the row structure either way.
"""
if detail is None:
detail = get_event_detail(eventid)
impacts = detail["properties"]["impacts"]
timeline_url = None
for imp in impacts:
url = imp.get("resource", {}).get("timeline")
if url:
timeline_url = url
break
if timeline_url is None:
raise NoTimelineError(f"no timeline resource for event {eventid}")
items = _get_json(timeline_url)["channel"]["item"]
if not items:
return TIMELINE_SCHEMA.validate(pd.DataFrame(columns=_TIMELINE_COLS))
rows = []
for item in items:
row = {k: item.get(k) for k in _TIMELINE_COLS if k != "storm_status"}
status = item.get("storm_status", [])
row["storm_status"] = status[0] if status else None
rows.append(row)
df = pd.DataFrame(rows)
# Numeric coercion is strict (errors propagate); a GDACS field
# appearing with an unparseable value should surface, not be
# silently NaN'd. Empty strings are treated as null below.
for col in _TIMELINE_NUMERIC_COLS:
df[col] = df[col].replace("", None)
df[col] = pd.to_numeric(df[col])
# GDACS timelines mix full and abbreviated month names within the same
# event ("01 January 2025" vs "01 Jun 2025"). Bare pd.to_datetime infers
# one format from the first row and chokes on the other; format="mixed"
# parses each value independently.
df["advisory_datetime"] = pd.to_datetime(
df["advisory_datetime"],
format="mixed",
)
df["advisory_number"] = df["advisory_number"].replace("", None)
df["advisory_number"] = pd.to_numeric(df["advisory_number"])
return TIMELINE_SCHEMA.validate(
df.sort_values("advisory_number").reset_index(drop=True)
)
_ADM0_COLUMNS = ["iso3", "country", "pop_affected", "distance_km"]
_ADM1_COLUMNS = [
"iso3",
"country",
"fips_admin",
"gmi_admin",
"admin_name",
"admin_type",
"pop_admin",
"pop_affected",
"distance_km",
]
[docs]
def get_exposure_adm0(
eventid: int,
episodeid: Optional[int] = None,
detail: Optional[Dict[str, Any]] = None,
) -> Dict[str, pd.DataFrame]:
"""Fetch country-level population exposure per wind buffer.
Reads ``datums[alias='country']`` from each buffer's impact JSON —
the canonical ADM0 rollup. One row per affected country, per
buffer.
Note: only returns data for events ~2022 onward. Earlier events
return empty results. Use :func:`get_timeline` for 2015+ coverage.
Parameters
----------
eventid : int
episodeid : int, optional
If provided, fetches that specific episode's snapshot via
:func:`get_episode_detail`. Otherwise uses the event-level
resource URLs from :func:`get_event_detail` (latest snapshot).
detail : dict, optional
Pre-fetched output of :func:`get_event_detail`. Pass it to
skip the internal detail fetch on the event-level path.
Mutually exclusive with ``episodeid`` — passing both raises
``ValueError`` (the two args describe different snapshots).
Returns
-------
dict
Maps buffer key (``"buffer39"`` or ``"buffer74"``) to a
DataFrame with columns ``iso3``, ``country``, ``pop_affected``,
``distance_km``.
"""
return _exposure_per_buffer(eventid, episodeid, _parse_adm0, detail=detail)
[docs]
def get_exposure_adm1(
eventid: int,
episodeid: Optional[int] = None,
detail: Optional[Dict[str, Any]] = None,
) -> Dict[str, pd.DataFrame]:
"""Fetch ADM1-grain population exposure per wind buffer.
Reads ``datums[alias='alert']`` from each buffer's impact JSON.
One row per affected sub-national admin unit, per buffer. Country
identifiers are attached so callers can roll up to ADM0 if needed.
Parameters
----------
eventid : int
episodeid : int, optional
Same semantics as :func:`get_exposure_adm0`.
detail : dict, optional
Same semantics as :func:`get_exposure_adm0`.
Returns
-------
dict
Maps buffer key to a DataFrame with columns ``iso3``,
``country``, ``fips_admin``, ``gmi_admin``, ``admin_name``,
``admin_type``, ``pop_admin``, ``pop_affected``,
``distance_km``.
"""
return _exposure_per_buffer(eventid, episodeid, _parse_adm1, detail=detail)
def _is_actual(actual: pd.Series) -> pd.Series:
"""Boolean mask for observed (vs forecast) advisories.
GDACS stores ``actual`` as a string (``"True"``/``"False"``);
compare case-insensitively rather than trusting the dtype.
"""
return actual.astype(str).str.lower() == "true"
def _split_timeline(
timeline: pd.DataFrame,
) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""Split a GDACS timeline into ``(observed, forecast)`` advisories.
Each timeline row is flagged ``actual``: observed positions
(``actual="True"``) and forecast-cone positions
(``actual="False"``). The two are matched differently — see
:func:`match_to_atcf`.
"""
actual_mask = _is_actual(timeline["actual"])
return timeline[actual_mask], timeline[~actual_mask]
def _nearest_atcf_at(
valid_time: pd.Timestamp,
lat: float,
lon: float,
nhc_tracks: pd.DataFrame,
max_dist_deg: float,
) -> Optional[str]:
"""Closest NHC ``atcf_id`` to ``(lat, lon)`` among rows at exactly
``valid_time``.
The shared primitive behind both matching strategies. Returns the
``atcf_id`` of the nearest row within ``max_dist_deg``, or
``None`` when no NHC row shares the timestamp or the closest one
is too far.
"""
at_time = nhc_tracks[nhc_tracks["valid_time"] == valid_time]
if at_time.empty:
return None
dist = np.sqrt((at_time["lat"] - lat) ** 2 + (at_time["lon"] - lon) ** 2)
best_idx = dist.idxmin()
if dist.loc[best_idx] > max_dist_deg:
return None
return at_time.loc[best_idx, "atcf_id"]
def _match_by_forecast_cone(
forecast: pd.DataFrame,
nhc_tracks: pd.DataFrame,
max_dist_deg: float,
) -> Optional[str]:
"""Strategy 1 (primary) — match on the shared forecast cone.
A GDACS forecast point and the NHC forecast for the same
``valid_time`` are the *same* NHC forecast package (GDACS scrapes
the TCM advisory, which carries the synoptic A-deck cone forward
unchanged), so for the correct storm they agree to ~0°. Each
forecast point that lands on an NHC row casts a vote for its
``atcf_id``; the most-voted ``atcf_id`` wins.
Voting across the whole cone is what makes this robust: a wrong
storm would have to share an exact forecast position at the same
valid_time, which doesn't happen. It also self-heals late
captures — it never needs the genesis cycle, only the cone of
whatever advisory we did capture.
Ties (a basin-crossing storm whose cone touches two ``atcf_id``
values) go to the earliest ``valid_time``, i.e. the genesis basin.
"""
votes = [
atcf
for _, row in forecast.sort_values("advisory_datetime").iterrows()
if (
atcf := _nearest_atcf_at(
row["advisory_datetime"],
row["latitude"],
row["longitude"],
nhc_tracks,
max_dist_deg,
)
)
is not None
]
if not votes:
return None
counts = Counter(votes)
most_votes = max(counts.values())
leaders = {atcf for atcf, n in counts.items() if n == most_votes}
# votes is in valid_time order, so the first leader encountered is
# the earliest-matching (genesis-basin) atcf_id.
return next(atcf for atcf in votes if atcf in leaders)
def _match_by_genesis(
observed: pd.DataFrame,
nhc_tracks: pd.DataFrame,
max_dist_deg: float,
) -> Optional[str]:
"""Strategy 2 (fallback) — match the genesis observed position.
Used only when the timeline carries no forecast cone (e.g. a
dissipated storm pulled from the archive). Takes the genesis
advisory (smallest ``advisory_number``) and requires an NHC row
at that exact ``valid_time``. It's a single point, so it depends
on that one genesis cycle being present in ``nhc_tracks`` — which
is exactly why it's the fallback, not the primary.
"""
if observed.empty:
return None
genesis = observed.sort_values("advisory_number").iloc[0]
return _nearest_atcf_at(
genesis["advisory_datetime"],
genesis["latitude"],
genesis["longitude"],
nhc_tracks,
max_dist_deg,
)
[docs]
def match_to_atcf(
timeline: pd.DataFrame,
nhc_tracks: pd.DataFrame,
max_dist_deg: float = 0.05,
) -> Optional[str]:
"""Match a GDACS event to an NHC ``atcf_id``.
GDACS and our NHC table are two views of the *same* NHC
forecaster output (GDACS scrapes the TCM Forecast/Advisory; we
store the A-deck OFCL). They're timestamped 3h apart — A-deck on
synoptic valid times (00/06/12/18Z), TCM on advisory issue times
(03/09/15/21Z) — but the **forecast cone (t>0) is identical
between them at shared valid_times**. Only the observed t=0
position differs (the storm moved in the intervening 3h).
That asymmetry drives a two-strategy match:
1. :func:`_match_by_forecast_cone` (primary) — vote the GDACS
forecast points onto NHC ``atcf_id`` values by exact valid_time.
Robust (many points), product-agnostic, and self-healing for
storms whose first advisories we never captured.
2. :func:`_match_by_genesis` (fallback) — single-point match on
the genesis observed advisory, for completed storms whose
timeline has no forecast cone left.
Parameters
----------
timeline : DataFrame
Output of :func:`get_timeline`. Required columns:
``advisory_number``, ``actual``, ``advisory_datetime``,
``latitude``, ``longitude``.
nhc_tracks : DataFrame
NHC tracks deduped to one row per ``(atcf_id, valid_time)``
at the freshest issuance (all leadtimes kept — the forecast
cone lives at leadtime>0). Required columns: ``atcf_id``,
``valid_time``, ``lat``, ``lon``. See
``load_freshest_nhc_tracks`` in the pipeline.
max_dist_deg : float, default 0.05
Spatial tolerance in degrees. Cone matches for the correct
storm are ~0°; 0.05° absorbs only rounding. Robustness comes
from voting across the cone, not from a loose tolerance.
Returns
-------
atcf_id (e.g. ``"AL142024"``) or ``None`` when neither strategy
finds a match — the correct answer for a non-NHC (JTWC/RSMC)
storm or a genuine gap in our NHC table.
"""
if timeline.empty or nhc_tracks.empty:
return None
observed, forecast = _split_timeline(timeline)
return _match_by_forecast_cone(
forecast, nhc_tracks, max_dist_deg
) or _match_by_genesis(observed, nhc_tracks, max_dist_deg)
def _exposure_per_buffer(
eventid: int,
episodeid: Optional[int],
parse,
detail: Optional[Dict[str, Any]] = None,
) -> Dict[str, pd.DataFrame]:
if episodeid is not None and detail is not None:
# The two args describe different snapshots (event-level vs.
# a specific episode). Accepting both silently would mean
# the caller's `detail` is wasted; force them to pick one.
raise ValueError("pass either `episodeid` or `detail`, not both")
if episodeid is not None:
detail = get_episode_detail(eventid, episodeid)
elif detail is None:
detail = get_event_detail(eventid)
impacts = detail["properties"]["impacts"]
result: Dict[str, pd.DataFrame] = {}
for imp in impacts:
# `resource` may legitimately omit individual buffer keys for
# events with no exposure at that wind threshold — handle
# missing keys explicitly rather than catching it everywhere.
resource = imp["resource"]
for key in _BUFFER_KEYS:
if key not in resource:
continue
result[key] = parse(_get_json(resource[key]))
return result
def _parse_adm0(data: Dict[str, Any]) -> pd.DataFrame:
"""Read ``alias='country'`` (ADM0 rollups) from a buffer JSON.
The ``iso3`` column holds GDACS' native ``GMI_CNTRY`` (which is
usually a valid ISO3 like "USA" but is X-prefixed for some
non-sovereign territories like "XJE"=Jersey). Callers should apply
:func:`to_iso3` to get a standardized ISO3 code.
``GMI_CNTRY`` is required — KeyError if GDACS omits it (contract
break). Population/distance fields are preserved as None when
missing (GDACS legitimately omits POP_AFFECTED when no population
is exposed at that wind threshold).
"""
rows = []
for dg in data.get("datums", []):
if dg.get("alias") != "country":
continue
for d in dg.get("datum", []):
scalars = _scalars(d)
rows.append(
{
"iso3": scalars["GMI_CNTRY"],
"country": scalars.get("CNTRY_NAME"),
"pop_affected": _to_nullable_int(
scalars.get("POP_AFFECTED")
),
"distance_km": _to_nullable_round(scalars.get("distance")),
}
)
return ADM0_EXPOSURE_SCHEMA.validate(
pd.DataFrame(rows, columns=_ADM0_COLUMNS)
)
def _parse_adm1(data: Dict[str, Any]) -> pd.DataFrame:
"""Read ``alias='alert'`` (ADM1 grain) from a buffer JSON.
Identifier fields (``GMI_CNTRY``, ``GMI_ADMIN``) are required —
KeyError if GDACS omits them. Pop/distance preserved as None
when missing.
"""
rows = []
for dg in data.get("datums", []):
if dg.get("alias") != "alert":
continue
for d in dg.get("datum", []):
scalars = _scalars(d)
rows.append(
{
"iso3": scalars["GMI_CNTRY"],
"country": scalars.get("CNTRY_NAME"),
"fips_admin": scalars.get("FIPS_ADMIN"),
"gmi_admin": scalars["GMI_ADMIN"],
"admin_name": scalars.get("ADMIN_NAME"),
"admin_type": scalars.get("TYPE_ENG"),
"pop_admin": _to_nullable_int(scalars.get("POP_ADMIN")),
"pop_affected": _to_nullable_int(
scalars.get("POP_AFFECTED")
),
"distance_km": _to_nullable_round(scalars.get("distance")),
}
)
return ADM1_EXPOSURE_SCHEMA.validate(
pd.DataFrame(rows, columns=_ADM1_COLUMNS)
)
# GDACS uses a couple of distinct "no data" sentinels in numeric scalar
# fields (POP_AFFECTED, POP_ADMIN, distance). Both normalize to None so
# downstream code sees one consistent missing representation.
# -99999 : structurally missing (e.g., field absent or zeroed-out)
# -1 : "data unavailable / not computed" (observed for POP_AFFECTED
# on some admin1 rows where POP_ADMIN itself is a real number,
# e.g. MARIO-25 over Baja California Sur)
_GDACS_MISSING_SENTINELS = {-99999, -1}
def _to_nullable(value: Any, coerce) -> Any:
"""Cast a GDACS scalar via ``coerce(float(value))``, normalizing
None, empty string, and the GDACS missing-data sentinels to None.
``coerce`` is the type-specific cast (e.g. ``int`` or
``lambda v: round(v, 1)``)."""
if value is None or value == "":
return None
out = coerce(float(value))
if out in _GDACS_MISSING_SENTINELS:
return None
return out
def _to_nullable_int(value: Any) -> Optional[int]:
return _to_nullable(value, int)
def _to_nullable_round(value: Any) -> Optional[float]:
return _to_nullable(value, lambda v: round(v, 1))
def _scalars(datum: Dict[str, Any]) -> Dict[str, Any]:
"""Flatten a GDACS ``scalars`` container into a name→value dict."""
return {
s["name"]: s["value"]
for s in datum.get("scalars", {}).get("scalar", [])
}
def _event_feature_to_row(feature: Dict[str, Any]) -> Dict[str, Any]:
"""Flatten a GDACS event feature JSON to a row dict.
Required fields (eventid, eventname, geometry coordinates,
fromdate/todate, alertlevel, episodealertlevel, iscurrent) use
direct subscripts — KeyError if GDACS drops them. Optional/
metadata fields (iso3, country, severitydata, source) use
``.get`` returning None — these are legitimately absent for
some events and downstream code is expected to handle null.
"""
p = feature["properties"]
coords = feature["geometry"]["coordinates"]
severity = p.get("severitydata") or {}
return {
"eventid": int(p["eventid"]),
"name": p["eventname"],
"alert_level": p["alertlevel"],
"episode_alert": p["episodealertlevel"],
"iso3": p.get("iso3"),
"country": p.get("country"),
"from_date": p["fromdate"],
"to_date": p["todate"],
"is_current": str(p["iscurrent"]).lower() == "true",
"severity_kmh": severity.get("severity"),
"severity_text": severity.get("severitytext"),
"source": p.get("source"),
"longitude": coords[0],
"latitude": coords[1],
}