Coverage for uqmodels / postprocessing / custom_UQKPI_Processor.py: 15%
55 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-09 08:15 +0000
1import numpy as np
2import uqmodels.postprocessing.anomaly_processing as anom_proc
3import uqmodels.postprocessing.UQKPI_Processor as UQKPI_proc
6class Multiscale_Anomscore_processor(UQKPI_proc.UQKPI_Processor):
7 """Processor aiming to produce an contextual deviation anomaly score from UQmeasure prediction and observation"""
9 def __init__(
10 self,
11 name="Multiscale_anomscore",
12 KPI_parameters=dict(),
13 cache=None,
14 random_state=None,
15 **kwargs
16 ):
17 """Initialization
19 Args:
20 name (str, optional): name. Defaults to 'Anomscore'.
21 KPI_parameters (_type_, optional): dict of parameters link to compute_score function.
22 Defaults to None will use predefined paramters.
23 cache (_type_, optional): Cache manager or none
24 """
26 KPI_parameters_default = {
27 "type_norm": "Nsigma_local",
28 "q_var": 1,
29 "d": 2,
30 "beta": 0.001,
31 "beta_source": 0.001,
32 "beta_global": 0.001,
33 "min_cut": 0.002,
34 "max_cut": 0.998,
35 "per_seuil": 0.999,
36 "dim_chan": 1,
37 "type_fusion": "mahalanobis",
38 "filt": [0.1, 0.2, 0.5, 0.2, 0.1],
39 }
41 for key in KPI_parameters_default.keys():
42 if key not in KPI_parameters.keys():
43 KPI_parameters[key] = KPI_parameters_default[key]
45 super().__init__(
46 name=name,
47 KPI_parameters=KPI_parameters,
48 cache=cache,
49 random_state=random_state,
50 **kwargs
51 )
53 self.params_ = None
55 def _aux_proc(
56 self,
57 list_UQ,
58 type_UQ,
59 list_pred,
60 list_y,
61 type_UQ_params=None,
62 ctx_mask=None,
63 mutliscale_anom_score_params_=None,
64 **kwargs
65 ):
66 """Auxialiar function that implement both fit and tranform procedure for Mutliscale_anom_score.
68 It is mainly based on the application of anom_proc.fit/compute_anom_score &
69 anom_proc.fit/compute_score_fusion.
71 Args:
72 UQ (np.array): UQmeasure provide by the UQestiamtor.
73 type_UQ (str): Type_UQ of the UQestimators.
74 pred (np.array): Prediction of the predictor or the UQestimator.
75 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit
76 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters.
77 Defaults to None.
78 """
80 n_dim = len(list_y)
81 n_chan = list_y[0].shape[2]
82 n_step = max([len(list_y[source_id]) for source_id in range(n_dim)])
84 S_anom_chan = np.zeros((n_step, n_dim * n_chan))
85 S_anom_source = np.zeros((n_step, n_dim))
86 list_bot = []
87 list_top = []
89 mode_fit = False
90 if mutliscale_anom_score_params_ is None:
91 mode_fit = True
92 mutliscale_anom_score_params_ = {
93 "anom_score_loc_params_": [],
94 "score_fusion_loc_params_": [],
95 "score_fusion_agg_params_": None,
96 }
98 for n_s in range(n_dim):
99 pred, UQ = list_pred[n_s], list_UQ[n_s]
100 y = list_y[n_s]
102 anom_score_loc_params_ = None
103 score_fusion_loc_params_ = None
104 if not mode_fit:
105 anom_score_loc_params_ = mutliscale_anom_score_params_[
106 "anom_score_loc_params_"
107 ][n_s]
109 score_fusion_loc_params_ = mutliscale_anom_score_params_[
110 "score_fusion_loc_params_"
111 ][n_s]
113 # Use fact that if params_ is none then fit is computed internall
114 (s_loc, born), anom_score_loc_params_ = anom_proc.compute_anom_score(
115 UQ=UQ,
116 type_UQ=type_UQ,
117 pred=pred,
118 y=y,
119 type_UQ_params=type_UQ_params,
120 ctx_mask=ctx_mask,
121 with_born=True,
122 params_=anom_score_loc_params_,
123 **self.KPI_parameters,
124 **kwargs
125 )
126 mutliscale_anom_score_params_["anom_score_loc_params_"].append(
127 anom_score_loc_params_
128 )
130 # Use fact that if params_ is none then fit is computed internally
131 s_agg, score_fusion_loc_params_ = anom_proc.compute_score_fusion(
132 s_loc,
133 ctx_mask=ctx_mask,
134 params_=score_fusion_loc_params_,
135 **self.KPI_parameters
136 )
138 mutliscale_anom_score_params_["score_fusion_loc_params_"].append(
139 score_fusion_loc_params_
140 )
142 mask = (np.arange(n_dim * n_chan) >= (n_s * n_chan)) & (
143 np.arange(n_dim * n_chan) < ((n_s + 1) * n_chan)
144 )
146 S_anom_chan[:, mask] = s_loc
147 S_anom_source[:, n_s] = s_agg[:, 0]
149 list_bot.append(born[0])
150 list_top.append(born[1])
152 # type_norm may be Chi² -> Exploratory works
154 score_fusion_agg_params_ = None
155 if not mode_fit:
156 score_fusion_agg_params_ = mutliscale_anom_score_params_[
157 "score_fusion_agg_params_"
158 ]
160 S_anom_agg, score_fusion_agg_params_ = anom_proc.compute_score_fusion(
161 S_anom_source,
162 ctx_mask=None,
163 beta=self.KPI_parameters["beta_global"],
164 type_fusion=self.KPI_parameters["type_fusion"],
165 type_norm="quantiles_global",
166 per_seuil=0.995,
167 d=2,
168 filt=self.KPI_parameters["filt"],
169 params_=score_fusion_agg_params_,
170 )
172 mutliscale_anom_score_params_["score_fusion_agg_params_"] = (
173 score_fusion_agg_params_
174 )
176 if mode_fit:
177 return mutliscale_anom_score_params_
179 else:
180 return S_anom_chan, S_anom_source, S_anom_agg, list_bot, list_top
182 def fit(
183 self,
184 list_UQ,
185 type_UQ,
186 list_pred,
187 list_y,
188 type_UQ_params=None,
189 ctx_mask=None,
190 **kwargs
191 ):
192 """Fitting procedure aim to estimate and store Multiscale_Anomscore_processor params for
193 Multiscale_Anomscore computation
195 Args:
196 UQ (np.array): UQmeasure provide by the UQestiamtor.
197 type_UQ (str): Type_UQ of the UQestimators.
198 pred (np.array): Prediction of the predictor or the UQestimator.
199 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit
200 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters.
201 Defaults to None.
202 """
203 self.params_ = self._aux_proc(
204 list_UQ,
205 type_UQ,
206 list_pred,
207 list_y,
208 type_UQ_params=type_UQ_params,
209 ctx_mask=ctx_mask,
210 )
211 super().fit(type_UQ=type_UQ)
213 def transform(
214 self,
215 list_UQ,
216 type_UQ,
217 list_pred,
218 list_y,
219 type_UQ_params=None,
220 ctx_mask=None,
221 **kwargs
222 ):
223 """Transform procedure aim to transform (predictor) & UQestimator output into a Multiscale_Anomscore
224 according to Multiscale_Anomscore_params
226 Args:
227 UQ (np.array): UQmeasure provide by the UQestiamtor.
228 type_UQ (str): Type_UQ of the UQestimators.
229 pred (np.array): Prediction of the predictor or the UQestimator.
230 y (np.array, optional): Targets/Observations, can be None if processor does't need y to fit
231 type_UQ_params (type_UQ_params, optional): Additional information about type_UQ parameters.
232 Defaults to None.
234 Returns:
235 A tuple (S_anom_chan, S_anom_agg, S_anom_source, list_bot, list_top)
236 where S_anom_chan is a mutli-dimensional providing Anom score for each channels of each source
237 on compute_anom_score,
238 S_anom_source is an aggregated score providing Anom score at source lvl on compute_score_fusion,
239 S_anom_agg is an aggregated score providing 1D Anom score based on compute_score_fusion,
240 list_bot is an anomalie lower threeshold for each chan of each sensors provided by compute_anom_score
241 and list_top is an anomalie upper threeshold for each chan of each sensors provided by compute_anom_score
242 """
243 super().transform(type_UQ=type_UQ)
244 S_anom_chan, S_anom_agg, S_anom_source, list_bot, list_top = self._aux_proc(
245 list_UQ,
246 type_UQ,
247 list_pred,
248 list_y,
249 type_UQ_params=type_UQ_params,
250 ctx_mask=ctx_mask,
251 mutliscale_anom_score_params_=self.params_,
252 **kwargs
253 )
254 return (S_anom_chan, S_anom_agg, S_anom_source, list_bot, list_top)