


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 							lowercase  :







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

           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 :						Optional[int]   ,   _lowercase :						Dict      ):
           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 :						Dict   ,   _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}\"."""      )
                       SCREAMING_SNAKE_CASE__       :  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:
                 SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		text







     def 							lowercase__       (						self :						int      ):
           if self.new_user_input:
                 self.past_user_inputs.append(self.new_user_input      )
           SCREAMING_SNAKE_CASE__       :  int						     =		None







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







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




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







     def __init__(						self :						Dict   ,   *_lowercase :						List[str]   ,   **_lowercase :						List[Any]      ):
           super().__init__(*_lowercase   ,   **_lowercase      )
           if self.tokenizer.pad_token_id is None:
                 SCREAMING_SNAKE_CASE__       :  Tuple						     =		self.tokenizer.eos_token







     def 							lowercase__       (						self :						str   ,   _lowercase :						List[Any]=None   ,   _lowercase :						Tuple=None   ,   _lowercase :						Tuple=None   ,   **_lowercase :						List[Any]      ):
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		{}
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		{}
           SCREAMING_SNAKE_CASE__       :  int						     =		{}

           if min_length_for_response is not None:
                 SCREAMING_SNAKE_CASE__       :  str						     =		min_length_for_response
           if minimum_tokens is not None:
                 SCREAMING_SNAKE_CASE__       :  Dict						     =		minimum_tokens

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

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







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







     def 							lowercase__       (						self :						int   ,   _lowercase :						Conversation   ,   _lowercase :						str=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'''      ):
                 SCREAMING_SNAKE_CASE__       :  Dict						     =		self.tokenizer._build_conversation_input_ids(_lowercase      )
           else:
                 # If the tokenizer cannot handle conversations, we default to only the old version
                 SCREAMING_SNAKE_CASE__       :  Any						     =		self._legacy_parse_and_tokenize(_lowercase      )

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







     def 							lowercase__       (						self :						str   ,   _lowercase :						Optional[Any]   ,   _lowercase :						Tuple=10   ,   **_lowercase :						Optional[Any]      ):
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		generate_kwargs.get('''max_length'''   ,   self.model.config.max_length      )

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







     def 							lowercase__       (						self :						Optional[Any]   ,   _lowercase :						str   ,   _lowercase :						int=True      ):
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		model_outputs['''output_ids''']
           SCREAMING_SNAKE_CASE__       :  str						     =		self.tokenizer.decode(
               output_ids[0]   ,   skip_special_tokens=_lowercase   ,   clean_up_tokenization_spaces=_lowercase   ,   )
           SCREAMING_SNAKE_CASE__       :  str						     =		model_outputs['''conversation''']
           conversation.mark_processed()
           conversation.append_response(_lowercase      )
           return conversation







     def 							lowercase__       (						self :						Union[str, Any]   ,   _lowercase :						Conversation      ):
           SCREAMING_SNAKE_CASE__       :  str						     =		self.tokenizer.eos_token_id
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		[]
           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:
                 SCREAMING_SNAKE_CASE__       :  Tuple						     =		input_ids[-self.tokenizer.model_max_length :]
           return input_ids













"""simple docstring"""





from dataclasses import dataclass
from typing import Optional

import torch
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .modeling_utils import ModelMixin






@dataclass
class    _lowerCAmelCase  (	a			):
       """simple docstring"""




       __magic_name__	:torch.FloatTensor






class    _lowerCAmelCase  (	a    ,  a			):
       """simple docstring"""




       @register_to_config
       def __init__(       self       ,					__UpperCAmelCase = 1_6       ,					__UpperCAmelCase = 8_8       ,					__UpperCAmelCase = None       ,					__UpperCAmelCase = None       ,					__UpperCAmelCase = 1       ,					__UpperCAmelCase = 0.0       ,					__UpperCAmelCase = 3_2       ,					__UpperCAmelCase = None       ,					__UpperCAmelCase = False       ,					__UpperCAmelCase = None       ,					__UpperCAmelCase = "geglu"       ,					__UpperCAmelCase = True       ,					__UpperCAmelCase = True       ,					):



         '''simple docstring'''

         super().__init__()
         lowerCAmelCase__     :Dict											=				num_attention_heads
         lowerCAmelCase__     :Any											=				attention_head_dim
         lowerCAmelCase__     :Optional[int]											=				num_attention_heads * attention_head_dim

         lowerCAmelCase__     :Any											=				in_channels

         lowerCAmelCase__     :str											=				torch.nn.GroupNorm(num_groups=__UpperCAmelCase       ,					num_channels=__UpperCAmelCase       ,					eps=1E-6       ,					affine=__UpperCAmelCase )
         lowerCAmelCase__     :int											=				nn.Linear(__UpperCAmelCase       ,					__UpperCAmelCase )

         # 3. Define transformers blocks
         lowerCAmelCase__     :List[Any]											=				nn.ModuleList(
             [
                 BasicTransformerBlock(
                     __UpperCAmelCase       ,					__UpperCAmelCase       ,					__UpperCAmelCase       ,					dropout=__UpperCAmelCase       ,					cross_attention_dim=__UpperCAmelCase       ,					activation_fn=__UpperCAmelCase       ,					attention_bias=__UpperCAmelCase       ,					double_self_attention=__UpperCAmelCase       ,					norm_elementwise_affine=__UpperCAmelCase       ,					)
                 for d in range(__UpperCAmelCase )
             ] )

         lowerCAmelCase__     :List[Any]											=				nn.Linear(__UpperCAmelCase       ,					__UpperCAmelCase )






       def 	snake_case		(       self       ,					__UpperCAmelCase       ,					__UpperCAmelCase=None       ,					__UpperCAmelCase=None       ,					__UpperCAmelCase=None       ,					__UpperCAmelCase=1       ,					__UpperCAmelCase=None       ,					__UpperCAmelCase = True       ,					):



         '''simple docstring'''

         lowerCAmelCase__				,							lowerCAmelCase__				,							lowerCAmelCase__				,							lowerCAmelCase__     :int											=				hidden_states.shape
         lowerCAmelCase__     :Tuple											=				batch_frames // num_frames

         lowerCAmelCase__     :str											=				hidden_states

         lowerCAmelCase__     :Union[str, Any]											=				hidden_states[None, :].reshape(__UpperCAmelCase       ,					__UpperCAmelCase       ,					__UpperCAmelCase       ,					__UpperCAmelCase       ,					__UpperCAmelCase )
         lowerCAmelCase__     :str											=				hidden_states.permute(0       ,					2       ,					1       ,					3       ,					4 )

         lowerCAmelCase__     :Optional[int]											=				self.norm(__UpperCAmelCase )
         lowerCAmelCase__     :Optional[Any]											=				hidden_states.permute(0       ,					3       ,					4       ,					2       ,					1 ).reshape(batch_size * height * width       ,					__UpperCAmelCase       ,					__UpperCAmelCase )

         lowerCAmelCase__     :List[Any]											=				self.proj_in(__UpperCAmelCase )

         # 2. Blocks
         for block in self.transformer_blocks:
           lowerCAmelCase__     :Optional[int]											=				block(
               __UpperCAmelCase       ,					encoder_hidden_states=__UpperCAmelCase       ,					timestep=__UpperCAmelCase       ,					cross_attention_kwargs=__UpperCAmelCase       ,					class_labels=__UpperCAmelCase       ,					)

         # 3. Output
         lowerCAmelCase__     :Any											=				self.proj_out(__UpperCAmelCase )
         lowerCAmelCase__     :Dict											=				(
             hidden_states[None, None, :]
             .reshape(__UpperCAmelCase       ,					__UpperCAmelCase       ,					__UpperCAmelCase       ,					__UpperCAmelCase       ,					__UpperCAmelCase )
             .permute(0       ,					3       ,					4       ,					1       ,					2 )
             .contiguous()
         )
         lowerCAmelCase__     :Optional[Any]											=				hidden_states.reshape(__UpperCAmelCase       ,					__UpperCAmelCase       ,					__UpperCAmelCase       ,					__UpperCAmelCase )

         lowerCAmelCase__     :List[Any]											=				hidden_states + residual

         if not return_dict:
           return (output,)

         return TransformerTemporalModelOutput(sample=__UpperCAmelCase )
