Coverage for uqmodels/postprocessing/UQKPI_Processor.py: 84%

1########################################################################## 

2# KPI_Processor : Processor object (fit/transform procedure) that aim to 

3# tranform predictor & UQestimator output into KPI. 

4# We define a KPI_processor view (Transform any kind of UQ measure in a 

5# specific KPI form) rather than an UQ_processor (express an UQ measure in 

6# several KPI forms) 

7 

8from copy import deepcopy 

9 

10import numpy as np 

11 

12import uqmodels.postprocessing.anomaly_processing as anom_proc 

13import uqmodels.postprocessing.UQ_processing as UQ_proc 

14from uqmodels.processing import Cache_manager, Processor 

15from uqmodels.utils import apply_middledim_reduction 

16 

17 

18class UQKPI_Processor(Processor): 

19 """KPI processor that aim to process (predictor) & UQestimator output into a KPI.""" 

20 

21 def __init__( 

22 self, 

23 name="UQ_processor", 

24 KPI_parameters={}, 

25 cache=Cache_manager(), 

26 random_state=None, 

27 **kwargs 

28 ): 

29 """Init UQKPI_Processor 

30 

31 Args: 

32 name (str, optional): name. Defaults to 'KPI_processor'. 

33 KPI_parameters: Set of parameters link to UQKPI to produce 

34 cache (cahe_manager or None, optional): cache_manager. if None no cache procedure 

35 random_state (bool): handle experimental random using seed. 

36 """ 

37 super().__init__( 

38 name=name, 

39 cache=cache, 

40 KPI_parameters=KPI_parameters, 

41 random_state=random_state, 

42 **kwargs 

43 ) 

44 

45 self.type_UQ = None 

46 

47 def fit( 

48 self, UQ=None, type_UQ=None, pred=None, y=None, type_UQ_params=None, **kwargs 

49 ): 

50 """fitting procedure aim to estimate and store KPI_processor params 

51 

52 Args: 

53 UQ (np.array): UQmeasure provide by the UQestiamtor. 

54 type_UQ (str): Type_UQ of the UQestimators. 

55 pred (np.array): Prediction of the predictor or the UQestimator. 

56 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit 

57 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters. 

58 Defaults to None. 

59 """ 

60 self.type_UQ = type_UQ 

61 super().fit(None) 

62 

63 def transform( 

64 self, UQ=None, type_UQ=None, pred=None, y=None, type_UQ_params=None, **kwargs 

65 ): 

66 """Transform procedure aim to transform (predictor) & UQestimator output into a KPI using fitted parameters. 

67 by default return row UQ provided by UQEstimators. 

68 Args: 

69 UQ (np.array): UQmeasure provide by the UQestiamtor. 

70 type_UQ (str): Type_UQ of the UQestimators. 

71 pred (np.array): Prediction of the predictor or the UQestimator. 

72 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit 

73 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters. 

74 Defaults to None. 

75 """ 

76 if self.type_UQ != type_UQ: 

77 print("Warning : fitted for", self.type_UQ, "give :", type_UQ) 

78 

79 if "reduc_filter" in self.KPI_parameters.keys(): 

80 if UQ is not None: 

81 UQ = np.array( 

82 [ 

83 apply_middledim_reduction( 

84 i, self.KPI_parameters["reduc_filter"] 

85 ) 

86 for i in UQ 

87 ] 

88 ) 

89 

90 if pred is not None: 

91 pred = apply_middledim_reduction( 

92 pred, self.KPI_parameters["reduc_filter"] 

93 ) 

94 

95 if "pred_and_UQ" in self.KPI_parameters.keys(): 

96 if self.KPI_parameters["pred_and_UQ"]: 

97 return (pred, UQ) 

98 

99 return UQ 

100 

101 

102# Predictive_interval_processor : 

103 

104 

105class NormalPIs_processor(UQKPI_Processor): 

106 """Processor aiming to transform UQmeasure in Normal predictive intervals""" 

