add comparision

.gitignore

@@ -3,4 +3,6 @@ __pycache__/
 *.pyc
 .npz
 ic_cache/
+era5_env_isolated/
+era5_static_cache/
 .idea/

aifs/era5_env.py

@@ -0,0 +1,128 @@
+"""
+aifs.era5_env
+=============
+Bootstraps and drives the isolated environment that talks to EarthMover's
+ERA5 Icechunk store (see :mod:`aifs.era5_worker`).
+
+Why isolated: ``icechunk`` requires ``zarr>=3``, but ``anemoi-datasets``
+(already required for AIFS inference in this Space) pins ``zarr<=2.18``.
+pip cannot satisfy both in one environment. Instead we ``pip install
+--target=`` a private directory on first use, and run the actual ERA5
+reads in a subprocess whose ``PYTHONPATH`` is prepended with that
+directory — the subprocess resolves ``zarr`` to the isolated v3 install
+regardless of what's importable from the main env's site-packages
+(verified: this is plain CPython import-order semantics, not a hack
+specific to zarr).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import platform
+import shutil
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+import numpy as np
+
+_HERE = Path(__file__).resolve().parent
+WORKER_SCRIPT = _HERE / "era5_worker.py"
+ISOLATED_DIR = _HERE.parent / "era5_env_isolated"
+SENTINEL = ISOLATED_DIR / ".bootstrap_ok"
+
+ISOLATED_PACKAGES = ["icechunk>=2.1", "zarr>=3", "pcodec"]
+
+
+def _fingerprint() -> str:
+    """
+    Identifies the machine/interpreter the isolated install's compiled
+    wheels (numpy, icechunk, pcodec) are built for. Compared against the
+    sentinel on every call so a directory built on one machine (e.g. a
+    dev sandbox) is never reused on another (e.g. the actual Space host)
+    — compiled extensions are platform- and Python-version-specific, and
+    a mismatch fails with a confusing "numpy C-extensions" ImportError
+    rather than anything that points at the real cause.
+    """
+    return f"{platform.system()}-{platform.machine()}-py{sys.version_info.major}.{sys.version_info.minor}"
+
+
+def ensure_bootstrapped(log=lambda msg: None) -> None:
+    """Install the isolated zarr>=3 / icechunk stack if not already present for this machine."""
+    fingerprint = _fingerprint()
+    if SENTINEL.exists() and SENTINEL.read_text().strip() == fingerprint:
+        return
+
+    if ISOLATED_DIR.exists():
+        log("🔄  Isolated ERA5 environment was built for a different machine — reinstalling…")
+        shutil.rmtree(ISOLATED_DIR)
+
+    ISOLATED_DIR.mkdir(parents=True, exist_ok=True)
+    log(f"📦  Setting up isolated ERA5 environment (one-time, ~30s)…")
+    result = subprocess.run(
+        [sys.executable, "-m", "pip", "install", "-q", "--target", str(ISOLATED_DIR), *ISOLATED_PACKAGES],
+        capture_output=True, text=True,
+    )
+    if result.returncode != 0:
+        raise RuntimeError(
+            f"Failed to bootstrap the isolated ERA5 environment:\n{result.stderr[-2000:]}"
+        )
+    SENTINEL.write_text(fingerprint)
+    log("✅  Isolated ERA5 environment ready.")
+
+
+def _subprocess_env() -> dict:
+    env = dict(os.environ)
+    existing = env.get("PYTHONPATH", "")
+    env["PYTHONPATH"] = f"{ISOLATED_DIR}{os.pathsep}{existing}" if existing else str(ISOLATED_DIR)
+    return env
+
+
+def fetch_era5_fields(requests: list[dict], log=lambda msg: None, timeout: int = 600) -> tuple[dict, dict]:
+    """
+    Run a batch of ERA5 reads in the isolated subprocess.
+
+    ``requests`` — list of ``{"group", "var", "level", "time_idx"}`` dicts.
+    Returns ``(arrays, meta)`` where ``arrays`` maps request index (int) to
+    a ``(721, 1440)`` float32 grid, and ``meta`` has ``"errors"`` (index ->
+    message, for requests that failed) and ``"resolved_levels"``.
+    """
+    ensure_bootstrapped(log)
+
+    with tempfile.TemporaryDirectory() as tmp:
+        request_path = Path(tmp) / "request.json"
+        response_prefix = Path(tmp) / "response"
+        request_path.write_text(json.dumps(requests))
+
+        proc = subprocess.Popen(
+            [sys.executable, str(WORKER_SCRIPT), str(request_path), str(response_prefix)],
+            env=_subprocess_env(),
+            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
+        )
+        stderr_tail = []
+        try:
+            for line in proc.stdout:
+                line = line.rstrip()
+                stderr_tail.append(line)
+                if line.startswith("PROGRESS") or line.startswith("RETRY"):
+                    log(f"📡  {line}")
+            proc.wait(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            raise RuntimeError("Timed out waiting for the ERA5 worker subprocess.")
+
+        if proc.returncode != 0:
+            raise RuntimeError(
+                "ERA5 worker subprocess failed:\n" + "\n".join(stderr_tail[-30:])
+            )
+
+        npz_path = f"{response_prefix}.npz"
+        meta_path = f"{response_prefix}.meta.json"
+        with np.load(npz_path) as npz:
+            arrays = {int(k): npz[k] for k in npz.files}
+        meta = json.loads(Path(meta_path).read_text())
+        meta["errors"] = {int(k): v for k, v in meta["errors"].items()}
+        meta["resolved_levels"] = {int(k): v for k, v in meta["resolved_levels"].items()}
+        return arrays, meta

aifs/era5_verify.py

@@ -0,0 +1,337 @@
+"""
+aifs.era5_verify
+=================
+Score an AIFS forecast (a list of state dicts from :func:`aifs.forecast.run_forecast`,
+initialised from a *historical* date via ``aifs.initial_conditions.load_ics(date=...)``)
+against EarthMover's public ERA5 reanalysis, plus a persistence baseline (ERA5 held
+fixed at the initial time — "assume nothing changes").
+
+Only forecasts initialised from a past date can be scored this way: a forecast's
+valid times are in the future relative to when it was run, and ERA5 (reanalysis)
+for those times doesn't exist until well after the fact.
+"""
+
+from __future__ import annotations
+
+import datetime
+
+import numpy as np
+
+from aifs.era5_env import fetch_era5_fields
+
+STEP_HOURS = 6  # AIFS advances in fixed 6-hour steps (see aifs.forecast)
+_EPOCH = datetime.datetime(1940, 1, 1)  # ERA5's valid_time units: "hours since 1940-01-01"
+
+#: AIFS field name -> (ERA5 group, ERA5 variable name, pressure level or None).
+#: Names/groups confirmed against the live store; wave fields (swh, mwp, ...) are
+#: NOT included here — EarthMover's ERA5 mirror doesn't carry the wave stream.
+EVAL_FIELD_MAP = {
+    "2t":    ("single",   "t2m", None),
+    "10u":   ("single",   "u10", None),
+    "10v":   ("single",   "v10", None),
+    "msl":   ("single",   "msl", None),
+    "sp":    ("single",   "sp",  None),
+    "tcw":   ("single",   "tcw", None),
+    "t_850": ("pressure", "t",   850),
+    "t_500": ("pressure", "t",   500),
+    "u_850": ("pressure", "u",   850),
+    "v_850": ("pressure", "v",   850),
+    "z_500": ("pressure", "z",   500),
+    "q_700": ("pressure", "q",   700),
+}
+
+#: ERA5's regular 0.25 degree grid (see aifs.era5_worker probing): lat descends
+#: 90 -> -90, lon ascends 0 -> 359.75 — same [0, 360) convention AIFS uses.
+_ERA5_LAT0, _ERA5_DLAT, _ERA5_NLAT = 90.0, -0.25, 721
+_ERA5_LON0, _ERA5_DLON, _ERA5_NLON = 0.0, 0.25, 1440
+
+
+def _time_idx(dt: datetime.datetime) -> int:
+    return int((dt - _EPOCH).total_seconds() // 3600)
+
+
+def _nearest_indices(lats: np.ndarray, lons: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
+    """Map each (lat, lon) point to its nearest cell in ERA5's regular grid."""
+    lat_idx = np.clip(np.round((lats - _ERA5_LAT0) / _ERA5_DLAT).astype(int), 0, _ERA5_NLAT - 1)
+    lon_idx = np.round((lons - _ERA5_LON0) / _ERA5_DLON).astype(int) % _ERA5_NLON
+    return lat_idx, lon_idx
+
+
+def sample_at_points(grid: np.ndarray, lat_idx: np.ndarray, lon_idx: np.ndarray) -> np.ndarray:
+    """Sample a (721, 1440) ERA5 grid at pre-computed nearest-neighbour indices."""
+    return grid[lat_idx, lon_idx]
+
+
+def metrics(forecast: np.ndarray, truth: np.ndarray) -> dict:
+    """RMSE / MAE / bias / correlation between two same-shape point arrays, NaN-safe."""
+    mask = ~(np.isnan(forecast) | np.isnan(truth))
+    f, t = np.asarray(forecast)[mask], np.asarray(truth)[mask]
+    if len(f) < 2:
+        return {"rmse": float("nan"), "mae": float("nan"), "bias": float("nan"), "corr": float("nan"), "n": int(len(f))}
+    diff = f - t
+    corr = float(np.corrcoef(f, t)[0, 1])
+    return {
+        "rmse": float(np.sqrt(np.mean(diff ** 2))),
+        "mae": float(np.mean(np.abs(diff))),
+        "bias": float(np.mean(diff)),
+        "corr": corr,
+        "n": int(len(f)),
+    }
+
+
+def _fetch_era5_bounds(group: str, extra_requests: list[dict], log):
+    """
+    Fetch a group's live ``valid_time`` coverage bounds together with
+    ``extra_requests`` in one batched call — avoids paying icechunk's
+    session-open cost twice. EarthMover's free ERA5 archive is NOT current
+    to "now" (it lags real-time by several months, and how far isn't
+    documented reliably), so callers check this live rather than trusting
+    a hardcoded date.
+
+    Returns ``(era5_start, era5_end, arrays, meta)`` — ``arrays``/``meta``
+    cover ALL requests: bounds are indices 0 and 1, ``extra_requests``
+    start at index 2.
+    """
+    bounds_requests = [
+        {"group": group, "var": "valid_time", "level": None, "time_idx": 0},
+        {"group": group, "var": "valid_time", "level": None, "time_idx": -1},
+    ]
+    arrays, meta = fetch_era5_fields(bounds_requests + extra_requests, log=log)
+    if 0 in meta["errors"] or 1 in meta["errors"]:
+        raise RuntimeError(
+            f"Could not read ERA5's time coverage for group '{group}': "
+            f"{meta['errors'].get(0) or meta['errors'].get(1)}"
+        )
+    era5_start = _EPOCH + datetime.timedelta(hours=float(arrays[0]))
+    era5_end = _EPOCH + datetime.timedelta(hours=float(arrays[1]))
+    return era5_start, era5_end, arrays, meta
+
+
+def evaluate_forecast(states: list[dict], field_name: str, log=lambda msg: None) -> dict:
+    """
+    Score every step of a historical-init forecast against ERA5, plus a
+    persistence baseline (ERA5 held fixed at t0).
+
+    Returns a dict: ``{"field", "lats", "lons", "per_step": [...]}`` — each
+    per_step entry has ``date``, ``lead_h``, ``forecast``, ``truth`` (both
+    sampled onto the AIFS grid), and ``model_*``/``persistence_*`` metrics.
+    """
+    if not states:
+        raise ValueError("No forecast states to evaluate — run a forecast first.")
+    if field_name not in EVAL_FIELD_MAP:
+        raise ValueError(
+            f"'{field_name}' has no ERA5 counterpart configured. "
+            f"Available: {sorted(EVAL_FIELD_MAP)}"
+        )
+
+    group, var, level = EVAL_FIELD_MAP[field_name]
+    lats = np.asarray(states[0]["latitudes"]).ravel()
+    lons = np.asarray(states[0]["longitudes"]).ravel()
+    lat_idx, lon_idx = _nearest_indices(lats, lons)
+
+    t0 = states[0]["date"] - datetime.timedelta(hours=STEP_HOURS)
+    valid_times = [t0] + [s["date"] for s in states]
+
+    # De-duplicate in case any two steps land on the same valid time.
+    unique_times = sorted(set(valid_times))
+    data_requests = [
+        {"group": group, "var": var, "level": level, "time_idx": _time_idx(t)}
+        for t in unique_times
+    ]
+
+    log(f"📡  Fetching ERA5 '{var}'" + (f"@{level}hPa" if level else "") +
+        f" for {len(unique_times)} valid time(s)…")
+    era5_start, era5_end, arrays, meta = _fetch_era5_bounds(group, data_requests, log)
+
+    earliest_needed, latest_needed = unique_times[0], unique_times[-1]
+    if latest_needed > era5_end or earliest_needed < era5_start:
+        raise ValueError(
+            f"This forecast needs ERA5 truth from {earliest_needed} to {latest_needed}, but "
+            f"EarthMover's free ERA5 archive currently only covers {era5_start} to {era5_end}. "
+            f"Pick an earlier historical init date — with enough room before {era5_end} to fit "
+            f"your full lead time — and run the forecast again."
+        )
+
+    data_errors = {i - 2: err for i, err in meta["errors"].items() if i >= 2}
+    if data_errors:
+        first_err = next(iter(data_errors.values()))
+        raise RuntimeError(f"Could not read '{field_name}' from EarthMover's ERA5 store: {first_err}")
+
+    truth_by_time = {
+        unique_times[i]: sample_at_points(arrays[i + 2], lat_idx, lon_idx)
+        for i in range(len(unique_times))
+    }
+
+    truth_t0 = truth_by_time[t0]
+    per_step = []
+    for i, state in enumerate(states, start=1):
+        truth = truth_by_time[state["date"]]
+        forecast = np.asarray(state["fields"][field_name])
+
+        m = metrics(forecast, truth)
+        p = metrics(truth_t0, truth)
+
+        per_step.append({
+            "date": str(state["date"]),
+            "lead_h": i * STEP_HOURS,
+            "forecast": forecast,
+            "truth": truth,
+            **{f"model_{k}": v for k, v in m.items()},
+            **{f"persistence_{k}": v for k, v in p.items()},
+        })
+
+    log(f"✅  Scored {len(per_step)} step(s) against ERA5 for '{field_name}'.")
+    return {"field": field_name, "lats": lats, "lons": lons, "per_step": per_step}
+
+
+def compute_climatology(state: dict, field_name: str, num_years: int = 10, log=lambda msg: None) -> dict:
+    """
+    Compare ONE forecast state's field against ERA5's climatology for that
+    calendar day/hour — the mean (and spread) over ``num_years`` years of
+    ERA5 history on the same month/day/hour.
+
+    Unlike :func:`evaluate_forecast`, this works for a LIVE (today's or a
+    future) forecast: no ERA5 truth is needed for the forecast's own valid
+    time, only for many *past* years on the same calendar day — sidestepping
+    ERA5's coverage-ceiling problem entirely. It answers "is this forecast
+    unusual for the time of year?", not "is this forecast correct?".
+    """
+    if field_name not in EVAL_FIELD_MAP:
+        raise ValueError(
+            f"'{field_name}' has no ERA5 counterpart configured. "
+            f"Available: {sorted(EVAL_FIELD_MAP)}"
+        )
+    if num_years < 2:
+        raise ValueError("num_years must be at least 2 to compute a meaningful climatology.")
+
+    group, var, level = EVAL_FIELD_MAP[field_name]
+    lats = np.asarray(state["latitudes"]).ravel()
+    lons = np.asarray(state["longitudes"]).ravel()
+    lat_idx, lon_idx = _nearest_indices(lats, lons)
+
+    valid_date: datetime.datetime = state["date"]
+
+    # Request a few extra candidate years beyond num_years so that skipped
+    # years (Feb 29 in a non-leap year, or anything ERA5 doesn't have yet
+    # for the most recent year or two) still leave enough successful
+    # samples — cheaper than pre-checking the archive's bounds separately.
+    BUFFER = 5
+    candidate_years = []
+    year = valid_date.year - 1  # climatology uses PAST years only
+    while len(candidate_years) < num_years + BUFFER and year >= 1940:
+        try:
+            candidate_years.append(valid_date.replace(year=year))
+        except ValueError:
+            pass  # Feb 29 in a non-leap year — skip it
+        year -= 1
+
+    requests = [
+        {"group": group, "var": var, "level": level, "time_idx": _time_idx(t)}
+        for t in candidate_years
+    ]
+    log(f"📡  Fetching up to {num_years} years of ERA5 '{var}'" + (f"@{level}hPa" if level else "") +
+        f" for {valid_date.strftime('%m-%d %H:%M')} UTC…")
+    arrays, meta = fetch_era5_fields(requests, log=log)
+
+    successful = [(candidate_years[i], arrays[i]) for i in range(len(candidate_years)) if i not in meta["errors"]]
+    if len(successful) < 2:
+        first_err = next(iter(meta["errors"].values()), "no data available")
+        raise RuntimeError(
+            f"Could not build a climatology for '{field_name}' — only {len(successful)} of "
+            f"{len(candidate_years)} candidate years had ERA5 data ({first_err})."
+        )
+    successful = successful[:num_years]
+    used_years = [t.year for t, _ in successful]
+
+    samples = np.stack([sample_at_points(grid, lat_idx, lon_idx) for _, grid in successful])
+    clim_mean = np.nanmean(samples, axis=0)
+    clim_std = np.nanstd(samples, axis=0)
+
+    forecast = np.asarray(state["fields"][field_name])
+    anomaly = forecast - clim_mean
+    zscore = np.full_like(anomaly, np.nan)
+    valid_std = clim_std > 1e-6
+    zscore[valid_std] = anomaly[valid_std] / clim_std[valid_std]
+
+    log(f"✅  Built a {len(used_years)}-year climatology ({min(used_years)}–{max(used_years)}) for '{field_name}'.")
+
+    return {
+        "field": field_name, "lats": lats, "lons": lons,
+        "date": str(valid_date), "years": used_years,
+        "forecast": forecast, "climatology_mean": clim_mean, "climatology_std": clim_std,
+        "anomaly": anomaly, "zscore": zscore,
+    }
+
+
+# ── Plotting (matches aifs.plot.plot_field's Cartopy tricontourf style) ────────
+
+def _map_figure(lats, lons, data, title: str, cmap: str = "RdBu_r"):
+    import cartopy.crs as ccrs
+    import cartopy.feature as cfeature
+    import matplotlib.pyplot as plt
+    import matplotlib.tri as tri
+
+    lons_plot = np.where(lons > 180, lons - 360, lons)
+    triangulation = tri.Triangulation(lons_plot, lats)
+
+    fig, ax = plt.subplots(figsize=(9, 5), subplot_kw={"projection": ccrs.PlateCarree()})
+    ax.coastlines()
+    ax.add_feature(cfeature.BORDERS, linestyle=":")
+    contour = ax.tricontourf(triangulation, data, levels=20, transform=ccrs.PlateCarree(), cmap=cmap)
+    fig.colorbar(contour, ax=ax, orientation="vertical", shrink=0.7)
+    ax.set_title(title, fontsize=11)
+    fig.tight_layout()
+    return fig
+
+
+def plot_eval_maps(eval_data: dict, step_index: int):
+    """Forecast / ERA5 truth / (forecast - truth) maps for one evaluated step."""
+    step = eval_data["per_step"][step_index]
+    lats, lons, field = eval_data["lats"], eval_data["lons"], eval_data["field"]
+
+    fig_fc = _map_figure(lats, lons, step["forecast"], f"AIFS forecast — {field} @ {step['date']}")
+    fig_truth = _map_figure(lats, lons, step["truth"], f"ERA5 truth — {field} @ {step['date']}")
+    fig_diff = _map_figure(
+        lats, lons, step["forecast"] - step["truth"],
+        f"Forecast − ERA5 — {field} @ {step['date']}",
+    )
+    return fig_fc, fig_truth, fig_diff
+
+
+def plot_climatology_maps(clim_data: dict):
+    """Forecast / ERA5 climatology-mean / anomaly maps for one field."""
+    lats, lons, field = clim_data["lats"], clim_data["lons"], clim_data["field"]
+    year_range = f"{min(clim_data['years'])}–{max(clim_data['years'])}"
+    n = len(clim_data["years"])
+
+    fig_fc = _map_figure(lats, lons, clim_data["forecast"], f"AIFS forecast — {field} @ {clim_data['date']}")
+    fig_clim = _map_figure(
+        lats, lons, clim_data["climatology_mean"],
+        f"ERA5 climatology mean — {field} ({year_range}, n={n})",
+    )
+    fig_anom = _map_figure(
+        lats, lons, clim_data["anomaly"],
+        f"Anomaly (forecast − climatology) — {field}",
+    )
+    return fig_fc, fig_clim, fig_anom
+
+
+def plot_skill_curve(eval_data: dict):
+    """RMSE vs lead time: AIFS forecast vs a persistence (ERA5-held-at-t0) baseline."""
+    import matplotlib.pyplot as plt
+
+    steps = eval_data["per_step"]
+    lead_h = [s["lead_h"] for s in steps]
+    model_rmse = [s["model_rmse"] for s in steps]
+    persistence_rmse = [s["persistence_rmse"] for s in steps]
+
+    fig, ax = plt.subplots(figsize=(7, 4))
+    ax.plot(lead_h, model_rmse, marker="o", label="AIFS forecast", color="#1f6feb")
+    ax.plot(lead_h, persistence_rmse, marker="o", linestyle="--", label="Persistence (t0 held fixed)", color="#8b949e")
+    ax.set_xlabel("Lead time (hours)")
+    ax.set_ylabel("RMSE")
+    ax.set_title(f"Skill vs ERA5 — {eval_data['field']}")
+    ax.legend()
+    ax.grid(alpha=0.3)
+    fig.tight_layout()
+    return fig

