


'''simple docstring'''





import gc
import random
import unittest

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

import diffusers
from diffusers import (
    AutoencoderKL,
    EulerDiscreteScheduler,
    StableDiffusionLatentUpscalePipeline,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.schedulers import KarrasDiffusionSchedulers
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


def        a__     (					UpperCamelCase_						:      Tuple						):
   UpperCAmelCase__	:List[Any]      =      [tensor.shape for tensor in tensor_list]
   return all(shape == shapes[0] for shape in shapes[1:]						)




class 				UpperCAmelCase			(			_snake_case						, _snake_case						, _snake_case						, unittest.TestCase						):
 UpperCAmelCase				=					StableDiffusionLatentUpscalePipeline
 UpperCAmelCase				=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
     "height",
     "width",
     "cross_attention_kwargs",
     "negative_prompt_embeds",
     "prompt_embeds",
 }
 UpperCAmelCase				=					PipelineTesterMixin.required_optional_params - {"num_images_per_prompt"}
 UpperCAmelCase				=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 UpperCAmelCase				=					frozenset(
     []						)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 UpperCAmelCase				=					frozenset([]						)

 UpperCAmelCase				=					True


 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	str ):
    UpperCAmelCase__	:List[str]      =      1
    UpperCAmelCase__	:str      =      4
    UpperCAmelCase__	:Dict      =      (1_6, 1_6)

    UpperCAmelCase__	:int      =      floats_tensor((batch_size, num_channels) + sizes			,       rng=random.Random(0 ) ).to(__lowerCamelCase )
    return image


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any] ):
    torch.manual_seed(0 )
    UpperCAmelCase__	:Union[str, Any]      =      UNetaDConditionModel(
        act_fn='''gelu'''			,       attention_head_dim=8			,       norm_num_groups=__lowerCamelCase			,       block_out_channels=[3_2, 3_2, 6_4, 6_4]			,       time_cond_proj_dim=1_6_0			,       conv_in_kernel=1			,       conv_out_kernel=1			,       cross_attention_dim=3_2			,       down_block_types=(
            '''KDownBlock2D''',
            '''KCrossAttnDownBlock2D''',
            '''KCrossAttnDownBlock2D''',
            '''KCrossAttnDownBlock2D''',
        )			,       in_channels=8			,       mid_block_type=__lowerCamelCase			,       only_cross_attention=__lowerCamelCase			,       out_channels=5			,       resnet_time_scale_shift='''scale_shift'''			,       time_embedding_type='''fourier'''			,       timestep_post_act='''gelu'''			,       up_block_types=('''KCrossAttnUpBlock2D''', '''KCrossAttnUpBlock2D''', '''KCrossAttnUpBlock2D''', '''KUpBlock2D''')			,       )
    UpperCAmelCase__	:List[Any]      =      AutoencoderKL(
        block_out_channels=[3_2, 3_2, 6_4, 6_4]			,       in_channels=3			,       out_channels=3			,       down_block_types=[
            '''DownEncoderBlock2D''',
            '''DownEncoderBlock2D''',
            '''DownEncoderBlock2D''',
            '''DownEncoderBlock2D''',
        ]			,       up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''', '''UpDecoderBlock2D''', '''UpDecoderBlock2D''']			,       latent_channels=4			,       )
    UpperCAmelCase__	:Dict      =      EulerDiscreteScheduler(prediction_type='''sample''' )
    UpperCAmelCase__	:Optional[Any]      =      CLIPTextConfig(
        bos_token_id=0			,       eos_token_id=2			,       hidden_size=3_2			,       intermediate_size=3_7			,       layer_norm_eps=1e-05			,       num_attention_heads=4			,       num_hidden_layers=5			,       pad_token_id=1			,       vocab_size=1_0_0_0			,       hidden_act='''quick_gelu'''			,       projection_dim=5_1_2			,       )
    UpperCAmelCase__	:Optional[int]      =      CLIPTextModel(__lowerCamelCase )
    UpperCAmelCase__	:Any      =      CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip''' )

    UpperCAmelCase__	:Any      =      {
        '''unet''': model.eval(),
        '''vae''': vae.eval(),
        '''scheduler''': scheduler,
        '''text_encoder''': text_encoder,
        '''tokenizer''': tokenizer,
    }

    return components


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[str]			,       __lowerCamelCase							:	Optional[Any]			,       __lowerCamelCase							:	str=0 ):
    if str(__lowerCamelCase ).startswith('''mps''' ):
       UpperCAmelCase__	:Union[str, Any]      =      torch.manual_seed(__lowerCamelCase )
    else:
       UpperCAmelCase__	:Optional[int]      =      torch.Generator(device=__lowerCamelCase ).manual_seed(__lowerCamelCase )
    UpperCAmelCase__	:int      =      {
        '''prompt''': '''A painting of a squirrel eating a burger''',
        '''image''': self.dummy_image.cpu(),
        '''generator''': generator,
        '''num_inference_steps''': 2,
        '''output_type''': '''numpy''',
    }
    return inputs


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any] ):
    UpperCAmelCase__	:List[str]      =      '''cpu'''

    UpperCAmelCase__	:Union[str, Any]      =      self.get_dummy_components()
    UpperCAmelCase__	:List[Any]      =      self.pipeline_class(**__lowerCamelCase )
    pipe.to(__lowerCamelCase )
    pipe.set_progress_bar_config(disable=__lowerCamelCase )

    UpperCAmelCase__	:int      =      self.get_dummy_inputs(__lowerCamelCase )
    UpperCAmelCase__	:List[str]      =      pipe(**__lowerCamelCase ).images
    UpperCAmelCase__	:Dict      =      image[0, -3:, -3:, -1]

    self.assertEqual(image.shape			,       (1, 2_5_6, 2_5_6, 3) )
    UpperCAmelCase__	:Optional[int]      =      np.array(
        [0.47_22_24_12, 0.41_92_16_33, 0.44_71_74_34, 0.46_87_41_92, 0.42_58_82_58, 0.46_15_07_26, 0.4_67_75_34, 0.45_58_38_32, 0.48_57_90_55] )
    UpperCAmelCase__	:Union[str, Any]      =      np.abs(image_slice.flatten() - expected_slice ).max()
    self.assertLessEqual(__lowerCamelCase			,       1e-3 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Tuple ):
    super().test_attention_slicing_forward_pass(expected_max_diff=7e-3 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[str] ):
    super().test_cpu_offload_forward_pass(expected_max_diff=3e-3 )


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


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any] ):
    super().test_inference_batch_single_identical(expected_max_diff=7e-3 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any] ):
    super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3e-3 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    super().test_save_load_local(expected_max_difference=3e-3 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any] ):
    super().test_save_load_optional_components(expected_max_difference=3e-3 )





 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    UpperCAmelCase__	:Dict      =      [
        '''DDIMScheduler''',
        '''DDPMScheduler''',
        '''PNDMScheduler''',
        '''HeunDiscreteScheduler''',
        '''EulerAncestralDiscreteScheduler''',
        '''KDPM2DiscreteScheduler''',
        '''KDPM2AncestralDiscreteScheduler''',
        '''DPMSolverSDEScheduler''',
    ]
    UpperCAmelCase__	:Any      =      self.get_dummy_components()
    UpperCAmelCase__	:Dict      =      self.pipeline_class(**__lowerCamelCase )

    # make sure that PNDM does not need warm-up
    pipe.scheduler.register_to_config(skip_prk_steps=__lowerCamelCase )

    pipe.to(__lowerCamelCase )
    pipe.set_progress_bar_config(disable=__lowerCamelCase )
    UpperCAmelCase__	:Dict      =      self.get_dummy_inputs(__lowerCamelCase )
    UpperCAmelCase__	:Tuple      =      2

    UpperCAmelCase__	:int      =      []
    for scheduler_enum in KarrasDiffusionSchedulers:
       if scheduler_enum.name in skip_schedulers:
          # no sigma schedulers are not supported
          # no schedulers
          continue

       UpperCAmelCase__	:List[Any]      =      getattr(__lowerCamelCase			,       scheduler_enum.name )
       UpperCAmelCase__	:str      =      scheduler_cls.from_config(pipe.scheduler.config )
       UpperCAmelCase__	:Union[str, Any]      =      pipe(**__lowerCamelCase )[0]
       outputs.append(__lowerCamelCase )

    assert check_same_shape(__lowerCamelCase )




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


 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[int] ):
    UpperCAmelCase__	:int      =      torch.manual_seed(3_3 )

    UpperCAmelCase__	:Optional[Any]      =      StableDiffusionPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''			,       torch_dtype=torch.floataa )
    pipe.to('''cuda''' )

    UpperCAmelCase__	:Any      =      StableDiffusionLatentUpscalePipeline.from_pretrained(
        '''stabilityai/sd-x2-latent-upscaler'''			,       torch_dtype=torch.floataa )
    upscaler.to('''cuda''' )

    UpperCAmelCase__	:Tuple      =      '''a photo of an astronaut high resolution, unreal engine, ultra realistic'''

    UpperCAmelCase__	:Dict      =      pipe(__lowerCamelCase			,       generator=__lowerCamelCase			,       output_type='''latent''' ).images

    UpperCAmelCase__	:Tuple      =      upscaler(
        prompt=__lowerCamelCase			,       image=__lowerCamelCase			,       num_inference_steps=2_0			,       guidance_scale=0			,       generator=__lowerCamelCase			,       output_type='''np'''			,       ).images[0]

    UpperCAmelCase__	:List[str]      =      load_numpy(
        '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/astronaut_1024.npy''' )
    assert np.abs((expected_image - image).mean() ) < 5e-2





 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any] ):
    UpperCAmelCase__	:int      =      torch.manual_seed(3_3 )

    UpperCAmelCase__	:Optional[Any]      =      StableDiffusionLatentUpscalePipeline.from_pretrained(
        '''stabilityai/sd-x2-latent-upscaler'''			,       torch_dtype=torch.floataa )
    upscaler.to('''cuda''' )

    UpperCAmelCase__	:Dict      =      '''the temple of fire by Ross Tran and Gerardo Dottori, oil on canvas'''

    UpperCAmelCase__	:Union[str, Any]      =      load_image(
        '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png''' )

    UpperCAmelCase__	:Any      =      upscaler(
        prompt=__lowerCamelCase			,       image=__lowerCamelCase			,       num_inference_steps=2_0			,       guidance_scale=0			,       generator=__lowerCamelCase			,       output_type='''np'''			,       ).images[0]

    UpperCAmelCase__	:str      =      load_numpy(
        '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_1024.npy''' )
    assert np.abs((expected_image - image).max() ) < 5e-2







'''simple docstring'''





from .glue import GlueDataset, GlueDataTrainingArguments
from .language_modeling import (
    LineByLineTextDataset,
    LineByLineWithRefDataset,
    LineByLineWithSOPTextDataset,
    TextDataset,
    TextDatasetForNextSentencePrediction,
)
from .squad import SquadDataset, SquadDataTrainingArguments



