
'''simple docstring'''





import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
      import torch


SCREAMING_SNAKE_CASE    =					logging.get_logger(__name__)






class        UpperCAmelCase_       :

 '''simple docstring'''







 def __init__(				self		:     List[str] ,							snake_case__		:     Optional[Any] ,							snake_case__		:     Tuple=21 ,							snake_case__		:     Optional[Any]=[] ,							snake_case__		:     Optional[int]=None   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Dict												=       model
   UpperCAmelCase__			:				Optional[Any]												=       trace_batch_nums
   UpperCAmelCase__			:				Tuple												=       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
   UpperCAmelCase__			:				Optional[int]												=       collections.deque([] ,							snake_case__   )
   UpperCAmelCase__			:				Optional[Any]												=       []
   UpperCAmelCase__			:				Optional[int]												=       0
   UpperCAmelCase__			:				str												=       0
   UpperCAmelCase__			:				Optional[Any]												=       False
   UpperCAmelCase__			:				Optional[int]												=       "                 "

   self.analyse_model()

   self.register_forward_hook()







 def        UpperCamelCase     (				self		:     Optional[int] ,							snake_case__		:     Union[str, Any]=None   ):



   '''simple docstring'''




   if frame is not None:
     self.expand_frame(snake_case__   )
   self.frames.append("\n".join(self.frame   )   )
   UpperCAmelCase__			:				Any												=       []  # start a new frame







 def        UpperCamelCase     (				self		:     Dict ,							snake_case__		:     str   ):



   '''simple docstring'''




   self.frame.append(snake_case__   )







 def        UpperCamelCase     (				self		:     List[str]   ):



   '''simple docstring'''




   print("\n".join(self.frames   )   )
   UpperCAmelCase__			:				Optional[Any]												=       []







 def        UpperCamelCase     (				self		:     Dict   ):



   '''simple docstring'''




   UpperCAmelCase__			:				List[Any]												=       []







 def        UpperCamelCase     (				self		:     str   ):



   '''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"   )
   UpperCAmelCase__			:				str												=       []







 def        UpperCamelCase     (				self		:     Any   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Union[str, 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		:     int ,							snake_case__		:     Optional[Any] ,							snake_case__		:     List[str]   ):



   '''simple docstring'''




   if torch.is_tensor(snake_case__   ):
     self.expand_frame(get_abs_min_max(snake_case__ ,							snake_case__   )   )
     if detect_overflow(snake_case__ ,							snake_case__   ):
       UpperCAmelCase__			:				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		:     int   ):



   '''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]   ):



   '''simple docstring'''




   self.expand_frame(F"""{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n"""   )







 def        UpperCamelCase     (				self		:     List[Any] ,							snake_case__		:     Optional[Any] ,							snake_case__		:     List[Any] ,							snake_case__		:     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=snake_case__   ):
     self.analyse_variable(snake_case__ ,							snake_case__   )

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

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

   self.save_frame()







 def        UpperCamelCase     (				self		:     int   ):



   '''simple docstring'''




   self.model.apply(self._register_forward_hook   )







 def        UpperCamelCase     (				self		:     Optional[Any] ,							snake_case__		:     Any   ):



   '''simple docstring'''




   module.register_forward_hook(self.forward_hook   )







 def        UpperCamelCase     (				self		:     Optional[int] ,							snake_case__		:     Dict ,							snake_case__		:     int ,							snake_case__		:     int   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Dict												=       False

   UpperCAmelCase__			:				Optional[Any]												=       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
     UpperCAmelCase__			:				List[Any]												=       True

   self.create_frame(snake_case__ ,							snake_case__ ,							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 snake_case_    (				lowercase__      ,					lowercase__		):
  UpperCAmelCase__			:				int												=       var.abs()
  return F"""{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"""







def snake_case_    (				lowercase__      ,					lowercase__		):
  UpperCAmelCase__			:				Optional[int]												=       False
  if torch.isnan(lowercase__		).any().item():
    UpperCAmelCase__			:				Dict												=       True
    print(F"""{ctx} has nans"""		)
  if torch.isinf(lowercase__		).any().item():
    UpperCAmelCase__			:				int												=       True
    print(F"""{ctx} has infs"""		)

  # if needed to monitor large elements can enable the following
  if 0:  # and detected:
    UpperCAmelCase__			:				str												=       var[torch.ge(var.abs()      ,					1_0_0		)]
    if naaa.numel() > 0:
      print(F"""{ctx}:  n100={naaa.numel()}"""		)
    UpperCAmelCase__			:				Union[str, Any]												=       var[torch.ge(var.abs()      ,					1_0_0_0		)]
    if naaaa.numel() > 0:
      print(F"""{ctx}: n1000={naaaa.numel()}"""		)
    UpperCAmelCase__			:				Optional[Any]												=       var[torch.ge(var.abs()      ,					1_0_0_0_0		)]
    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        UpperCAmelCase_       (						A							):

 '''simple docstring'''
 lowercase_				:						Any				=       "underflow_overflow"
 lowercase_				:						int				=       "tpu_metrics_debug"





'''simple docstring'''





def snake_case_    (				lowercase__      ,					lowercase__ = 1_3		):
  UpperCAmelCase__			:				Optional[Any]												=       ""
  for c in s:
    if "A" <= c <= "Z":
      out += chr(ord("A"		) + (ord(lowercase__		) - ord("A"		) + n) % 2_6		)
    elif "a" <= c <= "z":
      out += chr(ord("a"		) + (ord(lowercase__		) - ord("a"		) + n) % 2_6		)
    else:
      out += c
  return out







def snake_case_    (				):
  UpperCAmelCase__			:				Dict												=       input("Enter message: "		)

  UpperCAmelCase__			:				Optional[int]												=       dencrypt(lowercase__      ,					1_3		)
  print("Encryption:"      ,					lowercase__		)

  UpperCAmelCase__			:				Union[str, Any]												=       dencrypt(lowercase__      ,					1_3		)
  print("Decryption: "      ,					lowercase__		)


if __name__ == "__main__":
      import doctest

      doctest.testmod()
      main()