aifs/era5_worker.py

@@ -0,0 +1,130 @@
+"""
+aifs.era5_worker
+================
+Standalone subprocess entry point — reads EarthMover's public ERA5
+Icechunk/Zarr-v3 store on S3 (anonymous access, no account needed).
+
+This script is deliberately NOT imported by the rest of the app. It is
+invoked via ``subprocess`` with ``PYTHONPATH`` pointed at an isolated
+``pip install --target`` directory (see :mod:`aifs.era5_env`), because
+``icechunk`` requires ``zarr>=3`` while ``anemoi-datasets`` (already a
+hard dependency of this Space, for AIFS inference) pins ``zarr<=2.18``.
+The two cannot coexist in one interpreter's import path, so this script
+only ever runs in a separate process with its own isolated zarr install.
+
+Only stdlib + numpy + icechunk + zarr are imported here — keep it that
+way so it never accidentally picks up the main env's (incompatible)
+zarr via a transitive import.
+
+Usage
+-----
+    python3 era5_worker.py <request.json> <response_prefix>
+
+``request.json`` is a list of ``{"group", "var", "level", "time_idx"}``
+dicts. Writes ``<response_prefix>.npz`` (arrays keyed by request index,
+one per successfully-read request) and ``<response_prefix>.meta.json``
+(``{"errors": {...}, "resolved_levels": {...}}``). Progress is reported
+as ``PROGRESS <i>/<n> <group>/<var>`` lines on stdout.
+"""
+
+import json
+import sys
+import time
+
+import numpy as np
+
+BUCKET = "earthmover-icechunk-era5"
+PREFIX = "icechunkV2"
+REGION = "us-east-1"
+
+MAX_RETRIES = 6
+_RETRIABLE_KEYWORDS = (
+    "429", "rate limit", "too many requests", "timeout", "connection reset",
+    "503", "service unavailable", "throughput", "streaming error", "i/o error",
+)
+
+
+def _retriable(exc: Exception) -> bool:
+    msg = str(exc).lower()
+    return any(k in msg for k in _RETRIABLE_KEYWORDS)
+
+
+def _with_retry(fn, label: str):
+    for attempt in range(MAX_RETRIES):
+        try:
+            return fn()
+        except Exception as exc:
+            if attempt < MAX_RETRIES - 1 and _retriable(exc):
+                wait = min(3 * (2 ** attempt), 30)
+                print(f"RETRY {label}: {exc} (attempt {attempt + 2}/{MAX_RETRIES}, waiting {wait}s)", flush=True)
+                time.sleep(wait)
+            else:
+                raise
+
+
+def _open_store():
+    import icechunk
+
+    def _open():
+        storage = icechunk.s3_storage(bucket=BUCKET, prefix=PREFIX, region=REGION, anonymous=True)
+        repo = icechunk.Repository.open(storage)
+        session = repo.readonly_session("main")
+        return session.store
+
+    return _with_retry(_open, "open icechunk repo")
+
+
+def main():
+    request_path, response_prefix = sys.argv[1], sys.argv[2]
+    with open(request_path) as f:
+        requests = json.load(f)
+
+    import zarr
+
+    store = _open_store()
+    groups = {}
+    level_coords = {}
+
+    results = {}
+    errors = {}
+    resolved_levels = {}
+
+    n = len(requests)
+    for i, req in enumerate(requests):
+        group, var, level, time_idx = req["group"], req["var"], req.get("level"), req["time_idx"]
+        print(f"PROGRESS {i + 1}/{n} {group}/{var}" + (f"@{level}hPa" if level is not None else ""), flush=True)
+        try:
+            if group not in groups:
+                groups[group] = _with_retry(
+                    lambda g=group: zarr.open_group(store, mode="r", path=f"{g}/spatial"),
+                    f"open group {group}",
+                )
+            g = groups[group]
+
+            if var not in g.array_keys():
+                raise KeyError(f"'{var}' not found in ERA5 group '{group}' (have: {sorted(g.array_keys())})")
+
+            arr = g[var]
+
+            def _read():
+                if level is not None:
+                    if group not in level_coords:
+                        level_coords[group] = np.asarray(g["pressure_level"][:])
+                    levels = level_coords[group]
+                    pos = int(np.argmin(np.abs(levels - level)))
+                    resolved_levels[str(i)] = float(levels[pos])
+                    return arr[time_idx, pos]
+                return arr[time_idx]
+
+            data = _with_retry(_read, f"read {group}/{var}")
+            results[str(i)] = np.asarray(data, dtype=np.float32)
+        except Exception as exc:
+            errors[str(i)] = str(exc)
+
+    np.savez_compressed(f"{response_prefix}.npz", **results)
+    with open(f"{response_prefix}.meta.json", "w") as f:
+        json.dump({"errors": errors, "resolved_levels": resolved_levels}, f)
+
+
+if __name__ == "__main__":
+    main()