107 

108 def __init__( 

109 self, 

110 name="Normal_PIs_processor", 

111 KPI_parameters={"list_alpha": [0.05, 0.95], "with_epistemic": True}, 

112 cache=None, 

113 **kwargs 

114 ): 

115 """NormalPIs_processor initialization 

116 

117 Args: 

118 name (str, optional): name of processpr. Defaults to 'Normal_PIs_processor'. 

119 list_alpha (list, optional): Quantile lvl of predictives interval. Defaults to [0.05, 0.95]. 

120 with_epistemic (bool): if type_UQ is 'var_A&E and with epistemic is false, ignore epistemic 

121 and produce aleatoric confidence uncertainty 

122 cache (_type_, optional): Cache manager or None. 

123 """ 

124 super().__init__( 

125 name=name, KPI_parameters=KPI_parameters, cache=cache, **kwargs 

126 ) 

127 

128 self.ndUQ_ratio = None 

129 self.params_ = None 

130 

131 def fit(self, UQ, type_UQ, pred, y=None, type_UQ_params=None, **kwargs): 

132 """fitting procedure aim to estimate and store KPI_processor params for NormalPIs computation 

133 Args: 

134 UQ (np.array): UQmeasure provide by the UQestiamtor. 

135 type_UQ (str): Type_UQ of the UQestimators. 

136 pred (np.array): Prediction of the predictor or the UQestimator. 

137 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit 

138 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters. 

139 Defaults to None. 

140 """ 

141 UQ = deepcopy(UQ) 

142 

143 with_epistemic = True 

144 if "with_epistemic" in self.KPI_parameters.keys(): 

145 with_epistemic = self.KPI_parameters["with_epistemic"] 

146 

147 if not (with_epistemic): 

148 self.ndUQ_ratio = UQ_proc.get_nominal_disentangled_UQ_ratio( 

149 UQ, q_var=1, q_Eratio=3 

150 ) 

151 

152 sigma, sigma_E = UQ_proc.process_UQmeasure_to_TOT_and_E_sigma( 

153 UQ, 

154 type_UQ, 

155 pred=None, 

156 y=None, 

157 type_UQ_params=None, 

158 var_min=0, 

159 var_max=None, 

160 min_cut=0, 

161 max_cut=1, 

162 q_var=1, 

163 q_var_e=1, 

164 k_var_e=1, 

165 q_Eratio=3, 

166 ndUQ_ratio=self.ndUQ_ratio, 

167 reduc_filter=None, 

168 **self.KPI_parameters 

169 ) 

170 

171 UQ = np.power(sigma, 2), sigma_E * 0 

172 

173 self.params_ = UQ_proc.fit_PI( 

174 UQ, type_UQ, pred, y, type_UQ_params=type_UQ_params, **self.KPI_parameters 

175 ) 

176 

177 super().fit(None, type_UQ=type_UQ) 

178 

179 def transform(self, UQ, type_UQ, pred, y=None, type_UQ_params=None, **kwargs): 

180 """Transform procedure aim to transform (predictor) & UQestimator output into normal predictive intervales 

181 according to the list_alpha stored parameters. 

182 

183 Args: 

184 UQ (np.array): UQmeasure provide by the UQestiamtor. 

185 type_UQ (str): Type_UQ of the UQestimators. 

186 pred (np.array): Prediction of the predictor or the UQestimator. 

187 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit 

188 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters. 

189 Defaults to None. 

190 

191 Returns: 

192 list_PIs : Quantiles parametrized by list_alpha provided in the init procedure 

193 """ 

194 super().transform(None, type_UQ=type_UQ) 

195 UQ = deepcopy(UQ) 

196 

197 with_epistemic = True 

198 if "with_epistemic" in self.KPI_parameters.keys(): 

199 with_epistemic = self.KPI_parameters["with_epistemic"] 

200 

201 if type_UQ == "var_A&E": 

