



'''simple docstring'''




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 			UpperCamelCase_ (							__magic_name__      ,     __magic_name__      ,     unittest.TestCase			):
	lowercase          = CycleDiffusionPipeline
	lowercase          = TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
	    'negative_prompt',
	    'height',
	    'width',
	    'negative_prompt_embeds',
	}
	lowercase          = PipelineTesterMixin.required_optional_params - {'latents'}
	lowercase          = TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'source_prompt'}			)
	lowercase          = IMAGE_TO_IMAGE_IMAGE_PARAMS
	lowercase          = IMAGE_TO_IMAGE_IMAGE_PARAMS





	def  _lowercase(  self    )			->			Tuple:
								torch.manual_seed(0    )
								UpperCAmelCase     :		List[Any]							     =		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     :		Dict							     =		DDIMScheduler(
								    beta_start=0.0_0_0_8_5		,		beta_end=0.0_1_2		,		beta_schedule="""scaled_linear"""		,		num_train_timesteps=1000		,		clip_sample=A		,		set_alpha_to_one=A		,		)
								torch.manual_seed(0    )
								UpperCAmelCase     :		str							     =		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     :		Tuple							     =		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		,		)
								UpperCAmelCase     :		List[Any]							     =		CLIPTextModel(A    )
								UpperCAmelCase     :		Optional[Any]							     =		CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )

								UpperCAmelCase     :		Dict							     =		{
								    """unet""": unet,
								    """scheduler""": scheduler,
								    """vae""": vae,
								    """text_encoder""": text_encoder,
								    """tokenizer""": tokenizer,
								    """safety_checker""": None,
								    """feature_extractor""": None,
								}
								return components





	def  _lowercase(  self		,		A		,		A=0    )			->			Optional[int]:
								UpperCAmelCase     :		Union[str, Any]							     =		floats_tensor((1, 3, 32, 32)		,		rng=random.Random(A    )    ).to(A    )
								UpperCAmelCase     :		Dict							     =		image / 2 + 0.5
								if str(A    ).startswith("""mps"""    ):
															UpperCAmelCase     :		Optional[int]							     =		torch.manual_seed(A    )
								else:
															UpperCAmelCase     :		Union[str, Any]							     =		torch.Generator(device=A    ).manual_seed(A    )
								UpperCAmelCase     :		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  _lowercase(  self    )			->			str:
								UpperCAmelCase     :		List[str]							     =		"""cpu"""  # ensure determinism for the device-dependent torch.Generator

								UpperCAmelCase     :		Dict							     =		self.get_dummy_components()
								UpperCAmelCase     :		Dict							     =		CycleDiffusionPipeline(**A    )
								UpperCAmelCase     :		Optional[Any]							     =		pipe.to(A    )
								pipe.set_progress_bar_config(disable=A    )

								UpperCAmelCase     :		Optional[Any]							     =		self.get_dummy_inputs(A    )
								UpperCAmelCase     :		Dict							     =		pipe(**A    )
								UpperCAmelCase     :		int							     =		output.images

								UpperCAmelCase     :		Tuple							     =		images[0, -3:, -3:, -1]

								assert images.shape == (1, 32, 32, 3)
								UpperCAmelCase     :		str							     =		np.array([0.4_4_5_9, 0.4_9_4_3, 0.4_5_4_4, 0.6_6_4_3, 0.5_4_7_4, 0.4_3_2_7, 0.5_7_0_1, 0.5_9_5_9, 0.5_1_7_9]    )

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





	@unittest.skipIf(torch_device != """cuda"""		,		"""This test requires a GPU"""    )
	def  _lowercase(  self    )			->			Tuple:
								UpperCAmelCase     :		Any							     =		self.get_dummy_components()
								for name, module in components.items():
															if hasattr(A		,		"""half"""    ):
																						UpperCAmelCase     :		List[str]							     =		module.half()
								UpperCAmelCase     :		Optional[int]							     =		CycleDiffusionPipeline(**A    )
								UpperCAmelCase     :		str							     =		pipe.to(A    )
								pipe.set_progress_bar_config(disable=A    )

								UpperCAmelCase     :		List[str]							     =		self.get_dummy_inputs(A    )
								UpperCAmelCase     :		str							     =		pipe(**A    )
								UpperCAmelCase     :		Tuple							     =		output.images

								UpperCAmelCase     :		Optional[int]							     =		images[0, -3:, -3:, -1]

								assert images.shape == (1, 32, 32, 3)
								UpperCAmelCase     :		Any							     =		np.array([0.3_5_0_6, 0.4_5_4_3, 0.4_4_6, 0.4_5_7_5, 0.5_1_9_5, 0.4_1_5_5, 0.5_2_7_3, 0.5_1_8, 0.4_1_1_6]    )

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





	@skip_mps
	def  _lowercase(  self    )			->			int:
								return super().test_save_load_local()





	@unittest.skip("""non-deterministic pipeline"""    )
	def  _lowercase(  self    )			->			Dict:
								return super().test_inference_batch_single_identical()





	@skip_mps
	def  _lowercase(  self    )			->			Dict:
								return super().test_dict_tuple_outputs_equivalent()





	@skip_mps
	def  _lowercase(  self    )			->			Optional[int]:
								return super().test_save_load_optional_components()






	@skip_mps
	def  _lowercase(  self    )			->			Tuple:
								return super().test_attention_slicing_forward_pass()




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





	def  _lowercase(  self    )			->			Union[str, Any]:
								# clean up the VRAM after each test
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()





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

								UpperCAmelCase     :		Any							     =		"""CompVis/stable-diffusion-v1-4"""
								UpperCAmelCase     :		Optional[Any]							     =		DDIMScheduler.from_pretrained(A		,		subfolder="""scheduler"""    )
								UpperCAmelCase     :		Optional[int]							     =		CycleDiffusionPipeline.from_pretrained(
								    A		,		scheduler=A		,		safety_checker=A		,		torch_dtype=torch.floataa		,		revision="""fp16"""    )

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

								UpperCAmelCase     :		List[Any]							     =		"""A black colored car"""
								UpperCAmelCase     :		Any							     =		"""A blue colored car"""

								UpperCAmelCase     :		int							     =		torch.manual_seed(0    )
								UpperCAmelCase     :		Dict							     =		pipe(
								    prompt=A		,		source_prompt=A		,		image=A		,		num_inference_steps=100		,		eta=0.1		,		strength=0.8_5		,		guidance_scale=3		,		source_guidance_scale=1		,		generator=A		,		output_type="""np"""		,		)
								UpperCAmelCase     :		List[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  _lowercase(  self    )			->			int:
								UpperCAmelCase     :		Union[str, Any]							     =		load_image(
								    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
								    """/cycle-diffusion/black_colored_car.png"""    )
								UpperCAmelCase     :		Dict							     =		load_numpy(
								    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"""    )
								UpperCAmelCase     :		Tuple							     =		init_image.resize((512, 512)    )

								UpperCAmelCase     :		Tuple							     =		"""CompVis/stable-diffusion-v1-4"""
								UpperCAmelCase     :		Union[str, Any]							     =		DDIMScheduler.from_pretrained(A		,		subfolder="""scheduler"""    )
								UpperCAmelCase     :		Any							     =		CycleDiffusionPipeline.from_pretrained(A		,		scheduler=A		,		safety_checker=A    )

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

								UpperCAmelCase     :		Dict							     =		"""A black colored car"""
								UpperCAmelCase     :		List[str]							     =		"""A blue colored car"""

								UpperCAmelCase     :		List[Any]							     =		torch.manual_seed(0    )
								UpperCAmelCase     :		Union[str, Any]							     =		pipe(
								    prompt=A		,		source_prompt=A		,		image=A		,		num_inference_steps=100		,		eta=0.1		,		strength=0.8_5		,		guidance_scale=3		,		source_guidance_scale=1		,		generator=A		,		output_type="""np"""		,		)
								UpperCAmelCase     :		int							     =		output.images

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






'''simple docstring'''




from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
							from PIL import Image

							from ..image_utils import load_image

if is_torch_available():
							from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
							from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
							from ..tf_utils import stable_softmax

a   :			Union[str, Any]					  =    logging.get_logger(__name__)

@add_end_docstrings(__magic_name__			)
class 			UpperCamelCase_ (							__magic_name__			):





	def __init__(  self		,		**A    )			->			List[str]:
								super().__init__(**A    )

								requires_backends(self		,		"""vision"""    )
								self.check_model_type(
								    TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
								    if self.framework == """tf"""
								    else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING    )





	def __call__(  self		,		A		,		**A    )			->			Optional[Any]:
								return super().__call__(A		,		**A    )





	def  _lowercase(  self		,		**A    )			->			Optional[Any]:
								UpperCAmelCase     :		List[Any]							     =		{}
								if "candidate_labels" in kwargs:
															UpperCAmelCase     :		Dict							     =		kwargs["""candidate_labels"""]
								if "hypothesis_template" in kwargs:
															UpperCAmelCase     :		Optional[Any]							     =		kwargs["""hypothesis_template"""]

								return preprocess_params, {}, {}





	def  _lowercase(  self		,		A		,		A=None		,		A="This is a photo of {}."    )			->			Optional[Any]:
								UpperCAmelCase     :		int							     =		load_image(A    )
								UpperCAmelCase     :		List[str]							     =		self.image_processor(images=[image]		,		return_tensors=self.framework    )
								UpperCAmelCase     :		List[str]							     =		candidate_labels
								UpperCAmelCase     :		Tuple							     =		[hypothesis_template.format(A    ) for x in candidate_labels]
								UpperCAmelCase     :		Union[str, Any]							     =		self.tokenizer(A		,		return_tensors=self.framework		,		padding=A    )
								UpperCAmelCase     :		Union[str, Any]							     =		[text_inputs]
								return inputs





	def  _lowercase(  self		,		A    )			->			Optional[int]:
								UpperCAmelCase     :		List[Any]							     =		model_inputs.pop("""candidate_labels"""    )
								UpperCAmelCase     :		Optional[Any]							     =		model_inputs.pop("""text_inputs"""    )
								if isinstance(text_inputs[0]		,		A    ):
															UpperCAmelCase     :		Optional[Any]							     =		text_inputs[0]
								else:
															# Batching case.
															UpperCAmelCase     :		Any							     =		text_inputs[0][0]

								UpperCAmelCase     :		Dict							     =		self.model(**A		,		**A    )

								UpperCAmelCase     :		List[Any]							     =		{
								    """candidate_labels""": candidate_labels,
								    """logits""": outputs.logits_per_image,
								}
								return model_outputs






	def  _lowercase(  self		,		A    )			->			Union[str, Any]:
								UpperCAmelCase     :		int							     =		model_outputs.pop("""candidate_labels"""    )
								UpperCAmelCase     :		int							     =		model_outputs["""logits"""][0]
								if self.framework == "pt":
															UpperCAmelCase     :		Optional[int]							     =		logits.softmax(dim=-1    ).squeeze(-1    )
															UpperCAmelCase     :		Any							     =		probs.tolist()
															if not isinstance(A		,		A    ):
																						UpperCAmelCase     :		Any							     =		[scores]
								elif self.framework == "tf":
															UpperCAmelCase     :		List[str]							     =		stable_softmax(A		,		axis=-1    )
															UpperCAmelCase     :		Union[str, Any]							     =		probs.numpy().tolist()
								else:
															raise ValueError(f'''Unsupported framework: {self.framework}'''    )

								UpperCAmelCase     :		Any							     =		[
								    {"""score""": score, """label""": candidate_label}
								    for score, candidate_label in sorted(zip(A		,		A    )		,		key=lambda A    : -x[0]    )
								]
								return result

