
from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


_lowerCamelCase       :				int       =			logging.get_logger(__name__)




class    SCREAMING_SNAKE_CASE__     (	UpperCAmelCase					):





	'''simple docstring'''



	_UpperCAmelCase       :		Optional[int]       =					["input_features", "attention_mask"]



	def __init__(  self	:    str			,  lowercase	:    Any=80			,  lowercase	:    Any=16_000			,  lowercase	:    int=80			,  lowercase	:    List[Any]=0.0			,  lowercase	:    Union[str, Any]=True			,  lowercase	:    Union[str, Any]=True			,  lowercase	:    Optional[Any]=True			,  **lowercase	:    List[str]			,  ):


				'''simple docstring'''


				super().__init__(feature_size=lowercase			,  sampling_rate=lowercase			,  padding_value=lowercase			,  **lowercase	)
				_snake_case		=			num_mel_bins
				_snake_case		=			do_ceptral_normalize
				_snake_case		=			normalize_means
				_snake_case		=			normalize_vars
				_snake_case		=			True



	def  A							(  self	:    Tuple			,  lowercase	:    np.ndarray			,  ):


				'''simple docstring'''


				_snake_case		=			waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
				_snake_case		=			torch.from_numpy(lowercase	).unsqueeze(0	)
				_snake_case		=			ta_kaldi.fbank(lowercase			,  num_mel_bins=self.num_mel_bins			,  sample_frequency=self.sampling_rate	)
				return features.numpy()



	@staticmethod
	def  A							(  lowercase	:    np.ndarray			,  lowercase	:    int			,  lowercase	:    Optional[bool] = True			,  lowercase	:    Optional[bool] = True			,  lowercase	:    float = 0.0			,  ):


				'''simple docstring'''


				if normalize_means:
							_snake_case		=			x[:input_length].mean(axis=0	)
							_snake_case		=			np.subtract(lowercase			,  lowercase	)
				if normalize_vars:
							_snake_case		=			x[:input_length].std(axis=0	)
							_snake_case		=			np.divide(lowercase			,  lowercase	)

				if input_length < x.shape[0]:
							_snake_case		=			padding_value

				# make sure array is in float32
				_snake_case		=			x.astype(np.floataa	)

				return x



	def  A							(  self	:    Dict			,  lowercase	:    List[np.ndarray]			,  lowercase	:    Optional[np.ndarray] = None	):


				'''simple docstring'''


				_snake_case		=			attention_mask.sum(-1	) if attention_mask is not None else [x.shape[0] for x in input_features]
				return [
				    self.utterance_cmvn(lowercase			,  lowercase			,  self.normalize_means			,  self.normalize_vars			,  self.padding_value	)
				    for x, n in zip(lowercase			,  lowercase	)
				]





	def __call__(  self	:    int			,  lowercase	:    Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]			,  lowercase	:    Union[bool, str, PaddingStrategy] = False			,  lowercase	:    Optional[int] = None			,  lowercase	:    bool = False			,  lowercase	:    Optional[int] = None			,  lowercase	:    Optional[Union[str, TensorType]] = None			,  lowercase	:    Optional[int] = None			,  lowercase	:    Optional[bool] = None			,  **lowercase	:    Union[str, Any]			,  ):


				'''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(lowercase			,  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(lowercase			,  (list, tuple)	) and (isinstance(raw_speech[0]			,  (np.ndarray, tuple, list)	))
				)

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

				# always return batch
				if not is_batched:
							_snake_case		=			[raw_speech]

				# extract fbank features
				_snake_case		=			[self._extract_fbank_features(lowercase	) for waveform in raw_speech]

				# convert into correct format for padding
				_snake_case		=			BatchFeature({'input_features': features}	)

				_snake_case		=			self.pad(
				    lowercase			,  padding=lowercase			,  max_length=lowercase			,  truncation=lowercase			,  pad_to_multiple_of=lowercase			,  return_attention_mask=lowercase			,  **lowercase			,  )

				# make sure list is in array format
				_snake_case		=			padded_inputs.get('input_features'	)
				if isinstance(input_features[0]			,  lowercase	):
							_snake_case		=			[np.asarray(lowercase			,  dtype=np.floataa	) for feature in input_features]

				_snake_case		=			padded_inputs.get('attention_mask'	)
				if attention_mask is not None:
							_snake_case		=			[np.asarray(lowercase			,  dtype=np.intaa	) for array in attention_mask]

				# Utterance-level cepstral mean and variance normalization
				if self.do_ceptral_normalize:
							_snake_case		=			(
							    np.array(lowercase			,  dtype=np.intaa	)
							    if self._get_padding_strategies(lowercase			,  max_length=lowercase	) is not PaddingStrategy.DO_NOT_PAD
							    else None
							)
							_snake_case		=			self.normalize(
							    padded_inputs['input_features']			,  attention_mask=lowercase	)

				if return_tensors is not None:
							_snake_case		=			padded_inputs.convert_to_tensors(lowercase	)

				return padded_inputs

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

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
			from .tokenization_pegasus import PegasusTokenizer
else:
			_lowerCamelCase       :				int       =			None


_lowerCamelCase       :				List[str]       =			logging.get_logger(__name__)


_lowerCamelCase       :				Tuple       =			'''▁'''

_lowerCamelCase       :				Optional[Any]       =			{'''vocab_file''': '''spiece.model''', '''tokenizer_file''': '''tokenizer.json'''}

_lowerCamelCase       :				Any       =			{
    '''vocab_file''': {'''google/pegasus-xsum''': '''https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model'''},
    '''tokenizer_file''': {
        '''google/pegasus-xsum''': '''https://huggingface.co/google/pegasus-xsum/resolve/main/tokenizer.json'''
    },
}

