"""simple docstring"""


from dataclasses import dataclass
from typing import Dict, Optional, Union

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .attention_processor import AttentionProcessor, AttnProcessor
from .embeddings import TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin


@dataclass
class        a				(		lowerCAmelCase_   ):
    _snake_case     :  torch.FloatTensor



class        a				(		lowerCAmelCase_	,							lowerCAmelCase_   ):


    @register_to_config
    def __init__(					self	:  List[str]   ,				__lowerCAmelCase	:  int = 32   ,				__lowerCAmelCase	:  int = 64   ,				__lowerCAmelCase	:  int = 20   ,				__lowerCAmelCase	:  int = 768   ,				__lowerCAmelCase	:  Optional[int]=77   ,				__lowerCAmelCase	:  Tuple=4   ,				__lowerCAmelCase	:  float = 0.0   ,				__lowerCAmelCase	:  str = "silu"   ,				__lowerCAmelCase	:  Optional[str] = None   ,				__lowerCAmelCase	:  Optional[str] = None   ,				__lowerCAmelCase	:  Optional[str] = "linear"   ,				__lowerCAmelCase	:  Optional[str] = "prd"   ,				__lowerCAmelCase	:  Optional[int] = None   ,				__lowerCAmelCase	:  Optional[int] = None   ,				__lowerCAmelCase	:  Optional[int] = None   ,				):
      super().__init__()
      _UpperCAmelCase						 =      num_attention_heads
      _UpperCAmelCase						 =      attention_head_dim
      _UpperCAmelCase						 =      num_attention_heads * attention_head_dim
      _UpperCAmelCase						 =      additional_embeddings

      _UpperCAmelCase						 =      time_embed_dim or inner_dim
      _UpperCAmelCase						 =      embedding_proj_dim or embedding_dim
      _UpperCAmelCase						 =      clip_embed_dim or embedding_dim

      _UpperCAmelCase						 =      Timesteps(__lowerCAmelCase   ,				__lowerCAmelCase   ,				0					)
      _UpperCAmelCase						 =      TimestepEmbedding(__lowerCAmelCase   ,				__lowerCAmelCase   ,				out_dim=__lowerCAmelCase   ,				act_fn=__lowerCAmelCase					)

      _UpperCAmelCase						 =      nn.Linear(__lowerCAmelCase   ,				__lowerCAmelCase					)

      if embedding_proj_norm_type is None:
        _UpperCAmelCase						 =      None
      elif embedding_proj_norm_type == "layer":
        _UpperCAmelCase						 =      nn.LayerNorm(__lowerCAmelCase					)
      else:
        raise ValueError(f'''unsupported embedding_proj_norm_type: {embedding_proj_norm_type}'''					)

      _UpperCAmelCase						 =      nn.Linear(__lowerCAmelCase   ,				__lowerCAmelCase					)

      if encoder_hid_proj_type is None:
        _UpperCAmelCase						 =      None
      elif encoder_hid_proj_type == "linear":
        _UpperCAmelCase						 =      nn.Linear(__lowerCAmelCase   ,				__lowerCAmelCase					)
      else:
        raise ValueError(f'''unsupported encoder_hid_proj_type: {encoder_hid_proj_type}'''					)

      _UpperCAmelCase						 =      nn.Parameter(torch.zeros(1   ,				num_embeddings + additional_embeddings   ,				__lowerCAmelCase					)					)

      if added_emb_type == "prd":
        _UpperCAmelCase						 =      nn.Parameter(torch.zeros(1   ,				1   ,				__lowerCAmelCase					)					)
      elif added_emb_type is None:
        _UpperCAmelCase						 =      None
      else:
        raise ValueError(
            f'''`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `\'prd\'` or `None`.'''					)

      _UpperCAmelCase						 =      nn.ModuleList(
          [
              BasicTransformerBlock(
                  __lowerCAmelCase   ,				__lowerCAmelCase   ,				__lowerCAmelCase   ,				dropout=__lowerCAmelCase   ,				activation_fn="""gelu"""   ,				attention_bias=__lowerCAmelCase   ,				)
              for d in range(__lowerCAmelCase					)
          ]					)

      if norm_in_type == "layer":
        _UpperCAmelCase						 =      nn.LayerNorm(__lowerCAmelCase					)
      elif norm_in_type is None:
        _UpperCAmelCase						 =      None
      else:
        raise ValueError(f'''Unsupported norm_in_type: {norm_in_type}.'''					)

      _UpperCAmelCase						 =      nn.LayerNorm(__lowerCAmelCase					)

      _UpperCAmelCase						 =      nn.Linear(__lowerCAmelCase   ,				__lowerCAmelCase					)

      _UpperCAmelCase						 =      torch.full(
          [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]   ,				-10_000.0					)
      causal_attention_mask.triu_(1					)
      _UpperCAmelCase						 =      causal_attention_mask[None, ...]
      self.register_buffer("""causal_attention_mask"""   ,				__lowerCAmelCase   ,				persistent=__lowerCAmelCase					)

      _UpperCAmelCase						 =      nn.Parameter(torch.zeros(1   ,				__lowerCAmelCase					)					)
      _UpperCAmelCase						 =      nn.Parameter(torch.zeros(1   ,				__lowerCAmelCase					)					)


    @property
    # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
    def      lowerCAmelCase_  (					self	:  Union[str, Any]					):
      _UpperCAmelCase						 =      {}

      def fn_recursive_add_processors(__lowerCAmelCase	:  str   ,				__lowerCAmelCase	:  torch.nn.Module   ,				__lowerCAmelCase	:  Dict[str, AttentionProcessor]					):
        if hasattr(__lowerCAmelCase   ,				"""set_processor"""					):
          _UpperCAmelCase						 =      module.processor

        for sub_name, child in module.named_children():
          fn_recursive_add_processors(f'''{name}.{sub_name}'''   ,				__lowerCAmelCase   ,				__lowerCAmelCase					)

        return processors

      for name, module in self.named_children():
        fn_recursive_add_processors(__lowerCAmelCase   ,				__lowerCAmelCase   ,				__lowerCAmelCase					)

      return processors


    def      lowerCAmelCase_  (					self	:  int   ,				__lowerCAmelCase	:  Union[AttentionProcessor, Dict[str, AttentionProcessor]]					):
      _UpperCAmelCase						 =      len(self.attn_processors.keys()					)

      if isinstance(__lowerCAmelCase   ,				__lowerCAmelCase					) and len(__lowerCAmelCase					) != count:
        raise ValueError(
            f'''A dict of processors was passed, but the number of processors {len(__lowerCAmelCase					)} does not match the'''
            f''' number of attention layers: {count}. Please make sure to pass {count} processor classes.'''					)

      def fn_recursive_attn_processor(__lowerCAmelCase	:  str   ,				__lowerCAmelCase	:  torch.nn.Module   ,				__lowerCAmelCase	:  List[str]					):
        if hasattr(__lowerCAmelCase   ,				"""set_processor"""					):
          if not isinstance(__lowerCAmelCase   ,				__lowerCAmelCase					):
            module.set_processor(__lowerCAmelCase					)
          else:
            module.set_processor(processor.pop(f'''{name}.processor'''					)					)

        for sub_name, child in module.named_children():
          fn_recursive_attn_processor(f'''{name}.{sub_name}'''   ,				__lowerCAmelCase   ,				__lowerCAmelCase					)

      for name, module in self.named_children():
        fn_recursive_attn_processor(__lowerCAmelCase   ,				__lowerCAmelCase   ,				__lowerCAmelCase					)


    def      lowerCAmelCase_  (					self	:  Tuple					):
      self.set_attn_processor(AttnProcessor()					)


    def      lowerCAmelCase_  (					self	:  Any   ,				__lowerCAmelCase	:  Optional[int]   ,				__lowerCAmelCase	:  Union[torch.Tensor, float, int]   ,				__lowerCAmelCase	:  torch.FloatTensor   ,				__lowerCAmelCase	:  Optional[torch.FloatTensor] = None   ,				__lowerCAmelCase	:  Optional[torch.BoolTensor] = None   ,				__lowerCAmelCase	:  bool = True   ,				):
      _UpperCAmelCase						 =      hidden_states.shape[0]

      _UpperCAmelCase						 =      timestep
      if not torch.is_tensor(__lowerCAmelCase					):
        _UpperCAmelCase						 =      torch.tensor([timesteps]   ,				dtype=torch.long   ,				device=hidden_states.device					)
      elif torch.is_tensor(__lowerCAmelCase					) and len(timesteps.shape					) == 0:
        _UpperCAmelCase						 =      timesteps[None].to(hidden_states.device					)

      # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
      _UpperCAmelCase						 =      timesteps * torch.ones(__lowerCAmelCase   ,				dtype=timesteps.dtype   ,				device=timesteps.device					)

      _UpperCAmelCase						 =      self.time_proj(__lowerCAmelCase					)

      # timesteps does not contain any weights and will always return f32 tensors
      # but time_embedding might be fp16, so we need to cast here.
      _UpperCAmelCase						 =      timesteps_projected.to(dtype=self.dtype					)
      _UpperCAmelCase						 =      self.time_embedding(__lowerCAmelCase					)

      if self.embedding_proj_norm is not None:
        _UpperCAmelCase						 =      self.embedding_proj_norm(__lowerCAmelCase					)

      _UpperCAmelCase						 =      self.embedding_proj(__lowerCAmelCase					)
      if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
        _UpperCAmelCase						 =      self.encoder_hidden_states_proj(__lowerCAmelCase					)
      elif self.encoder_hidden_states_proj is not None and encoder_hidden_states is None:
        raise ValueError("""`encoder_hidden_states_proj` requires `encoder_hidden_states` to be set"""					)

      _UpperCAmelCase						 =      self.proj_in(__lowerCAmelCase					)

      _UpperCAmelCase						 =      self.positional_embedding.to(hidden_states.dtype					)

      _UpperCAmelCase						 =      []
      _UpperCAmelCase						 =      0

      if encoder_hidden_states is not None:
        additional_embeds.append(__lowerCAmelCase					)
        additional_embeddings_len += encoder_hidden_states.shape[1]

      if len(proj_embeddings.shape					) == 2:
        _UpperCAmelCase						 =      proj_embeddings[:, None, :]

      if len(hidden_states.shape					) == 2:
        _UpperCAmelCase						 =      hidden_states[:, None, :]

      _UpperCAmelCase						 =      additional_embeds + [
          proj_embeddings,
          time_embeddings[:, None, :],
          hidden_states,
      ]

      if self.prd_embedding is not None:
        _UpperCAmelCase						 =      self.prd_embedding.to(hidden_states.dtype					).expand(__lowerCAmelCase   ,				-1   ,				-1					)
        additional_embeds.append(__lowerCAmelCase					)

      _UpperCAmelCase						 =      torch.cat(
          __lowerCAmelCase   ,				dim=1   ,				)

      # Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
      _UpperCAmelCase						 =      additional_embeddings_len + proj_embeddings.shape[1] + 1
      if positional_embeddings.shape[1] < hidden_states.shape[1]:
        _UpperCAmelCase						 =      F.pad(
            __lowerCAmelCase   ,				(
                0,
                0,
                additional_embeddings_len,
                self.prd_embedding.shape[1] if self.prd_embedding is not None else 0,
            )   ,				value=0.0   ,				)

      _UpperCAmelCase						 =      hidden_states + positional_embeddings

      if attention_mask is not None:
        _UpperCAmelCase						 =      (1 - attention_mask.to(hidden_states.dtype					)) * -10_000.0
        _UpperCAmelCase						 =      F.pad(__lowerCAmelCase   ,				(0, self.additional_embeddings)   ,				value=0.0					)
        _UpperCAmelCase						 =      (attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype					)
        _UpperCAmelCase						 =      attention_mask.repeat_interleave(self.config.num_attention_heads   ,				dim=0					)

      if self.norm_in is not None:
        _UpperCAmelCase						 =      self.norm_in(__lowerCAmelCase					)

      for block in self.transformer_blocks:
        _UpperCAmelCase						 =      block(__lowerCAmelCase   ,				attention_mask=__lowerCAmelCase					)

      _UpperCAmelCase						 =      self.norm_out(__lowerCAmelCase					)

      if self.prd_embedding is not None:
        _UpperCAmelCase						 =      hidden_states[:, -1]
      else:
        _UpperCAmelCase						 =      hidden_states[:, additional_embeddings_len:]

      _UpperCAmelCase						 =      self.proj_to_clip_embeddings(__lowerCAmelCase					)

      if not return_dict:
        return (predicted_image_embedding,)

      return PriorTransformerOutput(predicted_image_embedding=__lowerCAmelCase					)


    def      lowerCAmelCase_  (					self	:  Optional[Any]   ,				__lowerCAmelCase	:  Dict					):
      _UpperCAmelCase						 =      (prior_latents * self.clip_std) + self.clip_mean
      return prior_latents

