





"""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							=      logging.get_logger(__name__)
class 		SCREAMING_SNAKE_CASE_							:
				"""simple docstring"""




				def __init__(     self  :Dict						,    __lowercase  :str = None						,    __lowercase  :uuid.UUID = None						,    __lowercase  :List[Any]=None						,    __lowercase  :Any=None			):
											if not conversation_id:
																		__lowerCamelCase      :      Dict         =uuid.uuida()
											if past_user_inputs is None:
																		__lowerCamelCase      :      Optional[int]         =[]
											if generated_responses is None:
																		__lowerCamelCase      :      List[Any]         =[]

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




				def __eq__(     self  :Optional[Any]						,    __lowercase  :Tuple			):
											if not isinstance(__lowercase						,    __lowercase			):
																		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 					__lowercase  (     self  :Any						,    __lowercase  :str						,    __lowercase  :bool = 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}".'			)
																									__lowerCamelCase      :      List[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:
																		__lowerCamelCase      :      Optional[int]         =text




				def 					__lowercase  (     self  :List[str]			):
											if self.new_user_input:
																		self.past_user_inputs.append(self.new_user_input			)
											__lowerCamelCase      :      Dict         =None




				def 					__lowercase  (     self  :List[Any]						,    __lowercase  :str			):
											self.generated_responses.append(__lowercase			)




				def 					__lowercase  (     self  :int			):
											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  :int			):
											__lowerCamelCase      :      Union[str, Any]         =f'Conversation id: {self.uuid} \n'
											for is_user, text in self.iter_texts():
																		__lowerCamelCase      :      Optional[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 		SCREAMING_SNAKE_CASE_							(    snake_case__   ):
				"""simple docstring"""




				def __init__(     self  :Tuple						,    *__lowercase  :int						,    **__lowercase  :Any			):
											super().__init__(*__lowercase						,    **__lowercase			)
											if self.tokenizer.pad_token_id is None:
																		__lowerCamelCase      :      int         =self.tokenizer.eos_token




				def 					__lowercase  (     self  :str						,    __lowercase  :int=None						,    __lowercase  :Dict=None						,    __lowercase  :Tuple=None						,    **__lowercase  :str			):
											__lowerCamelCase      :      Union[str, Any]         ={}
											__lowerCamelCase      :      str         ={}
											__lowerCamelCase      :      int         ={}

											if min_length_for_response is not None:
																		__lowerCamelCase      :      Optional[Any]         =min_length_for_response
											if minimum_tokens is not None:
																		__lowerCamelCase      :      Optional[int]         =minimum_tokens

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

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




				def __call__(     self  :Any						,    __lowercase  :Union[Conversation, List[Conversation]]						,    __lowercase  :List[str]=0						,    **__lowercase  :str			):
											__lowerCamelCase      :      List[str]         =super().__call__(__lowercase						,    num_workers=__lowercase						,    **__lowercase			)
											if isinstance(__lowercase						,    __lowercase			) and len(__lowercase			) == 1:
																		return outputs[0]
											return outputs




				def 					__lowercase  (     self  :List[Any]						,    __lowercase  :Conversation						,    __lowercase  :int=32			):
											if not isinstance(__lowercase						,    __lowercase			):
																		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'''			):
																		__lowerCamelCase      :      Optional[Any]         =self.tokenizer._build_conversation_input_ids(__lowercase			)
											else:
																		# If the tokenizer cannot handle conversations, we default to only the old version
																		__lowerCamelCase      :      Optional[int]         =self._legacy_parse_and_tokenize(__lowercase			)

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




				def 					__lowercase  (     self  :Optional[Any]						,    __lowercase  :Any						,    __lowercase  :List[str]=10						,    **__lowercase  :Union[str, Any]			):
											__lowerCamelCase      :      List[str]         =generate_kwargs.get('''max_length'''						,    self.model.config.max_length			)

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




				def 					__lowercase  (     self  :List[Any]						,    __lowercase  :int						,    __lowercase  :List[str]=True			):
											__lowerCamelCase      :      Any         =model_outputs['''output_ids''']
											__lowerCamelCase      :      List[str]         =self.tokenizer.decode(
											    output_ids[0]						,    skip_special_tokens=__lowercase						,    clean_up_tokenization_spaces=__lowercase						,    )
											__lowerCamelCase      :      Optional[Any]         =model_outputs['''conversation''']
											conversation.mark_processed()
											conversation.append_response(__lowercase			)
											return conversation




				def 					__lowercase  (     self  :str						,    __lowercase  :Conversation			):
											__lowerCamelCase      :      str         =self.tokenizer.eos_token_id
											__lowerCamelCase      :      List[str]         =[]
											for is_user, text in conversation.iter_texts():
																		if eos_token_id is not None:
																									input_ids.extend(self.tokenizer.encode(__lowercase						,    add_special_tokens=__lowercase			) + [eos_token_id]			)
																		else:
																									input_ids.extend(self.tokenizer.encode(__lowercase						,    add_special_tokens=__lowercase			)			)

											if len(__lowercase			) > self.tokenizer.model_max_length:
																		__lowerCamelCase      :      int         =input_ids[-self.tokenizer.model_max_length :]
											return input_ids











"""simple docstring"""





from typing import List, Optional, Union

import torch

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)


_UpperCamelCase							=      logging.get_logger(__name__)  # pylint: disable=invalid-name

_UpperCamelCase							=      '\n    Examples:\n        ```py\n        >>> import torch\n        >>> import numpy as np\n\n        >>> from diffusers import KandinskyV22PriorPipeline, KandinskyV22ControlnetPipeline\n        >>> from transformers import pipeline\n        >>> from diffusers.utils import load_image\n\n\n        >>> def make_hint(image, depth_estimator):\n        ...     image = depth_estimator(image)["depth"]\n        ...     image = np.array(image)\n        ...     image = image[:, :, None]\n        ...     image = np.concatenate([image, image, image], axis=2)\n        ...     detected_map = torch.from_numpy(image).float() / 255.0\n        ...     hint = detected_map.permute(2, 0, 1)\n        ...     return hint\n\n\n        >>> depth_estimator = pipeline("depth-estimation")\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16\n        ... )\n        >>> pipe_prior = pipe_prior.to("cuda")\n\n        >>> pipe = KandinskyV22ControlnetPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-controlnet-depth", torch_dtype=torch.float16\n        ... )\n        >>> pipe = pipe.to("cuda")\n\n\n        >>> img = load_image(\n        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"\n        ...     "/kandinsky/cat.png"\n        ... ).resize((768, 768))\n\n        >>> hint = make_hint(img, depth_estimator).unsqueeze(0).half().to("cuda")\n\n        >>> prompt = "A robot, 4k photo"\n        >>> negative_prior_prompt = "lowres, text, error, cropped, worst quality, low quality, jpeg artifacts, ugly, duplicate, morbid, mutilated, out of frame, extra fingers, mutated hands, poorly drawn hands, poorly drawn face, mutation, deformed, blurry, dehydrated, bad anatomy, bad proportions, extra limbs, cloned face, disfigured, gross proportions, malformed limbs, missing arms, missing legs, extra arms, extra legs, fused fingers, too many fingers, long neck, username, watermark, signature"\n\n        >>> generator = torch.Generator(device="cuda").manual_seed(43)\n\n        >>> image_emb, zero_image_emb = pipe_prior(\n        ...     prompt=prompt, negative_prompt=negative_prior_prompt, generator=generator\n        ... ).to_tuple()\n\n        >>> images = pipe(\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     hint=hint,\n        ...     num_inference_steps=50,\n        ...     generator=generator,\n        ...     height=768,\n        ...     width=768,\n        ... ).images\n\n        >>> images[0].save("robot_cat.png")\n        ```\n'



def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							Any    ,						SCREAMING_SNAKE_CASE       :							Tuple    ,						SCREAMING_SNAKE_CASE       :							Tuple=8   ):


							'''simple docstring'''







							__lowerCamelCase      :      Tuple         =height // scale_factor**2
							if height % scale_factor**2 != 0:
														new_height += 1
							__lowerCamelCase      :      Tuple         =width // scale_factor**2
							if width % scale_factor**2 != 0:
														new_width += 1
							return new_height * scale_factor, new_width * scale_factor
class 		SCREAMING_SNAKE_CASE_							(    snake_case__   ):
				"""simple docstring"""




				def __init__(     self  :Union[str, Any]						,    __lowercase  :UNetaDConditionModel						,    __lowercase  :DDPMScheduler						,    __lowercase  :VQModel						,    ):
											super().__init__()

											self.register_modules(
											    unet=__lowercase						,    scheduler=__lowercase						,    movq=__lowercase						,    )
											__lowerCamelCase      :      Any         =2 ** (len(self.movq.config.block_out_channels			) - 1)




				def 					__lowercase  (     self  :int						,    __lowercase  :Tuple						,    __lowercase  :Dict						,    __lowercase  :str						,    __lowercase  :Dict						,    __lowercase  :Union[str, Any]						,    __lowercase  :Dict			):
											if latents is None:
																		__lowerCamelCase      :      str         =randn_tensor(__lowercase						,    generator=__lowercase						,    device=__lowercase						,    dtype=__lowercase			)
											else:
																		if latents.shape != shape:
																									raise ValueError(f'Unexpected latents shape, got {latents.shape}, expected {shape}'			)
																		__lowerCamelCase      :      str         =latents.to(__lowercase			)

											__lowerCamelCase      :      Any         =latents * scheduler.init_noise_sigma
											return latents




				def 					__lowercase  (     self  :Dict						,    __lowercase  :Any=0			):
											if is_accelerate_available():
																		from accelerate import cpu_offload
											else:
																		raise ImportError('''Please install accelerate via `pip install accelerate`'''			)

											__lowerCamelCase      :      Any         =torch.device(f'cuda:{gpu_id}'			)

											__lowerCamelCase      :      Dict         =[
											    self.unet,
											    self.movq,
											]
											for cpu_offloaded_model in models:
																		if cpu_offloaded_model is not None:
																									cpu_offload(__lowercase						,    __lowercase			)




				def 					__lowercase  (     self  :Optional[int]						,    __lowercase  :List[Any]=0			):
											if is_accelerate_available() and is_accelerate_version('''>='''						,    '''0.17.0.dev0'''			):
																		from accelerate import cpu_offload_with_hook
											else:
																		raise ImportError('''`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.'''			)

											__lowerCamelCase      :      str         =torch.device(f'cuda:{gpu_id}'			)

											if self.device.type != "cpu":
																		self.to('''cpu'''						,    silence_dtype_warnings=__lowercase			)
																		torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

											__lowerCamelCase      :      Optional[Any]         =None
											for cpu_offloaded_model in [self.unet, self.movq]:
																		__lowerCamelCase    ,						__lowerCamelCase      :      Union[str, Any]         =cpu_offload_with_hook(__lowercase						,    __lowercase						,    prev_module_hook=__lowercase			)

											# We'll offload the last model manually.
											__lowerCamelCase      :      Dict         =hook




				@property
				# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
				def 					__lowercase  (     self  :Tuple			):
											if not hasattr(self.unet						,    '''_hf_hook'''			):
																		return self.device
											for module in self.unet.modules():
																		if (
																		    hasattr(__lowercase						,    '''_hf_hook'''			)
																		    and hasattr(module._hf_hook						,    '''execution_device'''			)
																		    and module._hf_hook.execution_device is not None
																		):
																									return torch.device(module._hf_hook.execution_device			)
											return self.device




				@torch.no_grad()
				@replace_example_docstring(__lowercase			)
				def __call__(     self  :Union[str, Any]						,    __lowercase  :Union[torch.FloatTensor, List[torch.FloatTensor]]						,    __lowercase  :Union[torch.FloatTensor, List[torch.FloatTensor]]						,    __lowercase  :torch.FloatTensor						,    __lowercase  :int = 512						,    __lowercase  :int = 512						,    __lowercase  :int = 100						,    __lowercase  :float = 4.0						,    __lowercase  :int = 1						,    __lowercase  :Optional[Union[torch.Generator, List[torch.Generator]]] = None						,    __lowercase  :Optional[torch.FloatTensor] = None						,    __lowercase  :Optional[str] = "pil"						,    __lowercase  :bool = True						,    ):
											__lowerCamelCase      :      Any         =self._execution_device

											__lowerCamelCase      :      Dict         =guidance_scale > 1.0

											if isinstance(__lowercase						,    __lowercase			):
																		__lowerCamelCase      :      Any         =torch.cat(__lowercase						,    dim=0			)
											if isinstance(__lowercase						,    __lowercase			):
																		__lowerCamelCase      :      Dict         =torch.cat(__lowercase						,    dim=0			)
											if isinstance(__lowercase						,    __lowercase			):
																		__lowerCamelCase      :      Dict         =torch.cat(__lowercase						,    dim=0			)

											__lowerCamelCase      :      List[Any]         =image_embeds.shape[0] * num_images_per_prompt

											if do_classifier_free_guidance:
																		__lowerCamelCase      :      Optional[int]         =image_embeds.repeat_interleave(__lowercase						,    dim=0			)
																		__lowerCamelCase      :      Any         =negative_image_embeds.repeat_interleave(__lowercase						,    dim=0			)
																		__lowerCamelCase      :      List[Any]         =hint.repeat_interleave(__lowercase						,    dim=0			)

											__lowerCamelCase      :      List[str]         =torch.cat([negative_image_embeds, image_embeds]						,    dim=0			).to(dtype=self.unet.dtype						,    device=__lowercase			)
											__lowerCamelCase      :      List[str]         =torch.cat([hint, hint]						,    dim=0			).to(dtype=self.unet.dtype						,    device=__lowercase			)
											self.scheduler.set_timesteps(__lowercase						,    device=__lowercase			)
											__lowerCamelCase      :      str         =self.scheduler.timesteps

											__lowerCamelCase      :      Tuple         =self.movq.config.latent_channels

											__lowerCamelCase    ,						__lowerCamelCase      :      List[str]         =downscale_height_and_width(__lowercase						,    __lowercase						,    self.movq_scale_factor			)

											# create initial latent
											__lowerCamelCase      :      Optional[int]         =self.prepare_latents(
											    (batch_size, num_channels_latents, height, width)						,    image_embeds.dtype						,    __lowercase						,    __lowercase						,    __lowercase						,    self.scheduler						,    )

											for i, t in enumerate(self.progress_bar(__lowercase			)			):
																		# expand the latents if we are doing classifier free guidance
																		__lowerCamelCase      :      int         =torch.cat([latents] * 2			) if do_classifier_free_guidance else latents

																		__lowerCamelCase      :      List[Any]         ={'''image_embeds''': image_embeds, '''hint''': hint}
																		__lowerCamelCase      :      Union[str, Any]         =self.unet(
																		    sample=__lowercase						,    timestep=__lowercase						,    encoder_hidden_states=__lowercase						,    added_cond_kwargs=__lowercase						,    return_dict=__lowercase						,    )[0]

																		if do_classifier_free_guidance:
																									__lowerCamelCase    ,						__lowerCamelCase      :      Dict         =noise_pred.split(latents.shape[1]						,    dim=1			)
																									__lowerCamelCase    ,						__lowerCamelCase      :      int         =noise_pred.chunk(2			)
																									__lowerCamelCase    ,						__lowerCamelCase      :      Tuple         =variance_pred.chunk(2			)
																									__lowerCamelCase      :      Tuple         =noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
																									__lowerCamelCase      :      Any         =torch.cat([noise_pred, variance_pred_text]						,    dim=1			)

																		if not (
																		    hasattr(self.scheduler.config						,    '''variance_type'''			)
																		    and self.scheduler.config.variance_type in ["learned", "learned_range"]
																		):
																									__lowerCamelCase    ,						__lowerCamelCase      :      Any         =noise_pred.split(latents.shape[1]						,    dim=1			)

																		# compute the previous noisy sample x_t -> x_t-1
																		__lowerCamelCase      :      Tuple         =self.scheduler.step(
																		    __lowercase						,    __lowercase						,    __lowercase						,    generator=__lowercase						,    )[0]
											# post-processing
											__lowerCamelCase      :      List[str]         =self.movq.decode(__lowercase						,    force_not_quantize=__lowercase			)['''sample''']

											if output_type not in ["pt", "np", "pil"]:
																		raise ValueError(f'Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}'			)

											if output_type in ["np", "pil"]:
																		__lowerCamelCase      :      List[str]         =image * 0.5 + 0.5
																		__lowerCamelCase      :      Optional[int]         =image.clamp(0						,    1			)
																		__lowerCamelCase      :      Union[str, Any]         =image.cpu().permute(0						,    2						,    3						,    1			).float().numpy()

											if output_type == "pil":
																		__lowerCamelCase      :      Optional[Any]         =self.numpy_to_pil(__lowercase			)

											if not return_dict:
																		return (image,)

											return ImagePipelineOutput(images=__lowercase			)




