





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__       (    SCREAMING_SNAKE_CASE_   :      List[Any]						)  ->  Any:






   """simple docstring"""







   _UpperCAmelCase           =   [tensor.shape for tensor in tensor_list]
   return all(shape == shapes[0] for shape in shapes[1:]						)




class 			__UpperCamelCase    (							A__	, A__	, A__	, unittest.TestCase    ):
     __A	:  str							  =   StableDiffusionLatentUpscalePipeline
     __A	:  int							  =   TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
         """height""",
         """width""",
         """cross_attention_kwargs""",
         """negative_prompt_embeds""",
         """prompt_embeds""",
     }
     __A	:  Dict							  =   PipelineTesterMixin.required_optional_params - {"""num_images_per_prompt"""}
     __A	:  Dict							  =   TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
     __A	:  Any							  =   frozenset(
         []    )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
     __A	:  Optional[int]							  =   frozenset([]    )

     __A	:  str							  =   True





     @property
     def 		UpperCamelCase(     self					):
        _UpperCAmelCase           =   1
        _UpperCAmelCase           =   4
        _UpperCAmelCase           =   (16, 16)

        _UpperCAmelCase           =   floats_tensor((batch_size, num_channels) + sizes				,							rng=random.Random(0					)					).to(_UpperCamelCase					)
        return image





     def 		UpperCamelCase(     self					):
        torch.manual_seed(0					)
        _UpperCAmelCase           =   UNetaDConditionModel(
            act_fn='''gelu'''				,							attention_head_dim=8				,							norm_num_groups=_UpperCamelCase				,							block_out_channels=[32, 32, 64, 64]				,							time_cond_proj_dim=160				,							conv_in_kernel=1				,							conv_out_kernel=1				,							cross_attention_dim=32				,							down_block_types=(
                '''KDownBlock2D''',
                '''KCrossAttnDownBlock2D''',
                '''KCrossAttnDownBlock2D''',
                '''KCrossAttnDownBlock2D''',
            )				,							in_channels=8				,							mid_block_type=_UpperCamelCase				,							only_cross_attention=_UpperCamelCase				,							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           =   AutoencoderKL(
            block_out_channels=[32, 32, 64, 64]				,							in_channels=3				,							out_channels=3				,							down_block_types=[
                '''DownEncoderBlock2D''',
                '''DownEncoderBlock2D''',
                '''DownEncoderBlock2D''',
                '''DownEncoderBlock2D''',
            ]				,							up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''', '''UpDecoderBlock2D''', '''UpDecoderBlock2D''']				,							latent_channels=4				,							)
        _UpperCAmelCase           =   EulerDiscreteScheduler(prediction_type='''sample'''					)
        _UpperCAmelCase           =   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='''quick_gelu'''				,							projection_dim=512				,							)
        _UpperCAmelCase           =   CLIPTextModel(_UpperCamelCase					)
        _UpperCAmelCase           =   CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

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

        return components





     def 		UpperCamelCase(     self				,							_UpperCamelCase				,							_UpperCamelCase=0					):
        if str(_UpperCamelCase					).startswith('''mps'''					):
           _UpperCAmelCase           =   torch.manual_seed(_UpperCamelCase					)
        else:
           _UpperCAmelCase           =   torch.Generator(device=_UpperCamelCase					).manual_seed(_UpperCamelCase					)
        _UpperCAmelCase           =   {
            '''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 		UpperCamelCase(     self					):
        _UpperCAmelCase           =   '''cpu'''

        _UpperCAmelCase           =   self.get_dummy_components()
        _UpperCAmelCase           =   self.pipeline_class(**_UpperCamelCase					)
        pipe.to(_UpperCamelCase					)
        pipe.set_progress_bar_config(disable=_UpperCamelCase					)

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

        self.assertEqual(image.shape				,							(1, 256, 256, 3)					)
        _UpperCAmelCase           =   np.array(
            [0.47222412, 0.41921633, 0.44717434, 0.46874192, 0.42588258, 0.46150726, 0.4677534, 0.45583832, 0.48579055]					)
        _UpperCAmelCase           =   np.abs(image_slice.flatten() - expected_slice					).max()
        self.assertLessEqual(_UpperCamelCase				,							1e-3					)





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





     def 		UpperCamelCase(     self					):
        super().test_cpu_offload_forward_pass(expected_max_diff=3e-3					)





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





     def 		UpperCamelCase(     self					):
        super().test_inference_batch_single_identical(expected_max_diff=7e-3					)





     def 		UpperCamelCase(     self					):
        super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3e-3					)





     def 		UpperCamelCase(     self					):
        super().test_save_load_local(expected_max_difference=3e-3					)





     def 		UpperCamelCase(     self					):
        super().test_save_load_optional_components(expected_max_difference=3e-3					)





     def 		UpperCamelCase(     self					):
        _UpperCAmelCase           =   [
            '''DDIMScheduler''',
            '''DDPMScheduler''',
            '''PNDMScheduler''',
            '''HeunDiscreteScheduler''',
            '''EulerAncestralDiscreteScheduler''',
            '''KDPM2DiscreteScheduler''',
            '''KDPM2AncestralDiscreteScheduler''',
            '''DPMSolverSDEScheduler''',
        ]
        _UpperCAmelCase           =   self.get_dummy_components()
        _UpperCAmelCase           =   self.pipeline_class(**_UpperCamelCase					)

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

        pipe.to(_UpperCamelCase					)
        pipe.set_progress_bar_config(disable=_UpperCamelCase					)
        _UpperCAmelCase           =   self.get_dummy_inputs(_UpperCamelCase					)
        _UpperCAmelCase           =   2

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

           _UpperCAmelCase           =   getattr(_UpperCamelCase				,							scheduler_enum.name					)
           _UpperCAmelCase           =   scheduler_cls.from_config(pipe.scheduler.config					)
           _UpperCAmelCase           =   pipe(**_UpperCamelCase					)[0]
           outputs.append(_UpperCamelCase					)

        assert check_same_shape(_UpperCamelCase					)




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





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





     def 		UpperCamelCase(     self					):
        _UpperCAmelCase           =   torch.manual_seed(33					)

        _UpperCAmelCase           =   StableDiffusionPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''				,							torch_dtype=torch.floataa					)
        pipe.to('''cuda'''					)

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

        _UpperCAmelCase           =   '''a photo of an astronaut high resolution, unreal engine, ultra realistic'''

        _UpperCAmelCase           =   pipe(_UpperCamelCase				,							generator=_UpperCamelCase				,							output_type='''latent'''					).images

        _UpperCAmelCase           =   upscaler(
            prompt=_UpperCamelCase				,							image=_UpperCamelCase				,							num_inference_steps=20				,							guidance_scale=0				,							generator=_UpperCamelCase				,							output_type='''np'''				,							).images[0]

        _UpperCAmelCase           =   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 		UpperCamelCase(     self					):
        _UpperCAmelCase           =   torch.manual_seed(33					)

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

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

        _UpperCAmelCase           =   load_image(
            '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png'''					)

        _UpperCAmelCase           =   upscaler(
            prompt=_UpperCamelCase				,							image=_UpperCamelCase				,							num_inference_steps=20				,							guidance_scale=0				,							generator=_UpperCamelCase				,							output_type='''np'''				,							).images[0]

        _UpperCAmelCase           =   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






