import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


lowerCamelCase__   = logging.get_logger(__name__)

lowerCamelCase__   = {
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

lowerCamelCase__   = {
    '''vocab_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/vocab.json'''},
    '''merges_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/merges.txt'''},
}

lowerCamelCase__   = {
    '''openai-gpt''': 512,
}


def 							lowerCAmelCase__    (       a__      )	->Optional[int]:




  '''simple docstring'''

  _UpperCamelCase				=  text.strip()
  if not text:
    return []
  _UpperCamelCase				=  text.split()
  return tokens




class  _UpperCAmelCase       (			A_   ):



 '''simple docstring'''




 def __init__(						self							:					int				,      lowercase_							:					List[str]=True				,      lowercase_							:					str=None				,      lowercase_							:					int=True				,      lowercase_							:					List[Any]=None				,      lowercase_							:					Any=True				,      )		->     Any:


   """simple docstring"""







   if never_split is None:
     _UpperCamelCase				=  []
   _UpperCamelCase				=  do_lower_case
   _UpperCamelCase				=  set(_A)
   _UpperCamelCase				=  tokenize_chinese_chars
   _UpperCamelCase				=  strip_accents
   _UpperCamelCase				=  do_split_on_punc







 def 						__UpperCAmelCase							(						self							:					Tuple				,      lowercase_							:					List[Any]				,      lowercase_							:					List[Any]=None)		->     Union[str, Any]:


   """simple docstring"""







   _UpperCamelCase				=  self.never_split.union(set(_A)) if never_split else self.never_split
   _UpperCamelCase				=  self._clean_text(_A)

   # This was added on November 1st, 2018 for the multilingual and Chinese
   # models. This is also applied to the English models now, but it doesn't
   # matter since the English models were not trained on any Chinese data
   # and generally don't have any Chinese data in them (there are Chinese
   # characters in the vocabulary because Wikipedia does have some Chinese
   # words in the English Wikipedia.).
   if self.tokenize_chinese_chars:
     _UpperCamelCase				=  self._tokenize_chinese_chars(_A)
   # prevents treating the same character with different unicode codepoints as different characters
   _UpperCamelCase				=  unicodedata.normalize("NFC"				,      _A)
   _UpperCamelCase				=  whitespace_tokenize(_A)
   _UpperCamelCase				=  []
   for token in orig_tokens:
     if token not in never_split:
       if self.do_lower_case:
         _UpperCamelCase				=  token.lower()
         if self.strip_accents is not False:
           _UpperCamelCase				=  self._run_strip_accents(_A)
       elif self.strip_accents:
         _UpperCamelCase				=  self._run_strip_accents(_A)
     split_tokens.extend(self._run_split_on_punc(_A				,      _A))

   _UpperCamelCase				=  whitespace_tokenize(" ".join(_A))
   return output_tokens







 def 						__UpperCAmelCase							(						self							:					Any				,      lowercase_							:					str)		->     str:


   """simple docstring"""







   _UpperCamelCase				=  unicodedata.normalize("NFD"				,      _A)
   _UpperCamelCase				=  []
   for char in text:
     _UpperCamelCase				=  unicodedata.category(_A)
     if cat == "Mn":
       continue
     output.append(_A)
   return "".join(_A)







 def 						__UpperCAmelCase							(						self							:					Any				,      lowercase_							:					str				,      lowercase_							:					str=None)		->     List[str]:


   """simple docstring"""







   if not self.do_split_on_punc or (never_split is not None and text in never_split):
     return [text]
   _UpperCamelCase				=  list(_A)
   _UpperCamelCase				=  0
   _UpperCamelCase				=  True
   _UpperCamelCase				=  []
   while i < len(_A):
     _UpperCamelCase				=  chars[i]
     if _is_punctuation(_A):
       output.append([char])
       _UpperCamelCase				=  True
     else:
       if start_new_word:
         output.append([])
       _UpperCamelCase				=  False
       output[-1].append(_A)
     i += 1

   return ["".join(_A) for x in output]







 def 						__UpperCAmelCase							(						self							:					int				,      lowercase_							:					Optional[int])		->     Tuple:


   """simple docstring"""







   _UpperCamelCase				=  []
   for char in text:
     _UpperCamelCase				=  ord(_A)
     if self._is_chinese_char(_A):
       output.append(" ")
       output.append(_A)
       output.append(" ")
     else:
       output.append(_A)
   return "".join(_A)







 def 						__UpperCAmelCase							(						self							:					int				,      lowercase_							:					Union[str, Any])		->     Union[str, Any]:


   """simple docstring"""







   if (
       (cp >= 0x4e00 and cp <= 0x9fff)
       or (cp >= 0x3400 and cp <= 0x4dbf)  #
       or (cp >= 0x2_0000 and cp <= 0x2_a6df)  #
       or (cp >= 0x2_a700 and cp <= 0x2_b73f)  #
       or (cp >= 0x2_b740 and cp <= 0x2_b81f)  #
       or (cp >= 0x2_b820 and cp <= 0x2_ceaf)  #
       or (cp >= 0xf900 and cp <= 0xfaff)
       or (cp >= 0x2_f800 and cp <= 0x2_fa1f)  #
   ):  #
     return True

   return False







 def 						__UpperCAmelCase							(						self							:					Optional[Any]				,      lowercase_							:					Optional[int])		->     List[Any]:


   """simple docstring"""







   _UpperCamelCase				=  []
   for char in text:
     _UpperCamelCase				=  ord(_A)
     if cp == 0 or cp == 0xfffd or _is_control(_A):
       continue
     if _is_whitespace(_A):
       output.append(" ")
     else:
       output.append(_A)
   return "".join(_A)


