


'''simple docstring'''







import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
   import torch


A__:   Dict    =	logging.get_logger(__name__)







class A__	:






      def __init__(			self      :int					,						SCREAMING_SNAKE_CASE      :Union[str, Any]					,						SCREAMING_SNAKE_CASE      :List[Any]=2_1					,						SCREAMING_SNAKE_CASE      :Optional[Any]=[]					,						SCREAMING_SNAKE_CASE      :List[str]=None     )	-> Optional[int]:




             '''simple docstring'''
             _a   :			str					 =model
             _a   :			str					 =trace_batch_nums
             _a   :			List[Any]					 =abort_after_batch_num

             # keep a LIFO buffer of frames to dump as soon as inf/nan is encountered to give context to the problem emergence
             _a   :			List[str]					 =collections.deque([]					,						SCREAMING_SNAKE_CASE     )
             _a   :			Union[str, Any]					 =[]
             _a   :			List[str]					 =0
             _a   :			List[str]					 =0
             _a   :			Any					 =False
             _a   :			Dict					 ="""                 """

             self.analyse_model()

             self.register_forward_hook()






      def __UpperCAmelCase	(			self      :Optional[Any]					,						SCREAMING_SNAKE_CASE      :Optional[int]=None     )	-> Any:




             '''simple docstring'''
             if frame is not None:
                    self.expand_frame(SCREAMING_SNAKE_CASE     )
             self.frames.append("""\n""".join(self.frame     )     )
             _a   :			str					 =[]  # start a new frame






      def __UpperCAmelCase	(			self      :int					,						SCREAMING_SNAKE_CASE      :str     )	-> str:




             '''simple docstring'''
             self.frame.append(SCREAMING_SNAKE_CASE     )






      def __UpperCAmelCase	(			self      :Dict     )	-> Dict:




             '''simple docstring'''
             print("""\n""".join(self.frames     )     )
             _a   :			Any					 =[]






      def __UpperCAmelCase	(			self      :List[Any]     )	-> List[Any]:




             '''simple docstring'''
             _a   :			Optional[int]					 =[]






      def __UpperCAmelCase	(			self      :str     )	-> Union[str, Any]:




             '''simple docstring'''
             print(f"\nDetected inf/nan during batch_number={self.batch_number}"     )
             print(f"Last {len(self.frames     )} forward frames:"     )
             print(f"{'abs min':8} {'abs max':8} metadata"     )
             print("""\n""".join(self.frames     )     )
             print("""\n\n"""     )
             _a   :			Tuple					 =[]






      def __UpperCAmelCase	(			self      :Optional[int]     )	-> Dict:




             '''simple docstring'''
             # extract the fully qualified module names, to be able to report at run time. e.g.:
             # encoder.block.2.layer.0.SelfAttention.o
             #
             # for shared weights only the first shared module name will be registered
             _a   :			Optional[Any]					 ={m: name for name, m in self.model.named_modules()}
             # self.longest_module_name = max(len(v) for v in self.module_names.values())






      def __UpperCAmelCase	(			self      :Any					,						SCREAMING_SNAKE_CASE      :str					,						SCREAMING_SNAKE_CASE      :Tuple     )	-> Dict:




             '''simple docstring'''
             if torch.is_tensor(SCREAMING_SNAKE_CASE     ):
                    self.expand_frame(get_abs_min_max(SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE     )     )
                    if detect_overflow(SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE     ):
                           _a   :			Union[str, Any]					 =True
             elif var is None:
                    self.expand_frame(f"{'None':>17} {ctx}"     )
             else:
                    self.expand_frame(f"{'not a tensor':>17} {ctx}"     )






      def __UpperCAmelCase	(			self      :List[Any]     )	-> Any:




             '''simple docstring'''
             self.expand_frame(f"\n\n{self.prefix} *** Starting batch number={self.batch_number} ***"     )
             self.expand_frame(f"{'abs min':8} {'abs max':8} metadata"     )






      def __UpperCAmelCase	(			self      :List[Any]     )	-> Optional[int]:




             '''simple docstring'''
             self.expand_frame(f"{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n"     )






      def __UpperCAmelCase	(			self      :Tuple					,						SCREAMING_SNAKE_CASE      :List[str]					,						SCREAMING_SNAKE_CASE      :Optional[int]					,						SCREAMING_SNAKE_CASE      :List[Any]     )	-> Dict:




             '''simple docstring'''
             self.expand_frame(f"{self.prefix} {self.module_names[module]} {module.__class__.__name__}"     )

             # params
             for name, p in module.named_parameters(recurse=SCREAMING_SNAKE_CASE     ):
                    self.analyse_variable(SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE     )

             # inputs
             if isinstance(SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE     ):
                    for i, x in enumerate(SCREAMING_SNAKE_CASE     ):
                           self.analyse_variable(SCREAMING_SNAKE_CASE					,						f"input[{i}]"     )
             else:
                    self.analyse_variable(SCREAMING_SNAKE_CASE					,						"""input"""     )

             # outputs
             if isinstance(SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE     ):
                    for i, x in enumerate(SCREAMING_SNAKE_CASE     ):
                           # possibly a tuple of tuples
                           if isinstance(SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE     ):
                                  for j, y in enumerate(SCREAMING_SNAKE_CASE     ):
                                         self.analyse_variable(SCREAMING_SNAKE_CASE					,						f"output[{i}][{j}]"     )
                           else:
                                  self.analyse_variable(SCREAMING_SNAKE_CASE					,						f"output[{i}]"     )
             else:
                    self.analyse_variable(SCREAMING_SNAKE_CASE					,						"""output"""     )

             self.save_frame()






      def __UpperCAmelCase	(			self      :List[Any]     )	-> Optional[Any]:




             '''simple docstring'''
             self.model.apply(self._register_forward_hook     )






      def __UpperCAmelCase	(			self      :Optional[int]					,						SCREAMING_SNAKE_CASE      :List[Any]     )	-> int:




             '''simple docstring'''
             module.register_forward_hook(self.forward_hook     )






      def __UpperCAmelCase	(			self      :int					,						SCREAMING_SNAKE_CASE      :int					,						SCREAMING_SNAKE_CASE      :Optional[int]					,						SCREAMING_SNAKE_CASE      :int     )	-> Tuple:




             '''simple docstring'''
             # - input is a tuple of packed inputs (could be non-Tensors)
             # - output could be a Tensor or a tuple of Tensors and non-Tensors

             _a   :			Optional[Any]					 =False

             _a   :			str					 =True if self.batch_number in self.trace_batch_nums else False
             if trace_mode:
                    self.reset_saved_frames()

             if self.total_calls == 0:
                    self.batch_start_frame()
             self.total_calls += 1

             # count batch numbers - the very first forward hook of the batch will be called when the
             # batch completes - i.e. it gets called very last - we know this batch has finished
             if module == self.model:
                    self.batch_number += 1
                    _a   :			Dict					 =True

             self.create_frame(SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE     )

             # if last_frame_of_batch:
             #     self.batch_end_frame()

             if trace_mode:
                    self.trace_frames()

             if last_frame_of_batch:
                    self.batch_start_frame()

             if self.detected_overflow and not trace_mode:
                    self.dump_saved_frames()

                    # now we can abort, as it's pointless to continue running
                    raise ValueError(
                        """DebugUnderflowOverflow: inf/nan detected, aborting as there is no point running further. """
                        """Please scroll up above this traceback to see the activation values prior to this event."""     )

             # abort after certain batch if requested to do so
             if self.abort_after_batch_num is not None and self.batch_number > self.abort_after_batch_num:
                    raise ValueError(
                        f"DebugUnderflowOverflow: aborting after {self.batch_number} batches due to"
                        f" `abort_after_batch_num={self.abort_after_batch_num}` arg"     )