202 if not (with_epistemic): 

203 sigma, sigma_E = UQ_proc.process_UQmeasure_to_TOT_and_E_sigma( 

204 UQ, 

205 type_UQ, 

206 pred=None, 

207 y=None, 

208 type_UQ_params=None, 

209 var_min=0, 

210 var_max=None, 

211 min_cut=0, 

212 max_cut=1, 

213 q_var=1, 

214 q_var_e=1, 

215 k_var_e=1, 

216 q_Eratio=3, 

217 ndUQ_ratio=self.ndUQ_ratio, 

218 **self.KPI_parameters 

219 ) 

220 UQ = np.power(sigma, 2), sigma_E * 0 

221 

222 list_PIs, _ = UQ_proc.compute_PI( 

223 UQ, 

224 type_UQ, 

225 pred, 

226 y, 

227 type_UQ_params, 

228 params_=self.params_, 

229 **self.KPI_parameters 

230 ) 

231 return list_PIs 

232 

233 

234class Epistemicscorelvl_processor(UQKPI_Processor): 

235 """Processor aiming to transform UQmeasure (containing epistemic measure) into Epistemic_score_lvl (ADD link)""" 

236 

237 def __init__( 

238 self, name="Epistemicscorelvl", KPI_parameters={}, cache=None, **kwargs 

239 ): 

240 """Initialization 

241 

242 Args: 

243 name (str, optional): name Defaults to 'Epistemicscorelvl'. 

244 KPI_paramsaters (_type_, optional): Parameters of processor : here none. Defaults to None. 

245 cache (_type_, optional): Cache manager or none 

246 """ 

247 super().__init__( 

248 name=name, KPI_parameters=KPI_parameters, cache=cache, **kwargs 

249 ) 

250 

251 self.params_ = None 

252 

253 def fit(self, UQ, type_UQ, pred, y=None, type_UQ_params=None, **kwargs): 

254 """Fitting procedure aim to estimate and store KPI_processor params for Epistemicscorelvl computation 

255 

256 Args: 

257 UQ (np.array): UQmeasure provide by the UQestiamtor. 

258 type_UQ (str): Type_UQ of the UQestimators. 

259 pred (np.array): Prediction of the predictor or the UQestimator. 

260 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit 

261 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters. 

262 Defaults to None. 

263 """ 

264 self.params_ = UQ_proc.fit_Epistemic_score( 

265 UQ, type_UQ, pred, y=y, type_UQ_params=type_UQ_params, **self.KPI_parameters 

266 ) 

267 super().fit(type_UQ=type_UQ) 

268 

269 def transform(self, UQ, type_UQ, pred, y=None, type_UQ_params=None, **kwargs): 

270 """Transform procedure aim to transform (predictor) & UQestimator output into a Epistemic_scorelvl 

271 according Episticscore quantile fitted during training procedure. 

272 

273 Args: 

274 UQ (np.array): UQmeasure provide by the UQestiamtor. 

275 type_UQ (str): Type_UQ of the UQestimators. 

276 pred (np.array): Prediction of the predictor or the UQestimator. 

277 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit 

278 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters. 

279 Defaults to None. 

280 

281 Returns: Epistemic_scorelvl : Epistemic_scorelvl that express the quantile class of Epistemic values 

282 amoung quantile [0.50, 0.80, 0.90, 0.95, 0.975, 0.99, 0.999] 

283 """ 

284 super().transform(None, type_UQ=type_UQ) 

285 Epistemic_scorelvl, _ = UQ_proc.compute_Epistemic_score( 

286 UQ, 

287 self.type_UQ, 

288 pred, 

289 y=y, 

290 type_UQ_params=type_UQ_params, 

291 params_=self.params_, 

292 **self.KPI_parameters 

293 ) 

294 return Epistemic_scorelvl 

295 

296 

297class Anomscore_processor(UQKPI_Processor): 

