Newer
Older
Karine PARRA
committed
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
from typing import cast
from source.data_source import *
from typing import Any
class DataMagest(DataSource):
limits = {
VariableFrLiq3D.GARONNE_SALINITY: {"min": 0, "max": 40},
VariableFrLiq3D.GARONNE_TEMPERATURE: {"min": 0, "max": 35},
VariableFrLiq3D.GARONNE_O2: {"min": 0, "max": 20},
VariableFrLiq3D.DORDOGNE_SALINITY: {"min": 0, "max": 40},
VariableFrLiq3D.DORDOGNE_TEMPERATURE: {"min": 0, "max": 35},
VariableFrLiq3D.DORDOGNE_O2: {"min": 0, "max": 20},
}
def parse_response_magest(self, api_response: list[dict[str, Any]], variable: VariableFrLiq3D,
start_date: datetime, end_date: datetime) -> None:
"""Parse the Magest API response to store the result in the source variables
:param api_response: the api response
:param variable : source variable
:param start_date : simulation start date
:param end_date : simulation end date
"""
if not api_response:
if variable == VariableFrLiq3D.GARONNE_SALINITY:
default_value = 0.15
elif variable == VariableFrLiq3D.DORDOGNE_SALINITY:
default_value = 0.10
elif variable == VariableFrLiq3D.GARONNE_TEMPERATURE:
default_value = 12.
elif variable == VariableFrLiq3D.DORDOGNE_TEMPERATURE:
default_value = 12.
elif variable == VariableFrLiq3D.GARONNE_O2:
default_value = 11.
else: # VariableFrLiq3D.DORDOGNE_O2:
default_value = 11.
# magest is supposed to return a value every 10 minutes
timestep = timedelta(seconds=600)
times: np.ndarray = np.arange(start_date, end_date + timestep, timestep)
values = np.ones(times.shape[0]) * default_value
else:
values = np.zeros(len(api_response))
times = np.zeros(len(api_response), dtype="datetime64[s]")
for idx, value in enumerate(api_response):
values[idx] = value["valueConcentree"]
times[idx] = value["data"]["date"]
# Assert that values are in the expected range
limits = self.limits[variable]
values[values > limits["max"]] = limits["max"]
values[values < limits["min"]] = limits["min"]
variable.values = values
variable.times = times
def download(self, start_date: datetime, end_date: datetime) -> bool:
"""Download liquid border forecast from magest.
:param Source self: mock source
:param datetime start_date: start date
:param datetime end_date: end date
:returns: bool
"""
logging.info("Downloading Magest datas")
real_start_date, real_end_date, mode = self.compute_real_dates(start_date, end_date)
for idx, param in enumerate(self.api_params[mode]):
logging.info(f"-- Parsing Magest datas for variable : {self.variables[idx]} ")
try:
req_url = (
f"{self.url[mode]}?{param}"
f"&dateDebut={real_start_date:%Y-%m-%d}"
f"&dateFin={real_end_date:%Y-%m-%d}"
)
response_str = get_curl(req_url).replace("true", "True").replace("false", "False")
response = ast.literal_eval(response_str)
if not response:
logging.info(" WARNING : No datas from Magest, parsing default datas")
self.parse_response_magest(response, cast(VariableFrLiq3D, self.variables[idx]), start_date, end_date)
except (Exception,):
logging.info(" ERROR : Error downloading Magest datas, parsing default datas")
self.parse_response_magest([], cast(VariableFrLiq3D, self.variables[idx]), start_date, end_date)
# Most of the time, in forecast mode, temperature along the garonne
# is not found. Use the same value as Dordogne in that case
temp_garonne_idx = self.variables.index(VariableFrLiq3D.GARONNE_TEMPERATURE)
temp_garonne = self.variables[temp_garonne_idx]
if math.isclose(temp_garonne.values.min(), 12.0) and math.isclose(temp_garonne.values.max(), 12.0):
temp_dordogne_idx = self.variables.index(VariableFrLiq3D.DORDOGNE_TEMPERATURE)
temp_dordogne = self.variables[temp_dordogne_idx]
temp_garonne.values = temp_dordogne.values
temp_garonne.times = temp_dordogne.times
return True