






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


a_				=     False


@skip_mps
class   lowercase__  (      A__,       A__,       A__,       unittest.TestCase						):
	a_       =StableDiffusionAttendAndExcitePipeline
	a_       =False
	a_       =TEXT_TO_IMAGE_PARAMS
	a_       =TEXT_TO_IMAGE_BATCH_PARAMS.union({"""token_indices"""}						)
	a_       =TEXT_TO_IMAGE_IMAGE_PARAMS
	a_       =TEXT_TO_IMAGE_IMAGE_PARAMS

	@classmethod
	def 		UpperCAmelCase	(				cls							)->       List[str]:







							'''simple docstring'''





							super().setUpClass()
							torch.use_deterministic_algorithms(_lowerCamelCase							)

	@classmethod
	def 		UpperCAmelCase	(				cls							)->       Tuple:







							'''simple docstring'''





							super().tearDownClass()
							torch.use_deterministic_algorithms(_lowerCamelCase							)

	def 		UpperCAmelCase	(				self							)->       Optional[int]:







							'''simple docstring'''





							torch.manual_seed(0							)
							lowerCAmelCase__			   =    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		,     )
							lowerCAmelCase__			   =    DDIMScheduler(
							    beta_start=0.00_085		,     beta_end=0.012		,     beta_schedule="scaled_linear"		,     clip_sample=_lowerCamelCase		,     set_alpha_to_one=_lowerCamelCase		,     )
							torch.manual_seed(0							)
							lowerCAmelCase__			   =    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							)
							lowerCAmelCase__			   =    CLIPTextConfig(
							    bos_token_id=0		,     eos_token_id=2		,     hidden_size=32		,     intermediate_size=37		,     layer_norm_eps=1E-0_5		,     num_attention_heads=4		,     num_hidden_layers=5		,     pad_token_id=1		,     vocab_size=1000		,     hidden_act="gelu"		,     projection_dim=512		,     )
							lowerCAmelCase__			   =    CLIPTextModel(_lowerCamelCase							)
							lowerCAmelCase__			   =    CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"							)

							lowerCAmelCase__			   =    {
							    'unet': unet,
							    'scheduler': scheduler,
							    'vae': vae,
							    'text_encoder': text_encoder,
							    'tokenizer': tokenizer,
							    'safety_checker': None,
							    'feature_extractor': None,
							}

							return components

	def 		UpperCAmelCase	(				self		,     __UpperCAmelCase		,     __UpperCAmelCase=0							)->       Any:







							'''simple docstring'''





							if str(_lowerCamelCase							).startswith("mps"							):
													lowerCAmelCase__			   =    torch.manual_seed(_lowerCamelCase							)
							else:
													lowerCAmelCase__			   =    torch.Generator(device=_lowerCamelCase							).manual_seed(_lowerCamelCase							)
							lowerCAmelCase__			   =    {
							    '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							)->       Dict:







							'''simple docstring'''





							lowerCAmelCase__			   =    'cpu'

							lowerCAmelCase__			   =    self.get_dummy_components()
							lowerCAmelCase__			   =    self.pipeline_class(**_lowerCamelCase							)
							pipe.to(_lowerCamelCase							)
							pipe.set_progress_bar_config(disable=_lowerCamelCase							)

							lowerCAmelCase__			   =    self.get_dummy_inputs(_lowerCamelCase							)
							lowerCAmelCase__			   =    pipe(**_lowerCamelCase							).images
							lowerCAmelCase__			   =    image[0, -3:, -3:, -1]

							self.assertEqual(image.shape		,     (1, 64, 64, 3)							)
							lowerCAmelCase__			   =    np.array(
							    [0.63_905_364, 0.62_897_307, 0.48_599_017, 0.5_133_624, 0.5_550_048, 0.45_769_516, 0.50_326_973, 0.5_023_139, 0.45_384_496]							)
							lowerCAmelCase__			   =    np.abs(image_slice.flatten() - expected_slice							).max()
							self.assertLessEqual(_lowerCamelCase		,     1E-3							)

	def 		UpperCAmelCase	(				self							)->       int:







							'''simple docstring'''





							super().test_cpu_offload_forward_pass(expected_max_diff=5E-4							)

	def 		UpperCAmelCase	(				self							)->       List[Any]:







							'''simple docstring'''





							self._test_inference_batch_consistent(batch_sizes=[1, 2]							)

	def 		UpperCAmelCase	(				self							)->       Tuple:







							'''simple docstring'''





							self._test_inference_batch_single_identical(batch_size=2		,     expected_max_diff=7E-4							)

	def 		UpperCAmelCase	(				self							)->       Any:







							'''simple docstring'''





							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3							)

	def 		UpperCAmelCase	(				self							)->       Tuple:







							'''simple docstring'''





							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5E-4							)

	def 		UpperCAmelCase	(				self							)->       Tuple:







							'''simple docstring'''





							super().test_save_load_local(expected_max_difference=5E-4							)


	def 		UpperCAmelCase	(				self							)->       Dict:







							'''simple docstring'''





							super().test_save_load_optional_components(expected_max_difference=4E-4							)





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

	@classmethod
	def 		UpperCAmelCase	(				cls							)->       Any:







							'''simple docstring'''





							super().setUpClass()
							torch.use_deterministic_algorithms(_lowerCamelCase							)

	@classmethod
	def 		UpperCAmelCase	(				cls							)->       int:







							'''simple docstring'''





							super().tearDownClass()
							torch.use_deterministic_algorithms(_lowerCamelCase							)

	def 		UpperCAmelCase	(				self							)->       Tuple:







							'''simple docstring'''





							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()


	def 		UpperCAmelCase	(				self							)->       Dict:







							'''simple docstring'''





							lowerCAmelCase__			   =    torch.manual_seed(51							)

							lowerCAmelCase__			   =    StableDiffusionAttendAndExcitePipeline.from_pretrained(
							    "CompVis/stable-diffusion-v1-4"		,     safety_checker=_lowerCamelCase		,     torch_dtype=torch.floataa							)
							pipe.to("cuda"							)

							lowerCAmelCase__			   =    'a painting of an elephant with glasses'
							lowerCAmelCase__			   =    [5, 7]

							lowerCAmelCase__			   =    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]

							lowerCAmelCase__			   =    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 os
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers.models.realm.configuration_realm import RealmConfig
from transformers.models.realm.retrieval_realm import _REALM_BLOCK_RECORDS_FILENAME, RealmRetriever
from transformers.models.realm.tokenization_realm import VOCAB_FILES_NAMES, RealmTokenizer