from typing import List

from .keymap import KEYMAP, get_character
def      A__       (    SCREAMING_SNAKE_CASE_   :      str						)  ->  List[str]:






   """simple docstring"""







   def decorator(SCREAMING_SNAKE_CASE_   :      List[Any]						):
      _UpperCAmelCase           =   getattr(SCREAMING_SNAKE_CASE_				,	'''handle_key'''				,	[]						)
      handle += [key]
      setattr(SCREAMING_SNAKE_CASE_				,	'''handle_key'''				,	SCREAMING_SNAKE_CASE_						)
      return func

   return decorator
def      A__       (    *SCREAMING_SNAKE_CASE_   :      List[str]						)  ->  Dict:






   """simple docstring"""







   def decorator(SCREAMING_SNAKE_CASE_   :      Any						):
      _UpperCAmelCase           =   getattr(SCREAMING_SNAKE_CASE_				,	'''handle_key'''				,	[]						)
      handle += keys
      setattr(SCREAMING_SNAKE_CASE_				,	'''handle_key'''				,	SCREAMING_SNAKE_CASE_						)
      return func

   return decorator




class 			__UpperCamelCase    (							A__    ):





     def __new__(     cls				,							_UpperCamelCase				,							_UpperCamelCase				,							_UpperCamelCase					):
        _UpperCAmelCase           =   super().__new__(cls				,							_UpperCamelCase				,							_UpperCamelCase				,							_UpperCamelCase					)
        if not hasattr(_UpperCamelCase				,							'''key_handler'''					):
           setattr(_UpperCamelCase				,							'''key_handler'''				,							{}					)
        setattr(_UpperCamelCase				,							'''handle_input'''				,							KeyHandler.handle_input					)

        for value in attrs.values():
           _UpperCAmelCase           =   getattr(_UpperCamelCase				,							'''handle_key'''				,							[]					)
           for key in handled_keys:
              _UpperCAmelCase           =   value
        return new_cls





     @staticmethod
     def 		UpperCamelCase(     cls					):
        _UpperCAmelCase           =   get_character()
        if char != KEYMAP["undefined"]:
           _UpperCAmelCase           =   ord(_UpperCamelCase					)
        _UpperCAmelCase           =   cls.key_handler.get(_UpperCamelCase					)
        if handler:
           _UpperCAmelCase           =   char
           return handler(cls					)
        else:
           return None




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






   """simple docstring"""







   return KeyHandler(cls.__name__				,	cls.__bases__				,	cls.__dict__.copy()						)