


import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, CycleDiffusionPipeline, DDIMScheduler, UNetaDConditionModel
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, skip_mps

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


enable_full_determinism()







class 						_snake_case    ( _snake_case		,       _snake_case		,       unittest.TestCase				):
 SCREAMING_SNAKE_CASE__ =			CycleDiffusionPipeline
 SCREAMING_SNAKE_CASE__ =			TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
     'negative_prompt',
     'height',
     'width',
     'negative_prompt_embeds',
 }
 SCREAMING_SNAKE_CASE__ =			PipelineTesterMixin.required_optional_params - {'latents'}
 SCREAMING_SNAKE_CASE__ =			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'source_prompt'}				)
 SCREAMING_SNAKE_CASE__ =			IMAGE_TO_IMAGE_IMAGE_PARAMS
 SCREAMING_SNAKE_CASE__ =			IMAGE_TO_IMAGE_IMAGE_PARAMS




 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       torch.manual_seed(0       )
       a		:str									=			UNetaDConditionModel(
           block_out_channels=(32, 64)   ,    layers_per_block=2   ,    sample_size=32   ,    in_channels=4   ,    out_channels=4   ,    down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')   ,    up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')   ,    cross_attention_dim=32   ,    )
       a		:Tuple									=			DDIMScheduler(
           beta_start=0.0_0085   ,    beta_end=0.012   ,    beta_schedule='''scaled_linear'''   ,    num_train_timesteps=1000   ,    clip_sample=_lowerCamelCase   ,    set_alpha_to_one=_lowerCamelCase   ,    )
       torch.manual_seed(0       )
       a		:Optional[int]									=			AutoencoderKL(
           block_out_channels=[32, 64]   ,    in_channels=3   ,    out_channels=3   ,    down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']   ,    up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']   ,    latent_channels=4   ,    )
       torch.manual_seed(0       )
       a		:List[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   ,    )
       a		:str									=			CLIPTextModel(_lowerCamelCase       )
       a		:Optional[int]									=			CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''       )

       a		:Optional[int]									=			{
           '''unet''': unet,
           '''scheduler''': scheduler,
           '''vae''': vae,
           '''text_encoder''': text_encoder,
           '''tokenizer''': tokenizer,
           '''safety_checker''': None,
           '''feature_extractor''': None,
       }
       return components




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase=0       ):
       a		:str									=			floats_tensor((1, 3, 32, 32)   ,    rng=random.Random(_lowerCamelCase       )       ).to(_lowerCamelCase       )
       a		:Optional[Any]									=			image / 2 + 0.5
       if str(_lowerCamelCase       ).startswith('''mps'''       ):
             a		:int									=			torch.manual_seed(_lowerCamelCase       )
       else:
             a		:Optional[Any]									=			torch.Generator(device=_lowerCamelCase       ).manual_seed(_lowerCamelCase       )
       a		:Optional[int]									=			{
           '''prompt''': '''An astronaut riding an elephant''',
           '''source_prompt''': '''An astronaut riding a horse''',
           '''image''': image,
           '''generator''': generator,
           '''num_inference_steps''': 2,
           '''eta''': 0.1,
           '''strength''': 0.8,
           '''guidance_scale''': 3,
           '''source_guidance_scale''': 1,
           '''output_type''': '''numpy''',
       }
       return inputs




 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       a		:Dict									=			'''cpu'''  # ensure determinism for the device-dependent torch.Generator

       a		:Tuple									=			self.get_dummy_components()
       a		:int									=			CycleDiffusionPipeline(**_lowerCamelCase       )
       a		:int									=			pipe.to(_lowerCamelCase       )
       pipe.set_progress_bar_config(disable=_lowerCamelCase       )

       a		:Any									=			self.get_dummy_inputs(_lowerCamelCase       )
       a		:int									=			pipe(**_lowerCamelCase       )
       a		:Optional[Any]									=			output.images

       a		:Union[str, Any]									=			images[0, -3:, -3:, -1]

       assert images.shape == (1, 32, 32, 3)
       a		:Tuple									=			np.array([0.4459, 0.4943, 0.4544, 0.6643, 0.5474, 0.4327, 0.5701, 0.5959, 0.5179]       )

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




 @unittest.skipIf(torch_device != '''cuda'''   ,    '''This test requires a GPU'''       )
 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       a		:str									=			self.get_dummy_components()
       for name, module in components.items():
             if hasattr(_lowerCamelCase   ,    '''half'''       ):
                   a		:List[Any]									=			module.half()
       a		:Optional[Any]									=			CycleDiffusionPipeline(**_lowerCamelCase       )
       a		:List[Any]									=			pipe.to(_lowerCamelCase       )
       pipe.set_progress_bar_config(disable=_lowerCamelCase       )

       a		:List[str]									=			self.get_dummy_inputs(_lowerCamelCase       )
       a		:Optional[int]									=			pipe(**_lowerCamelCase       )
       a		:List[Any]									=			output.images

       a		:Dict									=			images[0, -3:, -3:, -1]

       assert images.shape == (1, 32, 32, 3)
       a		:Optional[Any]									=			np.array([0.3506, 0.4543, 0.446, 0.4575, 0.5195, 0.4155, 0.5273, 0.518, 0.4116]       )

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




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




 @unittest.skip('''non-deterministic pipeline'''       )
 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       return super().test_inference_batch_single_identical()




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




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




 @skip_mps
 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       return super().test_attention_slicing_forward_pass()







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




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




 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       a		:Dict									=			load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
           '''/cycle-diffusion/black_colored_car.png'''       )
       a		:Any									=			load_numpy(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy'''       )
       a		:Optional[int]									=			init_image.resize((512, 512)       )

       a		:int									=			'''CompVis/stable-diffusion-v1-4'''
       a		:Any									=			DDIMScheduler.from_pretrained(_lowerCamelCase   ,    subfolder='''scheduler'''       )
       a		:Any									=			CycleDiffusionPipeline.from_pretrained(
           _lowerCamelCase   ,    scheduler=_lowerCamelCase   ,    safety_checker=_lowerCamelCase   ,    torch_dtype=torch.floataa   ,    revision='''fp16'''       )

       pipe.to(_lowerCamelCase       )
       pipe.set_progress_bar_config(disable=_lowerCamelCase       )
       pipe.enable_attention_slicing()

       a		:Any									=			'''A black colored car'''
       a		:Optional[int]									=			'''A blue colored car'''

       a		:str									=			torch.manual_seed(0       )
       a		:Any									=			pipe(
           prompt=_lowerCamelCase   ,    source_prompt=_lowerCamelCase   ,    image=_lowerCamelCase   ,    num_inference_steps=100   ,    eta=0.1   ,    strength=0.85   ,    guidance_scale=3   ,    source_guidance_scale=1   ,    generator=_lowerCamelCase   ,    output_type='''np'''   ,    )
       a		:Any									=			output.images

       # the values aren't exactly equal, but the images look the same visually
       assert np.abs(image - expected_image       ).max() < 5e-1




 def 			SCREAMING_SNAKE_CASE__							(						self       ):
       a		:Any									=			load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
           '''/cycle-diffusion/black_colored_car.png'''       )
       a		:Any									=			load_numpy(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy'''       )
       a		:Any									=			init_image.resize((512, 512)       )

       a		:Optional[int]									=			'''CompVis/stable-diffusion-v1-4'''
       a		:List[str]									=			DDIMScheduler.from_pretrained(_lowerCamelCase   ,    subfolder='''scheduler'''       )
       a		:List[str]									=			CycleDiffusionPipeline.from_pretrained(_lowerCamelCase   ,    scheduler=_lowerCamelCase   ,    safety_checker=_lowerCamelCase       )

       pipe.to(_lowerCamelCase       )
       pipe.set_progress_bar_config(disable=_lowerCamelCase       )
       pipe.enable_attention_slicing()

       a		:List[Any]									=			'''A black colored car'''
       a		:str									=			'''A blue colored car'''

       a		:str									=			torch.manual_seed(0       )
       a		:str									=			pipe(
           prompt=_lowerCamelCase   ,    source_prompt=_lowerCamelCase   ,    image=_lowerCamelCase   ,    num_inference_steps=100   ,    eta=0.1   ,    strength=0.85   ,    guidance_scale=3   ,    source_guidance_scale=1   ,    generator=_lowerCamelCase   ,    output_type='''np'''   ,    )
       a		:Any									=			output.images

       assert np.abs(image - expected_image       ).max() < 2e-2









