Skip to content

Commit c428bee

Browse files
mrmorawskifneumpre-commit-ci[bot]maurerleFabianHofmann
authored
Add alternative source for ERA5 data to eliminate queueing (#507)
* initial attemtp at opendap access * smarter CE plus semaphore to limit no of simultaneous connections * add caching * add deps for ncar * zarr instead of netcdf for cahing to improve parallelism, tests to compare against existing era5 cutout * test setup simplified, minor fix to downloaded coords * first pass at spatial subsetting * processing for solar * first attempt at regrid * regridding fixed * processing for remaining values implemented, passes tests at native resolution * merged tests for era5_ncar * precomputed time axis to save on fetches, parallel building, skip interpolation if grids are aligned, grid interpolation sanitise only if interpolatingfor consistency with cds * rate limit downloads through a custom dask pool instead of semaphore * todos after code review * style updated for consistency with era5.py, type annotations * added units to missing vars * invariant import handling simplified@ * per-variable tolerances for weird resolution cutout * numpy style docstrings * add time sampling tests * first pass at era5_ncar doc * finished docs * suppress pydap log, add regridding warning * updated comments + reorganised function - update incomplete * use circular azimuth error in tests, better zero meridian handling, comments * fix documentation typos * clearer comment structure, fix for edge case when retrieving data around 0 longitude index * add release note * move release note to correct place * clean up comments in era5_ncar tests * consistent formatting of log output in era5_ncar, suppress known warnings * clean up doc notebook output * more efficient zarr cache chunking, minor refactoring of logging setup * add all atlite vars downloaded from edh plus tests * update docs for EDH model * customizable chunk size support, cast all vars to float32 to save on space * udpate test tols for compatibility with float32 cast * updated documentation in notebooks * clearer comments * retries on failed chunk downloads, minor refactors * docs improvements * fix typo in test docstring Co-authored-by: Fabian Neumann <fabian.neumann@outlook.de> * replace fsspecstore with obstore for more robust connection * add EDH key to test CI * bump python version to support zarr v3 * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * update classifiers in pyproject to remove 3.10 * remove unnecessary tempdir from test on edh * performance optimisation to reduce ram usage. more sensible chunk size, reverse subset order, better retry config * rechunk for memory use optimisation, add example on using distributed scheduler to download large cutouts * fix example doc to render properly with pandoc * fix CI runners getting bounced from EDH due to multiple machines attempting to reach it at the same time with same creds * fix type errors in era5-edh * fix types and docstrings to match new ruff rules * move era5-edh to an upcoming release in release notes * refactor: reduce test retry timeout to fail faster in case of edh server/credential issue * chore: remove typing_extensions dependency post upgrade to python 3.11 * docs: fix typos in example and installation instructions Co-authored-by: Florian Maurer <fmaurer@disroot.org> * refactor: restore long edh reach retry times to deal with EDH timing out in CI, add a warning in the CI test job 9x runners simultaneously hammer EDH with the same API keys, which results in 403s. for now dealt with via long timeouts, final fix would be a two-step test process, separate creds, or some other form of rate limiting * refactor(era5-edh): share wind/influx processing with era5, add offline tests - Extract _process_wind/_process_influx helpers in era5.py (single_precision flag) - Add offline tests for resolution guard and credential resolution - Parametrize era5-edh comparison tests; fix _edh_reachable auth handling - Remove redundant comments; document ignored CDS-only get_data params * docs: add era5-edh usage note to release notes * refactor: edh reachability test retries on 429s sometimes many CI runners hit EDH at the same time, which results in 429s and rate limiting. retries are useful in that case --------- Co-authored-by: Fabian Neumann <fabian.neumann@outlook.de> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Florian Maurer <fmaurer@disroot.org> Co-authored-by: Fabian <fab.hof@gmx.de>
1 parent b8d9e4d commit c428bee

19 files changed

Lines changed: 1506 additions & 67 deletions

.github/workflows/test.yaml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ jobs:
5555
MPLBACKEND: Agg # https://github.com/orgs/community/discussions/26434
5656
CDSAPI_URL: ${{ vars.CDSAPI_URL }}
5757
CDSAPI_KEY: ${{ secrets.CDSAPI_TOKEN }}
58+
EARTHDATAHUB_API_KEY: ${{ secrets.EARTHDATAHUB_API_KEY}}
5859

5960
steps:
6061
- name: Setup cache and secrets (Linux & MacOS)

RELEASE_NOTES.rst

100755100644
Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,11 @@ Upcoming Release
2424
buffer that computes a geometrically accurate Euclidean buffer, correct in
2525
diagonal directions.
2626

27+
* Add an Earth Data Hub (EDH) source for ERA5 as a new dataset module ``era5-edh``,
28+
which serves data without the CDS processing queue. Configure a DestinE API key
29+
(see :doc:`installation`) and pass ``module="era5-edh"`` when creating a cutout.
30+
Only the native 0.25°×0.25° grid is supported; use ``module="era5"`` (CDS) for
31+
other resolutions.
2732
* Implement glofas dataset which contains daily river discharge. ``cutout.hydro()`` now
2833
returns discharge if ``cutout.module`` contains ``"glofas"``.
2934
(https://github.com/PyPSA/atlite/pull/498)

atlite/datasets/__init__.py

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,12 @@
44

55
"""atlite datasets."""
66

7-
from atlite.datasets import era5, gebco, glofas, sarah
7+
from atlite.datasets import era5, era5_edh, gebco, glofas, sarah
88

9-
modules = {"era5": era5, "sarah": sarah, "gebco": gebco, "glofas": glofas}
9+
modules = {
10+
"era5": era5,
11+
"era5-edh": era5_edh,
12+
"gebco": gebco,
13+
"glofas": glofas,
14+
"sarah": sarah,
15+
}

atlite/datasets/era5.py

Lines changed: 77 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121
import pandas as pd
2222
import xarray as xr
2323
from dask import compute, delayed
24-
from dask.array import arctan2, sqrt
2524
from numpy import atleast_1d
2625

2726
from atlite.datasets.cds_helper import (
@@ -133,6 +132,40 @@ def _rename_and_clean_coords(ds: xr.Dataset, add_lon_lat: bool = True) -> xr.Dat
133132
return ds.drop_vars(["expver", "number"], errors="ignore")
134133

135134

135+
def _process_wind(ds: xr.Dataset, single_precision: bool = False) -> xr.Dataset:
136+
"""
137+
Derive wind speed, shear exponent, azimuth and roughness from raw components.
138+
139+
Shared by the CDS (:mod:`atlite.datasets.era5`) and EDH
140+
(:mod:`atlite.datasets.era5_edh`) backends. Operates on a dataset carrying
141+
the raw ``u10``/``v10``/``u100``/``v100``/``fsr`` variables.
142+
``single_precision`` casts the float64-promoted shear and azimuth back to
143+
float32 (used by EDH to halve on-disk size).
144+
145+
Returns
146+
-------
147+
xr.Dataset
148+
Dataset with variables: wnd100m, wnd_shear_exp, wnd_azimuth, roughness.
149+
"""
150+
for h in (10, 100):
151+
units = ds[f"u{h}"].attrs.get("units", "m s**-1")
152+
ds[f"wnd{h}m"] = np.sqrt(ds[f"u{h}"] ** 2 + ds[f"v{h}"] ** 2).assign_attrs(
153+
units=units, long_name=f"{h} metre wind speed"
154+
)
155+
shear = (np.log(ds["wnd10m"] / ds["wnd100m"]) / np.log(10 / 100)).assign_attrs(
156+
units="", long_name="wind shear exponent"
157+
)
158+
ds["wnd_shear_exp"] = shear.astype(np.float32) if single_precision else shear
159+
160+
# span the whole circle: 0 is north, π/2 is east, -π is south, 3π/2 is west
161+
azimuth = np.arctan2(ds["u100"], ds["v100"])
162+
azimuth = azimuth.where(azimuth >= 0, azimuth + 2 * np.pi)
163+
ds["wnd_azimuth"] = azimuth.astype(np.float32) if single_precision else azimuth
164+
165+
ds = ds.drop_vars(["u100", "v100", "u10", "v10", "wnd10m"])
166+
return ds.rename({"fsr": "roughness"})
167+
168+
136169
def get_data_wind(retrieval_params: dict[str, Any]) -> xr.Dataset:
137170
"""
138171
Retrieve and compute wind speed variables from ERA5.
@@ -161,21 +194,7 @@ def get_data_wind(retrieval_params: dict[str, Any]) -> xr.Dataset:
161194
**retrieval_params,
162195
)
163196
ds = _rename_and_clean_coords(ds)
164-
165-
for h in [10, 100]:
166-
ds[f"wnd{h}m"] = sqrt(ds[f"u{h}"] ** 2 + ds[f"v{h}"] ** 2).assign_attrs(
167-
units=ds[f"u{h}"].attrs["units"], long_name=f"{h} metre wind speed"
168-
)
169-
ds["wnd_shear_exp"] = (
170-
np.log(ds["wnd10m"] / ds["wnd100m"]) / np.log(10 / 100)
171-
).assign_attrs(units="", long_name="wind shear exponent")
172-
173-
# span the whole circle: 0 is north, π/2 is east, -π is south, 3π/2 is west
174-
azimuth = arctan2(ds["u100"], ds["v100"])
175-
ds["wnd_azimuth"] = azimuth.where(azimuth >= 0, azimuth + 2 * np.pi)
176-
177-
ds = ds.drop_vars(["u100", "v100", "u10", "v10", "wnd10m"])
178-
return ds.rename({"fsr": "roughness"})
197+
return _process_wind(ds)
179198

180199

181200
def sanitize_wind(ds: xr.Dataset) -> xr.Dataset:
@@ -196,36 +215,22 @@ def sanitize_wind(ds: xr.Dataset) -> xr.Dataset:
196215
return ds
197216

198217

199-
def get_data_influx(retrieval_params: dict[str, Any]) -> xr.Dataset:
218+
def _process_influx(ds: xr.Dataset, single_precision: bool = False) -> xr.Dataset:
200219
"""
201-
Retrieve and compute solar radiation variables from ERA5.
220+
Derive influx variables and solar position from raw radiation fields.
202221
203-
Downloads radiation components, converts from J/m² to W/m², computes
204-
albedo, diffuse radiation, and solar position (altitude/azimuth).
205-
206-
Parameters
207-
----------
208-
retrieval_params : dict[str, Any]
209-
CDS API retrieval parameters including area, time, and format.
222+
Shared by the CDS (:mod:`atlite.datasets.era5`) and EDH
223+
(:mod:`atlite.datasets.era5_edh`) backends. Operates on a dataset carrying
224+
the raw ``ssrd``/``ssr``/``fdir``/``tisr`` variables. ``single_precision``
225+
casts the solar-position fields back to float32 (used by EDH to halve
226+
on-disk size).
210227
211228
Returns
212229
-------
213230
xr.Dataset
214231
Dataset with variables: influx_toa, influx_direct, influx_diffuse,
215232
albedo, solar_altitude, solar_azimuth.
216233
"""
217-
ds = retrieve_data(
218-
variable=[
219-
"surface_net_solar_radiation",
220-
"surface_solar_radiation_downwards",
221-
"toa_incident_solar_radiation",
222-
"total_sky_direct_solar_radiation_at_surface",
223-
],
224-
**retrieval_params,
225-
)
226-
227-
ds = _rename_and_clean_coords(ds)
228-
229234
ds = ds.rename({"fdir": "influx_direct", "tisr": "influx_toa"})
230235
ds["albedo"] = (
231236
((ds["ssrd"] - ds["ssr"]) / ds["ssrd"].where(ds["ssrd"] != 0))
@@ -249,13 +254,46 @@ def get_data_influx(retrieval_params: dict[str, Any]) -> xr.Dataset:
249254
# Suppress DeprecationWarning from new SolarPosition calculation (#199)
250255
with warnings.catch_warnings():
251256
warnings.simplefilter("ignore", DeprecationWarning)
252-
time_shift = pd.to_timedelta("-30 minutes")
253-
sp = SolarPosition(ds, time_shift=time_shift)
257+
sp = SolarPosition(ds, time_shift=pd.to_timedelta("-30 minutes"))
254258
sp = sp.rename({v: f"solar_{v}" for v in sp.data_vars})
259+
if single_precision:
260+
sp = sp.astype(np.float32)
255261

256262
return xr.merge([ds, sp])
257263

258264

265+
def get_data_influx(retrieval_params: dict[str, Any]) -> xr.Dataset:
266+
"""
267+
Retrieve and compute solar radiation variables from ERA5.
268+
269+
Downloads radiation components, converts from J/m² to W/m², computes
270+
albedo, diffuse radiation, and solar position (altitude/azimuth).
271+
272+
Parameters
273+
----------
274+
retrieval_params : dict[str, Any]
275+
CDS API retrieval parameters including area, time, and format.
276+
277+
Returns
278+
-------
279+
xr.Dataset
280+
Dataset with variables: influx_toa, influx_direct, influx_diffuse,
281+
albedo, solar_altitude, solar_azimuth.
282+
"""
283+
ds = retrieve_data(
284+
variable=[
285+
"surface_net_solar_radiation",
286+
"surface_solar_radiation_downwards",
287+
"toa_incident_solar_radiation",
288+
"total_sky_direct_solar_radiation_at_surface",
289+
],
290+
**retrieval_params,
291+
)
292+
293+
ds = _rename_and_clean_coords(ds)
294+
return _process_influx(ds)
295+
296+
259297
def sanitize_influx(ds: xr.Dataset) -> xr.Dataset:
260298
"""
261299
Clip negative radiation values to zero.

0 commit comments

Comments
 (0)