

import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
	import torch


__A	:   List[str]	=					logging.get_logger(__name__)


class 			_SCREAMING_SNAKE_CASE      :







						def __init__(			self     ,						_SCREAMING_SNAKE_CASE     ,						_SCREAMING_SNAKE_CASE=21     ,						_SCREAMING_SNAKE_CASE=[]     ,						_SCREAMING_SNAKE_CASE=None							)->							int:
													lowerCamelCase_								=model
													lowerCamelCase_								=trace_batch_nums
													lowerCamelCase_								=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_								=collections.deque([]     ,						_SCREAMING_SNAKE_CASE							)
													lowerCamelCase_								=[]
													lowerCamelCase_								=0
													lowerCamelCase_								=0
													lowerCamelCase_								=False
													lowerCamelCase_								="""                 """

													self.analyse_model()

													self.register_forward_hook()







						def        _snake_case				(			self     ,						_SCREAMING_SNAKE_CASE=None							)->							Union[str, Any]:
													if frame is not None:
																				self.expand_frame(_SCREAMING_SNAKE_CASE							)
													self.frames.append("""\n""".join(self.frame							)							)
													lowerCamelCase_								=[]  # start a new frame







						def        _snake_case				(			self     ,						_SCREAMING_SNAKE_CASE							)->							Any:
													self.frame.append(_SCREAMING_SNAKE_CASE							)







						def        _snake_case				(			self							)->							List[str]:
													print("""\n""".join(self.frames							)							)
													lowerCamelCase_								=[]







						def        _snake_case				(			self							)->							Dict:
													lowerCamelCase_								=[]







						def        _snake_case				(			self							)->							str:
													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_								=[]







						def        _snake_case				(			self							)->							Optional[int]:
													# 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_								={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     ,						_SCREAMING_SNAKE_CASE     ,						_SCREAMING_SNAKE_CASE							)->							Optional[Any]:
													if torch.is_tensor(_SCREAMING_SNAKE_CASE							):
																				self.expand_frame(get_abs_min_max(_SCREAMING_SNAKE_CASE     ,						_SCREAMING_SNAKE_CASE							)							)
																				if detect_overflow(_SCREAMING_SNAKE_CASE     ,						_SCREAMING_SNAKE_CASE							):
																											lowerCamelCase_								=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[str]:
													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							)->							Dict:
													self.expand_frame(f'{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n'							)







						def        _snake_case				(			self     ,						_SCREAMING_SNAKE_CASE     ,						_SCREAMING_SNAKE_CASE     ,						_SCREAMING_SNAKE_CASE							)->							Optional[Any]:
													self.expand_frame(f'{self.prefix} {self.module_names[module]} {module.__class__.__name__}'							)

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

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

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

													self.save_frame()







						def        _snake_case				(			self							)->							int:
													self.model.apply(self._register_forward_hook							)







						def        _snake_case				(			self     ,						_SCREAMING_SNAKE_CASE							)->							Dict:
													module.register_forward_hook(self.forward_hook							)







						def        _snake_case				(			self     ,						_SCREAMING_SNAKE_CASE     ,						_SCREAMING_SNAKE_CASE     ,						_SCREAMING_SNAKE_CASE							)->							int:
													# - 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_								=False

													lowerCamelCase_								=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_								=True

													self.create_frame(_SCREAMING_SNAKE_CASE     ,						_SCREAMING_SNAKE_CASE     ,						_SCREAMING_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     (      _A					:							int      ,  _A					:							Optional[int]		)				->Tuple:



							"""simple docstring"""
							lowerCamelCase_								=var.abs()
							return f'{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}'




def        __UpperCamelCase     (      _A					:							Dict      ,  _A					:							Tuple		)				->Optional[int]:



							"""simple docstring"""
							lowerCamelCase_								=False
							if torch.isnan(_A		).any().item():
														lowerCamelCase_								=True
														print(f'{ctx} has nans'		)
							if torch.isinf(_A		).any().item():
														lowerCamelCase_								=True
														print(f'{ctx} has infs'		)

							# if needed to monitor large elements can enable the following
							if 0:  # and detected:
														lowerCamelCase_								=var[torch.ge(var.abs()      ,  100		)]
														if naaa.numel() > 0:
																					print(f'{ctx}:  n100={naaa.numel()}'		)
														lowerCamelCase_								=var[torch.ge(var.abs()      ,  1000		)]
														if naaaa.numel() > 0:
																					print(f'{ctx}: n1000={naaaa.numel()}'		)
														lowerCamelCase_								=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 			_SCREAMING_SNAKE_CASE      (  lowerCAmelCase__):
						_UpperCamelCase:Dict	=		"underflow_overflow"
						_UpperCamelCase:List[Any]	=		"tpu_metrics_debug"



from typing import TYPE_CHECKING

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


__A	:   Optional[int]	=					{
    'configuration_mobilevitv2': [
        'MOBILEVITV2_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'MobileViTV2Config',
        'MobileViTV2OnnxConfig',
    ],
}


try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__A	:   Tuple	=					[
	    'MOBILEVITV2_PRETRAINED_MODEL_ARCHIVE_LIST',
	    'MobileViTV2ForImageClassification',
	    'MobileViTV2ForSemanticSegmentation',
	    'MobileViTV2Model',
	    'MobileViTV2PreTrainedModel',
	]

if TYPE_CHECKING:
	from .configuration_mobilevitva import (
	    MOBILEVITV2_PRETRAINED_CONFIG_ARCHIVE_MAP,
	    MobileViTVaConfig,
	    MobileViTVaOnnxConfig,
	)

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_mobilevitva import (
		    MOBILEVITV2_PRETRAINED_MODEL_ARCHIVE_LIST,
		    MobileViTVaForImageClassification,
		    MobileViTVaForSemanticSegmentation,
		    MobileViTVaModel,
		    MobileViTVaPreTrainedModel,
		)

else:
	import sys

	__A	:   List[str]	=					_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)
