


"""simple docstring"""





import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
   import torch


__UpperCAmelCase    				=      logging.get_logger(__name__)

class  lowercase_  :







      def __init__(      self :							Tuple		,  _lowercase :							Union[str, Any]		,  _lowercase :							Any=2_1		,  _lowercase :							str=[]		,  _lowercase :							List[str]=None    ):
        lowerCAmelCase__    :    Union[str, Any]								=					model
        lowerCAmelCase__    :    Union[str, Any]								=					trace_batch_nums
        lowerCAmelCase__    :    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
        lowerCAmelCase__    :    Optional[Any]								=					collections.deque([]		,  _lowercase    )
        lowerCAmelCase__    :    Any								=					[]
        lowerCAmelCase__    :    str								=					0
        lowerCAmelCase__    :    Dict								=					0
        lowerCAmelCase__    :    Any								=					False
        lowerCAmelCase__    :    Optional[Any]								=					"                 "

        self.analyse_model()

        self.register_forward_hook()







      def 				_lowerCAmelCase       (      self :							Union[str, Any]		,  _lowercase :							Union[str, Any]=None    ):
        if frame is not None:
          self.expand_frame(_lowercase    )
        self.frames.append("\n".join(self.frame    )    )
        lowerCAmelCase__    :    Optional[Any]								=					[]  # start a new frame







      def 				_lowerCAmelCase       (      self :							Union[str, Any]		,  _lowercase :							int    ):
        self.frame.append(_lowercase    )







      def 				_lowerCAmelCase       (      self :							List[str]    ):
        print("\n".join(self.frames    )    )
        lowerCAmelCase__    :    Dict								=					[]







      def 				_lowerCAmelCase       (      self :							Union[str, Any]    ):
        lowerCAmelCase__    :    Tuple								=					[]







      def 				_lowerCAmelCase       (      self :							Union[str, 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"    )
        lowerCAmelCase__    :    Dict								=					[]







      def 				_lowerCAmelCase       (      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
        lowerCAmelCase__    :    Tuple								=					{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 				_lowerCAmelCase       (      self :							Any		,  _lowercase :							Union[str, Any]		,  _lowercase :							Union[str, Any]    ):
        if torch.is_tensor(_lowercase    ):
          self.expand_frame(get_abs_min_max(_lowercase		,  _lowercase    )    )
          if detect_overflow(_lowercase		,  _lowercase    ):
            lowerCAmelCase__    :    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 				_lowerCAmelCase       (      self :							List[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 				_lowerCAmelCase       (      self :							Any    ):
        self.expand_frame(f"{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n"    )







      def 				_lowerCAmelCase       (      self :							List[str]		,  _lowercase :							List[Any]		,  _lowercase :							Dict		,  _lowercase :							int    ):
        self.expand_frame(f"{self.prefix} {self.module_names[module]} {module.__class__.__name__}"    )

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

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

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

        self.save_frame()







      def 				_lowerCAmelCase       (      self :							Tuple    ):
        self.model.apply(self._register_forward_hook    )







      def 				_lowerCAmelCase       (      self :							Dict		,  _lowercase :							int    ):
        module.register_forward_hook(self.forward_hook    )







      def 				_lowerCAmelCase       (      self :							Optional[int]		,  _lowercase :							int		,  _lowercase :							str		,  _lowercase :							List[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

        lowerCAmelCase__    :    str								=					False

        lowerCAmelCase__    :    Dict								=					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
          lowerCAmelCase__    :    List[str]								=					True

        self.create_frame(_lowercase		,  _lowercase		,  _lowercase    )

        # 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     lowercase__	(  lowerCamelCase     :  Optional[Any]				,	lowerCamelCase     :  List[str]				)					->     Dict:
  lowerCAmelCase__    :    Dict								=					var.abs()
  return F"{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"





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

  # if needed to monitor large elements can enable the following
  if 0:  # and detected:
    lowerCAmelCase__    :    List[str]								=					var[torch.ge(var.abs()				,	1_0_0				)]
    if naaa.numel() > 0:
      print(F"{ctx}:  n100={naaa.numel()}"				)
    lowerCAmelCase__    :    Tuple								=					var[torch.ge(var.abs()				,	1_0_0_0				)]
    if naaaa.numel() > 0:
      print(F"{ctx}: n1000={naaaa.numel()}"				)
    lowerCAmelCase__    :    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  lowercase_  ( a_						):
      __magic_name__     :							Optional[int]          =				"""underflow_overflow"""
      __magic_name__     :							Optional[int]          =				"""tpu_metrics_debug"""




"""simple docstring"""





from __future__ import annotations





def     lowercase__	(  lowerCamelCase     :  list				,	lowerCamelCase     :  list				)					->     list:
  lowerCAmelCase__    :    Dict								=					[None] * (len(lowerCamelCase				) + len(lowerCamelCase				))

  lowerCAmelCase__    :    Optional[Any]								=					0  # pointer to current index for left Half
  lowerCAmelCase__    :    Any								=					0  # pointer to current index for the right Half
  lowerCAmelCase__    :    Dict								=					0  # pointer to current index for the sorted array Half

  while pointera < len(lowerCamelCase				) and pointera < len(lowerCamelCase				):
    if left_half[pointera] < right_half[pointera]:
      lowerCAmelCase__    :    Tuple								=					left_half[pointera]
      pointera += 1
      index += 1
    else:
      lowerCAmelCase__    :    int								=					right_half[pointera]
      pointera += 1
      index += 1
  while pointera < len(lowerCamelCase				):
    lowerCAmelCase__    :    Tuple								=					left_half[pointera]
    pointera += 1
    index += 1

  while pointera < len(lowerCamelCase				):
    lowerCAmelCase__    :    Optional[int]								=					right_half[pointera]
    pointera += 1
    index += 1

  return sorted_array





def     lowercase__	(  lowerCamelCase     :  list				)					->     list:
  if len(lowerCamelCase				) <= 1:
    return array
  # the actual formula to calculate the middle element = left + (right - left) // 2
  # this avoids integer overflow in case of large N
  lowerCAmelCase__    :    Optional[int]								=					0 + (len(lowerCamelCase				) - 0) // 2

  # Split the array into halves till the array length becomes equal to One
  # merge the arrays of single length returned by mergeSort function and
  # pass them into the merge arrays function which merges the array
  lowerCAmelCase__    :    Union[str, Any]								=					array[:middle]
  lowerCAmelCase__    :    List[str]								=					array[middle:]

  return merge(merge_sort(lowerCamelCase				)				,	merge_sort(lowerCamelCase				)				)


if __name__ == "__main__":
   import doctest

   doctest.testmod()
