"""Resumable ECDS s2s-reforecasts downloader driven by hdate_2023.txt. Scheme (validated): for EC model version 2023, each MM-DD in hdate_2023.txt is a model-version date; hyear=2003..2022 (same MM-DD) yields the full reforecast set. Control forecast only, native 1.5deg (no grid key). Auth from ~/.cdsapirc_1. Variable GROUPS (constraint-validated live): sfc_core : single_level mx2t6, 2t, msl, 10u, 10v (6-hourly leads) pl_hi : pressure T,Z(geopot.height),U,V,q @ 850/700/500/200 hPa (24-h leads) Usage: python3 scripts/05_download_reforecasts.py --group sfc_core --dates 1 python3 scripts/05_download_reforecasts.py --group pl_hi --dates 1 --hyears 2003,2004 python3 scripts/05_download_reforecasts.py --group sfc_core --dates all # gate first! """ import argparse import os import time import yaml ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) OUT = os.path.join(ROOT, "data", "raw", "ecmwf_s2s_1p5deg") os.makedirs(OUT, exist_ok=True) CRED = os.path.expanduser("~/.cdsapirc_1") HDATES = os.path.join(ROOT, "hdate_2023.txt") DATASET = "s2s-reforecasts" MV_YEAR = "2023" # Predictor-branch domains (N,W,S,E). LOCAL = regional processes; LARGE = ENSO/ # MJO/monsoon/teleconnections. Pressure/circulation fields need the LARGE domain. AREA_LOCAL = [30, 95, 5, 120] # 95-120E, 5-30N AREA_LARGE = [50, 40, -20, 180] # 40-180E, 20S-50N SFC6_LEADS = [str(h) for h in range(6, 1105, 6)] # 184 six-hourly windows DAILY_LEADS = [str(h) for h in range(0, 1105, 24)] # 47 daily (24-hourly) steps DAILY_AVG_LEADS = [f"{s}_{s+24}" for s in range(0, 1081, 24)] # 46 daily-mean windows 0_24..1080_1104 # Variables are grouped by NATIVE step frequency so nothing is silently dropped: # mx2t6 is 6-hourly (needed for the daily max); instantaneous/daily fields and all # pressure fields are 24-hourly. GROUPS = { "sfc_tx": { # LOCAL: target-side daily-max Tx source "level_type": "single_level", "area": AREA_LOCAL, "domain": "loc", "variable": ["maximum_2_m_temperature_in_the_last_6_hours"], "leadtime_hour": SFC6_LEADS, }, "sfc_dyn": { # LOCAL: regional surface predictors (daily) "level_type": "single_level", "area": AREA_LOCAL, "domain": "loc", "variable": ["2_m_temperature", "mean_sea_level_pressure", "10_m_u_component_of_wind", "10_m_v_component_of_wind"], "leadtime_hour": DAILY_LEADS, }, "sfc_t2m": { # LOCAL: daily-MEAN 2 m temperature. 2t is archived ONLY as # 24-h averages (0_24..1080_1104); the plain-integer DAILY_LEADS request # in sfc_dyn returned a single 0_24 record, so 2t is pulled separately here. "level_type": "single_level", "area": AREA_LOCAL, "domain": "loc", "variable": ["2_m_temperature"], "leadtime_hour": DAILY_AVG_LEADS, }, "pl_hi": { # LARGE-SCALE: circulation (Z500,U200,T850,q,...) daily "level_type": "pressure", "area": AREA_LARGE, "domain": "ls", "variable": ["temperature", "geopotential_height", "u_component_of_wind", "v_component_of_wind", "specific_humidity"], "level_value": ["850_hpa", "700_hpa", "500_hpa", "200_hpa"], "leadtime_hour": DAILY_LEADS, }, } def read_hdates(): return [l.strip() for l in open(HDATES) if l.strip()] def build_request(mmdd, hyears, group): mm, dd = mmdd.split("-") g = GROUPS[group] req = { "origin": "ecmwf", "year": MV_YEAR, "month": mm, "day": dd, "time": "00:00", "hyear": [str(y) for y in hyears], "hmonth": [mm], "hday": [dd], "level_type": g["level_type"], "variable": g["variable"], "forecast_type": "control_forecast", "leadtime_hour": g["leadtime_hour"], "data_format": "grib", "area": g["area"], } if "level_value" in g: req["level_value"] = g["level_value"] return req def target_path(mmdd, hyears, group): mm, dd = mmdd.split("-") ytag = f"hy{hyears[0]}-{hyears[-1]}" if len(hyears) > 1 else f"hy{hyears[0]}" dtag = GROUPS[group]["domain"] return os.path.join(OUT, f"s2s_cf_{group}_{dtag}_mv{MV_YEAR}-{mm}-{dd}_{ytag}.grib") def main(): ap = argparse.ArgumentParser() ap.add_argument("--group", default="sfc_core", choices=list(GROUPS)) ap.add_argument("--dates", default="1", help="N (first N hdates) or 'all'") ap.add_argument("--hyears", default="2003-2022", help="e.g. 2003-2022 or 2003,2004") ap.add_argument("--which", default=None, help="explicit MM-DD list, comma-sep (overrides --dates)") ap.add_argument("--rc", default=CRED, help="path to cdsapirc credentials file (default ~/.cdsapirc_1)") a = ap.parse_args() if "-" in a.hyears and "," not in a.hyears: lo, hi = a.hyears.split("-"); hyears = list(range(int(lo), int(hi) + 1)) else: hyears = [int(x) for x in a.hyears.split(",")] dates = read_hdates() todo = a.which.split(",") if a.which else (dates if a.dates == "all" else dates[:int(a.dates)]) print(f"group={a.group} mv{MV_YEAR}: {len(todo)} date(s) {todo if len(todo)<=6 else todo[:6]+['...']}, " f"hyears {hyears[0]}..{hyears[-1]} ({len(hyears)})", flush=True) import cdsapi cred = yaml.safe_load(open(os.path.expanduser(a.rc))) print(f"creds: {a.rc}", flush=True) client = cdsapi.Client(url=cred["url"], key=cred["key"]) n_ok = n_fail = 0 for i, mmdd in enumerate(todo, 1): tgt = target_path(mmdd, hyears, a.group) if os.path.exists(tgt) and os.path.getsize(tgt) > 0: print(f"[{i}/{len(todo)}] skip {os.path.basename(tgt)}", flush=True) continue tmp = tgt + ".part" print(f"[{i}/{len(todo)}] {mmdd} -> {os.path.basename(tgt)}", flush=True) for attempt in range(1, 5): # retry-with-backoff; resumable across runs t0 = time.time() try: client.retrieve(DATASET, build_request(mmdd, hyears, a.group)).download(tmp) os.replace(tmp, tgt) print(f" OK {os.path.getsize(tgt)} bytes in {time.time()-t0:.0f}s", flush=True) n_ok += 1 break except Exception as e: if os.path.exists(tmp): os.remove(tmp) wait = 120 * attempt print(f" ERROR (try {attempt}/4) {type(e).__name__}: {e}", flush=True) if attempt < 4: print(f" backoff {wait}s ...", flush=True); time.sleep(wait) else: n_fail += 1 # give up this date; continue (re-run fills the gap) print(f"DONE group={a.group} ok={n_ok} failed={n_fail} (re-run to fill gaps)", flush=True) if __name__ == "__main__": main()