





import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
							import torch


UpperCAmelCase						:       int        =					logging.get_logger(__name__)





class 						_A:


				"""simple docstring"""


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

									self.analyse_model()

									self.register_forward_hook()



				def  UpperCAmelCase_		(      self					, _A=None					):
									if frame is not None:
														self.expand_frame(_A					)
									self.frames.append('\n'.join(self.frame					)					)
									__A     :						Any          =    []  # start a new frame



				def  UpperCAmelCase_		(      self					, _A					):
									self.frame.append(_A					)



				def  UpperCAmelCase_		(      self					):
									print('\n'.join(self.frames					)					)
									__A     :						str          =    []



				def  UpperCAmelCase_		(      self					):
									__A     :						Any          =    []



				def  UpperCAmelCase_		(      self					):
									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'					)
									__A     :						Optional[Any]          =    []



				def  UpperCAmelCase_		(      self					):
									# 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
									__A     :						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  UpperCAmelCase_		(      self					, _A					, _A					):
									if torch.is_tensor(_A					):
														self.expand_frame(get_abs_min_max(_A					, _A					)					)
														if detect_overflow(_A					, _A					):
																			__A     :						Tuple          =    True
									elif var is None:
														self.expand_frame(F"""{"None":>17} {ctx}"""					)
									else:
														self.expand_frame(F"""{"not a tensor":>17} {ctx}"""					)



				def  UpperCAmelCase_		(      self					):
									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  UpperCAmelCase_		(      self					):
									self.expand_frame(F"""{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n"""					)



				def  UpperCAmelCase_		(      self					, _A					, _A					, _A					):
									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  UpperCAmelCase_		(      self					):
									self.model.apply(self._register_forward_hook					)



				def  UpperCAmelCase_		(      self					, _A					):
									module.register_forward_hook(self.forward_hook					)





				def  UpperCAmelCase_		(      self					, _A					, _A					, _A					):
									# - input is a tuple of packed inputs (could be non-Tensors)
									# - output could be a Tensor or a tuple of Tensors and non-Tensors

									__A     :						int          =    False

									__A     :						int          =    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
														__A     :						Optional[Any]          =    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 			_SCREAMING_SNAKE_CASE    (       a ,						a  )				->				str:
					__A     :						Optional[Any]          =    var.abs()
					return F"""{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"""







def 			_SCREAMING_SNAKE_CASE    (       a ,						a  )				->				List[Any]:
					__A     :						Optional[Any]          =    False
					if torch.isnan(a  ).any().item():
										__A     :						List[str]          =    True
										print(F"""{ctx} has nans"""  )
					if torch.isinf(a  ).any().item():
										__A     :						List[str]          =    True
										print(F"""{ctx} has infs"""  )

					# if needed to monitor large elements can enable the following
					if 0:  # and detected:
										__A     :						Dict          =    var[torch.ge(var.abs() ,						1_00  )]
										if naaa.numel() > 0:
															print(F"""{ctx}:  n100={naaa.numel()}"""  )
										__A     :						List[Any]          =    var[torch.ge(var.abs() ,						10_00  )]
										if naaaa.numel() > 0:
															print(F"""{ctx}: n1000={naaaa.numel()}"""  )
										__A     :						Union[str, Any]          =    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 						_A( snake_case__       ):


				"""simple docstring"""


				UpperCamelCase   :  List[Any]    =     '''underflow_overflow'''
				UpperCamelCase   :  Tuple    =     '''tpu_metrics_debug'''











from .hash_table import HashTable





class 						_A( snake_case__       ):


				"""simple docstring"""


				def __init__(      self					, *_A					, **_A					):
									super().__init__(*_A					, **_A					)





				def  UpperCAmelCase_		(      self					, _A					, _A=None					):
									__A     :						Optional[int]          =    1
									__A     :						Union[str, Any]          =    self.hash_function(key + i * i					)

									while self.values[new_key] is not None and self.values[new_key] != key:
														i += 1
														__A     :						Tuple          =    (
														    self.hash_function(key + i * i					)
														    if not self.balanced_factor() >= self.lim_charge
														    else None
														)

														if new_key is None:
																			break

									return new_key




