

import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


UpperCamelCase__		   =    logging.get_logger(__name__)

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

UpperCamelCase__		   =    {
    '''vocab_file''': {
        '''roberta-base''': '''https://huggingface.co/roberta-base/resolve/main/vocab.json''',
        '''roberta-large''': '''https://huggingface.co/roberta-large/resolve/main/vocab.json''',
        '''roberta-large-mnli''': '''https://huggingface.co/roberta-large-mnli/resolve/main/vocab.json''',
        '''distilroberta-base''': '''https://huggingface.co/distilroberta-base/resolve/main/vocab.json''',
        '''roberta-base-openai-detector''': '''https://huggingface.co/roberta-base-openai-detector/resolve/main/vocab.json''',
        '''roberta-large-openai-detector''': (
            '''https://huggingface.co/roberta-large-openai-detector/resolve/main/vocab.json'''
        ),
    },
    '''merges_file''': {
        '''roberta-base''': '''https://huggingface.co/roberta-base/resolve/main/merges.txt''',
        '''roberta-large''': '''https://huggingface.co/roberta-large/resolve/main/merges.txt''',
        '''roberta-large-mnli''': '''https://huggingface.co/roberta-large-mnli/resolve/main/merges.txt''',
        '''distilroberta-base''': '''https://huggingface.co/distilroberta-base/resolve/main/merges.txt''',
        '''roberta-base-openai-detector''': '''https://huggingface.co/roberta-base-openai-detector/resolve/main/merges.txt''',
        '''roberta-large-openai-detector''': (
            '''https://huggingface.co/roberta-large-openai-detector/resolve/main/merges.txt'''
        ),
    },
}

UpperCamelCase__		   =    {
    '''roberta-base''': 512,
    '''roberta-large''': 512,
    '''roberta-large-mnli''': 512,
    '''distilroberta-base''': 512,
    '''roberta-base-openai-detector''': 512,
    '''roberta-large-openai-detector''': 512,
}







@lru_cache()
def 	lowerCamelCase__			(   ):


       """simple docstring"""




       __snake_case					  =       (
           list(range(ord("""!"""		)  ,ord("""~"""		) + 1		)		) + list(range(ord("""¡"""		)  ,ord("""¬"""		) + 1		)		) + list(range(ord("""®"""		)  ,ord("""ÿ"""		) + 1		)		)
       )
       __snake_case					  =       bs[:]
       __snake_case					  =       0
       for b in range(2**8		):
              if b not in bs:
                     bs.append(__A		)
                     cs.append(2**8 + n		)
                     n += 1
       __snake_case					  =       [chr(__A		) for n in cs]
       return dict(zip(__A  ,__A		)		)







def 	lowerCamelCase__			(   __A						:Any		):


       """simple docstring"""




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




