


"""simple docstring"""







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_	( UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase			,  unittest.TestCase):

	"""simple docstring"""





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




	def   UpperCAmelCase_      ( self			:				Tuple   )       ->	Dict:
						__SCREAMING_SNAKE_CASE						   =				3_2
						__SCREAMING_SNAKE_CASE						   =				embedder_hidden_size

						# image encoding components

						__SCREAMING_SNAKE_CASE						   =				CLIPImageProcessor(crop_size=3_2			,			size=3_2   )

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				CLIPVisionModelWithProjection(
						    CLIPVisionConfig(
						        hidden_size=UpperCAmelCase__			,			projection_dim=UpperCAmelCase__			,			num_hidden_layers=5			,			num_attention_heads=4			,			image_size=3_2			,			intermediate_size=3_7			,			patch_size=1			,			)   )

						# regular denoising components

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				StableUnCLIPImageNormalizer(embedding_dim=UpperCAmelCase__   )
						__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=UpperCAmelCase__			,			projection_dim=3_2			,			intermediate_size=3_7			,			layer_norm_eps=1E-05			,			num_attention_heads=4			,			num_hidden_layers=5			,			pad_token_id=1			,			vocab_size=1_0_0_0			,			)   )

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				UNetaDConditionModel(
						    sample_size=3_2			,			in_channels=4			,			out_channels=4			,			down_block_types=("CrossAttnDownBlock2D", "DownBlock2D")			,			up_block_types=("UpBlock2D", "CrossAttnUpBlock2D")			,			block_out_channels=(3_2, 6_4)			,			attention_head_dim=(2, 4)			,			class_embed_type="projection"			,			projection_class_embeddings_input_dim=embedder_projection_dim * 2			,			cross_attention_dim=UpperCAmelCase__			,			layers_per_block=1			,			upcast_attention=UpperCAmelCase__			,			use_linear_projection=UpperCAmelCase__			,			)

						torch.manual_seed(0   )
						__SCREAMING_SNAKE_CASE						   =				DDIMScheduler(
						    beta_schedule="scaled_linear"			,			beta_start=0.00_085			,			beta_end=0.012			,			prediction_type="v_prediction"			,			set_alpha_to_one=UpperCAmelCase__			,			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   UpperCAmelCase_      ( self			:				str			,			UpperCAmelCase__			:				Any			,			UpperCAmelCase__			:				int=0			,			UpperCAmelCase__			:				Union[str, Any]=True   )       ->	Optional[int]:
						if str(UpperCAmelCase__   ).startswith("mps"   ):
											__SCREAMING_SNAKE_CASE						   =				torch.manual_seed(UpperCAmelCase__   )
						else:
											__SCREAMING_SNAKE_CASE						   =				torch.Generator(device=UpperCAmelCase__   ).manual_seed(UpperCAmelCase__   )

						__SCREAMING_SNAKE_CASE						   =				floats_tensor((1, 3, 3_2, 3_2)			,			rng=random.Random(UpperCAmelCase__   )   ).to(UpperCAmelCase__   )

						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(UpperCAmelCase__   )[0]

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




	@skip_mps
	def   UpperCAmelCase_      ( self			:				Optional[int]   )       ->	List[str]:
						__SCREAMING_SNAKE_CASE						   =				"cpu"  # ensure determinism for the device-dependent torch.Generator
						__SCREAMING_SNAKE_CASE						   =				self.get_dummy_components()
						__SCREAMING_SNAKE_CASE						   =				StableUnCLIPImgaImgPipeline(**UpperCAmelCase__   )
						__SCREAMING_SNAKE_CASE						   =				sd_pipe.to(UpperCAmelCase__   )
						sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__   )

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

						assert image.shape == (1, 3_2, 3_2, 3)
						__SCREAMING_SNAKE_CASE						   =				np.array([0.3_872, 0.7_224, 0.5_601, 0.4_741, 0.6_872, 0.5_814, 0.4_636, 0.3_867, 0.5_078]   )

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




	def   UpperCAmelCase_      ( self			:				Optional[int]   )       ->	Union[str, Any]:
						__SCREAMING_SNAKE_CASE						   =				torch_device in ["cpu", "mps"]

						self._test_attention_slicing_forward_pass(test_max_difference=UpperCAmelCase__   )




	def   UpperCAmelCase_      ( self			:				Optional[Any]   )       ->	Union[str, Any]:
						__SCREAMING_SNAKE_CASE						   =				torch_device in ["cpu", "mps"]

						self._test_inference_batch_single_identical(test_max_difference=UpperCAmelCase__   )




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




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

	"""simple docstring"""





	def   UpperCAmelCase_      ( self			:				Optional[Any]   )       ->	int:
						# clean up the VRAM after each test
						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()




	def   UpperCAmelCase_      ( self			:				Any   )       ->	List[str]:
						__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(UpperCAmelCase__   )
						pipe.set_progress_bar_config(disable=UpperCAmelCase__   )
						# 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(UpperCAmelCase__			,			"anime turle"			,			generator=UpperCAmelCase__			,			output_type="np"   )

						__SCREAMING_SNAKE_CASE						   =				output.images[0]

						assert image.shape == (7_6_8, 7_6_8, 3)

						assert_mean_pixel_difference(UpperCAmelCase__			,			UpperCAmelCase__   )




	def   UpperCAmelCase_      ( self			:				Dict   )       ->	Dict:
						__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(UpperCAmelCase__   )
						pipe.set_progress_bar_config(disable=UpperCAmelCase__   )
						# 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(UpperCAmelCase__			,			"anime turle"			,			generator=UpperCAmelCase__			,			output_type="np"   )

						__SCREAMING_SNAKE_CASE						   =				output.images[0]

						assert image.shape == (7_6_8, 7_6_8, 3)

						assert_mean_pixel_difference(UpperCAmelCase__			,			UpperCAmelCase__   )




	def   UpperCAmelCase_      ( self			:				int   )       ->	Any:
						__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(UpperCAmelCase__   )
						pipe.set_progress_bar_config(disable=UpperCAmelCase__   )
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						__SCREAMING_SNAKE_CASE						   =				pipe(
						    UpperCAmelCase__			,			"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 * 1_0**9


import uuid
from typing import Any, Dict, List, Optional, Union

from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
		import tensorflow as tf

if is_torch_available():
		import torch


lowercase  :					List[str]													=       logging.get_logger(__name__)




class  lowerCamelCase__							:




	'''simple docstring'''




	def __init__(					self		:str     ,   a		:str = None     ,   a		:uuid.UUID = None     ,   a		:Tuple=None     ,   a		:Optional[Any]=None						)		-> str:
								if not conversation_id:
															__UpperCamelCase	:   Dict			   =					uuid.uuida()
								if past_user_inputs is None:
															__UpperCamelCase	:   List[Any]			   =					[]
								if generated_responses is None:
															__UpperCamelCase	:   Any			   =					[]

								__UpperCamelCase	:   uuid.UUID			   =					conversation_id
								__UpperCamelCase	:   List[str]			   =					past_user_inputs
								__UpperCamelCase	:   List[str]			   =					generated_responses
								__UpperCamelCase	:   Optional[str]			   =					text







	def __eq__(					self		:Optional[int]     ,   a		:Optional[int]						)		-> Union[str, Any]:
								if not isinstance(a     ,   a						):
															return False
								if self.uuid == other.uuid:
															return True
								return (
								    self.new_user_input == other.new_user_input
								    and self.past_user_inputs == other.past_user_inputs
								    and self.generated_responses == other.generated_responses
								)







	def    _lowerCamelCase				(					self		:Optional[int]     ,   a		:str     ,   a		:bool = False						)		-> str:
								if self.new_user_input:
															if overwrite:
																						logger.warning(
																						    f'User input added while unprocessed input was existing: "{self.new_user_input}" was overwritten '
																						    f'with: "{text}".'						)
																						__UpperCamelCase	:   Any			   =					text
															else:
																						logger.warning(
																						    f'User input added while unprocessed input was existing: "{self.new_user_input}" new input '
																						    f'ignored: "{text}". Set `overwrite` to True to overwrite unprocessed user input'						)
								else:
															__UpperCamelCase	:   int			   =					text







	def    _lowerCamelCase				(					self		:List[str]						)		-> int:
								if self.new_user_input:
															self.past_user_inputs.append(self.new_user_input						)
								__UpperCamelCase	:   Dict			   =					None







	def    _lowerCamelCase				(					self		:Optional[int]     ,   a		:str						)		-> Optional[int]:
								self.generated_responses.append(a						)







	def    _lowerCamelCase				(					self		:int						)		-> Optional[Any]:
								for user_input, generated_response in zip(self.past_user_inputs     ,   self.generated_responses						):
															yield True, user_input
															yield False, generated_response
								if self.new_user_input:
															yield True, self.new_user_input







	def __repr__(					self		:List[str]						)		-> List[Any]:
								__UpperCamelCase	:   Any			   =					f'Conversation id: {self.uuid} \n'
								for is_user, text in self.iter_texts():
															__UpperCamelCase	:   str			   =					"user" if is_user else "bot"
															output += f'{name} >> {text} \n'
								return output




@add_end_docstrings(
    __lowercase     ,						R'\n        min_length_for_response (`int`, *optional*, defaults to 32):\n            The minimum length (in number of tokens) for a response.\n        minimum_tokens (`int`, *optional*, defaults to 10):\n            The minimum length of tokens to leave for a response.\n    '     ,						)
class  lowerCamelCase__							(				__lowercase):




	'''simple docstring'''




	def __init__(					self		:Tuple     ,   *a		:Tuple     ,   **a		:List[str]						)		-> Tuple:
								super().__init__(*a     ,   **a						)
								if self.tokenizer.pad_token_id is None:
															__UpperCamelCase	:   int			   =					self.tokenizer.eos_token







	def    _lowerCamelCase				(					self		:Optional[int]     ,   a		:List[Any]=None     ,   a		:str=None     ,   a		:int=None     ,   **a		:str						)		-> List[str]:
								__UpperCamelCase	:   List[str]			   =					{}
								__UpperCamelCase	:   List[str]			   =					{}
								__UpperCamelCase	:   str			   =					{}

								if min_length_for_response is not None:
															__UpperCamelCase	:   Optional[Any]			   =					min_length_for_response
								if minimum_tokens is not None:
															__UpperCamelCase	:   List[str]			   =					minimum_tokens

								if "max_length" in generate_kwargs:
															__UpperCamelCase	:   List[Any]			   =					generate_kwargs["max_length"]
															# self.max_length = generate_kwargs.get("max_length", self.model.config.max_length)
								if clean_up_tokenization_spaces is not None:
															__UpperCamelCase	:   List[Any]			   =					clean_up_tokenization_spaces

								if generate_kwargs:
															forward_params.update(a						)
								return preprocess_params, forward_params, postprocess_params







	def __call__(					self		:Dict     ,   a		:Union[Conversation, List[Conversation]]     ,   a		:List[Any]=0     ,   **a		:Any						)		-> Union[str, Any]:
								__UpperCamelCase	:   Optional[int]			   =					super().__call__(a     ,   num_workers=a     ,   **a						)
								if isinstance(a     ,   a						) and len(a						) == 1:
															return outputs[0]
								return outputs







	def    _lowerCamelCase				(					self		:Tuple     ,   a		:Conversation     ,   a		:Dict=3_2						)		-> Dict[str, Any]:
								if not isinstance(a     ,   a						):
															raise ValueError("ConversationalPipeline, expects Conversation as inputs"						)
								if conversation.new_user_input is None:
															raise ValueError(
															    f'Conversation with UUID {type(conversation.uuid						)} does not contain new user input to process. '
															    "Add user inputs with the conversation's `add_user_input` method"						)
								if hasattr(self.tokenizer     ,   "_build_conversation_input_ids"						):
															__UpperCamelCase	:   str			   =					self.tokenizer._build_conversation_input_ids(a						)
								else:
															# If the tokenizer cannot handle conversations, we default to only the old version
															__UpperCamelCase	:   Optional[Any]			   =					self._legacy_parse_and_tokenize(a						)

								if self.framework == "pt":
															__UpperCamelCase	:   Dict			   =					torch.LongTensor([input_ids]						)
								elif self.framework == "tf":
															__UpperCamelCase	:   Any			   =					tf.constant([input_ids]						)
								return {"input_ids": input_ids, "conversation": conversation}







	def    _lowerCamelCase				(					self		:Any     ,   a		:List[Any]     ,   a		:Optional[Any]=1_0     ,   **a		:Tuple						)		-> List[str]:
								__UpperCamelCase	:   Union[str, Any]			   =					generate_kwargs.get("max_length"     ,   self.model.config.max_length						)

								__UpperCamelCase	:   Dict			   =					model_inputs["input_ids"].shape[1]
								if max_length - minimum_tokens < n:
															logger.warning(f'Conversation input is to long ({n}), trimming it to ({max_length} - {minimum_tokens})'						)
															__UpperCamelCase	:   Dict			   =					max_length - minimum_tokens
															__UpperCamelCase	:   Optional[int]			   =					model_inputs["input_ids"][:, -trim:]
															if "attention_mask" in model_inputs:
																						__UpperCamelCase	:   Dict			   =					model_inputs["attention_mask"][:, -trim:]
								__UpperCamelCase	:   List[str]			   =					model_inputs.pop("conversation"						)
								__UpperCamelCase	:   Optional[int]			   =					max_length
								__UpperCamelCase	:   str			   =					self.model.generate(**a     ,   **a						)
								if self.model.config.is_encoder_decoder:
															__UpperCamelCase	:   List[str]			   =					1
								else:
															__UpperCamelCase	:   Optional[int]			   =					n
								return {"output_ids": output_ids[:, start_position:], "conversation": conversation}







	def    _lowerCamelCase				(					self		:List[Any]     ,   a		:str     ,   a		:Optional[int]=True						)		-> Union[str, Any]:
								__UpperCamelCase	:   List[str]			   =					model_outputs["output_ids"]
								__UpperCamelCase	:   Any			   =					self.tokenizer.decode(
								    output_ids[0]     ,   skip_special_tokens=a     ,   clean_up_tokenization_spaces=a     ,   )
								__UpperCamelCase	:   int			   =					model_outputs["conversation"]
								conversation.mark_processed()
								conversation.append_response(a						)
								return conversation







	def    _lowerCamelCase				(					self		:str     ,   a		:Conversation						)		-> Dict:
								__UpperCamelCase	:   int			   =					self.tokenizer.eos_token_id
								__UpperCamelCase	:   Any			   =					[]
								for is_user, text in conversation.iter_texts():
															if eos_token_id is not None:
																						input_ids.extend(self.tokenizer.encode(a     ,   add_special_tokens=a						) + [eos_token_id]						)
															else:
																						input_ids.extend(self.tokenizer.encode(a     ,   add_special_tokens=a						)						)

								if len(a						) > self.tokenizer.model_max_length:
															__UpperCamelCase	:   Union[str, Any]			   =					input_ids[-self.tokenizer.model_max_length :]
								return input_ids