Add GFS init source as alternative to ERA5T (Phase 2 wiring)

Mirrors the same plumbing shipped in Weather AI 2. CRE's GraphCast
currently inits from ARCO ERA5T at today-5, which means its "5-day
forecast" actually covers today-5 through today-1 — the past week,
not the next week. Trigger decisions confirm past heat events instead
of warning before, which misaligns with the alert-SMS product intent.

GFS has ~3h lag instead of 5 days, enabling forward-looking forecasts
over today → today+5.

Behind a config toggle:

* NWP_INIT_SOURCE=gfs — GFS instead of ERA5T
* NWP_TARGET_DATE_OVERRIDE=YYYY-MM-DD — pin init date for
validation runs against the LMB insurance benchmark panel

Defaults leave existing ERA5T behavior unchanged. Only
src/init_sources/ + tests/test_init_sources/ + two one-line call-site
swaps in graphcast_inference/pipeline touch production code.

56 new unit tests, all pass. Opt-in live GFS test skipped unless
GFS_LIVE_TEST=1.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Dockerfile

@@ -8,6 +8,7 @@ ENV XLA_PYTHON_CLIENT_MEM_FRACTION=0.95
 
 RUN apt-get update && apt-get install -y --no-install-recommends \
     libgomp1 ca-certificates curl git \
