




"""simple docstring"""


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

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


A_						=				logging.get_logger(__name__)


A_						=				{
    '''vocab_file''': '''vocab.json''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
    '''merges_file''': '''merges.txt''',
}

A_						=				{
    '''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'''
        ),
    },
}

A_						=				'''</w>'''
A_						=				'''@@ '''


def 					UpperCAmelCase__       (snake_case__							:					List[Any]    ):
  """simple docstring"""

  _snake_case							:		Tuple								=  set()
  _snake_case							:		Optional[Any]								=  word[0]
  for char in word[1:]:
    pairs.add((prev_char, char)    )
    _snake_case							:		Tuple								=  char
  return pairs


# Speech2Text2 has no max input length
A_						=				{'''facebook/s2t-wav2vec2-large-en-de''': 10_24}






class 					lowercase(     UpperCAmelCase_ ):







 '''simple docstring'''





 lowercase__ 					=					VOCAB_FILES_NAMES
 lowercase__ 					=					PRETRAINED_VOCAB_FILES_MAP
 lowercase__ 					=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 lowercase__ 					=					['input_ids', 'attention_mask']


 def __init__(				self:		Tuple,      a_:		List[Any],      a_:		Dict="<s>",      a_:		Tuple="<pad>",      a_:		str="</s>",      a_:		int="<unk>",      a_:		List[str]=False,      a_:		str=None,      **a_:		Optional[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							:		Tuple								=  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							:		List[str]								=  None
     _snake_case							:		List[Any]								=  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[Any]								=  [tuple(merge.split()[:2]		) for merge in merges]
     _snake_case							:		Tuple								=  dict(zip(__UpperCAmelCase,      range(len(__UpperCAmelCase		)		)		)		)
     _snake_case							:		List[str]								=  {}


 @property
 def  UpperCamelCase_   (				self:		List[str]		):




   '''simple docstring'''



   return len(self.decoder		)


 def  UpperCamelCase_   (				self:		Union[str, Any]		):




   '''simple docstring'''



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


 def  UpperCamelCase_   (				self:		Dict,      a_:		Union[str, Any]		):




   '''simple docstring'''



   _snake_case							:		int								=  tuple(token[:-1]		) + (token[-1] + BPE_TOKEN_MERGES,)
   if token in self.cache:
     return self.cache[token]
   _snake_case							:		Optional[Any]								=  get_pairs(__UpperCAmelCase		)

   if not pairs:
     return token

   while True:
     _snake_case							:		int								=  min(__UpperCAmelCase,      key=lambda a_		: self.bpe_ranks.get(__UpperCAmelCase,      float("""inf"""		)		)		)
     if bigram not in self.bpe_ranks:
       break
     _snake_case		,					_snake_case							:		str								=  bigram
     _snake_case							:		Union[str, Any]								=  []
     _snake_case							:		List[Any]								=  0
     while i < len(__UpperCAmelCase		):
       try:
         _snake_case							:		List[Any]								=  word.index(__UpperCAmelCase,      __UpperCAmelCase		)
       except ValueError:
         new_word.extend(word[i:]		)
         break
       else:
         new_word.extend(word[i:j]		)
         _snake_case							:		Any								=  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							:		int								=  tuple(__UpperCAmelCase		)
     _snake_case							:		Dict								=  new_word
     if len(__UpperCAmelCase		) == 1:
       break
     else:
       _snake_case							:		List[str]								=  get_pairs(__UpperCAmelCase		)
   _snake_case							:		Union[str, Any]								=  """ """.join(__UpperCAmelCase		)
   if word == "\n  " + BPE_TOKEN_MERGES:
     _snake_case							:		Optional[Any]								=  """\n""" + BPE_TOKEN_MERGES

   if word.endswith(__UpperCAmelCase		):
     _snake_case							:		Tuple								=  word.replace(__UpperCAmelCase,      """"""		)

   _snake_case							:		Tuple								=  word.replace(""" """,      __UpperCAmelCase		)
   _snake_case							:		Any								=  word
   return word


 def  UpperCamelCase_   (				self:		Tuple,      a_:		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							:		List[str]								=  text.lower()

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

   _snake_case							:		List[str]								=  []
   for token in text:
     if token:
       split_tokens.extend(list(self.bpe(__UpperCAmelCase		).split(""" """		)		)		)

   return split_tokens


 def  UpperCamelCase_   (				self:		Union[str, Any],      a_:		str		):




   '''simple docstring'''



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


 def  UpperCamelCase_   (				self:		Any,      a_:		int		):




   '''simple docstring'''



   _snake_case							:		Any								=  self.decoder.get(__UpperCAmelCase,      self.unk_token		)
   return result


 def  UpperCamelCase_   (				self:		Dict,      a_:		List[str]		):




   '''simple docstring'''



   _snake_case							:		Optional[int]								=  """ """.join(__UpperCAmelCase		)

   # make sure @@ tokens are concatenated
   _snake_case							:		List[str]								=  """""".join(string.split(__UpperCAmelCase		)		)

   return string


 def  UpperCamelCase_   (				self:		Union[str, Any],      a_:		str,      a_:		Optional[str] = None		):




   '''simple docstring'''



   if not os.path.isdir(__UpperCAmelCase		):
     logger.error(f"Vocabulary path ({save_directory}) should be a directory"		)
     return
   _snake_case							:		List[Any]								=  os.path.join(
       __UpperCAmelCase,      (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]		)
   _snake_case							:		Dict								=  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							:		Any								=  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 a_		: 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							:		Any								=  token_index
       writer.write(""" """.join(__UpperCAmelCase		) + """\n"""		)
       index += 1

   return (vocab_file, merges_file)



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

import numpy as np

from ...audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
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      A							(				UpperCAmelCase_     ):
   __UpperCAmelCase : int      =			['input_values', 'attention_mask']


   def __init__(self     :						Any       ,  __UpperCAmelCase     :						int = 1       ,  __UpperCAmelCase     :						int = 1_6_0_0_0       ,  __UpperCAmelCase     :						float = 0.0       ,  __UpperCAmelCase     :						bool = False       ,  __UpperCAmelCase     :						int = 8_0       ,  __UpperCAmelCase     :						int = 1_6       ,  __UpperCAmelCase     :						int = 6_4       ,  __UpperCAmelCase     :						str = "hann_window"       ,  __UpperCAmelCase     :						float = 1.0       ,  __UpperCAmelCase     :						float = 8_0       ,  __UpperCAmelCase     :						float = 7_6_0_0       ,  __UpperCAmelCase     :						float = 1E-10       ,  __UpperCAmelCase     :						int = 2       ,  __UpperCAmelCase     :						bool = True       ,  **__UpperCAmelCase     :						Any       ,  )						->				str:
         """simple docstring"""



         super().__init__(feature_size=__UpperCAmelCase       ,  sampling_rate=__UpperCAmelCase       ,  padding_value=__UpperCAmelCase       ,  **__UpperCAmelCase		)
         UpperCAmelCase__								=  do_normalize
         UpperCAmelCase__								=  return_attention_mask

         UpperCAmelCase__								=  num_mel_bins
         UpperCAmelCase__								=  hop_length
         UpperCAmelCase__								=  win_length
         UpperCAmelCase__								=  win_function
         UpperCAmelCase__								=  frame_signal_scale
         UpperCAmelCase__								=  fmin
         UpperCAmelCase__								=  fmax
         UpperCAmelCase__								=  mel_floor
         UpperCAmelCase__								=  reduction_factor

         UpperCAmelCase__								=  win_length * sampling_rate // 1_0_0_0
         UpperCAmelCase__								=  hop_length * sampling_rate // 1_0_0_0
         UpperCAmelCase__								=  optimal_fft_length(self.sample_size		)
         UpperCAmelCase__								=  (self.n_fft // 2) + 1

         UpperCAmelCase__								=  window_function(window_length=self.sample_size       ,  name=self.win_function       ,  periodic=__UpperCAmelCase		)

         UpperCAmelCase__								=  mel_filter_bank(
             num_frequency_bins=self.n_freqs       ,  num_mel_filters=self.num_mel_bins       ,  min_frequency=self.fmin       ,  max_frequency=self.fmax       ,  sampling_rate=self.sampling_rate       ,  norm="slaney"       ,  mel_scale="slaney"       ,  )

         if frame_signal_scale != 1.0:
               warnings.warn(
                   "The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers"       ,  __UpperCAmelCase       ,  )
         if reduction_factor != 2.0:
               warnings.warn(
                   "The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers"       ,  __UpperCAmelCase       ,  )


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



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

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

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

         return normed_input_values


   def 			lowercase_      (self     :						Optional[int]       ,  __UpperCAmelCase     :						np.ndarray       ,  )						->				np.ndarray:
         """simple docstring"""



         UpperCAmelCase__								=  spectrogram(
             __UpperCAmelCase       ,  window=self.window       ,  frame_length=self.sample_size       ,  hop_length=self.sample_stride       ,  fft_length=self.n_fft       ,  mel_filters=self.mel_filters       ,  mel_floor=self.mel_floor       ,  log_mel="log10"       ,  )
         return log_mel_spec.T


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



         if audio is None and audio_target is None:
               raise ValueError("You must provide either `audio` or `audio_target` values."		)

         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 audio 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."		)

         if audio is not None:
               UpperCAmelCase__								=  self._process_audio(
                   __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  **__UpperCAmelCase       ,  )
         else:
               UpperCAmelCase__								=  None

         if audio_target is not None:
               UpperCAmelCase__								=  self._process_audio(
                   __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  __UpperCAmelCase       ,  **__UpperCAmelCase       ,  )

               if inputs is None:
                     return inputs_target
               else:
                     UpperCAmelCase__								=  inputs_target["input_values"]
                     UpperCAmelCase__								=  inputs_target.get("attention_mask"		)
                     if decoder_attention_mask is not None:
                           UpperCAmelCase__								=  decoder_attention_mask

         return inputs


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



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

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

         # always return batch
         if not is_batched:
               UpperCAmelCase__								=  [speech]

         # needed to make pad() work on spectrogram inputs
         UpperCAmelCase__								=  self.feature_size

         # convert into correct format for padding
         if is_target:
               UpperCAmelCase__								=  [self._extract_mel_features(__UpperCAmelCase		) for waveform in speech]
               UpperCAmelCase__								=  BatchFeature({"input_values": features}		)
               UpperCAmelCase__								=  self.num_mel_bins
         else:
               UpperCAmelCase__								=  BatchFeature({"input_values": speech}		)

         UpperCAmelCase__								=  self.pad(
             __UpperCAmelCase       ,  padding=__UpperCAmelCase       ,  max_length=__UpperCAmelCase       ,  truncation=__UpperCAmelCase       ,  pad_to_multiple_of=__UpperCAmelCase       ,  return_attention_mask=__UpperCAmelCase       ,  **__UpperCAmelCase       ,  )

         UpperCAmelCase__								=  feature_size_hack

         # convert input values to correct format
         UpperCAmelCase__								=  padded_inputs["input_values"]
         if not isinstance(input_values[0]       ,  np.ndarray		):
               UpperCAmelCase__								=  [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		)
         ):
               UpperCAmelCase__								=  [array.astype(np.floataa		) for array in input_values]
         elif isinstance(__UpperCAmelCase       ,  np.ndarray		) and input_values.dtype is np.dtype(np.floataa		):
               UpperCAmelCase__								=  input_values.astype(np.floataa		)

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

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

         if return_tensors is not None:
               UpperCAmelCase__								=  padded_inputs.convert_to_tensors(__UpperCAmelCase		)

         return padded_inputs





   def 			lowercase_      (self     :						Tuple		)						->				Dict[str, Any]:
         """simple docstring"""



         UpperCAmelCase__								=  super().to_dict()

         # Don't serialize these as they are derived from the other properties.
         UpperCAmelCase__								=  ["window", "mel_filters", "sample_size", "sample_stride", "n_fft", "n_freqs"]
         for name in names:
               if name in output:
                     del output[name]

         return output



