


"""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    _lowerCamelCase   (     lowerCamelCase__	:							Dict  ):
 lowercase__       :       Tuple				=						[tensor.shape for tensor in tensor_list]
 return all(shape == shapes[0] for shape in shapes[1:]  )

class 				_SCREAMING_SNAKE_CASE				(							__UpperCAmelCase     ,   __UpperCAmelCase     ,   __UpperCAmelCase     ,   unittest.TestCase  ):
   """simple docstring"""



   _a   :       List[Any]									=    StableDiffusionLatentUpscalePipeline
   _a   :       List[Any]									=    TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
       '''height''',
       '''width''',
       '''cross_attention_kwargs''',
       '''negative_prompt_embeds''',
       '''prompt_embeds''',
   }
   _a   :       Tuple									=    PipelineTesterMixin.required_optional_params - {'''num_images_per_prompt'''}
   _a   :       Optional[int]									=    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
   _a   :       List[str]									=    frozenset(
       []  )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
   _a   :       List[str]									=    frozenset([]  )

   _a   :       Optional[Any]									=    True


   @property
   def 					UpperCAmelCase__(						self						)      ->							Dict:
    lowercase__       :       int				=						1
    lowercase__       :       List[Any]				=						4
    lowercase__       :       Optional[Any]				=						(16, 16)

    lowercase__       :       List[str]				=						floats_tensor((batch_size, num_channels) + sizes ,     rng=random.Random(0						)						).to(lowerCamelCase__						)
    return image


   def 					UpperCAmelCase__(						self						)      ->							str:
    torch.manual_seed(0						)
    lowercase__       :       int				=						UNetaDConditionModel(
        act_fn="""gelu""" ,     attention_head_dim=8 ,     norm_num_groups=lowerCamelCase__ ,     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=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""") ,     )
    lowercase__       :       Dict				=						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 ,     )
    lowercase__       :       Union[str, Any]				=						EulerDiscreteScheduler(prediction_type="""sample"""						)
    lowercase__       :       Any				=						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 ,     )
    lowercase__       :       Any				=						CLIPTextModel(lowerCamelCase__						)
    lowercase__       :       Optional[int]				=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""						)

    lowercase__       :       List[Any]				=						{
        """unet""": model.eval(),
        """vae""": vae.eval(),
        """scheduler""": scheduler,
        """text_encoder""": text_encoder,
        """tokenizer""": tokenizer,
    }

    return components


   def 					UpperCAmelCase__(						self ,     lowerCamelCase__ ,     lowerCamelCase__=0						)      ->							str:
    if str(lowerCamelCase__						).startswith("""mps"""						):
     lowercase__       :       Tuple				=						torch.manual_seed(lowerCamelCase__						)
    else:
     lowercase__       :       List[Any]				=						torch.Generator(device=lowerCamelCase__						).manual_seed(lowerCamelCase__						)
    lowercase__       :       Tuple				=						{
        """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						)      ->							Tuple:
    lowercase__       :       str				=						"""cpu"""

    lowercase__       :       List[str]				=						self.get_dummy_components()
    lowercase__       :       List[Any]				=						self.pipeline_class(**lowerCamelCase__						)
    pipe.to(lowerCamelCase__						)
    pipe.set_progress_bar_config(disable=lowerCamelCase__						)

    lowercase__       :       Optional[Any]				=						self.get_dummy_inputs(lowerCamelCase__						)
    lowercase__       :       Optional[Any]				=						pipe(**lowerCamelCase__						).images
    lowercase__       :       str				=						image[0, -3:, -3:, -1]

    self.assertEqual(image.shape ,     (1, 256, 256, 3)						)
    lowercase__       :       Union[str, Any]				=						np.array(
        [0.4722_2412, 0.4192_1633, 0.4471_7434, 0.4687_4192, 0.4258_8258, 0.4615_0726, 0.467_7534, 0.4558_3832, 0.4857_9055]						)
    lowercase__       :       Dict				=						np.abs(image_slice.flatten() - expected_slice						).max()
    self.assertLessEqual(lowerCamelCase__ ,     1E-3						)


   def 					UpperCAmelCase__(						self						)      ->							Dict:
    super().test_attention_slicing_forward_pass(expected_max_diff=7E-3						)


   def 					UpperCAmelCase__(						self						)      ->							List[Any]:
    super().test_cpu_offload_forward_pass(expected_max_diff=3E-3						)


   def 					UpperCAmelCase__(						self						)      ->							List[Any]:
    super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3						)


   def 					UpperCAmelCase__(						self						)      ->							Tuple:
    super().test_inference_batch_single_identical(expected_max_diff=7E-3						)


   def 					UpperCAmelCase__(						self						)      ->							Union[str, Any]:
    super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3E-3						)


   def 					UpperCAmelCase__(						self						)      ->							Tuple:
    super().test_save_load_local(expected_max_difference=3E-3						)


   def 					UpperCAmelCase__(						self						)      ->							Optional[int]:
    super().test_save_load_optional_components(expected_max_difference=3E-3						)







   def 					UpperCAmelCase__(						self						)      ->							Union[str, Any]:
    lowercase__       :       Optional[Any]				=						[
        """DDIMScheduler""",
        """DDPMScheduler""",
        """PNDMScheduler""",
        """HeunDiscreteScheduler""",
        """EulerAncestralDiscreteScheduler""",
        """KDPM2DiscreteScheduler""",
        """KDPM2AncestralDiscreteScheduler""",
        """DPMSolverSDEScheduler""",
    ]
    lowercase__       :       Optional[int]				=						self.get_dummy_components()
    lowercase__       :       Optional[int]				=						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__						)
    lowercase__       :       Any				=						self.get_dummy_inputs(lowerCamelCase__						)
    lowercase__       :       Any				=						2

    lowercase__       :       Union[str, Any]				=						[]
    for scheduler_enum in KarrasDiffusionSchedulers:
     if scheduler_enum.name in skip_schedulers:
      # no sigma schedulers are not supported
      # no schedulers
      continue

     lowercase__       :       Optional[int]				=						getattr(lowerCamelCase__ ,     scheduler_enum.name						)
     lowercase__       :       str				=						scheduler_cls.from_config(pipe.scheduler.config						)
     lowercase__       :       Any				=						pipe(**lowerCamelCase__						)[0]
     outputs.append(lowerCamelCase__						)

    assert check_same_shape(lowerCamelCase__						)

@require_torch_gpu
@slow
class 				_SCREAMING_SNAKE_CASE				(							unittest.TestCase  ):
   """simple docstring"""



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


   def 					UpperCAmelCase__(						self						)      ->							Tuple:
    lowercase__       :       Optional[int]				=						torch.manual_seed(33						)

    lowercase__       :       Union[str, Any]				=						StableDiffusionPipeline.from_pretrained("""CompVis/stable-diffusion-v1-4""" ,     torch_dtype=torch.floataa						)
    pipe.to("""cuda"""						)

    lowercase__       :       Dict				=						StableDiffusionLatentUpscalePipeline.from_pretrained(
        """stabilityai/sd-x2-latent-upscaler""" ,     torch_dtype=torch.floataa						)
    upscaler.to("""cuda"""						)

    lowercase__       :       Tuple				=						"""a photo of an astronaut high resolution, unreal engine, ultra realistic"""

    lowercase__       :       Tuple				=						pipe(lowerCamelCase__ ,     generator=lowerCamelCase__ ,     output_type="""latent"""						).images

    lowercase__       :       Optional[int]				=						upscaler(
        prompt=lowerCamelCase__ ,     image=lowerCamelCase__ ,     num_inference_steps=20 ,     guidance_scale=0 ,     generator=lowerCamelCase__ ,     output_type="""np""" ,     ).images[0]

    lowercase__       :       Any				=						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						)      ->							Dict:
    lowercase__       :       int				=						torch.manual_seed(33						)

    lowercase__       :       int				=						StableDiffusionLatentUpscalePipeline.from_pretrained(
        """stabilityai/sd-x2-latent-upscaler""" ,     torch_dtype=torch.floataa						)
    upscaler.to("""cuda"""						)

    lowercase__       :       Union[str, Any]				=						"""the temple of fire by Ross Tran and Gerardo Dottori, oil on canvas"""

    lowercase__       :       List[str]				=						load_image(
        """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png"""						)

    lowercase__       :       Optional[int]				=						upscaler(
        prompt=lowerCamelCase__ ,     image=lowerCamelCase__ ,     num_inference_steps=20 ,     guidance_scale=0 ,     generator=lowerCamelCase__ ,     output_type="""np""" ,     ).images[0]

    lowercase__       :       List[Any]				=						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"""


import json
import os
import tempfile
import unittest

import numpy as np
from datasets import load_dataset

from transformers.testing_utils import require_torch, require_vision, slow
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
   import torch

if is_vision_available():
   from PIL import Image

   from transformers import ImageGPTImageProcessor

class 				_SCREAMING_SNAKE_CASE				(							unittest.TestCase  ):
   """simple docstring"""



   def __init__(						self ,     lowerCamelCase__ ,     lowerCamelCase__=7 ,     lowerCamelCase__=3 ,     lowerCamelCase__=18 ,     lowerCamelCase__=30 ,     lowerCamelCase__=400 ,     lowerCamelCase__=True ,     lowerCamelCase__=None ,     lowerCamelCase__=True ,     )      ->							int:
    lowercase__       :       int				=						size if size is not None else {"""height""": 18, """width""": 18}
    lowercase__       :       Optional[Any]				=						parent
    lowercase__       :       Dict				=						batch_size
    lowercase__       :       Union[str, Any]				=						num_channels
    lowercase__       :       Tuple				=						image_size
    lowercase__       :       str				=						min_resolution
    lowercase__       :       Optional[Any]				=						max_resolution
    lowercase__       :       Union[str, Any]				=						do_resize
    lowercase__       :       Dict				=						size
    lowercase__       :       Optional[Any]				=						do_normalize







   def 					UpperCAmelCase__(						self						)      ->							Any:
    return {
        # here we create 2 clusters for the sake of simplicity
        "clusters": np.asarray(
            [
                [0.8866_4436_3403_3203, 0.6618_8293_6954_4983, 0.3891_7464_0178_6804],
                [-0.6042_5591_4688_1104, -0.0_2295_0088_6052_8469, 0.5423_7973_6900_3296],
            ]						),
        "do_resize": self.do_resize,
        "size": self.size,
        "do_normalize": self.do_normalize,
    }

@require_torch
@require_vision
class 				_SCREAMING_SNAKE_CASE				(							__UpperCAmelCase     ,   unittest.TestCase  ):
   """simple docstring"""



   _a   :       List[Any]									=    ImageGPTImageProcessor if is_vision_available() else None


   def 					UpperCAmelCase__(						self						)      ->							Optional[int]:
    lowercase__       :       str				=						ImageGPTImageProcessingTester(self						)


   @property
   def 					UpperCAmelCase__(						self						)      ->							Optional[Any]:
    return self.image_processor_tester.prepare_image_processor_dict()


   def 					UpperCAmelCase__(						self						)      ->							int:
    lowercase__       :       List[Any]				=						self.image_processing_class(**self.image_processor_dict						)
    self.assertTrue(hasattr(lowerCamelCase__ ,     """clusters"""						)						)
    self.assertTrue(hasattr(lowerCamelCase__ ,     """do_resize"""						)						)
    self.assertTrue(hasattr(lowerCamelCase__ ,     """size"""						)						)
    self.assertTrue(hasattr(lowerCamelCase__ ,     """do_normalize"""						)						)


   def 					UpperCAmelCase__(						self						)      ->							Any:
    lowercase__       :       Optional[Any]				=						self.image_processing_class.from_dict(self.image_processor_dict						)
    self.assertEqual(image_processor.size ,     {"""height""": 18, """width""": 18}						)

    lowercase__       :       Union[str, Any]				=						self.image_processing_class.from_dict(self.image_processor_dict ,     size=42						)
    self.assertEqual(image_processor.size ,     {"""height""": 42, """width""": 42}						)


   def 					UpperCAmelCase__(						self						)      ->							List[str]:
    lowercase__       :       List[Any]				=						self.image_processing_class(**self.image_processor_dict						)
    lowercase__       :       Optional[Any]				=						json.loads(image_processor.to_json_string()						)
    for key, value in self.image_processor_dict.items():
     if key == "clusters":
      self.assertTrue(np.array_equal(lowerCamelCase__ ,     obj[key]						)						)
     else:
      self.assertEqual(obj[key] ,     lowerCamelCase__						)


   def 					UpperCAmelCase__(						self						)      ->							Dict:
    lowercase__       :       Optional[Any]				=						self.image_processing_class(**self.image_processor_dict						)

    with tempfile.TemporaryDirectory() as tmpdirname:
     lowercase__       :       Optional[int]				=						os.path.join(lowerCamelCase__ ,     """image_processor.json"""						)
     image_processor_first.to_json_file(lowerCamelCase__						)
     lowercase__       :       str				=						self.image_processing_class.from_json_file(lowerCamelCase__						).to_dict()

    lowercase__       :       Optional[int]				=						image_processor_first.to_dict()
    for key, value in image_processor_first.items():
     if key == "clusters":
      self.assertTrue(np.array_equal(lowerCamelCase__ ,     image_processor_second[key]						)						)
     else:
      self.assertEqual(image_processor_first[key] ,     lowerCamelCase__						)


   def 					UpperCAmelCase__(						self						)      ->							List[str]:
    lowercase__       :       List[Any]				=						self.image_processing_class(**self.image_processor_dict						)

    with tempfile.TemporaryDirectory() as tmpdirname:
     image_processor_first.save_pretrained(lowerCamelCase__						)
     lowercase__       :       Union[str, Any]				=						self.image_processing_class.from_pretrained(lowerCamelCase__						).to_dict()

    lowercase__       :       Dict				=						image_processor_first.to_dict()
    for key, value in image_processor_first.items():
     if key == "clusters":
      self.assertTrue(np.array_equal(lowerCamelCase__ ,     image_processor_second[key]						)						)
     else:
      self.assertEqual(image_processor_first[key] ,     lowerCamelCase__						)







   @unittest.skip("""ImageGPT requires clusters at initialization"""						)
   def 					UpperCAmelCase__(						self						)      ->							Dict:
    pass


def    _lowerCamelCase   (     ):
 lowercase__       :       Tuple				=						load_dataset("""hf-internal-testing/fixtures_image_utils"""       ,		split="""test"""  )

 lowercase__       :       Optional[int]				=						Image.open(dataset[4]["""file"""]  )
 lowercase__       :       Union[str, Any]				=						Image.open(dataset[5]["""file"""]  )

 lowercase__       :       Optional[int]				=						[imagea, imagea]

 return images






@require_vision
@require_torch
class 				_SCREAMING_SNAKE_CASE				(							unittest.TestCase  ):
   """simple docstring"""



   @slow
   def 					UpperCAmelCase__(						self						)      ->							str:
    lowercase__       :       Optional[int]				=						ImageGPTImageProcessor.from_pretrained("""openai/imagegpt-small"""						)

    lowercase__       :       int				=						prepare_images()

    # test non-batched
    lowercase__       :       int				=						image_processing(images[0] ,     return_tensors="""pt"""						)

    self.assertIsInstance(encoding.input_ids ,     torch.LongTensor						)
    self.assertEqual(encoding.input_ids.shape ,     (1, 1024)						)

    lowercase__       :       Any				=						[306, 191, 191]
    self.assertEqual(encoding.input_ids[0, :3].tolist() ,     lowerCamelCase__						)

    # test batched
    lowercase__       :       str				=						image_processing(lowerCamelCase__ ,     return_tensors="""pt"""						)

    self.assertIsInstance(encoding.input_ids ,     torch.LongTensor						)
    self.assertEqual(encoding.input_ids.shape ,     (2, 1024)						)

    lowercase__       :       Optional[int]				=						[303, 13, 13]
    self.assertEqual(encoding.input_ids[1, -3:].tolist() ,     lowerCamelCase__						)