






"""simple docstring"""


import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
		import torch


__SCREAMING_SNAKE_CASE      =      logging.get_logger(__name__)




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

											self.analyse_model()

											self.register_forward_hook()
					def 					UpperCAmelCase__   (       self       :  Optional[Any]    ,   UpperCAmelCase       :  List[Any]=None      )  ->					Any:
											if frame is not None:
																	self.expand_frame(UpperCAmelCase      )
											self.frames.append('\n'.join(self.frame      )      )
											lowerCAmelCase      :str				  = []  # start a new frame
					def 					UpperCAmelCase__   (       self       :  Dict    ,   UpperCAmelCase       :  List[str]      )  ->					Union[str, Any]:
											self.frame.append(UpperCAmelCase      )
					def 					UpperCAmelCase__   (       self       :  List[str]      )  ->					Tuple:
											print('\n'.join(self.frames      )      )
											lowerCAmelCase      :int				  = []
					def 					UpperCAmelCase__   (       self       :  int      )  ->					Union[str, Any]:
											lowerCAmelCase      :List[Any]				  = []
					def 					UpperCAmelCase__   (       self       :  int      )  ->					Any:
											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      :Optional[Any]				  = []
					def 					UpperCAmelCase__   (       self       :  Tuple      )  ->					str:
											# 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      :str				  = {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       :  Optional[int]    ,   UpperCAmelCase       :  List[str]    ,   UpperCAmelCase       :  Any      )  ->					Optional[int]:
											if torch.is_tensor(UpperCAmelCase      ):
																	self.expand_frame(get_abs_min_max(UpperCAmelCase    ,   UpperCAmelCase      )      )
																	if detect_overflow(UpperCAmelCase    ,   UpperCAmelCase      ):
																							lowerCAmelCase      :str				  = 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       :  Tuple      )  ->					Optional[int]:
											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       :  Optional[Any]      )  ->					int:
											self.expand_frame(f"""{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n"""      )
					def 					UpperCAmelCase__   (       self       :  Dict    ,   UpperCAmelCase       :  List[str]    ,   UpperCAmelCase       :  Optional[int]    ,   UpperCAmelCase       :  int      )  ->					Tuple:
											self.expand_frame(f"""{self.prefix} {self.module_names[module]} {module.__class__.__name__}"""      )

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

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

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

											self.save_frame()
					def 					UpperCAmelCase__   (       self       :  Dict      )  ->					int:
											self.model.apply(self._register_forward_hook      )
					def 					UpperCAmelCase__   (       self       :  str    ,   UpperCAmelCase       :  Tuple      )  ->					List[str]:
											module.register_forward_hook(self.forward_hook      )
					def 					UpperCAmelCase__   (       self       :  Optional[Any]    ,   UpperCAmelCase       :  Optional[int]    ,   UpperCAmelCase       :  Union[str, Any]    ,   UpperCAmelCase       :  Dict      )  ->					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

											lowerCAmelCase      :List[str]				  = False

											lowerCAmelCase      :Any				  = 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      :Dict				  = True

											self.create_frame(UpperCAmelCase    ,   UpperCAmelCase    ,   UpperCAmelCase      )

											# 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__    ,			a__   ):





						'''simple docstring'''






						lowerCAmelCase      :Dict				  = var.abs()
						return F"""{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"""


def 					UpperCAmelCase			(				a__    ,			a__   ):





						'''simple docstring'''






						lowerCAmelCase      :Any				  = False
						if torch.isnan(a__   ).any().item():
												lowerCAmelCase      :List[Any]				  = True
												print(F"""{ctx} has nans"""   )
						if torch.isinf(a__   ).any().item():
												lowerCAmelCase      :Optional[Any]				  = True
												print(F"""{ctx} has infs"""   )

						# if needed to monitor large elements can enable the following
						if 0:  # and detected:
												lowerCAmelCase      :Optional[int]				  = var[torch.ge(var.abs()    ,			1_00   )]
												if naaa.numel() > 0:
																		print(F"""{ctx}:  n100={naaa.numel()}"""   )
												lowerCAmelCase      :List[str]				  = var[torch.ge(var.abs()    ,			10_00   )]
												if naaaa.numel() > 0:
																		print(F"""{ctx}: n1000={naaaa.numel()}"""   )
												lowerCAmelCase      :Optional[int]				  = 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 	__UpperCamelCase    (    UpperCamelCase	):
					lowercase_						:							Optional[Any]			=     """underflow_overflow"""
					lowercase_						:							Optional[int]			=     """tpu_metrics_debug"""












