

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


lowercase_									=   logging.get_logger(__name__)


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

# See all BART models at https://huggingface.co/models?filter=bart
lowercase_									=   {
    'vocab_file': {
        'facebook/bart-base': 'https://huggingface.co/facebook/bart-base/resolve/main/vocab.json',
        'facebook/bart-large': 'https://huggingface.co/facebook/bart-large/resolve/main/vocab.json',
        'facebook/bart-large-mnli': 'https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json',
        'facebook/bart-large-cnn': 'https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json',
        'facebook/bart-large-xsum': 'https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json',
        'yjernite/bart_eli5': 'https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json',
    },
    'merges_file': {
        'facebook/bart-base': 'https://huggingface.co/facebook/bart-base/resolve/main/merges.txt',
        'facebook/bart-large': 'https://huggingface.co/facebook/bart-large/resolve/main/merges.txt',
        'facebook/bart-large-mnli': 'https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt',
        'facebook/bart-large-cnn': 'https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt',
        'facebook/bart-large-xsum': 'https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt',
        'yjernite/bart_eli5': 'https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt',
    },
}

lowercase_									=   {
    'facebook/bart-base': 1_0_2_4,
    'facebook/bart-large': 1_0_2_4,
    'facebook/bart-large-mnli': 1_0_2_4,
    'facebook/bart-large-cnn': 1_0_2_4,
    'facebook/bart-large-xsum': 1_0_2_4,
    'yjernite/bart_eli5': 1_0_2_4,
}
@lru_cache()
def      a      (			)    ->      Optional[Any]:





      """simple docstring"""



      _lowercase	       =(
          list(range(ord('!'     )      ,   ord('~'     ) + 1     )     ) + list(range(ord('¡'     )      ,   ord('¬'     ) + 1     )     ) + list(range(ord('®'     )      ,   ord('ÿ'     ) + 1     )     )
      )
      _lowercase	       =bs[:]
      _lowercase	       =0
      for b in range(2**8     ):
            if b not in bs:
                  bs.append(A__     )
                  cs.append(2**8 + n     )
                  n += 1
      _lowercase	       =[chr(A__     ) for n in cs]
      return dict(zip(A__      ,   A__     )     )
def      a      (			A__	:   List[str]     )    ->      str:





      """simple docstring"""



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







