smashbox.init.multimodel_statistics
Created on Tue Nov 4 13:23:42 2025
@author: maxime
1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3""" 4Created on Tue Nov 4 13:23:42 2025 5 6@author: maxime 7""" 8 9from smashbox.stats import mystats 10import numpy as np 11 12 13class multimodel_statistics: 14 """ 15 Class which handle computation of statistics for multi-model containers. Statistics for each model container must be calculated first. Mean, median, variance, min, max and the distribution of every scores of each model containers are computed. 16 """ 17 18 def __init__(self, parent_class): 19 self._parent_class = parent_class 20 """_parent_class attribute stores the parent_class src.init.smashbox.SmashBox() 21 to be able to access to the result of any model""" 22 self.target = "mysmashmodel" 23 """The target to plot. Default is mysmashmodel""" 24 25 self.multimodel_misfit_stats = _copy_stats_attr( 26 mystats.misfit_results() 27 ) 28 """Attribute misfit_stats_results stores the results of the multimodel statistics for the misfit.""" 29 30 self.multimodel_quantile_stats = None 31 """Attribute quantile_stats_results store the results of the multimodel statistics for the spatial quantile""" 32 33 self.multimodel_spatial_stats = _copy_stats_attr( 34 mystats.spatial_stats_results() 35 ) 36 """Atrribute spatial_stats_results store the results of the multimodel statistics for the spatial results""" 37 38 self.multimodel_outlets_stats_sim = _copy_stats_attr( 39 mystats.outlets_stats_results() 40 ) 41 """Atrribute outlets_stats store the resutls of the multimodel statistics for the simulted outlets stats""" 42 43 self.multimodel_outlets_stats_obs = _copy_stats_attr( 44 mystats.outlets_stats_results() 45 ) 46 """Atrribute outlets_stats store the resutls of the multimodel statistics for the observed outlets stats""" 47 48 def _get_model_list(self): 49 50 model_list = list() 51 52 for key, attr in self._parent_class.__dict__.items(): 53 54 if hasattr(attr, self.target): 55 56 current_model = getattr(self._parent_class, key) 57 58 if ( 59 current_model._myparam.param.bbox 60 == self._parent_class.myparam.param.bbox 61 ): 62 model_list.append(key) 63 64 return model_list 65 66 def compute_multimodel_statistics(self, model_list=None, target=None): 67 """ 68 Compute the multimodel statistics: mean, median, variance, min, max and the distribution. These statistics 69 are computed over all models containers which contains statistics and match the bounding box in object myparam. 70 71 :param model_list: list of the model container, defaults to None 72 :type model_list: list, optional 73 74 """ 75 76 if target in [ 77 "mysmashmodel", 78 "optimize_model", 79 "validation_model", 80 "warmup_model", 81 ]: 82 self.targe = target 83 84 if model_list is None: 85 model_list = self._get_model_list() 86 87 self.compute_multimodel_statistics_misfit(model_list=model_list) 88 self.compute_multimodel_statistics_outlets(model_list=model_list) 89 self.compute_multimodel_statistics_outlets( 90 model_list=model_list, obs=False 91 ) 92 self.compute_multimodel_statistics_outlets( 93 model_list=model_list, obs=True 94 ) 95 self.compute_multimodel_statistics_spatial(model_list=model_list) 96 self.compute_multimodel_statistics_quantile(model_list=model_list) 97 98 def compute_multimodel_statistics_misfit(self, model_list=None): 99 """ 100 Compute the multimodel statistics for the misfit criteria. 101 """ 102 if model_list is None: 103 model_list = self._get_model_list() 104 105 dict_results = {} 106 for model in model_list: 107 current_model = getattr(self._parent_class, model) 108 109 targeted_model = getattr(current_model, self.target) 110 misfit_list = targeted_model.mystats.misfit_stats.results.__dict__ 111 112 for misfit, results in misfit_list.items(): 113 114 if results is None: 115 116 results = ( 117 np.zeros( 118 shape=(len(current_model.mymesh.mesh["code"])) 119 ) 120 * np.nan 121 ) 122 print( 123 f"</> Warning misfit {misfit} is None for model {model}." 124 ) 125 126 if misfit in dict_results.keys(): 127 dict_results[misfit] = np.stack( 128 (dict_results[misfit], np.array(results)) 129 ) 130 else: 131 dict_results.update({misfit: np.array(results)}) 132 133 for key, values in dict_results.items(): 134 stats = getattr(self.multimodel_misfit_stats, key) 135 stats._fill_stats_attributes(values) 136 137 def compute_multimodel_statistics_outlets( 138 self, model_list=None, obs=False 139 ): 140 """ 141 Compute the multimodel statistics for the outlets criteria. 142 """ 143 if model_list is None: 144 model_list = self._get_model_list() 145 146 dict_results = {} 147 for model in model_list: 148 current_model = getattr(self._parent_class, model) 149 150 targeted_model = getattr(current_model, self.target) 151 152 if obs is False: 153 misfit_list = ( 154 targeted_model.mystats.outlets_stats.results_sim.__dict__ 155 ) 156 else: 157 misfit_list = ( 158 targeted_model.mystats.outlets_stats.results_obs.__dict__ 159 ) 160 161 for misfit, results in misfit_list.items(): 162 163 if results is None: 164 165 results = ( 166 np.zeros( 167 shape=(len(current_model.mymesh.mesh["code"])) 168 ) 169 * np.nan 170 ) 171 print( 172 f"</> Warning misfit {misfit} is None for model {model}." 173 ) 174 175 if misfit in dict_results.keys(): 176 dict_results[misfit] = np.stack( 177 (dict_results[misfit], np.array(results)) 178 ) 179 else: 180 dict_results.update({misfit: np.array(results)}) 181 182 for key, values in dict_results.items(): 183 if obs is False: 184 stats = getattr(self.multimodel_outlets_stats_sim, key) 185 else: 186 stats = getattr(self.multimodel_outlets_stats_obs, key) 187 stats._fill_stats_attributes(values) 188 189 def compute_multimodel_statistics_spatial(self, model_list=None): 190 """ 191 Compute the multimodel statistics for the spatial statisctics. 192 """ 193 if model_list is None: 194 model_list = self._get_model_list() 195 196 dict_results = {} 197 for model in model_list: 198 current_model = getattr(self._parent_class, model) 199 targeted_model = getattr(current_model, self.target) 200 201 stats_list = targeted_model.mystats.spatial_stats.results.__dict__ 202 203 for stats, results in stats_list.items(): 204 205 if results is None: 206 207 results = ( 208 np.zeros( 209 shape=( 210 ( 211 current_model.mymesh.mesh["nrow"], 212 current_model.mymesh.mesh["ncol"], 213 ) 214 ) 215 ) 216 * np.nan 217 ) 218 print( 219 f"</> Warning stats {stats} is None for model {model}." 220 ) 221 222 if stats in dict_results.keys(): 223 dict_results[stats] = np.stack( 224 (dict_results[stats], np.array(results)) 225 ) 226 else: 227 dict_results.update({stats: np.array(results)}) 228 229 for key, values in dict_results.items(): 230 stats = getattr(self.multimodel_spatial_stats, key) 231 stats._fill_stats_attributes(values) 232 233 def compute_multimodel_statistics_quantile(self, model_list=None): 234 """ 235 Compute the multimodel statistics for the quantile statistic. 236 """ 237 if model_list is None: 238 model_list = self._get_model_list() 239 240 quantile_matrix_dict_results = {} 241 spatial_dict_results = {} 242 vector_dict_results = {} 243 244 self.multimodel_quantile_stats = None 245 246 for model in model_list: 247 248 current_model = getattr(self._parent_class, model) 249 targeted_model = getattr(current_model, self.target) 250 251 if self.multimodel_quantile_stats is None: 252 self.multimodel_quantile_stats = _copy_quantile_attr( 253 targeted_model.mystats.quantile_stats 254 ) 255 256 all_quantile_results = ( 257 targeted_model.mystats.quantile_stats.__dict__ 258 ) 259 260 quantile_matrix_results = dict( 261 filter( 262 lambda key: not key[0].startswith("Quantile_"), 263 all_quantile_results.items(), 264 ) 265 ) 266 267 quantile_xxh_results = dict( 268 filter( 269 lambda key: key[0].startswith("Quantile_"), 270 all_quantile_results.items(), 271 ) 272 ) 273 274 for key, quantile_matrix in quantile_matrix_results.items(): 275 276 # if not key in quantile_matrix_dict_results.keys(): 277 # quantile_matrix_dict_results.update({key: {}}) 278 279 if quantile_matrix is None: 280 281 quantile_matrix = np.array([np.nan]) 282 print( 283 f"</> Warning quantile attr {key} is None for model {model}." 284 ) 285 286 if key in quantile_matrix_dict_results.keys(): 287 quantile_matrix_dict_results[key] = np.concat( 288 ( 289 quantile_matrix_dict_results[key], 290 np.array(quantile_matrix)[np.newaxis, :], 291 ), 292 axis=0, 293 ) 294 else: 295 quantile_matrix_dict_results.update( 296 {key: np.array(quantile_matrix)[np.newaxis, :]} 297 ) 298 299 for key, quantile_xxh in quantile_xxh_results.items(): 300 301 if not key in spatial_dict_results.keys(): 302 spatial_dict_results.update({key: {}}) 303 304 if not key in vector_dict_results.keys(): 305 vector_dict_results.update({key: {}}) 306 307 spatial_attr_list = [ 308 "Q_th", 309 "maxima", 310 "Umin", 311 "Umax", 312 "fit_shape", 313 "fit_scale", 314 "fit_loc", 315 ] 316 317 for attr in spatial_attr_list: 318 319 results = getattr(quantile_xxh, attr) 320 321 if results is None: 322 323 results = ( 324 np.zeros( 325 shape=( 326 ( 327 current_model.mymesh.mesh["nrow"], 328 current_model.mymesh.mesh["ncol"], 329 ) 330 ) 331 ) 332 * np.nan 333 ) 334 print( 335 f"</> Warning quantile attr {attr} is None for model {model}." 336 ) 337 338 if attr in spatial_dict_results[key].keys(): 339 spatial_dict_results[key][attr] = np.concat( 340 ( 341 spatial_dict_results[key][attr], 342 np.array(results)[np.newaxis, :], 343 ), 344 axis=0, 345 ) 346 else: 347 spatial_dict_results[key].update( 348 {attr: np.array(results)[np.newaxis, :]} 349 ) 350 351 vector_attr = [ 352 "T", 353 "T_emp", 354 "nb_chunks", 355 "chunk_size", 356 "fit", 357 "duration", 358 ] 359 360 for attr in vector_attr: 361 362 results = getattr(quantile_xxh_results[key], attr) 363 364 if results is None: 365 366 results = np.array([np.nan]) 367 print( 368 f"</> Warning attr {attr} is None for model {model}." 369 ) 370 371 if isinstance(results, str | int | float): 372 results = [results] 373 374 if attr in vector_dict_results[key].keys(): 375 vector_dict_results[key][attr] = np.concat( 376 ( 377 vector_dict_results[key][attr], 378 np.array(results)[np.newaxis, :], 379 ), 380 axis=0, 381 ) 382 else: 383 vector_dict_results[key].update( 384 {attr: np.array(results)[np.newaxis, :]} 385 ) 386 387 for key, values in quantile_matrix_dict_results.items(): 388 389 if np.any(values != np.nan): 390 stats = getattr(self.multimodel_quantile_stats, key) 391 stats._fill_stats_attributes(values) 392 393 for key, quantile in quantile_xxh_results.items(): 394 395 quantileXXh = getattr(self.multimodel_quantile_stats, key) 396 397 for var_name, values in spatial_dict_results[key].items(): 398 stats = getattr(quantileXXh, var_name) 399 stats._fill_stats_attributes(values) 400 401 for var_name, values in vector_dict_results[key].items(): 402 # stats = getattr(quantileXXh, var_name) 403 # stats = values 404 setattr(quantileXXh, var_name, values) 405 406 407class _copy_stats_attr: 408 409 def __init__(self, example_class): 410 411 for key in example_class.__dict__.keys(): 412 setattr(self, key, _statistics()) 413 414 415class _copy_quantile_attr: 416 417 def __init__(self, quantile_class): 418 419 for key in quantile_class.__dict__.keys(): 420 if key.startswith("Quantile_"): 421 setattr( 422 self, key, _copy_stats_attr(getattr(quantile_class, key)) 423 ) 424 else: 425 setattr(self, key, _statistics()) 426 427 428class _statistics: 429 """ 430 Class statistics for every statistics used for the multimodel statistics: mean, median, variance, min, max and data (the distribution) 431 """ 432 433 def __init__(self): 434 435 self.mean = None 436 437 self.median = None 438 439 self.variance = None 440 441 self.min = None 442 443 self.max = None 444 445 self.data = None 446 447 def _fill_stats_attributes(self, distrib): 448 self.mean = np.mean(distrib, axis=0) 449 self.median = np.median(distrib, axis=0) 450 self.variance = np.var(distrib, axis=0) 451 self.min = np.min(distrib, axis=0) 452 self.max = np.max(distrib, axis=0) 453 self.data = distrib 454 455 # def _fill_spatial_stats_attributes(self, matrix): 456 # self.mean = np.mean(matrix, axis=0) 457 # self.median = np.median(matrix, axis=0) 458 # self.variance = np.var(matrix, axis=0) 459 # self.min = np.min(matrix, axis=0) 460 # self.max = np.max(matrix, axis=0) 461 # self.data = matrix
class
multimodel_statistics:
14class multimodel_statistics: 15 """ 16 Class which handle computation of statistics for multi-model containers. Statistics for each model container must be calculated first. Mean, median, variance, min, max and the distribution of every scores of each model containers are computed. 17 """ 18 19 def __init__(self, parent_class): 20 self._parent_class = parent_class 21 """_parent_class attribute stores the parent_class src.init.smashbox.SmashBox() 22 to be able to access to the result of any model""" 23 self.target = "mysmashmodel" 24 """The target to plot. Default is mysmashmodel""" 25 26 self.multimodel_misfit_stats = _copy_stats_attr( 27 mystats.misfit_results() 28 ) 29 """Attribute misfit_stats_results stores the results of the multimodel statistics for the misfit.""" 30 31 self.multimodel_quantile_stats = None 32 """Attribute quantile_stats_results store the results of the multimodel statistics for the spatial quantile""" 33 34 self.multimodel_spatial_stats = _copy_stats_attr( 35 mystats.spatial_stats_results() 36 ) 37 """Atrribute spatial_stats_results store the results of the multimodel statistics for the spatial results""" 38 39 self.multimodel_outlets_stats_sim = _copy_stats_attr( 40 mystats.outlets_stats_results() 41 ) 42 """Atrribute outlets_stats store the resutls of the multimodel statistics for the simulted outlets stats""" 43 44 self.multimodel_outlets_stats_obs = _copy_stats_attr( 45 mystats.outlets_stats_results() 46 ) 47 """Atrribute outlets_stats store the resutls of the multimodel statistics for the observed outlets stats""" 48 49 def _get_model_list(self): 50 51 model_list = list() 52 53 for key, attr in self._parent_class.__dict__.items(): 54 55 if hasattr(attr, self.target): 56 57 current_model = getattr(self._parent_class, key) 58 59 if ( 60 current_model._myparam.param.bbox 61 == self._parent_class.myparam.param.bbox 62 ): 63 model_list.append(key) 64 65 return model_list 66 67 def compute_multimodel_statistics(self, model_list=None, target=None): 68 """ 69 Compute the multimodel statistics: mean, median, variance, min, max and the distribution. These statistics 70 are computed over all models containers which contains statistics and match the bounding box in object myparam. 71 72 :param model_list: list of the model container, defaults to None 73 :type model_list: list, optional 74 75 """ 76 77 if target in [ 78 "mysmashmodel", 79 "optimize_model", 80 "validation_model", 81 "warmup_model", 82 ]: 83 self.targe = target 84 85 if model_list is None: 86 model_list = self._get_model_list() 87 88 self.compute_multimodel_statistics_misfit(model_list=model_list) 89 self.compute_multimodel_statistics_outlets(model_list=model_list) 90 self.compute_multimodel_statistics_outlets( 91 model_list=model_list, obs=False 92 ) 93 self.compute_multimodel_statistics_outlets( 94 model_list=model_list, obs=True 95 ) 96 self.compute_multimodel_statistics_spatial(model_list=model_list) 97 self.compute_multimodel_statistics_quantile(model_list=model_list) 98 99 def compute_multimodel_statistics_misfit(self, model_list=None): 100 """ 101 Compute the multimodel statistics for the misfit criteria. 102 """ 103 if model_list is None: 104 model_list = self._get_model_list() 105 106 dict_results = {} 107 for model in model_list: 108 current_model = getattr(self._parent_class, model) 109 110 targeted_model = getattr(current_model, self.target) 111 misfit_list = targeted_model.mystats.misfit_stats.results.__dict__ 112 113 for misfit, results in misfit_list.items(): 114 115 if results is None: 116 117 results = ( 118 np.zeros( 119 shape=(len(current_model.mymesh.mesh["code"])) 120 ) 121 * np.nan 122 ) 123 print( 124 f"</> Warning misfit {misfit} is None for model {model}." 125 ) 126 127 if misfit in dict_results.keys(): 128 dict_results[misfit] = np.stack( 129 (dict_results[misfit], np.array(results)) 130 ) 131 else: 132 dict_results.update({misfit: np.array(results)}) 133 134 for key, values in dict_results.items(): 135 stats = getattr(self.multimodel_misfit_stats, key) 136 stats._fill_stats_attributes(values) 137 138 def compute_multimodel_statistics_outlets( 139 self, model_list=None, obs=False 140 ): 141 """ 142 Compute the multimodel statistics for the outlets criteria. 143 """ 144 if model_list is None: 145 model_list = self._get_model_list() 146 147 dict_results = {} 148 for model in model_list: 149 current_model = getattr(self._parent_class, model) 150 151 targeted_model = getattr(current_model, self.target) 152 153 if obs is False: 154 misfit_list = ( 155 targeted_model.mystats.outlets_stats.results_sim.__dict__ 156 ) 157 else: 158 misfit_list = ( 159 targeted_model.mystats.outlets_stats.results_obs.__dict__ 160 ) 161 162 for misfit, results in misfit_list.items(): 163 164 if results is None: 165 166 results = ( 167 np.zeros( 168 shape=(len(current_model.mymesh.mesh["code"])) 169 ) 170 * np.nan 171 ) 172 print( 173 f"</> Warning misfit {misfit} is None for model {model}." 174 ) 175 176 if misfit in dict_results.keys(): 177 dict_results[misfit] = np.stack( 178 (dict_results[misfit], np.array(results)) 179 ) 180 else: 181 dict_results.update({misfit: np.array(results)}) 182 183 for key, values in dict_results.items(): 184 if obs is False: 185 stats = getattr(self.multimodel_outlets_stats_sim, key) 186 else: 187 stats = getattr(self.multimodel_outlets_stats_obs, key) 188 stats._fill_stats_attributes(values) 189 190 def compute_multimodel_statistics_spatial(self, model_list=None): 191 """ 192 Compute the multimodel statistics for the spatial statisctics. 193 """ 194 if model_list is None: 195 model_list = self._get_model_list() 196 197 dict_results = {} 198 for model in model_list: 199 current_model = getattr(self._parent_class, model) 200 targeted_model = getattr(current_model, self.target) 201 202 stats_list = targeted_model.mystats.spatial_stats.results.__dict__ 203 204 for stats, results in stats_list.items(): 205 206 if results is None: 207 208 results = ( 209 np.zeros( 210 shape=( 211 ( 212 current_model.mymesh.mesh["nrow"], 213 current_model.mymesh.mesh["ncol"], 214 ) 215 ) 216 ) 217 * np.nan 218 ) 219 print( 220 f"</> Warning stats {stats} is None for model {model}." 221 ) 222 223 if stats in dict_results.keys(): 224 dict_results[stats] = np.stack( 225 (dict_results[stats], np.array(results)) 226 ) 227 else: 228 dict_results.update({stats: np.array(results)}) 229 230 for key, values in dict_results.items(): 231 stats = getattr(self.multimodel_spatial_stats, key) 232 stats._fill_stats_attributes(values) 233 234 def compute_multimodel_statistics_quantile(self, model_list=None): 235 """ 236 Compute the multimodel statistics for the quantile statistic. 237 """ 238 if model_list is None: 239 model_list = self._get_model_list() 240 241 quantile_matrix_dict_results = {} 242 spatial_dict_results = {} 243 vector_dict_results = {} 244 245 self.multimodel_quantile_stats = None 246 247 for model in model_list: 248 249 current_model = getattr(self._parent_class, model) 250 targeted_model = getattr(current_model, self.target) 251 252 if self.multimodel_quantile_stats is None: 253 self.multimodel_quantile_stats = _copy_quantile_attr( 254 targeted_model.mystats.quantile_stats 255 ) 256 257 all_quantile_results = ( 258 targeted_model.mystats.quantile_stats.__dict__ 259 ) 260 261 quantile_matrix_results = dict( 262 filter( 263 lambda key: not key[0].startswith("Quantile_"), 264 all_quantile_results.items(), 265 ) 266 ) 267 268 quantile_xxh_results = dict( 269 filter( 270 lambda key: key[0].startswith("Quantile_"), 271 all_quantile_results.items(), 272 ) 273 ) 274 275 for key, quantile_matrix in quantile_matrix_results.items(): 276 277 # if not key in quantile_matrix_dict_results.keys(): 278 # quantile_matrix_dict_results.update({key: {}}) 279 280 if quantile_matrix is None: 281 282 quantile_matrix = np.array([np.nan]) 283 print( 284 f"</> Warning quantile attr {key} is None for model {model}." 285 ) 286 287 if key in quantile_matrix_dict_results.keys(): 288 quantile_matrix_dict_results[key] = np.concat( 289 ( 290 quantile_matrix_dict_results[key], 291 np.array(quantile_matrix)[np.newaxis, :], 292 ), 293 axis=0, 294 ) 295 else: 296 quantile_matrix_dict_results.update( 297 {key: np.array(quantile_matrix)[np.newaxis, :]} 298 ) 299 300 for key, quantile_xxh in quantile_xxh_results.items(): 301 302 if not key in spatial_dict_results.keys(): 303 spatial_dict_results.update({key: {}}) 304 305 if not key in vector_dict_results.keys(): 306 vector_dict_results.update({key: {}}) 307 308 spatial_attr_list = [ 309 "Q_th", 310 "maxima", 311 "Umin", 312 "Umax", 313 "fit_shape", 314 "fit_scale", 315 "fit_loc", 316 ] 317 318 for attr in spatial_attr_list: 319 320 results = getattr(quantile_xxh, attr) 321 322 if results is None: 323 324 results = ( 325 np.zeros( 326 shape=( 327 ( 328 current_model.mymesh.mesh["nrow"], 329 current_model.mymesh.mesh["ncol"], 330 ) 331 ) 332 ) 333 * np.nan 334 ) 335 print( 336 f"</> Warning quantile attr {attr} is None for model {model}." 337 ) 338 339 if attr in spatial_dict_results[key].keys(): 340 spatial_dict_results[key][attr] = np.concat( 341 ( 342 spatial_dict_results[key][attr], 343 np.array(results)[np.newaxis, :], 344 ), 345 axis=0, 346 ) 347 else: 348 spatial_dict_results[key].update( 349 {attr: np.array(results)[np.newaxis, :]} 350 ) 351 352 vector_attr = [ 353 "T", 354 "T_emp", 355 "nb_chunks", 356 "chunk_size", 357 "fit", 358 "duration", 359 ] 360 361 for attr in vector_attr: 362 363 results = getattr(quantile_xxh_results[key], attr) 364 365 if results is None: 366 367 results = np.array([np.nan]) 368 print( 369 f"</> Warning attr {attr} is None for model {model}." 370 ) 371 372 if isinstance(results, str | int | float): 373 results = [results] 374 375 if attr in vector_dict_results[key].keys(): 376 vector_dict_results[key][attr] = np.concat( 377 ( 378 vector_dict_results[key][attr], 379 np.array(results)[np.newaxis, :], 380 ), 381 axis=0, 382 ) 383 else: 384 vector_dict_results[key].update( 385 {attr: np.array(results)[np.newaxis, :]} 386 ) 387 388 for key, values in quantile_matrix_dict_results.items(): 389 390 if np.any(values != np.nan): 391 stats = getattr(self.multimodel_quantile_stats, key) 392 stats._fill_stats_attributes(values) 393 394 for key, quantile in quantile_xxh_results.items(): 395 396 quantileXXh = getattr(self.multimodel_quantile_stats, key) 397 398 for var_name, values in spatial_dict_results[key].items(): 399 stats = getattr(quantileXXh, var_name) 400 stats._fill_stats_attributes(values) 401 402 for var_name, values in vector_dict_results[key].items(): 403 # stats = getattr(quantileXXh, var_name) 404 # stats = values 405 setattr(quantileXXh, var_name, values)
Class which handle computation of statistics for multi-model containers. Statistics for each model container must be calculated first. Mean, median, variance, min, max and the distribution of every scores of each model containers are computed.
multimodel_statistics(parent_class)
19 def __init__(self, parent_class): 20 self._parent_class = parent_class 21 """_parent_class attribute stores the parent_class src.init.smashbox.SmashBox() 22 to be able to access to the result of any model""" 23 self.target = "mysmashmodel" 24 """The target to plot. Default is mysmashmodel""" 25 26 self.multimodel_misfit_stats = _copy_stats_attr( 27 mystats.misfit_results() 28 ) 29 """Attribute misfit_stats_results stores the results of the multimodel statistics for the misfit.""" 30 31 self.multimodel_quantile_stats = None 32 """Attribute quantile_stats_results store the results of the multimodel statistics for the spatial quantile""" 33 34 self.multimodel_spatial_stats = _copy_stats_attr( 35 mystats.spatial_stats_results() 36 ) 37 """Atrribute spatial_stats_results store the results of the multimodel statistics for the spatial results""" 38 39 self.multimodel_outlets_stats_sim = _copy_stats_attr( 40 mystats.outlets_stats_results() 41 ) 42 """Atrribute outlets_stats store the resutls of the multimodel statistics for the simulted outlets stats""" 43 44 self.multimodel_outlets_stats_obs = _copy_stats_attr( 45 mystats.outlets_stats_results() 46 ) 47 """Atrribute outlets_stats store the resutls of the multimodel statistics for the observed outlets stats"""
multimodel_misfit_stats
Attribute misfit_stats_results stores the results of the multimodel statistics for the misfit.
multimodel_quantile_stats
Attribute quantile_stats_results store the results of the multimodel statistics for the spatial quantile
multimodel_spatial_stats
Atrribute spatial_stats_results store the results of the multimodel statistics for the spatial results
multimodel_outlets_stats_sim
Atrribute outlets_stats store the resutls of the multimodel statistics for the simulted outlets stats
multimodel_outlets_stats_obs
Atrribute outlets_stats store the resutls of the multimodel statistics for the observed outlets stats
def
compute_multimodel_statistics(self, model_list=None, target=None):
67 def compute_multimodel_statistics(self, model_list=None, target=None): 68 """ 69 Compute the multimodel statistics: mean, median, variance, min, max and the distribution. These statistics 70 are computed over all models containers which contains statistics and match the bounding box in object myparam. 71 72 :param model_list: list of the model container, defaults to None 73 :type model_list: list, optional 74 75 """ 76 77 if target in [ 78 "mysmashmodel", 79 "optimize_model", 80 "validation_model", 81 "warmup_model", 82 ]: 83 self.targe = target 84 85 if model_list is None: 86 model_list = self._get_model_list() 87 88 self.compute_multimodel_statistics_misfit(model_list=model_list) 89 self.compute_multimodel_statistics_outlets(model_list=model_list) 90 self.compute_multimodel_statistics_outlets( 91 model_list=model_list, obs=False 92 ) 93 self.compute_multimodel_statistics_outlets( 94 model_list=model_list, obs=True 95 ) 96 self.compute_multimodel_statistics_spatial(model_list=model_list) 97 self.compute_multimodel_statistics_quantile(model_list=model_list)
Compute the multimodel statistics: mean, median, variance, min, max and the distribution. These statistics are computed over all models containers which contains statistics and match the bounding box in object myparam.
Parameters
- model_list: list of the model container, defaults to None
def
compute_multimodel_statistics_misfit(self, model_list=None):
99 def compute_multimodel_statistics_misfit(self, model_list=None): 100 """ 101 Compute the multimodel statistics for the misfit criteria. 102 """ 103 if model_list is None: 104 model_list = self._get_model_list() 105 106 dict_results = {} 107 for model in model_list: 108 current_model = getattr(self._parent_class, model) 109 110 targeted_model = getattr(current_model, self.target) 111 misfit_list = targeted_model.mystats.misfit_stats.results.__dict__ 112 113 for misfit, results in misfit_list.items(): 114 115 if results is None: 116 117 results = ( 118 np.zeros( 119 shape=(len(current_model.mymesh.mesh["code"])) 120 ) 121 * np.nan 122 ) 123 print( 124 f"</> Warning misfit {misfit} is None for model {model}." 125 ) 126 127 if misfit in dict_results.keys(): 128 dict_results[misfit] = np.stack( 129 (dict_results[misfit], np.array(results)) 130 ) 131 else: 132 dict_results.update({misfit: np.array(results)}) 133 134 for key, values in dict_results.items(): 135 stats = getattr(self.multimodel_misfit_stats, key) 136 stats._fill_stats_attributes(values)
Compute the multimodel statistics for the misfit criteria.
def
compute_multimodel_statistics_outlets(self, model_list=None, obs=False):
138 def compute_multimodel_statistics_outlets( 139 self, model_list=None, obs=False 140 ): 141 """ 142 Compute the multimodel statistics for the outlets criteria. 143 """ 144 if model_list is None: 145 model_list = self._get_model_list() 146 147 dict_results = {} 148 for model in model_list: 149 current_model = getattr(self._parent_class, model) 150 151 targeted_model = getattr(current_model, self.target) 152 153 if obs is False: 154 misfit_list = ( 155 targeted_model.mystats.outlets_stats.results_sim.__dict__ 156 ) 157 else: 158 misfit_list = ( 159 targeted_model.mystats.outlets_stats.results_obs.__dict__ 160 ) 161 162 for misfit, results in misfit_list.items(): 163 164 if results is None: 165 166 results = ( 167 np.zeros( 168 shape=(len(current_model.mymesh.mesh["code"])) 169 ) 170 * np.nan 171 ) 172 print( 173 f"</> Warning misfit {misfit} is None for model {model}." 174 ) 175 176 if misfit in dict_results.keys(): 177 dict_results[misfit] = np.stack( 178 (dict_results[misfit], np.array(results)) 179 ) 180 else: 181 dict_results.update({misfit: np.array(results)}) 182 183 for key, values in dict_results.items(): 184 if obs is False: 185 stats = getattr(self.multimodel_outlets_stats_sim, key) 186 else: 187 stats = getattr(self.multimodel_outlets_stats_obs, key) 188 stats._fill_stats_attributes(values)
Compute the multimodel statistics for the outlets criteria.
def
compute_multimodel_statistics_spatial(self, model_list=None):
190 def compute_multimodel_statistics_spatial(self, model_list=None): 191 """ 192 Compute the multimodel statistics for the spatial statisctics. 193 """ 194 if model_list is None: 195 model_list = self._get_model_list() 196 197 dict_results = {} 198 for model in model_list: 199 current_model = getattr(self._parent_class, model) 200 targeted_model = getattr(current_model, self.target) 201 202 stats_list = targeted_model.mystats.spatial_stats.results.__dict__ 203 204 for stats, results in stats_list.items(): 205 206 if results is None: 207 208 results = ( 209 np.zeros( 210 shape=( 211 ( 212 current_model.mymesh.mesh["nrow"], 213 current_model.mymesh.mesh["ncol"], 214 ) 215 ) 216 ) 217 * np.nan 218 ) 219 print( 220 f"</> Warning stats {stats} is None for model {model}." 221 ) 222 223 if stats in dict_results.keys(): 224 dict_results[stats] = np.stack( 225 (dict_results[stats], np.array(results)) 226 ) 227 else: 228 dict_results.update({stats: np.array(results)}) 229 230 for key, values in dict_results.items(): 231 stats = getattr(self.multimodel_spatial_stats, key) 232 stats._fill_stats_attributes(values)
Compute the multimodel statistics for the spatial statisctics.
def
compute_multimodel_statistics_quantile(self, model_list=None):
234 def compute_multimodel_statistics_quantile(self, model_list=None): 235 """ 236 Compute the multimodel statistics for the quantile statistic. 237 """ 238 if model_list is None: 239 model_list = self._get_model_list() 240 241 quantile_matrix_dict_results = {} 242 spatial_dict_results = {} 243 vector_dict_results = {} 244 245 self.multimodel_quantile_stats = None 246 247 for model in model_list: 248 249 current_model = getattr(self._parent_class, model) 250 targeted_model = getattr(current_model, self.target) 251 252 if self.multimodel_quantile_stats is None: 253 self.multimodel_quantile_stats = _copy_quantile_attr( 254 targeted_model.mystats.quantile_stats 255 ) 256 257 all_quantile_results = ( 258 targeted_model.mystats.quantile_stats.__dict__ 259 ) 260 261 quantile_matrix_results = dict( 262 filter( 263 lambda key: not key[0].startswith("Quantile_"), 264 all_quantile_results.items(), 265 ) 266 ) 267 268 quantile_xxh_results = dict( 269 filter( 270 lambda key: key[0].startswith("Quantile_"), 271 all_quantile_results.items(), 272 ) 273 ) 274 275 for key, quantile_matrix in quantile_matrix_results.items(): 276 277 # if not key in quantile_matrix_dict_results.keys(): 278 # quantile_matrix_dict_results.update({key: {}}) 279 280 if quantile_matrix is None: 281 282 quantile_matrix = np.array([np.nan]) 283 print( 284 f"</> Warning quantile attr {key} is None for model {model}." 285 ) 286 287 if key in quantile_matrix_dict_results.keys(): 288 quantile_matrix_dict_results[key] = np.concat( 289 ( 290 quantile_matrix_dict_results[key], 291 np.array(quantile_matrix)[np.newaxis, :], 292 ), 293 axis=0, 294 ) 295 else: 296 quantile_matrix_dict_results.update( 297 {key: np.array(quantile_matrix)[np.newaxis, :]} 298 ) 299 300 for key, quantile_xxh in quantile_xxh_results.items(): 301 302 if not key in spatial_dict_results.keys(): 303 spatial_dict_results.update({key: {}}) 304 305 if not key in vector_dict_results.keys(): 306 vector_dict_results.update({key: {}}) 307 308 spatial_attr_list = [ 309 "Q_th", 310 "maxima", 311 "Umin", 312 "Umax", 313 "fit_shape", 314 "fit_scale", 315 "fit_loc", 316 ] 317 318 for attr in spatial_attr_list: 319 320 results = getattr(quantile_xxh, attr) 321 322 if results is None: 323 324 results = ( 325 np.zeros( 326 shape=( 327 ( 328 current_model.mymesh.mesh["nrow"], 329 current_model.mymesh.mesh["ncol"], 330 ) 331 ) 332 ) 333 * np.nan 334 ) 335 print( 336 f"</> Warning quantile attr {attr} is None for model {model}." 337 ) 338 339 if attr in spatial_dict_results[key].keys(): 340 spatial_dict_results[key][attr] = np.concat( 341 ( 342 spatial_dict_results[key][attr], 343 np.array(results)[np.newaxis, :], 344 ), 345 axis=0, 346 ) 347 else: 348 spatial_dict_results[key].update( 349 {attr: np.array(results)[np.newaxis, :]} 350 ) 351 352 vector_attr = [ 353 "T", 354 "T_emp", 355 "nb_chunks", 356 "chunk_size", 357 "fit", 358 "duration", 359 ] 360 361 for attr in vector_attr: 362 363 results = getattr(quantile_xxh_results[key], attr) 364 365 if results is None: 366 367 results = np.array([np.nan]) 368 print( 369 f"</> Warning attr {attr} is None for model {model}." 370 ) 371 372 if isinstance(results, str | int | float): 373 results = [results] 374 375 if attr in vector_dict_results[key].keys(): 376 vector_dict_results[key][attr] = np.concat( 377 ( 378 vector_dict_results[key][attr], 379 np.array(results)[np.newaxis, :], 380 ), 381 axis=0, 382 ) 383 else: 384 vector_dict_results[key].update( 385 {attr: np.array(results)[np.newaxis, :]} 386 ) 387 388 for key, values in quantile_matrix_dict_results.items(): 389 390 if np.any(values != np.nan): 391 stats = getattr(self.multimodel_quantile_stats, key) 392 stats._fill_stats_attributes(values) 393 394 for key, quantile in quantile_xxh_results.items(): 395 396 quantileXXh = getattr(self.multimodel_quantile_stats, key) 397 398 for var_name, values in spatial_dict_results[key].items(): 399 stats = getattr(quantileXXh, var_name) 400 stats._fill_stats_attributes(values) 401 402 for var_name, values in vector_dict_results[key].items(): 403 # stats = getattr(quantileXXh, var_name) 404 # stats = values 405 setattr(quantileXXh, var_name, values)
Compute the multimodel statistics for the quantile statistic.