



import shutil
import tempfile
import unittest

import numpy as np

from transformers.testing_utils import (
    is_pt_tf_cross_test,
    require_tf,
    require_torch,
    require_torchvision,
    require_vision,
)
from transformers.utils import is_tf_available, is_torch_available, is_vision_available


if is_vision_available():
   from PIL import Image

   from transformers import AutoProcessor, SamImageProcessor, SamProcessor

if is_torch_available():
   import torch

if is_tf_available():
   import tensorflow as tf



@require_vision
@require_torchvision
class      _UpperCAmelCase      (							unittest.TestCase  ):




     def 		snake_case_		(					self):
         A__					=       tempfile.mkdtemp()
         A__					=       SamImageProcessor()
         A__					=       SamProcessor(a__)
         processor.save_pretrained(self.tmpdirname)




     def 		snake_case_		(					self			,    **a__):
         return AutoProcessor.from_pretrained(self.tmpdirname			,    **a__).image_processor




     def 		snake_case_		(					self):
         shutil.rmtree(self.tmpdirname)




     def 		snake_case_		(					self):
         A__					=       [np.random.randint(2_5_5			,    size=(3, 3_0, 4_0_0)			,    dtype=np.uinta)]

         A__					=       [Image.fromarray(np.moveaxis(a__			,    0			,    -1)) for x in image_inputs]

         return image_inputs




     def 		snake_case_		(					self):
         A__					=       SamProcessor(image_processor=self.get_image_processor())
         processor.save_pretrained(self.tmpdirname)

         A__					=       self.get_image_processor(do_normalize=a__			,    padding_value=1.0)

         A__					=       SamProcessor.from_pretrained(self.tmpdirname			,    do_normalize=a__			,    padding_value=1.0)

         self.assertEqual(processor.image_processor.to_json_string()			,    image_processor_add_kwargs.to_json_string())
         self.assertIsInstance(processor.image_processor			,    a__)




     def 		snake_case_		(					self):
         A__					=       self.get_image_processor()

         A__					=       SamProcessor(image_processor=a__)

         A__					=       self.prepare_image_inputs()

         A__					=       image_processor(a__			,    return_tensors='''np''')
         A__					=       processor(images=a__			,    return_tensors='''np''')

         input_feat_extract.pop('''original_sizes''')  # pop original_sizes as it is popped in the processor
         input_feat_extract.pop('''reshaped_input_sizes''')  # pop original_sizes as it is popped in the processor

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




     @require_torch
     def 		snake_case_		(					self):
         A__					=       self.get_image_processor()

         A__					=       SamProcessor(image_processor=a__)
         A__					=       [torch.ones((1, 3, 5, 5))]

         A__					=       [[1_7_6_4, 2_6_4_6]]

         A__					=       [[6_8_3, 1_0_2_4]]
         A__					=       processor.post_process_masks(a__			,    a__			,    a__)
         self.assertEqual(masks[0].shape			,    (1, 3, 1_7_6_4, 2_6_4_6))

         A__					=       processor.post_process_masks(
             a__			,    torch.tensor(a__)			,    torch.tensor(a__))
         self.assertEqual(masks[0].shape			,    (1, 3, 1_7_6_4, 2_6_4_6))

         # should also work with np
         A__					=       [np.ones((1, 3, 5, 5))]
         A__					=       processor.post_process_masks(a__			,    np.array(a__)			,    np.array(a__))

         self.assertEqual(masks[0].shape			,    (1, 3, 1_7_6_4, 2_6_4_6))

         A__					=       [[1, 0], [0, 1]]
         with self.assertRaises(a__):
             A__					=       processor.post_process_masks(a__			,    np.array(a__)			,    np.array(a__))