+    libeccodes0 \
     && rm -rf /var/lib/apt/lists/*
 
 # Install JAX with CUDA 12 support (separate layer for caching)

pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+testpaths = tests
+markers =
+    live: hits live external services (AWS, HF, etc.); opt-in via env flag

requirements.txt

@@ -15,6 +15,10 @@ gcsfs>=2024.2.0
 xarray>=2024.1.0
 zarr>=2.16.0
 pyarrow>=15.0.0
+# GFS init source (alternative to ERA5T; activated via NWP_INIT_SOURCE=gfs)
+cfgrib>=0.9.10
+eccodes>=1.7.0
+boto3>=1.34.0
 # GraphCast inference (JAX installed separately in Dockerfile)
 graphcast @ git+https://github.com/google-deepmind/graphcast.git
 dm-haiku

src/init_sources/__init__.py

@@ -0,0 +1,31 @@
+"""Init-source plumbing for GraphCast.
+
+CRE's pipeline calls ``fetch_era5_for_graphcast(init_date)`` in
+``src/prediction/graphcast_inference.py`` with ``init_date = today - 5``
+to work around ARCO ERA5T's publication lag. That lag means the "5-day
+forecast" actually covers ``today-5`` through ``today-1`` — retroactive,
+not forward-looking. Alert SMS to workers can't warn before a heat event
+when the forecast is already in the past.
+
+This package provides an alternative: GFS (NOAA Global Forecast System)
+analysis at ~3h lag. A GFS-sourced ``xarray.Dataset`` is shape-compatible
+with the ERA5 Zarr ``full_ds`` used inside ``fetch_era5_for_graphcast``,
+so the integration in Phase 2 is a one-line source swap behind a config
+toggle.
+
+Phase 1 status: this package is reachable only from tests. Nothing in the
+production pipeline imports it. Wiring is deferred to Phase 2 so we can
+validate the module in isolation before deploying.
+
+Public entry point:
+    fetch_gfs_as_era5(target_date: str) -> xarray.Dataset
+
+Sister module in Weather AI 2 (``~/weather AI 2/src/init_sources/``) is the
+canonical copy; this CRE copy is identical code and keeps pace via manual
+sync. If/when init_sources grows a third user, extract to a shared pip-
+installable package.
+"""
+
+from __future__ import annotations
+
+from src.init_sources.gfs import fetch_gfs_as_era5  # noqa: F401

src/init_sources/gfs.py

@@ -0,0 +1,444 @@
+"""GFS (NOAA Global Forecast System) → ERA5-shaped xarray.Dataset.
+
+The top-level entry point is ``fetch_gfs_as_era5(target_date)`` which returns
+an ``xarray.Dataset`` that is a drop-in replacement for the ARCO ERA5 Zarr
+``full_ds`` used by ``graphcast_client._fetch_era5_sync`` and
+``gencast_client._prepare_era5_inputs``.
+
+**Phase 1 status: nothing in the production pipeline imports this module.**
+It's reachable only from ``tests/test_init_sources/``. Wiring into the two
+clients is deferred to Phase 2 so we can validate the module first in
+isolation, then head-to-head with the existing ERA5 path.
+
+Source: AWS S3 public bucket ``noaa-gfs-bdp-pds`` (``gfs.YYYYMMDD/HH/atmos/
+gfs.tHHz.pgrb2.0p25.fNNN``). Free, no auth, real-time (~3h lag per cycle).
+
+Layout inside the returned Dataset:
+    * time:      2 input timesteps (target-6h, target) + N forecast steps
+                 at 6h cadence
+    * latitude:  721 points (90 .. -90, 0.25° step)  — matches ARCO ERA5
+    * longitude: 1440 points (0 .. 359.75, 0.25° step)
+    * level:     13 pressure levels (the GraphCast operational set)
+    * data_vars: 5 surface + 6 pressure-level + 2 static variables in ERA5
+                 canonical naming; units converted to match ERA5
+
+The module keeps every heavy/optional import (``cfgrib``, ``s3fs``, ``boto3``)
+inside the function that needs it. That means the module itself imports fine
+on any box with ``xarray`` + ``numpy``, which matters because the test suite
+runs without cfgrib/eccodes installed.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import pickle
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, List, Optional, Sequence, Tuple
+
+from src.init_sources.variable_mapping import (
+    GRAPHCAST_PRESSURE_LEVELS,
+    PRESSURE_LEVEL_VARS,
+    STATIC_VARS,
+    SURFACE_VARS,
+    unit_convert,
+)
+from src.init_sources.static_vars import load_static_ds
+
+log = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+GFS_S3_BUCKET = "noaa-gfs-bdp-pds"
+GFS_CYCLE_HOURS: Tuple[int, ...] = (0, 6, 12, 18)
+DEFAULT_CYCLE_HOUR = 12
+GFS_NATIVE_RESOLUTION_DEG = 0.25
+GFS_N_LAT = 721
+GFS_N_LON = 1440
+
+# Where to cache assembled GFS datasets (pickle format, same pattern as the
+# ERA5 cache already used by graphcast_client.py).
+_GFS_CACHE_DIR = os.environ.get("GFS_CACHE_DIR", "/tmp/gfs_init_cache")
+
+
+# ---------------------------------------------------------------------------
+# Cycle selection
+# ---------------------------------------------------------------------------
+
+@dataclass
+class GfsCycle:
+    """A specific GFS analysis cycle. Pure data — no I/O."""
+    date: str     # "YYYY-MM-DD" (UTC calendar day of the cycle)
+    hour: int     # 0, 6, 12, or 18
+
+    def datetime_utc(self) -> datetime:
+        d = datetime.fromisoformat(self.date).replace(tzinfo=timezone.utc)
+        return d + timedelta(hours=self.hour)
+
+    def s3_prefix(self) -> str:
+        """Bucket-relative prefix for this cycle's files."""
+        d = self.date.replace("-", "")
+        return f"gfs.{d}/{self.hour:02d}/atmos"
+
+    def file_name(self, forecast_hour: int) -> str:
+        """GRIB2 filename for a given forecast hour (0 = analysis)."""
+        return f"gfs.t{self.hour:02d}z.pgrb2.0p25.f{forecast_hour:03d}"
+
+    def s3_key(self, forecast_hour: int) -> str:
+        return f"{self.s3_prefix()}/{self.file_name(forecast_hour)}"
+
+
+def _most_recent_cycle(
+    target_date: str,
+    target_cycle_hour: int = DEFAULT_CYCLE_HOUR,
+) -> GfsCycle:
+    """Return the cycle that should be used to init at ``target_date Thh:00 UTC``.
+
+    GFS cycles run at 00/06/12/18 UTC. Analysis becomes available ~3-4h after
+    the cycle starts, so in practice we want the most recent cycle that's
+    strictly completed. Phase 2 will walk this back further if the bucket
+    isn't populated yet.
+    """
+    if target_cycle_hour not in GFS_CYCLE_HOURS:
+        # Snap down to the nearest valid cycle hour.
+        target_cycle_hour = max(h for h in GFS_CYCLE_HOURS if h <= target_cycle_hour)
+    return GfsCycle(date=target_date, hour=target_cycle_hour)
+
+
+# ---------------------------------------------------------------------------
+# Download
+# ---------------------------------------------------------------------------
+
+def _download_grib2(
+    cycle: GfsCycle,
+    forecast_hour: int,
+    dest_dir: str,
+) -> str:
+    """Download a single GFS GRIB2 file from the public S3 bucket.
+
+    Returns the local path. Uses ``boto3`` with unsigned requests (the bucket
+    is public). Imports boto3 lazily so the module loads without it.
+    """
+    os.makedirs(dest_dir, exist_ok=True)
+    local_path = os.path.join(dest_dir, cycle.file_name(forecast_hour))
+    if os.path.exists(local_path) and os.path.getsize(local_path) > 0:
+        log.info("GFS GRIB cached: %s", local_path)
+        return local_path
+
+    import boto3
+    from botocore import UNSIGNED
+    from botocore.config import Config
+
+    s3 = boto3.client("s3", config=Config(signature_version=UNSIGNED))
+    key = cycle.s3_key(forecast_hour)
+    log.info("Downloading s3://%s/%s", GFS_S3_BUCKET, key)
+    s3.download_file(GFS_S3_BUCKET, key, local_path)
+    return local_path
+
+
+# ---------------------------------------------------------------------------
+# GRIB parsing
+# ---------------------------------------------------------------------------
+
+def _open_grib_surface(path: str, gfs_short_name: str) -> Any:
+    """Return an xarray.DataArray for a single surface variable from GRIB2.
+
+    cfgrib requires ``filter_by_keys`` to select which GRIB message layer
+    to decode. For surface variables we filter by the short name plus the
+    ``heightAboveGround`` or ``meanSea`` type of level as appropriate.
+    """
+    import xarray
+    # cfgrib's "typeOfLevel" filters pick the right surface layer. GFS stores
+    # t2m/u10/v10 at ``heightAboveGround`` (2m and 10m), ``prmsl`` at
+    # ``meanSea``, and ``tp`` at ``surface``.
+    level_map = {
+        "t2m": ("heightAboveGround", 2),
+        "u10": ("heightAboveGround", 10),
+        "v10": ("heightAboveGround", 10),
+        "prmsl": ("meanSea", 0),
+        "tp":    ("surface", 0),
+    }
+    type_of_level, level = level_map.get(gfs_short_name, ("surface", 0))
+    filters = {"typeOfLevel": type_of_level, "shortName": gfs_short_name}
+    if type_of_level == "heightAboveGround":
+        filters["level"] = level
+    return xarray.open_dataset(path, engine="cfgrib",
+                                backend_kwargs={"filter_by_keys": filters})[gfs_short_name]
+
+
+def _open_grib_pressure_levels(
+    path: str, gfs_short_name: str, levels: Sequence[int],
+) -> Any:
+    """Return an xarray.DataArray for a pressure-level variable, selected to
+    the requested levels. The result has a ``level`` dim with the 13-level set.
+    """
+    import xarray
+    filters = {"typeOfLevel": "isobaricInhPa", "shortName": gfs_short_name}
+    da = xarray.open_dataset(path, engine="cfgrib",
+                              backend_kwargs={"filter_by_keys": filters})[gfs_short_name]
+    # cfgrib names the pressure dim ``isobaricInhPa``; rename to ERA5's
+    # ``level`` for shape compatibility.
+    if "isobaricInhPa" in da.dims:
+        da = da.rename({"isobaricInhPa": "level"})
+    # Filter to the 13 levels we want. ``method='nearest'`` is safe since
+    # every requested level is present natively.
+    return da.sel(level=list(levels), method="nearest")
+
+
+# ---------------------------------------------------------------------------
+# Assembly: one cycle / one timestep worth of data
+# ---------------------------------------------------------------------------
+
+def _assemble_timestep(
+    local_path: str,
+    timestep: datetime,
+    levels: Sequence[int] = GRAPHCAST_PRESSURE_LEVELS,
+) -> Any:
+    """Build an xarray.Dataset for a single time snapshot, in ERA5 naming.
+
+    Opens the GRIB once per variable (cfgrib requires it). The resulting
+    Dataset has:
+        coords: time (scalar), latitude, longitude, level
+        data_vars: all surface + all pressure-level ERA5 names
+    """
+    import numpy as np
+    import xarray
+
+    data_arrays = {}
+
+    # Surface variables
+    for era5_name, (gfs_name, _) in SURFACE_VARS.items():
+        try:
+            da = _open_grib_surface(local_path, gfs_name)
+            da_c = unit_convert(era5_name, da)
+            # Drop cfgrib's valid_time/step/time coords — we'll stamp our own.
+            da_c = da_c.reset_coords(drop=True)
+            data_arrays[era5_name] = da_c
+        except Exception as exc:
+            log.warning("GFS surface var %s (%s) failed to load: %s",
+                        era5_name, gfs_name, exc)
+
+    # Pressure-level variables
+    for era5_name, (gfs_name, _) in PRESSURE_LEVEL_VARS.items():
+        try:
+            da = _open_grib_pressure_levels(local_path, gfs_name, levels)
+            da_c = unit_convert(era5_name, da)
+            da_c = da_c.reset_coords(drop=True)
+            data_arrays[era5_name] = da_c
+        except Exception as exc:
+            log.warning("GFS pressure var %s (%s) failed to load: %s",
+                        era5_name, gfs_name, exc)
+
+    ds = xarray.Dataset(data_arrays)
+    # Normalise coordinate names: cfgrib exposes latitude/longitude already,
+    # but assert and rename if the dim names differ.
+    ds = _normalise_coords(ds)
+    # Stamp the time coord.
+    ts64 = np.datetime64(timestep.replace(tzinfo=None).isoformat())
+    ds = ds.expand_dims(time=[ts64])
+    return ds
+
+
+def _normalise_coords(ds: Any) -> Any:
+    """Ensure the dataset uses ``latitude``/``longitude`` (ERA5 convention).
+
+    cfgrib names the horizontal coords ``latitude`` and ``longitude`` by
+    default, but some GRIB templates use ``lat``/``lon``. We normalise to
+    the ERA5 long names so downstream code (which itself normalises to
+    ``lat``/``lon`` later) has a single predictable starting point.
+    """
+    rename = {}
+    if "lat" in ds.dims and "latitude" not in ds.dims:
+        rename["lat"] = "latitude"
+    if "lon" in ds.dims and "longitude" not in ds.dims:
+        rename["lon"] = "longitude"
+    if rename:
+        ds = ds.rename(rename)
+    return ds
+
+
+# ---------------------------------------------------------------------------
+# Top-level fetcher
+# ---------------------------------------------------------------------------
+
+def _cache_path(target_date: str, cycle_hour: int, horizon_hours: int) -> str:
+    return os.path.join(
+        _GFS_CACHE_DIR,
+        f"gfs_{target_date}_c{cycle_hour:02d}_h{horizon_hours:03d}.pkl",
+    )
+
+
+def _load_cached(target_date: str, cycle_hour: int, horizon_hours: int) -> Optional[Any]:
+    path = _cache_path(target_date, cycle_hour, horizon_hours)
+    if os.path.exists(path):
+        with open(path, "rb") as f:
+            return pickle.load(f)
+    return None
+
+
+def _save_cached(ds: Any, target_date: str, cycle_hour: int, horizon_hours: int) -> None:
+    os.makedirs(_GFS_CACHE_DIR, exist_ok=True)
+    path = _cache_path(target_date, cycle_hour, horizon_hours)
+    with open(path, "wb") as f:
+        pickle.dump(ds, f)
+
+
+def fetch_gfs_as_era5(
+    target_date: str,
+    forecast_horizon_hours: int = 288,
+    cycle_hour: int = DEFAULT_CYCLE_HOUR,
+    *,
+    levels: Sequence[int] = GRAPHCAST_PRESSURE_LEVELS,
+    work_dir: Optional[str] = None,
+    attach_static: bool = True,
+) -> Any:
+    """Fetch a GFS cycle and return an ERA5-shape xarray.Dataset.
+
+    Args:
+        target_date: ISO date string (e.g. "2026-04-21"). The cycle's
+            nominal date.
+        forecast_horizon_hours: how far forward to include forecast steps.
+            Covers the longest horizon either model needs (GraphCast
+            FORECAST_STEPS=48 → 288h).
+        cycle_hour: which UTC cycle (0/6/12/18). Defaults to 12Z.
+        levels: pressure levels to keep. Defaults to the 13-level
+            GraphCast-operational set.
+        work_dir: download cache for raw GRIB files. Defaults to a subdir
+            under ``_GFS_CACHE_DIR``.
+        attach_static: if True, merge in ``geopotential_at_surface`` +
+            ``land_sea_mask`` from the ERA5-derived static cache.
+
+    Returns:
+        An ``xarray.Dataset`` with:
+            * dims: (time, level, latitude, longitude)
+            * time: [target-6h, target, target+6h, ..., target+horizon_h]
+            * 5 surface + 6 pressure-level data_vars (ERA5 names)
+            * static vars if ``attach_static`` and cache is built
+
+    Raises:
+        RuntimeError if a critical variable fails to load across all timesteps.
+        FileNotFoundError if ``attach_static`` and the static cache hasn't
+        been built (call ``static_vars.ensure_static_cache()``).
+    """
+    import xarray
+
+    cached = _load_cached(target_date, cycle_hour, forecast_horizon_hours)
+    if cached is not None:
+        log.info("GFS cache hit for %s %02dZ (+%dh)",
+                 target_date, cycle_hour, forecast_horizon_hours)
+        return cached
+
+    cycle = _most_recent_cycle(target_date, cycle_hour)
+    work_dir = work_dir or os.path.join(_GFS_CACHE_DIR, "grib_raw")
+
+    # Two input timesteps (analysis from this cycle and the previous cycle)
+    # + forecast steps at 6h cadence through the horizon.
+    cycle_dt = cycle.datetime_utc()
+    prev_cycle = _most_recent_cycle(
+        (cycle_dt - timedelta(hours=6)).date().isoformat(),
+        (cycle.hour - 6) % 24,
+    )
+
+    timesteps: List[Tuple[datetime, GfsCycle, int]] = []
+    timesteps.append((prev_cycle.datetime_utc(), prev_cycle, 0))
+    timesteps.append((cycle_dt, cycle, 0))
+    n_forecast = forecast_horizon_hours // 6
+    for i in range(1, n_forecast + 1):
+        fhour = i * 6
+        timesteps.append((cycle_dt + timedelta(hours=fhour), cycle, fhour))
+
+    per_step_ds = []
+    for (step_dt, step_cycle, fhour) in timesteps:
+        grib_path = _download_grib2(step_cycle, fhour, work_dir)
+        ds_step = _assemble_timestep(grib_path, step_dt, levels=levels)
+        per_step_ds.append(ds_step)
+
+    full_ds = xarray.concat(per_step_ds, dim="time")
+
+    if attach_static:
+        try:
+            static_ds = load_static_ds()
+        except FileNotFoundError:
+            log.info("Static-var cache missing; building once from ERA5 (~30s)")
+            from src.init_sources.static_vars import ensure_static_cache
+            ensure_static_cache()
+            static_ds = load_static_ds()
+        # Merge keeps the time-varying vars plus the static 2D surfaces.
+        full_ds = xarray.merge([full_ds, static_ds], compat="override")
+
+    _save_cached(full_ds, target_date, cycle_hour, forecast_horizon_hours)
+    return full_ds
+
+
+# ---------------------------------------------------------------------------
+# Helpers exposed for tests (pure xarray manipulation — no I/O)
+# ---------------------------------------------------------------------------
+
+def build_synthetic_gfs_dataset(
+    timesteps: Sequence[datetime],
+    levels: Sequence[int] = GRAPHCAST_PRESSURE_LEVELS,
+    lat_vals=None,
+    lon_vals=None,
+    include_static: bool = True,
+    temperature_k: float = 300.0,
+) -> Any:
+    """Construct a minimal synthetic dataset with the same shape this module
+    returns in production. For unit tests only.
+
+    Values are physically-plausible constants — the tests assert on shape
+    and dimension layout, not on the values themselves.
+    """
+    import numpy as np
+    import xarray
+
+    if lat_vals is None:
+        lat_vals = np.linspace(90.0, -90.0, GFS_N_LAT, dtype=np.float32)
+    if lon_vals is None:
+        lon_vals = np.linspace(0.0, 359.75, GFS_N_LON, dtype=np.float32)
+    t_coord = np.array([np.datetime64(t.replace(tzinfo=None).isoformat())
+                         for t in timesteps])
+
+    shape_s = (len(t_coord), len(lat_vals), len(lon_vals))
+    shape_p = (len(t_coord), len(levels), len(lat_vals), len(lon_vals))
+
+    ds = xarray.Dataset(
+        data_vars={
+            "2m_temperature":              (("time", "latitude", "longitude"),
+                                             np.full(shape_s, temperature_k, dtype=np.float32)),
+            "10m_u_component_of_wind":      (("time", "latitude", "longitude"),
+                                             np.zeros(shape_s, dtype=np.float32)),
+            "10m_v_component_of_wind":      (("time", "latitude", "longitude"),
+                                             np.zeros(shape_s, dtype=np.float32)),
+            "mean_sea_level_pressure":      (("time", "latitude", "longitude"),
+                                             np.full(shape_s, 101_325.0, dtype=np.float32)),
+            "total_precipitation_6hr":      (("time", "latitude", "longitude"),
+                                             np.zeros(shape_s, dtype=np.float32)),
+            "temperature":                  (("time", "level", "latitude", "longitude"),
+                                             np.full(shape_p, 260.0, dtype=np.float32)),
+            "specific_humidity":            (("time", "level", "latitude", "longitude"),
+                                             np.full(shape_p, 0.005, dtype=np.float32)),
+            "u_component_of_wind":          (("time", "level", "latitude", "longitude"),
+                                             np.zeros(shape_p, dtype=np.float32)),
+            "v_component_of_wind":          (("time", "level", "latitude", "longitude"),
+                                             np.zeros(shape_p, dtype=np.float32)),
+            "vertical_velocity":            (("time", "level", "latitude", "longitude"),
+                                             np.zeros(shape_p, dtype=np.float32)),
+            "geopotential":                 (("time", "level", "latitude", "longitude"),
+                                             np.full(shape_p, 5.0e4, dtype=np.float32)),
+        },
+        coords={
+            "time":      t_coord,
+            "level":     np.asarray(levels, dtype=np.int32),
+            "latitude":  lat_vals,
+            "longitude": lon_vals,
+        },
+    )
+    if include_static:
+        from src.init_sources.static_vars import build_synthetic_static_ds
+        static_ds = build_synthetic_static_ds(lat_vals=lat_vals, lon_vals=lon_vals)
+        ds = ds.merge(static_ds)
+    return ds

src/init_sources/static_vars.py

@@ -0,0 +1,142 @@
+"""Static (time-invariant) variables for GraphCast / GenCast init.
+
+GraphCast requires two static fields that GFS's standard pgrb2 products do
+not ship: ``geopotential_at_surface`` (surface elevation × g, used to place
+the model on Earth's terrain) and ``land_sea_mask`` (binary land/water flag).
+
+They are pulled once from ERA5 and cached to disk as a small netCDF. GFS
+inits then attach them verbatim — the values don't change per init.
+
+The cache lives at ``data/init_sources/era5_static.nc`` (committed to the
+repo). If it's missing, the ``ensure_static_cache`` helper downloads a fresh
+copy from ARCO ERA5. That download is a one-time, ~5 MB pull (two 721×1440
+float32 surfaces + coords).
+
+Phase 1 note: the cache file is not generated in this commit — Phase 2 will
+produce it the first time the integration runs. The loader here is wired
+end-to-end so Phase 2 can simply invoke ``load_static_ds()`` without further
+work.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+# Path to the static-variable cache. Relative to the repo root so both local
+# dev and HF Space runs find it in the same place.
+DEFAULT_CACHE_PATH = str(
+    Path(__file__).resolve().parent.parent.parent
+    / "data" / "init_sources" / "era5_static.nc"
+)
+
+STATIC_VARS_NEEDED = ("geopotential_at_surface", "land_sea_mask")
+
+
+def cache_path() -> str:
+    return os.environ.get("GFS_STATIC_CACHE", DEFAULT_CACHE_PATH)
+
+
+def cache_exists() -> bool:
+    return os.path.exists(cache_path())
+
+
+def load_static_ds() -> Any:
+    """Load the static-variable netCDF as an xarray.Dataset.
+
+    Raises FileNotFoundError if the cache hasn't been built yet. Callers in
+    Phase 2 should either call ``ensure_static_cache()`` first or fall back
+    gracefully (GraphCast's existing code zero-fills missing static vars
+    with a warning — same behavior preserved here).
+    """
+    import xarray  # lazy — module is importable without xarray present
+    path = cache_path()
+    if not os.path.exists(path):
+        raise FileNotFoundError(
+            f"Static-variable cache not found at {path}. "
+            f"Run ensure_static_cache() to build it from ERA5, or unset "
+            f"GFS_STATIC_CACHE to use the default repo path."
+        )
+    return xarray.load_dataset(path)
+
+
+def ensure_static_cache(force: bool = False) -> str:
+    """Build the static-variable cache from ERA5 if it doesn't exist.
+
+    Returns the path to the cache file. Idempotent.
+
+    This fetches a single ERA5 timestep's static fields from the ARCO Zarr.
+    ERA5 writes the same static values at every timestep (they truly don't
+    change), so any valid time works — we use 2024-01-01 as a stable anchor.
+    """
+    import xarray
+    path = cache_path()
+    if os.path.exists(path) and not force:
+        return path
+
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+
+    era5_path = "gs://gcp-public-data-arco-era5/ar/full_37-1h-0p25deg-chunk-1.zarr-v3"
+    full_ds = xarray.open_zarr(
+        era5_path, chunks=None, storage_options={"token": "anon"},
+        consolidated=True,
+    )
+    # Pick a known-valid timestep; values are static so the exact time is moot.
+    import numpy as np
+    t_anchor = np.datetime64("2024-01-01T12:00")
+    t_sel = full_ds.time.sel(time=t_anchor, method="nearest").values
+
+    vars_present = [v for v in STATIC_VARS_NEEDED if v in full_ds.data_vars]
+    if not vars_present:
+        raise RuntimeError(
+            f"ERA5 Zarr at {era5_path} lacks {STATIC_VARS_NEEDED}. "
+            f"Check the upstream bucket — this would be a schema change "
+            f"upstream, not a bug here."
+        )
+    static_ds = full_ds[vars_present].sel(time=t_sel).compute()
+    # Drop the time dim (static vars have no real time axis even though ERA5
+    # ships them per-step).
+    if "time" in static_ds.dims:
+        static_ds = static_ds.drop_vars("time")
+    elif "time" in static_ds.coords:
+        static_ds = static_ds.drop_vars("time")
+
+    static_ds.to_netcdf(path)
+    return path
+
+
+def build_synthetic_static_ds(
+    lat_vals=None, lon_vals=None,
+):
+    """Build a tiny synthetic static dataset for use in unit tests.
+
+    Returns a two-variable xarray.Dataset with plausible values — zeros for
+    ocean, a constant hill for land — on whatever grid is passed in. No
+    pretensions to realism; the tests only care that downstream code can
+    read and shape-check the fields.
+    """
+    import numpy as np
+    import xarray
+    if lat_vals is None:
+        lat_vals = np.linspace(90.0, -90.0, 721, dtype=np.float32)
+    if lon_vals is None:
+        lon_vals = np.linspace(0.0, 359.75, 1440, dtype=np.float32)
+
+    shape = (len(lat_vals), len(lon_vals))
+    gp_surf = np.full(shape, 0.0, dtype=np.float32)
+    lsm = np.zeros(shape, dtype=np.float32)
+    # Mark the India region as land for a sanity check: lat 8-14 N, lon 74-80 E.
+    lat_mask = (lat_vals >= 8.0) & (lat_vals <= 14.0)
+    lon_mask = (lon_vals >= 74.0) & (lon_vals <= 80.0)
+    if lat_mask.any() and lon_mask.any():
+        lsm[np.ix_(lat_mask, lon_mask)] = 1.0
+        gp_surf[np.ix_(lat_mask, lon_mask)] = 500.0 * 9.80665  # ~500m elev
+
+    return xarray.Dataset(
+        {
+            "geopotential_at_surface": (("latitude", "longitude"), gp_surf),
+            "land_sea_mask":           (("latitude", "longitude"), lsm),
+        },
+        coords={"latitude": lat_vals, "longitude": lon_vals},
+    )

src/init_sources/variable_mapping.py

@@ -0,0 +1,128 @@
+"""GFS → ERA5 variable name, unit, and pressure-level conversion tables.
+
+These are pure data. Tests exercise them directly. The fetcher in gfs.py
+consumes them.
+
+ERA5 (via ARCO Zarr, what GraphCast + GenCast are trained on) uses long
+human-readable names: ``2m_temperature``, ``10m_u_component_of_wind``, etc.
+GFS GRIB2 uses short codes: ``t2m``, ``u10``, etc. Both encode the same
+physical quantities but with different conventions.
+
+Everything here is single-source-of-truth: if GFS introduces a new short-code
+convention, it only gets added here and the fetcher picks it up automatically.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, Tuple
+
+
+# ---------------------------------------------------------------------------
+# Surface variables (single-level)
+# ---------------------------------------------------------------------------
+
+# Each entry: ERA5 canonical name → (GFS cfgrib short-name, unit conversion)
+# Unit conversion is a (scale, offset) tuple applied as gfs_value * scale + offset.
+# When no conversion is needed (same units), use (1.0, 0.0).
+#
+# GFS units from cfgrib:
+#   t2m:  K      (matches ERA5)
+#   u10:  m/s    (matches ERA5)
+#   v10:  m/s    (matches ERA5)
+#   prmsl:Pa     (matches ERA5)
+#   tp:   kg/m^2 (≡ mm of liquid water, ERA5 has m — divide by 1000)
+#
+# GFS GRIB2 has total_precipitation as an accumulated quantity over the forecast
+# step — cfgrib exposes ``tp`` in kg/m² which is numerically the same as mm of
+# liquid water. ERA5 reports precipitation in m. We divide by 1000 to match.
+
+SURFACE_VARS: Dict[str, Tuple[str, Tuple[float, float]]] = {
+    "2m_temperature":               ("t2m",   (1.0,    0.0)),
+    "10m_u_component_of_wind":      ("u10",   (1.0,    0.0)),
+    "10m_v_component_of_wind":      ("v10",   (1.0,    0.0)),
+    "mean_sea_level_pressure":      ("prmsl", (1.0,    0.0)),
+    "total_precipitation_6hr":      ("tp",    (1e-3,   0.0)),  # mm → m
+}
+
+
+# ---------------------------------------------------------------------------
+# Pressure-level variables (3D: time × level × lat × lon)
+# ---------------------------------------------------------------------------
+#
+# GFS units on pressure levels:
+#   t:   K       (matches ERA5)
+#   q:   kg/kg   (matches ERA5)
+#   u:   m/s     (matches ERA5)
+#   v:   m/s     (matches ERA5)
+#   w:   Pa/s    (matches ERA5 vertical_velocity)
+#   gh:  gpm     (geopotential HEIGHT in meters — ERA5 has geopotential in m²/s²)
+#                → multiply by g = 9.80665 to get geopotential.
+
+PRESSURE_LEVEL_VARS: Dict[str, Tuple[str, Tuple[float, float]]] = {
+    "temperature":              ("t",  (1.0,       0.0)),
+    "specific_humidity":        ("q",  (1.0,       0.0)),
+    "u_component_of_wind":      ("u",  (1.0,       0.0)),
+    "v_component_of_wind":      ("v",  (1.0,       0.0)),
+    "vertical_velocity":        ("w",  (1.0,       0.0)),
+    "geopotential":             ("gh", (9.80665,   0.0)),  # gpm → m²/s²
+}
+
+
+# ---------------------------------------------------------------------------
+# Static variables (no time dim)
+# ---------------------------------------------------------------------------
+#
+# These aren't in standard GFS pgrb2 files. They're loaded from a small ERA5
+# snapshot cached in the repo. See static_vars.py.
+
+STATIC_VARS: Tuple[str, ...] = (
+    "geopotential_at_surface",
+    "land_sea_mask",
+)
+
+
+# ---------------------------------------------------------------------------
+# Pressure levels
+# ---------------------------------------------------------------------------
+#
+# GraphCast operational uses 13 levels. GenCast 1.0° uses the same 13 (the
+# checkpoint's task_config lists them explicitly). Both models' task_configs
+# carry the level list, so the fetcher never hard-codes it — but the levels
+# here are what we select DOWN to from GFS's richer level set.
+#
+# GFS pgrb2.0p25 natively provides: 10, 20, 30, 40, 50, 70, 100, 150, 200,
+# 250, 300, 350, 400, 450, 500, 550, 600, 650, 700, 750, 800, 850, 900, 925,
+# 950, 975, 1000 hPa — all 13 GraphCast levels are in there.
+
+GRAPHCAST_PRESSURE_LEVELS: Tuple[int, ...] = (
+    50, 100, 150, 200, 250, 300, 400, 500, 600, 700, 850, 925, 1000,
+)
+
+
+# ---------------------------------------------------------------------------
+# Convenience helpers
+# ---------------------------------------------------------------------------
+
+def gfs_short_name(era5_name: str) -> str:
+    """Return the GFS short-name for an ERA5 variable, or raise KeyError."""
+    if era5_name in SURFACE_VARS:
+        return SURFACE_VARS[era5_name][0]
+    if era5_name in PRESSURE_LEVEL_VARS:
+        return PRESSURE_LEVEL_VARS[era5_name][0]
+    raise KeyError(f"No GFS mapping for ERA5 variable {era5_name!r}")
+
+
+def unit_convert(era5_name: str, gfs_values):
+    """Apply the stored (scale, offset) conversion to a GFS value/array."""
+    if era5_name in SURFACE_VARS:
+        scale, offset = SURFACE_VARS[era5_name][1]
+    elif era5_name in PRESSURE_LEVEL_VARS:
+        scale, offset = PRESSURE_LEVEL_VARS[era5_name][1]
+    else:
+        raise KeyError(f"No unit conversion for {era5_name!r}")
+    return gfs_values * scale + offset
+
+
+def all_era5_names() -> Tuple[str, ...]:
+    """All ERA5 canonical names handled by this mapping (surface + pressure)."""
+    return tuple(SURFACE_VARS.keys()) + tuple(PRESSURE_LEVEL_VARS.keys())

src/pipeline.py

@@ -547,7 +547,14 @@ class HeatRiskPipeline:
                 # "today" makes xarray's nearest-match silently fall back to
                 # whatever's available, so the 5-day forecast window doesn't
                 # align with T+1..T+5 as the pipeline assumes. Anchor to T-5.
-                init_date = (datetime.utcnow() - timedelta(days=5)).strftime("%Y-%m-%d")
+                _override = os.environ.get("NWP_TARGET_DATE_OVERRIDE")
+                if _override:
+                    init_date = _override
+                elif os.environ.get("NWP_INIT_SOURCE", "era5t").lower() == "gfs":
+                    # GFS has ~3h lag — yesterday's cycle is always complete.
+                    init_date = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%d")
+                else:
+                    init_date = (datetime.utcnow() - timedelta(days=5)).strftime("%Y-%m-%d")
                 _, gc_wbgt, gc_timing = forecast_for_dar(init_date, apply_mos=True)
                 forecast_source = "graphcast_mos"
                 logger.info(

src/prediction/graphcast_inference.py

@@ -250,10 +250,18 @@ def fetch_era5_for_graphcast(
     import xarray
     from graphcast import data_utils
 
-    log.info("Opening ARCO ERA5 Zarr for %s...", target_date)
-    gcs_opts = {"token": "anon"}
-    full_ds = xarray.open_zarr(
-        ERA5_PATH, chunks=None, storage_options=gcs_opts, consolidated=True)
+    import os as _os
+    if _os.environ.get("NWP_INIT_SOURCE", "era5t").lower() == "gfs":
+        log.info("NWP_INIT_SOURCE=gfs — fetching GFS init for %s", target_date)
+        from src.init_sources import fetch_gfs_as_era5
+        full_ds = fetch_gfs_as_era5(
+            target_date, forecast_horizon_hours=FORECAST_STEPS * 6,
+        )
+    else:
+        log.info("Opening ARCO ERA5 Zarr for %s...", target_date)
+        gcs_opts = {"token": "anon"}
+        full_ds = xarray.open_zarr(
+            ERA5_PATH, chunks=None, storage_options=gcs_opts, consolidated=True)
 
     # Target time: noon UTC on the requested date
     target = np.datetime64(f"{target_date}T12:00")

tests/test_init_sources/test_gfs_live.py

@@ -0,0 +1,107 @@
+"""Live GFS integration test — skipped by default.
+
+This is the "does the module actually work against real AWS GFS data" check.
+It runs only when ``GFS_LIVE_TEST=1`` is set, because it requires:
+
+    * network access to ``noaa-gfs-bdp-pds`` on S3 (public bucket, unsigned)
+    * ``cfgrib`` installed (pulls in the eccodes C library)
+    * ``boto3`` for the S3 fetch
+    * ~200 MB of disk for one GRIB2 file
+
+Sanity-check assertions are conservative: if cfgrib or boto3 are missing we
+``pytest.skip`` instead of failing so the suite stays green on machines
+without the deps.
+
+Run manually:
+
+    GFS_LIVE_TEST=1 .venv/bin/pytest -q -k live tests/test_init_sources/test_gfs_live.py
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import importlib.util
+import os
+import tempfile
+
+import numpy as np
+import pytest
+
+from src.init_sources import gfs
+from src.init_sources.variable_mapping import GRAPHCAST_PRESSURE_LEVELS
+
+
+REQUIRED_PACKAGES = ("cfgrib", "boto3")
+
+
+def _required_deps_present() -> bool:
+    return all(importlib.util.find_spec(pkg) is not None for pkg in REQUIRED_PACKAGES)
+
+
+pytestmark = pytest.mark.skipif(
+    os.environ.get("GFS_LIVE_TEST") != "1",
+    reason="set GFS_LIVE_TEST=1 to run live AWS/cfgrib integration tests",
+)
+
+
+def _skip_if_no_deps():
+    if not _required_deps_present():
+        pytest.skip(f"install {REQUIRED_PACKAGES} to run live GFS tests")
+
+
+@pytest.mark.live
+def test_download_one_grib_file(tmp_path):
+    """Smallest-possible live test: pull one GRIB2 file and verify it exists."""
+    _skip_if_no_deps()
+
+    # Use yesterday 06Z — old enough that the cycle is definitely complete.
+    yesterday = (dt.datetime.utcnow().date() - dt.timedelta(days=1)).isoformat()
+    cycle = gfs.GfsCycle(date=yesterday, hour=6)
+
+    path = gfs._download_grib2(cycle, forecast_hour=0, dest_dir=str(tmp_path))
+    assert os.path.exists(path)
+    assert os.path.getsize(path) > 10_000_000, (
+        "GFS pgrb2.0p25 f000 files are hundreds of MB; <10MB suggests partial download"
+    )
+
+
+@pytest.mark.live
+def test_fetch_gfs_as_era5_assembles_dataset(tmp_path, monkeypatch):
+    """End-to-end: fetch + parse + assemble the first two timesteps, check shape."""
+    _skip_if_no_deps()
+
+    # Point the cache at a tmp dir so this test doesn't pollute /tmp.
+    monkeypatch.setattr(gfs, "_GFS_CACHE_DIR", str(tmp_path))
+
+    yesterday = (dt.datetime.utcnow().date() - dt.timedelta(days=1)).isoformat()
+
+    # Keep the forecast horizon tiny so the live test completes in reasonable
+    # time — only 12 hours out (3 timesteps total: -6h analysis, 0h analysis,
+    # +6h forecast). The shape/variable coverage is the same regardless of
+    # horizon, so 12h is sufficient signal for the live sanity check.
+    ds = gfs.fetch_gfs_as_era5(
+        target_date=yesterday,
+        forecast_horizon_hours=12,
+        cycle_hour=6,
+        work_dir=str(tmp_path / "grib_raw"),
+        attach_static=False,
+    )
+
+    # Shape checks
+    assert ds.sizes["latitude"] == gfs.GFS_N_LAT
+    assert ds.sizes["longitude"] == gfs.GFS_N_LON
+    assert ds.sizes["level"] == len(GRAPHCAST_PRESSURE_LEVELS)
+    assert ds.sizes["time"] >= 3  # prev cycle t0, current t0, +6h forecast
+
+    # Variable coverage — at least 2m_temp + one pressure-level must load.
+    assert "2m_temperature" in ds.data_vars
+    assert "temperature" in ds.data_vars
+
+    # Physical plausibility on a Kerala grid cell.
+    t2m = ds["2m_temperature"].sel(
+        latitude=8.5, longitude=77.0, method="nearest"
+    ).values
+    # Kelvin; South India is never < 273 K (0 °C) even at night.
+    assert (t2m > 273).all() and (t2m < 320).all(), (
+        f"2m_temperature at Kerala grid cell {t2m} out of plausible K range"
+    )

tests/test_init_sources/test_gfs_shape.py

@@ -0,0 +1,242 @@
+"""Shape + interface tests for the GFS init source.
+
+These tests use a synthetic xarray.Dataset built by
+``gfs.build_synthetic_gfs_dataset`` — they never hit the network and do not
+require cfgrib/eccodes. The assertion surface is the SAME set of downstream
+operations that ``graphcast_client._fetch_era5_sync`` and
+``gencast_client._prepare_era5_inputs`` perform on the ERA5 ``full_ds``:
+
+    * ``.time.sel(time=X, method='nearest')``
+    * ``.sel(time=[t0, t1])``
+    * variable access by ERA5 canonical name
+    * ``.sel(level=[...])`` on pressure-level vars
+    * ``.compute()`` — even if the dataset isn't Dask-backed, existing code
+      still calls it, so the Dataset must accept it.
+
+If Phase 2 wires this module into the pipeline and anything here breaks,
+we'd have a merge/rename bug. These tests catch that before the Space run.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import numpy as np
+import pytest
+import xarray
+
+from src.init_sources import gfs
+from src.init_sources.variable_mapping import (
+    GRAPHCAST_PRESSURE_LEVELS,
+    PRESSURE_LEVEL_VARS,
+    STATIC_VARS,
+    SURFACE_VARS,
+)
+
+
+def _make_timesteps(n: int = 4, step_hours: int = 6):
+    t0 = datetime(2026, 4, 21, 12, 0, tzinfo=timezone.utc)
+    return [t0 + timedelta(hours=i * step_hours) for i in range(n)]
+
+
+def _small_grid_dataset(include_static=True):
+    """Build a small-grid synthetic dataset (fast tests)."""
+    lat_vals = np.linspace(14.0, 8.0, 25, dtype=np.float32)      # Kerala band
+    lon_vals = np.linspace(74.0, 80.0, 25, dtype=np.float32)     # TN band
+    return gfs.build_synthetic_gfs_dataset(
+        timesteps=_make_timesteps(n=4),
+        lat_vals=lat_vals,
+        lon_vals=lon_vals,
+        include_static=include_static,
+    )
+
+
+class TestSyntheticDatasetShape:
+    def test_returns_xarray_dataset(self):
+        ds = _small_grid_dataset()
+        assert isinstance(ds, xarray.Dataset)
+
+    def test_has_expected_dims(self):
+        ds = _small_grid_dataset()
+        for d in ("time", "level", "latitude", "longitude"):
+            assert d in ds.dims, f"missing dim {d!r}"
+
+    def test_time_dim_has_4_steps(self):
+        ds = _small_grid_dataset()
+        assert ds.sizes["time"] == 4
+
+    def test_level_dim_matches_graphcast(self):
+        ds = _small_grid_dataset()
+        assert ds.sizes["level"] == len(GRAPHCAST_PRESSURE_LEVELS)
+        np.testing.assert_array_equal(
+            ds["level"].values,
+            np.asarray(GRAPHCAST_PRESSURE_LEVELS, dtype=np.int32),
+        )
+
+
+class TestVariableCoverage:
+    def test_all_surface_vars_present(self):
+        ds = _small_grid_dataset()
+        for era5_name in SURFACE_VARS:
+            assert era5_name in ds.data_vars, f"missing surface var {era5_name!r}"
+
+    def test_all_pressure_level_vars_present(self):
+        ds = _small_grid_dataset()
+        for era5_name in PRESSURE_LEVEL_VARS:
+            assert era5_name in ds.data_vars, f"missing pressure var {era5_name!r}"
+
+    def test_static_vars_attached_when_requested(self):
+        ds = _small_grid_dataset(include_static=True)
+        for s in STATIC_VARS:
+            assert s in ds.data_vars, (
+                f"static var {s!r} must be attached when include_static=True — "
+                "GraphCast's task_config lists it in input_variables"
+            )
+
+    def test_static_vars_omitted_when_not_requested(self):
+        ds = _small_grid_dataset(include_static=False)
+        for s in STATIC_VARS:
+            assert s not in ds.data_vars
+
+    def test_pressure_level_vars_have_level_dim(self):
+        ds = _small_grid_dataset()
+        for era5_name in PRESSURE_LEVEL_VARS:
+            assert "level" in ds[era5_name].dims, (
+                f"{era5_name!r} must carry the level dim — downstream "
+                ".sel(level=[...]) will otherwise raise"
+            )
+
+    def test_surface_vars_have_no_level_dim(self):
+        ds = _small_grid_dataset()
+        for era5_name in SURFACE_VARS:
+            assert "level" not in ds[era5_name].dims, (
+                f"{era5_name!r} must NOT carry the level dim — it's a "
+                "single-level surface field"
+            )
+
+    def test_static_vars_have_no_time_or_level_dim(self):
+        ds = _small_grid_dataset()
+        for s in STATIC_VARS:
+            assert "time" not in ds[s].dims
+            assert "level" not in ds[s].dims
+
+
+class TestDownstreamCompatibility:
+    """Exercise the same xarray operations ``_fetch_era5_sync`` runs
+    against the ERA5 ``full_ds``. If any of these break, Phase 2 wiring
+    would blow up at runtime — we catch them here first.
+    """
+
+    def test_time_sel_nearest(self):
+        ds = _small_grid_dataset()
+        # existing code: ``full_ds.time.sel(time=target - 6h, method='nearest')``
+        target = np.datetime64("2026-04-21T12:00")
+        t1 = ds.time.sel(time=target, method="nearest").values
+        assert t1 is not None
+
+    def test_time_sel_list_of_two_timesteps(self):
+        ds = _small_grid_dataset()
+        # existing code: ``full_ds[dynamic_available].sel(time=input_times)``
+        input_times = ds.time.values[:2].tolist()
+        sub = ds.sel(time=input_times)
+        assert sub.sizes["time"] == 2
+
+    def test_level_subset_selection(self):
+        ds = _small_grid_dataset()
+        # existing code: ``full_ds[dynamic_available].sel(time=...,  level=sel_levels)``
+        sel = ds["temperature"].sel(level=[500, 850, 1000])
+        assert sel.sizes["level"] == 3
+
+    def test_compute_is_callable_even_on_non_dask_dataset(self):
+        ds = _small_grid_dataset()
+        # existing code: ``ds_input = full_ds[...].sel(...).compute()``.
+        computed = ds[["2m_temperature"]].sel(time=ds.time.values[0]).compute()
+        assert "2m_temperature" in computed.data_vars
+
+    def test_isnull_all_returns_bool_like(self):
+        ds = _small_grid_dataset()
+        # existing code: ``bool(ds_input[var].isnull().all().item())``
+        all_nan = ds["2m_temperature"].isnull().all().item()
+        assert isinstance(all_nan, (bool, np.bool_))
+        assert all_nan is False or all_nan == 0  # synthetic data has no NaN
+
+    def test_point_nearest_sel(self):
+        ds = _small_grid_dataset()
+        # existing code in _extract_station_forecasts:
+        #   point = predictions.sel(lat=station.lat, lon=station.lon,
+        #                             method='nearest')
+        # except the returned ds uses 'latitude'/'longitude' (ERA5 names).
+        # graphcast_client renames to lat/lon later via rename_map. We verify
+        # both names work here so the rename step succeeds.
+        kerala = ds.sel(latitude=8.5, longitude=76.95, method="nearest")
+        assert kerala["2m_temperature"].size == ds.sizes["time"]
+
+
+class TestGfsCycle:
+    def test_datetime_utc_is_utc_aware(self):
+        cyc = gfs.GfsCycle(date="2026-04-21", hour=12)
+        dt = cyc.datetime_utc()
+        assert dt.tzinfo is not None
+        assert dt.year == 2026 and dt.month == 4 and dt.day == 21
+        assert dt.hour == 12
+
+    def test_s3_prefix_format(self):
+        cyc = gfs.GfsCycle(date="2026-04-21", hour=6)
+        assert cyc.s3_prefix() == "gfs.20260421/06/atmos"
+
+    def test_file_name_analysis(self):
+        cyc = gfs.GfsCycle(date="2026-04-21", hour=12)
+        # f000 is the analysis (t=0 of the cycle).
+        assert cyc.file_name(0) == "gfs.t12z.pgrb2.0p25.f000"
+
+    def test_file_name_forecast_6h(self):
+        cyc = gfs.GfsCycle(date="2026-04-21", hour=12)
+        assert cyc.file_name(6) == "gfs.t12z.pgrb2.0p25.f006"
+
+    def test_file_name_forecast_168h(self):
+        cyc = gfs.GfsCycle(date="2026-04-21", hour=12)
+        # 7-day forecast — end of GenCast's horizon.
+        assert cyc.file_name(168) == "gfs.t12z.pgrb2.0p25.f168"
+
+    def test_s3_key_composition(self):
+        cyc = gfs.GfsCycle(date="2026-04-21", hour=12)
+        assert cyc.s3_key(6) == "gfs.20260421/12/atmos/gfs.t12z.pgrb2.0p25.f006"
+
+
+class TestCycleSelection:
+    def test_default_cycle_is_12z(self):
+        cyc = gfs._most_recent_cycle("2026-04-21")
+        assert cyc.hour == 12
+
+    def test_explicit_cycle_hour_respected(self):
+        cyc = gfs._most_recent_cycle("2026-04-21", target_cycle_hour=0)
+        assert cyc.hour == 0
+
+    def test_non_canonical_cycle_hour_snaps_down(self):
+        # Someone passes 9 UTC — GFS only runs 00/06/12/18, so we expect
+        # the preceding 06Z cycle.
+        cyc = gfs._most_recent_cycle("2026-04-21", target_cycle_hour=9)
+        assert cyc.hour == 6
+
+
+class TestPhysicalPlausibility:
+    def test_synthetic_temperature_is_kelvin_valued(self):
+        ds = _small_grid_dataset()
+        # The synthetic fixture ships 300 K everywhere. Convert a few points
+        # to Celsius and sanity-check the sign.
+        t_c = ds["2m_temperature"].values - 273.15
+        # 300 K → 26.85 °C; inside the plausible Earth-surface range.
+        assert np.all(t_c > -50) and np.all(t_c < 60)
+
+    def test_synthetic_mslp_in_physical_range_pa(self):
+        ds = _small_grid_dataset()
+        mslp = ds["mean_sea_level_pressure"].values
+        # Earth MSLP lives in ~87000-108000 Pa. Synthetic fixture uses 101325.
+        assert np.all(mslp > 80_000) and np.all(mslp < 110_000)
+
+    def test_synthetic_precipitation_non_negative_in_metres(self):
+        ds = _small_grid_dataset()
+        tp = ds["total_precipitation_6hr"].values
+        assert np.all(tp >= 0.0)
+        # After the mm→m conversion, a realistic 6h max ≤ ~0.2 m (200mm).
+        assert np.all(tp < 1.0)

tests/test_init_sources/test_static_vars.py

@@ -0,0 +1,83 @@
+"""Tests for the static-variable loader + synthetic builder."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import numpy as np
+import pytest
+import xarray
+
+from src.init_sources import static_vars
+
+
+class TestSyntheticStaticDS:
+    def test_has_both_static_vars(self):
+        ds = static_vars.build_synthetic_static_ds()
+        for name in static_vars.STATIC_VARS_NEEDED:
+            assert name in ds.data_vars
+
+    def test_full_0p25_grid_shape(self):
+        ds = static_vars.build_synthetic_static_ds()
+        assert ds.sizes["latitude"] == 721
+        assert ds.sizes["longitude"] == 1440
+
+    def test_small_grid_shape(self):
+        lat = np.linspace(14.0, 8.0, 25, dtype=np.float32)
+        lon = np.linspace(74.0, 80.0, 25, dtype=np.float32)
+        ds = static_vars.build_synthetic_static_ds(lat_vals=lat, lon_vals=lon)
+        assert ds.sizes["latitude"] == 25
+        assert ds.sizes["longitude"] == 25
+
+    def test_india_region_marked_as_land(self):
+        ds = static_vars.build_synthetic_static_ds()
+        lsm_india = ds["land_sea_mask"].sel(
+            latitude=slice(14.0, 8.0), longitude=slice(74.0, 80.0)
+        )
+        # At least some India grid cells are land — sanity check the synthetic
+        # setup, not a climatological claim.
+        assert (lsm_india.values > 0).any()
+
+    def test_static_ds_has_no_time_or_level_dims(self):
+        ds = static_vars.build_synthetic_static_ds()
+        assert "time" not in ds.dims
+        assert "level" not in ds.dims
+
+
+class TestCachePathResolution:
+    def test_default_path_is_inside_repo(self):
+        # The default resolves to data/init_sources/era5_static.nc under the
+        # project root. Phase 2 will build it on first use.
+        p = static_vars.cache_path()
+        assert p.endswith(os.path.join("data", "init_sources", "era5_static.nc")), p
+
+    def test_env_var_override(self, tmp_path, monkeypatch):
+        override = str(tmp_path / "static_override.nc")
+        monkeypatch.setenv("GFS_STATIC_CACHE", override)
+        assert static_vars.cache_path() == override
+
+
+class TestLoadStaticDS:
+    def test_raises_helpfully_when_cache_missing(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("GFS_STATIC_CACHE", str(tmp_path / "nope.nc"))
+        with pytest.raises(FileNotFoundError):
+            static_vars.load_static_ds()
+
+    def test_loads_a_round_tripped_cache(self, tmp_path, monkeypatch):
+        # Build a synthetic cache, write to disk, verify the loader reads it.
+        cache_path = tmp_path / "fake_static.nc"
+        monkeypatch.setenv("GFS_STATIC_CACHE", str(cache_path))
+
+        ds = static_vars.build_synthetic_static_ds(
+            lat_vals=np.array([10.0, 0.0], dtype=np.float32),
+            lon_vals=np.array([77.0, 80.0], dtype=np.float32),
+        )
+        ds.to_netcdf(cache_path)
+
+        loaded = static_vars.load_static_ds()
+        for name in static_vars.STATIC_VARS_NEEDED:
+            assert name in loaded.data_vars
+        np.testing.assert_array_equal(
+            loaded["land_sea_mask"].values, ds["land_sea_mask"].values
+        )

tests/test_init_sources/test_variable_mapping.py

@@ -0,0 +1,120 @@
+"""Tests for the GFS ↔ ERA5 variable mapping tables.
+
+These are pure-data tests — no network, no cfgrib, no xarray I/O. They
+protect against accidental breakage of the lookup tables that the GFS
+fetcher in ``src/init_sources/gfs.py`` depends on.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from src.init_sources import variable_mapping as vm
+
+
+class TestSurfaceVarsLookup:
+    def test_all_surface_vars_return_short_name_and_scale(self):
+        for era5_name, (short, (scale, offset)) in vm.SURFACE_VARS.items():
+            assert isinstance(era5_name, str) and era5_name
+            assert isinstance(short, str) and short
+            assert isinstance(scale, float)
+            assert isinstance(offset, float)
+
+    def test_expected_surface_vars_present(self):
+        # These 5 surface fields are load-bearing for GraphCast init.
+        expected = {
+            "2m_temperature", "10m_u_component_of_wind",
+            "10m_v_component_of_wind", "mean_sea_level_pressure",
+            "total_precipitation_6hr",
+        }
+        assert expected <= set(vm.SURFACE_VARS.keys()), (
+            "Missing surface vars — GraphCast will refuse to init without them"
+        )
+
+    def test_precipitation_converts_mm_to_m(self):
+        # GFS ``tp`` is kg/m² (≡ mm of liquid water); ERA5 is m.
+        # 12 mm of rain → 0.012 m.
+        scale, offset = vm.SURFACE_VARS["total_precipitation_6hr"][1]
+        assert scale == pytest.approx(1e-3)
+        assert offset == 0.0
+
+    def test_temperature_is_already_kelvin(self):
+        # Both GFS and ERA5 ship t2m in Kelvin — no conversion.
+        scale, offset = vm.SURFACE_VARS["2m_temperature"][1]
+        assert scale == 1.0
+        assert offset == 0.0
+
+
+class TestPressureLevelVarsLookup:
+    def test_geopotential_converts_gpm_to_m2s2(self):
+        # GFS `gh` is geopotential HEIGHT in geopotential metres; ERA5
+        # stores geopotential (height × g) in m²/s². Check the scale is g.
+        scale, offset = vm.PRESSURE_LEVEL_VARS["geopotential"][1]
+        assert scale == pytest.approx(9.80665)
+        assert offset == 0.0
+
+    def test_six_pressure_level_variables_present(self):
+        expected = {
+            "temperature", "specific_humidity", "u_component_of_wind",
+            "v_component_of_wind", "vertical_velocity", "geopotential",
+        }
+        assert expected <= set(vm.PRESSURE_LEVEL_VARS.keys())
+
+
+class TestPressureLevelsSelection:
+    def test_graphcast_has_13_canonical_levels(self):
+        assert len(vm.GRAPHCAST_PRESSURE_LEVELS) == 13
+
+    def test_levels_in_descending_model_atmosphere(self):
+        # GraphCast and GenCast list their levels ascending in pressure
+        # (top of atmosphere → surface). 50 .. 1000 hPa strictly increasing.
+        levels = list(vm.GRAPHCAST_PRESSURE_LEVELS)
+        assert levels == sorted(levels), "levels must be ascending pressure"
+        assert min(levels) >= 50, "lowest level must be ≥ 50 hPa"
+        assert max(levels) <= 1000, "highest level must be ≤ 1000 hPa"
+
+    def test_levels_available_in_gfs_pgrb2_native(self):
+        # GFS pgrb2.0p25 natively provides these levels. If any GraphCast
+        # level isn't in this list, the select-to-13 step would need to
+        # interpolate instead of just subsetting.
+        gfs_native = {
+            10, 20, 30, 40, 50, 70, 100, 150, 200, 250, 300, 350, 400, 450,
+            500, 550, 600, 650, 700, 750, 800, 850, 900, 925, 950, 975, 1000,
+        }
+        assert set(vm.GRAPHCAST_PRESSURE_LEVELS) <= gfs_native
+
+
+class TestHelpers:
+    def test_gfs_short_name_surface(self):
+        assert vm.gfs_short_name("2m_temperature") == "t2m"
+        assert vm.gfs_short_name("mean_sea_level_pressure") == "prmsl"
+
+    def test_gfs_short_name_pressure_level(self):
+        assert vm.gfs_short_name("temperature") == "t"
+        assert vm.gfs_short_name("geopotential") == "gh"
+
+    def test_gfs_short_name_raises_on_unknown(self):
+        with pytest.raises(KeyError):
+            vm.gfs_short_name("not_a_variable")
+
+    def test_unit_convert_temperature_is_identity(self):
+        xs = np.array([270.0, 290.0, 310.0], dtype=np.float32)
+        ys = vm.unit_convert("2m_temperature", xs)
+        np.testing.assert_allclose(ys, xs)
+
+    def test_unit_convert_precipitation_mm_to_m(self):
+        mm = np.array([0.0, 5.0, 12.5], dtype=np.float32)
+        meters = vm.unit_convert("total_precipitation_6hr", mm)
+        np.testing.assert_allclose(meters, [0.0, 0.005, 0.0125], rtol=1e-5)
+
+    def test_unit_convert_geopotential_applies_g(self):
+        gpm = np.array([0.0, 5000.0, 10_000.0], dtype=np.float32)
+        m2s2 = vm.unit_convert("geopotential", gpm)
+        np.testing.assert_allclose(
+            m2s2, [0.0, 5000.0 * 9.80665, 10_000.0 * 9.80665], rtol=1e-6
+        )
+
+    def test_all_era5_names_union(self):
+        names = vm.all_era5_names()
+        assert set(names) == set(vm.SURFACE_VARS) | set(vm.PRESSURE_LEVEL_VARS)