Coverage for quality/views.py: 58%

1from itertools import groupby 

2from typing import Any 

3 

4import numpy as np 

5import numpy.typing as npt 

6import plotly.graph_objects as go 

7from django.core.exceptions import BadRequest 

8from django.db.models import Max, Min 

9from django.http import HttpRequest, HttpResponse 

10from django.urls import reverse 

11from django.views import View 

12from django.views.generic import DetailView, ListView, TemplateView 

13from django.views.generic.base import ContextMixin 

14from pendulum.date import Date 

15from plotly.subplots import make_subplots 

16 

17from quality.forms import FilterForm 

18 

19from .layouts import ( 

20 HEATMAP_LAYOUT_COMPLETENESS, 

21 HEATMAP_LAYOUT_NETWORK, 

22 HEATMAP_LAYOUT_QUALITY, 

23 HEATMAP_LAYOUT_STATION, 

24) 

25from .models import ( 

26 RESULT_PONDERATION, 

27 Channel, 

28 Check, 

29 Network, 

30 Operator, 

31 Station, 

32) 

33from .utils import Filter, get_rtserved_stations, get_station_last_rt_data 

34 

35 

36class HealthCheck(View): 

37 def get(self, _request: HttpRequest) -> HttpResponse: 

38 return HttpResponse("OK") 

39 

40 

41class CheckFormMixin(ContextMixin): 

42 def get_context_data(self, **kwargs: object) -> dict[str, object]: 

43 context = super().get_context_data(**kwargs) 

44 extra = { 

45 "networks": Network.objects.all(), 

46 "stations": Station.objects.all().order_by("code"), 

47 "operators": Operator.objects.checks_exist() 

48 .only("agency") 

49 .order_by("agency"), 

50 "channels": Channel.objects.checks_exist().instruments(), 

51 } 

52 context.update(extra) 

53 return context 

54 

55 

56class HomePage(CheckFormMixin, TemplateView): 

57 template_name = "quality/homepage.html" 

58 

59 

60class OperatorList(ListView[Operator]): 

61 model = Operator 

62 ordering = ["agency"] 

63 

64 

65class NetworkList(ListView[Network]): 

66 model = Network 

67 

68 

69class NetworkOverview(DetailView[Network]): 

70 model = Network 

71 template_name = "quality/network_overview.html" 

72 

73 def get_context_data(self, **kwargs: object) -> dict[str, object]: 

74 context = super().get_context_data(**kwargs) 

75 query = Station.objects.overview(network_id=self.object.pk) 

76 context.update( 

77 { 

78 "stations": query.filter(triggered=False), 

79 "triggered_stations": query.filter(triggered=True), 

80 "rtserved": get_rtserved_stations(network=self.object.code), 

81 } 

82 ) 

83 return context 

84 

85 

86class Heatmap(ContextMixin): 

87 @staticmethod 

88 def completeness_trace(mean: npt.NDArray[np.int8]) -> go.Heatmap: 

89 return go.Heatmap( 

90 y=[ 

91 (f"<span id='{index}'>{completeness}%</span>") 

92 for index, completeness in enumerate(mean) 

93 ], 

94 z=np.expand_dims(mean, axis=1), 

95 **HEATMAP_LAYOUT_COMPLETENESS, 

96 ) 

97 

98 def figure( 

99 self, 

100 results: npt.NDArray[np.int8], 

101 items: list[str], 

102 dates: list[Date], 

103 ) -> go.Figure: 

104 mean = np.round(np.mean(results[:, :, 0], axis=1), 2) 

105 # Height = items * 15 px + padding 

106 minimum_height = 400 

107 layout_height = max(minimum_height, len(items) * 15 + 180) 

108 

109 figure = make_subplots(specs=[[{"secondary_y": True}]]) 

110 figure.add_trace( 

111 go.Heatmap( 

112 x=dates, 

113 y=items, 

114 z=results[:, :, 0] + results[:, :, 1], 

115 customdata=results, 

116 **HEATMAP_LAYOUT_NETWORK, 

117 ), 

118 secondary_y=False, 

119 ) 