class  __lowerCAmelCase     (					SCREAMING_SNAKE_CASE							):
  _a									=  VOCAB_FILES_NAMES
  _a									=  PRETRAINED_VOCAB_FILES_MAP
  _a									=  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  _a									=  ["""input_ids""", """attention_mask"""]




  def __init__(				self							,   lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase="replace"							,   lowerCAmelCase="<s>"							,   lowerCAmelCase="</s>"							,   lowerCAmelCase="</s>"							,   lowerCAmelCase="<s>"							,   lowerCAmelCase="<unk>"							,   lowerCAmelCase="<pad>"							,   lowerCAmelCase="<mask>"							,   lowerCAmelCase=False							,   **lowerCAmelCase							,   )  -> Dict:
        '''simple docstring'''






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

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

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

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

        # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
        _lowercase	       =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					)  -> Any:
        '''simple docstring'''






        return len(self.encoder					)




  def   A__     (				self					)  -> str:
        '''simple docstring'''






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




  def   A__     (				self							,   lowerCAmelCase					)  -> Union[str, Any]:
        '''simple docstring'''






        if token in self.cache:
              return self.cache[token]
        _lowercase	       =tuple(lowerCAmelCase					)
        _lowercase	       =get_pairs(lowerCAmelCase					)

        if not pairs:
              return token

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

                    if word[i] == first and i < len(lowerCAmelCase					) - 1 and word[i + 1] == second:
                          new_word.append(first + second					)
                          i += 2
                    else:
                          new_word.append(word[i]					)
                          i += 1
              _lowercase	       =tuple(lowerCAmelCase					)
              _lowercase	       =new_word
              if len(lowerCAmelCase					) == 1:
                    break
              else:
                    _lowercase	       =get_pairs(lowerCAmelCase					)
        _lowercase	       =' '.join(lowerCAmelCase					)
        _lowercase	       =word
        return word




  def   A__     (				self							,   lowerCAmelCase					)  -> Any:
        '''simple docstring'''






        _lowercase	       =[]
        for token in re.findall(self.pat							,   lowerCAmelCase					):
              _lowercase	       =''.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(lowerCAmelCase					).split(' '					)					)
        return bpe_tokens




  def   A__     (				self							,   lowerCAmelCase					)  -> List[Any]:
        '''simple docstring'''






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




  def   A__     (				self							,   lowerCAmelCase					)  -> Any:
        '''simple docstring'''






        return self.decoder.get(lowerCAmelCase					)




  def   A__     (				self							,   lowerCAmelCase					)  -> List[Any]:
        '''simple docstring'''






        _lowercase	       =''.join(lowerCAmelCase					)
        _lowercase	       =bytearray([self.byte_decoder[c] for c in text]					).decode('utf-8'							,   errors=self.errors					)
        return text




  def   A__     (				self							,   lowerCAmelCase							,   lowerCAmelCase = None					)  -> Tuple[str]:
        '''simple docstring'''






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

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

        _lowercase	       =0
        with open(lowerCAmelCase							,   '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 lowerCAmelCase					: 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!'					)
                          _lowercase	       =token_index
                    writer.write(' '.join(lowerCAmelCase					) + '\n'					)
                    index += 1

        return vocab_file, merge_file




  def   A__     (				self							,   lowerCAmelCase							,   lowerCAmelCase = None					)  -> List[int]:
        '''simple docstring'''






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




  def   A__     (				self							,   lowerCAmelCase							,   lowerCAmelCase = None							,   lowerCAmelCase = False					)  -> List[int]:
        '''simple docstring'''






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

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




  def   A__     (				self							,   lowerCAmelCase							,   lowerCAmelCase = None					)  -> List[int]:
        '''simple docstring'''






        _lowercase	       =[self.sep_token_id]
        _lowercase	       =[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							,   lowerCAmelCase							,   lowerCAmelCase=False							,   **lowerCAmelCase					)  -> Union[str, Any]:
        '''simple docstring'''






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





import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


lowercase_									=   logging.get_logger(__name__)





