


"""simple docstring"""



import json
import os
from typing import Dict, List, Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


__snake_case       =       logging.get_logger(__name__)


__snake_case       =       {
    """vocab_file""": """vocab.json""",
    """tokenizer_config_file""": """tokenizer_config.json""",
    """merges_file""": """merges.txt""",
}

__snake_case       =       {
    """vocab_file""": {
        """facebook/s2t-wav2vec2-large-en-de""": (
            """https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/vocab.json"""
        ),
    },
    """tokenizer_config_file""": {
        """facebook/s2t-wav2vec2-large-en-de""": (
            """https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/tokenizer_config.json"""
        ),
    },
    """merges_file""": {
        """facebook/s2t-wav2vec2-large-en-de""": (
            """https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/merges.txt"""
        ),
    },
}

__snake_case       =       """</w>"""
__snake_case       =       """@@ """







def     __lowerCAmelCase		(		lowercase    :    Tuple					)      ->		List[str]:


    """simple docstring"""





    snake_case       : Union[str, Any]    =       set()
    snake_case       : int    =       word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)					)
        snake_case       : Tuple    =       char
    return pairs


# Speech2Text2 has no max input length
__snake_case       =       {"""facebook/s2t-wav2vec2-large-en-de""": 1024}







class _lowerCAmelCase							( snake_case_      ):
   __UpperCAmelCase					:			Tuple				    =    VOCAB_FILES_NAMES
   __UpperCAmelCase					:			List[str]				    =    PRETRAINED_VOCAB_FILES_MAP
   __UpperCAmelCase					:			Union[str, Any]				    =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   __UpperCAmelCase					:			Tuple				    =    ['''input_ids''', '''attention_mask''']







   def __init__(      self		,      UpperCamelCase__		,      UpperCamelCase__="<s>"		,      UpperCamelCase__="<pad>"		,      UpperCamelCase__="</s>"		,      UpperCamelCase__="<unk>"		,      UpperCamelCase__=False		,      UpperCamelCase__=None		,      **UpperCamelCase__		,      )      ->						List[Any]:

       '''simple docstring'''




       super().__init__(
           unk_token=UpperCamelCase__		,      bos_token=UpperCamelCase__		,      eos_token=UpperCamelCase__		,      pad_token=UpperCamelCase__		,      do_lower_case=UpperCamelCase__		,      **UpperCamelCase__		,      )

       snake_case       : int    =       do_lower_case

       with open(UpperCamelCase__		,      encoding="utf-8"		) as vocab_handle:
           snake_case       : int    =       json.load(UpperCamelCase__		)
       snake_case       : List[str]    =       {v: k for k, v in self.encoder.items()}

       if merges_file is None:
           logger.info(F'No merges files provided. {self.__class__.__name__} can only be used for decoding.'		)

           snake_case       : Any    =       None
           snake_case       : Tuple    =       None
       else:
           with open(UpperCamelCase__		,      encoding="utf-8"		) as merges_handle:
               snake_case       : Union[str, Any]    =       merges_handle.read().split("\n"		)[:-1]

           snake_case       : List[str]    =       [tuple(merge.split()[:2]		) for merge in merges]
           snake_case       : List[str]    =       dict(zip(UpperCamelCase__		,      range(len(UpperCamelCase__		)		)		)		)
           snake_case       : Dict    =       {}







   @property
   def 					lowerCamelCase						(      self		)      ->						int:

       '''simple docstring'''




       return len(self.decoder		)







   def 					lowerCamelCase						(      self		)      ->						Dict:

       '''simple docstring'''




       return dict(self.encoder		,      **self.added_tokens_encoder		)







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

       '''simple docstring'''




       snake_case       : List[Any]    =       tuple(token[:-1]		) + (token[-1] + BPE_TOKEN_MERGES,)
       if token in self.cache:
           return self.cache[token]
       snake_case       : List[str]    =       get_pairs(UpperCamelCase__		)

       if not pairs:
           return token

       while True:
           snake_case       : Dict    =       min(UpperCamelCase__		,      key=lambda UpperCamelCase__		: self.bpe_ranks.get(UpperCamelCase__		,      float("inf"		)		)		)
           if bigram not in self.bpe_ranks:
               break
           snake_case   ,snake_case       : str    =       bigram
           snake_case       : Tuple    =       []
           snake_case       : Optional[int]    =       0
           while i < len(UpperCamelCase__		):
               try:
                   snake_case       : Tuple    =       word.index(UpperCamelCase__		,      UpperCamelCase__		)
               except ValueError:
                   new_word.extend(word[i:]		)
                   break
               else:
                   new_word.extend(word[i:j]		)
                   snake_case       : str    =       j

               if word[i] == first and i < len(UpperCamelCase__		) - 1 and word[i + 1] == second:
                   new_word.append(first + second		)
                   i += 2
               else:
                   new_word.append(word[i]		)
                   i += 1
           snake_case       : Any    =       tuple(UpperCamelCase__		)
           snake_case       : List[Any]    =       new_word
           if len(UpperCamelCase__		) == 1:
               break
           else:
               snake_case       : Optional[int]    =       get_pairs(UpperCamelCase__		)
       snake_case       : Optional[int]    =       " ".join(UpperCamelCase__		)
       if word == "\n  " + BPE_TOKEN_MERGES:
           snake_case       : List[Any]    =       "\n" + BPE_TOKEN_MERGES

       if word.endswith(UpperCamelCase__		):
           snake_case       : str    =       word.replace(UpperCamelCase__		,      ""		)

       snake_case       : Dict    =       word.replace(" "		,      UpperCamelCase__		)
       snake_case       : List[str]    =       word
       return word







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		)      ->						int:

       '''simple docstring'''




       if self.bpe_ranks is None:
           raise ValueError(
               "This tokenizer was instantiated without a `merges.txt` file, so"
               " that it can only be used for decoding, not for encoding."
               "Make sure to provide `merges.txt` file at instantiation to enable "
               "encoding."		)

       if self.do_lower_case:
           snake_case       : int    =       text.lower()

       snake_case       : Union[str, Any]    =       text.split()

       snake_case       : List[Any]    =       []
       for token in text:
           if token:
               split_tokens.extend(list(self.bpe(UpperCamelCase__		).split(" "		)		)		)

       return split_tokens







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		)      ->						int:

       '''simple docstring'''




       return self.encoder.get(UpperCamelCase__		,      self.encoder.get(self.unk_token		)		)







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		)      ->						str:

       '''simple docstring'''




       snake_case       : Union[str, Any]    =       self.decoder.get(UpperCamelCase__		,      self.unk_token		)
       return result







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		)      ->						str:

       '''simple docstring'''




       snake_case       : Tuple    =       " ".join(UpperCamelCase__		)

       # make sure @@ tokens are concatenated
       snake_case       : Optional[Any]    =       "".join(string.split(UpperCamelCase__		)		)

       return string







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		,      UpperCamelCase__ = None		)      ->						Tuple[str]:

       '''simple docstring'''




       if not os.path.isdir(UpperCamelCase__		):
           logger.error(F'Vocabulary path ({save_directory}) should be a directory'		)
           return
       snake_case       : Optional[Any]    =       os.path.join(
           UpperCamelCase__		,      (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]		)
       snake_case       : Optional[Any]    =       os.path.join(
           UpperCamelCase__		,      (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]		)

       with open(UpperCamelCase__		,      "w"		,      encoding="utf-8"		) as f:
           f.write(json.dumps(self.encoder		,      indent=2		,      sort_keys=UpperCamelCase__		,      ensure_ascii=UpperCamelCase__		) + "\n"		)

       snake_case       : int    =       0
       if self.bpe_ranks is None:
           return (vocab_file,)

       with open(UpperCamelCase__		,      "w"		,      encoding="utf-8"		) as writer:
           for bpe_tokens, token_index in sorted(self.bpe_ranks.items()		,      key=lambda UpperCamelCase__		: kv[1]		):
               if index != token_index:
                   logger.warning(
                       F'Saving vocabulary to {merges_file}: BPE merge indices are not consecutive.'
                       " Please check that the tokenizer is not corrupted!"		)
                   snake_case       : Optional[int]    =       token_index
               writer.write(" ".join(UpperCamelCase__		) + "\n"		)
               index += 1

       return (vocab_file, merges_file)









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





