Coverage for uqmodels/UQModel.py: 68%

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

2# UQModel : "Model than apply an modeling pipeline composed of an predictor(optional), UQestimator and KPI

3# To write a custom UQModel, inherite from UQModel class an use supoer().__init__()

5# Class UQmodels : from an UQestimators : perform dadada

7import inspect

8import os

10import numpy as np

11from sklearn.base import BaseEstimator

13import uqmodels.modelization.ML_estimator.random_forest_UQ as RF_UQ

14import uqmodels.postprocessing.UQKPI_Processor as UQProc

15import uqmodels.preprocessing.preprocessing as pre_pre

16import uqmodels.preprocessing.structure as pre_struc

17from uqmodels.processing import Cache_manager

18from uqmodels.utils import add_random_state, apply_mask, apply_middledim_reduction, cut

21class UQModel(BaseEstimator):

22 """

23 UQModel class : instanciate a pipeline of model estimation, model UQ estimation and post processing as

24 a (Fit,Predict,Score) triplet of method

25 """

27 def __init__(

28 self,

29 UQEstimator_initializer=RF_UQ.RF_UQEstimator,

30 UQEstimator_params=RF_UQ.get_params_dict(),

31 # Necesite with_prediction or to provide a predictor.

32 name="UQModel",

33 predictor=None,

34 preprocessor=None,

35 list_predict_KPI_processors=None, # List of UQProcessor

36 list_score_KPI_processors=None,

37 reduc_filter=None,

38 roll=None,

39 cache_manager=Cache_manager(),

40 save_result=False,

41 save_models=False,

42 random_state=None,

43 **kwargs

44 ):

45 """Initialize UQModel class

47 Args:

48 UQEstimator_initializer (obj_init): init method for instanciate an UQEstimator

49 UQEstimator_params (dict_params): params for the init method

50 name (str, optional): Wrapper name . Defaults to 'UQModel'.

51 predictor (predictor or None, optional): Predictor instanciate if None,

52 assume that UQEstiamtor also make prediction. Defaults to None.

53 list_predict_KPI_processors (list_processor or None, optional): List of instanciated

54 predict_KPI_processors. Defaults to None.

55 list_score_KPI_processors (list_processor or None, optional): List of instanciated

56 score_KPI_processors. Defaults to None

57 reduc_filter (np.array or None): Weigth ponderation for reduc middle dimension (if needed)

58 cache_manager (cache_manager or None, optional): Cache manager. Defaults to None.

59 random_state : Controls randoms

60 """

62 self.name = name

64 self.predictor = predictor

65 if predictor is not None:

66 self.predictor_initializer = predictor.__init__()

67 self.predictor_params = predictor.get_params()

69 self.UQEstimator_initializer = UQEstimator_initializer

70 self.preprocessor = preprocessor

71 self.UQEstimator_params = UQEstimator_params

72 self.list_predict_KPI_processors = list_predict_KPI_processors

73 self.list_score_KPI_processors = list_score_KPI_processors

74 self.reduc_filter = reduc_filter

75 self.roll = roll

76 self.cache_manager = cache_manager

77 self.random_state = random_state

78 if self.random_state is not None:

79 self.UQEstimator_params["random_state"] = random_state

81 for n, predict_KPI_processor in enumerate(list_predict_KPI_processors):

82 predict_KPI_processor.random_state = add_random_state(random_state, n)

84 for n, score_KPI_processor in enumerate(list_score_KPI_processors):

85 score_KPI_processor.random_state = add_random_state(

86 random_state, 10 + n

87 )

89 self.UQEstimator = None

90 self.is_fitted = None

91 self.type_UQ = None

92 self.type_UQ_params = None

93 self.save_result = save_result

94 self.save_models = save_models

96 for key_arg in kwargs.keys():

97 setattr(self, key_arg, kwargs[key_arg])

99 def _load(self, entity, path, name):

100 """Auxiliar load procedure for an specific entity : try to use entity.load method

101 or force save using UQModel cache_manager

102

103 Todo:

104 * Raise fail error -> part already handle by cache manager

105

106 Args:

107 entity (obj): Object to load

108 path (str): Path to load

109 name (str): name to load

110 """

111 if hasattr(entity, "load"):

112 arg_list = list(inspect.signature(entity.load).parameters)

113 if "name" in arg_list:

114 entity.load(path, name=name)

115 else:

116 path = os.path.join(path, name)

117 entity.load(path)