@require_vision
@require_tf
class      _UpperCAmelCase      (							unittest.TestCase  ):




     def 		snake_case_		(					self):
         A__					=       tempfile.mkdtemp()
         A__					=       SamImageProcessor()
         A__					=       SamProcessor(a__)
         processor.save_pretrained(self.tmpdirname)




     def 		snake_case_		(					self			,    **a__):
         return AutoProcessor.from_pretrained(self.tmpdirname			,    **a__).image_processor




     def 		snake_case_		(					self):
         shutil.rmtree(self.tmpdirname)




     def 		snake_case_		(					self):
         A__					=       [np.random.randint(2_5_5			,    size=(3, 3_0, 4_0_0)			,    dtype=np.uinta)]

         A__					=       [Image.fromarray(np.moveaxis(a__			,    0			,    -1)) for x in image_inputs]

         return image_inputs




     def 		snake_case_		(					self):
         A__					=       SamProcessor(image_processor=self.get_image_processor())
         processor.save_pretrained(self.tmpdirname)

         A__					=       self.get_image_processor(do_normalize=a__			,    padding_value=1.0)

         A__					=       SamProcessor.from_pretrained(self.tmpdirname			,    do_normalize=a__			,    padding_value=1.0)

         self.assertEqual(processor.image_processor.to_json_string()			,    image_processor_add_kwargs.to_json_string())
         self.assertIsInstance(processor.image_processor			,    a__)




     def 		snake_case_		(					self):
         A__					=       self.get_image_processor()

         A__					=       SamProcessor(image_processor=a__)

         A__					=       self.prepare_image_inputs()

         A__					=       image_processor(a__			,    return_tensors='''np''')
         A__					=       processor(images=a__			,    return_tensors='''np''')

         input_feat_extract.pop('''original_sizes''')  # pop original_sizes as it is popped in the processor
         input_feat_extract.pop('''reshaped_input_sizes''')  # pop reshaped_input_sizes as it is popped in the processor

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




     @require_tf
     def 		snake_case_		(					self):
         A__					=       self.get_image_processor()

         A__					=       SamProcessor(image_processor=a__)
         A__					=       [tf.ones((1, 3, 5, 5))]

         A__					=       [[1_7_6_4, 2_6_4_6]]

         A__					=       [[6_8_3, 1_0_2_4]]
         A__					=       processor.post_process_masks(a__			,    a__			,    a__			,    return_tensors='''tf''')
         self.assertEqual(masks[0].shape			,    (1, 3, 1_7_6_4, 2_6_4_6))

         A__					=       processor.post_process_masks(
             a__			,    tf.convert_to_tensor(a__)			,    tf.convert_to_tensor(a__)			,    return_tensors='''tf'''			,    )
         self.assertEqual(masks[0].shape			,    (1, 3, 1_7_6_4, 2_6_4_6))

         # should also work with np
         A__					=       [np.ones((1, 3, 5, 5))]
         A__					=       processor.post_process_masks(
             a__			,    np.array(a__)			,    np.array(a__)			,    return_tensors='''tf''')

         self.assertEqual(masks[0].shape			,    (1, 3, 1_7_6_4, 2_6_4_6))

         A__					=       [[1, 0], [0, 1]]
         with self.assertRaises(tf.errors.InvalidArgumentError):
             A__					=       processor.post_process_masks(
                 a__			,    np.array(a__)			,    np.array(a__)			,    return_tensors='''tf''')







@require_vision
@require_torchvision
class      _UpperCAmelCase      (							unittest.TestCase  ):




     def 		snake_case_		(					self):
         A__					=       tempfile.mkdtemp()
         A__					=       SamImageProcessor()
         A__					=       SamProcessor(a__)
         processor.save_pretrained(self.tmpdirname)




     def 		snake_case_		(					self			,    **a__):
         return AutoProcessor.from_pretrained(self.tmpdirname			,    **a__).image_processor




     def 		snake_case_		(					self):
         shutil.rmtree(self.tmpdirname)




     def 		snake_case_		(					self):
         A__					=       [np.random.randint(2_5_5			,    size=(3, 3_0, 4_0_0)			,    dtype=np.uinta)]

         A__					=       [Image.fromarray(np.moveaxis(a__			,    0			,    -1)) for x in image_inputs]

         return image_inputs




     @is_pt_tf_cross_test
     def 		snake_case_		(					self):
         A__					=       self.get_image_processor()

         A__					=       SamProcessor(image_processor=a__)
         A__					=       np.random.randint(0			,    2			,    size=(1, 3, 5, 5)).astype(np.floataa)
         A__					=       [tf.convert_to_tensor(a__)]
         A__					=       [torch.tensor(a__)]

         A__					=       [[1_7_6_4, 2_6_4_6]]

         A__					=       [[6_8_3, 1_0_2_4]]
         A__					=       processor.post_process_masks(
             a__			,    a__			,    a__			,    return_tensors='''tf''')
         A__					=       processor.post_process_masks(
             a__			,    a__			,    a__			,    return_tensors='''pt''')

         self.assertTrue(np.all(tf_masks[0].numpy() == pt_masks[0].numpy()))




     @is_pt_tf_cross_test
     def 		snake_case_		(					self):
         A__					=       self.get_image_processor()

         A__					=       SamProcessor(image_processor=a__)

         A__					=       self.prepare_image_inputs()

         A__					=       image_processor(a__			,    return_tensors='''pt''')['''pixel_values'''].numpy()
         A__					=       processor(images=a__			,    return_tensors='''pt''')['''pixel_values'''].numpy()

         A__					=       image_processor(a__			,    return_tensors='''tf''')['''pixel_values'''].numpy()
         A__					=       processor(images=a__			,    return_tensors='''tf''')['''pixel_values'''].numpy()

         self.assertTrue(np.allclose(a__			,    a__))
         self.assertTrue(np.allclose(a__			,    a__))
         self.assertTrue(np.allclose(a__			,    a__))





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__     (       UpperCamelCase_					:							Union[str, Any]						)->	List[str]:
    A__					=       [tensor.shape for tensor in tensor_list]
    return all(shape == shapes[0] for shape in shapes[1:]						)



