



'''simple docstring'''





from typing import Callable, List, Optional, Tuple, Union

import torch
from transformers import CLIPTextModel, CLIPTokenizer

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin, TransformeraDModel, VQModel
from ...schedulers import VQDiffusionScheduler
from ...utils import logging
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


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






class        lowerCAmelCase_(				SCREAMING_SNAKE_CASE_						, SCREAMING_SNAKE_CASE_     ):
  '''simple docstring'''




  @register_to_config
  def __init__(    self  ,__UpperCAmelCase  ,__UpperCAmelCase = None  ,__UpperCAmelCase = None      )    ->		Dict:
      super().__init__()

      lowerCAmelCase__ :  List[Any]							      =					learnable

      if self.learnable:
          assert hidden_size is not None, "learnable=True requires `hidden_size` to be set"
          assert length is not None, "learnable=True requires `length` to be set"

          lowerCAmelCase__ :  Dict							      =					torch.zeros(__UpperCAmelCase  ,__UpperCAmelCase      )
      else:
          lowerCAmelCase__ :  Optional[int]							      =					None

      lowerCAmelCase__ :  Union[str, Any]							      =					torch.nn.Parameter(__UpperCAmelCase      )







class        lowerCAmelCase_(				SCREAMING_SNAKE_CASE_     ):
  '''simple docstring'''




  __lowercase							:      VQModel
  __lowercase							:      CLIPTextModel
  __lowercase							:      CLIPTokenizer
  __lowercase							:      TransformeraDModel
  __lowercase							:      LearnedClassifierFreeSamplingEmbeddings
  __lowercase							:      VQDiffusionScheduler




  def __init__(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,)    ->		Union[str, Any]:
      super().__init__()

      self.register_modules(
          vqvae=__UpperCAmelCase  ,transformer=__UpperCAmelCase  ,text_encoder=__UpperCAmelCase  ,tokenizer=__UpperCAmelCase  ,scheduler=__UpperCAmelCase  ,learned_classifier_free_sampling_embeddings=__UpperCAmelCase  ,)




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase      )    ->		Optional[int]:
      lowerCAmelCase__ :  Optional[int]							      =					len(__UpperCAmelCase      ) if isinstance(__UpperCAmelCase  ,__UpperCAmelCase      ) else 1

      # get prompt text embeddings
      lowerCAmelCase__ :  int							      =					self.tokenizer(
          __UpperCAmelCase  ,padding="""max_length"""  ,max_length=self.tokenizer.model_max_length  ,return_tensors="""pt"""  ,)
      lowerCAmelCase__ :  Dict							      =					text_inputs.input_ids

      if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
          lowerCAmelCase__ :  List[Any]							      =					self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]      )
          logger.warning(
              """The following part of your input was truncated because CLIP can only handle sequences up to"""
              F""" {self.tokenizer.model_max_length} tokens: {removed_text}"""      )
          lowerCAmelCase__ :  Optional[Any]							      =					text_input_ids[:, : self.tokenizer.model_max_length]
      lowerCAmelCase__ :  Union[str, Any]							      =					self.text_encoder(text_input_ids.to(self.device      )      )[0]

      # NOTE: This additional step of normalizing the text embeddings is from VQ-Diffusion.
      # While CLIP does normalize the pooled output of the text transformer when combining
      # the image and text embeddings, CLIP does not directly normalize the last hidden state.
      #
      # CLIP normalizing the pooled output.
      # https://github.com/huggingface/transformers/blob/d92e22d1f28324f513f3080e5c47c071a3916721/src/transformers/models/clip/modeling_clip.py#L1052-L1053
      lowerCAmelCase__ :  List[str]							      =					prompt_embeds / prompt_embeds.norm(dim=-1  ,keepdim=__UpperCAmelCase      )

      # duplicate text embeddings for each generation per prompt
      lowerCAmelCase__ :  Any							      =					prompt_embeds.repeat_interleave(__UpperCAmelCase  ,dim=0      )

      if do_classifier_free_guidance:
          if self.learned_classifier_free_sampling_embeddings.learnable:
              lowerCAmelCase__ :  Dict							      =					self.learned_classifier_free_sampling_embeddings.embeddings
              lowerCAmelCase__ :  Optional[Any]							      =					negative_prompt_embeds.unsqueeze(0      ).repeat(__UpperCAmelCase  ,1  ,1      )
          else:
              lowerCAmelCase__ :  Optional[Any]							      =					[""""""] * batch_size

              lowerCAmelCase__ :  int							      =					text_input_ids.shape[-1]
              lowerCAmelCase__ :  str							      =					self.tokenizer(
                  __UpperCAmelCase  ,padding="""max_length"""  ,max_length=__UpperCAmelCase  ,truncation=__UpperCAmelCase  ,return_tensors="""pt"""  ,)
              lowerCAmelCase__ :  Union[str, Any]							      =					self.text_encoder(uncond_input.input_ids.to(self.device      )      )[0]
              # See comment for normalizing text embeddings
              lowerCAmelCase__ :  Union[str, Any]							      =					negative_prompt_embeds / negative_prompt_embeds.norm(dim=-1  ,keepdim=__UpperCAmelCase      )

          # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
          lowerCAmelCase__ :  Optional[int]							      =					negative_prompt_embeds.shape[1]
          lowerCAmelCase__ :  int							      =					negative_prompt_embeds.repeat(1  ,__UpperCAmelCase  ,1      )
          lowerCAmelCase__ :  Tuple							      =					negative_prompt_embeds.view(batch_size * num_images_per_prompt  ,__UpperCAmelCase  ,-1      )

          # For classifier free guidance, we need to do two forward passes.
          # Here we concatenate the unconditional and text embeddings into a single batch
          # to avoid doing two forward passes
          lowerCAmelCase__ :  List[Any]							      =					torch.cat([negative_prompt_embeds, prompt_embeds]      )

      return prompt_embeds




  @torch.no_grad()
  def __call__(    self  ,__UpperCAmelCase  ,__UpperCAmelCase = 100  ,__UpperCAmelCase = 5.0  ,__UpperCAmelCase = 1.0  ,__UpperCAmelCase = 1  ,__UpperCAmelCase = None  ,__UpperCAmelCase = None  ,__UpperCAmelCase = "pil"  ,__UpperCAmelCase = True  ,__UpperCAmelCase = None  ,__UpperCAmelCase = 1  ,)    ->		Union[ImagePipelineOutput, Tuple]:
      if isinstance(__UpperCAmelCase  ,__UpperCAmelCase      ):
          lowerCAmelCase__ :  Tuple							      =					1
      elif isinstance(__UpperCAmelCase  ,__UpperCAmelCase      ):
          lowerCAmelCase__ :  int							      =					len(__UpperCAmelCase      )
      else:
          raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(__UpperCAmelCase      )}"""      )

      lowerCAmelCase__ :  Optional[Any]							      =					batch_size * num_images_per_prompt

      lowerCAmelCase__ :  Union[str, Any]							      =					guidance_scale > 1.0

      lowerCAmelCase__ :  Any							      =					self._encode_prompt(__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase      )

      if (callback_steps is None) or (
          callback_steps is not None and (not isinstance(__UpperCAmelCase  ,__UpperCAmelCase      ) or callback_steps <= 0)
      ):
          raise ValueError(
              F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
              F""" {type(__UpperCAmelCase      )}."""      )

      # get the initial completely masked latents unless the user supplied it

      lowerCAmelCase__ :  Dict							      =					(batch_size, self.transformer.num_latent_pixels)
      if latents is None:
          lowerCAmelCase__ :  List[Any]							      =					self.transformer.num_vector_embeds - 1
          lowerCAmelCase__ :  int							      =					torch.full(__UpperCAmelCase  ,__UpperCAmelCase      ).to(self.device      )
      else:
          if latents.shape != latents_shape:
              raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""      )
          if (latents < 0).any() or (latents >= self.transformer.num_vector_embeds).any():
              raise ValueError(
                  """Unexpected latents value(s). All latents be valid embedding indices i.e. in the range 0,"""
                  F""" {self.transformer.num_vector_embeds - 1} (inclusive)."""      )
          lowerCAmelCase__ :  List[str]							      =					latents.to(self.device      )

      # set timesteps
      self.scheduler.set_timesteps(__UpperCAmelCase  ,device=self.device      )

      lowerCAmelCase__ :  Any							      =					self.scheduler.timesteps.to(self.device      )

      lowerCAmelCase__ :  int							      =					latents

      for i, t in enumerate(self.progress_bar(__UpperCAmelCase      )      ):
          # expand the sample if we are doing classifier free guidance
          lowerCAmelCase__ :  Union[str, Any]							      =					torch.cat([sample] * 2      ) if do_classifier_free_guidance else sample

          # predict the un-noised image
          # model_output == `log_p_x_0`
          lowerCAmelCase__ :  List[str]							      =					self.transformer(__UpperCAmelCase  ,encoder_hidden_states=__UpperCAmelCase  ,timestep=__UpperCAmelCase      ).sample

          if do_classifier_free_guidance:
              lowerCAmelCase__      ,   lowerCAmelCase__ :  List[str]							      =					model_output.chunk(2      )
              lowerCAmelCase__ :  List[Any]							      =					model_output_uncond + guidance_scale * (model_output_text - model_output_uncond)
              model_output -= torch.logsumexp(__UpperCAmelCase  ,dim=1  ,keepdim=__UpperCAmelCase      )

          lowerCAmelCase__ :  List[str]							      =					self.truncate(__UpperCAmelCase  ,__UpperCAmelCase      )

          # remove `log(0)`'s (`-inf`s)
          lowerCAmelCase__ :  Optional[Any]							      =					model_output.clamp(-70      )

          # compute the previous noisy sample x_t -> x_t-1
          lowerCAmelCase__ :  List[str]							      =					self.scheduler.step(__UpperCAmelCase  ,timestep=__UpperCAmelCase  ,sample=__UpperCAmelCase  ,generator=__UpperCAmelCase      ).prev_sample

          # call the callback, if provided
          if callback is not None and i % callback_steps == 0:
              callback(__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase      )

      lowerCAmelCase__ :  Tuple							      =					self.vqvae.config.vq_embed_dim
      lowerCAmelCase__ :  Optional[int]							      =					(batch_size, self.transformer.height, self.transformer.width, embedding_channels)
      lowerCAmelCase__ :  List[str]							      =					self.vqvae.quantize.get_codebook_entry(__UpperCAmelCase  ,shape=__UpperCAmelCase      )
      lowerCAmelCase__ :  int							      =					self.vqvae.decode(__UpperCAmelCase  ,force_not_quantize=__UpperCAmelCase      ).sample

      lowerCAmelCase__ :  Dict							      =					(image / 2 + 0.5).clamp(0  ,1      )
      lowerCAmelCase__ :  Union[str, Any]							      =					image.cpu().permute(0  ,2  ,3  ,1      ).numpy()

      if output_type == "pil":
          lowerCAmelCase__ :  Dict							      =					self.numpy_to_pil(__UpperCAmelCase      )

      if not return_dict:
          return (image,)

      return ImagePipelineOutput(images=__UpperCAmelCase      )






  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase      )    ->		torch.FloatTensor:
      lowerCAmelCase__      ,   lowerCAmelCase__ :  Any							      =					torch.sort(__UpperCAmelCase  ,1  ,descending=__UpperCAmelCase      )
      lowerCAmelCase__ :  Any							      =					torch.exp(__UpperCAmelCase      )
      lowerCAmelCase__ :  Dict							      =					sorted_p_x_0.cumsum(dim=1      ) < truncation_rate

      # Ensure that at least the largest probability is not zeroed out
      lowerCAmelCase__ :  Dict							      =					torch.full_like(keep_mask[:, 0:1, :]  ,__UpperCAmelCase      )
      lowerCAmelCase__ :  Optional[int]							      =					torch.cat((all_true, keep_mask)  ,dim=1      )
      lowerCAmelCase__ :  str							      =					keep_mask[:, :-1, :]

      lowerCAmelCase__ :  Optional[Any]							      =					keep_mask.gather(1  ,indices.argsort(1      )      )

      lowerCAmelCase__ :  List[Any]							      =					log_p_x_0.clone()

      lowerCAmelCase__ :  Optional[Any]							      =					-torch.inf  # -inf = log(0)

      return rv







'''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        lowerCAmelCase_:
  '''simple docstring'''




  def __init__(    self  ,__UpperCAmelCase = None  ,__UpperCAmelCase = None  ,__UpperCAmelCase=None  ,__UpperCAmelCase=None      )    ->		str:
      if not conversation_id:
          lowerCAmelCase__ :  List[str]							      =					uuid.uuida()
      if past_user_inputs is None:
          lowerCAmelCase__ :  List[Any]							      =					[]
      if generated_responses is None:
          lowerCAmelCase__ :  str							      =					[]

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




  def __eq__(    self  ,__UpperCAmelCase      )    ->		Dict:
      if not isinstance(__UpperCAmelCase  ,__UpperCAmelCase      ):
          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  ,__UpperCAmelCase  ,__UpperCAmelCase = False      )    ->		Optional[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}\"."""      )
              lowerCAmelCase__ :  Optional[int]							      =					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[Any]							      =					text




  def 							UpperCAmelCase_	(    self      )    ->		List[Any]:
      if self.new_user_input:
          self.past_user_inputs.append(self.new_user_input      )
      lowerCAmelCase__ :  Union[str, Any]							      =					None




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase      )    ->		Tuple:
      self.generated_responses.append(__UpperCAmelCase      )




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