118 else:

119 query = {"storing": path, "name": name}

120 entity = self.cache_manager.load(query)

121 return entity

122

123 def load(self, path=None, name="UQModel"):

124 """UQ model load procedure : recovers UQ model structure, then predictor (if existe), then UQestimators

125

126 Todo:

127 * Raise fail error -> part already handle by cache manager

128

129 Args:

130 path (str, optional): path of UQmodels save if none use path store in cache manager

131 name (str, optional): name of UQmodels to load uf none use UQModel name

132 """

133 if path is None:

134 path = self.cache_manager.storing

135

136 query = {"storing": path, "name": name}

137 _UQModel = self.cache_manager.load(query)

138

139 for key in _UQModel.__dict__.keys():

140 self.__setattr__(key, _UQModel.__getattribute__(key))

141

142 if self.predictor:

143 self.predictor = self.predictor_initializer(**self.predictor_params)

144 self.predictor = self._load(self.predictor, path, "predictor")

145

146 if self.UQEstimator:

147 self.UQEstimator = self.UQEstimator_initializer(**self.UQEstimator_params)

148 self.UQEstimator = self._load(self.UQEstimator, path, "UQEstimator")

149

150 # TO do :

151 # Specific load procedure for predict_KPI_processors and score_KPI_processors

152 # Specific load procedure for preprocessor object

153 # Hold currently by global __dict__ parameters save

154

155 def _save(self, entity, path, name, UQModel_name=None):

156 """Auxiliar save procedure for an specific entity : try to use entity.save

157 method or force save using UQmMdels cache_manager

158

159 Todo:

160 * Raise fail error -> part already handle by cache manager

161

162 Args:

163 entity (obj): Object to save

164 path (str): Path to save

165 name (str): name to save

166 """

167 if UQModel_name is None:

168 UQModel_name = self.name

169

170 if hasattr(entity, "save"):

171 arg_list = list(inspect.signature(entity.save).parameters)

172 if "name" in arg_list:

173 path = os.path.join(path, UQModel_name)

174 entity.save(path, name=name)

175 else:

176 path = os.path.join(path, UQModel_name, name)

177 entity.save(path)

178 else:

179 query = {"storing": path, "keys": [UQModel_name, name]}

180 if self.cache_manager is not None:

181 self.cache_manager.save(query, entity)

182 else:

183 print("No_cache_manager : cannot save")

184

185 def save(self, path=None, name=None):

186 """UQ model save procedure : recovers UQ model structure, then predictor (if existe), then UQestimators

187

188 Todo:

189 * Raise fail error

190

191 Args:

192 path (path, optional): path of UQmodels save

193 name (name, optional): name of UQmodels to load

194 """

195 if name is None:

196 name = self.name

197

198 if path is None:

199 path = self.cache_manager.storing

200

201 if self.predictor is not None:

202 self._save(self.predictor, path, "predictor", UQModel_name=name)

203 tmp_predictor = self.predictor

204 self.predictor = True

205

206 if self.UQEstimator:

207 self._save(self.UQEstimator, path, "UQEstimator", UQModel_name=name)

208 tmp_UQestimator = self.UQEstimator

209 self.UQEstimator = True

210

211 # TO do :

212 # Specific save procedure for predict_KPI_processors and score_KPI_processors

213 # Specific save procedure for preprocessor object

214 # Hold currently by global __dict__ parameters save

215

216 query = {"storing": path, "keys": [name, "UQModel"]}

217 self.cache_manager.save(query, self)

218

219 if self.predictor is not None:

220 self.predictor = tmp_predictor

221

222 if self.UQEstimator:

223 self.UQEstimator = tmp_UQestimator

224

225 def _init_UQEstimator(self, X=None, y=None):

226 if not self.UQEstimator:

227 self.UQEstimator = self.UQEstimator_initializer(**self.UQEstimator_params)

228

229 if hasattr(self.UQEstimator, "factory"):

230 _ = self.UQEstimator.factory(X, y, only_fit_scaler=True)

231

232 self.type_UQ = self.UQEstimator.type_UQ

233 if hasattr(self.UQEstimator, "type_UQ_params"):

234 self.type_UQ_params = self.UQEstimator.type_UQ_params

235

236 def _fit_predict_KPI_processors(self, UQ, pred, y, **kwargs):

237 """Auxiliar method that apply fit method of predict_KPI_processors

238

239 Args:

240 UQ (np.array): Features

241 pred (np.array): Prediction from predictor or UQEstimator

242 y (np.array): Targets/observations

243 """

