Coverage for uqmodels/UQModel.py: 67%

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

9from sklearn.base import BaseEstimator

11import uqmodels.modelization.ML_estimator.random_forest_UQ as RF_UQ

12from uqmodels.processing import Cache_manager

13from uqmodels.utils import add_random_state, apply_mask, apply_middledim_reduction

16class UQModel(BaseEstimator):

17 """

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

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

20 """

22 def __init__(

23 self,

24 UQEstimator_initializer=RF_UQ.RF_UQEstimator,

25 UQEstimator_params=RF_UQ.get_params_dict(),

26 # Necesite with_prediction or to provide a predictor.

27 name="UQModel",

28 predictor=None,

29 preprocessor=None,

30 list_predict_KPI_processors=None, # List of UQProcessor

31 list_score_KPI_processors=None,

32 reduc_filter=None,

33 roll=None,

34 cache_manager=Cache_manager(),

35 save_result=False,

36 save_models=False,

37 random_state=None,

38 **kwargs

39 ):

40 """Initialize UQModel class

42 Args:

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

44 UQEstimator_params (dict_params): params for the init method

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

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

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

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

49 predict_KPI_processors. Defaults to None.

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

51 score_KPI_processors. Defaults to None

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

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

54 random_state : Controls randoms

55 """

57 self.name = name

59 self.predictor = predictor

60 if predictor is not None:

61 self.predictor_initializer = predictor.__init__()

62 self.predictor_params = predictor.get_params()

64 self.UQEstimator_initializer = UQEstimator_initializer

65 self.preprocessor = preprocessor

66 self.UQEstimator_params = UQEstimator_params

67 self.list_predict_KPI_processors = list_predict_KPI_processors

68 self.list_score_KPI_processors = list_score_KPI_processors

69 self.reduc_filter = reduc_filter

70 self.roll = roll

71 self.cache_manager = cache_manager

72 self.random_state = random_state

73 if self.random_state is not None:

74 self.UQEstimator_params["random_state"] = random_state

76 for n, predict_KPI_processor in enumerate(list_predict_KPI_processors):

77 predict_KPI_processor.random_state = add_random_state(random_state, n)

79 for n, score_KPI_processor in enumerate(list_score_KPI_processors):

80 score_KPI_processor.random_state = add_random_state(

81 random_state, 10 + n

82 )

84 self.UQEstimator = None

85 self.is_fitted = None

86 self.type_UQ = None

87 self.type_UQ_params = None

88 self.save_result = save_result

89 self.save_models = save_models

91 for key_arg in kwargs.keys():

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

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

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

96 or force save using UQModel cache_manager

98 Todo:

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

100

101 Args:

102 entity (obj): Object to load

103 path (str): Path to load

104 name (str): name to load

105 """

106 if hasattr(entity, "load"):

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

108 if "name" in arg_list:

109 entity.load(path, name=name)

110 else:

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

112 entity.load(path)

113 else:

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

115 entity = self.cache_manager.load(query)

116 return entity

117

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

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

120

121 Todo:

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

123

124 Args:

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

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

127 """

128 if path is None:

129 path = self.cache_manager.storing

130

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

132 _UQModel = self.cache_manager.load(query)

133

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

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

136

137 if self.predictor:

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

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

140

141 if self.UQEstimator:

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

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

144

145 # TO do :

146 # Specific load procedure for predict_KPI_processors and score_KPI_processors

147 # Specific load procedure for preprocessor object

148 # Hold currently by global __dict__ parameters save

149

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

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

152 method or force save using UQmMdels cache_manager

153

154 Todo:

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

156

157 Args:

158 entity (obj): Object to save

159 path (str): Path to save

160 name (str): name to save

161 """

162 if UQModel_name is None:

163 UQModel_name = self.name

164

165 if hasattr(entity, "save"):

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

167 if "name" in arg_list:

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

169 entity.save(path, name=name)

170 else:

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

172 entity.save(path)

173 else:

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

175 if self.cache_manager is not None:

176 self.cache_manager.save(query, entity)

177 else:

178 print("No_cache_manager : cannot save")

179

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

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

182

183 Todo:

184 * Raise fail error

185

186 Args:

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

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

189 """

