




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        UpperCAmelCase__				(							UpperCAmelCase__      ,		UpperCAmelCase__      ,		unittest.TestCase				):



     '''simple docstring'''
     _SCREAMING_SNAKE_CASE      :					Tuple     					=    StableDiffusionXLPipeline
     _SCREAMING_SNAKE_CASE      :					int     					=    TEXT_TO_IMAGE_PARAMS
     _SCREAMING_SNAKE_CASE      :					Dict     					=    TEXT_TO_IMAGE_BATCH_PARAMS
     _SCREAMING_SNAKE_CASE      :					List[str]     					=    TEXT_TO_IMAGE_IMAGE_PARAMS
     _SCREAMING_SNAKE_CASE      :					Optional[Any]     					=    TEXT_TO_IMAGE_IMAGE_PARAMS
     def        lowerCAmelCase__  (				self	):
            torch.manual_seed(0	)
            a			   =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 ,    )
            a			   =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	)
            a			   =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	)
            a			   =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 ,    )
            a			   =CLIPTextModel(_lowerCAmelCase	)
            a			   =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" ,    local_files_only=_lowerCAmelCase	)

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

            a			   ={
                """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        lowerCAmelCase__  (				self ,    _lowerCAmelCase ,    _lowerCAmelCase=0	):
            if str(_lowerCAmelCase	).startswith("""mps"""	):
                   a			   =torch.manual_seed(_lowerCAmelCase	)
            else:
                   a			   =torch.Generator(device=_lowerCAmelCase	).manual_seed(_lowerCAmelCase	)
            a			   ={
                """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        lowerCAmelCase__  (				self	):
            a			   ="""cpu"""  # ensure determinism for the device-dependent torch.Generator
            a			   =self.get_dummy_components()
            a			   =StableDiffusionXLPipeline(**_lowerCAmelCase	)
            a			   =sd_pipe.to(_lowerCAmelCase	)
            sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase	)

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

            assert image.shape == (1, 64, 64, 3)
            a			   =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        lowerCAmelCase__  (				self	):
            a			   =self.get_dummy_components()
            a			   =StableDiffusionXLPipeline(**_lowerCAmelCase	)
            a			   =sd_pipe.to(_lowerCAmelCase	)
            a			   =sd_pipe.to(_lowerCAmelCase	)
            sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase	)

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

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

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

            (
                (
                a
            )   ,      (
                a
            )   ,      (
                a
            )   ,      (
                a
            )   ,      
            )			   =sd_pipe.encode_prompt(_lowerCAmelCase ,    negative_prompt=_lowerCAmelCase	)

            a			   =sd_pipe(
                **_lowerCAmelCase ,    prompt_embeds=_lowerCAmelCase ,    negative_prompt_embeds=_lowerCAmelCase ,    pooled_prompt_embeds=_lowerCAmelCase ,    negative_pooled_prompt_embeds=_lowerCAmelCase ,    )
            a			   =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        lowerCAmelCase__  (				self	):
            super().test_attention_slicing_forward_pass(expected_max_diff=3E-3	)






     def        lowerCAmelCase__  (				self	):
            super().test_inference_batch_single_identical(expected_max_diff=3E-3	)



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



     '''simple docstring'''
     def        lowerCAmelCase__  (				self	):
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()
     def        lowerCAmelCase__  (				self ,    _lowerCAmelCase ,    _lowerCAmelCase="cpu" ,    _lowerCAmelCase=torch.floataa ,    _lowerCAmelCase=0	):
            a			   =torch.Generator(device=_lowerCAmelCase	).manual_seed(_lowerCAmelCase	)
            a			   =np.random.RandomState(_lowerCAmelCase	).standard_normal((1, 4, 64, 64)	)
            a			   =torch.from_numpy(_lowerCAmelCase	).to(device=_lowerCAmelCase ,    dtype=_lowerCAmelCase	)
            a			   ={
                """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        lowerCAmelCase__  (				self	):
            a			   =DiffusionPipeline.from_pretrained("""stabilityai/stable-diffusion-2-base"""	)
            pipe.to(_lowerCAmelCase	)
            pipe.set_progress_bar_config(disable=_lowerCAmelCase	)

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

            assert image.shape == (1, 512, 512, 3)
            a			   =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






import functools
import operator

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_lowerCamelCase							=      logging.get_logger(__name__)

_lowerCamelCase							=      {
    '''facebook/hubert-base-ls960''': '''https://huggingface.co/facebook/hubert-base-ls960/resolve/main/config.json''',
    # See all Hubert models at https://huggingface.co/models?filter=hubert
}


class        UpperCAmelCase__				(							UpperCAmelCase__				):



     '''simple docstring'''
     _SCREAMING_SNAKE_CASE      :					Dict     					=    "hubert"
     def __init__(				self ,    _lowerCAmelCase=32 ,    _lowerCAmelCase=768 ,    _lowerCAmelCase=12 ,    _lowerCAmelCase=12 ,    _lowerCAmelCase=3_072 ,    _lowerCAmelCase="gelu" ,    _lowerCAmelCase=0.1 ,    _lowerCAmelCase=0.1 ,    _lowerCAmelCase=0.1 ,    _lowerCAmelCase=True ,    _lowerCAmelCase=0.0 ,    _lowerCAmelCase=0.1 ,    _lowerCAmelCase=0.1 ,    _lowerCAmelCase=0.02 ,    _lowerCAmelCase=1E-5 ,    _lowerCAmelCase="group" ,    _lowerCAmelCase="gelu" ,    _lowerCAmelCase=(512, 512, 512, 512, 512, 512, 512) ,    _lowerCAmelCase=(5, 2, 2, 2, 2, 2, 2) ,    _lowerCAmelCase=(10, 3, 3, 3, 3, 2, 2) ,    _lowerCAmelCase=False ,    _lowerCAmelCase=128 ,    _lowerCAmelCase=16 ,    _lowerCAmelCase=False ,    _lowerCAmelCase=True ,    _lowerCAmelCase=0.05 ,    _lowerCAmelCase=10 ,    _lowerCAmelCase=2 ,    _lowerCAmelCase=0.0 ,    _lowerCAmelCase=10 ,    _lowerCAmelCase=0 ,    _lowerCAmelCase="sum" ,    _lowerCAmelCase=False ,    _lowerCAmelCase=False ,    _lowerCAmelCase=256 ,    _lowerCAmelCase=0 ,    _lowerCAmelCase=1 ,    _lowerCAmelCase=2 ,    **_lowerCAmelCase ,    ):
            super().__init__(**_lowerCAmelCase ,    pad_token_id=_lowerCAmelCase ,    bos_token_id=_lowerCAmelCase ,    eos_token_id=_lowerCAmelCase	)
            a			   =hidden_size
            a			   =feat_extract_norm
            a			   =feat_extract_activation
            a			   =list(_lowerCAmelCase	)
            a			   =list(_lowerCAmelCase	)
            a			   =list(_lowerCAmelCase	)
            a			   =conv_bias
            a			   =num_conv_pos_embeddings
            a			   =num_conv_pos_embedding_groups
            a			   =len(self.conv_dim	)
            a			   =num_hidden_layers
            a			   =intermediate_size
            a			   =hidden_act
            a			   =num_attention_heads
            a			   =hidden_dropout
            a			   =attention_dropout
            a			   =activation_dropout
            a			   =feat_proj_layer_norm
            a			   =feat_proj_dropout
            a			   =final_dropout
            a			   =layerdrop
            a			   =layer_norm_eps
            a			   =initializer_range
            a			   =vocab_size
            a			   =do_stable_layer_norm
            a			   =use_weighted_layer_sum
            a			   =classifier_proj_size

            if (
                (len(self.conv_stride	) != self.num_feat_extract_layers)
                or (len(self.conv_kernel	) != self.num_feat_extract_layers)
                or (len(self.conv_dim	) != self.num_feat_extract_layers)
            ):
                   raise ValueError(
                       """Configuration for convolutional layers is incorrect. It is required that `len(config.conv_dim)` =="""
                       """ `len(config.conv_stride)` == `len(config.conv_kernel)`, but is `len(config.conv_dim) ="""
                       F''' {len(self.conv_dim	)}`, `len(config.conv_stride) = {len(self.conv_stride	)}`,'''
                       F''' `len(config.conv_kernel) = {len(self.conv_kernel	)}`.'''	)

            # fine-tuning config parameters for SpecAugment: https://arxiv.org/abs/1904.08779
            a			   =apply_spec_augment
            a			   =mask_time_prob
            a			   =mask_time_length
            a			   =mask_time_min_masks
            a			   =mask_feature_prob
            a			   =mask_feature_length
            a			   =mask_feature_min_masks

            # ctc loss
            a			   =ctc_loss_reduction
            a			   =ctc_zero_infinity






     @property
     def        lowerCAmelCase__  (				self	):
            return functools.reduce(operator.mul ,    self.conv_stride ,    1	)
