'''simple docstring'''

from .imports import is_tqdm_available


if is_tqdm_available():
  from tqdm.auto import tqdm as _tqdm

from ..state import PartialState







def __UpperCamelCase( _A       :					bool = True	,      *_A       :					str	,      **_A       :					List[Any]     ):
   '''simple docstring'''





   if not is_tqdm_available():
      raise ImportError('''Accelerate\'s `tqdm` module requires `tqdm` to be installed. Please run `pip install tqdm`.'''     )
   UpperCAmelCase__				:					Dict 					=	False
   if main_process_only:
      UpperCAmelCase__				:					int 					=	PartialState().local_process_index == 0
   return _tqdm(*_A	,      **_A	,      disable=_A     )








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


__snake_case       =       logging.get_logger(__name__)







class _lowerCAmelCase							:







   def __init__(      self		,      UpperCamelCase__ = None		,      UpperCamelCase__ = None		,      UpperCamelCase__=None		,      UpperCamelCase__=None		)      ->						Optional[Any]:

       '''simple docstring'''




       if not conversation_id:
           snake_case       : Union[str, Any]    =       uuid.uuida()
       if past_user_inputs is None:
           snake_case       : Optional[Any]    =       []
       if generated_responses is None:
           snake_case       : Optional[Any]    =       []

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







   def __eq__(      self		,      UpperCamelCase__		)      ->						Any:

       '''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 					lowerCamelCase						(      self		,      UpperCamelCase__		,      UpperCamelCase__ = False		)      ->						Dict:

       '''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}".'		)
               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:
           snake_case       : Any    =       text







   def 					lowerCamelCase						(      self		)      ->						List[str]:

       '''simple docstring'''




       if self.new_user_input:
           self.past_user_inputs.append(self.new_user_input		)
       snake_case       : Dict    =       None







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		)      ->						List[str]:

       '''simple docstring'''




       self.generated_responses.append(UpperCamelCase__		)







   def 					lowerCamelCase						(      self		)      ->						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]:

       '''simple docstring'''




       snake_case       : Optional[Any]    =       F'Conversation id: {self.uuid} \n'
       for is_user, text in self.iter_texts():
           snake_case       : List[str]    =       "user" if is_user else "bot"
           output += F'{name} >> {text} \n'
       return output







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







   def __init__(      self		,      *UpperCamelCase__		,      **UpperCamelCase__		)      ->						str:

       '''simple docstring'''




       super().__init__(*UpperCamelCase__		,      **UpperCamelCase__		)
       if self.tokenizer.pad_token_id is None:
           snake_case       : str    =       self.tokenizer.eos_token







   def 					lowerCamelCase						(      self		,      UpperCamelCase__=None		,      UpperCamelCase__=None		,      UpperCamelCase__=None		,      **UpperCamelCase__		)      ->						Union[str, Any]:

       '''simple docstring'''




       snake_case       : Tuple    =       {}
       snake_case       : Optional[int]    =       {}
       snake_case       : Optional[Any]    =       {}

       if min_length_for_response is not None:
           snake_case       : int    =       min_length_for_response
       if minimum_tokens is not None:
           snake_case       : Dict    =       minimum_tokens

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

       '''simple docstring'''




       snake_case       : int    =       super().__call__(UpperCamelCase__		,      num_workers=UpperCamelCase__		,      **UpperCamelCase__		)
       if isinstance(UpperCamelCase__		,      UpperCamelCase__		) and len(UpperCamelCase__		) == 1:
           return outputs[0]
       return outputs







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

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







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		,      UpperCamelCase__=10		,      **UpperCamelCase__		)      ->						Union[str, Any]:

       '''simple docstring'''




       snake_case       : Dict    =       generate_kwargs.get("max_length"		,      self.model.config.max_length		)

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







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		,      UpperCamelCase__=True		)      ->						Tuple:

       '''simple docstring'''




       snake_case       : Union[str, Any]    =       model_outputs["output_ids"]
       snake_case       : str    =       self.tokenizer.decode(
           output_ids[0]		,      skip_special_tokens=UpperCamelCase__		,      clean_up_tokenization_spaces=UpperCamelCase__		,      )
       snake_case       : List[Any]    =       model_outputs["conversation"]
       conversation.mark_processed()
       conversation.append_response(UpperCamelCase__		)
       return conversation







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		)      ->						Dict:

       '''simple docstring'''




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