244 for processor in self.list_predict_KPI_processors:

245 processor.fit(

246 UQ,

247 self.type_UQ,

248 pred,

249 y=y,

250 type_UQ_params=self.type_UQ_params,

251 **kwargs

252 )

253

254 def _fit_score_KPI_processors(self, UQ, pred, y, **kwargs):

255 """Auxiliar method that apply fit method of score_KPI_processors

256

257 Args:

258 UQ (np.array): Features

259 pred (np.array): Prediction from predictor or UQEstimator

260 y (np.array): Targets/observations

261 """

262 for processor in self.list_score_KPI_processors:

263 processor.fit(

264 UQ, self.type_UQ, pred, y, type_UQ_params=self.type_UQ_params, **kwargs

265 )

266

267 def fit(self, X, y=None, sample_weight=None, skip_UQEstimator=False, **kwargs):

268 """Fit method that apply fit method of (predictor), UQEstimators, predict_KPI_processors, score_KPI_processors

269 Args:

270 X (np.array): Features

271 y (np.array): Targets/observations

272 sample_weight (np.array or None, optional): sample_weight. Defaults to None.

273 """

274

275 if self.preprocessor is not None:

276 self.preprocessor.fit(X)

277 X, y = self.preprocessor.transform(X)

278

279 y_bis = y

280

281 # --- Fit predictor if it's not direclty the UQEstimator

282 if self.predictor is not None:

283 if hasattr(

284 self.predictor, "is_fitted"

285 ): # To do replace with sklearn is_fitted check

286 pred = self.predictor.fit(X)

287 y_bis = y - pred

288

289 if self.save_models:

290 self.save()

291

292 if "res" not in self.type_UQ:

293 y_bis = y

294 else:

295 print(

296 "Warning use predictor with non residual base UQestimator > UQMeasure"

297 " will not based on model output"

298 )

299

300 # Call to fit method of model to perform multivarite fiting.

301

302 self._init_UQEstimator(X, y_bis)

303

304 # ------

305 # Format data

306 if hasattr(self.UQEstimator, "factory"): # Apply factory to transform input

307 generator = False

308 if (

309 (hasattr(self.UQEstimator, "training_parameters"))

310 and ("generator" in self.UQEstimator.training_parameters.keys())

311 and (self.UQEstimator.training_parameters["generator"])

312 ):

313 # Apply factory by batch during generator unfolding

314 generator = self.UQEstimator.training_parameters["generator"]

315 X_norm, y_bis_norm = X, y_bis

316

317 # But recover totality of model target to provide it to KPI-Score-KPIProcessor

318 _, y_bis, _ = self.UQEstimator.factory(None, y_bis, "transform")

319

320 else: # Apply factory

321 X_norm, y_bis_norm, w_ = self.UQEstimator.factory(X, y_bis, "transform")

322 y_bis = y_bis_norm

323

324 # Unormalized model target that can be multi-horizon object

325 _, y_bis = self.UQEstimator._format(None, y_bis, "inverse_transform")

326

327 else: # Data normalised is row data

328 X_norm, y_bis_norm = X, y_bis

329 # ---

330

331 if not (skip_UQEstimator):

332 self.UQEstimator.fit(

333 X_norm,

334 y_bis_norm,

335 sample_weight,

336 skip_format=True,

337 generator=generator,

338 **kwargs

339 )

340 self.is_fitted = True

341

342 pred = None

343 if self.predictor is not None:

344 pred = self.predictor.predict(X)

345

346 # Call to "fit" function of processor

347 pred_bis, UQ = self.UQEstimator.predict(

348 X_norm, skip_format=True, generator=generator, **kwargs

349 )

350

351 if pred is None:

352 pred = pred_bis

353

354 # Side effect fit KPI processor

355 self._fit_predict_KPI_processors(UQ, pred, y_bis, **kwargs)

356 self._fit_score_KPI_processors(UQ, pred, y_bis, **kwargs)

357

358 def fit_with_generator(self, data_generator, shuffle=True, **kwargs):

359 """Specific fit method that handle data_generator

360

361 Args:

362 data_generator (datagenerator): Iterative object that provide : X, y, sample_weight, x_split, context,

363 objective, source (see data_generator)

364 shuffle (bool, optional): use shuffle or not. Defaults to True.

365 """

366 for _n, dataset in enumerate(data_generator):

