


"""simple docstring"""
import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
			import torch


SCREAMING_SNAKE_CASE					:       Optional[int]							   =							logging.get_logger(__name__)



class      _UpperCAmelCase					:






		'''simple docstring'''

		def __init__(self							,			a_							,			a_=21							,			a_=[]							,			a_=None   ):





			'''simple docstring'''

			__snake_case      :		Optional[int]         =							model
			__snake_case      :		List[Any]         =							trace_batch_nums
			__snake_case      :		str         =							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      :		Any         =							collections.deque([]							,			a_   )
			__snake_case      :		List[str]         =							[]
			__snake_case      :		Optional[int]         =							0
			__snake_case      :		Optional[Any]         =							0
			__snake_case      :		List[Any]         =							False
			__snake_case      :		Optional[Any]         =							'                 '

			self.analyse_model()

			self.register_forward_hook()



		def 							SCREAMING_SNAKE_CASE    (self							,			a_=None   ):





			'''simple docstring'''

			if frame is not None:
				self.expand_frame(a_   )
			self.frames.append('''\n'''.join(self.frame   )   )
			__snake_case      :		List[str]         =							[]  # start a new frame



		def 							SCREAMING_SNAKE_CASE    (self							,			a_   ):





			'''simple docstring'''

			self.frame.append(a_   )



		def 							SCREAMING_SNAKE_CASE    (self   ):





			'''simple docstring'''

			print('''\n'''.join(self.frames   )   )
			__snake_case      :		Optional[int]         =							[]



		def 							SCREAMING_SNAKE_CASE    (self   ):





			'''simple docstring'''

			__snake_case      :		Tuple         =							[]



		def 							SCREAMING_SNAKE_CASE    (self   ):





			'''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      :		Tuple         =							[]



		def 							SCREAMING_SNAKE_CASE    (self   ):





			'''simple docstring'''

			__snake_case      :		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 							SCREAMING_SNAKE_CASE    (self							,			a_							,			a_   ):





			'''simple docstring'''

			if torch.is_tensor(a_   ):
				self.expand_frame(get_abs_min_max(a_							,			a_   )   )
				if detect_overflow(a_							,			a_   ):
					__snake_case      :		Optional[int]         =							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   ):





			'''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   ):





			'''simple docstring'''

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



		def 							SCREAMING_SNAKE_CASE    (self							,			a_							,			a_							,			a_   ):





			'''simple docstring'''

			self.expand_frame(f"""{self.prefix} {self.module_names[module]} {module.__class__.__name__}"""   )

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

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

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

			self.save_frame()



		def 							SCREAMING_SNAKE_CASE    (self   ):





			'''simple docstring'''

			self.model.apply(self._register_forward_hook   )



		def 							SCREAMING_SNAKE_CASE    (self							,			a_   ):





			'''simple docstring'''

			module.register_forward_hook(self.forward_hook   )



		def 							SCREAMING_SNAKE_CASE    (self							,			a_							,			a_							,			a_   ):





			'''simple docstring'''

			__snake_case      :		List[Any]         =							False

			__snake_case      :		Tuple         =							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(a_							,			a_							,			a_   )

			# 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				(	_snake_case		: str	,  _snake_case		: str	)	->Optional[int]:
	"""simple docstring"""





	__snake_case      :		Optional[int]         =							var.abs()
	return f"""{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"""






def 						lowercase				(	_snake_case		: Dict	,  _snake_case		: str	)	->Optional[int]:
	"""simple docstring"""





	__snake_case      :		Any         =							False
	if torch.isnan(_snake_case	).any().item():
		__snake_case      :		Tuple         =							True
		print(f"""{ctx} has nans"""	)
	if torch.isinf(_snake_case	).any().item():
		__snake_case      :		List[str]         =							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()	,  100	)]
		if naaa.numel() > 0:
			print(f"""{ctx}:  n100={naaa.numel()}"""	)
		__snake_case      :		Dict         =							var[torch.ge(var.abs()	,  1_000	)]
		if naaaa.numel() > 0:
			print(f"""{ctx}: n1000={naaaa.numel()}"""	)
		__snake_case      :		Tuple         =							var[torch.ge(var.abs()	,  10_000	)]
		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'










'''simple docstring'''







from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available, is_vision_available


SCREAMING_SNAKE_CASE__     : Dict				=   {"""configuration_detr""": ["""DETR_PRETRAINED_CONFIG_ARCHIVE_MAP""", """DetrConfig""", """DetrOnnxConfig"""]}

try:
							if not is_vision_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							SCREAMING_SNAKE_CASE__     : List[Any]				=   ["""DetrFeatureExtractor"""]
							SCREAMING_SNAKE_CASE__     : str				=   ["""DetrImageProcessor"""]

try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							SCREAMING_SNAKE_CASE__     : Optional[Any]				=   [
							    """DETR_PRETRAINED_MODEL_ARCHIVE_LIST""",
							    """DetrForObjectDetection""",
							    """DetrForSegmentation""",
							    """DetrModel""",
							    """DetrPreTrainedModel""",
							]


if TYPE_CHECKING:
							from .configuration_detr import DETR_PRETRAINED_CONFIG_ARCHIVE_MAP, DetrConfig, DetrOnnxConfig

							try:
														if not is_vision_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .feature_extraction_detr import DetrFeatureExtractor
														from .image_processing_detr import DetrImageProcessor

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_detr import (
														    DETR_PRETRAINED_MODEL_ARCHIVE_LIST,
														    DetrForObjectDetection,
														    DetrForSegmentation,
														    DetrModel,
														    DetrPreTrainedModel,
														)

else:
							import sys

							SCREAMING_SNAKE_CASE__     : Optional[Any]				=   _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)
