

import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
  import torch


__A    	=			logging.get_logger(__name__)



class 						lowercase_					:



      def __init__(    self   :       List[Any]    ,      A__   :       List[Any]    ,      A__   :       Tuple=21    ,      A__   :       str=[]    ,      A__   :       str=None	)					->  Dict:
        _snake_case		=							model
        _snake_case		=							trace_batch_nums
        _snake_case		=							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
        _snake_case		=							collections.deque([]    ,      __UpperCamelCase	)
        _snake_case		=							[]
        _snake_case		=							0
        _snake_case		=							0
        _snake_case		=							False
        _snake_case		=							'''                 '''

        self.analyse_model()

        self.register_forward_hook()



      def 		UpperCamelCase_	(    self   :       Union[str, Any]    ,      A__   :       int=None	)					->  Tuple:
        if frame is not None:
          self.expand_frame(__UpperCamelCase	)
        self.frames.append('''\n'''.join(self.frame	)	)
        _snake_case		=							[]  # start a new frame



      def 		UpperCamelCase_	(    self   :       str    ,      A__   :       Any	)					->  str:
        self.frame.append(__UpperCamelCase	)



      def 		UpperCamelCase_	(    self   :       List[Any]	)					->  Optional[Any]:
        print('''\n'''.join(self.frames	)	)
        _snake_case		=							[]



      def 		UpperCamelCase_	(    self   :       Dict	)					->  List[str]:
        _snake_case		=							[]



      def 		UpperCamelCase_	(    self   :       Optional[int]	)					->  Tuple:
        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'''	)
        _snake_case		=							[]



      def 		UpperCamelCase_	(    self   :       List[str]	)					->  Optional[Any]:
        # 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
        _snake_case		=							{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    ,      A__   :       List[Any]    ,      A__   :       Optional[int]	)					->  List[str]:
        if torch.is_tensor(__UpperCamelCase	):
          self.expand_frame(get_abs_min_max(__UpperCamelCase    ,      __UpperCamelCase	)	)
          if detect_overflow(__UpperCamelCase    ,      __UpperCamelCase	):
            _snake_case		=							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   :       Dict	)					->  Any:
        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   :       Dict	)					->  Union[str, Any]:
        self.expand_frame(f"""{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n"""	)



      def 		UpperCamelCase_	(    self   :       List[Any]    ,      A__   :       str    ,      A__   :       Optional[int]    ,      A__   :       Dict	)					->  List[str]:
        self.expand_frame(f"""{self.prefix} {self.module_names[module]} {module.__class__.__name__}"""	)

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

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

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

        self.save_frame()



      def 		UpperCamelCase_	(    self   :       Optional[Any]	)					->  Union[str, Any]:
        self.model.apply(self._register_forward_hook	)



      def 		UpperCamelCase_	(    self   :       Union[str, Any]    ,      A__   :       List[Any]	)					->  Optional[int]:
        module.register_forward_hook(self.forward_hook	)







      def 		UpperCamelCase_	(    self   :       List[Any]    ,      A__   :       int    ,      A__   :       int    ,      A__   :       List[str]	)					->  Any:
        # - input is a tuple of packed inputs (could be non-Tensors)
        # - output could be a Tensor or a tuple of Tensors and non-Tensors

        _snake_case		=							False

        _snake_case		=							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
          _snake_case		=							True

        self.create_frame(__UpperCamelCase    ,      __UpperCamelCase    ,      __UpperCamelCase	)

        # 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    snake_case_(_UpperCamelCase				, _UpperCamelCase  )							->			str:



  """simple docstring"""






  _snake_case		=							var.abs()
  return F"""{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"""



def    snake_case_(_UpperCamelCase				, _UpperCamelCase  )							->			int:



  """simple docstring"""






  _snake_case		=							False
  if torch.isnan(lowercase_  ).any().item():
    _snake_case		=							True
    print(F"""{ctx} has nans"""  )
  if torch.isinf(lowercase_  ).any().item():
    _snake_case		=							True
    print(F"""{ctx} has infs"""  )

  # if needed to monitor large elements can enable the following
  if 0:  # and detected:
    _snake_case		=							var[torch.ge(var.abs()				, 100  )]
    if naaa.numel() > 0:
      print(F"""{ctx}:  n100={naaa.numel()}"""  )
    _snake_case		=							var[torch.ge(var.abs()				, 1_000  )]
    if naaaa.numel() > 0:
      print(F"""{ctx}: n1000={naaaa.numel()}"""  )
    _snake_case		=							var[torch.ge(var.abs()				, 10_000  )]
    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 						lowercase_					(						UpperCAmelCase_  ):
      UpperCamelCase_    :	Any		   =       "underflow_overflow"
      UpperCamelCase_    :	List[str]		   =       "tpu_metrics_debug"




from collections import namedtuple

__A    	=			namedtuple('''from_to''', '''from_ to''')

__A    	=			{
    '''atm''': from_to(1, 1),
    '''pascal''': from_to(0.000_0098, 10_13_25),
    '''bar''': from_to(0.98_6923, 1.0_1325),
    '''kilopascal''': from_to(0.0098_6923, 101.325),
    '''megapascal''': from_to(9.8_6923, 0.10_1325),
    '''psi''': from_to(0.06_8046, 14.6959),
    '''inHg''': from_to(0.033_4211, 29.9213),
    '''torr''': from_to(0.0013_1579, 7_60),
}



def    snake_case_(_UpperCamelCase				, _UpperCamelCase				, _UpperCamelCase  )							->			float:



  """simple docstring"""






  if from_type not in PRESSURE_CONVERSION:
    raise ValueError(
        F"""Invalid 'from_type' value: {from_type!r}  Supported values are:\n"""
        + ''', '''.join(_UpperCamelCase  )  )
  if to_type not in PRESSURE_CONVERSION:
    raise ValueError(
        F"""Invalid 'to_type' value: {to_type!r}.  Supported values are:\n"""
        + ''', '''.join(_UpperCamelCase  )  )
  return (
      value * PRESSURE_CONVERSION[from_type].from_ * PRESSURE_CONVERSION[to_type].to
  )


if __name__ == "__main__":
 import doctest

 doctest.testmod()