def 							lowerCAmelCase__    (       a__      )	->Optional[Any]:




  '''simple docstring'''

  _UpperCamelCase				=  set()
  _UpperCamelCase				=  word[0]
  for char in word[1:]:
    pairs.add((prev_char, char)      )
    _UpperCamelCase				=  char
  return pairs


def 							lowerCAmelCase__    (       a__      )	->int:




  '''simple docstring'''

  _UpperCamelCase				=  text.replace("—"  ,			"-"      )
  _UpperCamelCase				=  text.replace("–"  ,			"-"      )
  _UpperCamelCase				=  text.replace("―"  ,			"-"      )
  _UpperCamelCase				=  text.replace("…"  ,			"..."      )
  _UpperCamelCase				=  text.replace("´"  ,			"'"      )
  _UpperCamelCase				=  re.sub(r"(-+|~+|!+|\"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)"  ,			r" \1 "  ,			A__      )
  _UpperCamelCase				=  re.sub(r"\s*\n\s*"  ,			" \n "  ,			A__      )
  _UpperCamelCase				=  re.sub(r"[^\S\n]+"  ,			" "  ,			A__      )
  return text.strip()







class  _UpperCAmelCase       (			A_   ):



 '''simple docstring'''




 __A           = VOCAB_FILES_NAMES
 __A           = PRETRAINED_VOCAB_FILES_MAP
 __A           = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 __A           = ["input_ids", "attention_mask"]







 def __init__(						self							:					Union[str, Any]				,      lowercase_							:					Union[str, Any]				,      lowercase_							:					str				,      lowercase_							:					Tuple="<unk>"				,      **lowercase_							:					List[Any])		->     Any:


   """simple docstring"""







   super().__init__(unk_token=_A				,      **_A)

   try:
     import ftfy
     from spacy.lang.en import English

     _UpperCamelCase				=  English()
     _UpperCamelCase				=  _nlp.tokenizer
     _UpperCamelCase				=  ftfy.fix_text
   except ImportError:
     logger.warning("ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.")
     _UpperCamelCase				=  BasicTokenizer(do_lower_case=_A)
     _UpperCamelCase				=  None

   with open(_A				,      encoding="utf-8") as vocab_handle:
     _UpperCamelCase				=  json.load(_A)
   _UpperCamelCase				=  {v: k for k, v in self.encoder.items()}
   with open(_A				,      encoding="utf-8") as merges_handle:
     _UpperCamelCase				=  merges_handle.read().split("\n")[1:-1]
   _UpperCamelCase				=  [tuple(merge.split()) for merge in merges]
   _UpperCamelCase				=  dict(zip(_A				,      range(len(_A))))
   _UpperCamelCase				=  {}







 @property
 def 						__UpperCAmelCase							(						self							:					int)		->     Tuple:


   """simple docstring"""







   return True







 @property
 def 						__UpperCAmelCase							(						self							:					Optional[Any])		->     str:


   """simple docstring"""







   return len(self.encoder)







 def 						__UpperCAmelCase							(						self							:					List[Any])		->     List[Any]:


   """simple docstring"""







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







 def 						__UpperCAmelCase							(						self							:					str				,      lowercase_							:					Union[str, Any])		->     List[str]:


   """simple docstring"""







   _UpperCamelCase				=  tuple(token[:-1]) + (token[-1] + "</w>",)
   if token in self.cache:
     return self.cache[token]
   _UpperCamelCase				=  get_pairs(_A)

   if not pairs:
     return token + "</w>"

   while True:
     _UpperCamelCase				=  min(_A				,      key=lambda lowercase_: self.bpe_ranks.get(_A				,      float("inf")))
     if bigram not in self.bpe_ranks:
       break
     _UpperCamelCase	,		_UpperCamelCase				=  bigram
     _UpperCamelCase				=  []
     _UpperCamelCase				=  0
     while i < len(_A):
       try:
         _UpperCamelCase				=  word.index(_A				,      _A)
       except ValueError:
         new_word.extend(word[i:])
         break
       else:
         new_word.extend(word[i:j])
         _UpperCamelCase				=  j

       if word[i] == first and i < len(_A) - 1 and word[i + 1] == second:
         new_word.append(first + second)
         i += 2
       else:
         new_word.append(word[i])
         i += 1
     _UpperCamelCase				=  tuple(_A)
     _UpperCamelCase				=  new_word
     if len(_A) == 1:
       break
     else:
       _UpperCamelCase				=  get_pairs(_A)
   _UpperCamelCase				=  " ".join(_A)
   if word == "\n  </w>":
     _UpperCamelCase				=  "\n</w>"
   _UpperCamelCase				=  word
   return word







 def 						__UpperCAmelCase							(						self							:					List[Any]				,      lowercase_							:					Any)		->     Any:


   """simple docstring"""







   _UpperCamelCase				=  []
   if self.fix_text is None:
     # Using BERT's BasicTokenizer
     _UpperCamelCase				=  self.nlp.tokenize(_A)
     for token in text:
       split_tokens.extend(list(self.bpe(_A).split(" ")))
   else:
     # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
     _UpperCamelCase				=  self.nlp(text_standardize(self.fix_text(_A)))
     for token in text:
       split_tokens.extend(list(self.bpe(token.text.lower()).split(" ")))
   return split_tokens







 def 						__UpperCAmelCase							(						self							:					Dict				,      lowercase_							:					Any)		->     List[Any]:


   """simple docstring"""







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







 def 						__UpperCAmelCase							(						self							:					List[str]				,      lowercase_							:					Any)		->     Optional[Any]:


   """simple docstring"""







   return self.decoder.get(_A				,      self.unk_token)







 def 						__UpperCAmelCase							(						self							:					Tuple				,      lowercase_							:					Tuple)		->     Tuple:


   """simple docstring"""







   _UpperCamelCase				=  "".join(_A).replace("</w>"				,      " ").strip()
   return out_string







 def 						__UpperCAmelCase							(						self							:					List[Any]				,      lowercase_							:					str				,      lowercase_							:					Optional[str] = None)		->     Tuple[str]:


   """simple docstring"""







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

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

   _UpperCamelCase				=  0
   with open(_A				,      "w"				,      encoding="utf-8") as writer:
     writer.write("#version: 0.2\n")
     for bpe_tokens, token_index in sorted(self.bpe_ranks.items()				,      key=lambda lowercase_: kv[1]):
       if index != token_index:
         logger.warning(
             f'Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'
             " Please check that the tokenizer is not corrupted!")
         _UpperCamelCase				=  token_index
       writer.write(" ".join(_A) + "\n")
       index += 1

   return vocab_file, merge_file






