code: 
  
  
  
  
  
  '''simple docstring'''
  
  
  import collections
  
  from .utils import ExplicitEnum, is_torch_available, logging
  
  
  if is_torch_available():
      import torch
  
  
  _lowerCamelCase				:	Any				    =       logging.get_logger(__name__)
  
  
  
  class 							__UpperCAmelCase  :
  
  
  
  
  
  
  
    '''simple docstring'''
  
  
  
  
  
  
  
    def __init__(self						:      Optional[int]      ,      _lowerCAmelCase						:      str      ,      _lowerCAmelCase						:      Optional[Any]=21      ,      _lowerCAmelCase						:      Tuple=[]      ,      _lowerCAmelCase						:      Optional[Any]=None      ):
     A							  =      model
     A							  =      trace_batch_nums
     A							  =      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							  =      collections.deque([]      ,      _lowerCAmelCase      )
     A							  =      []
     A							  =      0
     A							  =      0
     A							  =      False
     A							  =      """                 """
  
     self.analyse_model()
  
     self.register_forward_hook()
  
  
  
  
  
  
  
    def        A     (self						:      int      ,      _lowerCAmelCase						:      Dict=None      ):
     if frame is not None:
      self.expand_frame(_lowerCAmelCase      )
     self.frames.append("""\n""".join(self.frame      )      )
     A							  =      []  # start a new frame
  
  
  
  
  
  
  
    def        A     (self						:      Optional[int]      ,      _lowerCAmelCase						:      str      ):
     self.frame.append(_lowerCAmelCase      )
  
  
  
  
  
  
  
    def        A     (self						:      Any      ):
     print("""\n""".join(self.frames      )      )
     A							  =      []
  
  
  
  
  
  
  
    def        A     (self						:      int      ):
     A							  =      []
  
  
  
  
  
  
  
    def        A     (self						:      List[Any]      ):
     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							  =      []
  
  
  
  
  
  
  
    def        A     (self						:      Dict      ):
     # 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							  =      {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        A     (self						:      Any      ,      _lowerCAmelCase						:      Optional[int]      ,      _lowerCAmelCase						:      List[str]      ):
     if torch.is_tensor(_lowerCAmelCase      ):
      self.expand_frame(get_abs_min_max(_lowerCAmelCase      ,      _lowerCAmelCase      )      )
      if detect_overflow(_lowerCAmelCase      ,      _lowerCAmelCase      ):
       A							  =      True
     elif var is None:
      self.expand_frame(F"""{"None":>17} {ctx}"""      )
     else:
      self.expand_frame(F"""{"not a tensor":>17} {ctx}"""      )
  
  
  
  
  
  
  
    def        A     (self						:      Tuple      ):
     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        A     (self						:      Dict      ):
     self.expand_frame(F"""{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n"""      )
  
  
  
  
  
  
  
    def        A     (self						:      Optional[int]      ,      _lowerCAmelCase						:      int      ,      _lowerCAmelCase						:      Dict      ,      _lowerCAmelCase						:      Any      ):
     self.expand_frame(F"""{self.prefix} {self.module_names[module]} {module.__class__.__name__}"""      )
  
     # params
     for name, p in module.named_parameters(recurse=_lowerCAmelCase      ):
      self.analyse_variable(_lowerCAmelCase      ,      _lowerCAmelCase      )
  
     # inputs
     if isinstance(_lowerCAmelCase      ,      _lowerCAmelCase      ):
      for i, x in enumerate(_lowerCAmelCase      ):
       self.analyse_variable(_lowerCAmelCase      ,      F"""input[{i}]"""      )
     else:
      self.analyse_variable(_lowerCAmelCase      ,      """input"""      )
  
     # outputs
     if isinstance(_lowerCAmelCase      ,      _lowerCAmelCase      ):
      for i, x in enumerate(_lowerCAmelCase      ):
       # possibly a tuple of tuples
       if isinstance(_lowerCAmelCase      ,      _lowerCAmelCase      ):
        for j, y in enumerate(_lowerCAmelCase      ):
         self.analyse_variable(_lowerCAmelCase      ,      F"""output[{i}][{j}]"""      )
       else:
        self.analyse_variable(_lowerCAmelCase      ,      F"""output[{i}]"""      )
     else:
      self.analyse_variable(_lowerCAmelCase      ,      """output"""      )
  
     self.save_frame()
  
  
  
  
  
  
  
    def        A     (self						:      List[Any]      ):
     self.model.apply(self._register_forward_hook      )
  
  
  
  
  
  
  
    def        A     (self						:      Optional[int]      ,      _lowerCAmelCase						:      Union[str, Any]      ):
     module.register_forward_hook(self.forward_hook      )
  
  
  
  
  
  
  
    def        A     (self						:      int      ,      _lowerCAmelCase						:      str      ,      _lowerCAmelCase						:      str      ,      _lowerCAmelCase						:      str      ):
     # - 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							  =      False
  
     A							  =      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							  =      True
  
     self.create_frame(_lowerCAmelCase      ,      _lowerCAmelCase      ,      _lowerCAmelCase      )
  
     # 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    __a (     UpperCAmelCase       ,			UpperCAmelCase    )      ->Optional[Any]:
  
  
  
   """simple docstring"""
  
  
  
  
  
  
  
   A							  =      var.abs()
   return f"""{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"""
  
  
  
  
  def    __a (     UpperCAmelCase       ,			UpperCAmelCase    )      ->int:
  
  
  
   """simple docstring"""
  
  
  
  
  
  
  
   A							  =      False
   if torch.isnan(__lowerCamelCase    ).any().item():
    A							  =      True
    print(f"""{ctx} has nans"""    )
   if torch.isinf(__lowerCamelCase    ).any().item():
    A							  =      True
    print(f"""{ctx} has infs"""    )
  
   # if needed to monitor large elements can enable the following
   if 0:  # and detected:
    A							  =      var[torch.ge(var.abs()       ,			100    )]
    if naaa.numel() > 0:
     print(f"""{ctx}:  n100={naaa.numel()}"""    )
    A							  =      var[torch.ge(var.abs()       ,			1000    )]
    if naaaa.numel() > 0:
     print(f"""{ctx}: n1000={naaaa.numel()}"""    )
    A							  =      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 							__UpperCAmelCase  (       __lowerCamelCase						):
  
  
  
  
  
  
  
    '''simple docstring'''
  
  
  
  
  
  
  
    __lowerCAmelCase 				=    '''underflow_overflow'''
    __lowerCAmelCase 				=    '''tpu_metrics_debug'''
  
  