"""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        a				:


    def __init__(					self	:  Tuple   ,				__lowerCAmelCase	:  str = None   ,				__lowerCAmelCase	:  uuid.UUID = None   ,				__lowerCAmelCase	:  Any=None   ,				__lowerCAmelCase	:  Tuple=None					):
      if not conversation_id:
        _UpperCAmelCase						 =      uuid.uuida()
      if past_user_inputs is None:
        _UpperCAmelCase						 =      []
      if generated_responses is None:
        _UpperCAmelCase						 =      []

      _UpperCAmelCase						 =      conversation_id
      _UpperCAmelCase						 =      past_user_inputs
      _UpperCAmelCase						 =      generated_responses
      _UpperCAmelCase						 =      text


    def __eq__(					self	:  Optional[Any]   ,				__lowerCAmelCase	:  Union[str, Any]					):
      if not isinstance(__lowerCAmelCase   ,				__lowerCAmelCase					):
        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]   ,				__lowerCAmelCase	:  str   ,				__lowerCAmelCase	:  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}".'''					)
          _UpperCAmelCase						 =      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:
        _UpperCAmelCase						 =      text


    def      lowerCAmelCase_  (					self	:  int					):
      if self.new_user_input:
        self.past_user_inputs.append(self.new_user_input					)
      _UpperCAmelCase						 =      None


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


    def      lowerCAmelCase_  (					self	:  Dict					):
      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	:  Optional[int]					):
      _UpperCAmelCase						 =      f'''Conversation id: {self.uuid} \n'''
      for is_user, text in self.iter_texts():
        _UpperCAmelCase						 =      """user""" if is_user else """bot"""
        output += f'''{name} >> {text} \n'''
      return output



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


    def __init__(					self	:  List[Any]   ,				*__lowerCAmelCase	:  List[Any]   ,				**__lowerCAmelCase	:  Tuple					):
      super().__init__(*__lowerCAmelCase   ,				**__lowerCAmelCase					)
      if self.tokenizer.pad_token_id is None:
        _UpperCAmelCase						 =      self.tokenizer.eos_token


    def      lowerCAmelCase_  (					self	:  Optional[Any]   ,				__lowerCAmelCase	:  Dict=None   ,				__lowerCAmelCase	:  List[Any]=None   ,				__lowerCAmelCase	:  Optional[int]=None   ,				**__lowerCAmelCase	:  str					):
      _UpperCAmelCase						 =      {}
      _UpperCAmelCase						 =      {}
      _UpperCAmelCase						 =      {}

      if min_length_for_response is not None:
        _UpperCAmelCase						 =      min_length_for_response
      if minimum_tokens is not None:
        _UpperCAmelCase						 =      minimum_tokens

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

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


    def __call__(					self	:  Tuple   ,				__lowerCAmelCase	:  Union[Conversation, List[Conversation]]   ,				__lowerCAmelCase	:  int=0   ,				**__lowerCAmelCase	:  List[str]					):
      _UpperCAmelCase						 =      super().__call__(__lowerCAmelCase   ,				num_workers=__lowerCAmelCase   ,				**__lowerCAmelCase					)
      if isinstance(__lowerCAmelCase   ,				__lowerCAmelCase					) and len(__lowerCAmelCase					) == 1:
        return outputs[0]
      return outputs


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

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


    def      lowerCAmelCase_  (					self	:  Dict   ,				__lowerCAmelCase	:  Union[str, Any]   ,				__lowerCAmelCase	:  List[str]=10   ,				**__lowerCAmelCase	:  Union[str, Any]					):
      _UpperCAmelCase						 =      generate_kwargs.get("""max_length"""   ,				self.model.config.max_length					)

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


    def      lowerCAmelCase_  (					self	:  str   ,				__lowerCAmelCase	:  Any   ,				__lowerCAmelCase	:  Optional[Any]=True					):
      _UpperCAmelCase						 =      model_outputs["""output_ids"""]
      _UpperCAmelCase						 =      self.tokenizer.decode(
          output_ids[0]   ,				skip_special_tokens=__lowerCAmelCase   ,				clean_up_tokenization_spaces=__lowerCAmelCase   ,				)
      _UpperCAmelCase						 =      model_outputs["""conversation"""]
      conversation.mark_processed()
      conversation.append_response(__lowerCAmelCase					)
      return conversation


    def      lowerCAmelCase_  (					self	:  int   ,				__lowerCAmelCase	:  Conversation					):
      _UpperCAmelCase						 =      self.tokenizer.eos_token_id
      _UpperCAmelCase						 =      []
      for is_user, text in conversation.iter_texts():
        if eos_token_id is not None:
          input_ids.extend(self.tokenizer.encode(__lowerCAmelCase   ,				add_special_tokens=__lowerCAmelCase					) + [eos_token_id]					)
        else:
          input_ids.extend(self.tokenizer.encode(__lowerCAmelCase   ,				add_special_tokens=__lowerCAmelCase					)					)

      if len(__lowerCAmelCase					) > self.tokenizer.model_max_length:
        _UpperCAmelCase						 =      input_ids[-self.tokenizer.model_max_length :]
      return input_ids