class 						__snake_case						(							snake_case__     ):
       """simple docstring"""







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


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






              """simple docstring"""


              __snake_case					  =       AddedToken(_UpperCamelCase			,   lstrip=_UpperCamelCase			,   rstrip=_UpperCamelCase      ) if isinstance(_UpperCamelCase			,   _UpperCamelCase      ) else bos_token
              __snake_case					  =       AddedToken(_UpperCamelCase			,   lstrip=_UpperCamelCase			,   rstrip=_UpperCamelCase      ) if isinstance(_UpperCamelCase			,   _UpperCamelCase      ) else eos_token
              __snake_case					  =       AddedToken(_UpperCamelCase			,   lstrip=_UpperCamelCase			,   rstrip=_UpperCamelCase      ) if isinstance(_UpperCamelCase			,   _UpperCamelCase      ) else sep_token
              __snake_case					  =       AddedToken(_UpperCamelCase			,   lstrip=_UpperCamelCase			,   rstrip=_UpperCamelCase      ) if isinstance(_UpperCamelCase			,   _UpperCamelCase      ) else cls_token
              __snake_case					  =       AddedToken(_UpperCamelCase			,   lstrip=_UpperCamelCase			,   rstrip=_UpperCamelCase      ) if isinstance(_UpperCamelCase			,   _UpperCamelCase      ) else unk_token
              __snake_case					  =       AddedToken(_UpperCamelCase			,   lstrip=_UpperCamelCase			,   rstrip=_UpperCamelCase      ) if isinstance(_UpperCamelCase			,   _UpperCamelCase      ) else pad_token

              # Mask token behave like a normal word, i.e. include the space before it
              __snake_case					  =       AddedToken(_UpperCamelCase			,   lstrip=_UpperCamelCase			,   rstrip=_UpperCamelCase      ) if isinstance(_UpperCamelCase			,   _UpperCamelCase      ) else mask_token

              super().__init__(
                  errors=_UpperCamelCase			,   bos_token=_UpperCamelCase			,   eos_token=_UpperCamelCase			,   unk_token=_UpperCamelCase			,   sep_token=_UpperCamelCase			,   cls_token=_UpperCamelCase			,   pad_token=_UpperCamelCase			,   mask_token=_UpperCamelCase			,   add_prefix_space=_UpperCamelCase			,   **_UpperCamelCase			,   )

              with open(_UpperCamelCase			,   encoding="""utf-8"""      ) as vocab_handle:
                     __snake_case					  =       json.load(_UpperCamelCase      )
              __snake_case					  =       {v: k for k, v in self.encoder.items()}
              __snake_case					  =       errors  # how to handle errors in decoding
              __snake_case					  =       bytes_to_unicode()
              __snake_case					  =       {v: k for k, v in self.byte_encoder.items()}
              with open(_UpperCamelCase			,   encoding="""utf-8"""      ) as merges_handle:
                     __snake_case					  =       merges_handle.read().split("""\n"""      )[1:-1]
              __snake_case					  =       [tuple(merge.split()      ) for merge in bpe_merges]
              __snake_case					  =       dict(zip(_UpperCamelCase			,   range(len(_UpperCamelCase      )      )      )      )
              __snake_case					  =       {}
              __snake_case					  =       add_prefix_space

              # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
              __snake_case					  =       re.compile(r"""'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"""      )


       @property
       def  a		(		self      )			->       Union[str, Any]:






              """simple docstring"""


              return len(self.encoder      )


       def  a		(		self      )			->       str:






              """simple docstring"""


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


       def  a		(		self			,   _UpperCamelCase      )			->       Union[str, Any]:






              """simple docstring"""


              if token in self.cache:
                     return self.cache[token]
              __snake_case					  =       tuple(_UpperCamelCase      )
              __snake_case					  =       get_pairs(_UpperCamelCase      )

              if not pairs:
                     return token

              while True:
                     __snake_case					  =       min(_UpperCamelCase			,   key=lambda _UpperCamelCase      : self.bpe_ranks.get(_UpperCamelCase			,   float("""inf"""      )      )      )
                     if bigram not in self.bpe_ranks:
                            break
                     __snake_case  ,						__snake_case					  =       bigram
                     __snake_case					  =       []
                     __snake_case					  =       0
                     while i < len(_UpperCamelCase      ):
                            try:
                                   __snake_case					  =       word.index(_UpperCamelCase			,   _UpperCamelCase      )
                            except ValueError:
                                   new_word.extend(word[i:]      )
                                   break
                            else:
                                   new_word.extend(word[i:j]      )
                                   __snake_case					  =       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					  =       tuple(_UpperCamelCase      )
                     __snake_case					  =       new_word
                     if len(_UpperCamelCase      ) == 1:
                            break
                     else:
                            __snake_case					  =       get_pairs(_UpperCamelCase      )
              __snake_case					  =       """ """.join(_UpperCamelCase      )
              __snake_case					  =       word
              return word


       def  a		(		self			,   _UpperCamelCase      )			->       Dict:






              """simple docstring"""


              __snake_case					  =       []
              for token in re.findall(self.pat			,   _UpperCamelCase      ):
                     __snake_case					  =       """""".join(
                         self.byte_encoder[b] for b in token.encode("""utf-8"""      )      )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
                     bpe_tokens.extend(bpe_token for bpe_token in self.bpe(_UpperCamelCase      ).split(""" """      )      )
              return bpe_tokens


       def  a		(		self			,   _UpperCamelCase      )			->       List[str]:






              """simple docstring"""


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


       def  a		(		self			,   _UpperCamelCase      )			->       Dict:






              """simple docstring"""


              return self.decoder.get(_UpperCamelCase      )


       def  a		(		self			,   _UpperCamelCase      )			->       Tuple:






              """simple docstring"""


              __snake_case					  =       """""".join(_UpperCamelCase      )
              __snake_case					  =       bytearray([self.byte_decoder[c] for c in text]      ).decode("""utf-8"""			,   errors=self.errors      )
              return text


       def  a		(		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					  =       os.path.join(
                  _UpperCamelCase			,   (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]      )
              __snake_case					  =       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					  =       0
              with open(_UpperCamelCase			,   """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 _UpperCamelCase      : 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!"""      )
                                   __snake_case					  =       token_index
                            writer.write(""" """.join(_UpperCamelCase      ) + """\n"""      )
                            index += 1

              return vocab_file, merge_file


       def  a		(		self			,   _UpperCamelCase			,   _UpperCamelCase = None      )			->       List[int]:






              """simple docstring"""


              if token_ids_a is None:
                     return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
              __snake_case					  =       [self.cls_token_id]
              __snake_case					  =       [self.sep_token_id]
              return cls + token_ids_a + sep + sep + token_ids_a + sep


       def  a		(		self			,   _UpperCamelCase			,   _UpperCamelCase = None			,   _UpperCamelCase = False      )			->       List[int]:






              """simple docstring"""


              if already_has_special_tokens:
                     return super().get_special_tokens_mask(
                         token_ids_a=_UpperCamelCase			,   token_ids_a=_UpperCamelCase			,   already_has_special_tokens=_UpperCamelCase      )

              if token_ids_a is None:
                     return [1] + ([0] * len(_UpperCamelCase      )) + [1]
              return [1] + ([0] * len(_UpperCamelCase      )) + [1, 1] + ([0] * len(_UpperCamelCase      )) + [1]


       def  a		(		self			,   _UpperCamelCase			,   _UpperCamelCase = None      )			->       List[int]:






              """simple docstring"""


              __snake_case					  =       [self.sep_token_id]
              __snake_case					  =       [self.cls_token_id]

              if token_ids_a is None:
                     return len(cls + token_ids_a + sep      ) * [0]
              return len(cls + token_ids_a + sep + sep + token_ids_a + sep      ) * [0]






       def  a		(		self			,   _UpperCamelCase			,   _UpperCamelCase=False			,   **_UpperCamelCase      )			->       Tuple:






              """simple docstring"""


              __snake_case					  =       kwargs.pop("""add_prefix_space"""			,   self.add_prefix_space      )
              if (is_split_into_words or add_prefix_space) and (len(_UpperCamelCase      ) > 0 and not text[0].isspace()):
                     __snake_case					  =       """ """ + text
              return (text, kwargs)




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


UpperCamelCase__		   =    logging.get_logger(__name__)

class 						__snake_case						(							snake_case__     ):
       """simple docstring"""







       __SCREAMING_SNAKE_CASE													=       ["input_values", "attention_mask"]


       def __init__(		self			,   _UpperCamelCase=1			,   _UpperCamelCase=1_60_00			,   _UpperCamelCase=0.0			,   _UpperCamelCase=False			,   _UpperCamelCase=True			,   **_UpperCamelCase			,   )			->       List[Any]:






              """simple docstring"""


              super().__init__(feature_size=_UpperCamelCase			,   sampling_rate=_UpperCamelCase			,   padding_value=_UpperCamelCase			,   **_UpperCamelCase      )
              __snake_case					  =       return_attention_mask
              __snake_case					  =       do_normalize


       @staticmethod
       def  a		(		_UpperCamelCase			,   _UpperCamelCase			,   _UpperCamelCase = 0.0      )			->       List[np.ndarray]:






              """simple docstring"""


              if attention_mask is not None:
                     __snake_case					  =       np.array(_UpperCamelCase			,   np.intaa      )
                     __snake_case					  =       []

                     for vector, length in zip(_UpperCamelCase			,   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(_UpperCamelCase      )
              else:
                     __snake_case					  =       [(x - x.mean()) / np.sqrt(x.var() + 1E-7      ) for x in input_values]

              return normed_input_values






       def __call__(		self			,   _UpperCamelCase			,   _UpperCamelCase = False			,   _UpperCamelCase = None			,   _UpperCamelCase = False			,   _UpperCamelCase = None			,   _UpperCamelCase = None			,   _UpperCamelCase = None			,   _UpperCamelCase = None			,   **_UpperCamelCase			,   )			->       BatchFeature:






              """simple docstring"""


              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(_UpperCamelCase			,   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(_UpperCamelCase			,   (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(
                  _UpperCamelCase			,   padding=_UpperCamelCase			,   max_length=_UpperCamelCase			,   truncation=_UpperCamelCase			,   pad_to_multiple_of=_UpperCamelCase			,   return_attention_mask=_UpperCamelCase			,   )

              # convert input values to correct format
              __snake_case					  =       padded_inputs["""input_values"""]
              if not isinstance(input_values[0]			,   np.ndarray      ):
                     __snake_case					  =       [np.asarray(_UpperCamelCase			,   dtype=np.floataa      ) for array in input_values]
              elif (
                  not isinstance(_UpperCamelCase			,   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(_UpperCamelCase			,   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(_UpperCamelCase			,   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(_UpperCamelCase			,   max_length=_UpperCamelCase      ) is not PaddingStrategy.DO_NOT_PAD
                         else None
                     )
                     __snake_case					  =       self.zero_mean_unit_var_norm(
                         padded_inputs["""input_values"""]			,   attention_mask=_UpperCamelCase			,   padding_value=self.padding_value      )

              if return_tensors is not None:
                     __snake_case					  =       padded_inputs.convert_to_tensors(_UpperCamelCase      )

              return padded_inputs