style_context: 
  
  
  
  
  
  '''simple docstring'''
  
  
  from typing import Dict, List, Optional, Tuple
  
  from ...tokenization_utils import AddedToken, PreTrainedTokenizer
  from ...utils import logging
  
  
  _lowerCamelCase				:	int				    =       logging.get_logger(__name__)
  
  
  
  class 							__UpperCAmelCase  (       A__						):
  
  
  
  
  
  
  
  		'''simple docstring'''
  
  
  
  
  
  
  
  		__lowerCAmelCase 				=    ['''input_ids''', '''attention_mask''']
  
  
  
  
  
  
  
  		def __init__(self						:      Tuple      ,      _lowerCAmelCase						:      str="[PAD]"      ,      _lowerCAmelCase						:      Dict="[BOS]"      ,      _lowerCAmelCase						:      Any="[EOS]"      ,      _lowerCAmelCase						:      List[Any]="[MASK]"      ,      _lowerCAmelCase						:      int="[CLS]"      ,      _lowerCAmelCase						:      Optional[Any]="[SEP]"      ,      _lowerCAmelCase						:      List[str]=2048      ,      **_lowerCAmelCase						:      Dict      ,      ):
  			A							  =      AddedToken(_lowerCAmelCase      ,      lstrip=_lowerCAmelCase      ,      rstrip=_lowerCAmelCase      ) if isinstance(_lowerCAmelCase      ,      _lowerCAmelCase      ) else pad_token
  			A							  =      AddedToken(_lowerCAmelCase      ,      lstrip=_lowerCAmelCase      ,      rstrip=_lowerCAmelCase      ) if isinstance(_lowerCAmelCase      ,      _lowerCAmelCase      ) else bos_token
  			A							  =      AddedToken(_lowerCAmelCase      ,      lstrip=_lowerCAmelCase      ,      rstrip=_lowerCAmelCase      ) if isinstance(_lowerCAmelCase      ,      _lowerCAmelCase      ) else eos_token
  			A							  =      AddedToken(_lowerCAmelCase      ,      lstrip=_lowerCAmelCase      ,      rstrip=_lowerCAmelCase      ) if isinstance(_lowerCAmelCase      ,      _lowerCAmelCase      ) else mask_token
  			A							  =      AddedToken(_lowerCAmelCase      ,      lstrip=_lowerCAmelCase      ,      rstrip=_lowerCAmelCase      ) if isinstance(_lowerCAmelCase      ,      _lowerCAmelCase      ) else cls_token
  			A							  =      AddedToken(_lowerCAmelCase      ,      lstrip=_lowerCAmelCase      ,      rstrip=_lowerCAmelCase      ) if isinstance(_lowerCAmelCase      ,      _lowerCAmelCase      ) else sep_token
  
  			super().__init__(
  			    pad_token=_lowerCAmelCase      ,      bos_token=_lowerCAmelCase      ,      eos_token=_lowerCAmelCase      ,      mask_token=_lowerCAmelCase      ,      cls_token=_lowerCAmelCase      ,      sep_token=_lowerCAmelCase      ,      model_max_length=_lowerCAmelCase      ,      **_lowerCAmelCase      ,      )
  
  			A							  =      2**8  # utf is 8 bits
  
  			# define special tokens dict
  			A							  =      {
  			    self.pad_token: 0,
  			    self.bos_token: 1,
  			    self.eos_token: 2,
  			    self.mask_token: 3,
  			    self.cls_token: 4,
  			    self.sep_token: 5,
  			}
  			A							  =      len(self.special_tokens_encoder      )
  			A							  =      {v: k for k, v in self.special_tokens_encoder.items()}
  
  
  
  
  
  
  
  		def        A     (self						:      Tuple      ):
  			A							  =      self.special_tokens_encoder.copy()
  			vocab.update(self.added_tokens_encoder      )
  			for i in range(self._utf_vocab_size      ):
  				A							  =      chr(_lowerCAmelCase      )
  				A							  =      i + len(self.special_tokens_encoder      )
  			return vocab
  
  
  
  
  
  
  
  		@property
  		def        A     (self						:      Tuple      ):
  			return self._utf_vocab_size + self._num_special_tokens
  
  
  
  
  
  
  
  		def        A     (self						:      Union[str, Any]      ,      _lowerCAmelCase						:      List[int]      ,      _lowerCAmelCase						:      Optional[List[int]] = None      ,      _lowerCAmelCase						:      bool = False      ):
  			if already_has_special_tokens:
  				return super().get_special_tokens_mask(
  				    token_ids_a=_lowerCAmelCase      ,      token_ids_a=_lowerCAmelCase      ,      already_has_special_tokens=_lowerCAmelCase      )
  
  			# normal case: some special tokens
  			if token_ids_a is None:
  				return [1] + [0] * len(_lowerCAmelCase      ) + [1]
  			return [1] + ([0] * len(_lowerCAmelCase      )) + [1] + ([0] * len(_lowerCAmelCase      )) + [1]
  
  
  
  
  
  
  
  		def        A     (self						:      List[Any]      ,      _lowerCAmelCase						:      List[int]      ,      _lowerCAmelCase						:      Optional[List[int]] = None      ):
  			if token_ids_a is None:
  				return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
  			else:
  				return [self.cls_token_id] + token_ids_a + [self.sep_token_id] + token_ids_a + [self.sep_token_id]
  
  
  
  
  
  
  
  		def        A     (self						:      str      ,      _lowerCAmelCase						:      str      ):
  			A							  =      [chr(_lowerCAmelCase      ) for i in text.encode("""utf-8"""      )]
  			return tokens
  
  
  
  
  
  
  
  		def        A     (self						:      int      ,      _lowerCAmelCase						:      List[str]      ):
  			if token in self.special_tokens_encoder:
  				A							  =      self.special_tokens_encoder[token]
  			elif token in self.added_tokens_encoder:
  				A							  =      self.added_tokens_encoder[token]
  			elif len(_lowerCAmelCase      ) != 1:
  				A							  =      self.unk_token_id
  			else:
  				A							  =      ord(_lowerCAmelCase      ) + self._num_special_tokens
  			return token_id
  
  
  
  
  
  
  
  		def        A     (self						:      Optional[int]      ,      _lowerCAmelCase						:      Any      ):
  			if index in self.special_tokens_decoder:
  				A							  =      self.special_tokens_decoder[index]
  			elif index in self.added_tokens_decoder:
  				A							  =      self.added_tokens_decoder[index]
  			else:
  				A							  =      chr(index - self._num_special_tokens      )
  			return token
  
  
  
  
  
  
  
  		def        A     (self						:      Dict      ,      _lowerCAmelCase						:      List[str]      ):
  			A							  =      b""""""
  			for token in tokens:
  				if token in self.special_tokens_decoder:
  					A							  =      self.special_tokens_decoder[token].encode("""utf-8"""      )
  				elif token in self.added_tokens_decoder:
  					A							  =      self.special_tokens_decoder[token].encode("""utf-8"""      )
  				elif token in self.special_tokens_encoder:
  					A							  =      token.encode("""utf-8"""      )
  				elif token in self.added_tokens_encoder:
  					A							  =      token.encode("""utf-8"""      )
  				else:
  					A							  =      bytes([ord(_lowerCAmelCase      )]      )
  				bstring += tok_string
  			A							  =      bstring.decode("""utf-8"""      ,      errors="""replace"""      )
  			return string
  
  
  
  
  
  
  
  		def        A     (self						:      Tuple      ,      _lowerCAmelCase						:      str      ,      _lowerCAmelCase						:      Optional[str] = None      ):
  			return ()
  
  