from typing import List, Optional, Union

import numpy as np

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


_A  					= logging.get_logger(__name__)







class      lowerCamelCase		(  A_  ):
     UpperCAmelCase__				:							int           =		["input_values", "attention_mask"]





     def __init__(self      :    Tuple				,    _A      :    Optional[Any]=1				,    _A      :    Dict=1_6_0_0_0				,    _A      :    int=0.0				,    _A      :    List[str]=False				,    _A      :    Optional[int]=True				,    **_A      :    List[Any]				,    )						->						str:
         super().__init__(feature_size=_A				,    sampling_rate=_A				,    padding_value=_A				,    **_A   )
         snake_case      =					return_attention_mask
         snake_case      =					do_normalize





     @staticmethod
     def 		UpperCAmelCase(_A      :    List[np.ndarray]				,    _A      :    List[np.ndarray]				,    _A      :    float = 0.0   )						->						List[np.ndarray]:
         if attention_mask is not None:
             snake_case      =					np.array(_A				,    np.intaa   )
             snake_case      =					[]

             for vector, length in zip(_A				,    attention_mask.sum(-1   )   ):
                 snake_case      =					(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7   )
                 if length < normed_slice.shape[0]:
                     snake_case      =					padding_value

                 normed_input_values.append(_A   )
         else:
             snake_case      =					[(x - x.mean()) / np.sqrt(x.var() + 1E-7   ) for x in input_values]

         return normed_input_values





     def __call__(self      :    Optional[Any]				,    _A      :    Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]				,    _A      :    Union[bool, str, PaddingStrategy] = False				,    _A      :    Optional[int] = None				,    _A      :    bool = False				,    _A      :    Optional[int] = None				,    _A      :    Optional[bool] = None				,    _A      :    Optional[Union[str, TensorType]] = None				,    _A      :    Optional[int] = None				,    **_A      :    List[Any]				,    )						->						BatchFeature:

         if sampling_rate is not None:
             if sampling_rate != self.sampling_rate:
                 raise ValueError(
                     f'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
                     f' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'
                     f' {self.sampling_rate} and not {sampling_rate}.'   )
         else:
             logger.warning(
                 "It is strongly recommended to pass the ``sampling_rate`` argument to this function. "
                 "Failing to do so can result in silent errors that might be hard to debug."   )

         snake_case      =					isinstance(_A				,    np.ndarray   ) and len(raw_speech.shape   ) > 1
         if is_batched_numpy and len(raw_speech.shape   ) > 2:
             raise ValueError(f'Only mono-channel audio is supported for input to {self}'   )
         snake_case      =					is_batched_numpy or (
             isinstance(_A				,    (list, tuple)   ) and (isinstance(raw_speech[0]				,    (np.ndarray, tuple, list)   ))
         )

         # always return batch
         if not is_batched:
             snake_case      =					[raw_speech]

         # convert into correct format for padding
         snake_case      =					BatchFeature({"input_values": raw_speech}   )

         snake_case      =					self.pad(
             _A				,    padding=_A				,    max_length=_A				,    truncation=_A				,    pad_to_multiple_of=_A				,    return_attention_mask=_A				,    )

         # convert input values to correct format
         snake_case      =					padded_inputs["input_values"]
         if not isinstance(input_values[0]				,    np.ndarray   ):
             snake_case      =					[np.asarray(_A				,    dtype=np.floataa   ) for array in input_values]
         elif (
             not isinstance(_A				,    np.ndarray   )
             and isinstance(input_values[0]				,    np.ndarray   )
             and input_values[0].dtype is np.dtype(np.floataa   )
         ):
             snake_case      =					[array.astype(np.floataa   ) for array in input_values]
         elif isinstance(_A				,    np.ndarray   ) and input_values.dtype is np.dtype(np.floataa   ):
             snake_case      =					input_values.astype(np.floataa   )

         # convert attention_mask to correct format
         snake_case      =					padded_inputs.get("attention_mask"   )
         if attention_mask is not None:
             snake_case      =					[np.asarray(_A				,    dtype=np.intaa   ) for array in attention_mask]

         # zero-mean and unit-variance normalization
         if self.do_normalize:
             snake_case      =					(
                 attention_mask
                 if self._get_padding_strategies(_A				,    max_length=_A   ) is not PaddingStrategy.DO_NOT_PAD
                 else None
             )
             snake_case      =					self.zero_mean_unit_var_norm(
                 padded_inputs["input_values"]				,    attention_mask=_A				,    padding_value=self.padding_value   )

         if return_tensors is not None:
             snake_case      =					padded_inputs.convert_to_tensors(_A   )

         return padded_inputs