class  __lowerCAmelCase     (					SCREAMING_SNAKE_CASE							):
  _a									=  ["""input_features"""]




  def __init__(				self							,   lowerCAmelCase=80							,   lowerCAmelCase=16_000							,   lowerCAmelCase=160							,   lowerCAmelCase=30							,   lowerCAmelCase=400							,   lowerCAmelCase=0.0							,   lowerCAmelCase=False							,   **lowerCAmelCase							,   )  -> Any:
        '''simple docstring'''






        super().__init__(
            feature_size=lowerCAmelCase							,   sampling_rate=lowerCAmelCase							,   padding_value=lowerCAmelCase							,   return_attention_mask=lowerCAmelCase							,   **lowerCAmelCase							,   )
        _lowercase	       =n_fft
        _lowercase	       =hop_length
        _lowercase	       =chunk_length
        _lowercase	       =chunk_length * sampling_rate
        _lowercase	       =self.n_samples // hop_length
        _lowercase	       =sampling_rate
        _lowercase	       =mel_filter_bank(
            num_frequency_bins=1 + n_fft // 2							,   num_mel_filters=lowerCAmelCase							,   min_frequency=0.0							,   max_frequency=8000.0							,   sampling_rate=lowerCAmelCase							,   norm='slaney'							,   mel_scale='slaney'							,   )




  def   A__     (				self							,   lowerCAmelCase					)  -> np.ndarray:
        '''simple docstring'''






        _lowercase	       =spectrogram(
            lowerCAmelCase							,   window_function(self.n_fft							,   'hann'					)							,   frame_length=self.n_fft							,   hop_length=self.hop_length							,   power=2.0							,   mel_filters=self.mel_filters							,   log_mel='log10'							,   )
        _lowercase	       =log_spec[:, :-1]
        _lowercase	       =np.maximum(lowerCAmelCase							,   log_spec.max() - 8.0					)
        _lowercase	       =(log_spec + 4.0) / 4.0
        return log_spec




  @staticmethod
  # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
  def   A__     (				lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase = 0.0					)  -> List[np.ndarray]:
        '''simple docstring'''






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

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

                    normed_input_values.append(lowerCAmelCase					)
        else:
              _lowercase	       =[(x - x.mean()) / np.sqrt(x.var() + 1e-7					) for x in input_values]

        return normed_input_values




  def __call__(				self							,   lowerCAmelCase							,   lowerCAmelCase = True							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = "max_length"							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   **lowerCAmelCase							,   )  -> 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.__class__.__name__} was trained using a'''
                        F''' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'''
                        F''' was sampled with {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.'					)

        _lowercase	       =isinstance(lowerCAmelCase							,   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}'''					)
        _lowercase	       =is_batched_numpy or (
            isinstance(lowerCAmelCase							,   (list, tuple)					) and (isinstance(raw_speech[0]							,   (np.ndarray, tuple, list)					))
        )

        if is_batched:
              _lowercase	       =[np.asarray([speech]							,   dtype=np.floataa					).T for speech in raw_speech]
        elif not is_batched and not isinstance(lowerCAmelCase							,   np.ndarray					):
              _lowercase	       =np.asarray(lowerCAmelCase							,   dtype=np.floataa					)
        elif isinstance(lowerCAmelCase							,   np.ndarray					) and raw_speech.dtype is np.dtype(np.floataa					):
              _lowercase	       =raw_speech.astype(np.floataa					)

        # always return batch
        if not is_batched:
              _lowercase	       =[np.asarray([raw_speech]					).T]

        _lowercase	       =BatchFeature({'input_features': raw_speech}					)

        # convert into correct format for padding

        _lowercase	       =self.pad(
            lowerCAmelCase							,   padding=lowerCAmelCase							,   max_length=max_length if max_length else self.n_samples							,   truncation=lowerCAmelCase							,   pad_to_multiple_of=lowerCAmelCase							,   return_attention_mask=return_attention_mask or do_normalize							,   )

        # zero-mean and unit-variance normalization
        if do_normalize:
              _lowercase	       =self.zero_mean_unit_var_norm(
                  padded_inputs['input_features']							,   attention_mask=padded_inputs['attention_mask']							,   padding_value=self.padding_value							,   )
              _lowercase	       =np.stack(padded_inputs['input_features']							,   axis=0					)

        # make sure list is in array format
        _lowercase	       =padded_inputs.get('input_features'					).transpose(2							,   0							,   1					)

        _lowercase	       =[self._np_extract_fbank_features(lowerCAmelCase					) for waveform in input_features[0]]

        if isinstance(input_features[0]							,   lowerCAmelCase					):
              _lowercase	       =[np.asarray(lowerCAmelCase							,   dtype=np.floataa					) for feature in input_features]
        else:
              _lowercase	       =input_features

        if return_attention_mask:
              # rescale from sample (48000) to feature (3000)
              _lowercase	       =padded_inputs['attention_mask'][:, :: self.hop_length]

        if return_tensors is not None:
              _lowercase	       =padded_inputs.convert_to_tensors(lowerCAmelCase					)

        return padded_inputs




  def   A__     (				self					)  -> Dict[str, Any]:
        '''simple docstring'''






        _lowercase	       =copy.deepcopy(self.__dict__					)
        _lowercase	       =self.__class__.__name__
        if "mel_filters" in output:
              del output["mel_filters"]
        return output


