





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


_lowerCamelCase :    Dict  							= logging.get_logger(__name__)


class 						lowerCamelCase       (__lowerCamelCase      ):







							"""simple docstring"""




							UpperCAmelCase_      						=   ["input_features"]






							def __init__( self     :					Tuple,			_UpperCAmelCase     :					List[Any]=8_0,			_UpperCAmelCase     :					Optional[int]=1_6_0_0_0,			_UpperCAmelCase     :					Optional[int]=1_6_0,			_UpperCAmelCase     :					int=3_0,			_UpperCAmelCase     :					Optional[Any]=4_0_0,			_UpperCAmelCase     :					Dict=0.0,			_UpperCAmelCase     :					Union[str, Any]=False,			**_UpperCAmelCase     :					Tuple,			)				->			Union[str, Any]:





														"""simple docstring"""
														super().__init__(
														    feature_size=_UpperCAmelCase,			sampling_rate=_UpperCAmelCase,			padding_value=_UpperCAmelCase,			return_attention_mask=_UpperCAmelCase,			**_UpperCAmelCase,			)
														SCREAMING_SNAKE_CASE__       :      List[Any]     =						n_fft
														SCREAMING_SNAKE_CASE__       :      List[str]     =						hop_length
														SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						chunk_length
														SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						chunk_length * sampling_rate
														SCREAMING_SNAKE_CASE__       :      str     =						self.n_samples // hop_length
														SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						sampling_rate
														SCREAMING_SNAKE_CASE__       :      Tuple     =						mel_filter_bank(
														    num_frequency_bins=1 + n_fft // 2,			num_mel_filters=_UpperCAmelCase,			min_frequency=0.0,			max_frequency=8000.0,			sampling_rate=_UpperCAmelCase,			norm="slaney",			mel_scale="slaney",			)






							def  A_     ( self     :					List[str],			_UpperCAmelCase     :					np.array						)				->			np.ndarray:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						spectrogram(
														    _UpperCAmelCase,			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",			)
														SCREAMING_SNAKE_CASE__       :      Any     =						log_spec[:, :-1]
														SCREAMING_SNAKE_CASE__       :      Optional[int]     =						np.maximum(_UpperCAmelCase,			log_spec.max() - 8.0						)
														SCREAMING_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  A_     ( _UpperCAmelCase     :					List[np.ndarray],			_UpperCAmelCase     :					List[np.ndarray],			_UpperCAmelCase     :					float = 0.0						)				->			List[np.ndarray]:





														"""simple docstring"""
														if attention_mask is not None:
																					SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						np.array(_UpperCAmelCase,			np.intaa						)
																					SCREAMING_SNAKE_CASE__       :      int     =						[]

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

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

														return normed_input_values






							def __call__( self     :					Optional[int],			_UpperCAmelCase     :					Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]],			_UpperCAmelCase     :					bool = True,			_UpperCAmelCase     :					Optional[int] = None,			_UpperCAmelCase     :					Optional[Union[str, TensorType]] = None,			_UpperCAmelCase     :					Optional[bool] = None,			_UpperCAmelCase     :					Optional[str] = "max_length",			_UpperCAmelCase     :					Optional[int] = None,			_UpperCAmelCase     :					Optional[int] = None,			_UpperCAmelCase     :					Optional[bool] = None,			**_UpperCAmelCase     :					Optional[Any],			)				->			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."						)

														SCREAMING_SNAKE_CASE__       :      List[Any]     =						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}'''						)
														SCREAMING_SNAKE_CASE__       :      str     =						is_batched_numpy or (
														    isinstance(_UpperCAmelCase,			(list, tuple)						) and (isinstance(raw_speech[0],			(np.ndarray, tuple, list)						))
														)

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

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

														SCREAMING_SNAKE_CASE__       :      Any     =						BatchFeature({"input_features": raw_speech}						)

														# convert into correct format for padding

														SCREAMING_SNAKE_CASE__       :      Tuple     =						self.pad(
														    _UpperCAmelCase,			padding=_UpperCAmelCase,			max_length=max_length if max_length else self.n_samples,			truncation=_UpperCAmelCase,			pad_to_multiple_of=_UpperCAmelCase,			return_attention_mask=return_attention_mask or do_normalize,			)

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

														# make sure list is in array format
														SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						padded_inputs.get("input_features"						).transpose(2,			0,			1						)

														SCREAMING_SNAKE_CASE__       :      Any     =						[self._np_extract_fbank_features(_UpperCAmelCase						) for waveform in input_features[0]]

														if isinstance(input_features[0],			_UpperCAmelCase						):
																					SCREAMING_SNAKE_CASE__       :      Optional[int]     =						[np.asarray(_UpperCAmelCase,			dtype=np.floataa						) for feature in input_features]
														else:
																					SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						input_features

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

														if return_tensors is not None:
																					SCREAMING_SNAKE_CASE__       :      Any     =						padded_inputs.convert_to_tensors(_UpperCAmelCase						)

														return padded_inputs






							def  A_     ( self     :					Any						)				->			Dict[str, Any]:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      Tuple     =						copy.deepcopy(self.__dict__						)
														SCREAMING_SNAKE_CASE__       :      int     =						self.__class__.__name__
														if "mel_filters" in output:
																					del output["mel_filters"]
														return output













import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


_lowerCamelCase :    int  							= logging.get_logger(__name__)

_lowerCamelCase :    Optional[Any]  							= '''▁'''

_lowerCamelCase :    Dict  							= {'''vocab_file''': '''sentencepiece.bpe.model'''}

_lowerCamelCase :    int  							= {
    '''vocab_file''': {
        '''xlm-roberta-base''': '''https://huggingface.co/xlm-roberta-base/resolve/main/sentencepiece.bpe.model''',
        '''xlm-roberta-large''': '''https://huggingface.co/xlm-roberta-large/resolve/main/sentencepiece.bpe.model''',
        '''xlm-roberta-large-finetuned-conll02-dutch''': (
            '''https://huggingface.co/xlm-roberta-large-finetuned-conll02-dutch/resolve/main/sentencepiece.bpe.model'''
        ),
        '''xlm-roberta-large-finetuned-conll02-spanish''': (
            '''https://huggingface.co/xlm-roberta-large-finetuned-conll02-spanish/resolve/main/sentencepiece.bpe.model'''
        ),
        '''xlm-roberta-large-finetuned-conll03-english''': (
            '''https://huggingface.co/xlm-roberta-large-finetuned-conll03-english/resolve/main/sentencepiece.bpe.model'''
        ),
        '''xlm-roberta-large-finetuned-conll03-german''': (
            '''https://huggingface.co/xlm-roberta-large-finetuned-conll03-german/resolve/main/sentencepiece.bpe.model'''
        ),
    }
}

