


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


A__						:   Tuple				=    logging.get_logger(__name__)





class   UpperCAmelCase_    :





    """simple docstring"""






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

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







    def __eq__(				self     ,	SCREAMING_SNAKE_CASE_      )							->	int:
         if not isinstance(SCREAMING_SNAKE_CASE_     ,	SCREAMING_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  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_ = False      )							->	Tuple:
         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[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  :      List[str]														=						text







    def  lowercase_							(				self      )							->	str:
         if self.new_user_input:
              self.past_user_inputs.append(self.new_user_input      )
         __lowerCamelCase  :      Any														=						None







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







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







@add_end_docstrings(
    _UpperCAmelCase   ,						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   UpperCAmelCase_    (_UpperCAmelCase    ):





    """simple docstring"""






    def __init__(				self     ,	*SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )							->	Optional[int]:
         super().__init__(*SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )
         if self.tokenizer.pad_token_id is None:
              __lowerCamelCase  :      Tuple														=						self.tokenizer.eos_token







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_=None     ,	SCREAMING_SNAKE_CASE_=None     ,	SCREAMING_SNAKE_CASE_=None     ,	**SCREAMING_SNAKE_CASE_      )							->	str:
         __lowerCamelCase  :      Union[str, Any]														=						{}
         __lowerCamelCase  :      int														=						{}
         __lowerCamelCase  :      Dict														=						{}

         if min_length_for_response is not None:
              __lowerCamelCase  :      Union[str, Any]														=						min_length_for_response
         if minimum_tokens is not None:
              __lowerCamelCase  :      Any														=						minimum_tokens

         if "max_length" in generate_kwargs:
              __lowerCamelCase  :      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:
              __lowerCamelCase  :      str														=						clean_up_tokenization_spaces

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







    def __call__(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_=0     ,	**SCREAMING_SNAKE_CASE_      )							->	Any:
         __lowerCamelCase  :      int														=						super().__call__(SCREAMING_SNAKE_CASE_     ,	num_workers=SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )
         if isinstance(SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_      ) and len(SCREAMING_SNAKE_CASE_      ) == 1:
              return outputs[0]
         return outputs







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

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







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_=10     ,	**SCREAMING_SNAKE_CASE_      )							->	List[str]:
         __lowerCamelCase  :      Dict														=						generate_kwargs.get('max_length'     ,	self.model.config.max_length      )

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







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_=True      )							->	Dict:
         __lowerCamelCase  :      Any														=						model_outputs['output_ids']
         __lowerCamelCase  :      Optional[int]														=						self.tokenizer.decode(
             output_ids[0]     ,	skip_special_tokens=SCREAMING_SNAKE_CASE_     ,	clean_up_tokenization_spaces=SCREAMING_SNAKE_CASE_     ,	)
         __lowerCamelCase  :      int														=						model_outputs['conversation']
         conversation.mark_processed()
         conversation.append_response(SCREAMING_SNAKE_CASE_      )
         return conversation







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_      )							->	Dict:
         __lowerCamelCase  :      Tuple														=						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(SCREAMING_SNAKE_CASE_     ,	add_special_tokens=SCREAMING_SNAKE_CASE_      ) + [eos_token_id]      )
              else:
                   input_ids.extend(self.tokenizer.encode(SCREAMING_SNAKE_CASE_     ,	add_special_tokens=SCREAMING_SNAKE_CASE_      )      )

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







import torch
from torch import nn

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin




class   _UpperCamelCase (		_UpperCAmelCase		,_UpperCAmelCase					):


			"""simple docstring"""






			@register_to_config
			def __init__(  self ,  *,
			    lowerCAmelCase__ = 4 ,  lowerCAmelCase__ = 7_68 ,  lowerCAmelCase__ ,  lowerCAmelCase__ ,  )	->   Dict:

							'''simple docstring'''


							super().__init__()

							__lowercase					     =				nn.Parameter(torch.zeros(lowerCAmelCase__		)		)

							# parameters for additional clip time embeddings
							__lowercase					     =				nn.Linear(lowerCAmelCase__ ,  lowerCAmelCase__		)
							__lowercase					     =				nn.Linear(lowerCAmelCase__ ,  lowerCAmelCase__		)

							# parameters for encoder hidden states
							__lowercase					     =				clip_extra_context_tokens
							__lowercase					     =				nn.Linear(
							    lowerCAmelCase__ ,  self.clip_extra_context_tokens * cross_attention_dim		)
							__lowercase					     =				nn.Linear(lowerCAmelCase__ ,  lowerCAmelCase__		)
							__lowercase					     =				nn.LayerNorm(lowerCAmelCase__		)





			def 			_SCREAMING_SNAKE_CASE     (  self ,  *, lowerCAmelCase__ ,  lowerCAmelCase__ ,  lowerCAmelCase__ ,  lowerCAmelCase__		)	->   List[str]:

							'''simple docstring'''


							if do_classifier_free_guidance:
											# Add the classifier free guidance embeddings to the image embeddings
											__lowercase					     =				image_embeddings.shape[0]
											__lowercase					     =				self.learned_classifier_free_guidance_embeddings.unsqueeze(0		)
											__lowercase					     =				classifier_free_guidance_embeddings.expand(
											    lowerCAmelCase__ ,  -1		)
											__lowercase					     =				torch.cat([classifier_free_guidance_embeddings, image_embeddings] ,  dim=0		)

							# The image embeddings batch size and the text embeddings batch size are equal
							assert image_embeddings.shape[0] == prompt_embeds.shape[0]

							__lowercase					     =				prompt_embeds.shape[0]

							# "Specifically, we modify the architecture described in Nichol et al. (2021) by projecting and
							# adding CLIP embeddings to the existing timestep embedding, ...
							__lowercase					     =				self.embedding_proj(lowerCAmelCase__		)
							__lowercase					     =				self.clip_image_embeddings_project_to_time_embeddings(lowerCAmelCase__		)
							__lowercase					     =				time_projected_image_embeddings + time_projected_prompt_embeds

							# ... and by projecting CLIP embeddings into four
							# extra tokens of context that are concatenated to the sequence of outputs from the GLIDE text encoder"
							__lowercase					     =				self.clip_extra_context_tokens_proj(lowerCAmelCase__		)
							__lowercase					     =				clip_extra_context_tokens.reshape(lowerCAmelCase__ ,  -1 ,  self.clip_extra_context_tokens		)
							__lowercase					     =				clip_extra_context_tokens.permute(0 ,  2 ,  1		)

							__lowercase					     =				self.encoder_hidden_states_proj(lowerCAmelCase__		)
							__lowercase					     =				self.text_encoder_hidden_states_norm(lowerCAmelCase__		)
							__lowercase					     =				torch.cat([clip_extra_context_tokens, text_encoder_hidden_states] ,  dim=1		)

							return text_encoder_hidden_states, additive_clip_time_embeddings