

"""simple docstring"""


import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
			import torch


__A		     =    logging.get_logger(__name__)





class 					snake_case			:

				def __init__(			self							:       str  ,						UpperCamelCase__							:       Union[str, Any]  ,						UpperCamelCase__							:       int=2_1  ,						UpperCamelCase__							:       Any=[]  ,						UpperCamelCase__							:       Any=None)->     Any:



						'''simple docstring'''


						__lowerCAmelCase: Dict					   =    model
						__lowerCAmelCase: Dict					   =    trace_batch_nums
						__lowerCAmelCase: 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
						__lowerCAmelCase: Any					   =    collections.deque([]  ,						UpperCamelCase__)
						__lowerCAmelCase: Optional[int]					   =    []
						__lowerCAmelCase: List[Any]					   =    0
						__lowerCAmelCase: List[Any]					   =    0
						__lowerCAmelCase: str					   =    False
						__lowerCAmelCase: str					   =    "                 "

						self.analyse_model()

						self.register_forward_hook()

				def 							lowercase_  (			self							:       int  ,						UpperCamelCase__							:       Any=None)->     List[Any]:



						'''simple docstring'''


						if frame is not None:
								self.expand_frame(UpperCamelCase__)
						self.frames.append("\n".join(self.frame))
						__lowerCAmelCase: int					   =    []  # start a new frame

				def 							lowercase_  (			self							:       Union[str, Any]  ,						UpperCamelCase__							:       int)->     List[Any]:



						'''simple docstring'''


						self.frame.append(UpperCamelCase__)

				def 							lowercase_  (			self							:       Any)->     Any:



						'''simple docstring'''


						print("\n".join(self.frames))
						__lowerCAmelCase: Union[str, Any]					   =    []

				def 							lowercase_  (			self							:       Optional[Any])->     Optional[Any]:



						'''simple docstring'''


						__lowerCAmelCase: Dict					   =    []

				def 							lowercase_  (			self							:       List[str])->     int:



						'''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")
						__lowerCAmelCase: Union[str, Any]					   =    []

				def 							lowercase_  (			self							:       List[str])->     Any:



						'''simple docstring'''


						__lowerCAmelCase: List[Any]					   =    {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 							lowercase_  (			self							:       Optional[int]  ,						UpperCamelCase__							:       str  ,						UpperCamelCase__							:       Optional[int])->     Tuple:



						'''simple docstring'''


						if torch.is_tensor(UpperCamelCase__):
								self.expand_frame(get_abs_min_max(UpperCamelCase__  ,						UpperCamelCase__))
								if detect_overflow(UpperCamelCase__  ,						UpperCamelCase__):
										__lowerCAmelCase: List[Any]					   =    True
						elif var is None:
								self.expand_frame(f"{'None':>17} {ctx}")
						else:
								self.expand_frame(f"{'not a tensor':>17} {ctx}")

				def 							lowercase_  (			self							:       List[str])->     int:



						'''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 							lowercase_  (			self							:       Tuple)->     List[Any]:



						'''simple docstring'''


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

				def 							lowercase_  (			self							:       Any  ,						UpperCamelCase__							:       Optional[Any]  ,						UpperCamelCase__							:       List[Any]  ,						UpperCamelCase__							:       List[Any])->     str:



						'''simple docstring'''


						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 							lowercase_  (			self							:       Optional[Any])->     Union[str, Any]:



						'''simple docstring'''


						self.model.apply(self._register_forward_hook)

				def 							lowercase_  (			self							:       str  ,						UpperCamelCase__							:       str)->     Any:



						'''simple docstring'''


						module.register_forward_hook(self.forward_hook)




				def 							lowercase_  (			self							:       int  ,						UpperCamelCase__							:       int  ,						UpperCamelCase__							:       Optional[Any]  ,						UpperCamelCase__							:       List[Any])->     List[Any]:



						'''simple docstring'''


						__lowerCAmelCase: int					   =    False

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






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

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





"""simple docstring"""


from __future__ import annotations






def 					a__					(					__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE				)			->			tuple[float, float]:
		assert (
		    c % greatest_common_divisor(__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE				) == 0
		)  # greatest_common_divisor(a,b) function implemented below
		((__lowerCAmelCase)   ,				(__lowerCAmelCase)   ,				(__lowerCAmelCase)): Dict					   =    extended_gcd(__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE				)  # extended_gcd(a,b) function implemented below
		__lowerCAmelCase: Optional[int]					   =    c / d
		return (r * x, r * y)






def 					a__					(					__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE = 2				)			->			None:
		((__lowerCAmelCase)   ,				(__lowerCAmelCase)): Optional[Any]					   =    diophantine(__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE				)  # Initial value
		__lowerCAmelCase: Optional[Any]					   =    greatest_common_divisor(__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE				)
		__lowerCAmelCase: Any					   =    a // d
		__lowerCAmelCase: str					   =    b // d

		for i in range(__SCREAMING_SNAKE_CASE				):
				__lowerCAmelCase: Dict					   =    xa + i * q
				__lowerCAmelCase: List[str]					   =    ya - i * p
				print(__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE				)






def 					a__					(					__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE				)			->			int:
		if a < b:
				__lowerCAmelCase   ,				__lowerCAmelCase: Any					   =    b, a

		while a % b != 0:
				__lowerCAmelCase   ,				__lowerCAmelCase: List[Any]					   =    b, a % b

		return b






def 					a__					(					__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE				)			->			tuple[int, int, int]:
		assert a >= 0 and b >= 0

		if b == 0:
				__lowerCAmelCase   ,				__lowerCAmelCase   ,				__lowerCAmelCase: List[str]					   =    a, 1, 0
		else:
				((__lowerCAmelCase)   ,				(__lowerCAmelCase)   ,				(__lowerCAmelCase)): Dict					   =    extended_gcd(__SCREAMING_SNAKE_CASE      ,				a % b				)
				__lowerCAmelCase: Dict					   =    q
				__lowerCAmelCase: int					   =    p - q * (a // b)

		assert a % d == 0 and b % d == 0
		assert d == a * x + b * y

		return (d, x, y)


if __name__ == "__main__":
			from doctest import testmod

			testmod(name="diophantine", verbose=True)
			testmod(name="diophantine_all_soln", verbose=True)
			testmod(name="extended_gcd", verbose=True)
			testmod(name="greatest_common_divisor", verbose=True)