aifs/initial_conditions.py

@@ -35,6 +35,32 @@ LEVELS   = [1000, 925, 850, 700, 600, 500, 400, 300, 250, 200, 150, 100, 50, 10]
 
 SOURCE = "ecmwf"
 
+#: ECMWF Open Data's primary endpoint ("ecmwf") only keeps a rolling ~4-day
+#: window. The "aws" named source (an S3 mirror, still via the same
+#: ecmwf-opendata/earthkit-data client) retains a much deeper archive
+#: (observed back to 2023-01-18) — used for historical initial conditions.
+HISTORICAL_SOURCE = "aws"
+
+#: Valid ECMWF synoptic run hours (UTC) — the archive only has data at these.
+VALID_RUN_HOURS = (0, 6, 12, 18)
+
+#: Earliest date the "aws" historical archive has been observed to serve.
+EARLIEST_HISTORICAL_DATE = datetime.date(2023, 1, 18)
+
+#: Before this date, ECMWF's 06/18 UTC runs used a reduced product ("scda" for
+#: pressure levels, "scwv" for waves) instead of the full "oper"/"wave" stream
+#: — confirmed missing: all pressure-level vars at 10 hPa, and everything in
+#: PARAM_WAVE except mwd/mwp/swh. 00/12 UTC always used the full stream, even
+#: before this cutover. This was unified across all run hours starting on this
+#: date, but that's *after* EarthMover's free ERA5 archive's coverage ends (see
+#: aifs.era5_verify) — so for this app's actual use case (historical init +
+#: ERA5 verification), 06/18 UTC is never a usable choice anyway.
+REDUCED_PRODUCT_CUTOVER = datetime.date(2026, 5, 12)
+REDUCED_PRODUCT_HOURS = (6, 18)
+
+#: Run hours guaranteed to carry the full field set, at any historical date.
+FULL_FIELD_RUN_HOURS = (0, 12)
+
 # ── Cache helpers ─────────────────────────────────────────────────────────────
 
 DEFAULT_CACHE_DIR = Path("ic_cache")