_lowerCamelCase :    Optional[Any]  							= {
    '''xlm-roberta-base''': 5_1_2,
    '''xlm-roberta-large''': 5_1_2,
    '''xlm-roberta-large-finetuned-conll02-dutch''': 5_1_2,
    '''xlm-roberta-large-finetuned-conll02-spanish''': 5_1_2,
    '''xlm-roberta-large-finetuned-conll03-english''': 5_1_2,
    '''xlm-roberta-large-finetuned-conll03-german''': 5_1_2,
}


class 						lowerCamelCase       (__lowerCamelCase      ):







							"""simple docstring"""




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






							def __init__( self     :					Dict,			_UpperCAmelCase     :					str,			_UpperCAmelCase     :					Optional[int]="<s>",			_UpperCAmelCase     :					Optional[int]="</s>",			_UpperCAmelCase     :					Dict="</s>",			_UpperCAmelCase     :					List[Any]="<s>",			_UpperCAmelCase     :					Union[str, Any]="<unk>",			_UpperCAmelCase     :					List[Any]="<pad>",			_UpperCAmelCase     :					str="<mask>",			_UpperCAmelCase     :					Optional[Dict[str, Any]] = None,			**_UpperCAmelCase     :					List[Any],			)				->			None:





														"""simple docstring"""
														# Mask token behave like a normal word, i.e. include the space before it
														SCREAMING_SNAKE_CASE__       :      int     =						AddedToken(_UpperCAmelCase,			lstrip=_UpperCAmelCase,			rstrip=_UpperCAmelCase						) if isinstance(_UpperCAmelCase,			_UpperCAmelCase						) else mask_token

														SCREAMING_SNAKE_CASE__       :      List[str]     =						{} if sp_model_kwargs is None else sp_model_kwargs

														super().__init__(
														    bos_token=_UpperCAmelCase,			eos_token=_UpperCAmelCase,			unk_token=_UpperCAmelCase,			sep_token=_UpperCAmelCase,			cls_token=_UpperCAmelCase,			pad_token=_UpperCAmelCase,			mask_token=_UpperCAmelCase,			sp_model_kwargs=self.sp_model_kwargs,			**_UpperCAmelCase,			)

														SCREAMING_SNAKE_CASE__       :      str     =						spm.SentencePieceProcessor(**self.sp_model_kwargs						)
														self.sp_model.Load(str(_UpperCAmelCase						)						)
														SCREAMING_SNAKE_CASE__       :      Tuple     =						vocab_file

														# Original fairseq vocab and spm vocab must be "aligned":
														# Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
														# -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
														# fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
														# spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

														# Mimic fairseq token-to-id alignment for the first 4 token
														SCREAMING_SNAKE_CASE__       :      List[str]     =						{"<s>": 0, "<pad>": 1, "</s>": 2, "<unk>": 3}

														# The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
														SCREAMING_SNAKE_CASE__       :      Dict     =						1

														SCREAMING_SNAKE_CASE__       :      int     =						len(self.sp_model						) + self.fairseq_offset
														SCREAMING_SNAKE_CASE__       :      Optional[int]     =						{v: k for k, v in self.fairseq_tokens_to_ids.items()}






							def __getstate__( self     :					str						)				->			Optional[Any]:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      List[Any]     =						self.__dict__.copy()
														SCREAMING_SNAKE_CASE__       :      List[Any]     =						None
														SCREAMING_SNAKE_CASE__       :      Dict     =						self.sp_model.serialized_model_proto()
														return state






							def __setstate__( self     :					int,			_UpperCAmelCase     :					List[Any]						)				->			Any:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      Tuple     =						d

														# for backward compatibility
														if not hasattr(self,			"sp_model_kwargs"						):
																					SCREAMING_SNAKE_CASE__       :      Dict     =						{}

														SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						spm.SentencePieceProcessor(**self.sp_model_kwargs						)
														self.sp_model.LoadFromSerializedProto(self.sp_model_proto						)






							def  A_     ( self     :					Any,			_UpperCAmelCase     :					List[int],			_UpperCAmelCase     :					Optional[List[int]] = None						)				->			List[int]:





														"""simple docstring"""

														if token_ids_a is None:
																					return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
														SCREAMING_SNAKE_CASE__       :      List[str]     =						[self.cls_token_id]
														SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						[self.sep_token_id]
														return cls + token_ids_a + sep + sep + token_ids_a + sep






							def  A_     ( self     :					List[Any],			_UpperCAmelCase     :					List[int],			_UpperCAmelCase     :					Optional[List[int]] = None,			_UpperCAmelCase     :					bool = 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     :					Union[str, Any],			_UpperCAmelCase     :					List[int],			_UpperCAmelCase     :					Optional[List[int]] = None						)				->			List[int]:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      str     =						[self.sep_token_id]
														SCREAMING_SNAKE_CASE__       :      Optional[int]     =						[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]






							@property
							def  A_     ( self     :					List[str]						)				->			List[str]:





														"""simple docstring"""
														return len(self.sp_model						) + self.fairseq_offset + 1  # Add the <mask> token






							def  A_     ( self     :					List[Any]						)				->			Optional[int]:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      str     =						{self.convert_ids_to_tokens(_UpperCAmelCase						): i for i in range(self.vocab_size						)}
														vocab.update(self.added_tokens_encoder						)
														return vocab






							def  A_     ( self     :					List[str],			_UpperCAmelCase     :					str						)				->			List[str]:





														"""simple docstring"""
														return self.sp_model.encode(_UpperCAmelCase,			out_type=_UpperCAmelCase						)






							def  A_     ( self     :					Optional[Any],			_UpperCAmelCase     :					Union[str, Any]						)				->			Union[str, Any]:





														"""simple docstring"""
														if token in self.fairseq_tokens_to_ids:
																					return self.fairseq_tokens_to_ids[token]
														SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						self.sp_model.PieceToId(_UpperCAmelCase						)

														# Need to return unknown token if the SP model returned 0
														return spm_id + self.fairseq_offset if spm_id else self.unk_token_id






							def  A_     ( self     :					Tuple,			_UpperCAmelCase     :					List[str]						)				->			List[str]:





														"""simple docstring"""
														if index in self.fairseq_ids_to_tokens:
																					return self.fairseq_ids_to_tokens[index]
														return self.sp_model.IdToPiece(index - self.fairseq_offset						)






							def  A_     ( self     :					Any,			_UpperCAmelCase     :					int						)				->			List[Any]:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      int     =						"".join(_UpperCAmelCase						).replace(_UpperCAmelCase,			" "						).strip()
														return out_string






							def  A_     ( self     :					Union[str, Any],			_UpperCAmelCase     :					str,			_UpperCAmelCase     :					Optional[str] = None						)				->			Tuple[str]:





														"""simple docstring"""
														if not os.path.isdir(_UpperCAmelCase						):
																					logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''						)
																					return
														SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						os.path.join(
														    _UpperCAmelCase,			(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]						)

														if os.path.abspath(self.vocab_file						) != os.path.abspath(_UpperCAmelCase						) and os.path.isfile(self.vocab_file						):
																					copyfile(self.vocab_file,			_UpperCAmelCase						)
														elif not os.path.isfile(self.vocab_file						):
																					with open(_UpperCAmelCase,			"wb"						) as fi:
																												SCREAMING_SNAKE_CASE__       :      Any     =						self.sp_model.serialized_model_proto()
																												fi.write(_UpperCAmelCase						)

														return (out_vocab_file,)






