diff --git a/docs/source/config-configuration.md b/docs/source/config-configuration.md index f6f6b38d..37252152 100644 --- a/docs/source/config-configuration.md +++ b/docs/source/config-configuration.md @@ -291,13 +291,68 @@ PyPSA-USA supports two sources for renewable capacity-factor time series, select - `atlite` — runtime computation of capacity factors from ERA5 cutouts, weighted by Copernicus / CORINE land-use exclusions (the legacy default; see [Renewable Capacity Factors](renewable_cfs) under Model Data). - `godeeep` — pre-computed regional climate-model capacity factors from the [GODEEEP](https://www.pnnl.gov/projects/godeeep) (Grid Operations, Decarbonization, Environmental and Energy Equity Platform) dataset, weighted at runtime by NREL reV land-access exclusions. -The `godeeep` path consumes two file families published on Zenodo, downloaded automatically by `scripts/zenodo_downloader.py` on first run: +The `godeeep` path consumes two file families: -- **Compressed GODEEEP capacity factors** — per-cell hourly capacity factors on the GODEEEP Lambert Conformal grid, uint8-quantized and zlib-compressed (~12× smaller than the raw aggregated files). One Zenodo record per `(tech, scenario)`: - - solar: [historical](https://doi.org/10.5281/zenodo.20127513), [rcp45hotter](https://doi.org/10.5281/zenodo.20127523), [rcp45cooler](https://doi.org/10.5281/zenodo.20127562), [rcp85hotter](https://doi.org/10.5281/zenodo.20127589), [rcp85cooler](https://doi.org/10.5281/zenodo.20127633) - - wind (125 m): [historical](https://doi.org/10.5281/zenodo.20127520), [rcp45hotter](https://doi.org/10.5281/zenodo.20127545), [rcp45cooler](https://doi.org/10.5281/zenodo.20127572), [rcp85hotter](https://doi.org/10.5281/zenodo.20127604), [rcp85cooler](https://doi.org/10.5281/zenodo.20127645) +- **Compressed GODEEEP capacity factors** — per-cell hourly capacity factors on the GODEEEP Lambert Conformal grid, uint8-quantized and zlib-compressed (~12× smaller than the raw aggregated files; ~350 MB per solar year, ~800 MB per wind year). One file per `(tech, hub height, scenario, year)`, retrieved by `rule retrieve_godeeep_cf` from the sources declared in `godeeep_cf_registry` (see below). - **NREL land-access artifacts** ([10.5281/zenodo.20127899](https://doi.org/10.5281/zenodo.20127899)) — `avail_{tech}_{access}[_cec|_boem].nc` per-cell availability rasters and `caps_{tech}_{access}[_cec|_boem].nc` per-bus rollups (`weight`, `p_nom_max`, `potential`, `average_distance`, and `underwater_fraction` for offshore). +#### How capacity-factor files are retrieved + +Capacity-factor files are no longer downloaded from inside `build_renewable_profiles`. A dedicated +rule, `retrieve_godeeep_cf`, places each file, driven by the `godeeep_cf_registry` block of +`config.common.yaml`. The registry is an *ordered* list of sources — a local mirror and Zenodo — +and each source declares exactly which `(dataset key, year)` pairs it holds. The first source +declaring the requested pair wins. + +There are **no fallback paths**: no default hub height, no nearest-year substitution, no swap to an +unscreened variant. A request for a dataset or year that no source declares fails at snakemake +parse time with an error naming the dataset key, the requested year and the years available from +each source. Local hits are symlinked into the run's `data/` directory (set `copy_local: true` to +copy instead) and verified against the mirror's `SHA256SUMS` manifest before use. + +#### Which data is available + +Dataset keys are `{tech}{wind_height}_{scenario}_compressed`. On Sherlock all of the following is +reachable out of the box, because the shipped config declares the group's Oak mirror +(`/oak/stanford/groups/iazevedo/GoDEEEP_Capacity_Factors_compressed`, group `oak_iazevedo`) as the +first source. Off Sherlock, edit `godeeep_cf_registry` to remove or repoint the `local` source — +only the Zenodo subset is then reachable. + +| Dataset key | `godeeep_wind_height` | Oak mirror | Zenodo | +|---|---|---|---| +| `solar_historical_compressed` | — | 1980–2022 | 2012 only ([20127513](https://doi.org/10.5281/zenodo.20127513)) | +| `wind_100m_historical_compressed` | `"_100m"` | 1980–2022 | not published | +| `wind_125m_historical_compressed` | `"_125m"` | 1980–2022 | 2012 only ([20127520](https://doi.org/10.5281/zenodo.20127520)) | +| `solar_rcp45hotter_compressed` | — | — | 2030 / 2040 / 2050 ([20127523](https://doi.org/10.5281/zenodo.20127523)) | +| `solar_rcp45cooler_compressed` | — | — | 2030 / 2040 / 2050 ([20127562](https://doi.org/10.5281/zenodo.20127562)) | +| `solar_rcp85hotter_compressed` | — | — | 2030 / 2040 / 2050 ([20127589](https://doi.org/10.5281/zenodo.20127589)) | +| `solar_rcp85cooler_compressed` | — | — | 2030 / 2040 / 2050 ([20127633](https://doi.org/10.5281/zenodo.20127633)) | +| `wind_125m_rcp45hotter_compressed` | `"_125m"` | — | **not published** (record 20127545 returns HTTP 404) | +| `wind_125m_rcp45cooler_compressed` | `"_125m"` | — | 2030 / 2040 / 2050 ([20127572](https://doi.org/10.5281/zenodo.20127572)) | +| `wind_125m_rcp85hotter_compressed` | `"_125m"` | — | 2030 / 2040 / 2050 ([20127604](https://doi.org/10.5281/zenodo.20127604)) | +| `wind_125m_rcp85cooler_compressed` | `"_125m"` | — | 2030 / 2040 / 2050 ([20127645](https://doi.org/10.5281/zenodo.20127645)) | + +Notable gaps: + +- **Full historical coverage (1980–2022) requires the Oak mirror.** From Zenodo the historical + scenario is a single year, 2012. +- **100 m hub-height wind exists only on the Oak mirror.** Zenodo publishes 125 m only. +- **`rcp45hotter` is solar-only.** The corresponding 125 m wind record was never registered, so the + registry declares it with an empty year list; requesting it fails with `available years: (none)` + rather than silently downloading nothing. + +The Oak mirror also holds an 80 m wind archive (1980–2022, sha256-verified alongside the rest), but +it is not yet wired into `godeeep_cf_registry` and the workflow accepts only `"_100m"` and +`"_125m"` for `godeeep_wind_height`. (`"_80m"` was the silent default behind +[#803](https://github.com/PyPSA/pypsa-usa/issues/803) and is now rejected.) + +```{eval-rst} +.. literalinclude:: ../../workflow/repo_data/config/config.common.yaml + :language: yaml + :start-after: # docs : GODEEEP_CF_REGISTRY + :end-before: # docs : +``` + #### Configuring a godeeep run A complete godeeep configuration requires four config blocks beyond the standard `electricity:` / `clustering:` settings: @@ -309,7 +364,10 @@ A complete godeeep configuration requires four config blocks beyond the standard dataset: godeeep # set to atlite for the ERA5 + CORINE workflow ``` -2. **Scenario and year selection**. The GODEEEP dataset has one historical record (2012) and four future climate scenarios (`rcp45hotter`, `rcp45cooler`, `rcp85hotter`, `rcp85cooler`) at planning horizons 2030 / 2040 / 2050. +2. **Scenario and year selection**. The GODEEEP dataset has a historical scenario and four future + climate scenarios (`rcp45hotter`, `rcp45cooler`, `rcp85hotter`, `rcp85cooler`) at planning + horizons 2030 / 2040 / 2050. Historical years run 1980–2022 via the Oak mirror; from Zenodo only + 2012 is available. Any `(scenario, year)` outside the table above fails at parse time. ```yaml renewable_scenarios: ["rcp45cooler"] # one of: historical | rcp45hotter | rcp45cooler | rcp85hotter | rcp85cooler @@ -318,6 +376,13 @@ A complete godeeep configuration requires four config blocks beyond the standard Future-scenario years come from the `planning_horizons` wildcard (under `scenario:`); the historical year comes from `renewable_weather_years`. + Wind runs must also set the hub height, which has **no default** — leaving it unset fails the + run: + + ```yaml + godeeep_wind_height: "_125m" # "_100m" (Oak mirror only) | "_125m" (Oak mirror + Zenodo) + ``` + 3. **Snapshots** — the temporal slice within the chosen year. For godeeep this controls how much of the 8760-hour GODEEEP CF is sampled: ```yaml diff --git a/docs/source/data-generators.md b/docs/source/data-generators.md index 39e21626..77735383 100644 --- a/docs/source/data-generators.md +++ b/docs/source/data-generators.md @@ -38,13 +38,13 @@ exclusion assumptions. #### GODEEEP (default) -The default capacity-factor source (`renewable.dataset: godeeep`) is the [GODEEEP](https://www.pnnl.gov/projects/godeeep) dataset — regional-climate-model capacity factors developed at Pacific Northwest National Laboratory under the Grid Operations, Decarbonization, Environmental and Energy Equity Platform. Designed for multi-year climate-change scenario studies, GODEEEP provides hourly solar PV and 125 m hub-height wind capacity factors on a 12 km Lambert Conformal grid for: +The default capacity-factor source (`renewable.dataset: godeeep`) is the [GODEEEP](https://www.pnnl.gov/projects/godeeep) dataset — regional-climate-model capacity factors developed at Pacific Northwest National Laboratory under the Grid Operations, Decarbonization, Environmental and Energy Equity Platform. Designed for multi-year climate-change scenario studies, GODEEEP provides hourly solar PV and 100 m / 125 m hub-height wind capacity factors on a 12 km Lambert Conformal grid for: -- **One historical year** (2012) calibrated against observed weather. +- **A historical period** (1980–2022) calibrated against observed weather. - **Four future climate scenarios** — `rcp45hotter`, `rcp45cooler`, `rcp85hotter`, `rcp85cooler` — under the RCP4.5 and RCP8.5 emissions pathways, downscaled with two GCM ensemble members per pathway. - **Three planning horizons** (2030, 2040, 2050) per future scenario, drawn from contiguous 20-year (wind) or 40-year (solar) windows. -The raw GODEEEP files are large (~4.4 GB per `(tech, scenario, year)` triple). PyPSA-USA consumes a uint8-quantized + zlib-compressed variant (~350 MB for solar, ~800 MB for wind) published as 10 Zenodo records keyed by `(tech, scenario)`. The compressed files are pulled automatically at runtime by `scripts/zenodo_downloader.py`. +The raw GODEEEP files are large (~4.4 GB per `(tech, scenario, year)` triple). PyPSA-USA consumes a uint8-quantized + zlib-compressed variant (~350 MB for solar, ~800 MB for wind). Each compressed file is placed by `rule retrieve_godeeep_cf` from the sources declared in the `godeeep_cf_registry` config block — a local mirror (on Sherlock, the full 1980–2022 historical archive at both hub heights) and Zenodo records keyed by `(tech, scenario)`. Only part of the matrix is on Zenodo; see [`renewable: godeeep`](godeeep_cf) for exactly which `(dataset, year)` pairs each source holds. GODEEEP capacity factors are re-aggregated to PyPSA-USA bus polygons using a runtime weighting step: diff --git a/docs/source/model-workflow.md b/docs/source/model-workflow.md index d7c2c3a7..8daf5f90 100644 --- a/docs/source/model-workflow.md +++ b/docs/source/model-workflow.md @@ -155,6 +155,7 @@ on first use: | `retrieve_eer_demand_data` | EER forecasted demand profiles (when `demand: profile: eer`) | | `retrieve_gridemissions_data` | Historical grid emissions data | | `retrieve_nrel_exclusion_artifact` | NREL reV-based land-access exclusion layers | +| `retrieve_godeeep_cf` | One compressed GODEEEP capacity-factor file, from the local mirror or Zenodo record declared in `godeeep_cf_registry` (godeeep runs) | | `retrieve_res_eulp` / `retrieve_com_eulp` | NREL End-Use Load Profiles (sector runs) | | `retrieve_sector_databundle` | Sector-coupling data bundle (sector runs) | | `retrieve_seismic_risk_mask` | Seismic-risk exclusion mask (EGS runs) | diff --git a/workflow/repo_data/config/config.common.yaml b/workflow/repo_data/config/config.common.yaml index c0bfb917..0aaa8ec5 100644 --- a/workflow/repo_data/config/config.common.yaml +++ b/workflow/repo_data/config/config.common.yaml @@ -163,12 +163,87 @@ renewable: renewable_land_access: # REQUIRED for godeeep; one of: reference | limited | open apply_cec_basescreen: false # overlay CEC Wind/Solar BaseScreen on onwind+solar (CA-only) apply_boem_osw: false # overlay BOEM offshore-wind planning areas on offwind + offwind_floating -godeeep_wind_height: "_100m" # raw GODEEEP wind-CF hub height; "_100m" | "_125m" +godeeep_wind_height: "_125m" # raw GODEEEP wind-CF hub height; "_100m" | "_125m" — NO DEFAULT is applied if unset, the run fails loudly. + # "_125m" is the only height published on Zenodo; "_100m" REQUIRES the local (Oak) source below. + # "_80m" is not a GODEEEP product — there is no data at that height. nrel_caps_reassign: # out-of-footprint caps recovery for footprint-scoped runs (caps are rolled up nationally) enable: false # when true, fold unmapped caps entries onto the nearest in-footprint bus (needs x/y in the caps file) max_km: 100 # only reassign entries within this distance (km); farther entries stay dropped (prevents cross-seam teleporting) +# ==================================================================== +# GODEEEP CF REGISTRY — where compressed capacity-factor files come from +# ==================================================================== +# Declares every (dataset key, year) the workflow is allowed to retrieve. +# `godeeep_cf_registry.py` walks `sources` IN ORDER and the FIRST source +# declaring the requested (dataset key, year) wins. There are NO fallback +# paths: a missing dataset/year raises an error naming the dataset key, the +# requested year and the available years per source — it never substitutes a +# different year, hub height or screening variant (issue #803). +# +# Dataset keys are "{tech}{wind_height}_{scenario}_compressed", e.g. +# `solar_historical_compressed`, `wind_125m_rcp85cooler_compressed`. +# `years:` takes an inclusive range string ("1980-2022") or an explicit list. +# docs : GODEEEP_CF_REGISTRY +godeeep_cf_registry: + copy_local: false # false = the retrieve rule symlinks local hits (no duplication); true = copy into the run's data/ + sources: + # ---- 1. Local Oak mirror: full 1980-2022 historical archive, both hub heights. + - kind: local + root: /oak/stanford/groups/iazevedo/GoDEEEP_Capacity_Factors_compressed + layout: "{scenario}/{tech_dir}/{filename}" # placeholders: scenario, tech_dir (solar|wind_100m|wind_125m), technology, wind_height, year, filename, dataset_key + datasets: + solar_historical_compressed: + years: "1980-2022" + wind_100m_historical_compressed: + years: "1980-2022" + wind_125m_historical_compressed: + years: "1980-2022" + # ---- 2. Zenodo: the published subset. Historical is a single year (2012); + # future scenarios hold 2030/2040/2050 per (tech, scenario) record. + # Every year list below was read off the live Zenodo API file listing + # (GET https://zenodo.org/api/records/), not assumed. + - kind: zenodo + datasets: + solar_historical_compressed: + record: 20127513 + years: [2012] + wind_125m_historical_compressed: + record: 20127520 + years: [2012] + solar_rcp45hotter_compressed: + record: 20127523 + years: [2030, 2040, 2050] + solar_rcp45cooler_compressed: + record: 20127562 + years: [2030, 2040, 2050] + solar_rcp85hotter_compressed: + record: 20127589 + years: [2030, 2040, 2050] + solar_rcp85cooler_compressed: + record: 20127633 + years: [2030, 2040, 2050] + wind_125m_rcp45cooler_compressed: + record: 20127572 + years: [2030, 2040, 2050] + wind_125m_rcp85hotter_compressed: + record: 20127604 + years: [2030, 2040, 2050] + wind_125m_rcp85cooler_compressed: + record: 20127645 + years: [2030, 2040, 2050] + # wind_125m_rcp45hotter is NOT published. Record 20127545 (hardcoded in the + # old zenodo_downloader.py) returns HTTP 404 "The persistent identifier is + # not registered", and no such record appears in a Zenodo search of the + # series — the other 9 records of the series do. Declaring it with an empty + # year list keeps the failure honest: requesting it raises "available years: + # (none)" instead of the legacy silent `None` download path. Use the local + # source or re-publish the record to make this scenario runnable. + wind_125m_rcp45hotter_compressed: + record: 20127545 + years: [] + + # ==================================================================== # ATLITE — ERA5 cutout definitions # ==================================================================== diff --git a/workflow/repo_data/config/config.default.yaml b/workflow/repo_data/config/config.default.yaml index 74a5d984..a0e9bed2 100644 --- a/workflow/repo_data/config/config.default.yaml +++ b/workflow/repo_data/config/config.default.yaml @@ -41,6 +41,12 @@ run: # Sets which precomputed/atlite-derived CF time series the renewable # attach pipeline consumes. `godeeep` requires `renewable_land_access` # below; `atlite` requires a populated cutout (see config.common.yaml). +# +# `godeeep` also retrieves its CF files through the `godeeep_cf_registry` +# block in config.common.yaml: every (dataset, year) the run may fetch is +# declared there, sources are tried in order, and an undeclared +# dataset/year fails at parse time naming the dataset key, the requested +# year and the years each source offers. Nothing is substituted silently. # docs : RENEWABLE_DATASET renewable: dataset: godeeep # atlite | godeeep — see docs/source/config-configuration.md#renewable-godeeep @@ -95,11 +101,12 @@ enable: # ==================================================================== # For Atlite (any year) and GODEEEP-historical, configure renewable_weather_years. # For GODEEEP future scenarios, the year is taken from planning_horizons above. +# Under `godeeep` the first entry must be a year declared for the requested +# dataset in `godeeep_cf_registry` (config.common.yaml) — the Oak mirror covers +# 1980-2022 historical, Zenodo publishes only 2012. An undeclared year is a +# parse-time error, never a silent swap to a neighbouring year. # docs : RENEWABLE_WEATHER_YEARS renewable_weather_years: [2019] # weather years to use (single year unless multi-horizon godeeep) -# renewable_weather_years_by_horizon: # OPTIONAL: map planning-horizon → weather year (multi-horizon atlite) -# 2030: 2019 -# 2040: 2019 # docs : SNAPSHOTS snapshots: @@ -108,10 +115,12 @@ snapshots: inclusive: "left" # pandas.date_range inclusive arg; left | right | both | neither # GODEEEP future-scenario controls. Used only when renewable.dataset = godeeep -# and renewable_scenarios != "historical". +# and renewable_scenarios != "historical". Future-scenario CF years come from +# scenario.planning_horizons above; the (scenario, hub height, year) triple must +# be declared in `godeeep_cf_registry` (config.common.yaml). Note rcp45hotter is +# published for solar only — its wind_125m Zenodo record is unregistered. # docs : RENEWABLE_SCENARIOS renewable_scenarios: ["rcp85cooler"] # climate scenario; historical | rcp45cooler | rcp45hotter | rcp85cooler | rcp85hotter -# renewable_scenario_years: [2030] # OPTIONAL: explicit list of GODEEEP scenario years; defaults to planning_horizons # For multi-horizon godeeep, the YEAR is taken from each planning_horizon # automatically — these knobs only set the month/day slice within that year. diff --git a/workflow/rules/build_electricity.smk b/workflow/rules/build_electricity.smk index daf03dc6..566d3982 100644 --- a/workflow/rules/build_electricity.smk +++ b/workflow/rules/build_electricity.smk @@ -2,6 +2,17 @@ from itertools import chain +# rules/common.smk materializes godeeep_cf_registry.py into the source-cache +# dir it puts on sys.path, so the CF registry is importable at parse time — +# where its validation belongs. +from godeeep_cf_registry import ( + cf_filename, + godeeep_tech_spec, + resolve_scenario, + resolve_weather_year, + validate_godeeep_cf_config, +) + rule build_shapes: params: @@ -160,13 +171,44 @@ if config["enable"].get("build_cutout", False): "../scripts/build_cutout.py" -# Only use planning_horizon for GoDEEEP future scenarios +# Resolve every configured (scenario, year, technology) CF request now, so a +# registry that cannot serve the run dies here — with one aggregated message +# naming each dataset key, the requested year and the available years — instead +# of hundreds of DAG nodes later. A non-godeeep config is a no-op. +validate_godeeep_cf_config(config) + +# Only use planning_horizon for GoDEEEP future scenarios. `and` short-circuits, +# so an atlite run never needs renewable_scenarios, and a godeeep run missing it +# gets the registry's error rather than a bare KeyError. godeeep_planning_horizon = ( config.get("renewable", {}).get("dataset") == "godeeep" - and config["renewable_scenarios"][0] != "historical" + and resolve_scenario(config) != "historical" ) +def godeeep_cf_artifact(wildcards): + """Input function for the GODEEEP capacity-factor file of a technology. + + The file is retrieved by ``rule retrieve_godeeep_cf`` rather than downloaded + inside ``build_renewable_profiles``. Returns no input unless the CF dataset + is ``godeeep``. Future scenarios take the CF year from the + ``{planning_horizon}`` wildcard, historical ones from + ``renewable_weather_years[0]``; both the year and the file name come from + ``godeeep_cf_registry`` so the rule, the retrieval and the parse-time + validation cannot drift apart. + """ + if config.get("renewable", {}).get("dataset") != "godeeep": + return [] + spec = godeeep_tech_spec(wildcards.technology, config) + scenario = resolve_scenario(config) + year = resolve_weather_year( + config, + int(wildcards.planning_horizon) if godeeep_planning_horizon else None, + ) + filename = cf_filename(spec.technology, spec.wind_height, scenario, year) + return DATA + f"godeeep/{scenario}/{filename}" + + def nrel_exclusion_artifact(kind): """Input function for the NREL exclusion artifact of the given ``kind``. @@ -234,6 +276,7 @@ rule build_renewable_profiles: ), nrel_avail=nrel_exclusion_artifact("avail"), nrel_caps=nrel_exclusion_artifact("caps"), + godeeep_cf=godeeep_cf_artifact, cutout=lambda wildcards: ( expand( "cutouts/" @@ -270,10 +313,14 @@ rule build_renewable_profiles: ) threads: ATLITE_NPROCESSES resources: + # godeeep does no atlite excluder work: it streams the compressed CF in + # 500-hour chunks and aggregates onto bus polygons, whose peak footprint + # is set by the chunk and the cell->bus mapping, not by input.size. mem_mb=lambda wildcards, input, attempt: ( - ATLITE_NPROCESSES * input.size // 2000000 - ) - * 2.5, + 32000 * attempt + if config["renewable"]["dataset"] == "godeeep" + else (ATLITE_NPROCESSES * input.size // 2000000) * 2.5 + ), walltime=config_provider( "walltime", "build_renewable_profiles", default="02:30:00" ), diff --git a/workflow/rules/common.smk b/workflow/rules/common.smk index 93a3d982..b7c44039 100644 --- a/workflow/rules/common.smk +++ b/workflow/rules/common.smk @@ -11,6 +11,7 @@ path = workflow.source_path("../scripts/_helpers.py") # Materialize the sibling modules imported below (e.g. constants.py) into the # same source-cache dir, or those imports fail under a fresh cache. workflow.source_path("../scripts/constants.py") +workflow.source_path("../scripts/godeeep_cf_registry.py") sys.path.insert(0, os.path.dirname(path)) from _helpers import validate_checksum, update_config_from_wildcards diff --git a/workflow/rules/retrieve.smk b/workflow/rules/retrieve.smk index 38cda565..77ef1136 100644 --- a/workflow/rules/retrieve.smk +++ b/workflow/rules/retrieve.smk @@ -301,6 +301,39 @@ rule retrieve_nrel_exclusion_artifact: "../scripts/retrieve_nrel_exclusion.py" +# GODEEEP climate scenarios; the scenario is a directory/record component, never +# part of the CF file name (see godeeep_cf_registry.cf_filename). +GODEEEP_SCENARIOS = ( + "historical", + "rcp45cooler", + "rcp45hotter", + "rcp85cooler", + "rcp85hotter", +) + + +rule retrieve_godeeep_cf: + """ + Place one compressed GODEEEP capacity-factor file where + build_renewable_profiles expects it. The source (local mirror or Zenodo + record) is declared per (dataset key, year) in the `godeeep_cf_registry` + config block; an undeclared dataset/year raises instead of falling back to + another year, hub height or screening variant. + """ + wildcard_constraints: + scenario="|".join(GODEEEP_SCENARIOS), + cf_file=r"(solar|wind)_gen_cf_\d{4}(_\d+m)?_compressed", + output: + DATA + "godeeep/{scenario}/{cf_file}.nc", + log: + LOGS + "retrieve/godeeep_cf_{scenario}_{cf_file}.log", + resources: + walltime="01:00:00", + mem_mb=2000, + script: + "../scripts/retrieve_godeeep_cf.py" + + if "EGS" in config["electricity"]["extendable_carriers"]["Generator"]: rule retrieve_egs: diff --git a/workflow/scripts/build_renewable_profiles.py b/workflow/scripts/build_renewable_profiles.py index 6586813e..cf8faf21 100644 --- a/workflow/scripts/build_renewable_profiles.py +++ b/workflow/scripts/build_renewable_profiles.py @@ -16,9 +16,9 @@ import xarray as xr from _helpers import configure_logging, get_snapshots from dask.distributed import Client +from godeeep_cf_registry import resolve_scenario, resolve_weather_year from pypsa.geo import haversine from shapely.geometry import LineString -from zenodo_downloader import ZenodoScenarioDownloader logger = logging.getLogger(__name__) @@ -434,7 +434,7 @@ def plot_data(data): if dataset == "godeeep": logger.info("Loading godeeep renewable data...") - scenario = snakemake.config["renewable_scenarios"][0] + scenario = resolve_scenario(snakemake.config) tech = snakemake.wildcards.technology access = snakemake.config.get("renewable_land_access") if not access: @@ -443,32 +443,20 @@ def plot_data(data): "'open') when renewable.dataset == 'godeeep'.", ) - # Determine year based on scenario type - if scenario == "historical": - year = snakemake.config["renewable_weather_years"][0] - logger.info(f"Using historical year: {year} (from renewable_weather_years)") - else: - year = snakemake.params.planning_horizon - logger.info(f"Using future scenario year: {year} (from planning_horizon wildcard)") + # Historical scenarios take the CF year from renewable_weather_years, + # future ones from the {planning_horizon} wildcard. + year = resolve_weather_year(snakemake.config, snakemake.params.planning_horizon) + logger.info(f"Using GODEEEP CF year {year} for scenario '{scenario}'") # Get snapshots with appropriate year renewable_sns = get_renewable_snapshots(snakemake.config, year) - downloader = ZenodoScenarioDownloader() - - # Technology configurations for filename construction - if tech in ["onwind", "offwind", "offwind_floating"]: - technology = "wind" - wind_height = snakemake.config.get("godeeep_wind_height", "_100m") - elif tech == "solar": - technology = "solar" - wind_height = "" - else: - raise ValueError("Invalid technology type. Choose 'onwind', 'offwind', 'offwind_floating' or 'solar'.") # ===== NREL access-scenario path ===== - # Download per-cell compressed CF and apply availability-weighted - # aggregation onto bus polygons at runtime. Capacity variables come - # from the NREL supply-curve rollup (caps file). + # Aggregate the per-cell compressed CF onto bus polygons weighted by + # availability. The CF file is retrieved by rule retrieve_godeeep_cf — + # which dataset/year/hub height it holds is declared in + # config["godeeep_cf_registry"], so this script never picks a file. + # Capacity variables come from the NREL supply-curve rollup (caps file). from nrel_exclusion.aggregate_godeeep_weighted import ( fix_godeeep_time, get_cell_to_bus_mapping, @@ -476,11 +464,7 @@ def plot_data(data): ) logger.info(f"NREL access scenario: {access}") - cf_filename = f"{technology}_gen_cf_{year}{wind_height}_compressed.nc" - # Compressed-CF records on Zenodo are split by (tech, scenario), not by - # year-window — one record holds 2030/2040/2050 for the same scenario. - cf_record_key = f"{technology}{wind_height}_{scenario}_compressed" - cf_filepath = downloader.download_scenario_file(cf_record_key, scenario, cf_filename) + cf_filepath = snakemake.input.godeeep_cf ds_cf = xr.open_dataset(cf_filepath) ds_cf = fix_godeeep_time(ds_cf, year) diff --git a/workflow/scripts/godeeep_cf_registry.py b/workflow/scripts/godeeep_cf_registry.py new file mode 100644 index 00000000..1bf6d4e0 --- /dev/null +++ b/workflow/scripts/godeeep_cf_registry.py @@ -0,0 +1,692 @@ +"""Config-declared registry for GODEEEP capacity-factor (CF) files. + +The GODEEEP CF retrieval used to be spread across ``build_renewable_profiles.py`` +(filename/record-key construction, wind-height defaulting, year selection) and +``zenodo_downloader.py`` (a hardcoded ``record_id`` table that returned ``None`` +when a lookup missed). A miss therefore surfaced hundreds of lines later as a +``TypeError`` on a ``None`` path, or — worse — silently resolved to a different +year/hub-height than the one requested. + +This module replaces that with a single declarative registry read from +``config["godeeep_cf_registry"]``: + +* every source (local Oak mirror, Zenodo) declares which ``(dataset key, year)`` + pairs it actually holds, +* :func:`resolve_cf` walks the sources IN CONFIG ORDER and the first source that + holds the requested pair wins, +* an unresolvable request raises :class:`CfNotAvailableError` naming the dataset + key, the requested year and the available years per source. + +There are NO fallback paths: no default hub height, no "closest year", no +substitution of an unscreened file. Every failure is loud and names what is +missing. :func:`validate_godeeep_cf_config` runs the same resolution eagerly at +snakemake parse time so a bad config fails before any rule executes. + +The module is deliberately stdlib-only so it stays importable from the +``Snakefile`` at parse time (``workflow/rules/common.smk`` puts +``workflow/scripts`` on ``sys.path``). +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field, replace +from pathlib import Path + +__all__ = [ + "CfNotAvailableError", + "CfResolution", + "CfSource", + "TechSpec", + "cf_filename", + "dataset_key", + "godeeep_tech_spec", + "load_sources", + "parse_years", + "resolve_cf", + "resolve_scenario", + "resolve_weather_year", + "validate_godeeep_cf_config", +] + +#: Hub heights with published GODEEEP wind CFs. "_80m" is deliberately absent — +#: no such dataset exists, and defaulting to it silently produced wrong profiles. +VALID_WIND_HEIGHTS = ("_100m", "_125m") + +#: pypsa-usa technology wildcards backed by the GODEEEP wind CFs. +WIND_TECHNOLOGIES = ("onwind", "offwind", "offwind_floating") + +#: pypsa-usa technology wildcards backed by the GODEEEP solar CFs. +SOLAR_TECHNOLOGIES = ("solar",) + +#: All pypsa-usa technology wildcards this registry can resolve. +GODEEEP_TECHNOLOGIES = SOLAR_TECHNOLOGIES + WIND_TECHNOLOGIES + +#: Registry key in the merged snakemake config. +REGISTRY_KEY = "godeeep_cf_registry" + +_YEAR_RANGE_RE = re.compile(r"^\s*(\d{4})\s*-\s*(\d{4})\s*$") + +_KNOWN_SOURCE_KINDS = ("local", "zenodo") + + +class CfNotAvailableError(Exception): + """A requested GODEEEP CF dataset/year is not declared by any config source. + + Raised instead of returning ``None`` or falling back to a different year, + hub height or screening variant. The message always names the dataset key, + the requested year and what each source actually offers. + """ + + +@dataclass(frozen=True) +class TechSpec: + """The GODEEEP-side identity of a pypsa-usa technology wildcard.""" + + technology: str + """GODEEEP technology family: ``"solar"`` or ``"wind"``.""" + + wind_height: str + """Hub-height suffix (``"_100m"``/``"_125m"``); empty string for solar.""" + + def __iter__(self): + # Tuple-unpackable so call sites can keep the original + # ``technology, wind_height = ...`` shape from build_renewable_profiles. + return iter((self.technology, self.wind_height)) + + @property + def tech_dir(self) -> str: + """Directory component of the local mirror layout (e.g. ``wind_125m``).""" + return f"{self.technology}{self.wind_height}" + + +@dataclass(frozen=True) +class CfSource: + """One entry of ``config["godeeep_cf_registry"]["sources"]``. + + A source declares the ``(dataset key -> years)`` availability it can serve + plus the kind-specific coordinates needed to actually fetch a file: a + filesystem ``root``/``layout`` for ``kind == "local"``, a Zenodo record id + per dataset key for ``kind == "zenodo"``. + """ + + kind: str + """``"local"`` or ``"zenodo"``.""" + + years: Mapping[str, tuple[int, ...]] + """Dataset key -> the years this source holds.""" + + root: str | None = None + """Filesystem root of a ``local`` source.""" + + layout: str = "{scenario}/{tech_dir}/{filename}" + """Path template of a ``local`` source, relative to ``root``.""" + + records: Mapping[str, str] = field(default_factory=dict) + """Dataset key -> Zenodo record id, for a ``zenodo`` source.""" + + copy_local: bool = False + """Copy rather than symlink the retrieved file (``local`` sources only).""" + + name: str = "" + """Human-readable label used in error messages.""" + + def has(self, key: str, year: int) -> bool: + """Whether this source declares ``year`` for dataset ``key``.""" + return int(year) in self.years.get(key, ()) + + def years_for(self, key: str) -> tuple[int, ...]: + """The years this source declares for dataset ``key`` (may be empty).""" + return self.years.get(key, ()) + + @classmethod + def from_config(cls, raw: Mapping, index: int) -> CfSource: + """Build a source from its raw config mapping. + + Parameters + ---------- + raw + One entry of ``godeeep_cf_registry: sources:``. + index + Position in the source list, used for error messages. + + Raises + ------ + ValueError + If the entry is malformed (unknown kind, missing root, unparseable + years, zenodo dataset without a record id). + """ + where = f"godeeep_cf_registry.sources[{index}]" + if not isinstance(raw, Mapping): + raise ValueError(f"{where} must be a mapping, got {type(raw).__name__}.") + + kind = raw.get("kind") + if kind not in _KNOWN_SOURCE_KINDS: + raise ValueError( + f"{where}.kind is {kind!r}; expected one of {list(_KNOWN_SOURCE_KINDS)}.", + ) + + datasets = raw.get("datasets") or {} + if not isinstance(datasets, Mapping): + raise ValueError(f"{where}.datasets must be a mapping of dataset key -> years.") + + years: dict[str, tuple[int, ...]] = {} + records: dict[str, str] = {} + for key, spec in datasets.items(): + if isinstance(spec, Mapping): + if "years" not in spec: + raise ValueError(f"{where}.datasets.{key} is missing the 'years' key.") + raw_years = spec["years"] + record = spec.get("record", spec.get("record_id")) + else: + raw_years = spec + record = None + try: + years[key] = tuple(parse_years(raw_years)) + except ValueError as exc: + raise ValueError(f"{where}.datasets.{key}: {exc}") from exc + if kind == "zenodo": + if record is None: + raise ValueError( + f"{where}.datasets.{key} is a zenodo dataset but declares no 'record' id.", + ) + records[key] = str(record) + + root = raw.get("root") + if kind == "local": + if not root: + raise ValueError(f"{where}.root is required for a local source.") + root = str(root) + + return cls( + kind=kind, + years=years, + root=root, + layout=str(raw.get("layout") or "{scenario}/{tech_dir}/{filename}"), + records=records, + copy_local=bool(raw.get("copy_local", False)), + name=str(raw.get("name") or (root if kind == "local" else "zenodo")), + ) + + +@dataclass(frozen=True) +class CfResolution: + """A fully resolved GODEEEP CF request — where exactly the file comes from.""" + + kind: str + """Source kind that won: ``"local"`` or ``"zenodo"``.""" + + dataset_key: str + """Registry key, e.g. ``wind_125m_historical_compressed``.""" + + scenario: str + """Climate scenario, e.g. ``historical`` / ``rcp85cooler``.""" + + technology: str + """GODEEEP technology family: ``solar`` / ``wind``.""" + + wind_height: str + """Hub-height suffix; empty for solar.""" + + year: int + """The requested (and confirmed available) year.""" + + filename: str + """Published file name inside the source.""" + + path: str | None = None + """Absolute path of the file for a ``local`` source, else ``None``.""" + + record_id: str | None = None + """Zenodo record id for a ``zenodo`` source, else ``None``.""" + + copy_local: bool = False + """Retrieve rule should copy rather than symlink a ``local`` hit.""" + + source_index: int = 0 + """Position of the winning source in the configured list.""" + + @property + def location(self) -> str: + """Source-specific location string (local path or ``zenodo:``).""" + return self.path if self.kind == "local" else f"zenodo:{self.record_id}" + + +def parse_years(spec) -> list[int]: + """Parse a year availability spec into an explicit, sorted list of years. + + Accepts an inclusive range string (``"1980-2022"``) or an explicit sequence + of years (``[2012]``, ``["2030", 2040]``). Anything else raises so a typo in + the config cannot degrade into an empty — silently unavailable — year set. + + Raises + ------ + ValueError + If ``spec`` is not a range string or a sequence of four-digit years. + """ + if isinstance(spec, bool): + raise ValueError(f"invalid year spec {spec!r}; expected 'YYYY-YYYY' or a list of years.") + if isinstance(spec, int): + return [spec] + if isinstance(spec, str): + match = _YEAR_RANGE_RE.match(spec) + if not match: + raise ValueError( + f"invalid year range {spec!r}; expected an inclusive range like '1980-2022'.", + ) + start, end = int(match.group(1)), int(match.group(2)) + if end < start: + raise ValueError(f"invalid year range {spec!r}; end year {end} precedes start year {start}.") + return list(range(start, end + 1)) + if isinstance(spec, Sequence) and not isinstance(spec, str | bytes): + years: list[int] = [] + for item in spec: + if isinstance(item, bool) or not isinstance(item, int | str): + raise ValueError(f"invalid year {item!r} in {spec!r}; expected an integer year.") + try: + years.append(int(item)) + except (TypeError, ValueError) as exc: + raise ValueError(f"invalid year {item!r} in {spec!r}; expected an integer year.") from exc + return sorted(set(years)) + raise ValueError(f"invalid year spec {spec!r}; expected 'YYYY-YYYY' or a list of years.") + + +def dataset_key(technology: str, wind_height: str, scenario: str) -> str: + """Registry key for a ``(technology, hub height, scenario)`` triple. + + Reproduces the ``cf_record_key`` string ``build_renewable_profiles.py`` used + to build inline, which is also the key space of the legacy + ``ZenodoScenarioDownloader.scenario_records`` table — e.g. + ``solar_historical_compressed``, ``wind_125m_rcp85cooler_compressed``. + Compressed CF records are split by ``(tech, scenario)``, not by year: one + record holds 2030/2040/2050 for the same scenario. + """ + return f"{technology}{wind_height}_{scenario}_compressed" + + +def cf_filename(technology: str, wind_height: str, scenario: str, year: int) -> str: + """Published file name of a compressed GODEEEP CF file. + + Identical for every scenario — the scenario lives in the containing record / + directory, never in the file name. Examples: + ``solar_gen_cf_2019_compressed.nc``, + ``wind_gen_cf_2019_125m_compressed.nc``. + + ``scenario`` is accepted (and unused) so callers can pass the full request + tuple and so a future scenario-dependent naming change has one place to go. + """ + del scenario # part of the request tuple, but not of the published name + return f"{technology}_gen_cf_{year}{wind_height}_compressed.nc" + + +def godeeep_tech_spec(technology: str, config: Mapping) -> TechSpec: + """Map a pypsa-usa technology wildcard onto its GODEEEP identity. + + Unlike the code this replaces there is NO ``"_100m"`` default: an absent or + invalid ``godeeep_wind_height`` raises rather than quietly picking a hub + height the configured source may not even hold (issue #803). + + Raises + ------ + ValueError + If ``technology`` is not a GODEEEP-backed carrier, or if + ``godeeep_wind_height`` is missing/invalid for a wind carrier. + """ + if technology in SOLAR_TECHNOLOGIES: + return TechSpec("solar", "") + if technology not in WIND_TECHNOLOGIES: + raise ValueError( + f"Invalid technology {technology!r} for the GODEEEP CF registry; " + f"choose one of {list(GODEEEP_TECHNOLOGIES)}.", + ) + + wind_height = config.get("godeeep_wind_height") + if wind_height is None: + raise ValueError( + f"godeeep_wind_height is not set, but technology {technology!r} needs a GODEEEP wind hub " + f"height. Set it to one of {list(VALID_WIND_HEIGHTS)} (there is no default: '_100m' data " + "is only published on the local Oak mirror, '_125m' on both Oak and Zenodo).", + ) + wind_height = str(wind_height) + if wind_height not in VALID_WIND_HEIGHTS: + raise ValueError( + f"godeeep_wind_height {wind_height!r} is not a published GODEEEP hub height; " + f"valid values are {list(VALID_WIND_HEIGHTS)} (note '_80m' has no data).", + ) + return TechSpec("wind", wind_height) + + +def resolve_scenario(config: Mapping) -> str: + """First entry of ``renewable_scenarios``, with a clear error when unusable. + + Replaces the bare ``config["renewable_scenarios"][0]`` indexing in + ``build_renewable_profiles.py`` and ``build_electricity.smk``, which raised + an unattributed ``KeyError``/``IndexError`` at parse time. + + Raises + ------ + ValueError + If ``renewable_scenarios`` is missing, not a list, or empty. + """ + if "renewable_scenarios" not in config: + raise ValueError( + "renewable_scenarios is not set. It is REQUIRED when renewable.dataset == 'godeeep'; " + "set it to a one-element list, e.g. renewable_scenarios: ['historical'].", + ) + scenarios = config["renewable_scenarios"] + if isinstance(scenarios, str): + raise ValueError( + f"renewable_scenarios must be a list, got the string {scenarios!r}; " + f"write renewable_scenarios: [{scenarios!r}].", + ) + if not isinstance(scenarios, Sequence) or not scenarios: + raise ValueError( + f"renewable_scenarios must be a non-empty list of scenario names, got {scenarios!r}.", + ) + return str(scenarios[0]) + + +def resolve_weather_year(config: Mapping, planning_horizon=None) -> int: + """The CF year implied by the configured scenario. + + Historical runs take the year from ``renewable_weather_years``; future + scenarios take it from the ``{planning_horizon}`` wildcard, exactly as + ``build_renewable_profiles.py`` did. + + Raises + ------ + ValueError + If the scenario is historical and ``renewable_weather_years`` is + missing/empty/non-integer, or if a future scenario is requested without + a ``planning_horizon``. + """ + scenario = resolve_scenario(config) + if scenario == "historical": + if "renewable_weather_years" not in config: + raise ValueError( + "renewable_weather_years is not set, but renewable_scenarios == ['historical'] takes " + "the GODEEEP CF year from it; set e.g. renewable_weather_years: [2019].", + ) + years = config["renewable_weather_years"] + if not isinstance(years, Sequence) or isinstance(years, str) or not years: + raise ValueError( + f"renewable_weather_years must be a non-empty list of years, got {years!r}.", + ) + try: + return int(years[0]) + except (TypeError, ValueError) as exc: + raise ValueError( + f"renewable_weather_years[0] must be an integer year, got {years[0]!r}.", + ) from exc + if planning_horizon is None: + raise ValueError( + f"renewable_scenarios == [{scenario!r}] is a future GODEEEP scenario, so the CF year comes " + "from the {planning_horizon} wildcard — but none was supplied. Either pass a planning " + "horizon or set renewable_scenarios: ['historical'].", + ) + try: + return int(planning_horizon) + except (TypeError, ValueError) as exc: + raise ValueError(f"planning_horizon must be an integer year, got {planning_horizon!r}.") from exc + + +def load_sources(config: Mapping) -> list[CfSource]: + """Parse ``config["godeeep_cf_registry"]["sources"]`` in declaration order. + + Raises + ------ + ValueError + If the registry block is missing or any source entry is malformed. + """ + registry = config.get(REGISTRY_KEY) + if registry is None: + raise ValueError( + f"{REGISTRY_KEY} is not configured. GODEEEP CF retrieval is declared in config: add a " + f"{REGISTRY_KEY}: sources: list (see workflow/repo_data/config/config.common.yaml).", + ) + if not isinstance(registry, Mapping): + raise ValueError(f"{REGISTRY_KEY} must be a mapping with a 'sources' list, got {type(registry).__name__}.") + + raw_sources = registry.get("sources") + if not isinstance(raw_sources, Sequence) or isinstance(raw_sources, str | bytes) or not raw_sources: + raise ValueError( + f"{REGISTRY_KEY}.sources must be a non-empty ordered list of sources, got {raw_sources!r}.", + ) + + default_copy_local = bool(registry.get("copy_local", False)) + sources: list[CfSource] = [] + for index, raw in enumerate(raw_sources): + source = CfSource.from_config(raw, index) + if isinstance(raw, Mapping) and "copy_local" not in raw: + source = replace(source, copy_local=default_copy_local) + sources.append(source) + return sources + + +def _availability_report(sources: Sequence[CfSource], key: str) -> str: + lines = [] + for source in sources: + if key not in source.years: + available = "(dataset key not declared)" + elif not source.years[key]: + # Declared on purpose with no years, e.g. a Zenodo record id that is + # not (yet) published — say so rather than pretending it is unknown. + available = "(declared, but no years published)" + else: + available = _format_years(source.years[key]) + lines.append(f" - [{source.kind}] {source.name}: {available}") + return "\n".join(lines) if lines else " (no sources configured)" + + +def _format_years(years: Sequence[int]) -> str: + """Render a year tuple compactly, collapsing contiguous runs to a range.""" + ordered = sorted(set(int(y) for y in years)) + if not ordered: + return "(none)" + if len(ordered) > 2 and ordered == list(range(ordered[0], ordered[-1] + 1)): + return f"{ordered[0]}-{ordered[-1]} (inclusive)" + return ", ".join(str(y) for y in ordered) + + +def resolve_cf(config: Mapping, technology: str, planning_horizon=None) -> CfResolution: + """Resolve a GODEEEP CF request to exactly one source, or raise. + + Walks ``godeeep_cf_registry.sources`` in configured order and returns the + first source declaring the requested ``(dataset key, year)``. There is no + fallback after resolution: the winning source must serve that exact file. + + Parameters + ---------- + config + Merged snakemake config. + technology + pypsa-usa technology wildcard (``solar``, ``onwind``, ``offwind``, + ``offwind_floating``). + planning_horizon + Planning-horizon wildcard; used as the CF year for future scenarios and + ignored for ``historical``. + + Returns + ------- + CfResolution + Source kind, location, dataset key, year and file name. + + Raises + ------ + ValueError + On a malformed config (bad wind height, unknown technology, malformed + registry, missing scenario/weather-year keys). + CfNotAvailableError + If no configured source declares the requested dataset key and year. + """ + spec = godeeep_tech_spec(technology, config) + scenario = resolve_scenario(config) + year = resolve_weather_year(config, planning_horizon) + key = dataset_key(spec.technology, spec.wind_height, scenario) + filename = cf_filename(spec.technology, spec.wind_height, scenario, year) + sources = load_sources(config) + + for index, source in enumerate(sources): + if not source.has(key, year): + continue + if source.kind == "local": + relative = source.layout.format( + scenario=scenario, + tech_dir=spec.tech_dir, + technology=spec.technology, + wind_height=spec.wind_height, + year=year, + filename=filename, + dataset_key=key, + ) + path = str(Path(source.root) / relative) + return CfResolution( + kind="local", + dataset_key=key, + scenario=scenario, + technology=spec.technology, + wind_height=spec.wind_height, + year=year, + filename=filename, + path=path, + copy_local=source.copy_local, + source_index=index, + ) + record_id = source.records.get(key) + if record_id is None: + raise ValueError( + f"godeeep_cf_registry.sources[{index}] declares years for {key!r} but no zenodo record id.", + ) + return CfResolution( + kind="zenodo", + dataset_key=key, + scenario=scenario, + technology=spec.technology, + wind_height=spec.wind_height, + year=year, + filename=filename, + record_id=str(record_id), + source_index=index, + ) + + raise CfNotAvailableError( + f"No configured GODEEEP CF source provides dataset {key!r} for year {year}.\n" + f" requested: technology={technology!r} scenario={scenario!r} " + f"hub_height={spec.wind_height or 'n/a'!r} file={filename!r}\n" + f" available years per source:\n{_availability_report(sources, key)}\n" + f" Fix: request a year listed above, add the dataset/year to " + f"{REGISTRY_KEY}.sources, or point a local source at a mirror that holds it.", + ) + + +def _technologies_to_validate(config: Mapping) -> list[str]: + """The GODEEEP-backed carriers present in the config (deduped by dataset).""" + renewable = config.get("renewable") or {} + if isinstance(renewable, Mapping): + techs = [tech for tech in GODEEEP_TECHNOLOGIES if tech in renewable] + if techs: + return techs + return ["solar", "onwind"] + + +def _years_to_validate(config: Mapping, scenario: str, problems: list[str]) -> list[int]: + if scenario == "historical": + years = config.get("renewable_weather_years") + if not years or isinstance(years, str) or not isinstance(years, Sequence): + problems.append( + "renewable_weather_years must be a non-empty list of years when " + f"renewable_scenarios == ['historical'] (got {years!r}).", + ) + return [] + return [years[0]] + horizons = (config.get("scenario") or {}).get("planning_horizons") + if not horizons or isinstance(horizons, str) or not isinstance(horizons, Sequence): + problems.append( + f"scenario.planning_horizons must be a non-empty list when renewable_scenarios == [{scenario!r}]: " + "future GODEEEP scenarios take the CF year from the planning horizon " + f"(got {horizons!r}).", + ) + return [] + return list(horizons) + + +def validate_godeeep_cf_config(config: Mapping) -> list[CfResolution]: + """Fail fast, and all at once, on a broken GODEEEP CF configuration. + + Intended to run at snakemake parse time. Every configured + ``scenario x year x technology`` combination is resolved eagerly; ALL + problems (missing keys, invalid hub height, malformed sources, unresolvable + combinations) are collected and reported in a single error rather than one + per run. + + A config whose ``renewable.dataset`` is not ``godeeep`` is skipped. + + Returns + ------- + list of CfResolution + Every combination that resolved, for logging/debugging. + + Raises + ------ + CfNotAvailableError + If any check failed; the message enumerates all failures. + """ + renewable = config.get("renewable") + if isinstance(renewable, Mapping) and renewable.get("dataset", "godeeep") != "godeeep": + return [] + + problems: list[str] = [] + resolved: list[CfResolution] = [] + + try: + sources = load_sources(config) + except ValueError as exc: + sources = [] + problems.append(str(exc)) + + try: + scenario = resolve_scenario(config) + except ValueError as exc: + problems.append(str(exc)) + scenario = None + + technologies = _technologies_to_validate(config) + specs: dict[str, TechSpec] = {} + for technology in technologies: + try: + specs[technology] = godeeep_tech_spec(technology, config) + except ValueError as exc: + problems.append(str(exc)) + + if scenario is not None and sources and specs: + years = _years_to_validate(config, scenario, problems) + seen: set[tuple[str, int]] = set() + for year in years: + try: + year = int(year) + except (TypeError, ValueError): + problems.append(f"CF year {year!r} for scenario {scenario!r} is not an integer.") + continue + for technology, spec in specs.items(): + key = dataset_key(spec.technology, spec.wind_height, scenario) + if (key, year) in seen: + continue + seen.add((key, year)) + horizon = None if scenario == "historical" else year + try: + resolved.append(resolve_cf(config, technology, horizon)) + except (CfNotAvailableError, ValueError) as exc: + problems.append(str(exc)) + + if problems: + # The three wind carriers share one TechSpec, so a bad hub height (etc.) + # produces the same message once per carrier — report each problem once. + problems = list(dict.fromkeys(problems)) + bullets = "\n\n".join(f" ({i + 1}) {problem}" for i, problem in enumerate(problems)) + raise CfNotAvailableError( + f"Invalid GODEEEP capacity-factor configuration — {len(problems)} problem(s):\n\n{bullets}\n\n" + f"All GODEEEP CF retrieval is declared under '{REGISTRY_KEY}' in the config; there are no " + "fallback paths (no default hub height, no nearest-year substitution).", + ) + return resolved diff --git a/workflow/scripts/nrel_exclusion/compress_godeeep_array.sbatch b/workflow/scripts/nrel_exclusion/compress_godeeep_array.sbatch new file mode 100644 index 00000000..e8e28a47 --- /dev/null +++ b/workflow/scripts/nrel_exclusion/compress_godeeep_array.sbatch @@ -0,0 +1,55 @@ +#!/bin/bash +#SBATCH -A iazevedo +#SBATCH -p serc +#SBATCH --job-name=godeeep-compress +#SBATCH --array=1-129%16 +#SBATCH --time=02:00:00 +#SBATCH --cpus-per-task=2 +#SBATCH --mem=12G +#SBATCH -o logs/run_slurm/godeeep-compress-%A_%a.out +#SBATCH -e logs/run_slurm/godeeep-compress-%A_%a.err + +# Sweep the GODEEEP capacity-factor compression: one array task per tasks.tsv row. +# +# The default array range (1-129) matches the full registry: 3 techs +# (solar, wind_125m, wind_100m) x 43 years (1980-2022). Regenerate the task list +# first and use the range it prints: +# +# cd workflow +# mkdir -p logs/run_slurm +# python scripts/nrel_exclusion/make_compress_tasks.py \ +# --output /oak/stanford/groups/iazevedo/GoDEEEP_Capacity_Factors_compressed/tasks.tsv +# sbatch scripts/nrel_exclusion/compress_godeeep_array.sbatch +# +# Tasks are idempotent (sha256 sidecar skip), so re-submitting the whole array +# after a partial sweep only rebuilds what is missing or corrupt. Extra flags are +# passed through to compress_godeeep_task.sh: +# +# sbatch scripts/nrel_exclusion/compress_godeeep_array.sbatch --dest-root /path/to/registry +# +# Submit from workflow/ (log paths are relative to $SLURM_SUBMIT_DIR) or set +# WORKFLOW_DIR. Not part of the Snakemake DAG — the registry is built once, out of +# band, then consumed read-only. + +set -euo pipefail +umask 0002 + +WORKFLOW_DIR="${WORKFLOW_DIR:-${SLURM_SUBMIT_DIR:-$PWD}}" +cd "$WORKFLOW_DIR" + +CONDA_SH="${CONDA_SH:-/home/groups/iazevedo/miniforge3/etc/profile.d/conda.sh}" +CONDA_ENV="${CONDA_ENV:-pypsa-usa}" +set +u # conda's activation hooks trip over `set -u` +# shellcheck disable=SC1090 +source "$CONDA_SH" +conda activate "$CONDA_ENV" +set -u + +# Keep BLAS/HDF5 from oversubscribing the shared node. +export OPENBLAS_NUM_THREADS=2 +export OMP_NUM_THREADS=2 +export MKL_NUM_THREADS=2 + +bash scripts/nrel_exclusion/compress_godeeep_task.sh \ + "${SLURM_ARRAY_TASK_ID:?run under sbatch --array, or pass an index to compress_godeeep_task.sh directly}" \ + "$@" diff --git a/workflow/scripts/nrel_exclusion/compress_godeeep_task.sh b/workflow/scripts/nrel_exclusion/compress_godeeep_task.sh new file mode 100755 index 00000000..9fa824f7 --- /dev/null +++ b/workflow/scripts/nrel_exclusion/compress_godeeep_task.sh @@ -0,0 +1,182 @@ +#!/bin/bash +# Compress one GODEEEP capacity-factor year: stage -> compress -> sanity -> publish. +# +# This is the per-task driver behind compress_godeeep_array.sbatch and the script +# that produced the Oak registry at +# /oak/stanford/groups/iazevedo/GoDEEEP_Capacity_Factors_compressed. +# +# The task index is the 1-based line number in tasks.tsv (see make_compress_tasks.py) +# and doubles as $SLURM_ARRAY_TASK_ID: +# +# idx tech year zip member member_bytes dest_filename +# +# Flow, per task: +# 1. idempotent skip — destination + SHA256SUMS.d sidecar agree -> exit 0 +# 2. stage — member out of the zip64 archive onto node-local disk +# ($L_SCRATCH when set, else $SCRATCH), removed by an EXIT trap +# 3. compress — compress_godeeep.py (uint8, scale_factor 1/254, zlib 4) +# 4. sanity — dims, uint8 dtype, NaN mask, 12-hour max|err| <= 1/254 vs raw +# 5. publish — copy to .part, chmod 0664, re-hash, atomic rename, +# then write the sidecar +# +# Nothing here is wired into the Snakemake DAG: the registry is built once, out of +# band, and consumed read-only by the workflow. +# +# Usage: +# bash compress_godeeep_task.sh 42 +# bash compress_godeeep_task.sh --index 42 --dest-root /path/to/registry --tasks /path/to/tasks.tsv +# +# Environment overrides: DEST_ROOT, TASKS_TSV, PYTHON, STAGE_ROOT, +# EXPECT_TIME / EXPECT_SOUTH_NORTH / EXPECT_WEST_EAST. + +set -euo pipefail +umask 0002 + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +DEST_ROOT="${DEST_ROOT:-/oak/stanford/groups/iazevedo/GoDEEEP_Capacity_Factors_compressed}" +TASKS_TSV="${TASKS_TSV:-}" +PYTHON="${PYTHON:-python}" +# Raw GODEEEP grid: 8760 hours on the 299x424 Lambert Conformal CONUS domain. +EXPECT_TIME="${EXPECT_TIME:-8760}" +EXPECT_SOUTH_NORTH="${EXPECT_SOUTH_NORTH:-299}" +EXPECT_WEST_EAST="${EXPECT_WEST_EAST:-424}" + +TASK_IDX="" + +die() { + echo "[task] ERROR: $*" >&2 + exit 1 +} + +usage() { + sed -n '2,/^$/p' "${BASH_SOURCE[0]}" | sed 's/^# \?//' +} + +while [ $# -gt 0 ]; do + case "$1" in + --index) TASK_IDX="${2:?--index needs a value}"; shift 2 ;; + --tasks) TASKS_TSV="${2:?--tasks needs a value}"; shift 2 ;; + --dest-root) DEST_ROOT="${2:?--dest-root needs a value}"; shift 2 ;; + -h|--help) usage; exit 0 ;; + -*) die "unknown flag $1" ;; + *) TASK_IDX="$1"; shift ;; + esac +done + +: "${TASKS_TSV:=$DEST_ROOT/tasks.tsv}" +[ -n "$TASK_IDX" ] || die "no task index given (positional or --index); see --help" +[ -f "$TASKS_TSV" ] || die "task list not found: $TASKS_TSV (generate it with make_compress_tasks.py)" + +# ---- resolve the task ------------------------------------------------------ +line="$(awk -v n="$TASK_IDX" 'NR == n' "$TASKS_TSV")" +[ -n "$line" ] || die "task index $TASK_IDX is out of range for $TASKS_TSV ($(wc -l < "$TASKS_TSV") tasks)" + +IFS=$'\t' read -r row_idx tech year zip_path member member_bytes dest_name <<<"$line" +[ "$row_idx" = "$TASK_IDX" ] || die "$TASKS_TSV line $TASK_IDX carries idx $row_idx — task list is malformed" + +rel="historical/$tech/$dest_name" +dest="$DEST_ROOT/$rel" +sidecar_dir="$DEST_ROOT/SHA256SUMS.d" +sidecar="$sidecar_dir/$dest_name.sha256" + +echo "[task $TASK_IDX] $tech $year -> $rel" + +# ---- 1. idempotent skip ---------------------------------------------------- +if [ -f "$dest" ] && [ -f "$sidecar" ]; then + want="$(awk '{print $1; exit}' "$sidecar")" + have="$(sha256sum "$dest" | awk '{print $1}')" + if [ "$want" = "$have" ]; then + echo "[task $TASK_IDX] SKIP $rel (sha256 verified against sidecar)" + exit 0 + fi + echo "[task $TASK_IDX] sidecar mismatch ($have != $want) — rebuilding" +fi + +# ---- 2. stage -------------------------------------------------------------- +if [ -z "${STAGE_ROOT:-}" ]; then + if [ -n "${L_SCRATCH:-}" ]; then + STAGE_ROOT="$L_SCRATCH" + else + STAGE_ROOT="${SCRATCH:?neither L_SCRATCH nor SCRATCH is set; pass STAGE_ROOT explicitly}" + fi +fi +tmpd="$STAGE_ROOT/godeeep_compress/task_$TASK_IDX" +rm -rf "$tmpd" +mkdir -p "$tmpd" +trap 'rm -rf "$tmpd"; rm -f "$dest.part"' EXIT + +raw="$tmpd/$(basename "$member")" +comp="$tmpd/$dest_name" + +t0=$SECONDS +"$PYTHON" "$SCRIPT_DIR/stage_godeeep_member.py" \ + --zip "$zip_path" --member "$member" --output "$raw" --expect-bytes "$member_bytes" +t_stage=$((SECONDS - t0)) + +# ---- 3. compress ----------------------------------------------------------- +t0=$SECONDS +"$PYTHON" "$SCRIPT_DIR/compress_godeeep.py" --input "$raw" --output "$comp" +t_comp=$((SECONDS - t0)) + +# ---- 4. sanity: dims, dtype, NaN mask, 12-hour round-trip vs the raw file --- +"$PYTHON" - "$raw" "$comp" "$EXPECT_TIME" "$EXPECT_SOUTH_NORTH" "$EXPECT_WEST_EAST" <<'PY' +import sys + +import netCDF4 +import numpy as np + +raw_path, comp_path = sys.argv[1], sys.argv[2] +expected = dict(zip(("Time", "south_north", "west_east"), (int(a) for a in sys.argv[3:6]))) +tol = 1.0 / 254.0 + + +def fail(msg): + raise SystemExit(f"[sanity] FAIL {msg}") + + +with netCDF4.Dataset(comp_path) as dc, netCDF4.Dataset(raw_path) as dr: + dims = {name: len(dim) for name, dim in dc.dimensions.items()} + for name, size in expected.items(): + if dims.get(name) != size: + fail(f"dimension {name}={dims.get(name)}, expected {size} (got {dims})") + + cf = dc.variables["capacity_factor"] + if cf.dtype != np.uint8: + fail(f"capacity_factor dtype {cf.dtype}, expected uint8") + + raw_cf = dr.variables["capacity_factor"] + raw_cf.set_auto_mask(False) # raw carries NaN in-band; keep it as float, not masked + for hour in np.linspace(0, expected["Time"] - 1, 12).astype(int): + decoded = cf[int(hour)] # auto-scaled to float, 255 -> masked + decoded = np.ma.filled(decoded, np.nan) if np.ma.isMaskedArray(decoded) else np.asarray(decoded) + decoded = decoded.astype(np.float64) + reference = np.asarray(raw_cf[int(hour)], dtype=np.float64) + + nan_decoded, nan_reference = np.isnan(decoded), np.isnan(reference) + if not np.array_equal(nan_decoded, nan_reference): + fail(f"NaN mask differs at hour {hour}: {int(nan_decoded.sum())} vs {int(nan_reference.sum())}") + + valid = ~nan_reference + if valid.any(): + err = float(np.abs(decoded[valid] - reference[valid]).max()) + if err > tol + 1e-9: + fail(f"round-trip error {err:.6g} > {tol:.6g} at hour {hour}") + +print("[sanity] OK dims/dtype/NaN-mask/round-trip", flush=True) +PY + +# ---- 5. publish (atomic) --------------------------------------------------- +digest="$(sha256sum "$comp" | awk '{print $1}')" +mkdir -p "$(dirname "$dest")" "$sidecar_dir" + +cp "$comp" "$dest.part" +chmod 0664 "$dest.part" +copied="$(sha256sum "$dest.part" | awk '{print $1}')" +[ "$copied" = "$digest" ] || die "destination hash mismatch after copy ($copied != $digest)" +mv -f "$dest.part" "$dest" + +printf '%s %s\n' "$digest" "$rel" > "$sidecar" +chmod 0664 "$sidecar" + +echo "[task $TASK_IDX] OK $rel stage=${t_stage}s comp=${t_comp}s size=$(stat -c %s "$dest") sha256=$digest" diff --git a/workflow/scripts/nrel_exclusion/make_compress_tasks.py b/workflow/scripts/nrel_exclusion/make_compress_tasks.py new file mode 100644 index 00000000..26f4833f --- /dev/null +++ b/workflow/scripts/nrel_exclusion/make_compress_tasks.py @@ -0,0 +1,175 @@ +""" +Enumerate the raw GODEEEP capacity-factor archives and emit the compression task list. + +The raw archives ship one NetCDF per (tech, year) inside a small number of zip64 +files. This script reads only their *central directories* (no decompression, no +staging), so it is cheap enough to run on a login node, and writes a tab-separated +task list consumed by ``compress_godeeep_task.sh``: + + idx tech year zip member member_bytes dest_filename + +``idx`` is the 1-based line number and doubles as the Slurm array task ID. Rows are +grouped by tech in the order given on the command line and, within a tech, ordered +by descending year (recent years first, so a partial sweep still yields the years +most runs care about). + +Coverage is asserted, not repaired: every requested year must resolve to exactly one +member of exactly one archive, otherwise the script fails naming the tech and the +missing years. There is no partial task list. + +Usage: + python make_compress_tasks.py --output tasks.tsv + python make_compress_tasks.py --techs solar --first-year 2000 --output solar.tsv +""" + +import argparse +import re +import sys +import zipfile +from dataclasses import astuple, dataclass +from pathlib import Path + +DEFAULT_ZIP_DIR = Path("/oak/stanford/groups/iazevedo/GoDEEEP_Capacity_Factors") +DEFAULT_TECHS = ("solar", "wind_125m", "wind_100m") +DEFAULT_FIRST_YEAR = 1980 +DEFAULT_LAST_YEAR = 2022 + + +@dataclass(frozen=True) +class Task: + """One (tech, year) compression job: where the raw member lives, what to call the output.""" + + idx: int + tech: str + year: int + zip_path: str + member: str + member_bytes: int + dest_filename: str + + def as_tsv(self) -> str: + """Render the task as one tab-separated ``tasks.tsv`` line (no trailing newline).""" + return "\t".join(str(field) for field in astuple(self)) + + +def member_pattern(tech: str) -> re.Pattern[str]: + """Regex matching the raw member *basenames* of ``tech``, capturing the year.""" + if tech == "solar": + return re.compile(r"^solar_gen_cf_(?P\d{4})\.nc$") + wind = re.fullmatch(r"wind_(?P\d+m)", tech) + if wind is None: + raise ValueError(f"unsupported tech {tech!r}: expected 'solar' or 'wind_m' (e.g. wind_100m)") + return re.compile(rf"^wind_gen_cf_(?P\d{{4}})_{wind['height']}\.nc$") + + +def dest_filename(member: str) -> str: + """Registry filename for a raw member (``.../wind_gen_cf_2012_125m.nc`` -> ``..._compressed.nc``).""" + return Path(member).stem + "_compressed.nc" + + +def scan_tech(zip_dir: Path, tech: str, years: range) -> dict[int, tuple[Path, str, int]]: + """ + Map ``year -> (zip_path, member, member_bytes)`` for one tech. + + Raises if the archives are missing, a year is served by two members, or any + requested year is absent — the caller gets a complete map or an exception. + """ + pattern = member_pattern(tech) + archives = sorted(zip_dir.glob(f"{tech}_historical_*.zip")) + if not archives: + raise FileNotFoundError(f"{tech}: no archives matching '{tech}_historical_*.zip' under {zip_dir}") + + wanted = set(years) + found: dict[int, tuple[Path, str, int]] = {} + for archive in archives: + with zipfile.ZipFile(archive) as zf: + for info in zf.infolist(): + if info.is_dir(): + continue + match = pattern.match(Path(info.filename).name) + if match is None: + continue + year = int(match["year"]) + if year not in wanted: + continue + if year in found: + prior_zip, prior_member, _ = found[year] + raise ValueError( + f"{tech} {year}: duplicate members " + f"{prior_zip.name}:{prior_member} and {archive.name}:{info.filename}" + ) + found[year] = (archive, info.filename, info.file_size) + + missing = sorted(wanted - found.keys()) + if missing: + raise ValueError( + f"{tech}: incomplete coverage — missing years {missing} " + f"(searched {[a.name for a in archives]} in {zip_dir}); " + f"found {len(found)}/{len(wanted)} of {min(years)}-{max(years)}" + ) + return found + + +def enumerate_tasks( + zip_dir: Path, + techs: tuple[str, ...] = DEFAULT_TECHS, + years: range = range(DEFAULT_FIRST_YEAR, DEFAULT_LAST_YEAR + 1), +) -> list[Task]: + """Build the full, contiguously numbered task list: techs in order, years descending.""" + tasks: list[Task] = [] + for tech in techs: + found = scan_tech(zip_dir, tech, years) + for year in sorted(found, reverse=True): + archive, member, size = found[year] + tasks.append( + Task( + idx=len(tasks) + 1, + tech=tech, + year=year, + zip_path=str(archive), + member=member, + member_bytes=size, + dest_filename=dest_filename(member), + ) + ) + return tasks + + +def write_tasks_tsv(tasks: list[Task], out_path: str | Path) -> Path: + """Write ``tasks`` to ``out_path`` (headerless TSV) and return the path.""" + out = Path(out_path) + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text("".join(task.as_tsv() + "\n" for task in tasks)) + return out + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--zip-dir", type=Path, default=DEFAULT_ZIP_DIR, help="Directory holding the raw GODEEEP zips.") + ap.add_argument( + "--techs", + nargs="+", + default=list(DEFAULT_TECHS), + help="Techs to enumerate, in output order (solar, wind_100m, wind_125m, ...).", + ) + ap.add_argument("--first-year", type=int, default=DEFAULT_FIRST_YEAR, help="First year required per tech.") + ap.add_argument("--last-year", type=int, default=DEFAULT_LAST_YEAR, help="Last year required per tech.") + ap.add_argument("--output", type=Path, default=Path("tasks.tsv"), help="Destination TSV.") + args = ap.parse_args() + + years = range(args.first_year, args.last_year + 1) + tasks = enumerate_tasks(args.zip_dir, tuple(args.techs), years) + out = write_tasks_tsv(tasks, args.output) + + total_bytes = sum(task.member_bytes for task in tasks) + print(f"[tasks] wrote {out}: {len(tasks)} tasks, {total_bytes / 1e12:.2f} TB raw", flush=True) + for tech in args.techs: + rows = [task for task in tasks if task.tech == tech] + print( + f"[tasks] {tech}: {len(rows)} years {min(t.year for t in rows)}-{max(t.year for t in rows)}", flush=True + ) + print(f"[tasks] submit with --array=1-{len(tasks)}", flush=True) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/workflow/scripts/nrel_exclusion/stage_godeeep_member.py b/workflow/scripts/nrel_exclusion/stage_godeeep_member.py new file mode 100644 index 00000000..61bbfbfc --- /dev/null +++ b/workflow/scripts/nrel_exclusion/stage_godeeep_member.py @@ -0,0 +1,90 @@ +""" +Stream one member out of a raw GODEEEP zip64 archive onto local disk. + +The raw archives are zip64 (~90 GB each, ~4.4 GB members), which rules out the +system ``unzip`` on some nodes; python's ``zipfile`` handles them and streams in +16 MiB blocks so peak memory stays flat regardless of member size. + +The staged byte count is checked against the archive's central directory. A short +read (truncated stage, full scratch filesystem, interrupted job) therefore fails +here rather than surfacing later as a corrupt NetCDF. + +Usage: + python stage_godeeep_member.py \ + --zip /oak/.../GoDEEEP_Capacity_Factors/solar_historical_1980_2022.zip \ + --member solar_historical/solar_gen_cf_2012.nc \ + --output $L_SCRATCH/solar_gen_cf_2012.nc \ + [--expect-bytes 4443329482] +""" + +import argparse +import shutil +import sys +import time +import zipfile +from pathlib import Path + +COPY_BLOCK = 16 << 20 # bytes per copyfileobj block + + +def stage_member( + zip_path: str | Path, + member: str, + dest: str | Path, + expect_bytes: int | None = None, +) -> int: + """ + Extract ``member`` from ``zip_path`` to ``dest`` and return the staged byte count. + + ``expect_bytes`` (e.g. the ``member_bytes`` column of ``tasks.tsv``) is checked + against the archive's central directory before any bytes move, so a stale task + list fails fast instead of after a multi-GB copy. + """ + dest = Path(dest) + dest.parent.mkdir(parents=True, exist_ok=True) + + with zipfile.ZipFile(zip_path) as zf: + try: + info = zf.getinfo(str(member)) + except KeyError: + raise KeyError(f"member {member!r} not found in {zip_path}") from None + if expect_bytes is not None and info.file_size != expect_bytes: + raise ValueError( + f"{zip_path}:{member} central directory reports {info.file_size} bytes, task list expects {expect_bytes}" + ) + with zf.open(info) as src, open(dest, "wb") as out: + shutil.copyfileobj(src, out, length=COPY_BLOCK) + + staged = dest.stat().st_size + if staged != info.file_size: + raise OSError( + f"staged {staged} bytes to {dest}, central directory says {info.file_size} for {zip_path}:{member}" + ) + return staged + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--zip", required=True, help="Raw GODEEEP zip64 archive.") + ap.add_argument("--member", required=True, help="Member path inside the archive.") + ap.add_argument("--output", required=True, help="Destination file (parent directories are created).") + ap.add_argument( + "--expect-bytes", + type=int, + default=None, + help="Optional expected member size; cross-checked against the central directory.", + ) + args = ap.parse_args() + + t0 = time.monotonic() + staged = stage_member(args.zip, args.member, args.output, expect_bytes=args.expect_bytes) + elapsed = time.monotonic() - t0 + print( + f"[stage] {args.zip}:{args.member} -> {args.output} " + f"({staged / 1e9:.2f} GB in {elapsed:.1f}s, {staged / 1e6 / max(elapsed, 1e-9):.0f} MB/s)", + flush=True, + ) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/workflow/scripts/retrieve_godeeep_cf.py b/workflow/scripts/retrieve_godeeep_cf.py new file mode 100644 index 00000000..2f393e5a --- /dev/null +++ b/workflow/scripts/retrieve_godeeep_cf.py @@ -0,0 +1,267 @@ +"""Place one compressed GODEEEP capacity-factor file where the DAG expects it. + +Driven by ``rule retrieve_godeeep_cf`` (rules/retrieve.smk). Retrieval used to +happen inside ``build_renewable_profiles.py``, hidden behind a hardcoded Zenodo +record table that returned ``None`` on a miss. It is now an explicit DAG step +whose source — the local mirror or a Zenodo record — is declared in +``config["godeeep_cf_registry"]`` and resolved by ``godeeep_cf_registry``. + +There is NO fallback after resolution: the winning source must serve exactly the +``(scenario, technology, hub height, year)`` the output path names. A missing +file, a checksum mismatch or a resolution that disagrees with the requested +wildcards raises — nothing substitutes another year, hub height or screening +variant (issue #803). +""" + +import logging +import os +import re +import shutil +from pathlib import Path + +from _helpers import configure_logging, validate_checksum +from godeeep_cf_registry import CfResolution, cf_filename, load_sources, resolve_cf +from zenodo_downloader import ZenodoScenarioDownloader + +logger = logging.getLogger(__name__) + +#: The ``{cf_file}`` wildcard of rule retrieve_godeeep_cf, e.g. +#: ``wind_gen_cf_2019_125m_compressed`` / ``solar_gen_cf_2019_compressed``. +CF_FILE_RE = re.compile( + r"^(?Psolar|wind)_gen_cf_(?P\d{4})(?P_\d+m)?_compressed$", +) + +#: Optional checksum manifest at the root of a local source: one +#: `` `` line per published file. +CHECKSUM_MANIFEST = "SHA256SUMS" + + +def parse_cf_file(cf_file: str) -> tuple[str, str, int]: + """Split the ``{cf_file}`` wildcard into ``(technology, hub height, year)``. + + Raises + ------ + ValueError + If ``cf_file`` is not a published GODEEEP CF file name. The rule's + ``wildcard_constraints`` already enforce this shape; the check is + repeated here so the script is safe to call directly. + """ + match = CF_FILE_RE.match(cf_file) + if match is None: + raise ValueError( + f"{cf_file!r} is not a GODEEEP capacity-factor file name; expected e.g. " + "'solar_gen_cf_2019_compressed' or 'wind_gen_cf_2019_125m_compressed'.", + ) + return ( + match.group("technology"), + match.group("wind_height") or "", + int(match.group("year")), + ) + + +def pypsa_technology(technology: str) -> str: + """A pypsa-usa carrier standing in for a GODEEEP technology family. + + ``resolve_cf`` takes the pypsa-usa technology wildcard, while the output + path only carries the GODEEEP family. Every wind carrier shares one + ``TechSpec`` (the hub height comes from ``godeeep_wind_height``), so any of + them resolves the same file. + """ + return "solar" if technology == "solar" else "onwind" + + +def resolve_request(config, scenario: str, cf_file: str) -> CfResolution: + """Resolve the requested output path against the configured registry. + + The resolution is cross-checked against the wildcards: the registry + resolves from the config alone, so a config that has drifted from the DAG + (different hub height, different weather year) must fail loudly instead of + filling the output path with a different file. + + Raises + ------ + ValueError + If the wildcards are malformed, the config is invalid, or the resolved + file is not the one the output path names. + CfNotAvailableError + If no configured source declares the requested dataset and year. + """ + technology, wind_height, year = parse_cf_file(cf_file) + requested = cf_filename(technology, wind_height, scenario, year) + + resolution = resolve_cf(config, pypsa_technology(technology), planning_horizon=year) + + if (resolution.scenario, resolution.filename) != (scenario, requested): + raise ValueError( + f"rule retrieve_godeeep_cf was asked for '{scenario}/{requested}', but the configured " + f"godeeep_cf_registry resolves to '{resolution.scenario}/{resolution.filename}' " + f"(dataset {resolution.dataset_key!r}, year {resolution.year}). Retrieving it would " + "silently substitute a different scenario, hub height or year. Fix the mismatch " + "between the requested path and renewable_scenarios / renewable_weather_years / " + "godeeep_wind_height, or re-run after clearing the stale target.", + ) + return resolution + + +def local_source_root(config, resolution: CfResolution) -> Path: + """Filesystem root of the local source that won the resolution.""" + return Path(load_sources(config)[resolution.source_index].root) + + +def read_checksum_manifest(manifest: Path) -> dict[str, str]: + """Parse a ``sha256sum``-style manifest into ``{path: hash}``. + + Entry paths are kept as published (relative to the manifest's directory); + the leading ``*``/space binary marker of ``sha256sum`` output is stripped. + """ + entries: dict[str, str] = {} + for line in manifest.read_text().splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + checksum, _, name = line.partition(" ") + name = name.strip().lstrip("*") + if checksum and name: + entries[os.path.normpath(name)] = checksum + return entries + + +def expected_checksum(entries: dict[str, str], relative: str, manifest: Path) -> str: + """The published hash of ``relative``, keyed by path or by unique basename. + + Mirrors generated with ``sha256sum /*.nc`` key their entries on the + bare file name, so both spellings are accepted — but only when the name is + unambiguous. + + Raises + ------ + ValueError + If the manifest has no entry for the file, or several entries share its + basename with different hashes. + """ + relative = os.path.normpath(relative) + if relative in entries: + return entries[relative] + + basename = os.path.basename(relative) + matches = {checksum for name, checksum in entries.items() if os.path.basename(name) == basename} + if len(matches) == 1: + return matches.pop() + if len(matches) > 1: + raise ValueError( + f"{manifest} lists several conflicting hashes for files named {basename!r} and no entry " + f"for {relative!r}; cannot tell which one belongs to the requested file.", + ) + raise ValueError( + f"{manifest} exists but publishes no checksum for {relative!r}, so the file's integrity " + "cannot be verified. Regenerate the manifest at the source root, or remove it to skip " + "checksum verification for this source.", + ) + + +def verify_local_checksum(source_path: Path, root: Path) -> None: + """Verify ``source_path`` against the source's manifest, if it has one. + + A source without a ``SHA256SUMS`` manifest is used as-is (the file's + existence is still required); a source with one must cover the file. + + Raises + ------ + ValueError + If the manifest does not cover the file or the hash does not match. + """ + manifest = root / CHECKSUM_MANIFEST + if not manifest.is_file(): + logger.info(f"No {CHECKSUM_MANIFEST} at {root}; skipping checksum verification.") + return + + relative = os.path.relpath(source_path, root) + checksum = expected_checksum(read_checksum_manifest(manifest), relative, manifest) + try: + validate_checksum(source_path, checksum=f"sha256:{checksum}") + except AssertionError as exc: + raise ValueError( + f"Checksum mismatch for {source_path}: {manifest} publishes sha256 {checksum}. " + "The local mirror is corrupt or out of date; re-stage the file rather than using it.", + ) from exc + logger.info(f"Verified sha256 of {relative} against {manifest}.") + + +def place_local(source_path: Path, out_path: Path, copy: bool) -> None: + """Symlink (default) or copy the mirrored file to the rule's output path.""" + out_path.parent.mkdir(parents=True, exist_ok=True) + # Snakemake does not clear an existing output before a script runs, and + # os.symlink refuses an existing target. + if out_path.is_symlink() or out_path.exists(): + out_path.unlink() + + if copy: + logger.info(f"Copying {source_path} -> {out_path}") + shutil.copy2(source_path, out_path) + else: + logger.info(f"Linking {source_path} -> {out_path}") + os.symlink(source_path, out_path) + + +def retrieve_local(config, resolution: CfResolution, out_path: Path) -> None: + """Serve a resolved file from a local mirror. + + Raises + ------ + FileNotFoundError + If the mirror declares the dataset/year but does not hold the file. + """ + source_path = Path(resolution.path) + if not source_path.is_file(): + raise FileNotFoundError( + f"The local GODEEEP CF source declares {resolution.dataset_key!r} for year " + f"{resolution.year}, but {source_path} does not exist. Either stage the file there or " + "drop the year from that source in godeeep_cf_registry so another source can serve it.", + ) + verify_local_checksum(source_path, local_source_root(config, resolution)) + place_local(source_path, out_path, copy=resolution.copy_local) + + +def retrieve_zenodo(resolution: CfResolution, out_path: Path) -> None: + """Download a resolved file from its Zenodo record (checksum-verified).""" + out_path.parent.mkdir(parents=True, exist_ok=True) + downloader = ZenodoScenarioDownloader(download_dir=out_path.parent) + logger.info( + f"Downloading {resolution.filename} from Zenodo record {resolution.record_id} " + f"({resolution.dataset_key}, year {resolution.year}).", + ) + downloader.download_to_path(resolution.record_id, resolution.filename, out_path) + + +def retrieve_cf(config, scenario: str, cf_file: str, out_path) -> CfResolution: + """Resolve and retrieve one GODEEEP CF file; raise on any failure.""" + out_path = Path(out_path) + resolution = resolve_request(config, scenario, cf_file) + logger.info( + f"Resolved {resolution.dataset_key} year {resolution.year} to [{resolution.kind}] {resolution.location}.", + ) + + if resolution.kind == "local": + retrieve_local(config, resolution, out_path) + else: + retrieve_zenodo(resolution, out_path) + return resolution + + +if __name__ == "__main__": + if "snakemake" not in globals(): + from _helpers import mock_snakemake + + snakemake = mock_snakemake( + "retrieve_godeeep_cf", + scenario="historical", + cf_file="solar_gen_cf_2019_compressed", + ) + configure_logging(snakemake) + + retrieve_cf( + snakemake.config, + snakemake.wildcards.scenario, + snakemake.wildcards.cf_file, + snakemake.output[0], + ) diff --git a/workflow/scripts/test/test_compress_godeeep.py b/workflow/scripts/test/test_compress_godeeep.py new file mode 100644 index 00000000..47455a30 --- /dev/null +++ b/workflow/scripts/test/test_compress_godeeep.py @@ -0,0 +1,180 @@ +"""Unit tests for nrel_exclusion/compress_godeeep.py (the uint8 CF quantizer). + +``compress_one`` produced every file in the Oak registry +($OAK/GoDEEEP_Capacity_Factors_compressed), so its encoding contract is what the +registry readers decode against. These tests pin that contract on a miniature +grid shaped like the real thing (float ``capacity_factor`` over Time x +south_north x west_east, XLAT/XLONG aux vars carrying ``_FillValue``): + + * uint8 with scale_factor 1/254 and _FillValue 255, so xarray auto-decodes, + * round-trip error bounded by the quantization step, NaN in -> NaN out, + * coordinates, variable attributes and provenance survive the rewrite, + * compressing the same input twice yields byte-identical uint8 payloads — + the determinism the Phase 1 pilot leaned on when it proved the regenerated + 2012 files matched the published Zenodo artifacts. +""" + +import os +import sys + +import netCDF4 +import numpy as np +import pytest + +sys.path.append(os.path.join(os.path.dirname(__file__), "..")) + +from nrel_exclusion.compress_godeeep import FILL, SCALE, compress_one + +pytestmark = pytest.mark.fast + +N_TIME, N_SOUTH_NORTH, N_WEST_EAST = 7, 3, 4 +TOL = 1.0 / 254.0 +PROJECTION = "LambertConformal(stand_lon=-97.0, moad_cen_lat=39.0)" + + +def make_raw_cf() -> np.ndarray: + """Miniature capacity-factor field: full 0-1 range, a NaN row and a lone NaN cell.""" + rng = np.random.default_rng(20260831) + cf = rng.random((N_TIME, N_SOUTH_NORTH, N_WEST_EAST)).astype(np.float32) + cf[0, 0, 0] = 0.0 # exact endpoints must survive quantization + cf[0, 0, 1] = 1.0 + cf[0, 0, 2] = 0.5 + cf[1, 2, :] = np.nan # an offshore/out-of-domain row + cf[3, 1, 2] = np.nan # and a single dead cell + return cf + + +def write_raw(path, cf: np.ndarray) -> None: + """Write a raw-GODEEEP-shaped NetCDF (dims, aux vars and attrs mirror the real files).""" + lat = np.linspace(31.0, 49.0, N_SOUTH_NORTH * N_WEST_EAST, dtype=np.float32).reshape(N_SOUTH_NORTH, N_WEST_EAST) + lon = np.linspace(-124.0, -67.0, N_SOUTH_NORTH * N_WEST_EAST, dtype=np.float32).reshape(N_SOUTH_NORTH, N_WEST_EAST) + with netCDF4.Dataset(path, "w") as ds: + ds.createDimension("Time", N_TIME) + ds.createDimension("south_north", N_SOUTH_NORTH) + ds.createDimension("west_east", N_WEST_EAST) + + # Aux vars carrying _FillValue: it can only be set at create time, which is + # the branch compress_one has to get right. + for name, values in (("XLAT", lat), ("XLONG", lon)): + var = ds.createVariable(name, "f4", ("south_north", "west_east"), fill_value=np.float32(np.nan)) + var[:] = values + var.units = "degrees" + + ds.createVariable("Time", "i8", ("Time",))[:] = np.arange(N_TIME, dtype=np.int64) + ds.createVariable("south_north", "i8", ("south_north",))[:] = np.arange(N_SOUTH_NORTH, dtype=np.int64) + ds.createVariable("west_east", "i8", ("west_east",))[:] = np.arange(N_WEST_EAST, dtype=np.int64) + + cf_var = ds.createVariable( + "capacity_factor", + "f4", + ("Time", "south_north", "west_east"), + fill_value=np.float32(np.nan), + ) + cf_var.set_auto_mask(False) + cf_var[:] = cf + cf_var.projection = PROJECTION + cf_var.coordinates = "XLAT XLONG" + cf_var.missing_value = np.float32(-999.0) # dropped by compress_one's skip list + ds.title = "synthetic GODEEEP capacity factors" + + +@pytest.fixture +def raw_cf() -> np.ndarray: + return make_raw_cf() + + +@pytest.fixture +def raw_path(tmp_path, raw_cf): + path = tmp_path / "solar_gen_cf_2012.nc" + write_raw(path, raw_cf) + return path + + +@pytest.fixture +def compressed_path(tmp_path, raw_path): + out = tmp_path / "solar_gen_cf_2012_compressed.nc" + compress_one(str(raw_path), str(out), chunk_t=3) # 7 hours / 3 => a partial final chunk + return out + + +def read_packed(path) -> np.ndarray: + """The stored uint8 payload, with auto-scaling and auto-masking switched off.""" + with netCDF4.Dataset(path) as ds: + var = ds.variables["capacity_factor"] + var.set_auto_scale(False) + var.set_auto_mask(False) + return np.asarray(var[:]) + + +def test_encoding_is_uint8_scaled_by_1_over_254(compressed_path): + with netCDF4.Dataset(compressed_path) as ds: + var = ds.variables["capacity_factor"] + assert var.dtype == np.uint8 + assert var.getncattr("scale_factor") == pytest.approx(1.0 / 254.0) + assert var.getncattr("scale_factor") == SCALE + assert var.getncattr("add_offset") == pytest.approx(0.0) + assert int(var.getncattr("_FillValue")) == int(FILL) == 255 + assert var.dimensions == ("Time", "south_north", "west_east") + + +def test_roundtrip_error_within_quantization_step(compressed_path, raw_cf): + with netCDF4.Dataset(compressed_path) as ds: + decoded = ds.variables["capacity_factor"][:] # auto-decoded to float, 255 -> masked + decoded = np.ma.filled(decoded, np.nan).astype(np.float64) + + valid = ~np.isnan(raw_cf) + err = np.abs(decoded[valid] - raw_cf.astype(np.float64)[valid]).max() + assert err <= TOL + 1e-9 + # The endpoints of the 0-1 range are representable exactly. + assert decoded[0, 0, 0] == pytest.approx(0.0, abs=1e-9) + assert decoded[0, 0, 1] == pytest.approx(1.0, abs=1e-6) + + +def test_nan_survives_as_fill_255(compressed_path, raw_cf): + packed = read_packed(compressed_path) + nan_in = np.isnan(raw_cf) + assert np.array_equal(packed == int(FILL), nan_in) + + with netCDF4.Dataset(compressed_path) as ds: + decoded = ds.variables["capacity_factor"][:] + decoded = np.ma.filled(decoded, np.nan) + assert np.array_equal(np.isnan(decoded), nan_in) + assert nan_in.sum() == N_WEST_EAST + 1 # the NaN row plus the dead cell + + +def test_coords_and_attrs_preserved(compressed_path, raw_path): + with netCDF4.Dataset(raw_path) as src, netCDF4.Dataset(compressed_path) as dst: + assert {name: len(dim) for name, dim in dst.dimensions.items()} == { + "Time": N_TIME, + "south_north": N_SOUTH_NORTH, + "west_east": N_WEST_EAST, + } + for name in ("XLAT", "XLONG", "Time", "south_north", "west_east"): + assert np.array_equal(np.asarray(src.variables[name][:]), np.asarray(dst.variables[name][:])), name + assert dst.variables[name].dtype == src.variables[name].dtype, name + + # _FillValue is re-created (not copied as a plain attribute) on aux vars. + assert np.isnan(dst.variables["XLAT"].getncattr("_FillValue")) + assert dst.variables["XLAT"].getncattr("units") == "degrees" + + cf_var = dst.variables["capacity_factor"] + assert cf_var.getncattr("projection") == PROJECTION + assert cf_var.getncattr("coordinates") == "XLAT XLONG" + # The source's float encoding attributes must not leak onto the uint8 field. + assert "missing_value" not in cf_var.ncattrs() + + +def test_provenance_attrs_stamped(compressed_path, raw_path): + with netCDF4.Dataset(compressed_path) as ds: + assert ds.getncattr("compression_source") == str(raw_path) + assert ds.getncattr("compression_encoding") == "uint8 scale_factor=1/254 zlib=4" + + +def test_compressing_twice_is_deterministic(tmp_path, raw_path): + """Two runs over the same input must agree bit-for-bit on the packed payload.""" + first = tmp_path / "first.nc" + second = tmp_path / "second.nc" + compress_one(str(raw_path), str(first), chunk_t=3) + compress_one(str(raw_path), str(second), chunk_t=5) # chunking must not change the payload + + assert read_packed(first).tobytes() == read_packed(second).tobytes() diff --git a/workflow/scripts/test/test_godeeep_cf_registry.py b/workflow/scripts/test/test_godeeep_cf_registry.py new file mode 100644 index 00000000..3e108dcf --- /dev/null +++ b/workflow/scripts/test/test_godeeep_cf_registry.py @@ -0,0 +1,466 @@ +"""Unit tests for the config-declared GODEEEP CF registry (godeeep_cf_registry.py). + +The registry replaces the inline record-key/filename construction in +build_renewable_profiles.py and the hardcoded record table in +zenodo_downloader.py, which together produced issue #803: a config asking for +``godeeep_wind_height: "_100m"`` against Zenodo-only sources fell through +``ZenodoScenarioDownloader.download_scenario_file`` -> ``None`` and only blew up +much later as a ``TypeError`` on a ``None`` path. + +These tests pin the invariants that make that impossible: + * published file names and dataset keys are locked to the strings on disk / + on Zenodo, so a refactor cannot rename them silently, + * there is no ``"_100m"`` hub-height default and no ``"_80m"`` dataset, + * sources are tried in configured order — first hit wins, no fallback after, + * an unavailable (dataset, year) raises CfNotAvailableError whose message + names the dataset key, the requested year and the available years, + * validate_godeeep_cf_config aggregates every problem into one parse-time error. + +Everything is built from inline dicts: no filesystem, no network. +""" + +import os +import sys + +import pytest + +sys.path.append(os.path.join(os.path.dirname(__file__), "..")) + +from godeeep_cf_registry import ( + CfNotAvailableError, + cf_filename, + dataset_key, + godeeep_tech_spec, + load_sources, + parse_years, + resolve_cf, + resolve_weather_year, + validate_godeeep_cf_config, +) + +pytestmark = pytest.mark.fast + +OAK_ROOT = "/oak/stanford/groups/iazevedo/GoDEEEP_Capacity_Factors_compressed" + +LOCAL_SOURCE = { + "kind": "local", + "root": OAK_ROOT, + "layout": "{scenario}/{tech_dir}/{filename}", + "datasets": { + "solar_historical_compressed": {"years": "1980-2022"}, + "wind_100m_historical_compressed": {"years": "1980-2022"}, + "wind_125m_historical_compressed": {"years": "1980-2022"}, + }, +} + +ZENODO_SOURCE = { + "kind": "zenodo", + "datasets": { + "solar_historical_compressed": {"record": 20127513, "years": [2012]}, + "wind_125m_historical_compressed": {"record": 20127520, "years": [2012]}, + "solar_rcp85cooler_compressed": {"record": 20127633, "years": [2030, 2040, 2050]}, + "wind_125m_rcp85cooler_compressed": {"record": 20127645, "years": [2030, 2040, 2050]}, + }, +} + + +def make_config(sources, **overrides): + """A minimal godeeep config; ``overrides`` replace/add top-level keys.""" + config = { + "renewable": {"dataset": "godeeep", "solar": {}, "onwind": {}}, + "renewable_scenarios": ["historical"], + "renewable_weather_years": [2012], + "godeeep_wind_height": "_125m", + "scenario": {"planning_horizons": [2030]}, + "godeeep_cf_registry": {"copy_local": False, "sources": sources}, + } + config.update(overrides) + return config + + +# -------------------------------------------------------------------------- +# parse_years +# -------------------------------------------------------------------------- + + +def test_parse_years_inclusive_range_string(): + years = parse_years("1980-2022") + assert years[0] == 1980 + assert years[-1] == 2022 + # 1980..2022 inclusive is the 43 files actually present on the Oak mirror. + assert len(years) == 43 + + +def test_parse_years_explicit_list(): + assert parse_years([2030, 2040, 2050]) == [2030, 2040, 2050] + assert parse_years([2012]) == [2012] + assert parse_years([]) == [] + # Deduped and sorted so availability checks are order-independent. + assert parse_years([2040, 2030, 2040]) == [2030, 2040] + + +def test_parse_years_accepts_numeric_strings(): + assert parse_years(["2030", 2040]) == [2030, 2040] + + +@pytest.mark.parametrize( + "garbage", + [ + "1980_2022", # wrong separator + "1980-2022-2030", # not a two-endpoint range + "1980 to 2022", + "2022-1980", # reversed endpoints would silently yield an empty set + "historical", + "", + None, + True, + {"start": 1980}, + [1980, "twenty-twenty"], + [None], + ], +) +def test_parse_years_rejects_garbage(garbage): + with pytest.raises(ValueError): + parse_years(garbage) + + +# -------------------------------------------------------------------------- +# dataset_key / cf_filename — locked to the published strings +# -------------------------------------------------------------------------- + + +def test_dataset_key_matches_legacy_record_keys(): + # Exactly the keys build_renewable_profiles.py built inline and + # zenodo_downloader.py's scenario_records table used. + assert dataset_key("solar", "", "historical") == "solar_historical_compressed" + assert dataset_key("wind", "_125m", "historical") == "wind_125m_historical_compressed" + assert dataset_key("wind", "_100m", "historical") == "wind_100m_historical_compressed" + assert dataset_key("wind", "_125m", "rcp85cooler") == "wind_125m_rcp85cooler_compressed" + assert dataset_key("solar", "", "rcp45hotter") == "solar_rcp45hotter_compressed" + + +def test_cf_filename_locks_published_names(): + assert cf_filename("solar", "", "historical", 2019) == "solar_gen_cf_2019_compressed.nc" + assert cf_filename("wind", "_125m", "historical", 2019) == "wind_gen_cf_2019_125m_compressed.nc" + assert cf_filename("wind", "_100m", "historical", 2019) == "wind_gen_cf_2019_100m_compressed.nc" + + +def test_cf_filename_is_scenario_independent(): + # The scenario lives in the containing record/directory, never in the name — + # the Zenodo future records publish e.g. solar_gen_cf_2030_compressed.nc. + assert cf_filename("solar", "", "rcp85cooler", 2030) == "solar_gen_cf_2030_compressed.nc" + assert cf_filename("wind", "_125m", "rcp45cooler", 2050) == "wind_gen_cf_2050_125m_compressed.nc" + for scenario in ("historical", "rcp45hotter", "rcp85cooler"): + assert cf_filename("solar", "", scenario, 2030) == "solar_gen_cf_2030_compressed.nc" + + +# -------------------------------------------------------------------------- +# godeeep_tech_spec — no hub-height default +# -------------------------------------------------------------------------- + + +def test_godeeep_tech_spec_solar_ignores_wind_height(): + spec = godeeep_tech_spec("solar", {"godeeep_wind_height": "_125m"}) + assert (spec.technology, spec.wind_height) == ("solar", "") + assert spec.tech_dir == "solar" + # Still tuple-unpackable like the code it replaces. + technology, wind_height = spec + assert (technology, wind_height) == ("solar", "") + + +@pytest.mark.parametrize("tech", ["onwind", "offwind", "offwind_floating"]) +def test_godeeep_tech_spec_wind_uses_configured_height(tech): + spec = godeeep_tech_spec(tech, {"godeeep_wind_height": "_100m"}) + assert (spec.technology, spec.wind_height) == ("wind", "_100m") + assert spec.tech_dir == "wind_100m" + + +def test_godeeep_tech_spec_raises_when_wind_height_missing(): + # The old code silently defaulted to "_100m", which Zenodo does not publish. + with pytest.raises(ValueError) as excinfo: + godeeep_tech_spec("onwind", {}) + message = str(excinfo.value) + assert "godeeep_wind_height" in message + assert "_100m" in message and "_125m" in message + + +def test_godeeep_tech_spec_rejects_80m(): + with pytest.raises(ValueError) as excinfo: + godeeep_tech_spec("onwind", {"godeeep_wind_height": "_80m"}) + message = str(excinfo.value) + assert "_80m" in message + assert "_100m" in message and "_125m" in message + + +def test_godeeep_tech_spec_rejects_unknown_technology(): + with pytest.raises(ValueError) as excinfo: + godeeep_tech_spec("hydro", {"godeeep_wind_height": "_125m"}) + assert "hydro" in str(excinfo.value) + + +# -------------------------------------------------------------------------- +# resolve_weather_year / resolve_scenario +# -------------------------------------------------------------------------- + + +def test_resolve_weather_year_historical_reads_weather_years(): + config = make_config([LOCAL_SOURCE], renewable_weather_years=[2019, 2020]) + assert resolve_weather_year(config, planning_horizon=2030) == 2019 + + +def test_resolve_weather_year_future_reads_planning_horizon(): + config = make_config([ZENODO_SOURCE], renewable_scenarios=["rcp85cooler"]) + assert resolve_weather_year(config, planning_horizon=2040) == 2040 + + +def test_resolve_weather_year_reports_missing_renewable_scenarios(): + config = make_config([LOCAL_SOURCE]) + del config["renewable_scenarios"] + with pytest.raises(ValueError) as excinfo: + resolve_weather_year(config, planning_horizon=2019) + assert "renewable_scenarios" in str(excinfo.value) + + +# -------------------------------------------------------------------------- +# resolve_cf — source precedence, no fallback +# -------------------------------------------------------------------------- + + +def test_resolve_cf_local_source_wins_over_zenodo_for_overlapping_year(): + # 2012 is offered by BOTH sources; the local source is listed first. + config = make_config([LOCAL_SOURCE, ZENODO_SOURCE], renewable_weather_years=[2012]) + resolution = resolve_cf(config, "onwind") + assert resolution.kind == "local" + assert resolution.source_index == 0 + assert resolution.dataset_key == "wind_125m_historical_compressed" + assert resolution.year == 2012 + assert resolution.filename == "wind_gen_cf_2012_125m_compressed.nc" + assert resolution.path == f"{OAK_ROOT}/historical/wind_125m/wind_gen_cf_2012_125m_compressed.nc" + assert resolution.record_id is None + + +def test_resolve_cf_order_is_config_order_not_kind_priority(): + # Same two sources, zenodo declared first -> zenodo wins for 2012. + config = make_config([ZENODO_SOURCE, LOCAL_SOURCE], renewable_weather_years=[2012]) + resolution = resolve_cf(config, "onwind") + assert resolution.kind == "zenodo" + assert resolution.source_index == 0 + assert resolution.record_id == "20127520" + assert resolution.path is None + assert resolution.location == "zenodo:20127520" + + +def test_resolve_cf_falls_through_to_zenodo_only_when_local_lacks_the_dataset(): + # Future scenario: local mirror holds historical only. + config = make_config( + [LOCAL_SOURCE, ZENODO_SOURCE], + renewable_scenarios=["rcp85cooler"], + ) + resolution = resolve_cf(config, "solar", planning_horizon=2040) + assert resolution.kind == "zenodo" + assert resolution.record_id == "20127633" + assert resolution.dataset_key == "solar_rcp85cooler_compressed" + assert resolution.filename == "solar_gen_cf_2040_compressed.nc" + + +def test_resolve_cf_local_path_uses_configured_layout(): + config = make_config( + [{**LOCAL_SOURCE, "root": "/mnt/cf", "layout": "{scenario}/{dataset_key}/{year}/{filename}"}], + renewable_weather_years=[1995], + ) + resolution = resolve_cf(config, "solar") + assert resolution.path == ("/mnt/cf/historical/solar_historical_compressed/1995/solar_gen_cf_1995_compressed.nc") + + +def test_resolve_cf_copy_local_defaults_from_registry_block(): + config = make_config([LOCAL_SOURCE], renewable_weather_years=[2012]) + config["godeeep_cf_registry"]["copy_local"] = True + assert resolve_cf(config, "solar").copy_local is True + + +def test_resolve_cf_raises_with_dataset_key_year_and_available_years(): + # 1999 exists on Oak but not on Zenodo; a Zenodo-only registry must refuse it. + config = make_config([ZENODO_SOURCE], renewable_weather_years=[1999]) + with pytest.raises(CfNotAvailableError) as excinfo: + resolve_cf(config, "solar") + message = str(excinfo.value) + assert "solar_historical_compressed" in message # dataset key + assert "1999" in message # requested year + assert "2012" in message # available years + assert "solar_gen_cf_1999_compressed.nc" in message + + +def test_resolve_cf_never_substitutes_a_neighbouring_year(): + config = make_config([LOCAL_SOURCE], renewable_weather_years=[2023]) + with pytest.raises(CfNotAvailableError) as excinfo: + resolve_cf(config, "solar") + # 2022 is available and adjacent — it must be reported, not silently used. + assert "1980-2022" in str(excinfo.value) + + +def test_resolve_cf_reports_unpublished_dataset_with_empty_year_list(): + # Mirrors wind_125m_rcp45hotter: a record id exists in config but Zenodo + # returns 404, so it is declared with no years. + source = { + "kind": "zenodo", + "datasets": {"wind_125m_rcp45hotter_compressed": {"record": 20127545, "years": []}}, + } + config = make_config([source], renewable_scenarios=["rcp45hotter"]) + with pytest.raises(CfNotAvailableError) as excinfo: + resolve_cf(config, "onwind", planning_horizon=2030) + message = str(excinfo.value) + assert "wind_125m_rcp45hotter_compressed" in message + assert "2030" in message + # Distinguished from "not declared at all" so the operator knows the record + # id is configured but the record itself publishes nothing. + assert "no years published" in message + + +def test_resolve_cf_803_regression_100m_against_zenodo_only_sources(): + """Issue #803: this combination used to yield ``None``, then a TypeError.""" + config = make_config( + [ZENODO_SOURCE], + godeeep_wind_height="_100m", + renewable_weather_years=[2012], + ) + with pytest.raises(CfNotAvailableError) as excinfo: + resolve_cf(config, "onwind") + message = str(excinfo.value) + assert "wind_100m_historical_compressed" in message + assert "2012" in message + assert "dataset key not declared" in message + # And the same request against the Oak mirror resolves cleanly. + ok = resolve_cf( + make_config([LOCAL_SOURCE], godeeep_wind_height="_100m", renewable_weather_years=[2012]), + "onwind", + ) + assert ok.path == f"{OAK_ROOT}/historical/wind_100m/wind_gen_cf_2012_100m_compressed.nc" + + +# -------------------------------------------------------------------------- +# load_sources — malformed registries +# -------------------------------------------------------------------------- + + +def test_load_sources_preserves_declared_order(): + sources = load_sources(make_config([LOCAL_SOURCE, ZENODO_SOURCE])) + assert [source.kind for source in sources] == ["local", "zenodo"] + assert sources[0].years_for("solar_historical_compressed")[0] == 1980 + + +def test_load_sources_requires_the_registry_block(): + config = make_config([LOCAL_SOURCE]) + del config["godeeep_cf_registry"] + with pytest.raises(ValueError) as excinfo: + load_sources(config) + assert "godeeep_cf_registry" in str(excinfo.value) + + +@pytest.mark.parametrize( + "bad_source", + [ + {"kind": "s3", "datasets": {}}, # unknown kind + {"kind": "local", "datasets": {}}, # local without a root + {"kind": "local", "root": "/tmp", "datasets": {"k": "nineteen-eighty"}}, # bad years + {"kind": "zenodo", "datasets": {"k": {"years": [2012]}}}, # zenodo without a record + {"kind": "zenodo", "datasets": [1, 2, 3]}, # datasets not a mapping + ], +) +def test_load_sources_rejects_malformed_sources(bad_source): + with pytest.raises(ValueError): + load_sources(make_config([bad_source])) + + +# -------------------------------------------------------------------------- +# validate_godeeep_cf_config — parse-time gate +# -------------------------------------------------------------------------- + + +def test_validate_passes_for_a_resolvable_config(): + config = make_config([LOCAL_SOURCE, ZENODO_SOURCE], renewable_weather_years=[2019]) + resolved = validate_godeeep_cf_config(config) + assert {resolution.dataset_key for resolution in resolved} == { + "solar_historical_compressed", + "wind_125m_historical_compressed", + } + assert all(resolution.kind == "local" for resolution in resolved) + + +def test_validate_fails_loudly_for_1999_against_a_zenodo_only_registry(): + config = make_config([ZENODO_SOURCE], renewable_weather_years=[1999]) + with pytest.raises(CfNotAvailableError) as excinfo: + validate_godeeep_cf_config(config) + message = str(excinfo.value) + assert "1999" in message + assert "solar_historical_compressed" in message + assert "wind_125m_historical_compressed" in message + assert "2012" in message + + +def test_validate_reports_missing_renewable_scenarios_instead_of_keyerror(): + # build_electricity.smk / build_renewable_profiles.py did a bare + # config["renewable_scenarios"][0]; the KeyError named nothing useful. + config = make_config([LOCAL_SOURCE]) + del config["renewable_scenarios"] + with pytest.raises(CfNotAvailableError) as excinfo: + validate_godeeep_cf_config(config) + assert "renewable_scenarios" in str(excinfo.value) + + +def test_validate_reports_missing_weather_years(): + config = make_config([LOCAL_SOURCE]) + del config["renewable_weather_years"] + with pytest.raises(CfNotAvailableError) as excinfo: + validate_godeeep_cf_config(config) + assert "renewable_weather_years" in str(excinfo.value) + + +def test_validate_aggregates_every_problem_into_one_error(): + config = make_config( + [ZENODO_SOURCE], + godeeep_wind_height="_80m", + renewable_weather_years=[1999], + ) + del config["renewable_scenarios"] + with pytest.raises(CfNotAvailableError) as excinfo: + validate_godeeep_cf_config(config) + message = str(excinfo.value) + assert "_80m" in message + assert "renewable_scenarios" in message + assert "problem(s)" in message + # Numbered bullets: more than one failure survived into the single report. + assert "(1)" in message and "(2)" in message + + +def test_validate_803_regression_100m_against_zenodo_only_registry(): + config = make_config( + [ZENODO_SOURCE], + godeeep_wind_height="_100m", + renewable_weather_years=[2012], + ) + with pytest.raises(CfNotAvailableError) as excinfo: + validate_godeeep_cf_config(config) + message = str(excinfo.value) + assert "wind_100m_historical_compressed" in message + assert "2012" in message + + +def test_validate_checks_every_planning_horizon_of_a_future_scenario(): + config = make_config( + [ZENODO_SOURCE], + renewable_scenarios=["rcp85cooler"], + scenario={"planning_horizons": [2030, 2040, 2060]}, + ) + with pytest.raises(CfNotAvailableError) as excinfo: + validate_godeeep_cf_config(config) + message = str(excinfo.value) + assert "2060" in message + # 2030/2040 resolve, so they must not be reported as problems. + assert message.count("2060") >= 1 + assert "problem(s)" in message + + +def test_validate_skips_non_godeeep_datasets(): + config = make_config([ZENODO_SOURCE], renewable_weather_years=[1999]) + config["renewable"] = {"dataset": "atlite"} + assert validate_godeeep_cf_config(config) == [] diff --git a/workflow/scripts/test/test_zenodo_downloader.py b/workflow/scripts/test/test_zenodo_downloader.py new file mode 100644 index 00000000..6e3d1b07 --- /dev/null +++ b/workflow/scripts/test/test_zenodo_downloader.py @@ -0,0 +1,257 @@ +"""Unit tests for the Zenodo downloader's fail-loud contract. + +The CF retrieval registry has zero fallback paths: a bad dataset key, a missing +file inside a record, an exhausted retry budget or a checksum mismatch must all +raise ``ZenodoDownloadError`` with an actionable message — never return ``None`` +and never leave a truncated artifact on disk for a later rule to open. + +Every test monkeypatches ``requests.get``; nothing here touches the network. +""" + +import hashlib +import os +import sys + +import pytest +import requests + +sys.path.append(os.path.join(os.path.dirname(__file__), "..")) + +import zenodo_downloader as zd +from zenodo_downloader import ZenodoDownloadError, ZenodoScenarioDownloader + +pytestmark = pytest.mark.fast + +RECORD_ID = 20316475 +FILENAME = "caps_solar_reference.nc" +PAYLOAD = b"netcdf-bytes" * 64 + + +def md5(payload: bytes) -> str: + return f"md5:{hashlib.md5(payload).hexdigest()}" + + +def record_metadata(files=None): + """A Zenodo record manifest shaped like the real API response.""" + files = files if files is not None else [(FILENAME, PAYLOAD), ("avail_solar_reference.nc", b"other")] + return { + "files": [ + { + "key": key, + "size": len(content), + "checksum": md5(content), + "links": {"self": f"https://zenodo.org/api/records/{RECORD_ID}/files/{key}/content"}, + } + for key, content in files + ], + } + + +class FakeResponse: + """The slice of `requests.Response` the downloader actually touches.""" + + def __init__(self, *, json_data=None, content=b"", status_code=200): + self._json = json_data + self.content = content + self.status_code = status_code + self.headers = {"content-length": str(len(content))} + + def raise_for_status(self): + """Raise an HTTPError carrying this response, as requests does.""" + if self.status_code >= 400: + raise requests.exceptions.HTTPError(f"{self.status_code} Server Error", response=self) + + def json(self): + """Return the JSON body, or raise like requests on a non-JSON body.""" + if self._json is None: + raise ValueError("no JSON body") + return self._json + + def iter_content(self, chunk_size=8192): + """Yield the body in chunks, as a streamed response does.""" + for i in range(0, len(self.content), chunk_size): + yield self.content[i : i + chunk_size] + + +@pytest.fixture(autouse=True) +def no_sleep(monkeypatch): + """Record backoff delays instead of sleeping through them.""" + delays = [] + monkeypatch.setattr(zd.time, "sleep", delays.append) + return delays + + +@pytest.fixture +def no_network(monkeypatch): + """Fail the test loudly if any code path reaches requests.get.""" + + def boom(*args, **kwargs): + raise AssertionError(f"unexpected network call: {args} {kwargs}") + + monkeypatch.setattr(zd.requests, "get", boom) + + +def install_fake_get(monkeypatch, *, metadata=None, file_responses=None): + """Route metadata URLs to `metadata` and file URLs through `file_responses`. + + `file_responses` is a list of per-attempt outcomes: either an exception to + raise or a FakeResponse to return. Returns the list of requested URLs so + tests can assert on the attempt count. + """ + metadata = record_metadata() if metadata is None else metadata + file_responses = list(file_responses or []) + calls = [] + + def fake_get(url, **kwargs): + calls.append(url) + if url.endswith(f"/api/records/{RECORD_ID}"): + return FakeResponse(json_data=metadata) + outcome = file_responses.pop(0) if file_responses else FakeResponse(content=PAYLOAD) + if isinstance(outcome, Exception): + raise outcome + return outcome + + monkeypatch.setattr(zd.requests, "get", fake_get) + return calls + + +def file_calls(calls): + return [url for url in calls if not url.endswith(f"/api/records/{RECORD_ID}")] + + +def test_unknown_dataset_key_raises_listing_available_keys(tmp_path, no_network): + downloader = ZenodoScenarioDownloader(download_dir=tmp_path / "data") + + with pytest.raises(ZenodoDownloadError) as excinfo: + downloader.download_scenario_file("solar_rcp45hotter_compressed", "rcp45hotter", FILENAME) + + message = str(excinfo.value) + assert "solar_rcp45hotter_compressed" in message + assert "nrel_exclusion_v1" in message # available keys are named + + +def test_compressed_cf_records_are_no_longer_hardcoded(tmp_path): + downloader = ZenodoScenarioDownloader(download_dir=tmp_path / "data") + assert not [key for key in downloader.scenario_records if key.endswith("_compressed")] + + +def test_missing_file_in_record_raises_listing_available_files(tmp_path, monkeypatch): + install_fake_get(monkeypatch) + downloader = ZenodoScenarioDownloader(download_dir=tmp_path / "data") + + with pytest.raises(ZenodoDownloadError) as excinfo: + downloader.download_to_path(RECORD_ID, "caps_wind_open.nc", tmp_path / "out.nc") + + message = str(excinfo.value) + assert "caps_wind_open.nc" in message + assert FILENAME in message + assert "avail_solar_reference.nc" in message + assert not (tmp_path / "out.nc").exists() + + +def test_transient_errors_are_retried_then_succeed(tmp_path, monkeypatch, no_sleep): + calls = install_fake_get( + monkeypatch, + file_responses=[ + requests.exceptions.ConnectionError("connection reset"), + FakeResponse(status_code=503), + FakeResponse(content=PAYLOAD), + ], + ) + downloader = ZenodoScenarioDownloader(download_dir=tmp_path / "data") + out_path = tmp_path / "out" / "caps.nc" + + result = downloader.download_to_path(RECORD_ID, FILENAME, out_path) + + assert result == str(out_path) + assert out_path.read_bytes() == PAYLOAD + assert len(file_calls(calls)) == 3 + assert no_sleep == [2.0, 4.0] # exponential backoff between the 3 attempts + + +def test_permanent_failure_raises_after_three_attempts(tmp_path, monkeypatch, no_sleep): + calls = install_fake_get( + monkeypatch, + file_responses=[requests.exceptions.ConnectionError("no route to host")] * 3, + ) + downloader = ZenodoScenarioDownloader(download_dir=tmp_path / "data") + out_path = tmp_path / "out" / "caps.nc" + + with pytest.raises(ZenodoDownloadError) as excinfo: + downloader.download_to_path(RECORD_ID, FILENAME, out_path) + + assert "after 3 attempts" in str(excinfo.value) + assert FILENAME in str(excinfo.value) + assert len(file_calls(calls)) == zd.MAX_DOWNLOAD_ATTEMPTS == 3 + assert not out_path.exists() + + +def test_non_transient_http_error_is_not_retried(tmp_path, monkeypatch): + calls = install_fake_get(monkeypatch, file_responses=[FakeResponse(status_code=404)]) + downloader = ZenodoScenarioDownloader(download_dir=tmp_path / "data") + + with pytest.raises(ZenodoDownloadError, match="404"): + downloader.download_to_path(RECORD_ID, FILENAME, tmp_path / "out.nc") + + assert len(file_calls(calls)) == 1 + + +def test_checksum_mismatch_raises_and_removes_partial_file(tmp_path, monkeypatch): + install_fake_get(monkeypatch, file_responses=[FakeResponse(content=b"truncated")]) + downloader = ZenodoScenarioDownloader(download_dir=tmp_path / "data") + out_path = tmp_path / "out" / "caps.nc" + + with pytest.raises(ZenodoDownloadError) as excinfo: + downloader.download_to_path(RECORD_ID, FILENAME, out_path) + + assert "Checksum mismatch" in str(excinfo.value) + assert md5(PAYLOAD) in str(excinfo.value) + assert not out_path.exists() # the truncated body must not survive + + +def test_record_without_checksum_raises_before_downloading(tmp_path, monkeypatch): + metadata = record_metadata() + del metadata["files"][0]["checksum"] + calls = install_fake_get(monkeypatch, metadata=metadata) + downloader = ZenodoScenarioDownloader(download_dir=tmp_path / "data") + + with pytest.raises(ZenodoDownloadError, match="no checksum"): + downloader.download_to_path(RECORD_ID, FILENAME, tmp_path / "out.nc") + + assert file_calls(calls) == [] + + +def test_metadata_fetch_failure_raises(tmp_path, monkeypatch, no_sleep): + def fake_get(url, **kwargs): + raise requests.exceptions.Timeout("read timed out") + + monkeypatch.setattr(zd.requests, "get", fake_get) + downloader = ZenodoScenarioDownloader(download_dir=tmp_path / "data") + + with pytest.raises(ZenodoDownloadError) as excinfo: + downloader.download_to_path(RECORD_ID, FILENAME, tmp_path / "out.nc") + + assert str(RECORD_ID) in str(excinfo.value) + assert "after 3 attempts" in str(excinfo.value) + + +def test_scenario_download_verifies_checksum_and_returns_path(tmp_path, monkeypatch): + install_fake_get(monkeypatch) + download_dir = tmp_path / "data" + downloader = ZenodoScenarioDownloader(download_dir=download_dir) + + result = downloader.download_scenario_file("nrel_exclusion_v1", "historical", FILENAME) + + expected = download_dir / "zenodo" / "historical" / FILENAME + assert result == str(expected) + assert expected.read_bytes() == PAYLOAD + + +def test_existing_file_is_reused_without_network(tmp_path, monkeypatch, no_network): + download_dir = tmp_path / "data" + target = download_dir / "zenodo" / "historical" / FILENAME + target.parent.mkdir(parents=True) + target.write_bytes(PAYLOAD) + downloader = ZenodoScenarioDownloader(download_dir=download_dir) + + assert downloader.download_scenario_file("nrel_exclusion_v1", "historical", FILENAME) == str(target) diff --git a/workflow/scripts/zenodo_downloader.py b/workflow/scripts/zenodo_downloader.py index 653bfc9a..b9df6d08 100644 --- a/workflow/scripts/zenodo_downloader.py +++ b/workflow/scripts/zenodo_downloader.py @@ -1,8 +1,45 @@ -"""Download scenarios from Zenodo.""" +"""Download scenarios from Zenodo. +Every failure path in this module raises :class:`ZenodoDownloadError`. Nothing +returns ``None`` and nothing silently substitutes a different record, file or +partially-downloaded artifact: a caller that gets a path back is guaranteed a +complete file whose checksum matches the one Zenodo publishes for it. + +Record IDs for the GODEEEP ``*_compressed.nc`` capacity factors are no longer +hardcoded here — they are declared in the ``godeeep_cf_registry`` config block +and reach this module through ``download_to_path``. +""" + +import time from pathlib import Path import requests +from _helpers import validate_checksum + +# Transient-failure retry policy for both metadata and file requests. +MAX_DOWNLOAD_ATTEMPTS = 3 +BACKOFF_BASE_SECONDS = 2.0 +METADATA_TIMEOUT_SECONDS = 30 +# (connect, read) — the read budget has to cover a slow chunk, not the file. +DOWNLOAD_TIMEOUT_SECONDS = (10, 120) +RETRYABLE_STATUS_CODES = frozenset({429, 500, 502, 503, 504}) + + +class ZenodoDownloadError(Exception): + """Raised when a Zenodo record, file or download cannot be resolved.""" + + +def _is_transient(exc: Exception) -> bool: + """True for errors worth retrying (connection resets, timeouts, 5xx/429).""" + if isinstance( + exc, + requests.exceptions.Timeout | requests.exceptions.ConnectionError | requests.exceptions.ChunkedEncodingError, + ): + return True + if isinstance(exc, requests.exceptions.HTTPError): + status = getattr(exc.response, "status_code", None) + return status is not None and (status >= 500 or status in RETRYABLE_STATUS_CODES) + return False class ZenodoScenarioDownloader: @@ -10,47 +47,47 @@ class ZenodoScenarioDownloader: def __init__(self, download_dir="./data"): self.download_dir = Path(download_dir) - self.download_dir.mkdir(exist_ok=True) + self.download_dir.mkdir(parents=True, exist_ok=True) # NREL land-access availability + bus-capacity artifacts; one record # holds all 48 avail_*.nc / caps_*.nc files. - # GODEEEP _compressed.nc records used by the NREL land-access path. - # Keyed by (tech + wind_height + scenario) — no year_range — because - # one record holds all years per (tech, scenario). self.scenario_records = { "nrel_exclusion_v1": 20316475, - "solar_historical_compressed": 20127513, - "solar_rcp45hotter_compressed": 20127523, - "solar_rcp45cooler_compressed": 20127562, - "solar_rcp85hotter_compressed": 20127589, - "solar_rcp85cooler_compressed": 20127633, - "wind_125m_historical_compressed": 20127520, - "wind_125m_rcp45hotter_compressed": 20127545, - "wind_125m_rcp45cooler_compressed": 20127572, - "wind_125m_rcp85hotter_compressed": 20127604, - "wind_125m_rcp85cooler_compressed": 20127645, } # Cache for record metadata to avoid repeated API calls self._metadata_cache = {} def get_record_metadata(self, record_id): - """Get metadata for a record (with caching).""" + """ + Get metadata for a record (with caching). + + Raises + ------ + ZenodoDownloadError + If the record cannot be fetched or does not parse as JSON. + """ if record_id in self._metadata_cache: return self._metadata_cache[record_id] url = f"https://zenodo.org/api/records/{record_id}" + what = f"metadata for Zenodo record {record_id} ({url})" + + for attempt in range(1, MAX_DOWNLOAD_ATTEMPTS + 1): + try: + response = requests.get(url, timeout=METADATA_TIMEOUT_SECONDS) + response.raise_for_status() + break + except requests.exceptions.RequestException as e: + self._retry_or_raise(e, attempt, what) try: - response = requests.get(url) - response.raise_for_status() metadata = response.json() - self._metadata_cache[record_id] = metadata - return metadata + except ValueError as e: + raise ZenodoDownloadError(f"Zenodo returned a non-JSON response for {what}: {e}") from e - except requests.exceptions.RequestException as e: - print(f"Failed to get metadata for record {record_id}: {e}") - return None + self._metadata_cache[record_id] = metadata + return metadata def download_scenario_file(self, scenario_final, scenario, filename, force_redownload=False): """ @@ -59,37 +96,47 @@ def download_scenario_file(self, scenario_final, scenario, filename, force_redow Parameters ---------- scenario_final : str - Lookup key into scenario_records (e.g. "solar_rcp45hotter_compressed"). + Lookup key into scenario_records (e.g. "nrel_exclusion_v1"). scenario : str Climate-scenario name used to choose the on-disk subdir (e.g. "historical", "rcp45hotter"). filename : str - Name of the file to download, e.g., "solar_gen_cf_2030_compressed.nc". + Name of the file to download, e.g., "caps_solar_reference.nc". force_redownload : bool, optional If True, re-download the file even if it exists locally. Default is False. + + Returns + ------- + str + Path to the complete, checksum-verified local file. + + Raises + ------ + ZenodoDownloadError + If `scenario_final` is not a known key, if `filename` is not in the + record, or if the download fails or fails checksum validation. """ - (self.download_dir / "zenodo" / scenario).mkdir(exist_ok=True) - local_filepath = f"{self.download_dir}/zenodo/{scenario}/{filename}" + # Validated up front, even when the file is already on disk: an unknown + # key is a config bug and must never resolve to a cached artifact. + try: + record_id = self.scenario_records[scenario_final] + except KeyError as e: + available = ", ".join(sorted(self.scenario_records)) or "" + raise ZenodoDownloadError( + f"Unknown Zenodo dataset key '{scenario_final}'. Available keys: {available}.", + ) from e + + local_filepath = self.download_dir / "zenodo" / scenario / filename + local_filepath.parent.mkdir(parents=True, exist_ok=True) # Check if file already exists locally and skip Zenodo - if Path(local_filepath).exists() and not force_redownload: + if local_filepath.exists() and not force_redownload: print( f"File already exists locally: {local_filepath}. Skipping download. Use force_redownload=True to re-download.", ) return str(local_filepath) - # Only check record_id if we need to download - else: - record_id = self.scenario_records.get(scenario_final) - - if not record_id: - print(f"No record ID found for scenario: {scenario_final}") - print("Available scenarios with record IDs:") - for scenario, rec_id in self.scenario_records.items(): - if rec_id is not None: - print(f" - {scenario} (ID: {rec_id})") - return None - return self._download_file(record_id, filename, Path(local_filepath), force_redownload) + return self._download_file(record_id, filename, local_filepath, force_redownload) def download_to_path(self, record_id, filename, out_path, force_redownload=False): """ @@ -108,6 +155,17 @@ def download_to_path(self, record_id, filename, out_path, force_redownload=False record's manifest). out_path : str or Path Destination path. Parent directories are created if missing. + + Returns + ------- + str + Path to the complete, checksum-verified local file. + + Raises + ------ + ZenodoDownloadError + If `filename` is not in the record, or if the download fails or + fails checksum validation. """ out_path = Path(out_path) if out_path.exists() and not force_redownload: @@ -121,62 +179,93 @@ def _download_file(self, record_id, filename, local_filepath, force_redownload=F This is only called after confirming the file doesn't exist locally. """ - # Ensure directory exists + local_filepath = Path(local_filepath) local_filepath.parent.mkdir(parents=True, exist_ok=True) - # Get record metadata metadata = self.get_record_metadata(record_id) - if not metadata: - return None + record_files = metadata.get("files", []) # Find the specific file - target_file = None - for file_info in metadata.get("files", []): - if file_info["key"] == filename: - target_file = file_info - break + target_file = next((f for f in record_files if f["key"] == filename), None) + if target_file is None: + available = ", ".join(sorted(f["key"] for f in record_files)) or "" + raise ZenodoDownloadError( + f"File '{filename}' not found in Zenodo record {record_id}. Available files: {available}.", + ) - if not target_file: - print(f"File '{filename}' not found in record {record_id}") - print("Available files:") - for file_info in metadata.get("files", []): - print(f" - {file_info['key']}") - return None + # Checked before spending bandwidth: an unverifiable download is a + # failure, not something to wave through. + checksum = target_file.get("checksum") + if not checksum: + raise ZenodoDownloadError( + f"Zenodo record {record_id} publishes no checksum for '{filename}'; " + "refusing to download a file whose integrity cannot be verified.", + ) - # Download the file download_url = target_file["links"]["self"] - file_size_mb = target_file["size"] / (1024 * 1024) - print(f"Downloading {filename} from record {record_id}...") - print(f"Size: {file_size_mb:.1f} MB") + if target_file.get("size"): + print(f"Size: {target_file['size'] / (1024 * 1024):.1f} MB") print(f"Saving to: {local_filepath}") - try: - response = requests.get(download_url, stream=True) - response.raise_for_status() - - total_size = int(response.headers.get("content-length", 0)) - downloaded_size = 0 - - with open(local_filepath, "wb") as f: - for chunk in response.iter_content(chunk_size=8192): - if chunk: - f.write(chunk) - downloaded_size += len(chunk) - - # Show progress for large files - if total_size > 10 * 1024 * 1024: # Show progress for files > 10MB - progress = (downloaded_size / total_size) * 100 - print(f"\rProgress: {progress:.1f}%", end="", flush=True) - - if total_size > 10 * 1024 * 1024: - print() # New line after progress - - print(f"Successfully downloaded {filename}") - return str(local_filepath) + what = f"'{filename}' from Zenodo record {record_id} ({download_url})" + for attempt in range(1, MAX_DOWNLOAD_ATTEMPTS + 1): + try: + response = requests.get(download_url, stream=True, timeout=DOWNLOAD_TIMEOUT_SECONDS) + response.raise_for_status() + self._write_stream(response, local_filepath) + break + except requests.exceptions.RequestException as e: + local_filepath.unlink(missing_ok=True) # never keep a torn file + self._retry_or_raise(e, attempt, what) - except requests.exceptions.RequestException as e: - print(f"Download failed: {e}") - if Path(local_filepath).exists(): - Path(local_filepath).unlink() # Remove partial file - return None + try: + validate_checksum(local_filepath, checksum=checksum) + except (AssertionError, ValueError) as e: + local_filepath.unlink(missing_ok=True) + raise ZenodoDownloadError( + f"Checksum mismatch for {what}: expected {checksum}. " + f"The partial or corrupt file {local_filepath} has been removed; re-run the rule.", + ) from e + + print(f"Successfully downloaded {filename}") + return str(local_filepath) + + @staticmethod + def _write_stream(response, local_filepath): + """Stream a response body to disk, reporting progress on large files.""" + total_size = int(response.headers.get("content-length", 0)) + downloaded_size = 0 + + with open(local_filepath, "wb") as f: + for chunk in response.iter_content(chunk_size=8192): + if chunk: + f.write(chunk) + downloaded_size += len(chunk) + + # Show progress for large files + if total_size > 10 * 1024 * 1024: # Show progress for files > 10MB + progress = (downloaded_size / total_size) * 100 + print(f"\rProgress: {progress:.1f}%", end="", flush=True) + + if total_size > 10 * 1024 * 1024: + print() # New line after progress + + @staticmethod + def _retry_or_raise(exc, attempt, what): + """Sleep before the next attempt, or raise once retries are exhausted. + + Non-transient failures (404, 403, ...) are raised immediately — retrying + them would only delay the error. + """ + if not _is_transient(exc): + raise ZenodoDownloadError(f"Failed to fetch {what}: {exc}") from exc + if attempt >= MAX_DOWNLOAD_ATTEMPTS: + raise ZenodoDownloadError( + f"Failed to fetch {what} after {MAX_DOWNLOAD_ATTEMPTS} attempts: {exc}", + ) from exc + delay = BACKOFF_BASE_SECONDS * 2 ** (attempt - 1) + print( + f"Transient error fetching {what} (attempt {attempt}/{MAX_DOWNLOAD_ATTEMPTS}): {exc}. Retrying in {delay:.0f}s..." + ) + time.sleep(delay)