
import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
   import torch


__lowerCamelCase							= logging.get_logger(__name__)






class 						UpperCAmelCase	:



   def __init__(self	:      Optional[Any]					,		snake_case__	:      Optional[Any]					,		snake_case__	:      Union[str, Any]=21					,		snake_case__	:      Optional[int]=[]					,		snake_case__	:      Dict=None				)	->			Union[str, Any]:


          '''simple docstring'''


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

          self.analyse_model()

          self.register_forward_hook()



   def        _SCREAMING_SNAKE_CASE     (self	:      int					,		snake_case__	:      List[str]=None				)	->			List[str]:


          '''simple docstring'''


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



   def        _SCREAMING_SNAKE_CASE     (self	:      str					,		snake_case__	:      Optional[int]				)	->			List[str]:


          '''simple docstring'''


          self.frame.append(snake_case__				)



   def        _SCREAMING_SNAKE_CASE     (self	:      List[str]				)	->			Any:


          '''simple docstring'''


          print("\n".join(self.frames				)				)
          snake_case :     str           =   []



   def        _SCREAMING_SNAKE_CASE     (self	:      Optional[Any]				)	->			Optional[Any]:


          '''simple docstring'''


          snake_case :     Union[str, Any]           =   []



   def        _SCREAMING_SNAKE_CASE     (self	:      int				)	->			Optional[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"				)
          snake_case :     Any           =   []



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


          '''simple docstring'''


          snake_case :     Dict           =   {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        _SCREAMING_SNAKE_CASE     (self	:      Optional[int]					,		snake_case__	:      int					,		snake_case__	:      Any				)	->			Any:


          '''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__				):
                        snake_case :     str           =   True
          elif var is None:
                 self.expand_frame(f"""{'None':>17} {ctx}"""				)
          else:
                 self.expand_frame(f"""{'not a tensor':>17} {ctx}"""				)



   def        _SCREAMING_SNAKE_CASE     (self	:      Union[str, Any]				)	->			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        _SCREAMING_SNAKE_CASE     (self	:      int				)	->			str:


          '''simple docstring'''


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



   def        _SCREAMING_SNAKE_CASE     (self	:      Optional[int]					,		snake_case__	:      int					,		snake_case__	:      str					,		snake_case__	:      Dict				)	->			List[str]:


          '''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        _SCREAMING_SNAKE_CASE     (self	:      Any				)	->			Any:


          '''simple docstring'''


          self.model.apply(self._register_forward_hook				)



   def        _SCREAMING_SNAKE_CASE     (self	:      str					,		snake_case__	:      Tuple				)	->			Optional[Any]:


          '''simple docstring'''


          module.register_forward_hook(self.forward_hook				)



   def        _SCREAMING_SNAKE_CASE     (self	:      Union[str, Any]					,		snake_case__	:      Tuple					,		snake_case__	:      List[Any]					,		snake_case__	:      int				)	->			int:


          '''simple docstring'''


          snake_case :     List[Any]           =   False

          snake_case :     List[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
                 snake_case :     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     UpperCamelCase      (					__lowerCamelCase  :				Tuple       ,      __lowerCamelCase  :				Optional[Any]					):
       snake_case :     Tuple           =   var.abs()
       return f"""{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"""


def     UpperCamelCase      (					__lowerCamelCase  :				Tuple       ,      __lowerCamelCase  :				List[Any]					):
       snake_case :     str           =   False
       if torch.isnan(__lowerCamelCase					).any().item():
              snake_case :     str           =   True
              print(f"""{ctx} has nans"""					)
       if torch.isinf(__lowerCamelCase					).any().item():
              snake_case :     str           =   True
              print(f"""{ctx} has infs"""					)

       # if needed to monitor large elements can enable the following
       if 0:  # and detected:
              snake_case :     Optional[Any]           =   var[torch.ge(var.abs()       ,      100					)]
              if naaa.numel() > 0:
                     print(f"""{ctx}:  n100={naaa.numel()}"""					)
              snake_case :     Dict           =   var[torch.ge(var.abs()       ,      1000					)]
              if naaaa.numel() > 0:
                     print(f"""{ctx}: n1000={naaaa.numel()}"""					)
              snake_case :     Union[str, 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 						UpperCAmelCase	(		A_				):
   A__      :	str		  =  "underflow_overflow"
   A__      :	Any		  =  "tpu_metrics_debug"







import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import Mock, patch

import diffusers.utils.hub_utils






class 						UpperCAmelCase	(		unittest.TestCase				):



   @patch("diffusers.utils.hub_utils.get_full_repo_name"				)
   def        _SCREAMING_SNAKE_CASE     (self	:      Any					,		snake_case__	:      Mock				)	->			None:


          '''simple docstring'''


          snake_case :     str           =   "full_repo_name"
          with TemporaryDirectory() as tmpdir:
                 # Dummy args values
                 snake_case :     List[str]           =   Mock()
                 snake_case :     List[str]           =   tmpdir
                 snake_case :     Optional[int]           =   0
                 snake_case :     int           =   "hub_token"
                 snake_case :     Union[str, Any]           =   "dataset_name"
                 snake_case :     Tuple           =   0.01
                 snake_case :     int           =   10_00_00
                 snake_case :     Optional[Any]           =   1_00_00
                 snake_case :     Optional[Any]           =   0.01
                 snake_case :     Dict           =   0.02
                 snake_case :     List[Any]           =   0.03
                 snake_case :     str           =   0.0005
                 snake_case :     List[str]           =   0.000001
                 snake_case :     Tuple           =   1
                 snake_case :     Optional[int]           =   10
                 snake_case :     List[str]           =   0.001
                 snake_case :     List[Any]           =   0.1
                 snake_case :     List[str]           =   0.2
                 snake_case :     Tuple           =   True

                 # Model card mush be rendered and saved
                 diffusers.utils.hub_utils.create_model_card(snake_case__					,		model_name="model_name"				)
                 self.assertTrue((Path(snake_case__				) / "README.md").is_file()				)





