


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 							lowerCamelCase__						(				__SCREAMING_SNAKE_CASE       ,						__SCREAMING_SNAKE_CASE       ,						unittest.TestCase   ):






			"""simple docstring"""






			_UpperCamelCase		:       Any 			=		CycleDiffusionPipeline
			_UpperCamelCase		:       Tuple 			=		TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
			    'negative_prompt',
			    'height',
			    'width',
			    'negative_prompt_embeds',
			}
			_UpperCamelCase		:       int 			=		PipelineTesterMixin.required_optional_params - {'latents'}
			_UpperCamelCase		:       List[Any] 			=		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'source_prompt'}   )
			_UpperCamelCase		:       Optional[int] 			=		IMAGE_TO_IMAGE_IMAGE_PARAMS
			_UpperCamelCase		:       Optional[Any] 			=		IMAGE_TO_IMAGE_IMAGE_PARAMS




			def 					snake_case__					(				self  ):

						'''simple docstring'''


						torch.manual_seed(0  )
						UpperCamelCase__       =							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   ,    )
						UpperCamelCase__       =							DDIMScheduler(
						    beta_start=0.00085   ,    beta_end=0.012   ,    beta_schedule="scaled_linear"   ,    num_train_timesteps=1000   ,    clip_sample=lowerCAmelCase__   ,    set_alpha_to_one=lowerCAmelCase__   ,    )
						torch.manual_seed(0  )
						UpperCamelCase__       =							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  )
						UpperCamelCase__       =							CLIPTextConfig(
						    bos_token_id=0   ,    eos_token_id=2   ,    hidden_size=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   ,    )
						UpperCamelCase__       =							CLIPTextModel(lowerCAmelCase__  )
						UpperCamelCase__       =							CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"  )

						UpperCamelCase__       =							{
						    'unet': unet,
						    'scheduler': scheduler,
						    'vae': vae,
						    'text_encoder': text_encoder,
						    'tokenizer': tokenizer,
						    'safety_checker': None,
						    'feature_extractor': None,
						}
						return components




			def 					snake_case__					(				self   ,    snake_case   ,    snake_case=0  ):

						'''simple docstring'''


						UpperCamelCase__       =							floats_tensor((1, 3, 32, 32)   ,    rng=random.Random(lowerCAmelCase__  )  ).to(lowerCAmelCase__  )
						UpperCamelCase__       =							image / 2 + 0.5
						if str(lowerCAmelCase__  ).startswith("mps"  ):
									UpperCamelCase__       =							torch.manual_seed(lowerCAmelCase__  )
						else:
									UpperCamelCase__       =							torch.Generator(device=lowerCAmelCase__  ).manual_seed(lowerCAmelCase__  )
						UpperCamelCase__       =							{
						    '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 					snake_case__					(				self  ):

						'''simple docstring'''


						UpperCamelCase__       =							'cpu'  # ensure determinism for the device-dependent torch.Generator

						UpperCamelCase__       =							self.get_dummy_components()
						UpperCamelCase__       =							CycleDiffusionPipeline(**lowerCAmelCase__  )
						UpperCamelCase__       =							pipe.to(lowerCAmelCase__  )
						pipe.set_progress_bar_config(disable=lowerCAmelCase__  )

						UpperCamelCase__       =							self.get_dummy_inputs(lowerCAmelCase__  )
						UpperCamelCase__       =							pipe(**lowerCAmelCase__  )
						UpperCamelCase__       =							output.images

						UpperCamelCase__       =							images[0, -3:, -3:, -1]

						assert images.shape == (1, 32, 32, 3)
						UpperCamelCase__       =							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 					snake_case__					(				self  ):

						'''simple docstring'''


						UpperCamelCase__       =							self.get_dummy_components()
						for name, module in components.items():
									if hasattr(lowerCAmelCase__   ,    "half"  ):
												UpperCamelCase__       =							module.half()
						UpperCamelCase__       =							CycleDiffusionPipeline(**lowerCAmelCase__  )
						UpperCamelCase__       =							pipe.to(lowerCAmelCase__  )
						pipe.set_progress_bar_config(disable=lowerCAmelCase__  )

						UpperCamelCase__       =							self.get_dummy_inputs(lowerCAmelCase__  )
						UpperCamelCase__       =							pipe(**lowerCAmelCase__  )
						UpperCamelCase__       =							output.images

						UpperCamelCase__       =							images[0, -3:, -3:, -1]

						assert images.shape == (1, 32, 32, 3)
						UpperCamelCase__       =							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 					snake_case__					(				self  ):

						'''simple docstring'''


						return super().test_save_load_local()




			@unittest.skip("non-deterministic pipeline"  )
			def 					snake_case__					(				self  ):

						'''simple docstring'''


						return super().test_inference_batch_single_identical()




			@skip_mps
			def 					snake_case__					(				self  ):

						'''simple docstring'''


						return super().test_dict_tuple_outputs_equivalent()




			@skip_mps
			def 					snake_case__					(				self  ):

						'''simple docstring'''


						return super().test_save_load_optional_components()







			@skip_mps
			def 					snake_case__					(				self  ):

						'''simple docstring'''


						return super().test_attention_slicing_forward_pass()







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






			"""simple docstring"""






			def 					snake_case__					(				self  ):

						'''simple docstring'''


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




			def 					snake_case__					(				self  ):

						'''simple docstring'''


						UpperCamelCase__       =							load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/cycle-diffusion/black_colored_car.png"  )
						UpperCamelCase__       =							load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy"  )
						UpperCamelCase__       =							init_image.resize((512, 512)  )

						UpperCamelCase__       =							'CompVis/stable-diffusion-v1-4'
						UpperCamelCase__       =							DDIMScheduler.from_pretrained(lowerCAmelCase__   ,    subfolder="scheduler"  )
						UpperCamelCase__       =							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()

						UpperCamelCase__       =							'A black colored car'
						UpperCamelCase__       =							'A blue colored car'

						UpperCamelCase__       =							torch.manual_seed(0  )
						UpperCamelCase__       =							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"   ,    )
						UpperCamelCase__       =							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 					snake_case__					(				self  ):

						'''simple docstring'''


						UpperCamelCase__       =							load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/cycle-diffusion/black_colored_car.png"  )
						UpperCamelCase__       =							load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"  )
						UpperCamelCase__       =							init_image.resize((512, 512)  )

						UpperCamelCase__       =							'CompVis/stable-diffusion-v1-4'
						UpperCamelCase__       =							DDIMScheduler.from_pretrained(lowerCAmelCase__   ,    subfolder="scheduler"  )
						UpperCamelCase__       =							CycleDiffusionPipeline.from_pretrained(lowerCAmelCase__   ,    scheduler=lowerCAmelCase__   ,    safety_checker=lowerCAmelCase__  )

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

						UpperCamelCase__       =							'A black colored car'
						UpperCamelCase__       =							'A blue colored car'

						UpperCamelCase__       =							torch.manual_seed(0  )
						UpperCamelCase__       =							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"   ,    )
						UpperCamelCase__       =							output.images

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






from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowerCAmelCase__			:      Union[str, Any]								={
    'configuration_blip_2': [
        'BLIP_2_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'Blip2Config',
        'Blip2QFormerConfig',
        'Blip2VisionConfig',
    ],
    'processing_blip_2': ['Blip2Processor'],
}

try:
				if not is_torch_available():
								raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
				pass
else:
				lowerCAmelCase__			:      Dict								=[
				    'BLIP_2_PRETRAINED_MODEL_ARCHIVE_LIST',
				    'Blip2Model',
				    'Blip2QFormerModel',
				    'Blip2PreTrainedModel',
				    'Blip2ForConditionalGeneration',
				    'Blip2VisionModel',
				]

if TYPE_CHECKING:
				from .configuration_blip_a import (
				    BLIP_2_PRETRAINED_CONFIG_ARCHIVE_MAP,
				    BlipaConfig,
				    BlipaQFormerConfig,
				    BlipaVisionConfig,
				)
				from .processing_blip_a import BlipaProcessor

				try:
								if not is_torch_available():
												raise OptionalDependencyNotAvailable()
				except OptionalDependencyNotAvailable:
								pass
				else:
								from .modeling_blip_a import (
								    BLIP_2_PRETRAINED_MODEL_ARCHIVE_LIST,
								    BlipaForConditionalGeneration,
								    BlipaModel,
								    BlipaPreTrainedModel,
								    BlipaQFormerModel,
								    BlipaVisionModel,
								)

else:
				import sys

				lowerCAmelCase__			:      Union[str, Any]								=_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)





