Coverage for mlair/reference_models/reference_model_cams.py: 0%
44 statements
« prev ^ index » next coverage.py v6.4.2, created at 2023-06-30 10:22 +0000
« prev ^ index » next coverage.py v6.4.2, created at 2023-06-30 10:22 +0000
1__author__ = "Lukas Leufen"
2__date__ = "2022-06-27"
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
12class CAMSforecast(AbstractReferenceModel):
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.
19 :param ref_name:
20 :param ref_store_path:
21 :param data_path:
22 :param interp_method:
23 """
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"
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))
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