120 figure.add_trace(self.completeness_trace(mean), secondary_y=True) 

121 figure.update_layout( 

122 xaxis_side="top", 

123 height=layout_height, 

124 ) 

125 return figure.to_html(full_html=False, include_plotlyjs=False) 

126 

127 def get_summary(self, _filters: Filter) -> object: 

128 msg = "Subclass must implement abstract method" 

129 raise NotImplementedError(msg) 

130 

131 @staticmethod 

132 def build_x_values(query: Any) -> list[Date]: 

133 return sorted({key for key, _ in groupby(query, lambda x: x[2])}) 

134 

135 def build_y_values(self, query: Any) -> list[str]: 

136 return [ 

137 "<a href='{}?{}' target='_self'>{}.{}</a>".format( 

138 reverse("quality:station-detail", kwargs={"pk": pk}), 

139 self.request.GET.urlencode(), # type: ignore[attr-defined] 

140 network, 

141 station, 

142 ) 

143 for (pk, station, network), _ in groupby( 

144 query, 

145 lambda x: (x[1], x[0], x[7]), 

146 ) 

147 ] 

148 

149 @staticmethod 

150 def build_z_values(query: Any) -> npt.NDArray[np.int8]: 

151 # build z values (check results) 

152 # Warning : Here, we make the assumption that there is a check 

153 # for channel each day 

154 time_to_percent_ratio = 100 / (86400 * 1000) 

155 return np.array( 

156 [ 

157 [ 

158 [ 

159 check[3] * time_to_percent_ratio, 

160 RESULT_PONDERATION[check[4]], 

161 check[5], 

162 check[6], 

163 ] 

164 for check in checks 

165 ] 

166 for _, checks in groupby(query, lambda x: x[0]) 

167 ], 

168 ) 

169 

170 def years_available(self, _filters: Filter) -> list[int]: 

171 msg = "Subclass must implement abstract method" 

172 raise NotImplementedError(msg) 

173 

174 def get_context_data(self, **kwargs: object) -> dict[str, object]: 

175 context = super().get_context_data(**kwargs) 

176 form = FilterForm(self.request.GET) # type: ignore[attr-defined] 

177 if not form.is_valid(): 

178 raise BadRequest(form.errors.as_data()) 

179 filters: Filter = form.cleaned_data # type: ignore[assignment] 

180 query = self.get_summary(filters) 

181 if query: 

182 dates = self.build_x_values(query) 

183 items = self.build_y_values(query) 

184 results = self.build_z_values(query) 

185 years_available = self.years_available(filters) 

186 

187 context.update( 

188 { 

189 "figure": self.figure(results=results, items=items, dates=dates), 

190 "years_available": years_available, 

191 }, 

192 ) 

193 return context 

194 

195 

196class FullHeatmap(Heatmap): 

197 @staticmethod 

198 def build_y_values(query: Any) -> list[str]: 

199 return [ 

200 key 

201 for key, _ in groupby( 

202 query, 

203 lambda x: f"{x[9]}.{x[8]}.{x[7]}.{x[1]}", 

204 ) 

205 ] 

206 

207 

208class NetworkDetail(Heatmap, DetailView[Network]): 

209 model = Network 

210 

211 def years_available(self, _filters: Filter) -> list[int]: 

212 dates = Check.objects.filter( 

213 channel__station__network_id=self.object.id, 

214 ).aggregate(min=Min("date"), max=Max("date")) 

215 return list(range(dates["min"].year, dates["max"].year + 1)) 

216 

217 def get_summary(self, filters: Filter) -> object: 

218 return ( 

219 Station.objects.filter(network_id=self.object.pk) 

220 .channels(filters["channel"]) 

221 .year(filters["year"]) 

222 .triggered(filters["triggered"]) 

223 .summary() 

224 ) 

225 

226 

227class NetworkFullDetail(FullHeatmap, NetworkDetail): 

228 def get_summary(self, filters: Filter) -> object: 

229 return ( 

230 Check.objects.filter(channel__station__network_id=self.object.pk) 

231 .year(filters["year"]) 

232 .triggered(filters["triggered"]) 

233 .summary() 

234 ) 