@@ -97,7 +123,7 @@ def _fetch_with_retry(ekd, ekr, date, param, max_retries: int = 6, **kwargs):
                 raise
 
 
-def _fetch_fields(ekd, ekr, date, param, levelist=None, **kwargs) -> dict:
+def _fetch_fields(ekd, ekr, date, param, levelist=None, source=SOURCE, **kwargs) -> dict:
     """
     Download ``param`` for two time-steps (t-6h, t) and return a dict
     ``{variable_name: np.ndarray shape (2, N320_nodes)}``.
@@ -111,7 +137,7 @@ def _fetch_fields(ekd, ekr, date, param, levelist=None, **kwargs) -> dict:
             date=t,
             param=param,
             levelist=levelist,
-            source=SOURCE,
+            source=source,
             **kwargs,
         )
         for field in dataset:
@@ -134,7 +160,7 @@ def _fetch_fields(ekd, ekr, date, param, levelist=None, **kwargs) -> dict:
 
 
 
-def _build_fields(ekd, ekr, date: datetime.datetime):
+def _build_fields(ekd, ekr, date: datetime.datetime, source: str = SOURCE):
     """Download and transform all required fields for ``date``."""
     fields: dict = {}
     log_lines: list[str] = []
@@ -148,6 +174,7 @@ def _build_fields(ekd, ekr, date: datetime.datetime):
         """Yield log messages then return the result dict."""
         yield log(label)
         result = None
+        kwargs.setdefault("source", source)
         for kind, payload in _fetch_with_retry(ekd, ekr, *args, **kwargs):
             if kind == "log":
                 yield log(f"  {payload}")
@@ -216,10 +243,21 @@ def _build_fields(ekd, ekr, date: datetime.datetime):
 def load_ics(
     cache_dir: Path | str = DEFAULT_CACHE_DIR,
     force: bool = False,
+    date: datetime.datetime | None = None,
 ):
     """
     Generator version of load_ics.
 
