






import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


SCREAMING_SNAKE_CASE__		:							int 				=				False







@skip_mps
class 				UpperCAmelCase_  (		__lowerCamelCase							,      __lowerCamelCase							,      __lowerCamelCase							,      unittest.TestCase		):
     __lowerCamelCase      				=  StableDiffusionAttendAndExcitePipeline
     __lowerCamelCase      				=  False
     __lowerCamelCase      				=  TEXT_TO_IMAGE_PARAMS
     __lowerCamelCase      				=  TEXT_TO_IMAGE_BATCH_PARAMS.union({'token_indices'}		)
     __lowerCamelCase      				=  TEXT_TO_IMAGE_IMAGE_PARAMS
     __lowerCamelCase      				=  TEXT_TO_IMAGE_IMAGE_PARAMS




     @classmethod
     def   __UpperCAmelCase  (				cls		):
          super().setUpClass()
          torch.use_deterministic_algorithms(_lowerCAmelCase		)




     @classmethod
     def   __UpperCAmelCase  (				cls		):
          super().tearDownClass()
          torch.use_deterministic_algorithms(_lowerCAmelCase		)




     def   __UpperCAmelCase  (				self		):
          torch.manual_seed(0		)
          UpperCAmelCase__      :  Optional[int]				      =			UNetaDConditionModel(
              block_out_channels=(32, 64)	,						layers_per_block=1	,						sample_size=32	,						in_channels=4	,						out_channels=4	,						down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")	,						up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")	,						cross_attention_dim=32	,						attention_head_dim=(2, 4)	,						use_linear_projection=_lowerCAmelCase	,						)
          UpperCAmelCase__      :  Optional[int]				      =			DDIMScheduler(
              beta_start=0.0_0_0_8_5	,						beta_end=0.0_1_2	,						beta_schedule="""scaled_linear"""	,						clip_sample=_lowerCAmelCase	,						set_alpha_to_one=_lowerCAmelCase	,						)
          torch.manual_seed(0		)
          UpperCAmelCase__      :  List[Any]				      =			AutoencoderKL(
              block_out_channels=[32, 64]	,						in_channels=3	,						out_channels=3	,						down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]	,						up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]	,						latent_channels=4	,						sample_size=128	,						)
          torch.manual_seed(0		)
          UpperCAmelCase__      :  str				      =			CLIPTextConfig(
              bos_token_id=0	,						eos_token_id=2	,						hidden_size=32	,						intermediate_size=37	,						layer_norm_eps=1e-05	,						num_attention_heads=4	,						num_hidden_layers=5	,						pad_token_id=1	,						vocab_size=1000	,						hidden_act="""gelu"""	,						projection_dim=512	,						)
          UpperCAmelCase__      :  Optional[int]				      =			CLIPTextModel(_lowerCAmelCase		)
          UpperCAmelCase__      :  Tuple				      =			CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""		)

          UpperCAmelCase__      :  Union[str, Any]				      =			{
              """unet""": unet,
              """scheduler""": scheduler,
              """vae""": vae,
              """text_encoder""": text_encoder,
              """tokenizer""": tokenizer,
              """safety_checker""": None,
              """feature_extractor""": None,
          }

          return components




     def   __UpperCAmelCase  (				self	,						_lowerCAmelCase	,						_lowerCAmelCase=0		):
          if str(_lowerCAmelCase		).startswith("""mps"""		):
               UpperCAmelCase__      :  Any				      =			torch.manual_seed(_lowerCAmelCase		)
          else:
               UpperCAmelCase__      :  Any				      =			torch.Generator(device=_lowerCAmelCase		).manual_seed(_lowerCAmelCase		)
          UpperCAmelCase__      :  List[str]				      =			{
              """prompt""": """a cat and a frog""",
              """token_indices""": [2, 5],
              """generator""": generator,
              """num_inference_steps""": 1,
              """guidance_scale""": 6.0,
              """output_type""": """numpy""",
              """max_iter_to_alter""": 2,
              """thresholds""": {0: 0.7},
          }
          return inputs




     def   __UpperCAmelCase  (				self		):
          UpperCAmelCase__      :  Any				      =			"""cpu"""

          UpperCAmelCase__      :  Optional[int]				      =			self.get_dummy_components()
          UpperCAmelCase__      :  List[str]				      =			self.pipeline_class(**_lowerCAmelCase		)
          pipe.to(_lowerCAmelCase		)
          pipe.set_progress_bar_config(disable=_lowerCAmelCase		)

          UpperCAmelCase__      :  int				      =			self.get_dummy_inputs(_lowerCAmelCase		)
          UpperCAmelCase__      :  Tuple				      =			pipe(**_lowerCAmelCase		).images
          UpperCAmelCase__      :  List[Any]				      =			image[0, -3:, -3:, -1]

          self.assertEqual(image.shape	,						(1, 64, 64, 3)		)
          UpperCAmelCase__      :  str				      =			np.array(
              [0.6_3_9_0_5_3_6_4, 0.6_2_8_9_7_3_0_7, 0.4_8_5_9_9_0_1_7, 0.5_1_3_3_6_2_4, 0.5_5_5_0_0_4_8, 0.4_5_7_6_9_5_1_6, 0.5_0_3_2_6_9_7_3, 0.5_0_2_3_1_3_9, 0.4_5_3_8_4_4_9_6]		)
          UpperCAmelCase__      :  Optional[Any]				      =			np.abs(image_slice.flatten() - expected_slice		).max()
          self.assertLessEqual(_lowerCAmelCase	,						1e-3		)




     def   __UpperCAmelCase  (				self		):
          super().test_cpu_offload_forward_pass(expected_max_diff=5e-4		)




     def   __UpperCAmelCase  (				self		):
          # NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
          self._test_inference_batch_consistent(batch_sizes=[1, 2]		)




     def   __UpperCAmelCase  (				self		):
          self._test_inference_batch_single_identical(batch_size=2	,						expected_max_diff=7e-4		)




     def   __UpperCAmelCase  (				self		):
          super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3		)




     def   __UpperCAmelCase  (				self		):
          super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4		)




     def   __UpperCAmelCase  (				self		):
          super().test_save_load_local(expected_max_difference=5e-4		)




     def   __UpperCAmelCase  (				self		):
          super().test_save_load_optional_components(expected_max_difference=4e-4		)







