


'''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


UpperCAmelCase_						:  Tuple				  =						logging.get_logger(__name__)





class        lowercase__    :





							'''simple docstring'''


							def __init__(       self       ,   __snake_case = None       ,   __snake_case = None       ,   __snake_case=None       ,   __snake_case=None    ):
											if not conversation_id:
															_SCREAMING_SNAKE_CASE					:			Tuple							= uuid.uuida()
											if past_user_inputs is None:
															_SCREAMING_SNAKE_CASE					:			List[str]							= []
											if generated_responses is None:
															_SCREAMING_SNAKE_CASE					:			List[Any]							= []

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






							def __eq__(       self       ,   __snake_case    ):
											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 UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case = False    ):
											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}\"."""    )
																			_SCREAMING_SNAKE_CASE					:			Union[str, 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:
															_SCREAMING_SNAKE_CASE					:			Optional[Any]							= text






							def UpperCAmelCase_      (       self    ):
											if self.new_user_input:
															self.past_user_inputs.append(self.new_user_input    )
											_SCREAMING_SNAKE_CASE					:			str							= None






							def UpperCAmelCase_      (       self       ,   __snake_case    ):
											self.generated_responses.append(__snake_case    )






							def UpperCAmelCase_      (       self    ):
											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    ):
											_SCREAMING_SNAKE_CASE					:			Dict							= f"""Conversation id: {self.uuid} \n"""
											for is_user, text in self.iter_texts():
															_SCREAMING_SNAKE_CASE					:			Any							= """user""" if is_user else """bot"""
															output += f"""{name} >> {text} \n"""
											return output





@add_end_docstrings(
    _snake_case	,	r"""
        min_length_for_response (`int`, *optional*, defaults to 32):
            The minimum length (in number of tokens) for a response.
        minimum_tokens (`int`, *optional*, defaults to 10):
            The minimum length of tokens to leave for a response.
    """	,	)
class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							def __init__(       self       ,   *__snake_case       ,   **__snake_case    ):
											super().__init__(*__snake_case       ,   **__snake_case    )
											if self.tokenizer.pad_token_id is None:
															_SCREAMING_SNAKE_CASE					:			str							= self.tokenizer.eos_token






							def UpperCAmelCase_      (       self       ,   __snake_case=None       ,   __snake_case=None       ,   __snake_case=None       ,   **__snake_case    ):
											_SCREAMING_SNAKE_CASE					:			List[Any]							= {}
											_SCREAMING_SNAKE_CASE					:			str							= {}
											_SCREAMING_SNAKE_CASE					:			Tuple							= {}

											if min_length_for_response is not None:
															_SCREAMING_SNAKE_CASE					:			List[Any]							= min_length_for_response
											if minimum_tokens is not None:
															_SCREAMING_SNAKE_CASE					:			int							= minimum_tokens

											if "max_length" in generate_kwargs:
															_SCREAMING_SNAKE_CASE					:			List[str]							= 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:
															_SCREAMING_SNAKE_CASE					:			str							= clean_up_tokenization_spaces

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






							def __call__(       self       ,   __snake_case       ,   __snake_case=0       ,   **__snake_case    ):
											_SCREAMING_SNAKE_CASE					:			Dict							= 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 UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case=32    ):
											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"""    ):
															_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.tokenizer._build_conversation_input_ids(__snake_case    )
											else:
															# If the tokenizer cannot handle conversations, we default to only the old version
															_SCREAMING_SNAKE_CASE					:			str							= self._legacy_parse_and_tokenize(__snake_case    )

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






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case=10       ,   **__snake_case    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= generate_kwargs.get("""max_length"""       ,   self.model.config.max_length    )

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






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case=True    ):
											_SCREAMING_SNAKE_CASE					:			str							= model_outputs["""output_ids"""]
											_SCREAMING_SNAKE_CASE					:			int							= self.tokenizer.decode(
											    output_ids[0]       ,   skip_special_tokens=__snake_case       ,   clean_up_tokenization_spaces=__snake_case       ,   )
											_SCREAMING_SNAKE_CASE					:			Any							= model_outputs["""conversation"""]
											conversation.mark_processed()
											conversation.append_response(__snake_case    )
											return conversation






							def UpperCAmelCase_      (       self       ,   __snake_case    ):
											_SCREAMING_SNAKE_CASE					:			int							= self.tokenizer.eos_token_id
											_SCREAMING_SNAKE_CASE					:			Dict							= []
											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:
															_SCREAMING_SNAKE_CASE					:			List[str]							= 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        lowercase__    (		_snake_case	,	unittest.TestCase   ):





							'''simple docstring'''


							A_				: Dict       =    KandinskyVaaPipeline
							A_				: List[str]       =    [
							    """image_embeds""",
							    """negative_image_embeds""",
							]
							A_				: Optional[int]       =    ["""image_embeds""", """negative_image_embeds"""]
							A_				: Optional[Any]       =    [
							    """generator""",
							    """height""",
							    """width""",
							    """latents""",
							    """guidance_scale""",
							    """num_inference_steps""",
							    """return_dict""",
							    """guidance_scale""",
							    """num_images_per_prompt""",
							    """output_type""",
							    """return_dict""",
							]
							A_				: Dict       =    False






							@property
							def UpperCAmelCase_      (       self    ):
											return 32






							@property
							def UpperCAmelCase_      (       self    ):
											return 32






							@property
							def UpperCAmelCase_      (       self    ):
											return self.time_input_dim






							@property
							def UpperCAmelCase_      (       self    ):
											return self.time_input_dim * 4






							@property
							def UpperCAmelCase_      (       self    ):
											return 100






							@property
							def UpperCAmelCase_      (       self    ):
											torch.manual_seed(0    )

											_SCREAMING_SNAKE_CASE					:			Tuple							= {
											    """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,
											}

											_SCREAMING_SNAKE_CASE					:			Any							= UNetaDConditionModel(**__snake_case    )
											return model






							@property
							def UpperCAmelCase_      (       self    ):
											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    ):
											torch.manual_seed(0    )
											_SCREAMING_SNAKE_CASE					:			Tuple							= VQModel(**self.dummy_movq_kwargs    )
											return model






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			int							= self.dummy_unet
											_SCREAMING_SNAKE_CASE					:			List[str]							= self.dummy_movq

											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= DDIMScheduler(
											    num_train_timesteps=1000       ,   beta_schedule="""linear"""       ,   beta_start=0.00085       ,   beta_end=0.012       ,   clip_sample=__snake_case       ,   set_alpha_to_one=__snake_case       ,   steps_offset=1       ,   prediction_type="""epsilon"""       ,   thresholding=__snake_case       ,   )

											_SCREAMING_SNAKE_CASE					:			Dict							= {
											    """unet""": unet,
											    """scheduler""": scheduler,
											    """movq""": movq,
											}
											return components






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case=0    ):
											_SCREAMING_SNAKE_CASE					:			int							= floats_tensor((1, self.text_embedder_hidden_size)       ,   rng=random.Random(__snake_case    )    ).to(__snake_case    )
											_SCREAMING_SNAKE_CASE					:			List[str]							= floats_tensor((1, self.text_embedder_hidden_size)       ,   rng=random.Random(seed + 1    )    ).to(
											    __snake_case    )
											if str(__snake_case    ).startswith("""mps"""    ):
															_SCREAMING_SNAKE_CASE					:			str							= torch.manual_seed(__snake_case    )
											else:
															_SCREAMING_SNAKE_CASE					:			Dict							= torch.Generator(device=__snake_case    ).manual_seed(__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Tuple							= {
											    """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    ):
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """cpu"""

											_SCREAMING_SNAKE_CASE					:			Dict							= self.get_dummy_components()

											_SCREAMING_SNAKE_CASE					:			List[Any]							= self.pipeline_class(**__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= pipe.to(__snake_case    )

											pipe.set_progress_bar_config(disable=__snake_case    )

											_SCREAMING_SNAKE_CASE					:			Any							= pipe(**self.get_dummy_inputs(__snake_case    )    )
											_SCREAMING_SNAKE_CASE					:			Tuple							= output.images

											_SCREAMING_SNAKE_CASE					:			Dict							= pipe(
											    **self.get_dummy_inputs(__snake_case    )       ,   return_dict=__snake_case       ,   )[0]

											_SCREAMING_SNAKE_CASE					:			Tuple							= image[0, -3:, -3:, -1]
											_SCREAMING_SNAKE_CASE					:			List[Any]							= image_from_tuple[0, -3:, -3:, -1]

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

											_SCREAMING_SNAKE_CASE					:			Dict							= np.array(
											    [0.6237976, 1.0, 0.36441332, 1.0, 0.70639634, 0.29877186, 0.85652125, 0.5216843, 0.54454046]    )

											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        lowercase__    (		unittest.TestCase   ):





							'''simple docstring'''


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






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			List[str]							= load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
											    """/kandinskyv22/kandinskyv22_text2img_cat_fp16.npy"""    )

											_SCREAMING_SNAKE_CASE					:			List[Any]							= KandinskyVaaPriorPipeline.from_pretrained(
											    """kandinsky-community/kandinsky-2-2-prior"""       ,   torch_dtype=torch.floataa    )
											pipe_prior.to(__snake_case    )

											_SCREAMING_SNAKE_CASE					:			Tuple							= KandinskyVaaPipeline.from_pretrained(
											    """kandinsky-community/kandinsky-2-2-decoder"""       ,   torch_dtype=torch.floataa    )
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= pipeline.to(__snake_case    )
											pipeline.set_progress_bar_config(disable=__snake_case    )

											_SCREAMING_SNAKE_CASE					:			Dict							= """red cat, 4k photo"""

											_SCREAMING_SNAKE_CASE					:			List[str]							= torch.Generator(device="""cuda"""    ).manual_seed(0    )
											_SCREAMING_SNAKE_CASE						,			_SCREAMING_SNAKE_CASE					:			List[str]							= pipe_prior(
											    __snake_case       ,   generator=__snake_case       ,   num_inference_steps=5       ,   negative_prompt=""""""       ,   ).to_tuple()

											_SCREAMING_SNAKE_CASE					:			Dict							= torch.Generator(device="""cuda"""    ).manual_seed(0    )
											_SCREAMING_SNAKE_CASE					:			List[str]							= pipeline(
											    image_embeds=__snake_case       ,   negative_image_embeds=__snake_case       ,   generator=__snake_case       ,   num_inference_steps=100       ,   output_type="""np"""       ,   )

											_SCREAMING_SNAKE_CASE					:			List[Any]							= output.images[0]

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

											assert_mean_pixel_difference(__snake_case       ,   __snake_case    )




