



import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
						import torch


A__  :  Any       = logging.get_logger(__name__)



class     __snake_case						:




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

									self.analyse_model()

									self.register_forward_hook()




							def        UpperCAmelCase__      (	self					:	Optional[Any]    ,    A_					:	List[Any]=None):




									if frame is not None:
											self.expand_frame(A_)
									self.frames.append('''\n'''.join(self.frame))
									lowerCAmelCase_      :					Dict     =			[]  # start a new frame




							def        UpperCAmelCase__      (	self					:	Optional[Any]    ,    A_					:	List[str]):
									self.frame.append(A_)




							def        UpperCAmelCase__      (	self					:	Union[str, Any]):
									print('''\n'''.join(self.frames))
									lowerCAmelCase_      :					str     =			[]




							def        UpperCAmelCase__      (	self					:	Optional[Any]):
									lowerCAmelCase_      :					Dict     =			[]




							def        UpperCAmelCase__      (	self					:	Union[str, 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_      :					int     =			[]




							def        UpperCAmelCase__      (	self					:	List[str]):
									lowerCAmelCase_      :					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					:	Optional[int]    ,    A_					:	Any    ,    A_					:	int):




									if torch.is_tensor(A_):
											self.expand_frame(get_abs_min_max(A_    ,    A_))
											if detect_overflow(A_    ,    A_):
													lowerCAmelCase_      :					Dict     =			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					:	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        UpperCAmelCase__      (	self					:	Optional[Any]):
									self.expand_frame(F"""{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n""")




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




							def        UpperCAmelCase__      (	self					:	List[str]    ,    A_					:	str):
									module.register_forward_hook(self.forward_hook)







							def        UpperCAmelCase__      (	self					:	List[str]    ,    A_					:	Optional[int]    ,    A_					:	Optional[Any]    ,    A_					:	Dict):
									lowerCAmelCase_      :					Any     =			False

									lowerCAmelCase_      :					List[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_      :					Union[str, 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       UpperCamelCase(  __UpperCamelCase							:		Optional[Any]					,__UpperCamelCase							:		Dict			):
		lowerCAmelCase_      :					int     =			var.abs()
		return f"""{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"""


def       UpperCamelCase(  __UpperCamelCase							:		Union[str, Any]					,__UpperCamelCase							:		Union[str, Any]			):
		lowerCAmelCase_      :					Tuple     =			False
		if torch.isnan(__UpperCamelCase			).any().item():
				lowerCAmelCase_      :					int     =			True
				print(f"""{ctx} has nans"""			)
		if torch.isinf(__UpperCamelCase			).any().item():
				lowerCAmelCase_      :					str     =			True
				print(f"""{ctx} has infs"""			)

		# if needed to monitor large elements can enable the following
		if 0:  # and detected:
				lowerCAmelCase_      :					Optional[Any]     =			var[torch.ge(var.abs()					,100			)]
				if naaa.numel() > 0:
						print(f"""{ctx}:  n100={naaa.numel()}"""			)
				lowerCAmelCase_      :					Optional[int]     =			var[torch.ge(var.abs()					,1000			)]
				if naaaa.numel() > 0:
						print(f"""{ctx}: n1000={naaaa.numel()}"""			)
				lowerCAmelCase_      :					Union[str, Any]     =			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						(					UpperCAmelCase__						):
							_a      				=			'underflow_overflow'
							_a      				=			'tpu_metrics_debug'




import random
import unittest

import torch

from diffusers import IFImgaImgPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin



@skip_mps
class     __snake_case						(					UpperCamelCase_						,UpperCamelCase_						,unittest.TestCase						):
							_a      				=			IFImgaImgPipeline
							_a      				=			TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''width''', '''height'''}
							_a      				=			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
							_a      				=			PipelineTesterMixin.required_optional_params - {'''latents'''}




							def        UpperCAmelCase__      (	self					:	Any):
									return self._get_dummy_components()




							def        UpperCAmelCase__      (	self					:	str    ,    A_					:	Optional[int]    ,    A_					:	List[Any]=0):
									if str(A_).startswith('''mps'''):
											lowerCAmelCase_      :					Dict     =			torch.manual_seed(A_)
									else:
											lowerCAmelCase_      :					List[Any]     =			torch.Generator(device=A_).manual_seed(A_)

									lowerCAmelCase_      :					List[Any]     =			floats_tensor((1, 3, 3_2, 3_2)    ,    rng=random.Random(A_)).to(A_)

									lowerCAmelCase_      :					Any     =			{
									    '''prompt''': '''A painting of a squirrel eating a burger''',
									    '''image''': image,
									    '''generator''': generator,
									    '''num_inference_steps''': 2,
									    '''output_type''': '''numpy''',
									}

									return inputs




							def        UpperCAmelCase__      (	self					:	Union[str, Any]):
									self._test_save_load_optional_components()




							@unittest.skipIf(
							    torch_device != '''cuda''' or not is_xformers_available()    ,    reason='''XFormers attention is only available with CUDA and `xformers` installed'''    ,    )
							def        UpperCAmelCase__      (	self					:	List[str]):
									self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3)




							@unittest.skipIf(torch_device != '''cuda'''    ,    reason='''float16 requires CUDA''')
							def        UpperCAmelCase__      (	self					:	Dict):
									# Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
									super().test_save_load_floataa(expected_max_diff=1e-1)




							@unittest.skipIf(torch_device != '''cuda'''    ,    reason='''float16 requires CUDA''')
							def        UpperCAmelCase__      (	self					:	Tuple):
									super().test_floataa_inference(expected_max_diff=1e-1)




							def        UpperCAmelCase__      (	self					:	str):
									self._test_attention_slicing_forward_pass(expected_max_diff=1e-2)




							def        UpperCAmelCase__      (	self					:	int):
									self._test_save_load_local()







							def        UpperCAmelCase__      (	self					:	Union[str, Any]):
									self._test_inference_batch_single_identical(
									    expected_max_diff=1e-2    ,    )