def      SCREAMING_SNAKE_CASE_ (  _UpperCAmelCase     :     int   ,_UpperCAmelCase     :     Any  )  ->      Any:
       _a   :			str					 =var.abs()
       return F"{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"


def      SCREAMING_SNAKE_CASE_ (  _UpperCAmelCase     :     Optional[int]   ,_UpperCAmelCase     :     Tuple  )  ->      Union[str, Any]:
       _a   :			Tuple					 =False
       if torch.isnan(_UpperCAmelCase  ).any().item():
              _a   :			Any					 =True
              print(F"{ctx} has nans"  )
       if torch.isinf(_UpperCAmelCase  ).any().item():
              _a   :			List[Any]					 =True
              print(F"{ctx} has infs"  )

       # if needed to monitor large elements can enable the following
       if 0:  # and detected:
              _a   :			List[str]					 =var[torch.ge(var.abs()   ,100  )]
              if naaa.numel() > 0:
                     print(F"{ctx}:  n100={naaa.numel()}"  )
              _a   :			Dict					 =var[torch.ge(var.abs()   ,1000  )]
              if naaaa.numel() > 0:
                     print(F"{ctx}: n1000={naaaa.numel()}"  )
              _a   :			Any					 =var[torch.ge(var.abs()   ,10000  )]
              if naaaaa.numel() > 0:
                     print(F"{ctx}: n10000={naaaaa.numel()}"  )

       if 0:
              print(F"min={var.min():9.2e} max={var.max():9.2e}"  )

       if 0:
              print(F"min={var.min():9.2e} max={var.max():9.2e} var={var.var():9.2e} mean={var.mean():9.2e} ({ctx})"  )

       return detected