"""simple docstring"""


from .number_theory.prime_numbers import next_prime




class 	__UpperCamelCase    :
					def __init__(       self       :  List[str]    ,   UpperCAmelCase       :  int    ,   UpperCAmelCase       :  int | None = None    ,   UpperCAmelCase       :  float | None = None    ,   )  ->					None:
											lowerCAmelCase      :Tuple				  = size_table
											lowerCAmelCase      :List[Any]				  = [None] * self.size_table
											lowerCAmelCase      :Tuple				  = 0.7_5 if lim_charge is None else lim_charge
											lowerCAmelCase      :str				  = 1 if charge_factor is None else charge_factor
											lowerCAmelCase      :list				  = []
											lowerCAmelCase      :dict				  = {}
					def 					UpperCAmelCase__   (       self       :  Optional[int]      )  ->					Union[str, Any]:
											return self._keys
					def 					UpperCAmelCase__   (       self       :  Dict      )  ->					Optional[Any]:
											return sum(1 for slot in self.values if slot is not None      ) / (
											    self.size_table * self.charge_factor
											)
					def 					UpperCAmelCase__   (       self       :  Any    ,   UpperCAmelCase       :  List[str]      )  ->					int:
											return key % self.size_table
					def 					UpperCAmelCase__   (       self       :  Dict    ,   UpperCAmelCase       :  Union[str, Any]      )  ->					int:
											print(f"""step {step_ord}"""      )
											print(list(range(len(self.values      )      )      )      )
											print(self.values      )
					def 					UpperCAmelCase__   (       self       :  Any    ,   UpperCAmelCase       :  Dict      )  ->					Any:
											lowerCAmelCase      :List[str]				  = 1
											lowerCAmelCase      :Union[str, Any]				  = values
											for value in values:
																	self.insert_data(UpperCAmelCase      )
																	self._step_by_step(UpperCAmelCase      )
																	i += 1
					def 					UpperCAmelCase__   (       self       :  str    ,   UpperCAmelCase       :  Dict    ,   UpperCAmelCase       :  Union[str, Any]      )  ->					Optional[Any]:
											lowerCAmelCase      :List[Any]				  = data
											lowerCAmelCase      :str				  = data
					def 					UpperCAmelCase__   (       self       :  Optional[int]    ,   UpperCAmelCase       :  str    ,   UpperCAmelCase       :  str=None      )  ->					Tuple:
											lowerCAmelCase      :Optional[Any]				  = self.hash_function(key + 1      )

											while self.values[new_key] is not None and self.values[new_key] != key:
																	if self.values.count(UpperCAmelCase      ) > 0:
																							lowerCAmelCase      :Tuple				  = self.hash_function(new_key + 1      )
																	else:
																							lowerCAmelCase      :Union[str, Any]				  = None
																							break

											return new_key
					def 					UpperCAmelCase__   (       self       :  Any      )  ->					Any:
											lowerCAmelCase      :Optional[Any]				  = [value for value in self.values if value is not None]
											lowerCAmelCase      :List[str]				  = next_prime(self.size_table    ,   factor=2      )
											self._keys.clear()
											lowerCAmelCase      :int				  = [None] * self.size_table  # hell's pointers D: don't DRY ;/
											for value in survivor_values:
																	self.insert_data(UpperCAmelCase      )
					def 					UpperCAmelCase__   (       self       :  Optional[int]    ,   UpperCAmelCase       :  Tuple      )  ->					int:
											lowerCAmelCase      :Union[str, Any]				  = self.hash_function(UpperCAmelCase      )

											if self.values[key] is None:
																	self._set_value(UpperCAmelCase    ,   UpperCAmelCase      )

											elif self.values[key] == data:
																	pass

											else:
																	lowerCAmelCase      :Tuple				  = self._collision_resolution(UpperCAmelCase    ,   UpperCAmelCase      )
																	if collision_resolution is not None:
																							self._set_value(UpperCAmelCase    ,   UpperCAmelCase      )
																	else:
																							self.rehashing()
																							self.insert_data(UpperCAmelCase      )