@add_end_docstrings(
    SCREAMING_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        lowerCAmelCase_(				SCREAMING_SNAKE_CASE_     ):
  '''simple docstring'''




  def __init__(    self  ,*__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Tuple:
      super().__init__(*__UpperCAmelCase  ,**__UpperCAmelCase      )
      if self.tokenizer.pad_token_id is None:
          lowerCAmelCase__ :  Tuple							      =					self.tokenizer.eos_token




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase=None  ,__UpperCAmelCase=None  ,__UpperCAmelCase=None  ,**__UpperCAmelCase      )    ->		Optional[int]:
      lowerCAmelCase__ :  List[Any]							      =					{}
      lowerCAmelCase__ :  Optional[int]							      =					{}
      lowerCAmelCase__ :  List[str]							      =					{}

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

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

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




  def __call__(    self  ,__UpperCAmelCase  ,__UpperCAmelCase=0  ,**__UpperCAmelCase      )    ->		List[str]:
      lowerCAmelCase__ :  Optional[int]							      =					super().__call__(__UpperCAmelCase  ,num_workers=__UpperCAmelCase  ,**__UpperCAmelCase      )
      if isinstance(__UpperCAmelCase  ,__UpperCAmelCase      ) and len(__UpperCAmelCase      ) == 1:
          return outputs[0]
      return outputs




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase=32      )    ->		Dict[str, Any]:
      if not isinstance(__UpperCAmelCase  ,__UpperCAmelCase      ):
          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__ :  str							      =					self.tokenizer._build_conversation_input_ids(__UpperCAmelCase      )
      else:
          # If the tokenizer cannot handle conversations, we default to only the old version
          lowerCAmelCase__ :  List[Any]							      =					self._legacy_parse_and_tokenize(__UpperCAmelCase      )

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




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase=10  ,**__UpperCAmelCase      )    ->		Dict:
      lowerCAmelCase__ :  Optional[Any]							      =					generate_kwargs.get("""max_length"""  ,self.model.config.max_length      )

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




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






  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase      )    ->		Dict:
      lowerCAmelCase__ :  Dict							      =					self.tokenizer.eos_token_id
      lowerCAmelCase__ :  int							      =					[]
      for is_user, text in conversation.iter_texts():
          if eos_token_id is not None:
              input_ids.extend(self.tokenizer.encode(__UpperCAmelCase  ,add_special_tokens=__UpperCAmelCase      ) + [eos_token_id]      )
          else:
              input_ids.extend(self.tokenizer.encode(__UpperCAmelCase  ,add_special_tokens=__UpperCAmelCase      )      )

      if len(__UpperCAmelCase      ) > self.tokenizer.model_max_length:
          lowerCAmelCase__ :  Optional[Any]							      =					input_ids[-self.tokenizer.model_max_length :]
      return input_ids