235 

236 

237class CheckOverview(Heatmap, CheckFormMixin, TemplateView): 

238 template_name = "quality/check_overview.html" 

239 

240 def years_available(self, filters: Filter) -> list[int]: 

241 dates = ( 

242 Check.objects.networks(filters["network"]) 

243 .stations(filters["station"]) 

244 .channels(filters["channel"]) 

245 .operators(filters["operator"]) 

246 .triggered(filters["triggered"]) 

247 .aggregate(min=Min("date"), max=Max("date")) 

248 ) 

249 return list(range(dates["min"].year, dates["max"].year + 1)) 

250 

251 def get_summary(self, filters: Filter) -> object: 

252 return ( 

253 Station.objects.networks(filters["network"]) 

254 .stations(filters["station"]) 

255 .channels(filters["channel"]) 

256 .operators(filters["operator"]) 

257 .year(filters["year"]) 

258 .triggered(filters["triggered"]) 

259 .summary() 

260 ) 

261 

262 

263class CheckFullOverview(FullHeatmap, CheckOverview): 

264 def get_summary(self, filters: Filter) -> object: 

265 return ( 

266 Check.objects.networks(filters["network"]) 

267 .stations(filters["station"]) 

268 .channels(filters["channel"]) 

269 .operators(filters["operator"]) 

270 .year(filters["year"]) 

271 .triggered(filters["triggered"]) 

272 .summary() 

273 ) 

274 

275 

276class StationDetail(FullHeatmap, DetailView[Station]): 

277 model = Station 

278 queryset = Station.objects.all().select_related("network") 

279 

280 def figure( 

281 self, 

282 results: npt.NDArray[np.int8], 

283 items: list[str], 

284 dates: list[Date], 

285 ) -> go.Figure: 

286 minimum_height = 450 

287 layout_height = max(minimum_height, len(items) * 60 + 200) 

288 

289 mean = np.round(np.mean(results[:, :, 0], axis=1), 2) 

290 figure = make_subplots( 

291 rows=2, 

292 cols=1, 

293 shared_xaxes=True, 

294 specs=[[{"secondary_y": True}], [{"secondary_y": False}]], 

295 ) 

296 figure.add_trace( 

297 go.Heatmap( 

298 x=dates, 

299 y=items, 

300 z=results[:, :, 0] + results[:, :, 1], 

301 customdata=results, 

302 **HEATMAP_LAYOUT_QUALITY, 

303 ), 

304 row=1, 

305 col=1, 

306 secondary_y=False, 

307 ) 

308 figure.add_trace(self.completeness_trace(mean), secondary_y=True) 

309 # We need to know the maximum number of traces for all checks 

310 zmax = max(np.max(results[:, :, 2]), 4) # type: ignore [call-overload] 

311 HEATMAP_LAYOUT_STATION.update({"zmax": zmax}) 

312 figure.add_trace( 

313 go.Heatmap(x=dates, y=items, z=results[:, :, 2], **HEATMAP_LAYOUT_STATION), 

314 row=2, 

315 col=1, 

316 secondary_y=False, 

317 ) 

318 figure.update_xaxes(side="top") 

319 figure.update_layout(height=layout_height) 

320 return figure.to_html(full_html=False, include_plotlyjs=False) 

321 

322 def years_available(self, _filters: Filter) -> list[int]: 

323 dates = Check.objects.filter( 

324 channel__station_id=self.object.id, 

325 ).aggregate(min=Min("date"), max=Max("date")) 

326 return list(range(dates["min"].year, dates["max"].year + 1)) 

327 

328 def get_summary(self, filters: Filter) -> object: 

329 return ( 

330 Check.objects.filter(channel__station_id=self.object.pk) 

331 .year(filters["year"]) 

332 .summary() 

333 ) 

334 

335 def get_context_data(self, **kwargs: object) -> dict[str, object]: 

336 context = super().get_context_data(**kwargs) 

337 context["last_data"] = get_station_last_rt_data( 

338 network=self.object.network.code, station=self.object.code 

339 ) 

340 return context