o }o™iÈã@s6ddlZddlZddlmZdgZGdd„deƒZdS)éN)ÚMetricÚMultiBinaryAccuracyc sTeZdZdZdZd ‡fdd„ Z d dejdejdejdejfd d „Zd d „Z ‡Z S)ra This metric computes accuracies that are needed to evaluate multiple binary outputs. For example, if a model returns a set of multiple sigmoid outputs per each sample or at each time step, F1 score can be calculated to monitor Type 1 error and Type 2 error together. Example: def validation_step(self, batch, batch_idx): ... signals, signal_lengths, targets = batch preds, _ = self.forward(input_signal=signals, signal_lengths=signal_lengths, targets=targets) loss = self.loss(logits=preds, labels=targets) self._accuracy_valid(preds, targets, signal_lengths) f1_acc = self._accuracy.compute() self.val_outputs = {'val_loss': loss, 'val_f1_acc': f1_acc} return self.val_outputs def on_validation_epoch_end(self): ... val_loss_mean = torch.stack([x['val_loss'] for x in self.val_outputs]).mean() correct_counts = torch.stack([x['val_correct_counts'] for x in self.val_outputs]).sum(axis=0) total_counts = torch.stack([x['val_total_counts'] for x in self.val_outputs]).sum(axis=0) self._accuracy_valid.correct_counts_k = correct_counts self._accuracy_valid.total_counts_k = total_counts f1_acc = self._accuracy_valid.compute() self._accuracy_valid.reset() self.log('val_loss', val_loss_mean) self.log('val_f1_acc', f1_acc) self.val_outputs.clear() # free memory return {'val_loss': val_loss_mean, 'val_f1_acc': f1_acc} Args: preds (torch.Tensor): Predicted values which should be in range of [0, 1]. targets (torch.Tensor): Target values which should be in range of [0, 1]. signal_lengths (torch.Tensor): Length of each sequence in the batch input. signal_lengths values are used to filter out zero-padded parts in each sequence. Returns: f1_score (torch.Tensor): F1 score calculated from the predicted value and binarized target values. Fcs¨tƒj|d|jdt d¡ddd|jdt d¡ddd|jdt d¡ddd|jd t d¡ddd|jd t d¡ddd|jd t d¡dddd |_dS) N)Údist_sync_on_stepÚtotal_correct_countsrÚsumF)ÚdefaultÚdist_reduce_fxÚ persistentÚtotal_sample_countsÚtrue_positive_countÚfalse_positive_countÚfalse_negative_countÚpositive_countgíµ ÷Æ°>)ÚsuperÚ__init__Ú add_stateÚtorchÚtensorÚeps)Úselfr©Ú __class__©úa/home/ubuntu/.local/lib/python3.10/site-packages/nemo/collections/asr/metrics/multi_binary_acc.pyrJs zMultiBinaryAccuracy.__init__ÚpredsÚtargetsÚsignal_lengthsÚreturnc spt ¡(‡‡fdd„tˆjdƒDƒ}‡‡fdd„tˆjdƒDƒ}tj|dd|_tj|dd|_|j ¡ ¡|j ¡ ¡k|_ |j ¡ ¡|j ¡ ¡k|_ |j ¡ ¡dk|_ |j ¡ ¡dk|_ |rÏ|j t |j ¡ ¡dk¡7_ |jt t |j |j ¡¡7_|jt t |j |j ¡¡7_|jt t |j |j ¡¡7_|jt |j ¡ ¡|j ¡ ¡k¡7_|jt t |jj¡¡7_nVt |j ¡ ¡dk¡|_ t t |j |j ¡¡|_t t |j |j ¡¡|_t t |j |j ¡¡|_t |j ¡ ¡|j ¡ ¡k¡|_t t |jj¡¡|_WdƒdSWdƒdS1s1wYdS) a‹ Update the metric with the given predictions, targets, and signal lengths to the metric instance. Args: preds (torch.Tensor): Predicted values. targets (torch.Tensor): Target values. signal_lengths (torch.Tensor): Length of each sequence in the batch input. cumulative (bool): Whether to accumulate the values over time. cs&g|]}ˆ|dˆ|…dd…f‘qS©Nr©Ú.0Úk)rrrrÚ aó&z.MultiBinaryAccuracy.update..rcs&g|]}ˆ|dˆ|…dd…f‘qSrrr)rrrrr"br#)ÚdiméTN)rÚno_gradÚrangeÚshapeÚcatrrÚroundÚboolÚtrueÚfalseÚpositiveÚnegativerrr Ú logical_andr r rr Úprodr)rrrrÚ cumulativeÚ preds_listÚ targets_listr)rrrrÚupdateTs2 "   , $è$ïzMultiBinaryAccuracy.updatecCs€|j|j|j|j}|j|j|j|j}d|||||j ¡ ¡}t |¡r5t  d¡d}|  ¡|  ¡|  ¡fS)as Compute F1 score from the accumulated values. Return -1 if the F1 score is NaN. Returns: f1_score (torch.Tensor): F1 score calculated from the accumulated values. precision (torch.Tensor): Precision calculated from the accumulated values. recall (torch.Tensor): Recall calculated from the accumulated values. ézDself.f1_score contains NaN value. Returning -1 instead of NaN value.éÿÿÿÿ) r r rr ÚdetachÚclonerÚisnanÚloggingÚwarnÚfloat)rÚ precisionÚrecallÚf1_scorerrrÚcomputezs "  zMultiBinaryAccuracy.compute)F) Ú__name__Ú __module__Ú __qualname__Ú__doc__Úfull_state_updaterrÚTensorr5rAÚ __classcell__rrrrrs0 ÿÿÿÿ þ&)r;rÚ torchmetricsrÚ__all__rrrrrÚs