190 if name is None:

191 name = self.name

192

193 if path is None:

194 path = self.cache_manager.storing

195

196 if self.predictor is not None:

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

198 tmp_predictor = self.predictor

199 self.predictor = True

200

201 if self.UQEstimator:

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

203 tmp_UQestimator = self.UQEstimator

204 self.UQEstimator = True

205

206 # TO do :

207 # Specific save procedure for predict_KPI_processors and score_KPI_processors

208 # Specific save procedure for preprocessor object

209 # Hold currently by global __dict__ parameters save

210

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

212 self.cache_manager.save(query, self)

213

214 if self.predictor is not None:

215 self.predictor = tmp_predictor

216

217 if self.UQEstimator:

218 self.UQEstimator = tmp_UQestimator

219

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

221 if not self.UQEstimator:

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

223

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

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

226

227 self.type_UQ = self.UQEstimator.type_UQ

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

229 self.type_UQ_params = self.UQEstimator.type_UQ_params

230

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

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

233

234 Args:

235 UQ (np.array): Features

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

237 y (np.array): Targets/observations

238 """

239 for processor in self.list_predict_KPI_processors:

240 processor.fit(

241 UQ,

242 self.type_UQ,

243 pred,

244 y=y,

245 type_UQ_params=self.type_UQ_params,

246 **kwargs

247 )

248

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

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

251

252 Args:

253 UQ (np.array): Features

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

255 y (np.array): Targets/observations

256 """

257 for processor in self.list_score_KPI_processors:

258 processor.fit(

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

260 )

261

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

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

264 Args:

265 X (np.array): Features

266 y (np.array): Targets/observations

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

268 """

269

270 if self.preprocessor is not None:

271 self.preprocessor.fit(X)

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

273

274 y_bis = y

275

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

277 if self.predictor is not None:

278 if hasattr(

279 self.predictor, "is_fitted"

280 ): # To do replace with sklearn is_fitted check

281 pred = self.predictor.fit(X)

282 y_bis = y - pred

283

284 if self.save_models:

285 self.save()

286

287 if "res" not in self.type_UQ:

288 y_bis = y

289 else:

290 print(

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

292 " will not based on model output"

293 )

294

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

296

297 self._init_UQEstimator(X, y_bis)

298

299 # ------

300 # Format data

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

302 generator = False

303 if (

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

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

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

307 ):

308 # Apply factory by batch during generator unfolding

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

310 X_norm, y_bis_norm = X, y_bis

311

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

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

314

315 else: # Apply factory

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

317 y_bis = y_bis_norm

318

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

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

321

322 else: # Data normalised is row data

323 X_norm, y_bis_norm = X, y_bis

324 # ---

325

326 if not skip_UQEstimator:

327 self.UQEstimator.fit(

328 X_norm,

329 y_bis_norm,

330 sample_weight,

331 skip_format=True,

332 generator=generator,

333 **kwargs

334 )

335 self.is_fitted = True

336

337 pred = None

338 if self.predictor is not None:

339 pred = self.predictor.predict(X)

340

341 # Call to "fit" function of processor

342 pred_bis, UQ = self.UQEstimator.predict(

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

344 )

345

346 if pred is None:

347 pred = pred_bis

348

349 # Side effect fit KPI processor

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

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

352

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

354 """Specific fit method that handle data_generator

355

356 Args:

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

358 objective, source (see data_generator)

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

360 """

361 for _n, dataset in enumerate(data_generator):

362 # Recovers data from data_generator

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

364 train = None

365 if x_split is not None:

366 train = x_split == 1

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

368 if sample_weight is not None:

369 sample_weight = sample_weight[train]

370 # Called fit procedure for each dataset

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

372

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

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

375

376 Args:

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

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

379

380 Returns:

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

382 """

383 list_KPI_output = []

384 for processor in self.list_predict_KPI_processors:

385 KPI_output = processor.transform(

386 UQ=UQ,

387 type_UQ=self.type_UQ,

388 pred=pred,

389 y=None,

390 type_UQ_params=self.type_UQ_params,

391 **kwargs

392 )

393

394 list_KPI_output.append(KPI_output)

395