298 """Processor aiming to produce an contextual deviation anomaly score from UQmeasure prediction and observation""" 

299 

300 def __init__(self, name="Anomscore", KPI_parameters=dict(), cache=None, **kwargs): 

301 """Initialization 

302 

303 Args: 

304 name (str, optional): name. Defaults to 'Anomscore'. 

305 KPI_parameters (_type_, optional): dict of parameters link to compute_score function. 

306 Defaults to None will use predefined paramters. 

307 cache (_type_, optional): Cache manager or none 

308 """ 

309 super().__init__( 

310 name=name, KPI_parameters=KPI_parameters, cache=cache, **kwargs 

311 ) 

312 

313 self.params_ = None 

314 self.fusion_params_ = None 

315 

316 def fit( 

317 self, UQ, type_UQ, pred, y=None, type_UQ_params=None, ctx_mask=None, **kwargs 

318 ): 

319 """Fitting procedure aim to estimate and store Anomscore_processor params for Anomscore computation 

320 

321 Args: 

322 UQ (np.array): UQmeasure provide by the UQestiamtor. 

323 type_UQ (str): Type_UQ of the UQestimators. 

324 pred (np.array): Prediction of the predictor or the UQestimator. 

325 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit 

326 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters. 

327 Defaults to None. 

328 """ 

329 self.params_ = anom_proc.fit_anom_score( 

330 UQ, 

331 type_UQ, 

332 pred, 

333 y=y, 

334 type_UQ_params=type_UQ_params, 

335 ctx_mask=ctx_mask, 

336 **self.KPI_parameters, 

337 **kwargs 

338 ) 

339 

340 if "fusion" in self.KPI_parameters.keys(): 

341 if self.KPI_parameters["fusion"]: 

342 anom_score, _ = anom_proc.compute_anom_score( 

343 UQ=UQ, 

344 type_UQ=type_UQ, 

345 pred=pred, 

346 y=y, 

347 type_UQ_params=type_UQ_params, 

348 ctx_mask=ctx_mask, 

349 params_=self.params_, 

350 **self.KPI_parameters, 

351 **kwargs 

352 ) 

353 

354 self.fusion_params_ = anom_proc.fit_score_fusion( 

355 anom_score, **self.KPI_parameters 

356 ) 

357 

358 super().fit(None, type_UQ=type_UQ) 

359 

360 def transform( 

361 self, UQ, type_UQ, pred, y=None, type_UQ_params=None, ctx_mask=None, **kwargs 

362 ): 

363 """Transform procedure aim to transform (predictor) & UQestimator output into a Epistemic_scorelvl 

364 according Episticscore quantile fitted during training procedure. 

365 

366 Args: 

367 UQ (np.array): UQmeasure provide by the UQestimator. 

368 type_UQ (str): Type_UQ of the UQestimators. 

369 pred (np.array): Prediction of the predictor or the UQestimator. 

370 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit 

371 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters. 

372 Defaults to None. 

373 

374 Returns: anom_score : Deviation score (pred-y) process by anom_proc.compute_anom_score 

375 """ 

376 

377 # Create a bug for python >3.8 : wrapped() missing 1required positional argument :'X' 

378 # super().transform(type_UQ=type_UQ) 

379 

380 anom_score, _ = anom_proc.compute_anom_score( 

381 UQ=UQ, 

382 type_UQ=type_UQ, 

383 pred=pred, 

384 y=y, 

385 type_UQ_params=type_UQ_params, 

386 ctx_mask=ctx_mask, 

387 params_=self.params_, 

388 **self.KPI_parameters, 

389 **kwargs 

390 ) 

391 

392 if "fusion" in self.KPI_parameters.keys(): 

393 if self.KPI_parameters["fusion"]: 

394 anom_score, _ = anom_proc.compute_score_fusion( 

395 anom_score, params_=self.fusion_params_, **self.KPI_parameters 

396 ) 

397 anom_score = -anom_score 

398 return anom_score