367 # Recovers data from data_generator

368 X, y, sample_weight, x_split, _context, _objective, _source = dataset

369 train = None

370 if x_split is not None:

371 train = x_split == 1

372 X, y = apply_mask(X, train), apply_mask(y, train)

373 if sample_weight is not None:

374 sample_weight = sample_weight[train]

375 # Called fit procedure for each dataset

376 self.fit(X, y, sample_weight=sample_weight, shuffle=shuffle, **kwargs)

377

378 def _transform_predict_KPI_processors(self, UQ, pred, **kwargs):

379 """Auxiliar method that apply transform method of predict_KPI_processors

380

381 Args:

382 UQ (np.array): UQmeasure from an UQestimator

383 pred (np.array): Predictor from an predictor or an UQestimator

384

385 Returns:

386 list_KPI_output: list_KPI_output pr KPI if len(list==1)

387 """

388 list_KPI_output = []

389 for processor in self.list_predict_KPI_processors:

390 KPI_output = processor.transform(

391 UQ=UQ,

392 type_UQ=self.type_UQ,

393 pred=pred,

394 y=None,

395 type_UQ_params=self.type_UQ_params,

396 **kwargs

397 )

398

399 list_KPI_output.append(KPI_output)

400

401 if len(list_KPI_output) == 1:

402 list_KPI_output = list_KPI_output[0]

403

404 return list_KPI_output

405

406 def predict(self, X, name_save="output.p", **kwargs):

407 """Predict method that apply predictor and UQestimators predict method, then compute predict_KPIs by use

408 transform methods of predict_KPI_Processors

409

410 Args:

411 X (np.array): feature

412 name_save (str, optional): file_name for (Predictor) and UQEstimator outputs save file.

413 Defaults to "output.p".

414

415 Returns:

416 pred, list_KPI_output : prediction and list of KPIs or KPI if len(list==1)

417 """

418

419 if self.preprocessor is not None:

420 X, _ = self.preprocessor.transform(X, training=False)

421

422 # --- Run predictor if it's not direclty the UQEstimator

423 pred = None

424 if self.predictor is not None:

425 pred = self.predictor.predict(X)

426 if self.save_result:

427 query = {"name": name_save + "_predictor"}

428 self.cache_manager.save(query, pred)

429

430 # --- Format data

431 if hasattr(self.UQEstimator, "factory"):

432

433 X_norm, _, _ = self.UQEstimator.factory(X, None)

434

435 generator = False

436 if (

437 (hasattr(self.UQEstimator, "training_parameters"))

438 and ("generator" in self.UQEstimator.training_parameters.keys())

439 and (self.UQEstimator.training_parameters["generator"])

440 ):

441 generator = True

442 X_norm = X

443 else:

444 generator = False

445 X_norm, _, _ = self.UQEstimator.factory(X, None)

446

447 else:

448 X_norm = X

449 # ---

450

451 pred_bis, UQ = self.UQEstimator.predict(

452 X_norm, skip_format=True, generator=generator, **kwargs

453 )

454

455 if pred is None:

456 pred = pred_bis

457

458 list_KPI_output = self._transform_predict_KPI_processors(

459 UQ=UQ, pred=pred, **kwargs

460 )

461

462 if self.reduc_filter is not None:

463 pred = apply_middledim_reduction(

464 pred, reduc_filter=self.reduc_filter, roll=self.roll

465 )

466

467 if self.save_result:

468 query = {"name": name_save + "_UQEstimator"}

469 self.cache_manager.save(query, (pred, UQ))

470 query = {"name": name_save + "_predict_KPI_processors"}

471 self.cache_manager.save(query, list_KPI_output)

472

473 return pred, list_KPI_output

474

475 def predict_with_generator(self, data_generator, **kwargs):

476 """specific predict method that handle data_generator

477

478 Args:

479 data_generator (datagenerator): Iterative object that provide : X, y, sample_weight, x_split, context,

480 objective, source (see data_generator)

481

482 Returns:

483 list of (pred, UQ)

484 """

485 list_pred = []

486 for _n, dataset in enumerate(data_generator):

487 # Recovers data from data_generator

488 X, _y, _sample_weight, _x_split, _context, _objective, source = dataset

489 list_pred.append(self.predict(X, name_save=source + "_output.p", **kwargs))

490 return list_pred

491

492 def _transform_score_KPI_processors(self, UQ, pred, y, **kwargs):

