




'''simple docstring'''
import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
  import torch


A							:      Optional[Any] 	=       logging.get_logger(__name__)






class        SCREAMING_SNAKE_CASE:

       def __init__(    self					,		lowerCamelCase__					,		lowerCamelCase__=21					,		lowerCamelCase__=[]					,		lowerCamelCase__=None     )    ->					Optional[int]:

           """simple docstring"""



           __lowercase							=						model
           __lowercase							=						trace_batch_nums
           __lowercase							=						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
           __lowercase							=						collections.deque([]					,		lowerCamelCase__     )
           __lowercase							=						[]
           __lowercase							=						0
           __lowercase							=						0
           __lowercase							=						False
           __lowercase							=						"""                 """

           self.analyse_model()

           self.register_forward_hook()

       def     snake_case__			(    self					,		lowerCamelCase__=None     )    ->					Optional[int]:

           """simple docstring"""



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

       def     snake_case__			(    self					,		lowerCamelCase__     )    ->					int:

           """simple docstring"""



           self.frame.append(lowerCamelCase__     )

       def     snake_case__			(    self     )    ->					Any:

           """simple docstring"""



           print("""\n""".join(self.frames     )     )
           __lowercase							=						[]

       def     snake_case__			(    self     )    ->					Dict:

           """simple docstring"""



           __lowercase							=						[]

       def     snake_case__			(    self     )    ->					Dict:

           """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"""     )
           __lowercase							=						[]

       def     snake_case__			(    self     )    ->					int:

           """simple docstring"""



           __lowercase							=						{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     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__     )    ->					str:

           """simple docstring"""



           if torch.is_tensor(lowerCamelCase__     ):
               self.expand_frame(get_abs_min_max(lowerCamelCase__					,		lowerCamelCase__     )     )
               if detect_overflow(lowerCamelCase__					,		lowerCamelCase__     ):
                   __lowercase							=						True
           elif var is None:
               self.expand_frame(F'{"None":>17} {ctx}'     )
           else:
               self.expand_frame(F'{"not a tensor":>17} {ctx}'     )

       def     snake_case__			(    self     )    ->					List[Any]:

           """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     snake_case__			(    self     )    ->					Optional[int]:

           """simple docstring"""



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

       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__					,		lowerCamelCase__     )    ->					Optional[Any]:

           """simple docstring"""



           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     snake_case__			(    self     )    ->					Optional[Any]:

           """simple docstring"""



           self.model.apply(self._register_forward_hook     )

       def     snake_case__			(    self					,		lowerCamelCase__     )    ->					List[str]:

           """simple docstring"""



           module.register_forward_hook(self.forward_hook     )




       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__					,		lowerCamelCase__     )    ->					int:

           """simple docstring"""



           __lowercase							=						False

           __lowercase							=						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
               __lowercase							=						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 				snake_case_  (   a__		:	List[str]    ,a__		:	Union[str, Any]			):



    """simple docstring"""




    __lowercase							=						var.abs()
    return f'{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}'
def 				snake_case_  (   a__		:	int    ,a__		:	Any			):



    """simple docstring"""




    __lowercase							=						False
    if torch.isnan(a__			).any().item():
        __lowercase							=						True
        print(f'{ctx} has nans'			)
    if torch.isinf(a__			).any().item():
        __lowercase							=						True
        print(f'{ctx} has infs'			)

    # if needed to monitor large elements can enable the following
    if 0:  # and detected:
        __lowercase							=						var[torch.ge(var.abs()    ,1_00			)]
        if naaa.numel() > 0:
            print(f'{ctx}:  n100={naaa.numel()}'			)
        __lowercase							=						var[torch.ge(var.abs()    ,10_00			)]
        if naaaa.numel() > 0:
            print(f'{ctx}: n1000={naaaa.numel()}'			)
        __lowercase							=						var[torch.ge(var.abs()    ,1_00_00			)]
        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        SCREAMING_SNAKE_CASE(				__A     ):
       snake_case_ :     int    =					"""underflow_overflow"""
       snake_case_ :     List[Any]    =					"""tpu_metrics_debug"""










'''simple docstring'''
import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
  import torch

if is_vision_available():
  from PIL import Image

  from transformers import BlipImageProcessor






class        SCREAMING_SNAKE_CASE(				unittest.TestCase     ):

       def __init__(    self					,		lowerCamelCase__					,		lowerCamelCase__=7					,		lowerCamelCase__=3					,		lowerCamelCase__=18					,		lowerCamelCase__=30					,		lowerCamelCase__=400					,		lowerCamelCase__=True					,		lowerCamelCase__=None					,		lowerCamelCase__=True					,		lowerCamelCase__=False					,		lowerCamelCase__=[0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73]					,		lowerCamelCase__=[0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11]					,		lowerCamelCase__=True					,		)    ->					Union[str, Any]:

           """simple docstring"""



           __lowercase							=						size if size is not None else {"""height""": 20, """width""": 20}
           __lowercase							=						parent
           __lowercase							=						batch_size
           __lowercase							=						num_channels
           __lowercase							=						image_size
           __lowercase							=						min_resolution
           __lowercase							=						max_resolution
           __lowercase							=						do_resize
           __lowercase							=						size
           __lowercase							=						do_normalize
           __lowercase							=						image_mean
           __lowercase							=						image_std
           __lowercase							=						do_pad
           __lowercase							=						do_convert_rgb

       def     snake_case__			(    self     )    ->					Optional[int]:

           """simple docstring"""



           return {
               "do_resize": self.do_resize,
               "size": self.size,
               "do_normalize": self.do_normalize,
               "image_mean": self.image_mean,
               "image_std": self.image_std,
               "do_convert_rgb": self.do_convert_rgb,
               "do_pad": self.do_pad,
           }




       def     snake_case__			(    self					,		lowerCamelCase__=False					,		lowerCamelCase__=False					,		lowerCamelCase__=False     )    ->					Tuple:

           """simple docstring"""



           assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

           if equal_resolution:
               __lowercase							=						[]
               for i in range(self.batch_size     ):
                   image_inputs.append(
                       np.random.randint(
                           255					,		size=(self.num_channels, self.max_resolution, self.max_resolution)					,		dtype=np.uinta     )     )
           else:
               __lowercase							=						[]
               for i in range(self.batch_size     ):
                   __lowercase				,__lowercase							=						np.random.choice(np.arange(self.min_resolution					,		self.max_resolution     )					,		2     )
                   image_inputs.append(np.random.randint(255					,		size=(self.num_channels, width, height)					,		dtype=np.uinta     )     )

           if not numpify and not torchify:
               # PIL expects the channel dimension as last dimension
               __lowercase							=						[Image.fromarray(np.moveaxis(lowerCamelCase__					,		0					,		-1     )     ) for x in image_inputs]

           if torchify:
               __lowercase							=						[torch.from_numpy(lowerCamelCase__     ) for x in image_inputs]

           return image_inputs






@require_torch
@require_vision
class        SCREAMING_SNAKE_CASE(				__A   ,      unittest.TestCase     ):
       snake_case_ :     Optional[int]    =					BlipImageProcessor if is_vision_available() else None

       def     snake_case__			(    self     )    ->					List[Any]:

           """simple docstring"""



           __lowercase							=						BlipImageProcessingTester(self     )

       @property
       def     snake_case__			(    self     )    ->					Optional[int]:

           """simple docstring"""



           return self.image_processor_tester.prepare_image_processor_dict()

       def     snake_case__			(    self     )    ->					List[Any]:

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_resize"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""size"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_normalize"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""image_mean"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""image_std"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_convert_rgb"""     )     )

       def     snake_case__			(    self     )    ->					Any:

           """simple docstring"""



           pass

       def     snake_case__			(    self     )    ->					List[str]:

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           # create random PIL images
           __lowercase							=						self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__     )
           for image in image_inputs:
               self.assertIsInstance(lowerCamelCase__					,		Image.Image     )

           # Test not batched input
           __lowercase							=						image_processor(image_inputs[0]					,		return_tensors="""pt"""     ).pixel_values
           self.assertEqual(
               encoded_images.shape					,		(
                   1,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.size["""height"""],
                   self.image_processor_tester.size["""width"""],
               )					,		)

           # Test batched
           __lowercase							=						image_processor(lowerCamelCase__					,		return_tensors="""pt"""     ).pixel_values
           self.assertEqual(
               encoded_images.shape					,		(
                   self.image_processor_tester.batch_size,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.size["""height"""],
                   self.image_processor_tester.size["""width"""],
               )					,		)

       def     snake_case__			(    self     )    ->					int:

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           # create random numpy tensors
           __lowercase							=						self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__					,		numpify=lowerCamelCase__     )
           for image in image_inputs:
               self.assertIsInstance(lowerCamelCase__					,		np.ndarray     )

           # Test not batched input
           __lowercase							=						image_processor(image_inputs[0]					,		return_tensors="""pt"""     ).pixel_values
           self.assertEqual(
               encoded_images.shape					,		(
                   1,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.size["""height"""],
                   self.image_processor_tester.size["""width"""],
               )					,		)

           # Test batched
           __lowercase							=						image_processor(lowerCamelCase__					,		return_tensors="""pt"""     ).pixel_values
           self.assertEqual(
               encoded_images.shape					,		(
                   self.image_processor_tester.batch_size,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.size["""height"""],
                   self.image_processor_tester.size["""width"""],
               )					,		)




       def     snake_case__			(    self     )    ->					Any:

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           # create random PyTorch tensors
           __lowercase							=						self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__					,		torchify=lowerCamelCase__     )
           for image in image_inputs:
               self.assertIsInstance(lowerCamelCase__					,		torch.Tensor     )

           # Test not batched input
           __lowercase							=						image_processor(image_inputs[0]					,		return_tensors="""pt"""     ).pixel_values
           self.assertEqual(
               encoded_images.shape					,		(
                   1,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.size["""height"""],
                   self.image_processor_tester.size["""width"""],
               )					,		)

           # Test batched
           __lowercase							=						image_processor(lowerCamelCase__					,		return_tensors="""pt"""     ).pixel_values
           self.assertEqual(
               encoded_images.shape					,		(
                   self.image_processor_tester.batch_size,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.size["""height"""],
                   self.image_processor_tester.size["""width"""],
               )					,		)






