
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			:     str           =   logging.get_logger(__name__)




class        UpperCAmelCase_			:


  '''simple docstring'''







  def __init__( self			: Optional[int]			,							UpperCamelCase__			: str = None			,							UpperCamelCase__			: uuid.UUID = None			,							UpperCamelCase__			: Any=None			,							UpperCamelCase__			: List[str]=None    )			->	List[str]:







    """simple docstring"""






    if not conversation_id:
      __magic_name__				     =       uuid.uuida()
    if past_user_inputs is None:
      __magic_name__				     =       []
    if generated_responses is None:
      __magic_name__				     =       []

    __magic_name__				     =       conversation_id
    __magic_name__				     =       past_user_inputs
    __magic_name__				     =       generated_responses
    __magic_name__				     =       text
  def __eq__( self			: Union[str, Any]			,							UpperCamelCase__			: List[Any]    )			->	List[str]:







    """simple docstring"""






    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 							_lowercase							( self			: Optional[int]			,							UpperCamelCase__			: str			,							UpperCamelCase__			: bool = False    )			->	Tuple:







    """simple docstring"""






    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}".'''    )
        __magic_name__				     =       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:
      __magic_name__				     =       text
  def 							_lowercase							( self			: Union[str, Any]    )			->	str:







    """simple docstring"""






    if self.new_user_input:
      self.past_user_inputs.append(self.new_user_input    )
    __magic_name__				     =       None
  def 							_lowercase							( self			: Optional[Any]			,							UpperCamelCase__			: str    )			->	Optional[int]:







    """simple docstring"""






    self.generated_responses.append(UpperCamelCase__    )
  def 							_lowercase							( self			: Tuple    )			->	List[Any]:







    """simple docstring"""






    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			: Union[str, Any]    )			->	Tuple:







    """simple docstring"""






    __magic_name__				     =       F'''Conversation id: {self.uuid} \n'''
    for is_user, text in self.iter_texts():
      __magic_name__				     =       """user""" if is_user else """bot"""
      output += F'''{name} >> {text} \n'''
    return output







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


  '''simple docstring'''







  def __init__( self			: List[str]			,							*UpperCamelCase__			: List[Any]			,							**UpperCamelCase__			: List[Any]    )			->	str:







    """simple docstring"""






    super().__init__(*UpperCamelCase__			,							**UpperCamelCase__    )
    if self.tokenizer.pad_token_id is None:
      __magic_name__				     =       self.tokenizer.eos_token
  def 							_lowercase							( self			: str			,							UpperCamelCase__			: Optional[Any]=None			,							UpperCamelCase__			: Tuple=None			,							UpperCamelCase__			: Tuple=None			,							**UpperCamelCase__			: Tuple    )			->	Any:







    """simple docstring"""






    __magic_name__				     =       {}
    __magic_name__				     =       {}
    __magic_name__				     =       {}

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

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

    if generate_kwargs:
      forward_params.update(UpperCamelCase__    )
    return preprocess_params, forward_params, postprocess_params
  def __call__( self			: int			,							UpperCamelCase__			: Union[Conversation, List[Conversation]]			,							UpperCamelCase__			: Any=0			,							**UpperCamelCase__			: List[Any]    )			->	str:







    """simple docstring"""






    __magic_name__				     =       super().__call__(UpperCamelCase__			,							num_workers=UpperCamelCase__			,							**UpperCamelCase__    )
    if isinstance(UpperCamelCase__			,							UpperCamelCase__    ) and len(UpperCamelCase__    ) == 1:
      return outputs[0]
    return outputs
  def 							_lowercase							( self			: List[Any]			,							UpperCamelCase__			: Conversation			,							UpperCamelCase__			: Union[str, Any]=32    )			->	Dict[str, Any]:







    """simple docstring"""






    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"""    ):
      __magic_name__				     =       self.tokenizer._build_conversation_input_ids(UpperCamelCase__    )
    else:
      # If the tokenizer cannot handle conversations, we default to only the old version
      __magic_name__				     =       self._legacy_parse_and_tokenize(UpperCamelCase__    )

    if self.framework == "pt":
      __magic_name__				     =       torch.LongTensor([input_ids]    )
    elif self.framework == "tf":
      __magic_name__				     =       tf.constant([input_ids]    )
    return {"input_ids": input_ids, "conversation": conversation}
  def 							_lowercase							( self			: Dict			,							UpperCamelCase__			: Dict			,							UpperCamelCase__			: List[str]=10			,							**UpperCamelCase__			: int    )			->	Dict:







    """simple docstring"""






    __magic_name__				     =       generate_kwargs.get("""max_length"""			,							self.model.config.max_length    )

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







    """simple docstring"""






    __magic_name__				     =       model_outputs["""output_ids"""]
    __magic_name__				     =       self.tokenizer.decode(
        output_ids[0]			,							skip_special_tokens=UpperCamelCase__			,							clean_up_tokenization_spaces=UpperCamelCase__			,							)
    __magic_name__				     =       model_outputs["""conversation"""]
    conversation.mark_processed()
    conversation.append_response(UpperCamelCase__    )
    return conversation





  def 							_lowercase							( self			: List[Any]			,							UpperCamelCase__			: Conversation    )			->	Dict:







    """simple docstring"""






    __magic_name__				     =       self.tokenizer.eos_token_id
    __magic_name__				     =       []
    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:
      __magic_name__				     =       input_ids[-self.tokenizer.model_max_length :]
    return input_ids



import unittest

from transformers import (
    MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TextaTextGenerationPipeline,
    pipeline,
)
from transformers.testing_utils import is_pipeline_test, require_tf, require_torch
from transformers.utils import is_torch_available

from .test_pipelines_common import ANY


if is_torch_available():
    import torch




@is_pipeline_test
class        UpperCAmelCase_			( unittest.TestCase     ):


  '''simple docstring'''







  a__           =		MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
  a__           =		TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
  def 							_lowercase							( self			: str			,							UpperCamelCase__			: Optional[int]			,							UpperCamelCase__			: int			,							UpperCamelCase__			: int    )			->	Tuple:







    """simple docstring"""






    __magic_name__				     =       TextaTextGenerationPipeline(model=UpperCamelCase__			,							tokenizer=UpperCamelCase__    )
    return generator, ["Something to write", "Something else"]
  def 							_lowercase							( self			: Any			,							UpperCamelCase__			: Tuple			,							UpperCamelCase__			: List[str]    )			->	Optional[Any]:







    """simple docstring"""






    __magic_name__				     =       generator("""Something there"""    )
    self.assertEqual(UpperCamelCase__			,							[{"""generated_text""": ANY(UpperCamelCase__    )}]    )
    # These are encoder decoder, they don't just append to incoming string
    self.assertFalse(outputs[0]["""generated_text"""].startswith("""Something there"""    )    )

    __magic_name__				     =       generator(["""This is great !""", """Something else"""]			,							num_return_sequences=2			,							do_sample=UpperCamelCase__    )
    self.assertEqual(
        UpperCamelCase__			,							[
            [{"""generated_text""": ANY(UpperCamelCase__    )}, {"""generated_text""": ANY(UpperCamelCase__    )}],
            [{"""generated_text""": ANY(UpperCamelCase__    )}, {"""generated_text""": ANY(UpperCamelCase__    )}],
        ]			,							)

    __magic_name__				     =       generator(
        ["""This is great !""", """Something else"""]			,							num_return_sequences=2			,							batch_size=2			,							do_sample=UpperCamelCase__    )
    self.assertEqual(
        UpperCamelCase__			,							[
            [{"""generated_text""": ANY(UpperCamelCase__    )}, {"""generated_text""": ANY(UpperCamelCase__    )}],
            [{"""generated_text""": ANY(UpperCamelCase__    )}, {"""generated_text""": ANY(UpperCamelCase__    )}],
        ]			,							)

    with self.assertRaises(UpperCamelCase__    ):
      generator(4    )
  @require_torch
  def 							_lowercase							( self			: Optional[int]    )			->	Union[str, Any]:







    """simple docstring"""






    __magic_name__				     =       pipeline("""text2text-generation"""			,							model="""patrickvonplaten/t5-tiny-random"""			,							framework="""pt"""    )
    # do_sample=False necessary for reproducibility
    __magic_name__				     =       generator("""Something there"""			,							do_sample=UpperCamelCase__    )
    self.assertEqual(UpperCamelCase__			,							[{"""generated_text""": """"""}]    )

    __magic_name__				     =       3
    __magic_name__				     =       generator(
        """Something there"""			,							num_return_sequences=UpperCamelCase__			,							num_beams=UpperCamelCase__			,							)
    __magic_name__				     =       [
        {"""generated_text""": """Beide Beide Beide Beide Beide Beide Beide Beide Beide"""},
        {"""generated_text""": """Beide Beide Beide Beide Beide Beide Beide Beide"""},
        {"""generated_text""": """"""},
    ]
    self.assertEqual(UpperCamelCase__			,							UpperCamelCase__    )

    __magic_name__				     =       generator("""This is a test"""			,							do_sample=UpperCamelCase__			,							num_return_sequences=2			,							return_tensors=UpperCamelCase__    )
    self.assertEqual(
        UpperCamelCase__			,							[
            {"""generated_token_ids""": ANY(torch.Tensor    )},
            {"""generated_token_ids""": ANY(torch.Tensor    )},
        ]			,							)
    __magic_name__				     =       generator.model.config.eos_token_id
    __magic_name__				     =       """<pad>"""
    __magic_name__				     =       generator(
        ["""This is a test""", """This is a second test"""]			,							do_sample=UpperCamelCase__			,							num_return_sequences=2			,							batch_size=2			,							return_tensors=UpperCamelCase__			,							)
    self.assertEqual(
        UpperCamelCase__			,							[
            [
                {"""generated_token_ids""": ANY(torch.Tensor    )},
                {"""generated_token_ids""": ANY(torch.Tensor    )},
            ],
            [
                {"""generated_token_ids""": ANY(torch.Tensor    )},
                {"""generated_token_ids""": ANY(torch.Tensor    )},
            ],
        ]			,							)





  @require_tf
  def 							_lowercase							( self			: int    )			->	str:







    """simple docstring"""






    __magic_name__				     =       pipeline("""text2text-generation"""			,							model="""patrickvonplaten/t5-tiny-random"""			,							framework="""tf"""    )
    # do_sample=False necessary for reproducibility
    __magic_name__				     =       generator("""Something there"""			,							do_sample=UpperCamelCase__    )
    self.assertEqual(UpperCamelCase__			,							[{"""generated_text""": """"""}]    )