396 if len(list_KPI_output) == 1:

397 list_KPI_output = list_KPI_output[0]

398

399 return list_KPI_output

400

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

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

403 transform methods of predict_KPI_Processors

404

405 Args:

406 X (np.array): feature

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

408 Defaults to "output.p".

409

410 Returns:

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

412 """

413

414 if self.preprocessor is not None:

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

416

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

418 pred = None

419 if self.predictor is not None:

420 pred = self.predictor.predict(X)

421 if self.save_result:

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

423 self.cache_manager.save(query, pred)

424

425 # --- Format data

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

427

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

429

430 generator = False

431 if (

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

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

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

435 ):

436 generator = True

437 X_norm = X

438 else:

439 generator = False

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

441

442 else:

443 X_norm = X

444 # ---

445

446 pred_bis, UQ = self.UQEstimator.predict(

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

448 )

449

450 if pred is None:

451 pred = pred_bis

452

453 list_KPI_output = self._transform_predict_KPI_processors(

454 UQ=UQ, pred=pred, **kwargs

455 )

456

457 if self.reduc_filter is not None:

458 pred = apply_middledim_reduction(

459 pred, reduc_filter=self.reduc_filter, roll=self.roll

460 )

461

462 if self.save_result:

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

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

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

466 self.cache_manager.save(query, list_KPI_output)

467

468 return pred, list_KPI_output

469

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

471 """specific predict method that handle data_generator

472

473 Args:

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

475 objective, source (see data_generator)

476

477 Returns:

478 list of (pred, UQ)

479 """

480 list_pred = []

481 for _n, dataset in enumerate(data_generator):

482 # Recovers data from data_generator

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

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

485 return list_pred

486

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

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

489

490 Args:

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

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

493 y (np.array): Targets/Observation

494

495 Returns:

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

497 """

498

499 list_KPI_output = []

500 for processor in self.list_score_KPI_processors:

501 list_KPI_output.append(

502 processor.transform(

503 UQ,

504 self.type_UQ,

505 pred,

506 y,

507 type_UQ_params=self.type_UQ_params,

508 **kwargs

509 )

510 )

511

512 if len(list_KPI_output) == 1:

513 list_KPI_output = list_KPI_output[0]

514

515 return list_KPI_output

516

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

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

519 into a KPI according to score_KPI_processors

520

521 Args:

522 X (np.array): Features

523 y (np.array): Targets/obserbations

524

525 Returns:

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

527 """

528 if self.preprocessor is not None:

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

530

531 y_bis = y

532

533 # Run predictor if distinct of the UQEstimator

534 pred = None

535 if self.predictor is not None:

536 # To do replace with sklearn is_fitted check

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

538 pred = self.predictor.predict(X)

539 y_bis = y - pred

540

541 if self.save_result:

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

543 self.cache_manager.save(query, pred)

544

545 if "res" not in self.type_UQ:

546 y_bis = y

547 else:

548 print(

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

550 " will not based on model output"

551 )

552

553 # ---

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

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

556 generator = False

557 if (

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

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

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

561 ):

562 # Apply factory by batch during generator unfolding

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

564 X_norm, y_bis_norm = X, y

565

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

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

568

569 else: # Apply factory

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

571 y_bis = y_bis_norm

572

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

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

575

576 else: # Data normalised is row data

577 X_norm, y_bis_norm = X, y_bis

578 # ---

579

580 pred_bis, UQ = self.UQEstimator.predict(

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

582 )

583

584 if pred is None:

585 pred = pred_bis

586

587 list_KPI_output = self._transform_score_KPI_processors(

588 UQ, pred, y_bis, **kwargs

589 )

590 if self.save_result:

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

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

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

594 self.cache_manager.save(query, list_KPI_output)

595 return list_KPI_output

596

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

598 """specitic score method that handle data_generator

599

600 Args:

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

602 objective, source (see data_generator)

603

604 Returns:

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

606 """

607 list_res = []

608 n_req = 0

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

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

611 list_res.append(res)

612 n_req += 1

613

614 if n_req == 1:

615 list_res = list_res[0]

616

617 return list_res

Coverage for uqmodels / UQModel.py: 67%

248 statements