Coverage for mlair/helpers/data_sources/era5.py: 9%

1"""Methods to load era5 data.""" 

2__author__ = "Lukas Leufen" 

3__date__ = "2022-06-09" 

4 

5import logging 

6import os 

7from functools import partial 

8 

9import pandas as pd 

10import xarray as xr 

11 

12from mlair import helpers 

13from mlair.configuration.era5_settings import era5_settings 

14from mlair.configuration.toar_data_v2_settings import toar_data_v2_settings 

15from mlair.helpers.data_sources.toar_data_v2 import load_station_information, combine_meta_data, correct_timezone 

16from mlair.helpers.data_sources.data_loader import EmptyQueryResult 

17from mlair.helpers.meteo import relative_humidity_from_dewpoint 

18 

19 

20def load_era5(station_name, stat_var, sampling, data_origin, time_dim, window_dim, target_dim, era5_data_path=None, 

21 era5_file_names=None): 

22 

23 # make sure station_name parameter is a list 

24 station_name = helpers.to_list(station_name) 

25 

26 # get data path 

27 data_path, file_names = era5_settings(sampling, era5_data_path=era5_data_path, era5_file_names=era5_file_names) 

28 

29 # correct stat_var values if data is not aggregated (hourly) 

30 if sampling == "hourly": 

31 stat_var = {key: "values" for key in stat_var.keys()} 

32 else: 

33 raise ValueError(f"Given sampling {sampling} is not supported, only hourly sampling can be used.") 

34 

35 # load station meta using toar-data v2 API 

36 meta_url_base, headers = toar_data_v2_settings("meta") 

37 station_meta = load_station_information(station_name, meta_url_base, headers) 

38 

39 # sel data for station using sel method nearest 

40 logging.info(f"load data for {station_meta['codes'][0]} from ERA5") 

41 try: 

42 lon, lat = station_meta["coordinates"]["lng"], station_meta["coordinates"]["lat"] 

43 file_names = os.path.join(data_path, file_names) 

44 with xr.open_mfdataset(file_names, preprocess=partial(preprocess_era5_single_file, lon, lat)) as data: 

45 station_data = data.to_array().T.compute() 

46 except OSError as e: 

47 logging.info(f"Cannot load era5 data from path {data_path} and filenames {file_names} due to: {e}") 

48 return None, None 

49 

50 if "relhum" in stat_var: 

51 relhum = relative_humidity_from_dewpoint(station_data.sel(variable="D2M"), station_data.sel(variable="T2M")) 

52 station_data = xr.concat([station_data, relhum.expand_dims({"variable": ["RHw"]})], dim="variable") 

53 station_data.coords["variable"] = _rename_era5_variables(station_data.coords["variable"].values) 

54 

55 # check if all requested variables are available 

56 if set(stat_var).issubset(station_data.coords["variable"].values) is False: 

57 missing_variables = set(stat_var).difference(station_data.coords["variable"].values) 

58 origin = helpers.select_from_dict(data_origin, missing_variables) 

59 options = f"station={station_name}, origin={origin}" 

60 raise EmptyQueryResult(f"No data found for variables {missing_variables} and options {options} in JOIN.") 

61 else: 

62 station_data = station_data.sel(variable=list(stat_var.keys())) 

63 

64 # convert to local timezone 

65 station_data.coords["time"] = correct_timezone(station_data.to_pandas(), station_meta, sampling).index 

66 station_data = station_data.rename({"time": time_dim, "variable": target_dim}) 

67 

68 # expand window_dim 

69 station_data = station_data.expand_dims({window_dim: [0]}) 

70 

71 # create meta data 

72 variable_meta = _emulate_meta_data(station_data.coords[target_dim].values) 

73 meta = combine_meta_data(station_meta, variable_meta) 

74 meta = pd.DataFrame.from_dict(meta, orient='index') 

75 meta.columns = station_name 

76 return station_data, meta 

77 

78 

79def preprocess_era5_single_file(lon, lat, ds): 

80 """Select lon and lat from data file and transform valid time into lead time.""" 

81 ds = ds.sel(lon=lon, lat=lat, method="nearest", drop=True) 

82 return ds 

83 

84 

85def _emulate_meta_data(variables): 

86 general_meta = {"sampling_frequency": "hourly", "data_origin": "model", "data_origin_type": "model"} 

87 roles_meta = {"roles": [{"contact": {"organisation": {"name": "ERA5", "longname": "ECMWF"}}}]} 

88 variable_meta = {var: {"variable": {"name": var}, **roles_meta, ** general_meta} for var in variables} 

89 return variable_meta 

90 

91 

92def _rename_era5_variables(era5_names): 

93 mapper = {"SP": "press", "U10M": "u", "V10M": "v", "T2M": "temp", "D2M": "dew", "BLH": "pblheight", 

94 "TCC": "cloudcover", "RHw": "relhum"} 

95 era5_names = list(era5_names) 

96 try: 

97 join_names = list(map(lambda x: mapper[x], era5_names)) 

98 return join_names 

99 except KeyError as e: 

100 raise KeyError(f"Cannot map names from era5 to join naming convention: {e}")