





import gc
import random
import unittest

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

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()

class 	UpperCamelCase__				(   lowerCAmelCase_							,   lowerCAmelCase_							,   lowerCAmelCase_							,   unittest.TestCase     ):



     '''simple docstring'''

     __snake_case					:			Union[str, Any]				=    StableUnCLIPImgaImgPipeline
     __snake_case					:			Union[str, Any]				=    TEXT_GUIDED_IMAGE_VARIATION_PARAMS
     __snake_case					:			Union[str, Any]				=    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
     __snake_case					:			Tuple				=    frozenset(
         []     )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
     __snake_case					:			Optional[Any]				=    frozenset([]     )






     def        SCREAMING_SNAKE_CASE__						(							self		:					str )  ->	int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		32
           SCREAMING_SNAKE_CASE	=		embedder_hidden_size

           # image encoding components

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

           torch.manual_seed(0 )
           SCREAMING_SNAKE_CASE	=		CLIPVisionModelWithProjection(
               CLIPVisionConfig(
                   hidden_size=lowerCamelCase__		,projection_dim=lowerCamelCase__		,num_hidden_layers=5		,num_attention_heads=4		,image_size=32		,intermediate_size=37		,patch_size=1		,) )

           # regular denoising components

           torch.manual_seed(0 )
           SCREAMING_SNAKE_CASE	=		StableUnCLIPImageNormalizer(embedding_dim=lowerCamelCase__ )
           SCREAMING_SNAKE_CASE	=		DDPMScheduler(beta_schedule="""squaredcos_cap_v2""" )

           torch.manual_seed(0 )
           SCREAMING_SNAKE_CASE	=		CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" )

           torch.manual_seed(0 )
           SCREAMING_SNAKE_CASE	=		CLIPTextModel(
               CLIPTextConfig(
                   bos_token_id=0		,eos_token_id=2		,hidden_size=lowerCamelCase__		,projection_dim=32		,intermediate_size=37		,layer_norm_eps=1e-0_5		,num_attention_heads=4		,num_hidden_layers=5		,pad_token_id=1		,vocab_size=1000		,) )

           torch.manual_seed(0 )
           SCREAMING_SNAKE_CASE	=		UNetaDConditionModel(
               sample_size=32		,in_channels=4		,out_channels=4		,down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")		,up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")		,block_out_channels=(32, 64)		,attention_head_dim=(2, 4)		,class_embed_type="""projection"""		,projection_class_embeddings_input_dim=embedder_projection_dim * 2		,cross_attention_dim=lowerCamelCase__		,layers_per_block=1		,upcast_attention=lowerCamelCase__		,use_linear_projection=lowerCamelCase__		,)

           torch.manual_seed(0 )
           SCREAMING_SNAKE_CASE	=		DDIMScheduler(
               beta_schedule="""scaled_linear"""		,beta_start=0.00085		,beta_end=0.012		,prediction_type="""v_prediction"""		,set_alpha_to_one=lowerCamelCase__		,steps_offset=1		,)

           torch.manual_seed(0 )
           SCREAMING_SNAKE_CASE	=		AutoencoderKL()

           SCREAMING_SNAKE_CASE	=		{
               # image encoding components
               """feature_extractor""": feature_extractor,
               """image_encoder""": image_encoder.eval(),
               # image noising components
               """image_normalizer""": image_normalizer.eval(),
               """image_noising_scheduler""": image_noising_scheduler,
               # regular denoising components
               """tokenizer""": tokenizer,
               """text_encoder""": text_encoder.eval(),
               """unet""": unet.eval(),
               """scheduler""": scheduler,
               """vae""": vae.eval(),
           }

           return components






     def        SCREAMING_SNAKE_CASE__						(							self		:					List[Any]		,lowerCamelCase__		:					Union[str, Any]		,lowerCamelCase__		:					Optional[Any]=0		,lowerCamelCase__		:					Any=True )  ->	Optional[Any]:




           '''simple docstring'''


           if str(lowerCamelCase__ ).startswith("""mps""" ):
                 SCREAMING_SNAKE_CASE	=		torch.manual_seed(lowerCamelCase__ )
           else:
                 SCREAMING_SNAKE_CASE	=		torch.Generator(device=lowerCamelCase__ ).manual_seed(lowerCamelCase__ )

           SCREAMING_SNAKE_CASE	=		floats_tensor((1, 3, 32, 32)		,rng=random.Random(lowerCamelCase__ ) ).to(lowerCamelCase__ )

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

           return {
               "prompt": "An anime racoon running a marathon",
               "image": input_image,
               "generator": generator,
               "num_inference_steps": 2,
               "output_type": "np",
           }






     @skip_mps
     def        SCREAMING_SNAKE_CASE__						(							self		:					Union[str, Any] )  ->	Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
           SCREAMING_SNAKE_CASE	=		self.get_dummy_components()
           SCREAMING_SNAKE_CASE	=		StableUnCLIPImgaImgPipeline(**lowerCamelCase__ )
           SCREAMING_SNAKE_CASE	=		sd_pipe.to(lowerCamelCase__ )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase__ )

           SCREAMING_SNAKE_CASE	=		self.get_dummy_inputs(lowerCamelCase__ )
           inputs.update({"""image_embeds""": None} )
           SCREAMING_SNAKE_CASE	=		sd_pipe(**lowerCamelCase__ ).images
           SCREAMING_SNAKE_CASE	=		image[0, -3:, -3:, -1]

           assert image.shape == (1, 32, 32, 3)
           SCREAMING_SNAKE_CASE	=		np.array([0.3872, 0.7224, 0.5601, 0.4741, 0.6872, 0.5814, 0.4636, 0.3867, 0.5078] )

           assert np.abs(image_slice.flatten() - expected_slice ).max() < 1e-3






     def        SCREAMING_SNAKE_CASE__						(							self		:					List[Any] )  ->	str:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		torch_device in ["""cpu""", """mps"""]

           self._test_attention_slicing_forward_pass(test_max_difference=lowerCamelCase__ )






     def        SCREAMING_SNAKE_CASE__						(							self		:					Tuple )  ->	str:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		torch_device in ["""cpu""", """mps"""]

           self._test_inference_batch_single_identical(test_max_difference=lowerCamelCase__ )






     @unittest.skipIf(
         torch_device != """cuda""" or not is_xformers_available()		,reason="""XFormers attention is only available with CUDA and `xformers` installed"""		,)
     def        SCREAMING_SNAKE_CASE__						(							self		:					Tuple )  ->	Dict:




           '''simple docstring'''


           self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowerCamelCase__ )

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



     '''simple docstring'''

     def        SCREAMING_SNAKE_CASE__						(							self		:					Optional[int] )  ->	Union[str, Any]:




           '''simple docstring'''


           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()






     def        SCREAMING_SNAKE_CASE__						(							self		:					Dict )  ->	Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		load_image(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png""" )

           SCREAMING_SNAKE_CASE	=		load_numpy(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy""" )

           SCREAMING_SNAKE_CASE	=		StableUnCLIPImgaImgPipeline.from_pretrained(
               """fusing/stable-unclip-2-1-l-img2img"""		,torch_dtype=torch.floataa )
           pipe.to(lowerCamelCase__ )
           pipe.set_progress_bar_config(disable=lowerCamelCase__ )
           # stable unclip will oom when integration tests are run on a V100,
           # so turn on memory savings
           pipe.enable_attention_slicing()
           pipe.enable_sequential_cpu_offload()

           SCREAMING_SNAKE_CASE	=		torch.Generator(device="""cpu""" ).manual_seed(0 )
           SCREAMING_SNAKE_CASE	=		pipe(lowerCamelCase__		,"""anime turle"""		,generator=lowerCamelCase__		,output_type="""np""" )

           SCREAMING_SNAKE_CASE	=		output.images[0]

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

           assert_mean_pixel_difference(lowerCamelCase__		,lowerCamelCase__ )






     def        SCREAMING_SNAKE_CASE__						(							self		:					Tuple )  ->	str:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		load_image(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png""" )

           SCREAMING_SNAKE_CASE	=		load_numpy(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy""" )

           SCREAMING_SNAKE_CASE	=		StableUnCLIPImgaImgPipeline.from_pretrained(
               """fusing/stable-unclip-2-1-h-img2img"""		,torch_dtype=torch.floataa )
           pipe.to(lowerCamelCase__ )
           pipe.set_progress_bar_config(disable=lowerCamelCase__ )
           # stable unclip will oom when integration tests are run on a V100,
           # so turn on memory savings
           pipe.enable_attention_slicing()
           pipe.enable_sequential_cpu_offload()

           SCREAMING_SNAKE_CASE	=		torch.Generator(device="""cpu""" ).manual_seed(0 )
           SCREAMING_SNAKE_CASE	=		pipe(lowerCamelCase__		,"""anime turle"""		,generator=lowerCamelCase__		,output_type="""np""" )

           SCREAMING_SNAKE_CASE	=		output.images[0]

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

           assert_mean_pixel_difference(lowerCamelCase__		,lowerCamelCase__ )






     def        SCREAMING_SNAKE_CASE__						(							self		:					int )  ->	List[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		load_image(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png""" )

           torch.cuda.empty_cache()
           torch.cuda.reset_max_memory_allocated()
           torch.cuda.reset_peak_memory_stats()

           SCREAMING_SNAKE_CASE	=		StableUnCLIPImgaImgPipeline.from_pretrained(
               """fusing/stable-unclip-2-1-h-img2img"""		,torch_dtype=torch.floataa )
           SCREAMING_SNAKE_CASE	=		pipe.to(lowerCamelCase__ )
           pipe.set_progress_bar_config(disable=lowerCamelCase__ )
           pipe.enable_attention_slicing()
           pipe.enable_sequential_cpu_offload()

           SCREAMING_SNAKE_CASE	=		pipe(
               lowerCamelCase__		,"""anime turtle"""		,num_inference_steps=2		,output_type="""np"""		,)

           SCREAMING_SNAKE_CASE	=		torch.cuda.max_memory_allocated()
           # make sure that less than 7 GB is allocated
           assert mem_bytes < 7 * 10**9









import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto import CONFIG_MAPPING


SCREAMING_SNAKE_CASE_		     = logging.get_logger(__name__)

SCREAMING_SNAKE_CASE_		     = {
    """ut/deta""": """https://huggingface.co/ut/deta/resolve/main/config.json""",
}

class 	UpperCamelCase__				(   lowerCAmelCase_     ):



     '''simple docstring'''

     __snake_case					:			str				=    "deta"
     __snake_case					:			int				=    {
         "hidden_size": "d_model",
         "num_attention_heads": "encoder_attention_heads",
     }






     def __init__(							self		:					List[Any]		,lowerCamelCase__		:					Any=None		,lowerCamelCase__		:					Optional[int]=900		,lowerCamelCase__		:					Optional[Any]=2048		,lowerCamelCase__		:					Optional[Any]=6		,lowerCamelCase__		:					Dict=2048		,lowerCamelCase__		:					Optional[int]=8		,lowerCamelCase__		:					List[str]=6		,lowerCamelCase__		:					Union[str, Any]=1024		,lowerCamelCase__		:					str=8		,lowerCamelCase__		:					Union[str, Any]=0.0		,lowerCamelCase__		:					Dict=True		,lowerCamelCase__		:					Dict="relu"		,lowerCamelCase__		:					int=256		,lowerCamelCase__		:					Optional[Any]=0.1		,lowerCamelCase__		:					Dict=0.0		,lowerCamelCase__		:					Dict=0.0		,lowerCamelCase__		:					Dict=0.02		,lowerCamelCase__		:					List[Any]=1.0		,lowerCamelCase__		:					List[str]=True		,lowerCamelCase__		:					int=False		,lowerCamelCase__		:					int="sine"		,lowerCamelCase__		:					Any=5		,lowerCamelCase__		:					Optional[int]=4		,lowerCamelCase__		:					Optional[int]=4		,lowerCamelCase__		:					List[Any]=True		,lowerCamelCase__		:					Optional[int]=300		,lowerCamelCase__		:					Optional[Any]=True		,lowerCamelCase__		:					Optional[Any]=True		,lowerCamelCase__		:					Optional[int]=1		,lowerCamelCase__		:					Tuple=5		,lowerCamelCase__		:					Tuple=2		,lowerCamelCase__		:					Dict=1		,lowerCamelCase__		:					str=1		,lowerCamelCase__		:					List[Any]=5		,lowerCamelCase__		:					Tuple=2		,lowerCamelCase__		:					Union[str, Any]=0.1		,lowerCamelCase__		:					List[str]=0.25		,**lowerCamelCase__		:					int		,)  ->	List[Any]:




           '''simple docstring'''


           if backbone_config is None:
                 logger.info("""`backbone_config` is `None`. Initializing the config with the default `ResNet` backbone.""" )
                 SCREAMING_SNAKE_CASE	=		CONFIG_MAPPING["""resnet"""](out_features=["""stage2""", """stage3""", """stage4"""] )
           else:
                 if isinstance(lowerCamelCase__		,lowerCamelCase__ ):
                       SCREAMING_SNAKE_CASE	=		backbone_config.pop("""model_type""" )
                       SCREAMING_SNAKE_CASE	=		CONFIG_MAPPING[backbone_model_type]
                       SCREAMING_SNAKE_CASE	=		config_class.from_dict(lowerCamelCase__ )

           SCREAMING_SNAKE_CASE	=		backbone_config
           SCREAMING_SNAKE_CASE	=		num_queries
           SCREAMING_SNAKE_CASE	=		max_position_embeddings
           SCREAMING_SNAKE_CASE	=		d_model
           SCREAMING_SNAKE_CASE	=		encoder_ffn_dim
           SCREAMING_SNAKE_CASE	=		encoder_layers
           SCREAMING_SNAKE_CASE	=		encoder_attention_heads
           SCREAMING_SNAKE_CASE	=		decoder_ffn_dim
           SCREAMING_SNAKE_CASE	=		decoder_layers
           SCREAMING_SNAKE_CASE	=		decoder_attention_heads
           SCREAMING_SNAKE_CASE	=		dropout
           SCREAMING_SNAKE_CASE	=		attention_dropout
           SCREAMING_SNAKE_CASE	=		activation_dropout
           SCREAMING_SNAKE_CASE	=		activation_function
           SCREAMING_SNAKE_CASE	=		init_std
           SCREAMING_SNAKE_CASE	=		init_xavier_std
           SCREAMING_SNAKE_CASE	=		encoder_layerdrop
           SCREAMING_SNAKE_CASE	=		auxiliary_loss
           SCREAMING_SNAKE_CASE	=		position_embedding_type
           # deformable attributes
           SCREAMING_SNAKE_CASE	=		num_feature_levels
           SCREAMING_SNAKE_CASE	=		encoder_n_points
           SCREAMING_SNAKE_CASE	=		decoder_n_points
           SCREAMING_SNAKE_CASE	=		two_stage
           SCREAMING_SNAKE_CASE	=		two_stage_num_proposals
           SCREAMING_SNAKE_CASE	=		with_box_refine
           SCREAMING_SNAKE_CASE	=		assign_first_stage
           if two_stage is True and with_box_refine is False:
                 raise ValueError("""If two_stage is True, with_box_refine must be True.""" )
           # Hungarian matcher
           SCREAMING_SNAKE_CASE	=		class_cost
           SCREAMING_SNAKE_CASE	=		bbox_cost
           SCREAMING_SNAKE_CASE	=		giou_cost
           # Loss coefficients
           SCREAMING_SNAKE_CASE	=		mask_loss_coefficient
           SCREAMING_SNAKE_CASE	=		dice_loss_coefficient
           SCREAMING_SNAKE_CASE	=		bbox_loss_coefficient
           SCREAMING_SNAKE_CASE	=		giou_loss_coefficient
           SCREAMING_SNAKE_CASE	=		eos_coefficient
           SCREAMING_SNAKE_CASE	=		focal_alpha
           super().__init__(is_encoder_decoder=lowerCamelCase__		,**lowerCamelCase__ )






     @property
     def        SCREAMING_SNAKE_CASE__						(							self		:					str )  ->	int:




           '''simple docstring'''


           return self.encoder_attention_heads






     @property
     def        SCREAMING_SNAKE_CASE__						(							self		:					List[str] )  ->	int:




           '''simple docstring'''


           return self.d_model






     def        SCREAMING_SNAKE_CASE__						(							self		:					Optional[Any] )  ->	Optional[int]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		copy.deepcopy(self.__dict__ )
           SCREAMING_SNAKE_CASE	=		self.backbone_config.to_dict()
           SCREAMING_SNAKE_CASE	=		self.__class__.model_type
           return output


