dataset: add bias-corrected wind speeds from Copernicus

RELEASE_NOTES.rst

@@ -20,6 +20,12 @@ Upcoming Release
 **Features**
 
 * Add support for Python 3.14.
+* Add the ``era5-corrected`` module providing the bias-corrected ([``sis-energy-global-reanalysis``](https://cds.climate.copernicus.eu/datasets/sis-energy-global-reanalysis?tab=overview)) 
+  ERA5 100 m wind speed as ``wnd100m_corrected`` and ``wnd100m_corrected_extrapolated``, plus the power law coefficients as
+  ``wnd_shear_exp_corrected`` (a month-by-hour climatology, fetched in a separate request). 
+  Assign the preferred variables to ``cutout.data["wnd100m"]`` and
+  ``cutout.data["wnd_shear_exp"]`` to use them in ``cutout.wind``. Combine with ``era5`` for ``roughness``
+  (logarithmic extrapolation).
 
 **Bug fixes**
 

atlite/data.py

@@ -5,8 +5,10 @@
 
 from __future__ import annotations
 
+import gc
 import logging
 import os
+import time
 from functools import wraps
 from pathlib import Path
 from shutil import rmtree
@@ -32,6 +34,19 @@
 logger = logging.getLogger(__name__)
 
 
+def _rename_with_retry(source: Path, target: Path, attempts: int = 20) -> None:
+    """Rename a file, retrying while Windows still holds handles on it."""
+    for attempt in range(attempts):
+        try:
+            source.rename(target)
+            return
+        except PermissionError:
+            if attempt == attempts - 1:
+                raise
+            gc.collect()
+            time.sleep(0.25)
+
+
 def get_features(
     cutout: Cutout,
     module: str,
@@ -233,12 +248,12 @@ def cutout_prepare(
     dask_kwargs : dict, default {}
         Dict with keyword arguments passed to `dask.compute`.
     monthly_requests : bool, optional
-        If True, the data is requested on a monthly basis in ERA5. This is useful for
-        large cutouts, where the data is requested in smaller chunks. The
-        default is False
+        If True, ERA5 and ERA5-corrected data are requested on a monthly basis.
+        This is useful for large cutouts, where the data is requested in smaller
+        chunks. The default is False.
     concurrent_requests : bool, optional
-        If True, the monthly data requests are posted concurrently.
-        Only has an effect if `monthly_requests` is True. The default is False.
+        If True, the time-chunked data requests are posted concurrently. The
+        default is False.
 
     Returns
     -------
@@ -319,10 +334,14 @@ def cutout_prepare(
                 write_job.compute(**dask_kwargs)
         else:
             write_job.compute(**dask_kwargs)
+        del write_job
         if cutout.path.exists():
             cutout.data.close()
+            # release lazily opened handles of the old file (Windows cannot
+            # replace files that are still open)
+            gc.collect()
             cutout.path.unlink()
-        Path(tmp).rename(cutout.path)
+        _rename_with_retry(Path(tmp), cutout.path)
 
         cutout.data = xr.open_dataset(cutout.path, chunks=cutout.chunks)
 

atlite/datasets/__init__.py

@@ -4,10 +4,11 @@
 
 """atlite datasets."""
 
-from atlite.datasets import era5, era5_edh, gebco, glofas, sarah
+from atlite.datasets import era5, era5_corrected, era5_edh, gebco, glofas, sarah
 
 modules = {
     "era5": era5,
+    "era5-corrected": era5_corrected,
     "era5-edh": era5_edh,
     "gebco": gebco,
     "glofas": glofas,

atlite/datasets/era5.py

@@ -60,6 +60,7 @@ def nullcontext():  # noqa: D103
 features = {
     "height": ["height"],
     "wind": ["wnd100m", "wnd_shear_exp", "wnd_azimuth", "roughness"],
+    "roughness": ["roughness"],
     "influx": [
         "influx_toa",
         "influx_direct",
@@ -199,6 +200,30 @@ def get_data_wind(retrieval_params: dict[str, Any]) -> xr.Dataset:
     return _process_wind(ds)
 
 
+def get_data_roughness(retrieval_params: dict[str, Any]) -> xr.Dataset:
+    """
+    Retrieve the forecast surface roughness from ERA5.
+
+    Parameters
+    ----------
+    retrieval_params : dict[str, Any]
+        CDS API retrieval parameters including area, time, and format.
+
+    Returns
+    -------
+    xr.Dataset
+        Dataset with variable: roughness.
+    """
+    ds = retrieve_data(variable=["forecast_surface_roughness"], **retrieval_params)
+    ds = _rename_and_clean_coords(ds)
+    return ds.rename({"fsr": "roughness"})
+
+
+def sanitize_roughness(ds: xr.Dataset) -> xr.Dataset:
+    """Clip negative roughness values to a minimum of 2e-4."""
+    return sanitize_wind(ds)
+
+
 def sanitize_wind(ds: xr.Dataset) -> xr.Dataset:
     """
     Clip negative roughness values to a minimum of 2e-4.