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






class    UpperCAmelCase_				:





 def __init__(    self ,       lowercase_ = None ,       lowercase_ = None ,       lowercase_=None ,       lowercase_=None):
     if not conversation_id:
         snake_case_ :    Union[str, Any]     			=						uuid.uuida()
     if past_user_inputs is None:
         snake_case_ :    Tuple     			=						[]
     if generated_responses is None:
         snake_case_ :    str     			=						[]

     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 ,       lowercase_):
     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        snake_case__		(    self ,       lowercase_ ,       lowercase_ = 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}".')
             snake_case_ :    Union[str, 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:
         snake_case_ :    Dict     			=						text





 def        snake_case__		(    self):
     if self.new_user_input:
         self.past_user_inputs.append(self.new_user_input)
     snake_case_ :    Tuple     			=						None





 def        snake_case__		(    self ,       lowercase_):
     self.generated_responses.append(lowercase_)





 def        snake_case__		(    self):
     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):
     snake_case_ :    int     			=						F'Conversation id: {self.uuid} \n'
     for is_user, text in self.iter_texts():
         snake_case_ :    Optional[int]     			=						"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    UpperCAmelCase_				(	snake_case__					):





 def __init__(    self ,       *lowercase_ ,       **lowercase_):
     super().__init__(*lowercase_ ,       **lowercase_)
     if self.tokenizer.pad_token_id is None:
         snake_case_ :    Any     			=						self.tokenizer.eos_token





 def        snake_case__		(    self ,       lowercase_=None ,       lowercase_=None ,       lowercase_=None ,       **lowercase_):
     snake_case_ :    int     			=						{}
     snake_case_ :    Optional[Any]     			=						{}
     snake_case_ :    List[str]     			=						{}

     if min_length_for_response is not None:
         snake_case_ :    List[Any]     			=						min_length_for_response
     if minimum_tokens is not None:
         snake_case_ :    Optional[int]     			=						minimum_tokens

     if "max_length" in generate_kwargs:
         snake_case_ :    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:
         snake_case_ :    Union[str, Any]     			=						clean_up_tokenization_spaces

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





 def __call__(    self ,       lowercase_ ,       lowercase_=0 ,       **lowercase_):
     snake_case_ :    Optional[Any]     			=						super().__call__(lowercase_ ,       num_workers=lowercase_ ,       **lowercase_)
     if isinstance(lowercase_ ,       lowercase_) and len(lowercase_) == 1:
         return outputs[0]
     return outputs





 def        snake_case__		(    self ,       lowercase_ ,       lowercase_=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"):
         snake_case_ :    List[Any]     			=						self.tokenizer._build_conversation_input_ids(lowercase_)
     else:
         # If the tokenizer cannot handle conversations, we default to only the old version
         snake_case_ :    Union[str, Any]     			=						self._legacy_parse_and_tokenize(lowercase_)

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





 def        snake_case__		(    self ,       lowercase_ ,       lowercase_=10 ,       **lowercase_):
     snake_case_ :    Tuple     			=						generate_kwargs.get("max_length" ,       self.model.config.max_length)

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





 def        snake_case__		(    self ,       lowercase_ ,       lowercase_=True):
     snake_case_ :    List[Any]     			=						model_outputs["output_ids"]
     snake_case_ :    str     			=						self.tokenizer.decode(
         output_ids[0] ,       skip_special_tokens=lowercase_ ,       clean_up_tokenization_spaces=lowercase_ ,       )
     snake_case_ :    Tuple     			=						model_outputs["conversation"]
     conversation.mark_processed()
     conversation.append_response(lowercase_)
     return conversation





 def        snake_case__		(    self ,       lowercase_):
     snake_case_ :    str     			=						self.tokenizer.eos_token_id
     snake_case_ :    Optional[int]     			=						[]
     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:
         snake_case_ :    Optional[int]     			=						input_ids[-self.tokenizer.model_max_length :]
     return input_ids






'''simple docstring'''
import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


a_											=						logging.get_logger(__name__)






class    UpperCAmelCase_				:
 UpperCAmelCase_		    =						None


@experimental
def 						UpperCamelCase_     (	__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE       ):
    """simple docstring"""






    if ParallelBackendConfig.backend_name is None:
        return _map_with_multiprocessing_pool(
            __SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE       )

    return _map_with_joblib(__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE       )

def 						UpperCamelCase_     (	__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE       ):
    """simple docstring"""






    snake_case_ :    str     			=						num_proc if num_proc <= len(__SCREAMING_SNAKE_CASE       ) else len(__SCREAMING_SNAKE_CASE       )
    snake_case_ :    Optional[int]     			=						[]  # We organize the splits ourselve (contiguous splits)
    for index in range(__SCREAMING_SNAKE_CASE       ):
        snake_case_ :    List[str]     			=						len(__SCREAMING_SNAKE_CASE       ) // num_proc
        snake_case_ :    Optional[Any]     			=						len(__SCREAMING_SNAKE_CASE       ) % num_proc
        snake_case_ :    Optional[Any]     			=						div * index + min(__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE       )
        snake_case_ :    Union[str, Any]     			=						start + div + (1 if index < mod else 0)
        split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)       )

    if len(__SCREAMING_SNAKE_CASE       ) != sum(len(i[1]       ) for i in split_kwds       ):
        raise ValueError(
            f'Error dividing inputs iterable among processes. '
            f'Total number of objects {len(__SCREAMING_SNAKE_CASE       )}, '
            f'length: {sum(len(i[1]       ) for i in split_kwds       )}'       )

    logger.info(
        f'Spawning {num_proc} processes for {len(__SCREAMING_SNAKE_CASE       )} objects in slices of {[len(i[1]       ) for i in split_kwds]}'       )
    snake_case_			, snake_case_ :    Tuple     			=						None, None
    if not disable_tqdm:
        snake_case_			, snake_case_ :    int     			=						(RLock(),), tqdm.set_lock
    with Pool(__SCREAMING_SNAKE_CASE,		initargs=__SCREAMING_SNAKE_CASE,		initializer=__SCREAMING_SNAKE_CASE       ) as pool:
        snake_case_ :    Union[str, Any]     			=						pool.map(__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE       )
    logger.info(f'Finished {num_proc} processes'       )
    snake_case_ :    str     			=						[obj for proc_res in mapped for obj in proc_res]
    logger.info(f'Unpacked {len(__SCREAMING_SNAKE_CASE       )} objects'       )

    return mapped

def 						UpperCamelCase_     (	__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE       ):
    """simple docstring"""






    import joblib

    with joblib.parallel_backend(ParallelBackendConfig.backend_name,		n_jobs=__SCREAMING_SNAKE_CASE       ):
        return joblib.Parallel()(
            joblib.delayed(__SCREAMING_SNAKE_CASE       )((function, obj, types, None, True, None)       ) for obj in iterable       )


@experimental
@contextlib.contextmanager
def 						UpperCamelCase_     (	__SCREAMING_SNAKE_CASE       ):
    """simple docstring"""






    snake_case_ :    Union[str, Any]     			=						backend_name

    if backend_name == "spark":
        from joblibspark import register_spark

        register_spark()

        # TODO: call create_cache_and_write_probe if "download" in steps
        # TODO: raise NotImplementedError when Dataset.map etc is called

    try:
        yield
    finally:
        snake_case_ :    Any     			=						None