class 				a__					(						A__       ):



     def  lowerCamelCase_						(							self :List[str]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Optional[Any]   	=tempfile.mkdtemp()
      UpperCamelCase_     :					str   	=5

      # Realm tok
      UpperCamelCase_     :					List[Any]   	=[
          '[UNK]',
          '[CLS]',
          '[SEP]',
          '[PAD]',
          '[MASK]',
          'test',
          'question',
          'this',
          'is',
          'the',
          'first',
          'second',
          'third',
          'fourth',
          'fifth',
          'record',
          'want',
          '##want',
          '##ed',
          'wa',
          'un',
          'runn',
          '##ing',
          ',',
          'low',
          'lowest',
      ]
      UpperCamelCase_     :					List[str]   	=os.path.join(self.tmpdirname      ,  'realm_tokenizer'   )
      os.makedirs(_lowerCamelCase      ,  exist_ok=_lowerCamelCase   )
      UpperCamelCase_     :					int   	=os.path.join(_lowerCamelCase      ,  VOCAB_FILES_NAMES['vocab_file']   )
      with open(self.vocab_file      ,  'w'      ,  encoding='utf-8'   ) as vocab_writer:
       vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]   )   )

      UpperCamelCase_     :					str   	=os.path.join(self.tmpdirname      ,  'realm_block_records'   )
      os.makedirs(_lowerCamelCase      ,  exist_ok=_lowerCamelCase   )



     def  lowerCamelCase_						(							self :Union[str, Any]   ):







      '''simple docstring'''



      return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname      ,  'realm_tokenizer'   )   )



     def  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



      shutil.rmtree(self.tmpdirname   )



     def  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



      UpperCamelCase_     :					List[str]   	=RealmConfig(num_block_records=self.num_block_records   )
      return config



     def  lowerCamelCase_						(							self :Optional[int]   ):







      '''simple docstring'''



      UpperCamelCase_     :					str   	=Dataset.from_dict(
          {
              'id': ['0', '1'],
              'question': ['foo', 'bar'],
              'answers': [['Foo', 'Bar'], ['Bar']],
          }   )
      return dataset



     def  lowerCamelCase_						(							self :Optional[Any]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Union[str, Any]   	=np.array(
          [
              b'This is the first record',
              b'This is the second record',
              b'This is the third record',
              b'This is the fourth record',
              b'This is the fifth record',
              b'This is a longer longer longer record',
          ]      ,  dtype=_lowerCamelCase      ,  )
      return block_records



     def  lowerCamelCase_						(							self :Any   ):







      '''simple docstring'''



      UpperCamelCase_     :					List[Any]   	=RealmRetriever(
          block_records=self.get_dummy_block_records()      ,  tokenizer=self.get_tokenizer()      ,  )
      return retriever



     def  lowerCamelCase_						(							self :int   ):







      '''simple docstring'''



      UpperCamelCase_     :					Any   	=self.get_config()
      UpperCamelCase_     :					Tuple   	=self.get_dummy_retriever()
      UpperCamelCase_     :					Tuple   	=retriever.tokenizer

      UpperCamelCase_     :					List[str]   	=np.array([0, 3]      ,  dtype='long'   )
      UpperCamelCase_     :					Optional[Any]   	=tokenizer(['Test question']   ).input_ids
      UpperCamelCase_     :					Union[str, Any]   	=tokenizer(
          ['the fourth']      ,  add_special_tokens=_lowerCamelCase      ,  return_token_type_ids=_lowerCamelCase      ,  return_attention_mask=_lowerCamelCase      ,  ).input_ids
      UpperCamelCase_     :					Union[str, Any]   	=config.reader_seq_len

      UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_     :					Any   	=retriever(
          _lowerCamelCase      ,  _lowerCamelCase      ,  answer_ids=_lowerCamelCase      ,  max_length=_lowerCamelCase      ,  return_tensors='np'   )

      self.assertEqual(len(_lowerCamelCase   )      ,  2   )
      self.assertEqual(len(_lowerCamelCase   )      ,  2   )
      self.assertEqual(len(_lowerCamelCase   )      ,  2   )
      self.assertEqual(concat_inputs.input_ids.shape      ,  (2, 10)   )
      self.assertEqual(concat_inputs.attention_mask.shape      ,  (2, 10)   )
      self.assertEqual(concat_inputs.token_type_ids.shape      ,  (2, 10)   )
      self.assertEqual(concat_inputs.special_tokens_mask.shape      ,  (2, 10)   )
      self.assertEqual(
          tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[0]   )      ,  ['[CLS]', 'test', 'question', '[SEP]', 'this', 'is', 'the', 'first', 'record', '[SEP]']      ,  )
      self.assertEqual(
          tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[1]   )      ,  ['[CLS]', 'test', 'question', '[SEP]', 'this', 'is', 'the', 'fourth', 'record', '[SEP]']      ,  )



     def  lowerCamelCase_						(							self :Optional[int]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Optional[int]   	=self.get_config()
      UpperCamelCase_     :					int   	=self.get_dummy_retriever()
      UpperCamelCase_     :					Any   	=retriever.tokenizer

      UpperCamelCase_     :					Optional[int]   	=np.array([0, 3, 5]      ,  dtype='long'   )
      UpperCamelCase_     :					Optional[int]   	=tokenizer(['Test question']   ).input_ids
      UpperCamelCase_     :					int   	=tokenizer(
          ['the fourth', 'longer longer']      ,  add_special_tokens=_lowerCamelCase      ,  return_token_type_ids=_lowerCamelCase      ,  return_attention_mask=_lowerCamelCase      ,  ).input_ids
      UpperCamelCase_     :					Union[str, Any]   	=config.reader_seq_len

      UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_     :					Union[str, Any]   	=retriever(
          _lowerCamelCase      ,  _lowerCamelCase      ,  answer_ids=_lowerCamelCase      ,  max_length=_lowerCamelCase      ,  return_tensors='np'   )

      self.assertEqual([False, True, True]      ,  _lowerCamelCase   )
      self.assertEqual([[-1, -1, -1], [6, -1, -1], [6, 7, 8]]      ,  _lowerCamelCase   )
      self.assertEqual([[-1, -1, -1], [7, -1, -1], [7, 8, 9]]      ,  _lowerCamelCase   )




     def  lowerCamelCase_						(							self :List[Any]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Tuple   	=self.get_dummy_retriever()
      retriever.save_pretrained(os.path.join(self.tmpdirname      ,  'realm_block_records'   )   )

      # Test local path
      UpperCamelCase_     :					int   	=retriever.from_pretrained(os.path.join(self.tmpdirname      ,  'realm_block_records'   )   )
      self.assertEqual(retriever.block_records[0]      ,  b'This is the first record'   )

      # Test mocked remote path
      with patch('transformers.models.realm.retrieval_realm.hf_hub_download'   ) as mock_hf_hub_download:
       UpperCamelCase_     :					Dict   	=os.path.join(
           os.path.join(self.tmpdirname      ,  'realm_block_records'   )      ,  _REALM_BLOCK_RECORDS_FILENAME   )
       UpperCamelCase_     :					int   	=RealmRetriever.from_pretrained('google/realm-cc-news-pretrained-openqa'   )

      self.assertEqual(retriever.block_records[0]      ,  b'This is the first record'   )