@require_torch_gpu
@slow
class 				UpperCAmelCase_  (		unittest.TestCase		):




     @classmethod
     def   __UpperCAmelCase  (				cls		):
          super().setUpClass()
          torch.use_deterministic_algorithms(_lowerCAmelCase		)




     @classmethod
     def   __UpperCAmelCase  (				cls		):
          super().tearDownClass()
          torch.use_deterministic_algorithms(_lowerCAmelCase		)




     def   __UpperCAmelCase  (				self		):
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()




     def   __UpperCAmelCase  (				self		):
          UpperCAmelCase__      :  List[Any]				      =			torch.manual_seed(51		)

          UpperCAmelCase__      :  List[Any]				      =			StableDiffusionAttendAndExcitePipeline.from_pretrained(
              """CompVis/stable-diffusion-v1-4"""	,						safety_checker=_lowerCAmelCase	,						torch_dtype=torch.floataa		)
          pipe.to("""cuda"""		)

          UpperCAmelCase__      :  str				      =			"""a painting of an elephant with glasses"""
          UpperCAmelCase__      :  List[str]				      =			[5, 7]

          UpperCAmelCase__      :  Tuple				      =			pipe(
              prompt=_lowerCAmelCase	,						token_indices=_lowerCAmelCase	,						guidance_scale=7.5	,						generator=_lowerCAmelCase	,						num_inference_steps=5	,						max_iter_to_alter=5	,						output_type="""numpy"""	,						).images[0]

          UpperCAmelCase__      :  List[Any]				      =			load_numpy(
              """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy"""		)
          assert np.abs((expected_image - image).max()		) < 5e-1