class      _UpperCAmelCase      (							A__      ,   A__      ,   A__      ,   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 		snake_case_		(					self):
         A__					=       1
         A__					=       4
         A__					=       (1_6, 1_6)

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




     def 		snake_case_		(					self):
         torch.manual_seed(0)
         A__					=       UNetaDConditionModel(
             act_fn='''gelu'''			,    attention_head_dim=8			,    norm_num_groups=a__			,    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=a__			,    only_cross_attention=a__			,    out_channels=5			,    resnet_time_scale_shift='''scale_shift'''			,    time_embedding_type='''fourier'''			,    timestep_post_act='''gelu'''			,    up_block_types=('''KCrossAttnUpBlock2D''', '''KCrossAttnUpBlock2D''', '''KCrossAttnUpBlock2D''', '''KUpBlock2D''')			,    )
         A__					=       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			,    )
         A__					=       EulerDiscreteScheduler(prediction_type='''sample''')
         A__					=       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			,    )
         A__					=       CLIPTextModel(a__)
         A__					=       CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip''')

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

         return components




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

         A__					=       self.get_dummy_components()
         A__					=       self.pipeline_class(**a__)
         pipe.to(a__)
         pipe.set_progress_bar_config(disable=a__)

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

         self.assertEqual(image.shape			,    (1, 2_5_6, 2_5_6, 3))
         A__					=       np.array(
             [0.4_7_2_2_2_4_1_2, 0.4_1_9_2_1_6_3_3, 0.4_4_7_1_7_4_3_4, 0.4_6_8_7_4_1_9_2, 0.4_2_5_8_8_2_5_8, 0.4_6_1_5_0_7_2_6, 0.4_6_7_7_5_3_4, 0.4_5_5_8_3_8_3_2, 0.4_8_5_7_9_0_5_5])
         A__					=       np.abs(image_slice.flatten() - expected_slice).max()
         self.assertLessEqual(a__			,    1e-3)




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




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




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




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




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




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




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




     def 		snake_case_		(					self):
         A__					=       [
             '''DDIMScheduler''',
             '''DDPMScheduler''',
             '''PNDMScheduler''',
             '''HeunDiscreteScheduler''',
             '''EulerAncestralDiscreteScheduler''',
             '''KDPM2DiscreteScheduler''',
             '''KDPM2AncestralDiscreteScheduler''',
             '''DPMSolverSDEScheduler''',
         ]
         A__					=       self.get_dummy_components()
         A__					=       self.pipeline_class(**a__)

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

         pipe.to(a__)
         pipe.set_progress_bar_config(disable=a__)
         A__					=       self.get_dummy_inputs(a__)
         A__					=       2

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

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

         assert check_same_shape(a__)







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




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




     def 		snake_case_		(					self):
         A__					=       torch.manual_seed(3_3)

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

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

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

         A__					=       pipe(a__			,    generator=a__			,    output_type='''latent''').images

         A__					=       upscaler(
             prompt=a__			,    image=a__			,    num_inference_steps=2_0			,    guidance_scale=0			,    generator=a__			,    output_type='''np'''			,    ).images[0]

         A__					=       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 		snake_case_		(					self):
         A__					=       torch.manual_seed(3_3)

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

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

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

         A__					=       upscaler(
             prompt=a__			,    image=a__			,    num_inference_steps=2_0			,    guidance_scale=0			,    generator=a__			,    output_type='''np'''			,    ).images[0]

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