493 """auxilliair method that apply transform method of the score_KPI_processors

494

495 Args:

496 UQ (np.array): UQmeasure from an UQestimator

497 pred (np.array): Prediction for an predictot or UQestimator

498 y (np.array): Targets/Observation

499

500 Returns:

501 list_KPI_output: list_KPI_output or KPI_ouput if len(list)==1

502 """

503

504 list_KPI_output = []

505 for processor in self.list_score_KPI_processors:

506 list_KPI_output.append(

507 processor.transform(

508 UQ,

509 self.type_UQ,

510 pred,

511 y,

512 type_UQ_params=self.type_UQ_params,

513 **kwargs

514 )

515 )

516

517 if len(list_KPI_output) == 1:

518 list_KPI_output = list_KPI_output[0]

519

520 return list_KPI_output

521

522 def score(self, X, y=None, name_save="output", **kwargs):

523 """Score method that produce score KPI that transform y observation and (pred,UQ) UQestimator outputs

524 into a KPI according to score_KPI_processors

525

526 Args:

527 X (np.array): Features

528 y (np.array): Targets/obserbations

529

530 Returns:

531 list_KPI_output: list_KPI_output or KPI_ouput if len(list)==1

532 """

533 if self.preprocessor is not None:

534 X, y = self.preprocessor.transform(X)

535

536 y_bis = y

537

538 # Run predictor if distinct of the UQEstimator

539 pred = None

540 if self.predictor is not None:

541 # To do replace with sklearn is_fitted check

542 if hasattr(self.predictor, "is_fitted"):

543 pred = self.predictor.predict(X)

544 y_bis = y - pred

545

546 if self.save_result:

547 query = {"name": name_save + "_predictor"}

548 self.cache_manager.save(query, pred)

549

550 if "res" not in self.type_UQ:

551 y_bis = y

552 else:

553 print(

554 "Warning use predictor with non residual base UQestimator > UQMeasure"

555 " will not based on model output"

556 )

557

558 # ---

559 # Format data for UQEstimator distinct mode : With or without factory + with or without generator

560 if hasattr(self.UQEstimator, "factory"): # Apply factory to transform input

561 generator = False

562 if (

563 (hasattr(self.UQEstimator, "training_parameters"))

564 and ("generator" in self.UQEstimator.training_parameters.keys())

565 and (self.UQEstimator.training_parameters["generator"])

566 ):

567 # Apply factory by batch during generator unfolding

568 generator = self.UQEstimator.training_parameters["generator"]

569 X_norm, y_bis_norm = X, y

570

571 # But recover totality of model target to provide it to KPI-Score-KPIProcessor

572 _, y_bis, _ = self.UQEstimator.factory(None, y_bis, "transform")

573

574 else: # Apply factory

575 X_norm, y_bis_norm, w_ = self.UQEstimator.factory(X, y_bis, "transform")

576 y_bis = y_bis_norm

577

578 # Unormalized model target that can be multi-horizon object

579 _, y_bis = self.UQEstimator._format(None, y_bis, "inverse_transform")

580

581 else: # Data normalised is row data

582 X_norm, y_bis_norm = X, y_bis

583 # ---

584

585 pred_bis, UQ = self.UQEstimator.predict(

586 X_norm, skip_format=True, generator=generator, **kwargs

587 )

588

589 if pred is None:

590 pred = pred_bis

591

592 list_KPI_output = self._transform_score_KPI_processors(

593 UQ, pred, y_bis, **kwargs

594 )

595 if self.save_result:

596 query = {"name": name_save + "_UQEstimator"}

597 self.cache_manager.save(query, (pred_bis, UQ))

598 query = {"name": name_save + "_score_KPI_processors"}

599 self.cache_manager.save(query, list_KPI_output)

600 return list_KPI_output

601

602 def score_with_generator(self, data_generator, **kwargs):

603 """specitic score method that handle data_generator

604

605 Args:

606 data_generator (datagenerator): Iterative object that provide : X, y, sample_weight, x_split, context,

607 objective, source (see data_generator)

608

609 Returns:

610 list of score_KPI or score_KPI if len(list)==1

611 """

612 list_res = []

613 n_req = 0

614 for _n, (X, y, _, _, _, _, _source) in enumerate(data_generator):

615 res = self.score(X, y, **kwargs)

616 list_res.append(res)

617 n_req += 1

618

619 if n_req == 1:

620 list_res = list_res[0]

621

622 return list_res