from __future__ import annotations

import copy
import inspect
import json
import math
import os
import tempfile
import unittest
from importlib import import_module

import numpy as np

from transformers import ViTMAEConfig
from transformers.file_utils import cached_property, is_tf_available, is_vision_available
from transformers.testing_utils import require_tf, require_vision, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
      import tensorflow as tf

      from transformers import TFViTMAEForPreTraining, TFViTMAEModel


if is_vision_available():
      from PIL import Image

      from transformers import ViTImageProcessor





class      lowerCAmelCase						:



     def __init__(					self				,      lowerCAmelCase__				,      lowerCAmelCase__=13				,      lowerCAmelCase__=30				,      lowerCAmelCase__=2				,      lowerCAmelCase__=3				,      lowerCAmelCase__=True				,      lowerCAmelCase__=True				,      lowerCAmelCase__=32				,      lowerCAmelCase__=2				,      lowerCAmelCase__=4				,      lowerCAmelCase__=37				,      lowerCAmelCase__="gelu"				,      lowerCAmelCase__=0.1				,      lowerCAmelCase__=0.1				,      lowerCAmelCase__=10				,      lowerCAmelCase__=0.02				,      lowerCAmelCase__=3				,      lowerCAmelCase__=0.6				,      lowerCAmelCase__=None				,      ):
            _A=	parent
            _A=	batch_size
            _A=	image_size
            _A=	patch_size
            _A=	num_channels
            _A=	is_training
            _A=	use_labels
            _A=	hidden_size
            _A=	num_hidden_layers
            _A=	num_attention_heads
            _A=	intermediate_size
            _A=	hidden_act
            _A=	hidden_dropout_prob
            _A=	attention_probs_dropout_prob
            _A=	type_sequence_label_size
            _A=	initializer_range
            _A=	mask_ratio
            _A=	scope

            # in ViTMAE, the expected sequence length = (num_patches + 1) * (1 - config.mask_ratio), rounded above
            # (we add 1 for the [CLS] token)
            _A=	(image_size // patch_size) ** 2
            _A=	int(math.ceil((1 - mask_ratio) * (num_patches + 1)       )       )



     def 			a__  (					self       ):
            _A=	floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]       )

            _A=	None
            if self.use_labels:
                   _A=	ids_tensor([self.batch_size]				,      self.type_sequence_label_size       )

            _A=	self.get_config()

            return config, pixel_values, labels



     def 			a__  (					self       ):
            return ViTMAEConfig(
                image_size=self.image_size				,      patch_size=self.patch_size				,      num_channels=self.num_channels				,      hidden_size=self.hidden_size				,      num_hidden_layers=self.num_hidden_layers				,      num_attention_heads=self.num_attention_heads				,      intermediate_size=self.intermediate_size				,      decoder_hidden_size=self.hidden_size				,      decoder_num_hidden_layers=self.num_hidden_layers				,      decoder_num_attention_heads=self.num_attention_heads				,      decoder_intermediate_size=self.intermediate_size				,      hidden_act=self.hidden_act				,      hidden_dropout_prob=self.hidden_dropout_prob				,      attention_probs_dropout_prob=self.attention_probs_dropout_prob				,      is_decoder=lowerCAmelCase__				,      initializer_range=self.initializer_range				,      mask_ratio=self.mask_ratio				,      )



     def 			a__  (					self				,      lowerCAmelCase__				,      lowerCAmelCase__				,      lowerCAmelCase__       ):
            _A=	TFViTMAEModel(config=lowerCAmelCase__       )
            _A=	model(lowerCAmelCase__				,      training=lowerCAmelCase__       )
            self.parent.assertEqual(result.last_hidden_state.shape				,      (self.batch_size, self.seq_length, self.hidden_size)       )



     def 			a__  (					self				,      lowerCAmelCase__				,      lowerCAmelCase__				,      lowerCAmelCase__       ):
            _A=	TFViTMAEForPreTraining(lowerCAmelCase__       )
            _A=	model(lowerCAmelCase__				,      training=lowerCAmelCase__       )
            # expected sequence length = num_patches
            _A=	(self.image_size // self.patch_size) ** 2
            _A=	self.patch_size**2 * self.num_channels
            self.parent.assertEqual(result.logits.shape				,      (self.batch_size, num_patches, expected_num_channels)       )

            # test greyscale images
            _A=	1
            _A=	TFViTMAEForPreTraining(lowerCAmelCase__       )

            _A=	floats_tensor([self.batch_size, 1, self.image_size, self.image_size]       )
            _A=	model(lowerCAmelCase__				,      training=lowerCAmelCase__       )
            _A=	self.patch_size**2
            self.parent.assertEqual(result.logits.shape				,      (self.batch_size, num_patches, expected_num_channels)       )





     def 			a__  (					self       ):
            _A=	self.prepare_config_and_inputs()
            ((_A),			(_A),			(_A))=	config_and_inputs
            _A=	{'pixel_values': pixel_values}
            return config, inputs_dict





@require_tf
class      lowerCAmelCase						(					_a			,		_a			,		unittest.TestCase			):
     _SCREAMING_SNAKE_CASE     :     Any		=(TFViTMAEModel, TFViTMAEForPreTraining) if is_tf_available() else ()
     _SCREAMING_SNAKE_CASE     :     int		={"""feature-extraction""": TFViTMAEModel} if is_tf_available() else {}

     _SCREAMING_SNAKE_CASE     :     List[str]		=False
     _SCREAMING_SNAKE_CASE     :     Union[str, Any]		=False
     _SCREAMING_SNAKE_CASE     :     List[Any]		=False
     _SCREAMING_SNAKE_CASE     :     int		=False



     def 			a__  (					self       ):
            _A=	TFViTMAEModelTester(self       )
            _A=	ConfigTester(self				,      config_class=lowerCAmelCase__				,      has_text_modality=lowerCAmelCase__				,      hidden_size=37       )



     def 			a__  (					self       ):
            self.config_tester.run_common_tests()



     @unittest.skip(reason='ViTMAE does not use inputs_embeds'       )
     def 			a__  (					self       ):
            pass



     def 			a__  (					self       ):
            _A,			_A=	self.model_tester.prepare_config_and_inputs_for_common()

            for model_class in self.all_model_classes:
                   _A=	model_class(lowerCAmelCase__       )
                   self.assertIsInstance(model.get_input_embeddings()				,      (tf.keras.layers.Layer)       )
                   _A=	model.get_output_embeddings()
                   self.assertTrue(x is None or isinstance(lowerCAmelCase__				,      tf.keras.layers.Layer       )       )



     def 			a__  (					self       ):
            _A,			_A=	self.model_tester.prepare_config_and_inputs_for_common()

            for model_class in self.all_model_classes:
                   _A=	model_class(lowerCAmelCase__       )
                   _A=	inspect.signature(model.call       )
                   # signature.parameters is an OrderedDict => so arg_names order is deterministic
                   _A=	[*signature.parameters.keys()]

                   _A=	['pixel_values']
                   self.assertListEqual(arg_names[:1]				,      lowerCAmelCase__       )



     def 			a__  (					self       ):
            _A=	self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_model(*lowerCAmelCase__       )



     def 			a__  (					self       ):
            _A=	self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_for_pretraining(*lowerCAmelCase__       )



     def 			a__  (					self       ):
            # make the mask reproducible
            np.random.seed(2       )

            _A,			_A=	self.model_tester.prepare_config_and_inputs_for_common()

            _A=	int((config.image_size // config.patch_size) ** 2       )
            _A=	np.random.uniform(size=(self.model_tester.batch_size, num_patches)       )

            for model_class in self.all_model_classes:
                   _A=	model_class(lowerCAmelCase__       )
                   _A=	self._prepare_for_class(lowerCAmelCase__				,      lowerCAmelCase__       )

                   _A=	model(lowerCAmelCase__				,      noise=lowerCAmelCase__       )

                   _A=	copy.deepcopy(self._prepare_for_class(lowerCAmelCase__				,      lowerCAmelCase__       )       )
                   _A=	model(**lowerCAmelCase__				,      noise=lowerCAmelCase__       )
                   _A=	outputs_dict[0].numpy()
                   _A=	outputs_keywords[0].numpy()

                   self.assertLess(np.sum(np.abs(output_dict - output_keywords       )       )				,      1E-6       )



     def 			a__  (					self       ):
            # make the mask reproducible
            np.random.seed(2       )

            _A,			_A=	self.model_tester.prepare_config_and_inputs_for_common()

            _A=	int((config.image_size // config.patch_size) ** 2       )
            _A=	np.random.uniform(size=(self.model_tester.batch_size, num_patches)       )

            def prepare_numpy_arrays(lowerCAmelCase__       ):
                   _A=	{}
                   for k, v in inputs_dict.items():
                          if tf.is_tensor(lowerCAmelCase__       ):
                                 _A=	v.numpy()
                          else:
                                 _A=	np.array(lowerCAmelCase__       )

                   return inputs_np_dict

            for model_class in self.all_model_classes:
                   _A=	model_class(lowerCAmelCase__       )

                   _A=	self._prepare_for_class(lowerCAmelCase__				,      lowerCAmelCase__       )
                   _A=	prepare_numpy_arrays(lowerCAmelCase__       )

                   _A=	model(lowerCAmelCase__				,      noise=lowerCAmelCase__       )
                   _A=	model(**lowerCAmelCase__				,      noise=lowerCAmelCase__       )
                   self.assert_outputs_same(lowerCAmelCase__				,      lowerCAmelCase__       )



     def 			a__  (					self				,      lowerCAmelCase__				,      lowerCAmelCase__				,      lowerCAmelCase__       ):
            # make masks reproducible
            np.random.seed(2       )

            _A=	int((tf_model.config.image_size // tf_model.config.patch_size) ** 2       )
            _A=	np.random.uniform(size=(self.model_tester.batch_size, num_patches)       )
            _A=	tf.constant(lowerCAmelCase__       )

            # Add `noise` argument.
            # PT inputs will be prepared in `super().check_pt_tf_models()` with this added `noise` argument
            _A=	tf_noise

            super().check_pt_tf_models(lowerCAmelCase__				,      lowerCAmelCase__				,      lowerCAmelCase__       )



     def 			a__  (					self       ):
            # make mask reproducible
            np.random.seed(2       )

            _A,			_A=	self.model_tester.prepare_config_and_inputs_for_common()

            _A=	{
                module_member
                for model_class in self.all_model_classes
                for module in (import_module(model_class.__module__       ),)
                for module_member_name in dir(lowerCAmelCase__       )
                if module_member_name.endswith('MainLayer'       )
                # This condition is required, since `modeling_tf_clip.py` has 3 classes whose names end with `MainLayer`.
                and module_member_name[: -len('MainLayer'       )] == model_class.__name__[: -len('Model'       )]
                for module_member in (getattr(lowerCAmelCase__				,      lowerCAmelCase__       ),)
                if isinstance(lowerCAmelCase__				,      lowerCAmelCase__       )
                and tf.keras.layers.Layer in module_member.__bases__
                and getattr(lowerCAmelCase__				,      '_keras_serializable'				,      lowerCAmelCase__       )
            }

            _A=	int((config.image_size // config.patch_size) ** 2       )
            _A=	np.random.uniform(size=(self.model_tester.batch_size, num_patches)       )
            _A=	tf.convert_to_tensor(lowerCAmelCase__       )
            inputs_dict.update({'noise': noise}       )

            for main_layer_class in tf_main_layer_classes:
                   _A=	main_layer_class(lowerCAmelCase__       )

                   _A=	{
                       name: tf.keras.Input(tensor.shape[1:]				,      dtype=tensor.dtype       ) for name, tensor in inputs_dict.items()
                   }

                   _A=	tf.keras.Model(lowerCAmelCase__				,      outputs=main_layer(lowerCAmelCase__       )       )
                   _A=	model(lowerCAmelCase__       )

                   with tempfile.TemporaryDirectory() as tmpdirname:
                          _A=	os.path.join(lowerCAmelCase__				,      'keras_model.h5'       )
                          model.save(lowerCAmelCase__       )
                          _A=	tf.keras.models.load_model(
                              lowerCAmelCase__				,      custom_objects={main_layer_class.__name__: main_layer_class}       )
                          assert isinstance(lowerCAmelCase__				,      tf.keras.Model       )
                          _A=	model(lowerCAmelCase__       )
                          self.assert_outputs_same(lowerCAmelCase__				,      lowerCAmelCase__       )



     @slow
     def 			a__  (					self       ):
            # make mask reproducible
            np.random.seed(2       )

            _A,			_A=	self.model_tester.prepare_config_and_inputs_for_common()

            _A=	int((config.image_size // config.patch_size) ** 2       )
            _A=	np.random.uniform(size=(self.model_tester.batch_size, num_patches)       )

            for model_class in self.all_model_classes:
                   _A=	model_class(lowerCAmelCase__       )
                   _A=	self._prepare_for_class(lowerCAmelCase__				,      lowerCAmelCase__       )
                   _A=	model(lowerCAmelCase__				,      noise=lowerCAmelCase__       )

                   if model_class.__name__ == "TFViTMAEModel":
                          _A=	outputs.last_hidden_state.numpy()
                          _A=	0
                   else:
                          _A=	outputs.logits.numpy()
                          _A=	0

                   with tempfile.TemporaryDirectory() as tmpdirname:
                          model.save_pretrained(lowerCAmelCase__				,      saved_model=lowerCAmelCase__       )
                          _A=	model_class.from_pretrained(lowerCAmelCase__       )
                          _A=	model(lowerCAmelCase__				,      noise=lowerCAmelCase__       )

                          if model_class.__name__ == "TFViTMAEModel":
                                 _A=	after_outputs['last_hidden_state'].numpy()
                                 _A=	0
                          else:
                                 _A=	after_outputs['logits'].numpy()
                                 _A=	0

                          _A=	np.amax(np.abs(out_a - out_a       )       )
                          self.assertLessEqual(lowerCAmelCase__				,      1E-5       )



     def 			a__  (					self       ):
            # make mask reproducible
            np.random.seed(2       )

            _A,			_A=	self.model_tester.prepare_config_and_inputs_for_common()

            _A=	int((config.image_size // config.patch_size) ** 2       )
            _A=	np.random.uniform(size=(self.model_tester.batch_size, num_patches)       )

            for model_class in self.all_model_classes:
                   _A=	model_class(lowerCAmelCase__       )
                   _A=	self._prepare_for_class(lowerCAmelCase__				,      lowerCAmelCase__       )

                   _A=	model(lowerCAmelCase__				,      noise=lowerCAmelCase__       )
                   _A=	model.get_config()
                   # make sure that returned config is jsonifiable, which is required by keras
                   json.dumps(lowerCAmelCase__       )
                   _A=	model_class.from_config(model.get_config()       )
                   # make sure it also accepts a normal config
                   _A=	model_class.from_config(model.config       )
                   _A=	new_model(lowerCAmelCase__       )  # Build model
                   new_model.set_weights(model.get_weights()       )
                   _A=	new_model(lowerCAmelCase__				,      noise=lowerCAmelCase__       )

                   self.assert_outputs_same(lowerCAmelCase__				,      lowerCAmelCase__       )



     @unittest.skip(
         reason='ViTMAE returns a random mask + ids_restore in each forward pass. See test_save_load\n    to get deterministic results.'       )
     def 			a__  (					self       ):
            pass



     @unittest.skip(reason='ViTMAE returns a random mask + ids_restore in each forward pass. See test_save_load'       )
     def 			a__  (					self       ):
            pass





     @slow
     def 			a__  (					self       ):
            _A=	TFViTMAEModel.from_pretrained('google/vit-base-patch16-224'       )
            self.assertIsNotNone(lowerCAmelCase__       )


def 							UpperCamelCase				(    )		->						str:




       '''simple docstring'''






       _A=	Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'   )
       return image





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



     @cached_property
     def 			a__  (					self       ):
            return ViTImageProcessor.from_pretrained('facebook/vit-mae-base'       ) if is_vision_available() else None





     @slow
     def 			a__  (					self       ):
            # make random mask reproducible across the PT and TF model
            np.random.seed(2       )

            _A=	TFViTMAEForPreTraining.from_pretrained('facebook/vit-mae-base'       )

            _A=	self.default_image_processor
            _A=	prepare_img()
            _A=	image_processor(images=lowerCAmelCase__				,      return_tensors='tf'       )

            # prepare a noise vector that will be also used for testing the TF model
            # (this way we can ensure that the PT and TF models operate on the same inputs)
            _A=	ViTMAEConfig()
            _A=	int((vit_mae_config.image_size // vit_mae_config.patch_size) ** 2       )
            _A=	np.random.uniform(size=(1, num_patches)       )

            # forward pass
            _A=	model(**lowerCAmelCase__				,      noise=lowerCAmelCase__       )

            # verify the logits
            _A=	tf.convert_to_tensor([1, 196, 768]       )
            self.assertEqual(outputs.logits.shape				,      lowerCAmelCase__       )

            _A=	tf.convert_to_tensor(
                [[-0.0548, -1.7023, -0.9325], [0.3721, -0.5670, -0.2233], [0.8235, -1.3878, -0.3524]]       )

            tf.debugging.assert_near(outputs.logits[0, :3, :3]				,      lowerCAmelCase__				,      atol=1E-4       )
import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import (
    DiffusionPipeline,
    UnCLIPImageVariationPipeline,
    UnCLIPScheduler,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, load_image, require_torch_gpu, skip_mps

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()





class      lowerCAmelCase						(					_a			,		unittest.TestCase			):
     _SCREAMING_SNAKE_CASE     :     Union[str, Any]		=UnCLIPImageVariationPipeline
     _SCREAMING_SNAKE_CASE     :     Union[str, Any]		=IMAGE_VARIATION_PARAMS - {"""height""", """width""", """guidance_scale"""}
     _SCREAMING_SNAKE_CASE     :     Union[str, Any]		=IMAGE_VARIATION_BATCH_PARAMS

     _SCREAMING_SNAKE_CASE     :     int		=[
         """generator""",
         """return_dict""",
         """decoder_num_inference_steps""",
         """super_res_num_inference_steps""",
     ]
     _SCREAMING_SNAKE_CASE     :     Optional[int]		=False



     @property
     def 			a__  (					self       ):
            return 32



     @property
     def 			a__  (					self       ):
            return 32



     @property
     def 			a__  (					self       ):
            return self.time_input_dim



     @property
     def 			a__  (					self       ):
            return self.time_input_dim * 4



     @property
     def 			a__  (					self       ):
            return 100



     @property
     def 			a__  (					self       ):
            _A=	CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'       )
            return tokenizer



     @property
     def 			a__  (					self       ):
            torch.manual_seed(0       )
            _A=	CLIPTextConfig(
                bos_token_id=0				,      eos_token_id=2				,      hidden_size=self.text_embedder_hidden_size				,      projection_dim=self.text_embedder_hidden_size				,      intermediate_size=37				,      layer_norm_eps=1E-05				,      num_attention_heads=4				,      num_hidden_layers=5				,      pad_token_id=1				,      vocab_size=1000				,      )
            return CLIPTextModelWithProjection(lowerCAmelCase__       )



     @property
     def 			a__  (					self       ):
            torch.manual_seed(0       )
            _A=	CLIPVisionConfig(
                hidden_size=self.text_embedder_hidden_size				,      projection_dim=self.text_embedder_hidden_size				,      num_hidden_layers=5				,      num_attention_heads=4				,      image_size=32				,      intermediate_size=37				,      patch_size=1				,      )
            return CLIPVisionModelWithProjection(lowerCAmelCase__       )



     @property
     def 			a__  (					self       ):
            torch.manual_seed(0       )

            _A=	{
                'clip_embeddings_dim': self.text_embedder_hidden_size,
                'time_embed_dim': self.time_embed_dim,
                'cross_attention_dim': self.cross_attention_dim,
            }

            _A=	UnCLIPTextProjModel(**lowerCAmelCase__       )
            return model



     @property
     def 			a__  (					self       ):
            torch.manual_seed(0       )

            _A=	{
                'sample_size': 32,
                # RGB in channels
                'in_channels': 3,
                # Out channels is double in channels because predicts mean and variance
                'out_channels': 6,
                'down_block_types': ('ResnetDownsampleBlock2D', 'SimpleCrossAttnDownBlock2D'),
                'up_block_types': ('SimpleCrossAttnUpBlock2D', 'ResnetUpsampleBlock2D'),
                'mid_block_type': 'UNetMidBlock2DSimpleCrossAttn',
                'block_out_channels': (self.block_out_channels_a, self.block_out_channels_a * 2),
                'layers_per_block': 1,
                'cross_attention_dim': self.cross_attention_dim,
                'attention_head_dim': 4,
                'resnet_time_scale_shift': 'scale_shift',
                'class_embed_type': 'identity',
            }

            _A=	UNetaDConditionModel(**lowerCAmelCase__       )
            return model



     @property
     def 			a__  (					self       ):
            return {
                "sample_size": 64,
                "layers_per_block": 1,
                "down_block_types": ("ResnetDownsampleBlock2D", "ResnetDownsampleBlock2D"),
                "up_block_types": ("ResnetUpsampleBlock2D", "ResnetUpsampleBlock2D"),
                "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
                "in_channels": 6,
                "out_channels": 3,
            }



     @property
     def 			a__  (					self       ):
            torch.manual_seed(0       )

            _A=	UNetaDModel(**self.dummy_super_res_kwargs       )
            return model



     @property
     def 			a__  (					self       ):
            # seeded differently to get different unet than `self.dummy_super_res_first`
            torch.manual_seed(1       )

            _A=	UNetaDModel(**self.dummy_super_res_kwargs       )
            return model



     def 			a__  (					self       ):
            _A=	self.dummy_decoder
            _A=	self.dummy_text_proj
            _A=	self.dummy_text_encoder
            _A=	self.dummy_tokenizer
            _A=	self.dummy_super_res_first
            _A=	self.dummy_super_res_last

            _A=	UnCLIPScheduler(
                variance_type='learned_range'				,      prediction_type='epsilon'				,      num_train_timesteps=1000				,      )

            _A=	UnCLIPScheduler(
                variance_type='fixed_small_log'				,      prediction_type='epsilon'				,      num_train_timesteps=1000				,      )

            _A=	CLIPImageProcessor(crop_size=32				,      size=32       )

            _A=	self.dummy_image_encoder

            return {
                "decoder": decoder,
                "text_encoder": text_encoder,
                "tokenizer": tokenizer,
                "text_proj": text_proj,
                "feature_extractor": feature_extractor,
                "image_encoder": image_encoder,
                "super_res_first": super_res_first,
                "super_res_last": super_res_last,
                "decoder_scheduler": decoder_scheduler,
                "super_res_scheduler": super_res_scheduler,
            }



     def 			a__  (					self				,      lowerCAmelCase__				,      lowerCAmelCase__=0				,      lowerCAmelCase__=True       ):
            _A=	floats_tensor((1, 3, 32, 32)				,      rng=random.Random(lowerCAmelCase__       )       ).to(lowerCAmelCase__       )
            if str(lowerCAmelCase__       ).startswith('mps'       ):
                   _A=	torch.manual_seed(lowerCAmelCase__       )
            else:
                   _A=	torch.Generator(device=lowerCAmelCase__       ).manual_seed(lowerCAmelCase__       )

            if pil_image:
                   _A=	input_image * 0.5 + 0.5
                   _A=	input_image.clamp(0				,      1       )
                   _A=	input_image.cpu().permute(0				,      2				,      3				,      1       ).float().numpy()
                   _A=	DiffusionPipeline.numpy_to_pil(lowerCAmelCase__       )[0]

            return {
                "image": input_image,
                "generator": generator,
                "decoder_num_inference_steps": 2,
                "super_res_num_inference_steps": 2,
                "output_type": "np",
            }



     def 			a__  (					self       ):
            _A=	'cpu'

            _A=	self.get_dummy_components()

            _A=	self.pipeline_class(**lowerCAmelCase__       )
            _A=	pipe.to(lowerCAmelCase__       )

            pipe.set_progress_bar_config(disable=lowerCAmelCase__       )

            _A=	self.get_dummy_inputs(lowerCAmelCase__				,      pil_image=lowerCAmelCase__       )

            _A=	pipe(**lowerCAmelCase__       )
            _A=	output.images

            _A=	self.get_dummy_inputs(lowerCAmelCase__				,      pil_image=lowerCAmelCase__       )

            _A=	pipe(
                **lowerCAmelCase__				,      return_dict=lowerCAmelCase__				,      )[0]

            _A=	image[0, -3:, -3:, -1]
            _A=	image_from_tuple[0, -3:, -3:, -1]

            assert image.shape == (1, 64, 64, 3)

            _A=	np.array(
                [
                    0.9997,
                    0.0002,
                    0.9997,
                    0.9997,
                    0.9969,
                    0.0023,
                    0.9997,
                    0.9969,
                    0.9970,
                ]       )

            assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-2
            assert np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1E-2



     def 			a__  (					self       ):
            _A=	'cpu'

            _A=	self.get_dummy_components()

            _A=	self.pipeline_class(**lowerCAmelCase__       )
            _A=	pipe.to(lowerCAmelCase__       )

            pipe.set_progress_bar_config(disable=lowerCAmelCase__       )

            _A=	self.get_dummy_inputs(lowerCAmelCase__				,      pil_image=lowerCAmelCase__       )

            _A=	pipe(**lowerCAmelCase__       )
            _A=	output.images

            _A=	self.get_dummy_inputs(lowerCAmelCase__				,      pil_image=lowerCAmelCase__       )

            _A=	pipe(
                **lowerCAmelCase__				,      return_dict=lowerCAmelCase__				,      )[0]

            _A=	image[0, -3:, -3:, -1]
            _A=	image_from_tuple[0, -3:, -3:, -1]

            assert image.shape == (1, 64, 64, 3)

            _A=	np.array([0.9997, 0.0003, 0.9997, 0.9997, 0.9970, 0.0024, 0.9997, 0.9971, 0.9971]       )

            assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-2
            assert np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1E-2



     def 			a__  (					self       ):
            _A=	'cpu'

            _A=	self.get_dummy_components()

            _A=	self.pipeline_class(**lowerCAmelCase__       )
            _A=	pipe.to(lowerCAmelCase__       )

            pipe.set_progress_bar_config(disable=lowerCAmelCase__       )

            _A=	self.get_dummy_inputs(lowerCAmelCase__				,      pil_image=lowerCAmelCase__       )
            _A=	[
                pipeline_inputs['image'],
                pipeline_inputs['image'],
            ]

            _A=	pipe(**lowerCAmelCase__       )
            _A=	output.images

            _A=	self.get_dummy_inputs(lowerCAmelCase__				,      pil_image=lowerCAmelCase__       )
            _A=	[
                tuple_pipeline_inputs['image'],
                tuple_pipeline_inputs['image'],
            ]

            _A=	pipe(
                **lowerCAmelCase__				,      return_dict=lowerCAmelCase__				,      )[0]

            _A=	image[0, -3:, -3:, -1]
            _A=	image_from_tuple[0, -3:, -3:, -1]

            assert image.shape == (2, 64, 64, 3)

            _A=	np.array(
                [
                    0.9997,
                    0.9989,
                    0.0008,
                    0.0021,
                    0.9960,
                    0.0018,
                    0.0014,
                    0.0002,
                    0.9933,
                ]       )

            assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-2
            assert np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1E-2



     def 			a__  (					self       ):
            _A=	torch.device('cpu'       )





            class      lowerCAmelCase						:
                 _SCREAMING_SNAKE_CASE     :     List[str]		=1


            _A=	self.get_dummy_components()

            _A=	self.pipeline_class(**lowerCAmelCase__       )
            _A=	pipe.to(lowerCAmelCase__       )

            pipe.set_progress_bar_config(disable=lowerCAmelCase__       )

            _A=	torch.Generator(device=lowerCAmelCase__       ).manual_seed(0       )
            _A=	pipe.decoder.dtype
            _A=	1

            _A=	(
                batch_size,
                pipe.decoder.config.in_channels,
                pipe.decoder.config.sample_size,
                pipe.decoder.config.sample_size,
            )
            _A=	pipe.prepare_latents(
                lowerCAmelCase__				,      dtype=lowerCAmelCase__				,      device=lowerCAmelCase__				,      generator=lowerCAmelCase__				,      latents=lowerCAmelCase__				,      scheduler=DummyScheduler()       )

            _A=	(
                batch_size,
                pipe.super_res_first.config.in_channels // 2,
                pipe.super_res_first.config.sample_size,
                pipe.super_res_first.config.sample_size,
            )
            _A=	pipe.prepare_latents(
                lowerCAmelCase__				,      dtype=lowerCAmelCase__				,      device=lowerCAmelCase__				,      generator=lowerCAmelCase__				,      latents=lowerCAmelCase__				,      scheduler=DummyScheduler()       )

            _A=	self.get_dummy_inputs(lowerCAmelCase__				,      pil_image=lowerCAmelCase__       )

            _A=	pipe(
                **lowerCAmelCase__				,      decoder_latents=lowerCAmelCase__				,      super_res_latents=lowerCAmelCase__       ).images

            _A=	self.get_dummy_inputs(lowerCAmelCase__				,      pil_image=lowerCAmelCase__       )
            # Don't pass image, instead pass embedding
            _A=	pipeline_inputs.pop('image'       )
            _A=	pipe.image_encoder(lowerCAmelCase__       ).image_embeds

            _A=	pipe(
                **lowerCAmelCase__				,      decoder_latents=lowerCAmelCase__				,      super_res_latents=lowerCAmelCase__				,      image_embeddings=lowerCAmelCase__				,      ).images

            # make sure passing text embeddings manually is identical
            assert np.abs(img_out_a - img_out_a       ).max() < 1E-4



     @skip_mps
     def 			a__  (					self       ):
            _A=	torch_device == 'cpu'

            # Check is relaxed because there is not a torch 2.0 sliced attention added kv processor
            _A=	1E-2

            self._test_attention_slicing_forward_pass(
                test_max_difference=lowerCAmelCase__				,      expected_max_diff=lowerCAmelCase__       )



     @skip_mps
     def 			a__  (					self       ):
            _A=	torch_device == 'cpu'
            _A=	True
            _A=	[
                'decoder_num_inference_steps',
                'super_res_num_inference_steps',
            ]

            self._test_inference_batch_single_identical(
                test_max_difference=lowerCAmelCase__				,      relax_max_difference=lowerCAmelCase__				,      additional_params_copy_to_batched_inputs=lowerCAmelCase__				,      )



     def 			a__  (					self       ):
            _A=	[
                'decoder_num_inference_steps',
                'super_res_num_inference_steps',
            ]

            if torch_device == "mps":
                   # TODO: MPS errors with larger batch sizes
                   _A=	[2, 3]
                   self._test_inference_batch_consistent(
                       batch_sizes=lowerCAmelCase__				,      additional_params_copy_to_batched_inputs=lowerCAmelCase__				,      )
            else:
                   self._test_inference_batch_consistent(
                       additional_params_copy_to_batched_inputs=lowerCAmelCase__       )



     @skip_mps
     def 			a__  (					self       ):
            return super().test_dict_tuple_outputs_equivalent()



     @skip_mps
     def 			a__  (					self       ):
            return super().test_save_load_local()





     @skip_mps
     def 			a__  (					self       ):
            return super().test_save_load_optional_components()





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



     def 			a__  (					self       ):
            # clean up the VRAM after each test
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()





     def 			a__  (					self       ):
            _A=	load_image(
                'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png'       )
            _A=	load_numpy(
                'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
                '/unclip/karlo_v1_alpha_cat_variation_fp16.npy'       )

            _A=	UnCLIPImageVariationPipeline.from_pretrained(
                'kakaobrain/karlo-v1-alpha-image-variations'				,      torch_dtype=torch.floataa       )
            _A=	pipeline.to(lowerCAmelCase__       )
            pipeline.set_progress_bar_config(disable=lowerCAmelCase__       )

            _A=	torch.Generator(device='cpu'       ).manual_seed(0       )
            _A=	pipeline(
                lowerCAmelCase__				,      generator=lowerCAmelCase__				,      output_type='np'				,      )

            _A=	output.images[0]

            assert image.shape == (256, 256, 3)

            assert_mean_pixel_difference(lowerCAmelCase__				,      lowerCAmelCase__				,      15       )