






"""simple docstring"""





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


__lowerCamelCase							=  logging.get_logger(__name__)



class       __A		:

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

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

					def __eq__(			self  :      List[Any]     ,		__snake_case  :      int					)			->							List[str]:
											if not isinstance(__snake_case     ,		__snake_case					):
																	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  :      List[str]     ,		__snake_case  :      str     ,		__snake_case  :      bool = False					)			->							Any:
											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}".'					)
																							__magic_name__:	List[str]							=       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:
																	__magic_name__:	Union[str, Any]							=       text

					def   lowerCamelCase__   (			self  :      Any					)			->							Optional[int]:
											if self.new_user_input:
																	self.past_user_inputs.append(self.new_user_input					)
											__magic_name__:	Any							=       None

					def   lowerCamelCase__   (			self  :      Union[str, Any]     ,		__snake_case  :      str					)			->							str:
											self.generated_responses.append(__snake_case					)

					def   lowerCamelCase__   (			self  :      str					)			->							List[str]:
											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  :      Tuple					)			->							Tuple:
											__magic_name__:	Any							=       F'Conversation id: {self.uuid} \n'
											for is_user, text in self.iter_texts():
																	__magic_name__:	List[str]							=       """user""" if is_user else """bot"""
																	output += F'{name} >> {text} \n'
											return output



