
'''simple docstring'''



import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, 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 PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





class      A							(    __UpperCAmelCase						,  __UpperCAmelCase						,  unittest.TestCase			):
			lowerCamelCase							:							int    						=					StableDiffusionXLPipeline
			lowerCamelCase							:							List[Any]    						=					TEXT_TO_IMAGE_PARAMS
			lowerCamelCase							:							str    						=					TEXT_TO_IMAGE_BATCH_PARAMS
			lowerCamelCase							:							str    						=					TEXT_TO_IMAGE_IMAGE_PARAMS
			lowerCamelCase							:							List[Any]    						=					TEXT_TO_IMAGE_IMAGE_PARAMS



			def A__	(   self    )     ->						str:




									'''simple docstring'''



									torch.manual_seed(0    )
									lowercase__								=    UNetaDConditionModel(
									    block_out_channels=(32, 64)    ,       layers_per_block=2    ,       sample_size=32    ,       in_channels=4    ,       out_channels=4    ,       down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")    ,       up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")    ,       attention_head_dim=(2, 4)    ,       use_linear_projection=lowerCamelCase__    ,       addition_embed_type="""text_time"""    ,       addition_time_embed_dim=8    ,       transformer_layers_per_block=(1, 2)    ,       projection_class_embeddings_input_dim=80    ,       cross_attention_dim=64    ,       )
									lowercase__								=    EulerDiscreteScheduler(
									    beta_start=0.0_00_85    ,       beta_end=0.0_12    ,       steps_offset=1    ,       beta_schedule="""scaled_linear"""    ,       timestep_spacing="""leading"""    ,       )
									torch.manual_seed(0    )
									lowercase__								=    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    )
									lowercase__								=    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=1_000    ,       hidden_act="""gelu"""    ,       projection_dim=32    ,       )
									lowercase__								=    CLIPTextModel(lowerCamelCase__    )
									lowercase__								=    CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    ,       local_files_only=lowerCamelCase__    )

									lowercase__								=    CLIPTextModelWithProjection(lowerCamelCase__    )
									lowercase__								=    CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    ,       local_files_only=lowerCamelCase__    )

									lowercase__								=    {
									    """unet""": unet,
									    """scheduler""": scheduler,
									    """vae""": vae,
									    """text_encoder""": text_encoder,
									    """tokenizer""": tokenizer,
									    """text_encoder_2""": text_encoder_a,
									    """tokenizer_2""": tokenizer_a,
									    # "safety_checker": None,
									    # "feature_extractor": None,
									}
									return components



			def A__	(   self    ,       lowerCamelCase__    ,       lowerCamelCase__=0    )     ->						Union[str, Any]:




									'''simple docstring'''



									if str(lowerCamelCase__    ).startswith("""mps"""    ):
															lowercase__								=    torch.manual_seed(lowerCamelCase__    )
									else:
															lowercase__								=    torch.Generator(device=lowerCamelCase__    ).manual_seed(lowerCamelCase__    )
									lowercase__								=    {
									    """prompt""": """A painting of a squirrel eating a burger""",
									    """generator""": generator,
									    """num_inference_steps""": 2,
									    """guidance_scale""": 5.0,
									    """output_type""": """numpy""",
									}
									return inputs



			def A__	(   self    )     ->						List[str]:




									'''simple docstring'''



									lowercase__								=    """cpu"""  # ensure determinism for the device-dependent torch.Generator
									lowercase__								=    self.get_dummy_components()
									lowercase__								=    StableDiffusionXLPipeline(**lowerCamelCase__    )
									lowercase__								=    sd_pipe.to(lowerCamelCase__    )
									sd_pipe.set_progress_bar_config(disable=lowerCamelCase__    )

									lowercase__								=    self.get_dummy_inputs(lowerCamelCase__    )
									lowercase__								=    sd_pipe(**lowerCamelCase__    ).images
									lowercase__								=    image[0, -3:, -3:, -1]

									assert image.shape == (1, 64, 64, 3)
									lowercase__								=    np.array([0.58_73, 0.61_28, 0.47_97, 0.51_22, 0.56_74, 0.46_39, 0.52_27, 0.51_49, 0.47_47]    )

									assert np.abs(image_slice.flatten() - expected_slice    ).max() < 1e-2



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




									'''simple docstring'''



									lowercase__								=    self.get_dummy_components()
									lowercase__								=    StableDiffusionXLPipeline(**lowerCamelCase__    )
									lowercase__								=    sd_pipe.to(lowerCamelCase__    )
									lowercase__								=    sd_pipe.to(lowerCamelCase__    )
									sd_pipe.set_progress_bar_config(disable=lowerCamelCase__    )

									# forward without prompt embeds
									lowercase__								=    self.get_dummy_inputs(lowerCamelCase__    )
									lowercase__								=    3 * ["""this is a negative prompt"""]
									lowercase__								=    negative_prompt
									lowercase__								=    3 * [inputs["""prompt"""]]

									lowercase__								=    sd_pipe(**lowerCamelCase__    )
									lowercase__								=    output.images[0, -3:, -3:, -1]

									# forward with prompt embeds
									lowercase__								=    self.get_dummy_inputs(lowerCamelCase__    )
									lowercase__								=    3 * ["""this is a negative prompt"""]
									lowercase__								=    3 * [inputs.pop("""prompt"""    )]

									(
									    (
									    lowercase__
									)	,  (
									    lowercase__
									)	,  (
									    lowercase__
									)	,  (
									    lowercase__
									)	,  
									)								=    sd_pipe.encode_prompt(lowerCamelCase__    ,       negative_prompt=lowerCamelCase__    )

									lowercase__								=    sd_pipe(
									    **lowerCamelCase__    ,       prompt_embeds=lowerCamelCase__    ,       negative_prompt_embeds=lowerCamelCase__    ,       pooled_prompt_embeds=lowerCamelCase__    ,       negative_pooled_prompt_embeds=lowerCamelCase__    ,       )
									lowercase__								=    output.images[0, -3:, -3:, -1]

									# make sure that it's equal
									assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()    ).max() < 1e-4



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




									'''simple docstring'''



									super().test_attention_slicing_forward_pass(expected_max_diff=3e-3    )



			def A__	(   self    )     ->						str:




									'''simple docstring'''



									super().test_inference_batch_single_identical(expected_max_diff=3e-3    )





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



			def A__	(   self    )     ->						Union[str, Any]:




									'''simple docstring'''



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



			def A__	(   self    ,       lowerCamelCase__    ,       lowerCamelCase__="cpu"    ,       lowerCamelCase__=torch.floataa    ,       lowerCamelCase__=0    )     ->						Any:




									'''simple docstring'''



									lowercase__								=    torch.Generator(device=lowerCamelCase__    ).manual_seed(lowerCamelCase__    )
									lowercase__								=    np.random.RandomState(lowerCamelCase__    ).standard_normal((1, 4, 64, 64)    )
									lowercase__								=    torch.from_numpy(lowerCamelCase__    ).to(device=lowerCamelCase__    ,       dtype=lowerCamelCase__    )
									lowercase__								=    {
									    """prompt""": """a photograph of an astronaut riding a horse""",
									    """latents""": latents,
									    """generator""": generator,
									    """num_inference_steps""": 3,
									    """guidance_scale""": 7.5,
									    """output_type""": """numpy""",
									}
									return inputs



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




									'''simple docstring'''



									lowercase__								=    DiffusionPipeline.from_pretrained("""stabilityai/stable-diffusion-2-base"""    )
									pipe.to(lowerCamelCase__    )
									pipe.set_progress_bar_config(disable=lowerCamelCase__    )

									lowercase__								=    self.get_inputs(lowerCamelCase__    )
									lowercase__								=    pipe(**lowerCamelCase__    ).images
									lowercase__								=    image[0, -3:, -3:, -1].flatten()

									assert image.shape == (1, 512, 512, 3)
									lowercase__								=    np.array([0.4_94_93, 0.4_78_96, 0.4_07_98, 0.5_42_14, 0.5_32_12, 0.4_82_02, 0.4_76_56, 0.4_63_29, 0.4_85_06]    )
									assert np.abs(image_slice - expected_slice    ).max() < 7e-3








