



"""simple docstring"""







import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
			import torch


A    =	logging.get_logger(__name__)

class 			_a     :



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

									self.analyse_model()

									self.register_forward_hook()



			def 							__lowercase				( self       :							int      ,  _lowercase       :							List[Any]=None		)       ->				List[str]:
									if frame is not None:
															self.expand_frame(_lowercase		)
									self.frames.append("\n".join(self.frame		)		)
									snake_case      : Optional[Any]							=   []  # start a new frame



			def 							__lowercase				( self       :							Optional[Any]      ,  _lowercase       :							Union[str, Any]		)       ->				Optional[Any]:
									self.frame.append(_lowercase		)



			def 							__lowercase				( self       :							Union[str, Any]		)       ->				str:
									print("\n".join(self.frames		)		)
									snake_case      : Optional[Any]							=   []



			def 							__lowercase				( self       :							Optional[Any]		)       ->				Optional[Any]:
									snake_case      : Dict							=   []



			def 							__lowercase				( self       :							Dict		)       ->				Optional[int]:
									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      : str							=   []



			def 							__lowercase				( self       :							List[Any]		)       ->				Tuple:
									# 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
									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 							__lowercase				( self       :							Tuple      ,  _lowercase       :							Tuple      ,  _lowercase       :							int		)       ->				str:
									if torch.is_tensor(_lowercase		):
															self.expand_frame(get_abs_min_max(_lowercase      ,  _lowercase		)		)
															if detect_overflow(_lowercase      ,  _lowercase		):
																					snake_case      : Optional[Any]							=   True
									elif var is None:
															self.expand_frame(F'''{"None":>17} {ctx}'''		)
									else:
															self.expand_frame(F'''{"not a tensor":>17} {ctx}'''		)



			def 							__lowercase				( self       :							List[str]		)       ->				Union[str, 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 							__lowercase				( self       :							Union[str, Any]		)       ->				Union[str, Any]:
									self.expand_frame(F'''{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n'''		)



			def 							__lowercase				( self       :							Optional[Any]      ,  _lowercase       :							Any      ,  _lowercase       :							Union[str, Any]      ,  _lowercase       :							Union[str, Any]		)       ->				Tuple:
									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 							__lowercase				( self       :							Optional[Any]		)       ->				str:
									self.model.apply(self._register_forward_hook		)



			def 							__lowercase				( self       :							List[str]      ,  _lowercase       :							str		)       ->				int:
									module.register_forward_hook(self.forward_hook		)



			def 							__lowercase				( self       :							List[Any]      ,  _lowercase       :							Any      ,  _lowercase       :							Optional[int]      ,  _lowercase       :							List[Any]		)       ->				List[Any]:
									# - input is a tuple of packed inputs (could be non-Tensors)
									# - output could be a Tensor or a tuple of Tensors and non-Tensors

									snake_case      : Union[str, 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      : 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    __SCREAMING_SNAKE_CASE		( lowerCamelCase_:       List[str]    ,		lowerCamelCase_:       Optional[Any]   ):
						"""simple docstring"""






						snake_case      : Any							=   var.abs()
						return f'''{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}'''

def    __SCREAMING_SNAKE_CASE		( lowerCamelCase_:       List[Any]    ,		lowerCamelCase_:       str   ):
						"""simple docstring"""






						snake_case      : Any							=   False
						if torch.isnan(lowerCamelCase_   ).any().item():
												snake_case      : List[Any]							=   True
												print(f'''{ctx} has nans'''   )
						if torch.isinf(lowerCamelCase_   ).any().item():
												snake_case      : Union[str, Any]							=   True
												print(f'''{ctx} has infs'''   )

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


if is_torch_available():
			import torch

if is_vision_available():
			from PIL import Image

			from transformers import OwlViTImageProcessor

class 			_a     (   unittest.TestCase):



			def __init__( self       :							Optional[int]      ,  _lowercase       :							int      ,  _lowercase       :							Union[str, Any]=7      ,  _lowercase       :							Union[str, Any]=3      ,  _lowercase       :							int=18      ,  _lowercase       :							Any=30      ,  _lowercase       :							str=400      ,  _lowercase       :							str=True      ,  _lowercase       :							Dict=None      ,  _lowercase       :							int=True      ,  _lowercase       :							Union[str, Any]=None      ,  _lowercase       :							str=True      ,  _lowercase       :							Union[str, Any]=[0.48145466, 0.4578275, 0.40821073]      ,  _lowercase       :							str=[0.26862954, 0.26130258, 0.27577711]      ,  _lowercase       :							Optional[int]=True      ,  )       ->				Optional[int]:
									snake_case      : str							=   parent
									snake_case      : Any							=   batch_size
									snake_case      : Optional[int]							=   num_channels
									snake_case      : Any							=   image_size
									snake_case      : Optional[Any]							=   min_resolution
									snake_case      : int							=   max_resolution
									snake_case      : List[Any]							=   do_resize
									snake_case      : Dict							=   size if size is not None else {"height": 18, "width": 18}
									snake_case      : str							=   do_center_crop
									snake_case      : List[str]							=   crop_size if crop_size is not None else {"height": 18, "width": 18}
									snake_case      : str							=   do_normalize
									snake_case      : str							=   image_mean
									snake_case      : Optional[Any]							=   image_std
									snake_case      : str							=   do_convert_rgb



			def 							__lowercase				( self       :							Tuple		)       ->				Union[str, Any]:
									return {
									    "do_resize": self.do_resize,
									    "size": self.size,
									    "do_center_crop": self.do_center_crop,
									    "crop_size": self.crop_size,
									    "do_normalize": self.do_normalize,
									    "image_mean": self.image_mean,
									    "image_std": self.image_std,
									    "do_convert_rgb": self.do_convert_rgb,
									}



@require_torch
@require_vision
class 			_a     (   SCREAMING_SNAKE_CASE__    ,  unittest.TestCase):
			__magic_name__    				=    OwlViTImageProcessor if is_vision_available() else None



			def 							__lowercase				( self       :							Any		)       ->				List[Any]:
									snake_case      : Dict							=   OwlViTImageProcessingTester(self		)



			@property
			def 							__lowercase				( self       :							Optional[int]		)       ->				List[Any]:
									return self.image_processor_tester.prepare_image_processor_dict()



			def 							__lowercase				( self       :							List[Any]		)       ->				Union[str, Any]:
									snake_case      : List[Any]							=   self.image_processing_class(**self.image_processor_dict		)
									self.assertTrue(hasattr(_lowercase      ,  "do_resize"		)		)
									self.assertTrue(hasattr(_lowercase      ,  "size"		)		)
									self.assertTrue(hasattr(_lowercase      ,  "do_center_crop"		)		)
									self.assertTrue(hasattr(_lowercase      ,  "center_crop"		)		)
									self.assertTrue(hasattr(_lowercase      ,  "do_normalize"		)		)
									self.assertTrue(hasattr(_lowercase      ,  "image_mean"		)		)
									self.assertTrue(hasattr(_lowercase      ,  "image_std"		)		)
									self.assertTrue(hasattr(_lowercase      ,  "do_convert_rgb"		)		)



			def 							__lowercase				( self       :							Any		)       ->				Any:
									snake_case      : List[Any]							=   self.image_processing_class.from_dict(self.image_processor_dict		)
									self.assertEqual(image_processor.size      ,  {"height": 18, "width": 18}		)
									self.assertEqual(image_processor.crop_size      ,  {"height": 18, "width": 18}		)

									snake_case      : int							=   self.image_processing_class.from_dict(self.image_processor_dict      ,  size=42      ,  crop_size=84		)
									self.assertEqual(image_processor.size      ,  {"height": 42, "width": 42}		)
									self.assertEqual(image_processor.crop_size      ,  {"height": 84, "width": 84}		)



			def 							__lowercase				( self       :							Any		)       ->				Optional[int]:
									# Initialize image_processing
									snake_case      : List[Any]							=   self.image_processing_class(**self.image_processor_dict		)
									# create random PIL images
									snake_case      : Tuple							=   prepare_image_inputs(self.image_processor_tester      ,  equal_resolution=_lowercase		)

									for image in image_inputs:
															self.assertIsInstance(_lowercase      ,  Image.Image		)

									# Test not batched input
									snake_case      : Dict							=   image_processing(image_inputs[0]      ,  return_tensors="pt"		).pixel_values
									self.assertEqual(
									    encoded_images.shape      ,  (
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size["height"],
									        self.image_processor_tester.crop_size["width"],
									    )      ,  )

									# Test batched
									snake_case      : Optional[Any]							=   image_processing(_lowercase      ,  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.crop_size["height"],
									        self.image_processor_tester.crop_size["width"],
									    )      ,  )



			def 							__lowercase				( self       :							Optional[Any]		)       ->				Dict:
									# Initialize image_processing
									snake_case      : Optional[int]							=   self.image_processing_class(**self.image_processor_dict		)
									# create random numpy tensors
									snake_case      : int							=   prepare_image_inputs(self.image_processor_tester      ,  equal_resolution=_lowercase      ,  numpify=_lowercase		)
									for image in image_inputs:
															self.assertIsInstance(_lowercase      ,  np.ndarray		)

									# Test not batched input
									snake_case      : str							=   image_processing(image_inputs[0]      ,  return_tensors="pt"		).pixel_values
									self.assertEqual(
									    encoded_images.shape      ,  (
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size["height"],
									        self.image_processor_tester.crop_size["width"],
									    )      ,  )

									# Test batched
									snake_case      : str							=   image_processing(_lowercase      ,  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.crop_size["height"],
									        self.image_processor_tester.crop_size["width"],
									    )      ,  )



			def 							__lowercase				( self       :							List[Any]		)       ->				List[str]:
									# Initialize image_processing
									snake_case      : Any							=   self.image_processing_class(**self.image_processor_dict		)
									# create random PyTorch tensors
									snake_case      : Dict							=   prepare_image_inputs(self.image_processor_tester      ,  equal_resolution=_lowercase      ,  torchify=_lowercase		)
									for image in image_inputs:
															self.assertIsInstance(_lowercase      ,  torch.Tensor		)

									# Test not batched input
									snake_case      : Union[str, Any]							=   image_processing(image_inputs[0]      ,  return_tensors="pt"		).pixel_values
									self.assertEqual(
									    encoded_images.shape      ,  (
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size["height"],
									        self.image_processor_tester.crop_size["width"],
									    )      ,  )

									# Test batched
									snake_case      : Any							=   image_processing(_lowercase      ,  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.crop_size["height"],
									        self.image_processor_tester.crop_size["width"],
									    )      ,  )