@add_end_docstrings(
    SCREAMING_SNAKE_CASE_     ,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       __A		(   SCREAMING_SNAKE_CASE_			):

					def __init__(			self  :      Union[str, Any]     ,		*__snake_case  :      Any     ,		**__snake_case  :      Optional[int]					)			->							int:
											super().__init__(*__snake_case     ,		**__snake_case					)
											if self.tokenizer.pad_token_id is None:
																	__magic_name__:	Optional[int]							=       self.tokenizer.eos_token

					def   lowerCamelCase__   (			self  :      Optional[Any]     ,		__snake_case  :      str=None     ,		__snake_case  :      List[str]=None     ,		__snake_case  :      int=None     ,		**__snake_case  :      Optional[int]					)			->							Dict:
											__magic_name__:	List[Any]							=       {}
											__magic_name__:	int							=       {}
											__magic_name__:	List[str]							=       {}

											if min_length_for_response is not None:
																	__magic_name__:	Union[str, Any]							=       min_length_for_response
											if minimum_tokens is not None:
																	__magic_name__:	Optional[int]							=       minimum_tokens

											if "max_length" in generate_kwargs:
																	__magic_name__:	Tuple							=       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:
																	__magic_name__:	Optional[int]							=       clean_up_tokenization_spaces

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

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

					def   lowerCamelCase__   (			self  :      Union[str, Any]     ,		__snake_case  :      Conversation     ,		__snake_case  :      Tuple=3_2					)			->							Dict[str, Any]:
											if not isinstance(__snake_case     ,		__snake_case					):
																	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"""					):
																	__magic_name__:	int							=       self.tokenizer._build_conversation_input_ids(__snake_case					)
											else:
																	# If the tokenizer cannot handle conversations, we default to only the old version
																	__magic_name__:	Union[str, Any]							=       self._legacy_parse_and_tokenize(__snake_case					)

											if self.framework == "pt":
																	__magic_name__:	int							=       torch.LongTensor([input_ids]					)
											elif self.framework == "tf":
																	__magic_name__:	Tuple							=       tf.constant([input_ids]					)
											return {"input_ids": input_ids, "conversation": conversation}

					def   lowerCamelCase__   (			self  :      Optional[Any]     ,		__snake_case  :      Any     ,		__snake_case  :      int=1_0     ,		**__snake_case  :      Optional[Any]					)			->							Any:
											__magic_name__:	Any							=       generate_kwargs.get("""max_length"""     ,		self.model.config.max_length					)

											__magic_name__:	int							=       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})'					)
																	__magic_name__:	Any							=       max_length - minimum_tokens
																	__magic_name__:	Union[str, Any]							=       model_inputs["""input_ids"""][:, -trim:]
																	if "attention_mask" in model_inputs:
																							__magic_name__:	Union[str, Any]							=       model_inputs["""attention_mask"""][:, -trim:]
											__magic_name__:	Tuple							=       model_inputs.pop("""conversation"""					)
											__magic_name__:	Any							=       max_length
											__magic_name__:	Tuple							=       self.model.generate(**__snake_case     ,		**__snake_case					)
											if self.model.config.is_encoder_decoder:
																	__magic_name__:	List[str]							=       1
											else:
																	__magic_name__:	List[Any]							=       n
											return {"output_ids": output_ids[:, start_position:], "conversation": conversation}

					def   lowerCamelCase__   (			self  :      List[Any]     ,		__snake_case  :      str     ,		__snake_case  :      Tuple=True					)			->							Any:
											__magic_name__:	Tuple							=       model_outputs["""output_ids"""]
											__magic_name__:	List[Any]							=       self.tokenizer.decode(
											    output_ids[0]     ,		skip_special_tokens=__snake_case     ,		clean_up_tokenization_spaces=__snake_case     ,		)
											__magic_name__:	Tuple							=       model_outputs["""conversation"""]
											conversation.mark_processed()
											conversation.append_response(__snake_case					)
											return conversation






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

											if len(__snake_case					) > self.tokenizer.model_max_length:
																	__magic_name__:	Tuple							=       input_ids[-self.tokenizer.model_max_length :]
											return input_ids










"""simple docstring"""



import gc
import random
import unittest

import numpy as np
import torch

from diffusers import DDIMScheduler, KandinskyVaaPipeline, KandinskyVaaPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()






class 						__snake_case   (						SCREAMING_SNAKE_CASE__				,				unittest.TestCase	):





 """simple docstring"""


 _lowerCamelCase								=	KandinskyVaaPipeline
 _lowerCamelCase								=	[
     """image_embeds""",
     """negative_image_embeds""",
 ]
 _lowerCamelCase								=	["""image_embeds""", """negative_image_embeds"""]
 _lowerCamelCase								=	[
     """generator""",
     """height""",
     """width""",
     """latents""",
     """guidance_scale""",
     """num_inference_steps""",
     """return_dict""",
     """guidance_scale""",
     """num_images_per_prompt""",
     """output_type""",
     """return_dict""",
 ]
 _lowerCamelCase								=	False

 @property
 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     return 32

 @property
 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     return 32

 @property
 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     return self.time_input_dim

 @property
 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     return self.time_input_dim * 4

 @property
 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     return 100

 @property
 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     torch.manual_seed(0    )

     __A       :		Optional[Any]							=	{
         '''in_channels''': 4,
         # Out channels is double in channels because predicts mean and variance
         '''out_channels''': 8,
         '''addition_embed_type''': '''image''',
         '''down_block_types''': ('''ResnetDownsampleBlock2D''', '''SimpleCrossAttnDownBlock2D'''),
         '''up_block_types''': ('''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D'''),
         '''mid_block_type''': '''UNetMidBlock2DSimpleCrossAttn''',
         '''block_out_channels''': (self.block_out_channels_a, self.block_out_channels_a * 2),
         '''layers_per_block''': 1,
         '''encoder_hid_dim''': self.text_embedder_hidden_size,
         '''encoder_hid_dim_type''': '''image_proj''',
         '''cross_attention_dim''': self.cross_attention_dim,
         '''attention_head_dim''': 4,
         '''resnet_time_scale_shift''': '''scale_shift''',
         '''class_embed_type''': None,
     }

     __A       :		Optional[Any]							=	UNetaDConditionModel(**__lowerCamelCase    )
     return model

 @property
 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     return {
         "block_out_channels": [32, 64],
         "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
         "in_channels": 3,
         "latent_channels": 4,
         "layers_per_block": 1,
         "norm_num_groups": 8,
         "norm_type": "spatial",
         "num_vq_embeddings": 12,
         "out_channels": 3,
         "up_block_types": [
             "AttnUpDecoderBlock2D",
             "UpDecoderBlock2D",
         ],
         "vq_embed_dim": 4,
     }

 @property
 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     torch.manual_seed(0    )
     __A       :		Tuple							=	VQModel(**self.dummy_movq_kwargs    )
     return model

 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     __A       :		List[Any]							=	self.dummy_unet
     __A       :		Optional[int]							=	self.dummy_movq

     __A       :		List[Any]							=	DDIMScheduler(
         num_train_timesteps=1000   ,		beta_schedule='''linear'''   ,		beta_start=0.0_0_0_8_5   ,		beta_end=0.0_1_2   ,		clip_sample=__lowerCamelCase   ,		set_alpha_to_one=__lowerCamelCase   ,		steps_offset=1   ,		prediction_type='''epsilon'''   ,		thresholding=__lowerCamelCase   ,		)

     __A       :		Optional[int]							=	{
         '''unet''': unet,
         '''scheduler''': scheduler,
         '''movq''': movq,
     }
     return components

 def    UpperCamelCase__(       self   ,		__lowerCamelCase   ,		__lowerCamelCase=0    ):



     '''simple docstring'''




     __A       :		Optional[int]							=	floats_tensor((1, self.text_embedder_hidden_size)   ,		rng=random.Random(__lowerCamelCase    )    ).to(__lowerCamelCase    )
     __A       :		List[Any]							=	floats_tensor((1, self.text_embedder_hidden_size)   ,		rng=random.Random(seed + 1    )    ).to(
         __lowerCamelCase    )
     if str(__lowerCamelCase    ).startswith('''mps'''    ):
         __A       :		Optional[Any]							=	torch.manual_seed(__lowerCamelCase    )
     else:
         __A       :		Dict							=	torch.Generator(device=__lowerCamelCase    ).manual_seed(__lowerCamelCase    )
     __A       :		Dict							=	{
         '''image_embeds''': image_embeds,
         '''negative_image_embeds''': negative_image_embeds,
         '''generator''': generator,
         '''height''': 64,
         '''width''': 64,
         '''guidance_scale''': 4.0,
         '''num_inference_steps''': 2,
         '''output_type''': '''np''',
     }
     return inputs







 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     __A       :		str							=	'''cpu'''

     __A       :		str							=	self.get_dummy_components()

     __A       :		Dict							=	self.pipeline_class(**__lowerCamelCase    )
     __A       :		Tuple							=	pipe.to(__lowerCamelCase    )

     pipe.set_progress_bar_config(disable=__lowerCamelCase    )

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

     __A       :		Tuple							=	pipe(
         **self.get_dummy_inputs(__lowerCamelCase    )   ,		return_dict=__lowerCamelCase   ,		)[0]

     __A       :		Dict							=	image[0, -3:, -3:, -1]
     __A       :		Any							=	image_from_tuple[0, -3:, -3:, -1]

     assert image.shape == (1, 64, 64, 3)

     __A       :		Optional[int]							=	np.array(
         [0.6_2_3_7_9_7_6, 1.0, 0.3_6_4_4_1_3_3_2, 1.0, 0.7_0_6_3_9_6_3_4, 0.2_9_8_7_7_1_8_6, 0.8_5_6_5_2_1_2_5, 0.5_2_1_6_8_4_3, 0.5_4_4_5_4_0_4_6]    )

     assert (
         np.abs(image_slice.flatten() - expected_slice    ).max() < 1e-2
     ), F""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""

     assert (
         np.abs(image_from_tuple_slice.flatten() - expected_slice    ).max() < 1e-2
     ), F""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""






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





 """simple docstring"""


 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




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







 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     __A       :		Any							=	load_numpy(
         '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
         '''/kandinskyv22/kandinskyv22_text2img_cat_fp16.npy'''    )

     __A       :		Optional[Any]							=	KandinskyVaaPriorPipeline.from_pretrained(
         '''kandinsky-community/kandinsky-2-2-prior'''   ,		torch_dtype=torch.floataa    )
     pipe_prior.to(__lowerCamelCase    )

     __A       :		int							=	KandinskyVaaPipeline.from_pretrained(
         '''kandinsky-community/kandinsky-2-2-decoder'''   ,		torch_dtype=torch.floataa    )
     __A       :		Union[str, Any]							=	pipeline.to(__lowerCamelCase    )
     pipeline.set_progress_bar_config(disable=__lowerCamelCase    )

     __A       :		List[Any]							=	'''red cat, 4k photo'''

     __A       :		Union[str, Any]							=	torch.Generator(device='''cuda'''    ).manual_seed(0    )
     __A					,			__A       :		List[str]							=	pipe_prior(
         __lowerCamelCase   ,		generator=__lowerCamelCase   ,		num_inference_steps=5   ,		negative_prompt=''''''   ,		).to_tuple()

     __A       :		Any							=	torch.Generator(device='''cuda'''    ).manual_seed(0    )
     __A       :		Optional[int]							=	pipeline(
         image_embeds=__lowerCamelCase   ,		negative_image_embeds=__lowerCamelCase   ,		generator=__lowerCamelCase   ,		num_inference_steps=100   ,		output_type='''np'''   ,		)

     __A       :		Union[str, Any]							=	output.images[0]

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

     assert_mean_pixel_difference(__lowerCamelCase   ,		__lowerCamelCase    )