'''simple docstring'''



import json
import os
import shutil
import tempfile
import unittest

from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor, WavaVecaProcessor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.utils import FEATURE_EXTRACTOR_NAME

from .test_feature_extraction_wavaveca import floats_list





class      A							(    unittest.TestCase			):



			def A__	(   self    )     ->						Union[str, Any]:




									'''simple docstring'''



									lowercase__								=    """<pad> <s> </s> <unk> | E T A O N I H S R D L U M W C F G Y P B V K ' X J Q Z""".split(""" """    )
									lowercase__								=    dict(zip(lowerCamelCase__    ,       range(len(lowerCamelCase__    )    )    )    )

									lowercase__								=    {
									    """pad_token""": """<pad>""",
									    """unk_token""": """<unk>""",
									    """bos_token""": """<s>""",
									    """eos_token""": """</s>""",
									}
									lowercase__								=    {
									    """feature_size""": 1,
									    """padding_value""": 0.0,
									    """sampling_rate""": 16_000,
									    """return_attention_mask""": False,
									    """do_normalize""": True,
									}

									lowercase__								=    tempfile.mkdtemp()
									lowercase__								=    os.path.join(self.tmpdirname    ,       VOCAB_FILES_NAMES["""vocab_file"""]    )
									lowercase__								=    os.path.join(self.tmpdirname    ,       lowerCamelCase__    )
									with open(self.vocab_file    ,       """w"""    ,       encoding="""utf-8"""    ) as fp:
															fp.write(json.dumps(lowerCamelCase__    ) + """\n"""    )

									with open(self.feature_extraction_file    ,       """w"""    ,       encoding="""utf-8"""    ) as fp:
															fp.write(json.dumps(lowerCamelCase__    ) + """\n"""    )



			def A__	(   self    ,       **lowerCamelCase__    )     ->						int:




									'''simple docstring'''



									lowercase__								=    self.add_kwargs_tokens_map.copy()
									kwargs.update(lowerCamelCase__    )
									return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname    ,       **lowerCamelCase__    )



			def A__	(   self    ,       **lowerCamelCase__    )     ->						Dict:




									'''simple docstring'''



									return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname    ,       **lowerCamelCase__    )



			def A__	(   self    )     ->						Dict:




									'''simple docstring'''



									shutil.rmtree(self.tmpdirname    )



			def A__	(   self    )     ->						str:




									'''simple docstring'''



									lowercase__								=    self.get_tokenizer()
									lowercase__								=    self.get_feature_extractor()

									lowercase__								=    WavaVecaProcessor(tokenizer=lowerCamelCase__    ,       feature_extractor=lowerCamelCase__    )

									processor.save_pretrained(self.tmpdirname    )
									lowercase__								=    WavaVecaProcessor.from_pretrained(self.tmpdirname    )

									self.assertEqual(processor.tokenizer.get_vocab()    ,       tokenizer.get_vocab()    )
									self.assertIsInstance(processor.tokenizer    ,       lowerCamelCase__    )

									self.assertEqual(processor.feature_extractor.to_json_string()    ,       feature_extractor.to_json_string()    )
									self.assertIsInstance(processor.feature_extractor    ,       lowerCamelCase__    )



			def A__	(   self    )     ->						Tuple:




									'''simple docstring'''



									lowercase__								=    WavaVecaProcessor(tokenizer=self.get_tokenizer()    ,       feature_extractor=self.get_feature_extractor()    )
									processor.save_pretrained(self.tmpdirname    )

									lowercase__								=    self.get_tokenizer(bos_token="""(BOS)"""    ,       eos_token="""(EOS)"""    )
									lowercase__								=    self.get_feature_extractor(do_normalize=lowerCamelCase__    ,       padding_value=1.0    )

									lowercase__								=    WavaVecaProcessor.from_pretrained(
									    self.tmpdirname    ,       bos_token="""(BOS)"""    ,       eos_token="""(EOS)"""    ,       do_normalize=lowerCamelCase__    ,       padding_value=1.0    )

									self.assertEqual(processor.tokenizer.get_vocab()    ,       tokenizer_add_kwargs.get_vocab()    )
									self.assertIsInstance(processor.tokenizer    ,       lowerCamelCase__    )

									self.assertEqual(processor.feature_extractor.to_json_string()    ,       feature_extractor_add_kwargs.to_json_string()    )
									self.assertIsInstance(processor.feature_extractor    ,       lowerCamelCase__    )



			def A__	(   self    )     ->						List[str]:




									'''simple docstring'''



									lowercase__								=    self.get_feature_extractor()
									lowercase__								=    self.get_tokenizer()

									lowercase__								=    WavaVecaProcessor(tokenizer=lowerCamelCase__    ,       feature_extractor=lowerCamelCase__    )

									lowercase__								=    floats_list((3, 1_000)    )

									lowercase__								=    feature_extractor(lowerCamelCase__    ,       return_tensors="""np"""    )
									lowercase__								=    processor(lowerCamelCase__    ,       return_tensors="""np"""    )

									for key in input_feat_extract.keys():
															self.assertAlmostEqual(input_feat_extract[key].sum()    ,       input_processor[key].sum()    ,       delta=1e-2    )



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




									'''simple docstring'''



									lowercase__								=    self.get_feature_extractor()
									lowercase__								=    self.get_tokenizer()

									lowercase__								=    WavaVecaProcessor(tokenizer=lowerCamelCase__    ,       feature_extractor=lowerCamelCase__    )

									lowercase__								=    """This is a test string"""

									lowercase__								=    processor(text=lowerCamelCase__    )

									lowercase__								=    tokenizer(lowerCamelCase__    )

									for key in encoded_tok.keys():
															self.assertListEqual(encoded_tok[key]    ,       encoded_processor[key]    )



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




									'''simple docstring'''



									lowercase__								=    self.get_feature_extractor()
									lowercase__								=    self.get_tokenizer()

									lowercase__								=    WavaVecaProcessor(tokenizer=lowerCamelCase__    ,       feature_extractor=lowerCamelCase__    )

									lowercase__								=    [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

									lowercase__								=    processor.batch_decode(lowerCamelCase__    )
									lowercase__								=    tokenizer.batch_decode(lowerCamelCase__    )

									self.assertListEqual(lowerCamelCase__    ,       lowerCamelCase__    )



			def A__	(   self    )     ->						List[str]:




									'''simple docstring'''



									lowercase__								=    self.get_feature_extractor()
									lowercase__								=    self.get_tokenizer()

									lowercase__								=    WavaVecaProcessor(tokenizer=lowerCamelCase__    ,       feature_extractor=lowerCamelCase__    )

									self.assertListEqual(
									    processor.model_input_names    ,       feature_extractor.model_input_names    ,       msg="""`processor` and `feature_extractor` model input names do not match"""    ,       )