class A__	(						UpperCAmelCase__				):
      __UpperCamelCase	:  Dict     	=							"underflow_overflow"
      __UpperCamelCase	:  Tuple     	=							"tpu_metrics_debug"






'''simple docstring'''







from sklearn.metrics import mean_absolute_error

import datasets


A__:   Dict    =	'''\
@article{scikit-learn,
 title={Scikit-learn: Machine Learning in {P}ython},
 author={Pedregosa, F. and Varoquaux, G. and Gramfort, A. and Michel, V.
         and Thirion, B. and Grisel, O. and Blondel, M. and Prettenhofer, P.
         and Weiss, R. and Dubourg, V. and Vanderplas, J. and Passos, A. and
         Cournapeau, D. and Brucher, M. and Perrot, M. and Duchesnay, E.},
 journal={Journal of Machine Learning Research},
 volume={12},
 pages={2825--2830},
 year={2011}
}
'''

A__:   Union[str, Any]    =	'''\
Mean Absolute Error (MAE) is the mean of the magnitude of difference between the predicted and actual
values.
'''


A__:   str    =	'''
Args:
    predictions: array-like of shape (n_samples,) or (n_samples, n_outputs)
        Estimated target values.
    references: array-like of shape (n_samples,) or (n_samples, n_outputs)
        Ground truth (correct) target values.
    sample_weight: array-like of shape (n_samples,), default=None
        Sample weights.
    multioutput: {"raw_values", "uniform_average"} or array-like of shape (n_outputs,), default="uniform_average"
        Defines aggregating of multiple output values. Array-like value defines weights used to average errors.

                 "raw_values" : Returns a full set of errors in case of multioutput input.

                 "uniform_average" : Errors of all outputs are averaged with uniform weight.

Returns:
    mae : mean absolute error.
        If multioutput is "raw_values", then mean absolute error is returned for each output separately. If multioutput is "uniform_average" or an ndarray of weights, then the weighted average of all output errors is returned.
        MAE output is non-negative floating point. The best value is 0.0.
Examples:

    >>> mae_metric = datasets.load_metric("mae")
    >>> predictions = [2.5, 0.0, 2, 8]
    >>> references = [3, -0.5, 2, 7]
    >>> results = mae_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {\'mae\': 0.5}

    If you\'re using multi-dimensional lists, then set the config as follows :

    >>> mae_metric = datasets.load_metric("mae", "multilist")
    >>> predictions = [[0.5, 1], [-1, 1], [7, -6]]
    >>> references = [[0, 2], [-1, 2], [8, -5]]
    >>> results = mae_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {\'mae\': 0.75}
    >>> results = mae_metric.compute(predictions=predictions, references=references, multioutput=\'raw_values\')
    >>> print(results)
    {\'mae\': array([0.5, 1. ])}
'''







@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION			,					_KWARGS_DESCRIPTION				)
class A__	(						datasets.Metric				):






      def __UpperCAmelCase	(			self      :List[Any]     )	-> Dict:




             '''simple docstring'''
             return datasets.MetricInfo(
                 description=_DESCRIPTION					,						citation=_CITATION					,						inputs_description=_KWARGS_DESCRIPTION					,						features=datasets.Features(self._get_feature_types()     )					,						reference_urls=[
                     """https://scikit-learn.org/stable/modules/generated/sklearn.metrics.mean_absolute_error.html"""
                 ]					,						)






      def __UpperCAmelCase	(			self      :Dict     )	-> List[str]:




             '''simple docstring'''
             if self.config_name == "multilist":
                    return {
                        "predictions": datasets.Sequence(datasets.Value("""float"""     )     ),
                        "references": datasets.Sequence(datasets.Value("""float"""     )     ),
                    }
             else:
                    return {
                        "predictions": datasets.Value("""float"""     ),
                        "references": datasets.Value("""float"""     ),
                    }






      def __UpperCAmelCase	(			self      :Any					,						SCREAMING_SNAKE_CASE      :Union[str, Any]					,						SCREAMING_SNAKE_CASE      :str					,						SCREAMING_SNAKE_CASE      :Union[str, Any]=None					,						SCREAMING_SNAKE_CASE      :Any="uniform_average"     )	-> List[Any]:




             '''simple docstring'''
             _a   :			Tuple					 =mean_absolute_error(SCREAMING_SNAKE_CASE					,						SCREAMING_SNAKE_CASE					,						sample_weight=SCREAMING_SNAKE_CASE					,						multioutput=SCREAMING_SNAKE_CASE     )

             return {"mae": mae_score}