'''simple docstring'''
import logging
import os

import quant_trainer
import torch
from torch.utils.data import DataLoader

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput


lowerCAmelCase   : Dict             =				logging.getLogger(__name__)

if is_torch_tpu_available(check_device=False):
					import torch_xla.core.xla_model as xm
					import torch_xla.debug.metrics as met




class        _UpperCamelCase	(    SCREAMING_SNAKE_CASE):


				'''simple docstring'''


				def __init__(     self							,      *a_							,      a_=None							,      a_=None							,      a_=None							,      **a_							)    ->  Any:
					super().__init__(*a_							,      **a_							)
					lowercase						:		Optional[Any]   					=		eval_examples
					lowercase						:		Any   					=		post_process_function
					lowercase						:		Dict   					=		quant_trainer_args
					lowercase						:		List[Any]   					=		1_2_8  # default number of calibration samples

				def 	a__		(     self							,      a_=None							)    ->  List[str]:
					if calib_dataset is None and self.calib_dataset is None:
						raise ValueError("Trainer: calibration requires an calib_dataset."							)
					lowercase						:		Dict   					=		calib_dataset if calib_dataset is not None else self.calib_dataset

					lowercase						:		Optional[int]   					=		self._remove_unused_columns(a_							,      description="Calibration"							)

					return DataLoader(
					    a_							,      batch_size=self.args.eval_batch_size							,      collate_fn=self.data_collator							,      drop_last=self.args.dataloader_drop_last							,      num_workers=self.args.dataloader_num_workers							,      pin_memory=self.args.dataloader_pin_memory							,      shuffle=a_							,      )

				def 	a__		(     self							,      a_=None							)    ->  str:
					lowercase						:		Optional[Any]   					=		self.train_dataset if calib_dataset is None else calib_dataset
					lowercase						:		Optional[Any]   					=		self.get_calib_dataloader(a_							)

					lowercase						:		Union[str, Any]   					=		self.model
					quant_trainer.configure_model(a_							,      self.quant_trainer_args							,      calib=a_							)
					model.eval()
					quant_trainer.enable_calibration(a_							)

					logger.info("***** Running calibration *****"							)
					logger.info(F'''  Num examples = {self.calib_num}'''							)
					logger.info(F'''  Batch size = {calib_dataloader.batch_size}'''							)

					for step, inputs in enumerate(a_							):
						# Prediction step
						lowercase					,		lowercase					,		lowercase						:		Union[str, Any]   					=		self.prediction_step(a_							,      a_							,      prediction_loss_only=a_							)
						if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
							break

					quant_trainer.finish_calibration(a_							,      self.quant_trainer_args							)
					lowercase						:		List[Any]   					=		model

				def 	a__		(     self							,      a_=None							,      a_=None							,      a_=None							,      a_ = "eval"							)    ->  int:
					lowercase						:		List[str]   					=		self.eval_dataset if eval_dataset is None else eval_dataset
					lowercase						:		Union[str, Any]   					=		self.get_eval_dataloader(a_							)
					lowercase						:		Tuple   					=		self.eval_examples if eval_examples is None else eval_examples

					# Temporarily disable metric computation, we will do it in the loop here.
					lowercase						:		List[str]   					=		self.compute_metrics
					lowercase						:		Dict   					=		None
					lowercase						:		int   					=		self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
					try:
						lowercase						:		Tuple   					=		eval_loop(
						    a_							,      description="Evaluation"							,      prediction_loss_only=True if compute_metrics is None else None							,      ignore_keys=a_							,      )
					finally:
						lowercase						:		List[Any]   					=		compute_metrics

					if self.post_process_function is not None and self.compute_metrics is not None:
						lowercase						:		Union[str, Any]   					=		self.post_process_function(a_							,      a_							,      output.predictions							)
						lowercase						:		Union[str, Any]   					=		self.compute_metrics(a_							)

						# Prefix all keys with metric_key_prefix + '_'
						for key in list(metrics.keys()							):
							if not key.startswith(F'''{metric_key_prefix}_'''							):
								lowercase						:		Dict   					=		metrics.pop(a_							)

						self.log(a_							)
					else:
						lowercase						:		Tuple   					=		{}

					if self.args.tpu_metrics_debug or self.args.debug:
						# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
						xm.master_print(met.metrics_report()							)

					lowercase						:		int   					=		self.callback_handler.on_evaluate(self.args							,      self.state							,      self.control							,      a_							)
					return metrics

				def 	a__		(     self							,      a_							,      a_							,      a_=None							,      a_ = "test"							)    ->  Optional[int]:
					lowercase						:		Dict   					=		self.get_test_dataloader(a_							)

					# Temporarily disable metric computation, we will do it in the loop here.
					lowercase						:		List[str]   					=		self.compute_metrics
					lowercase						:		List[str]   					=		None
					lowercase						:		int   					=		self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
					try:
						lowercase						:		Optional[Any]   					=		eval_loop(
						    a_							,      description="Prediction"							,      prediction_loss_only=True if compute_metrics is None else None							,      ignore_keys=a_							,      )
					finally:
						lowercase						:		int   					=		compute_metrics

					if self.post_process_function is None or self.compute_metrics is None:
						return output

					lowercase						:		List[str]   					=		self.post_process_function(a_							,      a_							,      output.predictions							,      "predict"							)
					lowercase						:		Dict   					=		self.compute_metrics(a_							)

					# Prefix all keys with metric_key_prefix + '_'
					for key in list(metrics.keys()							):
						if not key.startswith(F'''{metric_key_prefix}_'''							):
							lowercase						:		Union[str, Any]   					=		metrics.pop(a_							)

					return PredictionOutput(predictions=predictions.predictions							,      label_ids=predictions.label_ids							,      metrics=a_							)


				def 	a__		(     self							,      a_="./"							)    ->  Optional[Any]:
					lowercase						:		str   					=		self.eval_dataset
					lowercase						:		Dict   					=		self.get_eval_dataloader(a_							)

					lowercase						:		Any   					=		next(iter(a_							)							)

					# saving device - to make it consistent
					lowercase						:		List[Any]   					=		torch.device("cuda" if torch.cuda.is_available() else "cpu"							)

					# convert to tuple
					lowercase						:		Tuple   					=		tuple(v.to(a_							) for k, v in batch.items()							)

					logger.info("Converting model to be onnx compatible"							)
					from pytorch_quantization.nn import TensorQuantizer

					lowercase						:		Optional[Any]   					=		True

					lowercase						:		int   					=		self.model.to(a_							)

					model.eval()
					model.float()

					lowercase						:		Union[str, Any]   					=		model.module if hasattr(a_							,      "module"							) else model
					quant_trainer.configure_model(a_							,      self.quant_trainer_args							)

					lowercase						:		List[str]   					=		os.path.join(a_							,      "model.onnx"							)
					logger.info(F'''exporting model to {output_model_file}'''							)

					lowercase						:		Any   					=		{0: "batch_size", 1: "seq_len"}

					torch.onnx.export(
					    a_							,      a_							,      a_							,      export_params=a_							,      opset_version=1_3							,      do_constant_folding=a_							,      input_names=["input_ids", "attention_mask", "token_type_ids"]							,      output_names=["output_start_logits", "output_end_logits"]							,      dynamic_axes={
					        "input_ids": axes,
					        "attention_mask": axes,
					        "token_type_ids": axes,
					        "output_start_logits": axes,
					        "output_end_logits": axes,
					    }							,      verbose=a_							,      )
					logger.info("onnx export finished"							)