from ...configuration_utils import PretrainedConfig
from ...utils import logging


snake_case							:       List[Any]										=	logging.get_logger(__name__)

snake_case							:       Dict										=	{
    '''uclanlp/visualbert-vqa''': '''https://huggingface.co/uclanlp/visualbert-vqa/resolve/main/config.json''',
    '''uclanlp/visualbert-vqa-pre''': '''https://huggingface.co/uclanlp/visualbert-vqa-pre/resolve/main/config.json''',
    '''uclanlp/visualbert-vqa-coco-pre''': (
        '''https://huggingface.co/uclanlp/visualbert-vqa-coco-pre/resolve/main/config.json'''
    ),
    '''uclanlp/visualbert-vcr''': '''https://huggingface.co/uclanlp/visualbert-vcr/resolve/main/config.json''',
    '''uclanlp/visualbert-vcr-pre''': '''https://huggingface.co/uclanlp/visualbert-vcr-pre/resolve/main/config.json''',
    '''uclanlp/visualbert-vcr-coco-pre''': (
        '''https://huggingface.co/uclanlp/visualbert-vcr-coco-pre/resolve/main/config.json'''
    ),
    '''uclanlp/visualbert-nlvr2''': '''https://huggingface.co/uclanlp/visualbert-nlvr2/resolve/main/config.json''',
    '''uclanlp/visualbert-nlvr2-pre''': '''https://huggingface.co/uclanlp/visualbert-nlvr2-pre/resolve/main/config.json''',
    '''uclanlp/visualbert-nlvr2-coco-pre''': (
        '''https://huggingface.co/uclanlp/visualbert-nlvr2-coco-pre/resolve/main/config.json'''
    )
    # See all VisualBERT models at https://huggingface.co/models?filter=visual_bert
}