_lowerCamelCase       :				Optional[int]       =			{
    '''google/pegasus-xsum''': 512,
}




class    SCREAMING_SNAKE_CASE__     (	UpperCAmelCase					):





	'''simple docstring'''



	_UpperCAmelCase       :		int       =					VOCAB_FILES_NAMES
	_UpperCAmelCase       :		Union[str, Any]       =					PRETRAINED_VOCAB_FILES_MAP
	_UpperCAmelCase       :		Union[str, Any]       =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	_UpperCAmelCase       :		Any       =					PegasusTokenizer
	_UpperCAmelCase       :		Dict       =					["input_ids", "attention_mask"]



	def __init__(  self	:    Tuple			,  lowercase	:    str=None			,  lowercase	:    Any=None			,  lowercase	:    List[Any]="<pad>"			,  lowercase	:    List[Any]="</s>"			,  lowercase	:    Tuple="<unk>"			,  lowercase	:    Any="<mask_2>"			,  lowercase	:    List[str]="<mask_1>"			,  lowercase	:    List[Any]=None			,  lowercase	:    Dict=103			,  **lowercase	:    Optional[Any]			,  ):


				'''simple docstring'''


				_snake_case		=			offset

				if additional_special_tokens is not None:
							if not isinstance(lowercase			,  lowercase	):
										raise TypeError(
										    f'''additional_special_tokens should be of type {type(lowercase	)}, but is'''
										    f''' {type(lowercase	)}'''	)

							_snake_case		=			(
							    ([mask_token_sent] + additional_special_tokens)
							    if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
							    else additional_special_tokens
							)
							# fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
							additional_special_tokens_extended += [
							    f'''<unk_{i}>''' for i in range(len(lowercase	)			,  self.offset - 1	)
							]

							if len(set(lowercase	)	) != len(lowercase	):
										raise ValueError(
										    'Please make sure that the provided additional_special_tokens do not contain an incorrectly'
										    f''' shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}.'''	)
							_snake_case		=			additional_special_tokens_extended
				else:
							_snake_case		=			[mask_token_sent] if mask_token_sent is not None else []
							additional_special_tokens += [f'''<unk_{i}>''' for i in range(2			,  self.offset	)]

				super().__init__(
				    lowercase			,  tokenizer_file=lowercase			,  pad_token=lowercase			,  eos_token=lowercase			,  unk_token=lowercase			,  mask_token=lowercase			,  mask_token_sent=lowercase			,  offset=lowercase			,  additional_special_tokens=lowercase			,  **lowercase			,  )
				_snake_case		=			vocab_file
				_snake_case		=			False if not self.vocab_file else True



	def  A							(  self	:    List[str]			,  lowercase	:    Optional[int]	):


				'''simple docstring'''


				_snake_case		=			set(self.all_special_ids	)  # call it once instead of inside list comp
				all_special_ids.remove(self.unk_token_id	)  # <unk> is only sometimes special

				if all_special_ids != set(range(len(self.additional_special_tokens	) + 3	)	):
							raise ValueError(
							    'There should be 3 special tokens: mask_token, pad_token, and eos_token +'
							    f''' {len(self.additional_special_tokens	)} additional_special_tokens, but got {all_special_ids}'''	)

				return [1 if x in all_special_ids else 0 for x in seq]



	def  A							(  self	:    List[Any]			,  lowercase	:    List			,  lowercase	:    Optional[List] = None			,  lowercase	:    bool = False	):


				'''simple docstring'''


				if already_has_special_tokens:
							return self._special_token_mask(lowercase	)
				elif token_ids_a is None:
							return self._special_token_mask(lowercase	) + [1]
				else:
							return self._special_token_mask(token_ids_a + token_ids_a	) + [1]



	def  A							(  self	:    Any			,  lowercase	:    Tuple			,  lowercase	:    Any=None	):


				'''simple docstring'''


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





	def  A							(  self	:    int			,  lowercase	:    str			,  lowercase	:    Optional[str] = None	):


				'''simple docstring'''


				if not self.can_save_slow_tokenizer:
							raise ValueError(
							    'Your fast tokenizer does not have the necessary information to save the vocabulary for a slow '
							    'tokenizer.'	)

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

				if os.path.abspath(self.vocab_file	) != os.path.abspath(lowercase	):
							copyfile(self.vocab_file			,  lowercase	)

				return (out_vocab_file,)