import requests
import pandas as pd
from io import BytesIO, StringIO
import zipfile
import time
VARIABLE_MAP = {
# short names
'd2m': 'temp_dew',
't2m': 'temp_air',
'sp': 'pressure',
'ssrd': 'ghi',
'fdir': 'bhi',
'tp': 'precipitation',
'strd': 'longwave_down',
# long names
'2m_dewpoint_temperature': 'temp_dew',
'2m_temperature': 'temp_air',
'surface_pressure': 'pressure',
'surface_solar_radiation_downwards': 'ghi',
'total_sky_direct_solar_radiation_at_surface': 'bhi',
'total_precipitation': 'precipitation',
'surface_thermal_radiation_downwards': 'longwave_down',
}
def _same(x):
return x
def _k_to_c(temp_k):
return temp_k - 273.15
def _j_to_w(j):
return j / 3600
def _m_to_cm(m):
return m * 100
UNITS = {
'u100': _same,
'v100': _same,
'u10': _same,
'v10': _same,
'd2m': _k_to_c,
't2m': _k_to_c,
'msl': _same,
'sst': _k_to_c,
'skt': _k_to_c,
'sp': _same,
'ssrd': _j_to_w,
'fdir': _j_to_w,
'strd': _j_to_w,
'tp': _m_to_cm,
}
[docs]
def get_era5(latitude, longitude, start, end, variables, api_key,
dataset="reanalysis-era5-single-levels-timeseries",
map_variables=True, timeout=60,
url='https://cds.climate.copernicus.eu/api/retrieve/v1/'):
"""
Retrieve ERA5 reanalysis data from the ECMWF's Copernicus Data Store.
A CDS API key is needed to access this API. Register for one at [1]_.
This API [2]_ provides a subset of parameters of the full ERA5 datasets,
see [3]_ for available variables. A comparison of ERA5 and ERA5-land is
available in [4]_.
Parameters
----------
latitude : float
In decimal degrees, north is positive (ISO 19115).
longitude: float
In decimal degrees, east is positive (ISO 19115).
start : datetime like or str
First day of the requested period. Assumed to be UTC if not localized.
end : datetime like or str
Last day of the requested period. Assumed to be UTC if not localized.
variables : list of str
List of variable names to retrieve, for example
``['ghi', 'bhi', 'temp_air']``. Both pvlib and ERA5 names can be used.
See [1]_ for additional options.
api_key : str
ECMWF CDS API key.
dataset : str, default ``"reanalysis-era5-single-levels-timeseries"``
The dataset to query. May be either
``"reanalysis-era5-single-levels-timeseries"`` or
``"reanalysis-era5-land-timeseries"``. See [4]_ for details.
map_variables : bool, default True
When true, renames columns of the DataFrame to pvlib variable names
where applicable. Also converts units of some variables. See variable
:const:`VARIABLE_MAP` and :const:`UNITS`.
timeout : int, default 60
Number of seconds to wait for the requested data to become available
before timeout.
url : str, optional
API endpoint URL.
Raises
------
Exception
If ``timeout`` is reached without the job finishing.
Returns
-------
data : pd.DataFrame
Time series data. The index corresponds to the end of the interval.
meta : dict
Metadata.
Notes
-----
Data are returned for the grid point nearest to the requested location.
The ERA5 dataset is available on a 0.25° x 0.25° grid, whereas ERA5-Land
is available on a finer 0.1° x 0.1° grid [4]_.
ERA5-Land only covers land surfaces. For locations near the coast, the
nearest grid point may lie over water, in which case ERA5-Land returns
only missing values (NaN) without raising an error. Similarly, for ERA5
the nearest grid point may be over water and therefore not
representative of conditions on land, e.g., for air temperature. Check
the returned ``meta['latitude']`` and ``meta['longitude']``, and
consider adjusting the requested coordinates slightly inland.
References
----------
.. [1] https://cds.climate.copernicus.eu/
.. [2] https://cds.climate.copernicus.eu/datasets/reanalysis-era5-single-levels-timeseries?tab=overview
.. [3] https://confluence.ecmwf.int/pages/viewpage.action?pageId=505390919
.. [4] https://confluence.ecmwf.int/display/CKB/The+family+of+ERA5+datasets
""" # noqa: E501
def _to_utc_dt_notz(dt):
dt = pd.to_datetime(dt)
if dt.tzinfo is not None:
dt = dt.tz_convert("UTC")
return dt
start = _to_utc_dt_notz(start).strftime("%Y-%m-%d")
end = _to_utc_dt_notz(end).strftime("%Y-%m-%d")
headers = {'PRIVATE-TOKEN': api_key}
# allow variables to be specified with pvlib names
reverse_map = {v: k for k, v in VARIABLE_MAP.items()}
variables = [reverse_map.get(k, k) for k in variables]
# Step 1: submit data request (add it to the queue)
params = {
"inputs": {
"variable": variables,
"location": {"longitude": longitude, "latitude": latitude},
"date": [f"{start}/{end}"],
"data_format": "csv"
}
}
slug = f"processes/{dataset}/execution"
response = requests.post(url + slug, json=params, headers=headers,
timeout=timeout)
submission_response = response.json()
if not response.ok:
raise Exception(submission_response) # likely need to accept license
job_id = submission_response['jobID']
# Step 2: poll until the data request is ready
slug = "jobs/" + job_id
poll_interval = 1
num_polls = 0
while True:
response = requests.get(url + slug, headers=headers, timeout=timeout)
poll_response = response.json()
job_status = poll_response['status']
if job_status == 'successful':
break # ready to proceed to next step
elif job_status == 'failed':
msg = (
'Request failed. Please check the ECMWF website for details: '
'https://cds.climate.copernicus.eu/requests?tab=all'
)
raise Exception(msg)
num_polls += 1
if num_polls * poll_interval > timeout:
raise requests.exceptions.Timeout(
'Request timed out. Try increasing the timeout parameter or '
'reducing the request size.'
)
time.sleep(1)
# Step 3: get the download link for our requested dataset
slug = "jobs/" + job_id + "/results"
response = requests.get(url + slug, headers=headers, timeout=timeout)
results_response = response.json()
download_url = results_response['asset']['value']['href']
# Step 4: finally, download our dataset. it's a zipfile of CSVs; some
# datasets (e.g., ERA5-Land) return one CSV per group of parameters
response = requests.get(download_url, timeout=timeout)
zipbuffer = BytesIO(response.content)
archive = zipfile.ZipFile(zipbuffer)
filenames = [file.filename for file in archive.filelist]
dfs = []
for filename in filenames:
csvbuffer = StringIO(archive.read(filename).decode('utf-8'))
dfs.append(pd.read_csv(csvbuffer))
# and parse into the usual formats
metadata = submission_response['metadata'] # include messages from ECMWF
metadata['jobID'] = job_id
if not dfs[0].empty:
metadata['latitude'] = dfs[0]['latitude'].values[0]
metadata['longitude'] = dfs[0]['longitude'].values[0]
for i, df in enumerate(dfs):
df.index = pd.to_datetime(df['valid_time']).dt.tz_localize('UTC')
dfs[i] = df.drop(columns=['valid_time', 'latitude', 'longitude'])
df = pd.concat(dfs, axis=1)
if map_variables:
# convert units and rename
for shortname in df.columns:
converter = UNITS.get(shortname, _same)
df[shortname] = converter(df[shortname])
df = df.rename(columns=VARIABLE_MAP)
return df, metadata