class 						_snake_case    ( _snake_case				):
 SCREAMING_SNAKE_CASE__ =			'visual_bert'




 def __init__(						self   ,    _lowerCamelCase=3_0522   ,    _lowerCamelCase=768   ,    _lowerCamelCase=512   ,    _lowerCamelCase=12   ,    _lowerCamelCase=12   ,    _lowerCamelCase=3072   ,    _lowerCamelCase="gelu"   ,    _lowerCamelCase=0.1   ,    _lowerCamelCase=0.1   ,    _lowerCamelCase=512   ,    _lowerCamelCase=2   ,    _lowerCamelCase=0.02   ,    _lowerCamelCase=1e-12   ,    _lowerCamelCase=False   ,    _lowerCamelCase=True   ,    _lowerCamelCase=1   ,    _lowerCamelCase=0   ,    _lowerCamelCase=2   ,    **_lowerCamelCase   ,    ):
       super().__init__(pad_token_id=_lowerCamelCase   ,    bos_token_id=_lowerCamelCase   ,    eos_token_id=_lowerCamelCase   ,    **_lowerCamelCase       )

       a		:Any									=			vocab_size
       a		:str									=			max_position_embeddings
       a		:str									=			hidden_size
       a		:List[Any]									=			visual_embedding_dim
       a		:str									=			num_hidden_layers
       a		:Optional[int]									=			num_attention_heads
       a		:int									=			intermediate_size
       a		:int									=			hidden_act
       a		:Union[str, Any]									=			hidden_dropout_prob
       a		:List[Any]									=			attention_probs_dropout_prob
       a		:int									=			initializer_range
       a		:List[Any]									=			type_vocab_size
       a		:str									=			layer_norm_eps
       a		:Optional[int]									=			bypass_transformer
       a		:str									=			special_visual_initialize





