An error occurred while loading the file. Please try again.
-
Le Roux Erwan authoredc7416d72
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
import os
import os.path as op
from collections import OrderedDict
from typing import List, Dict
import matplotlib.pyplot as plt
import pandas as pd
from netCDF4 import Dataset
from experiment.meteo_france_SCM_study.abstract_variable import AbstractVariable
from experiment.meteo_france_SCM_study.massif import safran_massif_names_from_datasets
from extreme_estimator.margin_fits.plot.create_shifted_cmap import get_color_rbga_shifted
from spatio_temporal_dataset.coordinates.abstract_coordinates import AbstractCoordinates
from spatio_temporal_dataset.coordinates.spatial_coordinates.abstract_spatial_coordinates import \
AbstractSpatialCoordinates
from spatio_temporal_dataset.spatio_temporal_observations.annual_maxima_observations import AnnualMaxima
from utils import get_full_path, cached_property
class AbstractStudy(object):
ALTITUDES = [1800, 2400]
def __init__(self, variable_class: type, altitude: int = 1800, year_min=1000, year_max=3000):
assert altitude in self.ALTITUDES, altitude
self.altitude = altitude
self.model_name = None
self.variable_class = variable_class
self.year_min = year_min
self.year_max = year_max
def write_to_file(self, df: pd.DataFrame):
if not op.exists(self.result_full_path):
os.makedirs(self.result_full_path, exist_ok=True)
df.to_csv(op.join(self.result_full_path, 'merged_array_{}_altitude.csv'.format(self.altitude)))
""" Data """
@property
def df_all_daily_time_series_concatenated(self) -> pd.DataFrame:
df_list = [pd.DataFrame(time_serie, columns=self.safran_massif_names) for time_serie in
self.year_to_daily_time_serie.values()]
df_concatenated = pd.concat(df_list)
return df_concatenated
@property
def observations_annual_maxima(self) -> AnnualMaxima:
return AnnualMaxima(df_maxima_gev=pd.DataFrame(self.year_to_annual_maxima, index=self.safran_massif_names))
@property
def df_annual_total(self) -> pd.DataFrame:
return pd.DataFrame(self.year_to_annual_total, index=self.safran_massif_names).transpose()
def annual_aggregation_function(self, *args, **kwargs):
raise NotImplementedError()
""" Load some attributes only once """
@cached_property
def year_to_dataset_ordered_dict(self) -> OrderedDict:
# Map each year to the correspond netCDF4 Dataset
year_to_dataset = OrderedDict()
nc_files = [(int(f.split('_')[-2][:4]), f) for f in os.listdir(self.safran_full_path) if f.endswith('.nc')]
for year, nc_file in sorted(nc_files, key=lambda t: t[0]):
if self.year_min <= year < self.year_max:
year_to_dataset[year] = Dataset(op.join(self.safran_full_path, nc_file))
print(year_to_dataset.keys())
return year_to_dataset
@cached_property
def year_to_daily_time_serie(self) -> OrderedDict:
return self._year_to_daily_time_serie
@cached_property
def year_to_annual_maxima(self) -> OrderedDict:
# Map each year to an array of size nb_massif
year_to_annual_maxima = OrderedDict()
for year, time_serie in self._year_to_max_daily_time_serie.items():
year_to_annual_maxima[year] = time_serie.max(axis=0)
return year_to_annual_maxima
@cached_property
def year_to_annual_total(self) -> OrderedDict:
# Map each year to an array of size nb_massif
year_to_annual_mean = OrderedDict()
for year, time_serie in self._year_to_daily_time_serie.items():
year_to_annual_mean[year] = self.annual_aggregation_function(time_serie, axis=0)
return year_to_annual_mean
def instantiate_variable_object(self, dataset) -> AbstractVariable:
return self.variable_class(dataset, self.altitude)
""" Private methods to be overwritten """
@property
def _year_to_daily_time_serie(self) -> OrderedDict:
# Map each year to a matrix of size 365-nb_days_consecutive+1 x nb_massifs
year_to_variable = {year: self.instantiate_variable_object(dataset) for year, dataset in
self.year_to_dataset_ordered_dict.items()}
year_to_daily_time_serie = OrderedDict()
for year in self.year_to_dataset_ordered_dict.keys():
year_to_daily_time_serie[year] = year_to_variable[year].daily_time_serie
return year_to_daily_time_serie
@property
def _year_to_max_daily_time_serie(self) -> OrderedDict:
return self._year_to_daily_time_serie
##########
@property
def safran_massif_names(self) -> List[str]:
return self.original_safran_massif_names
@property
def original_safran_massif_names(self) -> List[str]:
# Load the names of the massif as defined by SAFRAN
return safran_massif_names_from_datasets(list(self.year_to_dataset_ordered_dict.values()), self.altitude)
@property
def original_safran_massif_id_to_massif_name(self) -> Dict[int, str]:
return {massif_id: massif_name for massif_id, massif_name in enumerate(self.original_safran_massif_names)}
@cached_property
def massifs_coordinates(self) -> AbstractSpatialCoordinates:
# Coordinate object that represents the massif coordinates in Lambert extended
df_centroid = self.load_df_centroid()
for coord_column in [AbstractCoordinates.COORDINATE_X, AbstractCoordinates.COORDINATE_Y]:
df_centroid.loc[:, coord_column] = df_centroid[coord_column].str.replace(',', '.').astype(float)
# Build coordinate object from df_centroid
return AbstractSpatialCoordinates.from_df(df_centroid)
def load_df_centroid(self) -> pd.DataFrame:
df_centroid = pd.read_csv(op.join(self.map_full_path, 'coordonnees_massifs_alpes.csv'))
df_centroid.set_index('NOM', inplace=True)
df_centroid = df_centroid.loc[self.original_safran_massif_names]
return df_centroid
@property
def coordinate_id_to_massif_name(self) -> Dict[int, str]:
df_centroid = self.load_df_centroid()
return dict(zip(df_centroid['id'], df_centroid.index))
""" Visualization methods """
def visualize_study(self, ax=None, massif_name_to_value=None, show=True, fill=True, replace_blue_by_white=True,
label=None):
massif_names, values = list(zip(*massif_name_to_value.items()))
colors = get_color_rbga_shifted(ax, replace_blue_by_white, values, label=label)
massif_name_to_fill_kwargs = {massif_name: {'color': color} for massif_name, color in
zip(massif_names, colors)}
if ax is None:
ax = plt.gca()
df_massif = pd.read_csv(op.join(self.map_full_path, 'massifsalpes.csv'))
coord_tuples = [(row_massif['idx'], row_massif[AbstractCoordinates.COORDINATE_X],
row_massif[AbstractCoordinates.COORDINATE_Y])
for _, row_massif in df_massif.iterrows()]
for _, coordinate_id in enumerate(set([t[0] for t in coord_tuples])):
# Retrieve the list of coords (x,y) that define the contour of the massif of id coordinate_id
coords_list = [coords for idx, *coords in coord_tuples if idx == coordinate_id]
# if j == 0:
# mask_outside_polygon(poly_verts=l, ax=ax)
# Plot the contour of the massif
coords_list = list(zip(*coords_list))
ax.plot(*coords_list, color='black')
# Potentially, fill the inside of the polygon with some color
if fill:
massif_name = self.coordinate_id_to_massif_name[coordinate_id]
fill_kwargs = massif_name_to_fill_kwargs[massif_name] if massif_name_to_fill_kwargs is not None else {}
ax.fill(*coords_list, **fill_kwargs)
# x , y = list(self.massifs_coordinates.df_all_coordinates.loc[massif_name])
# x , y= coords_list[0][0], coords_list[0][1]
# print(x, y)
# print(massif_name)
# ax.scatter(x, y)
# ax.text(x, y, massif_name)
# Display the center of the massif
ax.scatter(self.massifs_coordinates.x_coordinates, self.massifs_coordinates.y_coordinates, s=1)
# Display the name of the massif
for _, row in self.massifs_coordinates.df_all_coordinates.iterrows():
x, y = list(row)
massif_name = row.name
value = massif_name_to_value[massif_name]
ax.text(x, y, str(round(value, 1)))
if show:
plt.show()
""" Some properties """
@property
def title(self):
return "{} at altitude {}m".format(self.variable_name, self.altitude)
@property
def variable_name(self):
return self.variable_class.NAME
@property
def relative_path(self) -> str:
return r'local/spatio_temporal_datasets'
@property
def full_path(self) -> str:
return get_full_path(relative_path=self.relative_path)
@property
def safran_full_path(self) -> str:
assert self.model_name in ['Safran', 'Crocus']
return op.join(self.full_path, 'safran-crocus_{}'.format(self.altitude), self.model_name)
@property
def map_full_path(self) -> str:
return op.join(self.full_path, 'map')
@property
def result_full_path(self) -> str:
return op.join(self.full_path, 'results')