@require_torch
@require_vision
class        SCREAMING_SNAKE_CASE(				__A   ,      unittest.TestCase     ):
       snake_case_ :     Union[str, Any]    =					BlipImageProcessor if is_vision_available() else None

       def     snake_case__			(    self     )    ->					List[str]:

           """simple docstring"""



           __lowercase							=						BlipImageProcessingTester(self					,		num_channels=4     )
           __lowercase							=						3

       @property
       def     snake_case__			(    self     )    ->					str:

           """simple docstring"""



           return self.image_processor_tester.prepare_image_processor_dict()

       def     snake_case__			(    self     )    ->					Any:

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_resize"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""size"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_normalize"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""image_mean"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""image_std"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_convert_rgb"""     )     )

       def     snake_case__			(    self     )    ->					Any:

           """simple docstring"""



           pass




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

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           # create random PIL images
           __lowercase							=						self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__     )
           for image in image_inputs:
               self.assertIsInstance(lowerCamelCase__					,		Image.Image     )

           # Test not batched input
           __lowercase							=						image_processor(image_inputs[0]					,		return_tensors="""pt"""     ).pixel_values
           self.assertEqual(
               encoded_images.shape					,		(
                   1,
                   self.expected_encoded_image_num_channels,
                   self.image_processor_tester.size["""height"""],
                   self.image_processor_tester.size["""width"""],
               )					,		)

           # Test batched
           __lowercase							=						image_processor(lowerCamelCase__					,		return_tensors="""pt"""     ).pixel_values
           self.assertEqual(
               encoded_images.shape					,		(
                   self.image_processor_tester.batch_size,
                   self.expected_encoded_image_num_channels,
                   self.image_processor_tester.size["""height"""],
                   self.image_processor_tester.size["""width"""],
               )					,		)