+    Parameters
+    ----------
+    date:
+        If ``None`` (default), fetch the latest available run from
+        ECMWF Open Data's primary endpoint — unchanged live behaviour.
+        If given, fetch that specific historical run instead, from the
+        "aws" S3 mirror (a much deeper archive than the ~4-day rolling
+        window of the primary endpoint — observed back to 2023-01-18).
+        Must fall on a synoptic run hour (00/06/12/18 UTC).
+
     Yields
     ------
     ("log", str)               -- progress messages
@@ -227,13 +265,29 @@ def load_ics(
     """
     import earthkit.data as ekd
     import earthkit.regrid as ekr
-    from ecmwf.opendata import Client as OpendataClient
 
     ekd.config.set({"cache-policy": "user"})
     cache_dir = Path(cache_dir)
 
-    date: datetime.datetime = OpendataClient(SOURCE).latest()
-    yield "log", f"📅  Latest ECMWF run: {date}"
+    if date is None:
+        from ecmwf.opendata import Client as OpendataClient
+
+        date = OpendataClient(SOURCE).latest()
+        yield "log", f"📅  Latest ECMWF run: {date}"
+        source = SOURCE
+    else:
+        if date.hour not in VALID_RUN_HOURS or date.minute or date.second or date.microsecond:
+            raise ValueError(
+                f"date must fall on a synoptic run hour {VALID_RUN_HOURS} UTC, got {date}"
+            )
+        if date.hour in REDUCED_PRODUCT_HOURS and date.date() < REDUCED_PRODUCT_CUTOVER:
+            raise ValueError(
+                f"{date} is a 06/18 UTC run before {REDUCED_PRODUCT_CUTOVER.isoformat()} — "
+                "ECMWF served a reduced product at those hours back then (missing 10 hPa "
+                "pressure levels and most wave fields AIFS needs). Use a 00 or 12 UTC run instead."
+            )
+        yield "log", f"📅  Historical ECMWF run: {date} (via '{HISTORICAL_SOURCE}' archive)"
+        source = HISTORICAL_SOURCE
 
     if not force:
         cached, path = _try_load(date, cache_dir)
@@ -246,7 +300,7 @@ def load_ics(
     yield "log", "⬇️   Downloading initial conditions …"
 
     fields = None
-    for kind, payload in _build_fields(ekd, ekr, date):
+    for kind, payload in _build_fields(ekd, ekr, date, source=source):
         if kind == "log":
             yield "log", payload
         else:  # "result"

app.py

@@ -1,3 +1,4 @@
+import datetime
 import os
 import ssl
 import warnings
@@ -19,6 +20,8 @@ import gradio as gr
 import tempfile
 
 from aifs.device import  device_label
+from aifs.initial_conditions import VALID_RUN_HOURS, EARLIEST_HISTORICAL_DATE, FULL_FIELD_RUN_HOURS
+from aifs.era5_verify import EVAL_FIELD_MAP
 
 import shutil
 
@@ -59,17 +62,40 @@ def _run_forecast_gpu(fields, date, lead_time, num_chunks):
     yield from _run_forecast(fields, date, lead_time=lead_time, num_chunks=num_chunks)
 
 
-def run_forecast(lead_time: int, num_chunks: int):
+def _parse_historical_date(date_str: str, hour_str: str) -> datetime.datetime:
+    """Parse and validate the historical-date UI inputs, or raise ValueError."""
+    try:
+        year, month, day = (int(p) for p in date_str.strip().split("-"))
+        picked = datetime.date(year, month, day)
+    except Exception:
+        raise ValueError(f"could not parse date '{date_str}' — use YYYY-MM-DD.")
+
+    if picked < EARLIEST_HISTORICAL_DATE or picked > datetime.date.today():
+        raise ValueError(
+            f"date must be between {EARLIEST_HISTORICAL_DATE.isoformat()} and today, got {picked}."
+        )
+    return datetime.datetime(picked.year, picked.month, picked.day, int(hour_str))
+
+
+def run_forecast(lead_time: int, num_chunks: int, ic_mode: str, hist_date_str: str, hist_hour: str):
     from aifs.initial_conditions import load_ics
 
     def emit(status, phase, dd=gr.update(choices=[]), states=[], btn=_BTN_RUNNING):
         return status, _phase(phase), dd, states, btn
 
+    historical_dt = None
+    if ic_mode == "Historical date":
+        try:
+            historical_dt = _parse_historical_date(hist_date_str, hist_hour)
+        except ValueError as exc:
+            yield emit(f"❌  {exc}", "❌ Invalid historical date", btn=_BTN_READY)
+            return
+
     yield emit("Starting up…", "📥 Downloading initial conditions from ECMWF…")
 
     try:
         fields = date = None
-        for kind, payload in load_ics(cache_dir="ic_cache"):
+        for kind, payload in load_ics(cache_dir="ic_cache", date=historical_dt):
             if kind == "log":
                 yield emit(payload, "📥 Downloading initial conditions from ECMWF…")
             else:
@@ -137,6 +163,193 @@ def plot_forecast_handler(field_name: str, timestamp: str, states: list):
     )
     return path, stats
 
+# ── ERA5 evaluation (historical runs only) ─────────────────────────────────────
+
+_EVAL_BTN_RUNNING = gr.update(interactive=False, value="⏳ Evaluating…")
+_EVAL_BTN_READY   = gr.update(interactive=True,  value="📡 Evaluate vs ERA5")
+
+
+def run_evaluation(field_name: str, states: list):
+    """
+    Thread + queue wrapper around aifs.era5_verify.evaluate_forecast (a plain
+    blocking call with a log callback) so progress streams to the UI the same
+    way run_forecast's GPU step does.
+    """
+    import queue
+    import threading
+
+    from aifs.era5_verify import evaluate_forecast
+
+    def emit(status, dd=gr.update(choices=[]), eval_data=None, skill_path=None, btn=_EVAL_BTN_RUNNING):
+        return status, dd, eval_data if eval_data is not None else {}, skill_path, btn
+
+    if not states:
+        yield emit("⚠️  Run a forecast from a historical date first, then evaluate it.", btn=_EVAL_BTN_READY)
+        return
+
+    result_queue: queue.Queue = queue.Queue()
+
+    def _worker():
+        try:
+            data = evaluate_forecast(states, field_name, log=lambda msg: result_queue.put(("log", msg)))
+            result_queue.put(("done", data))
+        except Exception as exc:
+            result_queue.put(("error", exc))
+
+    thread = threading.Thread(target=_worker, daemon=True)
+    thread.start()
+
+    log_lines: list[str] = []
+    eval_data = None
+    while True:
+        kind, payload = result_queue.get()
+        if kind == "log":
+            log_lines.append(payload)
+            yield emit("\n".join(log_lines))
+        elif kind == "done":
+            eval_data = payload
+            break
+        elif kind == "error":
+            yield emit("\n".join(log_lines) + f"\n❌  Error: {payload}", btn=_EVAL_BTN_READY)
+            return
+
+    thread.join(timeout=5.0)
+
+    from aifs.era5_verify import plot_skill_curve
+
+    skill_path = os.path.join(tempfile.gettempdir(), "aifs_eval_skill.png")
+    plot_skill_curve(eval_data).savefig(skill_path, dpi=150, bbox_inches="tight")
+
+    timestamps = [s["date"] for s in eval_data["per_step"]]
+    yield (
+        f"✅  Evaluated {len(timestamps)} step(s) against ERA5 for '{field_name}'.",
+        gr.update(choices=timestamps, value=timestamps[-1]),
+        eval_data,
+        skill_path,
+        _EVAL_BTN_READY,
+    )
+
+
+def plot_eval_handler(timestamp: str, eval_data: dict):
+    from aifs.era5_verify import plot_eval_maps
+
+    if not eval_data or not eval_data.get("per_step"):
+        return None, None, None, "Run an evaluation first."
+
+    per_step = eval_data["per_step"]
+    idx = next((i for i, s in enumerate(per_step) if s["date"] == timestamp), len(per_step) - 1)
+    step = per_step[idx]
+
+    fig_fc, fig_truth, fig_diff = plot_eval_maps(eval_data, idx)
+    paths = []
+    for name, fig in zip(("fc", "truth", "diff"), (fig_fc, fig_truth, fig_diff)):
+        path = os.path.join(tempfile.gettempdir(), f"aifs_eval_{name}.png")
+        fig.savefig(path, dpi=150, bbox_inches="tight")
+        paths.append(path)
+
+    units = UNITS_MAP.get(eval_data["field"], "")
+    stats = (
+        f"**{eval_data['field']}** @ {step['date']}  (lead +{step['lead_h']}h)\n\n"
+        f"| | AIFS forecast | Persistence (t0) |\n|---|---|---|\n"
+        f"| RMSE | `{step['model_rmse']:.4g}` {units} | `{step['persistence_rmse']:.4g}` {units} |\n"
+        f"| MAE | `{step['model_mae']:.4g}` {units} | `{step['persistence_mae']:.4g}` {units} |\n"
+        f"| Bias | `{step['model_bias']:.4g}` {units} | `{step['persistence_bias']:.4g}` {units} |\n"
+        f"| Corr. w/ ERA5 | `{step['model_corr']:.3f}` | `{step['persistence_corr']:.3f}` |\n"
+    )
+    return paths[0], paths[1], paths[2], stats
+
+# ── ERA5 climatology (works for any forecast — live or historical) ────────────
+
+_CLIM_BTN_RUNNING = gr.update(interactive=False, value="⏳ Computing…")
+_CLIM_BTN_READY   = gr.update(interactive=True,  value="🌡️ Compare vs Climatology")
+
+
+def run_climatology(field_name: str, timestamp: str, num_years: int, states: list):
+    """
+    Compares one forecast step's field to its ERA5 climatology (mean over
+    many past years on the same calendar day). Unlike the ERA5-truth
+    evaluation above, this works for ANY forecast — including a live one
+    initialised from "Latest" — since it never needs ERA5 truth for the
+    forecast's own (possibly future) valid time, only for past years.
+    """
+    import queue
+    import threading
+
+    import numpy as np
+
+    from aifs.era5_verify import compute_climatology
+
+    def emit(status, fc_path=None, clim_path=None, anom_path=None, stats="", btn=_CLIM_BTN_RUNNING):
+        return status, fc_path, clim_path, anom_path, stats, btn
+
+    if not states:
+        yield emit("⚠️  Run a forecast first, then compare it to climatology.", btn=_CLIM_BTN_READY)
+        return
+
+    state = next((s for s in states if str(s["date"]) == timestamp), states[-1])
+    if field_name not in state.get("fields", {}):
+        yield emit(f"⚠️  Field '{field_name}' not available in this forecast state.", btn=_CLIM_BTN_READY)
+        return
+
+    result_queue: queue.Queue = queue.Queue()
+
+    def _worker():
+        try:
+            data = compute_climatology(
+                state, field_name, num_years=int(num_years),
+                log=lambda msg: result_queue.put(("log", msg)),
+            )
+            result_queue.put(("done", data))
+        except Exception as exc:
+            result_queue.put(("error", exc))
+
+    thread = threading.Thread(target=_worker, daemon=True)
+    thread.start()
+
+    log_lines: list[str] = []
+    clim_data = None
+    while True:
+        kind, payload = result_queue.get()
+        if kind == "log":
+            log_lines.append(payload)
+            yield emit("\n".join(log_lines))
+        elif kind == "done":
+            clim_data = payload
+            break
+        elif kind == "error":
+            yield emit("\n".join(log_lines) + f"\n❌  Error: {payload}", btn=_CLIM_BTN_READY)
+            return
+
+    thread.join(timeout=5.0)
+
+    from aifs.era5_verify import plot_climatology_maps
+
+    fig_fc, fig_clim, fig_anom = plot_climatology_maps(clim_data)
+    paths = []
+    for name, fig in zip(("fc", "clim", "anom"), (fig_fc, fig_clim, fig_anom)):
+        path = os.path.join(tempfile.gettempdir(), f"aifs_clim_{name}.png")
+        fig.savefig(path, dpi=150, bbox_inches="tight")
+        paths.append(path)
+
+    units = UNITS_MAP.get(field_name, "")
+    anomaly, zscore = clim_data["anomaly"], clim_data["zscore"]
+    z_mask = ~np.isnan(zscore)
+    pct_unusual = float(np.mean(np.abs(zscore[z_mask]) > 2) * 100) if z_mask.any() else float("nan")
+    years = clim_data["years"]
+    stats = (
+        f"**{field_name}** @ {clim_data['date']}  ·  {len(years)}-year climatology "
+        f"({min(years)}–{max(years)})\n\n"
+        f"- Mean anomaly: `{np.nanmean(anomaly):.4g}` {units}\n"
+        f"- Mean |z-score|: `{np.nanmean(np.abs(zscore)):.3g}`\n"
+        f"- Area with |z| > 2 (\"unusual for the time of year\"): `{pct_unusual:.1f}%`\n"
+    )
+
+    yield (
+        f"✅  Built a {len(years)}-year climatology for '{field_name}'.",
+        paths[0], paths[1], paths[2], stats,
+        _CLIM_BTN_READY,
+    )
+
 # ── UI ────────────────────────────────────────────────────────────────────────
 DARK_CSS = """
 body, .gradio-container {
@@ -188,6 +401,27 @@ with gr.Blocks(css=DARK_CSS, title="AIFS Single v2 Forecast") as demo:
     with gr.Row():
         with gr.Column(scale=1, elem_classes="panel"):
             gr.Markdown("### ⚙️ Forecast Settings")
+            ic_mode_radio = gr.Radio(
+                ["Latest", "Historical date"], value="Latest",
+                label="Initial conditions",
+                info="Historical dates pull from ECMWF's deeper S3 archive (from "
+                     f"{EARLIEST_HISTORICAL_DATE.isoformat()}) instead of the live feed.",
+            )
+            with gr.Row(visible=False) as historical_row:
+                hist_date_tb = gr.Textbox(
+                    label="Date (UTC)", placeholder="YYYY-MM-DD",
+                    value=(datetime.date.today() - datetime.timedelta(days=7)).isoformat(),
+                )
+                hist_hour_dd = gr.Dropdown(
+                    [f"{h:02d}" for h in FULL_FIELD_RUN_HOURS], value="00",
+                    label="Run hour (UTC)",
+                    info="Limited to 00/12 UTC — 06/18 UTC used a reduced ECMWF product "
+                         "before 2026-05-12 that's missing fields AIFS needs.",
+                )
+            ic_mode_radio.change(
+                fn=lambda mode: gr.update(visible=(mode == "Historical date")),
+                inputs=ic_mode_radio, outputs=historical_row,
+            )
             lead_time_sl = gr.Slider(
                 minimum=6, maximum=96, step=6, value=6,
                 label="Lead time (hours)",
@@ -223,9 +457,79 @@ with gr.Blocks(css=DARK_CSS, title="AIFS Single v2 Forecast") as demo:
             map_img  = gr.Image(label="Global Map", type="filepath")
             stats_md = gr.Markdown()
 
+    gr.Markdown("---")
+    with gr.Row():
+        with gr.Column(scale=1, elem_classes="panel"):
+            gr.Markdown(
+                "### 📊 Evaluate vs ERA5\n"
+                "Only meaningful for a forecast initialised from a **historical date** "
+                "above — ERA5 (EarthMover's public reanalysis) is the ground truth being "
+                "compared against, and it only exists for times already in the past.\n\n"
+                "EarthMover's free ERA5 archive also isn't current to *now* — it lags "
+                "real-time by several months. If a date is too recent (init date + lead "
+                "time falls after the archive's latest coverage), evaluation will fail "
+                "with a message telling you the exact usable window — pick an earlier "
+                "historical date and re-run the forecast above."
+            )
+            eval_field_dd = gr.Dropdown(
+                choices=sorted(EVAL_FIELD_MAP), value="2t",
+                label="Field to evaluate",
+                info="Wave fields aren't available in EarthMover's ERA5 mirror, so they're excluded here.",
+            )
+            eval_btn = gr.Button("📡 Evaluate vs ERA5", variant="primary")
+            eval_status = gr.Textbox(
+                label="Detailed log", lines=6, interactive=False,
+                placeholder="Run a historical forecast above, then evaluate it here…",
+            )
+            eval_step_dd = gr.Dropdown(choices=[], label="Forecast step to inspect")
+            eval_show_btn = gr.Button("🔍 Show Maps for Step", variant="secondary")
+
+        with gr.Column(scale=2, elem_classes="panel"):
+            skill_img = gr.Image(label="Skill vs lead time (RMSE)", type="filepath")
+            with gr.Row():
+                eval_fc_img = gr.Image(label="AIFS forecast", type="filepath")
+                eval_truth_img = gr.Image(label="ERA5 truth", type="filepath")
+                eval_diff_img = gr.Image(label="Forecast − ERA5", type="filepath")
+            eval_stats_md = gr.Markdown()
+
+    eval_state = gr.State({})
+
+    gr.Markdown("---")
+    with gr.Row():
+        with gr.Column(scale=1, elem_classes="panel"):
+            gr.Markdown(
+                "### 🌡️ Compare vs ERA5 Climatology\n"
+                "Works for **any** forecast, including a live one initialised from "
+                "\"Latest\" — this compares a forecast step to the ERA5 average for "
+                "that calendar day over many *past* years, so it never needs ERA5 "
+                "truth for the forecast's own (possibly future) valid time. It answers "
+                "*\"is this forecast unusual for the time of year?\"*, not \"is it correct?\"."
+            )
+            clim_field_dd = gr.Dropdown(
+                choices=sorted(EVAL_FIELD_MAP), value="2t",
+                label="Field to compare",
+            )
+            clim_years_sl = gr.Slider(
+                minimum=3, maximum=30, step=1, value=10,
+                label="Climatology years",
+                info="How many past years of ERA5 to average — more is slower but smoother.",
+            )
+            clim_btn = gr.Button("🌡️ Compare vs Climatology", variant="primary")
+            clim_status = gr.Textbox(
+                label="Detailed log", lines=6, interactive=False,
+                placeholder="Run a forecast above, pick a step in \"Forecast step\", then compare here…",
+            )
+
+        with gr.Column(scale=2, elem_classes="panel"):
+            with gr.Row():
+                clim_fc_img = gr.Image(label="AIFS forecast", type="filepath")
+                clim_mean_img = gr.Image(label="ERA5 climatology mean", type="filepath")
+                clim_anom_img = gr.Image(label="Anomaly (forecast − climatology)", type="filepath")
+            clim_stats_md = gr.Markdown()
+
     run_btn.click(
         fn=run_forecast,
-        inputs=[lead_time_sl, num_chunks_sl],
+        inputs=[lead_time_sl, num_chunks_sl, ic_mode_radio, hist_date_tb, hist_hour_dd],
         outputs=[status_box, phase_md, timestamp_dd, forecast_state, run_btn],
     )
     plot_btn.click(
@@ -233,6 +537,21 @@ with gr.Blocks(css=DARK_CSS, title="AIFS Single v2 Forecast") as demo:
         inputs=[field_dd, timestamp_dd, forecast_state],
         outputs=[map_img, stats_md],
     )
+    eval_btn.click(
+        fn=run_evaluation,
+        inputs=[eval_field_dd, forecast_state],
+        outputs=[eval_status, eval_step_dd, eval_state, skill_img, eval_btn],
+    )
+    eval_show_btn.click(
+        fn=plot_eval_handler,
+        inputs=[eval_step_dd, eval_state],
+        outputs=[eval_fc_img, eval_truth_img, eval_diff_img, eval_stats_md],
+    )
+    clim_btn.click(
+        fn=run_climatology,
+        inputs=[clim_field_dd, timestamp_dd, clim_years_sl, forecast_state],
+        outputs=[clim_status, clim_fc_img, clim_mean_img, clim_anom_img, clim_stats_md, clim_btn],
+    )
 
     gr.Markdown(
         """