

import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
					import torch


lowerCAmelCase_ 			=			logging.get_logger(__name__)





class 			snake_case_     :







			'''simple docstring'''

			def __init__(  self				:       Dict							,     _UpperCamelCase				:       List[str]							,     _UpperCamelCase				:       Tuple=2_1							,     _UpperCamelCase				:       Dict=[]							,     _UpperCamelCase				:       Union[str, Any]=None							)   ->List[Any]:
									snake_case_     =			model
									snake_case_     =			trace_batch_nums
									snake_case_     =			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_     =			collections.deque([]							,     _UpperCamelCase							)
									snake_case_     =			[]
									snake_case_     =			0
									snake_case_     =			0
									snake_case_     =			False
									snake_case_     =			'''                 '''

									self.analyse_model()

									self.register_forward_hook()







			def        snake_case__(  self				:       str							,     _UpperCamelCase				:       Any=None							)   ->Dict:
									if frame is not None:
															self.expand_frame(_UpperCamelCase							)
									self.frames.append('''\n'''.join(self.frame							)							)
									snake_case_     =			[]  # start a new frame







			def        snake_case__(  self				:       Optional[int]							,     _UpperCamelCase				:       List[Any]							)   ->Union[str, Any]:
									self.frame.append(_UpperCamelCase							)







			def        snake_case__(  self				:       int							)   ->str:
									print('''\n'''.join(self.frames							)							)
									snake_case_     =			[]







			def        snake_case__(  self				:       List[Any]							)   ->Dict:
									snake_case_     =			[]







			def        snake_case__(  self				:       Union[str, Any]							)   ->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'''							)
									snake_case_     =			[]







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







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







			def        snake_case__(  self				:       Optional[int]							,     _UpperCamelCase				:       str							)   ->List[Any]:
									module.register_forward_hook(self.forward_hook							)







			def        snake_case__(  self				:       int							,     _UpperCamelCase				:       Any							,     _UpperCamelCase				:       Any							,     _UpperCamelCase				:       Any							)   ->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

									snake_case_     =			False

									snake_case_     =			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_     =			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 						__SCREAMING_SNAKE_CASE      (SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__ ):
						snake_case_     =			var.abs()
						return F'''{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}'''


def 						__SCREAMING_SNAKE_CASE      (SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__ ):
						snake_case_     =			False
						if torch.isnan(SCREAMING_SNAKE_CASE__ ).any().item():
												snake_case_     =			True
												print(F'''{ctx} has nans''' )
						if torch.isinf(SCREAMING_SNAKE_CASE__ ).any().item():
												snake_case_     =			True
												print(F'''{ctx} has infs''' )

						# if needed to monitor large elements can enable the following
						if 0:  # and detected:
												snake_case_     =			var[torch.ge(var.abs()						,	100 )]
												if naaa.numel() > 0:
																		print(F'''{ctx}:  n100={naaa.numel()}''' )
												snake_case_     =			var[torch.ge(var.abs()						,	1000 )]
												if naaaa.numel() > 0:
																		print(F'''{ctx}: n1000={naaaa.numel()}''' )
												snake_case_     =			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 			snake_case_     (	__A      ):







			'''simple docstring'''

			SCREAMING_SNAKE_CASE  :     List[str]		    =       "underflow_overflow"
			SCREAMING_SNAKE_CASE  :     Optional[int]		    =       "tpu_metrics_debug"


def 						__SCREAMING_SNAKE_CASE      (SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__ ):
						return (input_a, input_a).count(0 ) % 2


def 						__SCREAMING_SNAKE_CASE      ():
						assert xor_gate(0						,	0 ) == 0
						assert xor_gate(0						,	1 ) == 1
						assert xor_gate(1						,	0 ) == 1
						assert xor_gate(1						,	1 ) == 0


if __name__ == "__main__":
					print(xor_gate(0, 0))
					print(xor_gate(0, 1))