Coverage for mlair/reference_models/reference_model_cams.py: 0%

1__author__ = "Lukas Leufen" 

2__date__ = "2022-06-27" 

3 

4 

5from mlair.configuration.path_config import check_path_and_create 

6from mlair.reference_models.abstract_reference_model import AbstractReferenceModel 

7import os 

8import xarray as xr 

9import pandas as pd 

10 

11 

12class CAMSforecast(AbstractReferenceModel): 

13 

14 def __init__(self, ref_name: str, ref_store_path: str = None, data_path: str = None, interp_method: str = None): 

15 """ 

16 Use parameters `cams_data_path` to set `data_path` and `cams_interp_method` to set `interp_method` in MLAir 

17 run script. 

18 

19 :param ref_name: 

20 :param ref_store_path: 

21 :param data_path: 

22 :param interp_method: 

23 """ 

24 

25 super().__init__() 

26 self.ref_name = ref_name 

27 if ref_store_path is None: 

28 ref_store_path = f"{self.ref_name}/" 

29 self.ref_store_path = ref_store_path 

30 if data_path is None: 

31 self.data_path = os.path.abspath(".") 

32 else: 

33 self.data_path = os.path.abspath(data_path) 

34 self.interp_method = interp_method 

35 self.file_pattern = "forecasts_%s_test.nc" 

36 self.time_dim = "index" 

37 self.ahead_dim = "ahead" 

38 self.type_dim = "type" 

39 

40 def make_reference_available_locally(self, stations): 

41 "dma8eu_ENS_FORECAST_2019-04-09.nc" 

42 missing_stations = self.list_locally_available_references(self.ref_store_path, stations) 

43 if len(missing_stations) > 0: 

44 check_path_and_create(self.ref_store_path) 

45 dataset = xr.open_mfdataset(os.path.join(self.data_path, "dma8eu_ENS_FORECAST_*.nc")) 

46 darray = dataset.to_array().sortby(["longitude", "latitude"]) 

47 for station, coords in missing_stations.items(): 

48 lon, lat = coords["lon"], coords["lat"] 

49 if self.interp_method is None: 

50 station_data = darray.sel(longitude=lon, latitude=lat, method="nearest", drop=True).squeeze(drop=True) 

51 else: 

52 station_data = darray.interp(**{"longitude": lon, "latitude": lat}, method=self.interp_method) 

53 station_data = station_data.drop_vars(["longitude", "latitude"]).squeeze(drop=True) 

54 station_data = station_data.expand_dims(dim={self.type_dim: [self.ref_name]}).compute() 

55 station_data.coords[self.time_dim] = station_data.coords[self.time_dim] - pd.Timedelta(days=1) 

56 station_data.coords[self.ahead_dim] = station_data.coords[self.ahead_dim] + 1 

57 file_name = self.file_pattern % str(station) 

58 station_data.to_netcdf(os.path.join(self.ref_store_path, file_name)) 

59 

60 @staticmethod 

61 def list_locally_available_references(reference_path, stations) -> dict: 

62 try: 

63 file_list = os.listdir(reference_path) 

64 if len(file_list) > 0: 

65 res = {k: v for k, v in stations.items() if all(k not in x for x in file_list)} 

66 else: 

67 res = stations 

68 except FileNotFoundError: 

69 res = stations 

70 return res