



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


a_  :List[Any]     = logging.get_logger(__name__)




class    snake_case__		(			lowerCamelCase__	):






     """simple docstring"""


     _SCREAMING_SNAKE_CASE          =      ["""input_features"""]






     def __init__(			self    :    Optional[int],					_snake_case    :    List[Any]=8_0,					_snake_case    :    List[Any]=1_6_0_0_0,					_snake_case    :    Optional[int]=1_6_0,					_snake_case    :    List[str]=3_0,					_snake_case    :    Dict=4_0_0,					_snake_case    :    List[str]=0.0,					_snake_case    :    Any=False,					**_snake_case    :    Union[str, Any],					)  ->Optional[int]:
            super().__init__(
                feature_size=a__,					sampling_rate=a__,					padding_value=a__,					return_attention_mask=a__,					**a__,					)
            snake_case__				:      int        =      n_fft
            snake_case__				:      List[Any]        =      hop_length
            snake_case__				:      Tuple        =      chunk_length
            snake_case__				:      Optional[Any]        =      chunk_length * sampling_rate
            snake_case__				:      Optional[Any]        =      self.n_samples // hop_length
            snake_case__				:      List[str]        =      sampling_rate
            snake_case__				:      Optional[int]        =      mel_filter_bank(
                num_frequency_bins=1 + n_fft // 2,					num_mel_filters=a__,					min_frequency=0.0,					max_frequency=8_0_0_0.0,					sampling_rate=a__,					norm='slaney',					mel_scale='slaney',					)






     def   lowercase_  (			self    :    Union[str, Any],					_snake_case    :    Optional[int]   )  ->List[str]:
            snake_case__				:      Any        =      spectrogram(
                a__,					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',					)
            snake_case__				:      str        =      log_spec[:, :-1]
            snake_case__				:      Union[str, Any]        =      np.maximum(a__,					log_spec.max() - 8.0   )
            snake_case__				:      Optional[int]        =      (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   lowercase_  (			_snake_case    :    Dict,					_snake_case    :    List[str],					_snake_case    :    List[Any] = 0.0   )  ->Optional[Any]:





            if attention_mask is not None:
                   snake_case__				:      Dict        =      np.array(a__,					np.intaa   )
                   snake_case__				:      Any        =      []

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

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

            return normed_input_values






     def __call__(			self    :    Tuple,					_snake_case    :    Union[str, Any],					_snake_case    :    Tuple = True,					_snake_case    :    List[Any] = None,					_snake_case    :    List[str] = None,					_snake_case    :    List[Any] = None,					_snake_case    :    Any = "max_length",					_snake_case    :    Tuple = None,					_snake_case    :    int = None,					_snake_case    :    List[str] = None,					**_snake_case    :    str,					)  ->List[Any]:





            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.'   )

            snake_case__				:      Dict        =      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__				:      int        =      is_batched_numpy or (
                isinstance(a__,					(list, tuple)   ) and (isinstance(raw_speech[0],					(np.ndarray, tuple, list)   ))
            )

            if is_batched:
                   snake_case__				:      List[Any]        =      [np.asarray([speech],					dtype=np.floataa   ).T for speech in raw_speech]
            elif not is_batched and not isinstance(a__,					np.ndarray   ):
                   snake_case__				:      str        =      np.asarray(a__,					dtype=np.floataa   )
            elif isinstance(a__,					np.ndarray   ) and raw_speech.dtype is np.dtype(np.floataa   ):
                   snake_case__				:      str        =      raw_speech.astype(np.floataa   )

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

            snake_case__				:      List[str]        =      BatchFeature({'input_features': raw_speech}   )

            # convert into correct format for padding

            snake_case__				:      Optional[Any]        =      self.pad(
                a__,					padding=a__,					max_length=max_length if max_length else self.n_samples,					truncation=a__,					pad_to_multiple_of=a__,					return_attention_mask=return_attention_mask or do_normalize,					)

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

            # make sure list is in array format
            snake_case__				:      Optional[Any]        =      padded_inputs.get('input_features'   ).transpose(2,					0,					1   )

            snake_case__				:      Tuple        =      [self._np_extract_fbank_features(a__   ) for waveform in input_features[0]]

            if isinstance(input_features[0],					a__   ):
                   snake_case__				:      str        =      [np.asarray(a__,					dtype=np.floataa   ) for feature in input_features]
            else:
                   snake_case__				:      Dict        =      input_features

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

            if return_tensors is not None:
                   snake_case__				:      str        =      padded_inputs.convert_to_tensors(a__   )

            return padded_inputs






     def   lowercase_  (			self    :    List[Any]   )  ->Dict:
            snake_case__				:      List[Any]        =      copy.deepcopy(self.__dict__   )
            snake_case__				:      int        =      self.__class__.__name__
            if "mel_filters" in output:
                   del output["mel_filters"]
            return output













import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

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


_lowerCamelCase      							=   logging.get_logger(__name__)

_lowerCamelCase      							=   '▁'

_lowerCamelCase      							=   {
    'vocab_file': 'vocab.json',
    'spm_file': 'sentencepiece.bpe.model',
}

_lowerCamelCase      							=   {
    'vocab_file': {
        'facebook/s2t-small-librispeech-asr': (
            'https://huggingface.co/facebook/s2t-small-librispeech-asr/resolve/main/vocab.json'
        ),
    },
    'spm_file': {
        'facebook/s2t-small-librispeech-asr': (
            'https://huggingface.co/facebook/s2t-small-librispeech-asr/resolve/main/sentencepiece.bpe.model'
        )
    },
}

_lowerCamelCase      							=   {
    'facebook/s2t-small-librispeech-asr': 1024,
}

_lowerCamelCase      							=   ['pt', 'fr', 'ru', 'nl', 'ro', 'it', 'es', 'de']

_lowerCamelCase      							=   {'mustc': MUSTC_LANGS}







class       __A							(   lowerCamelCase__	):




      """simple docstring"""







      UpperCAmelCase__							=   VOCAB_FILES_NAMES
      UpperCAmelCase__							=   PRETRAINED_VOCAB_FILES_MAP
      UpperCAmelCase__							=   MAX_MODEL_INPUT_SIZES
      UpperCAmelCase__							=   ["""input_ids""", """attention_mask"""]

      UpperCAmelCase__							=   []


      def __init__(		self   ,				a__   ,				a__   ,				a__="<s>"   ,				a__="</s>"   ,				a__="<pad>"   ,				a__="<unk>"   ,				a__=False   ,				a__=False   ,				a__=None   ,				a__=None   ,				a__ = None   ,				**a__   ,				):



             """simple docstring"""





             _lowerCamelCase				:  str						=		{} if sp_model_kwargs is None else sp_model_kwargs

             super().__init__(
                 bos_token=a__   ,				eos_token=a__   ,				unk_token=a__   ,				pad_token=a__   ,				do_upper_case=a__   ,				do_lower_case=a__   ,				tgt_lang=a__   ,				lang_codes=a__   ,				sp_model_kwargs=self.sp_model_kwargs   ,				**a__   ,				)
             _lowerCamelCase				:  Optional[int]						=		do_upper_case
             _lowerCamelCase				:  Optional[Any]						=		do_lower_case

             _lowerCamelCase				:  Tuple						=		load_json(a__)
             _lowerCamelCase				:  Union[str, Any]						=		{v: k for k, v in self.encoder.items()}
             _lowerCamelCase				:  Tuple						=		spm_file
             _lowerCamelCase				:  Any						=		load_spm(a__   ,				self.sp_model_kwargs)

             if lang_codes is not None:
                    _lowerCamelCase				:  List[Any]						=		lang_codes
                    _lowerCamelCase				:  List[str]						=		LANGUAGES[lang_codes]
                    _lowerCamelCase				:  Any						=		[F"""<lang:{lang}>""" for lang in self.langs]
                    _lowerCamelCase				:  Optional[Any]						=		{lang: self.sp_model.PieceToId(F"""<lang:{lang}>""") for lang in self.langs}

                    _lowerCamelCase				:  List[str]						=		self.lang_tokens
                    _lowerCamelCase				:  str						=		tgt_lang if tgt_lang is not None else self.langs[0]

                    self.set_tgt_lang_special_tokens(self._tgt_lang)
             else:
                    _lowerCamelCase				:  Any						=		{}


      @property
      def 					__snake_case      (		self):



             """simple docstring"""





             return len(self.encoder)


      @property
      def 					__snake_case      (		self):



             """simple docstring"""





             return self._tgt_lang


      @tgt_lang.setter
      def 					__snake_case      (		self   ,				a__):



             """simple docstring"""





             _lowerCamelCase				:  Any						=		new_tgt_lang
             self.set_tgt_lang_special_tokens(a__)


      def 					__snake_case      (		self   ,				a__):



             """simple docstring"""





             _lowerCamelCase				:  Optional[Any]						=		self.lang_code_to_id[tgt_lang]
             _lowerCamelCase				:  Any						=		[lang_code_id]


      def 					__snake_case      (		self   ,				a__):



             """simple docstring"""





             return self.sp_model.encode(a__   ,				out_type=a__)


      def 					__snake_case      (		self   ,				a__):



             """simple docstring"""





             return self.encoder.get(a__   ,				self.encoder[self.unk_token])


      def 					__snake_case      (		self   ,				a__):



             """simple docstring"""





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


      def 					__snake_case      (		self   ,				a__):



             """simple docstring"""





             _lowerCamelCase				:  Union[str, Any]						=		[]
             _lowerCamelCase				:  List[str]						=		''''''
             for token in tokens:
                    # make sure that special tokens are not decoded using sentencepiece model
                    if token in self.all_special_tokens:
                           _lowerCamelCase				:  List[Any]						=		self.sp_model.decode(a__)
                           out_string += (decoded.upper() if self.do_upper_case else decoded) + token + " "
                           _lowerCamelCase				:  Optional[int]						=		[]
                    else:
                           current_sub_tokens.append(a__)
             _lowerCamelCase				:  Tuple						=		self.sp_model.decode(a__)
             out_string += decoded.upper() if self.do_upper_case else decoded
             return out_string.strip()


      def 					__snake_case      (		self   ,				a__   ,				a__=None):



             """simple docstring"""





             if token_ids_a is None:
                    return self.prefix_tokens + token_ids_a + [self.eos_token_id]
             # We don't expect to process pairs, but leave the pair logic for API consistency
             return self.prefix_tokens + token_ids_a + token_ids_a + [self.eos_token_id]


      def 					__snake_case      (		self   ,				a__   ,				a__ = None   ,				a__ = False):



             """simple docstring"""





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

             _lowerCamelCase				:  Tuple						=		[1] * len(self.prefix_tokens)
             _lowerCamelCase				:  Tuple						=		[1]
             if token_ids_a is None:
                    return prefix_ones + ([0] * len(a__)) + suffix_ones
             return prefix_ones + ([0] * len(a__)) + ([0] * len(a__)) + suffix_ones


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Union[str, Any]						=		self.encoder.copy()
             vocab.update(self.added_tokens_encoder)
             return vocab


      def __getstate__(		self):



             """simple docstring"""





             _lowerCamelCase				:  Union[str, Any]						=		self.__dict__.copy()
             _lowerCamelCase				:  str						=		None
             return state


      def __setstate__(		self   ,				a__):



             """simple docstring"""





             _lowerCamelCase				:  Optional[int]						=		d

             # for backward compatibility
             if not hasattr(self   ,				'''sp_model_kwargs'''):
                    _lowerCamelCase				:  List[Any]						=		{}

             _lowerCamelCase				:  Dict						=		load_spm(self.spm_file   ,				self.sp_model_kwargs)



      def 					__snake_case      (		self   ,				a__   ,				a__ = None):



             """simple docstring"""





             _lowerCamelCase				:  str						=		Path(a__)
             assert save_dir.is_dir(), F"""{save_directory} should be a directory"""
             _lowerCamelCase				:  Any						=		save_dir / (
                 (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''vocab_file''']
             )
             _lowerCamelCase				:  Optional[Any]						=		save_dir / (
                 (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''spm_file''']
             )

             save_json(self.encoder   ,				a__)

             if os.path.abspath(self.spm_file) != os.path.abspath(a__) and os.path.isfile(self.spm_file):
                    copyfile(self.spm_file   ,				a__)
             elif not os.path.isfile(self.spm_file):
                    with open(a__   ,				'''wb''') as fi:
                           _lowerCamelCase				:  List[str]						=		self.sp_model.serialized_model_proto()
                           fi.write(a__)

             return (str(a__), str(a__))




def  __UpperCAmelCase(   lowercase_			,     lowercase_   ):
       _lowerCamelCase				:  Optional[Any]						=		sentencepiece.SentencePieceProcessor(**lowercase_   )
       spm.Load(str(lowercase_   )   )
       return spm




def  __UpperCAmelCase(   lowercase_   ):
       with open(lowercase_			,     '''r'''   ) as f:
              return json.load(lowercase_   )




def  __UpperCAmelCase(   lowercase_			,     lowercase_   ):
       with open(lowercase_			,     '''w'''   ) as f:
              json.dump(lowercase_			,     lowercase_			,     indent=2   )

