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): 

15 

16 super().__init__() 

17 self.ref_name = ref_name 

18 if ref_store_path is None: 

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

20 self.ref_store_path = ref_store_path 

21 if data_path is None: 

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

23 else: 

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

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

26 self.time_dim = "index" 

27 self.ahead_dim = "ahead" 

28 self.type_dim = "type" 

29 

30 def make_reference_available_locally(self, stations): 

31 "dma8eu_ENS_FORECAST_2019-04-09.nc" 

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

33 if len(missing_stations) > 0: 

34 check_path_and_create(self.ref_store_path) 

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

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

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

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

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

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

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

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

43 file_name = self.file_pattern % str(station) 

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

45 

46 @staticmethod 

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

48 try: 

49 file_list = os.listdir(reference_path) 

50 if len(file_list) > 0: 

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

52 else: 

53 res = stations 

54 except FileNotFoundError: 

55 res = stations 

56 return res