




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

import numpy as np
import torch

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


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

class 						__lowerCAmelCase      (      __magic_name__					):
							UpperCamelCase__   	=    ['''input_features''', '''is_longer''']
							def __init__(		self		:str						, __magic_name__		:Union[str, Any]=64						, __magic_name__		:Optional[int]=4_8000						, __magic_name__		:str=480						, __magic_name__		:int=10						, __magic_name__		:List[Any]=1024						, __magic_name__		:List[Any]=0.0						, __magic_name__		:Tuple=False						, __magic_name__		:float = 0						, __magic_name__		:float = 1_4000						, __magic_name__		:int = None						, __magic_name__		:str = "fusion"						, __magic_name__		:str = "repeatpad"						, **__magic_name__		:List[Any]						, ):







								'''simple docstring'''







								super().__init__(
								    feature_size=__magic_name__						, sampling_rate=__magic_name__						, padding_value=__magic_name__						, return_attention_mask=__magic_name__						, **__magic_name__						, )
								a									=				top_db
								a									=				truncation
								a									=				padding
								a									=				fft_window_size
								a									=				(fft_window_size >> 1) + 1
								a									=				hop_length
								a									=				max_length_s
								a									=				max_length_s * sampling_rate
								a									=				sampling_rate
								a									=				frequency_min
								a									=				frequency_max
								a									=				mel_filter_bank(
								    num_frequency_bins=self.nb_frequency_bins						, num_mel_filters=__magic_name__						, min_frequency=__magic_name__						, max_frequency=__magic_name__						, sampling_rate=__magic_name__						, norm=__magic_name__						, mel_scale="""htk"""						, )
								a									=				mel_filter_bank(
								    num_frequency_bins=self.nb_frequency_bins						, num_mel_filters=__magic_name__						, min_frequency=__magic_name__						, max_frequency=__magic_name__						, sampling_rate=__magic_name__						, norm="""slaney"""						, mel_scale="""slaney"""						, )
							def  lowerCamelCase__				(		self		:Any							):







								'''simple docstring'''







								a									=				copy.deepcopy(self.__dict__							)
								a									=				self.__class__.__name__
								if "mel_filters" in output:
									del output["mel_filters"]
								if "mel_filters_slaney" in output:
									del output["mel_filters_slaney"]
								return output
							def  lowerCamelCase__				(		self		:Dict						, __magic_name__		:np.array						, __magic_name__		:Optional[np.array] = None							):







								'''simple docstring'''







								a									=				spectrogram(
								    __magic_name__						, window_function(self.fft_window_size						, """hann"""							)						, frame_length=self.fft_window_size						, hop_length=self.hop_length						, power=2.0						, mel_filters=__magic_name__						, log_mel="""dB"""						, )
								return log_mel_spectrogram.T
							def  lowerCamelCase__				(		self		:Optional[Any]						, __magic_name__		:Dict						, __magic_name__		:int						, __magic_name__		:Union[str, Any]							):







								'''simple docstring'''







								a									=				np.array_split(list(range(0						, total_frames - chunk_frames + 1							)							)						, 3							)
								if len(ranges[1]							) == 0:
									# if the audio is too short, we just use the first chunk
									a									=				[0]
								if len(ranges[2]							) == 0:
									# if the audio is too short, we just use the first chunk
									a									=				[0]
								# randomly choose index for each part
								a									=				np.random.choice(ranges[0]							)
								a									=				np.random.choice(ranges[1]							)
								a									=				np.random.choice(ranges[2]							)

								a									=				mel[idx_front : idx_front + chunk_frames, :]
								a									=				mel[idx_middle : idx_middle + chunk_frames, :]
								a									=				mel[idx_back : idx_back + chunk_frames, :]

								a									=				torch.tensor(mel[None, None, :]							)
								a									=				torch.nn.functional.interpolate(
								    __magic_name__						, size=[chunk_frames, 64]						, mode="""bilinear"""						, align_corners=__magic_name__							)
								a									=				mel_shrink[0][0].numpy()
								a									=				np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]						, axis=0							)
								return mel_fusion
							def  lowerCamelCase__				(		self		:Any						, __magic_name__		:np.array						, __magic_name__		:Union[str, Any]						, __magic_name__		:int						, __magic_name__		:List[Any]							):







								'''simple docstring'''







								if waveform.shape[0] > max_length:
									if truncation == "rand_trunc":
										a									=				True
										# random crop to max_length (for compatibility) -> this should be handled by self.pad
										a									=				len(__magic_name__							) - max_length
										a									=				np.random.randint(0						, overflow + 1							)
										a									=				waveform[idx : idx + max_length]
										a									=				self._np_extract_fbank_features(__magic_name__						, self.mel_filters_slaney							)[None, :]
									elif truncation == "fusion":
										a									=				self._np_extract_fbank_features(__magic_name__						, self.mel_filters							)
										a									=				max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
										a									=				mel.shape[0]
										if chunk_frames == total_frames:
											# there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
											# In this case, we just use the whole audio.
											a									=				np.stack([mel, mel, mel, mel]						, axis=0							)
											a									=				False
										else:
											a									=				self._random_mel_fusion(__magic_name__						, __magic_name__						, __magic_name__							)
											a									=				True
									else:
										raise NotImplementedError(F'data_truncating {truncation} not implemented'							)

								else:
									a									=				False
									# only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
									if waveform.shape[0] < max_length:
										if padding == "repeat":
											a									=				int(max_length / len(__magic_name__							)							)
											a									=				np.stack(np.tile(__magic_name__						, n_repeat + 1							)							)[:max_length]
										if padding == "repeatpad":
											a									=				int(max_length / len(__magic_name__							)							)
											a									=				np.stack(np.tile(__magic_name__						, __magic_name__							)							)
										a									=				np.pad(__magic_name__						, (0, max_length - waveform.shape[0])						, mode="""constant"""						, constant_values=0							)

									if truncation == "fusion":
										a									=				self._np_extract_fbank_features(__magic_name__						, self.mel_filters							)
										a									=				np.stack([input_mel, input_mel, input_mel, input_mel]						, axis=0							)
									else:
										a									=				self._np_extract_fbank_features(__magic_name__						, self.mel_filters_slaney							)[None, :]

								return input_mel, longer





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







								'''simple docstring'''







								a									=				truncation if truncation is not None else self.truncation
								a									=				padding if padding else self.padding

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

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

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

								# always return batch
								if not is_batched:
									a									=				[np.asarray(__magic_name__							)]

								# convert to mel spectrogram, truncate and pad if needed.
								a									=				[
								    self._get_input_mel(__magic_name__						, max_length if max_length else self.nb_max_samples						, __magic_name__						, __magic_name__							)
								    for waveform in raw_speech
								]

								a									=				[]
								a									=				[]
								for mel, longer in padded_inputs:
									input_mel.append(__magic_name__							)
									is_longer.append(__magic_name__							)

								if truncation == "fusion" and sum(__magic_name__							) == 0:
									# if no audio is longer than 10s, then randomly select one audio to be longer
									a									=				np.random.randint(0						, len(__magic_name__							)							)
									a									=				True

								if isinstance(input_mel[0]						, __magic_name__							):
									a									=				[np.asarray(__magic_name__						, dtype=np.floataa							) for feature in input_mel]

								# is_longer is a list of bool
								a									=				[[longer] for longer in is_longer]

								a									=				{"""input_features""": input_mel, """is_longer""": is_longer}
								a									=				BatchFeature(__magic_name__							)

								if return_tensors is not None:
									a									=				input_features.convert_to_tensors(__magic_name__							)

								return input_features







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

import sentencepiece as spm

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


__UpperCamelCase   :			Union[str, Any]          = logging.get_logger(__name__)

__UpperCamelCase   :			Optional[Any]          = "▁"

__UpperCamelCase   :			Any          = {"vocab_file": "sentencepiece.bpe.model"}

__UpperCamelCase   :			int          = {
    "vocab_file": {
        "facebook/xglm-564M": "https://huggingface.co/facebook/xglm-564M/resolve/main/sentencepiece.bpe.model",
    }
}

__UpperCamelCase   :			Dict          = {
    "facebook/xglm-564M": 2_048,
}

class 						__lowerCAmelCase      (      __magic_name__					):
							UpperCamelCase__   	=    VOCAB_FILES_NAMES
							UpperCamelCase__   	=    PRETRAINED_VOCAB_FILES_MAP
							UpperCamelCase__   	=    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							UpperCamelCase__   	=    ['''input_ids''', '''attention_mask''']
							def __init__(		self		:Optional[Any]						, __magic_name__		:List[str]						, __magic_name__		:Tuple="<s>"						, __magic_name__		:List[Any]="</s>"						, __magic_name__		:List[str]="</s>"						, __magic_name__		:Any="<s>"						, __magic_name__		:Tuple="<unk>"						, __magic_name__		:int="<pad>"						, __magic_name__		:Optional[Dict[str, Any]] = None						, **__magic_name__		:Optional[Any]						, ):







								'''simple docstring'''







								a									=				{} if sp_model_kwargs is None else sp_model_kwargs

								# Compatibility with the original tokenizer
								a									=				7
								a									=				[F'<madeupword{i}>' for i in range(self.num_madeup_words							)]

								a									=				kwargs.get("""additional_special_tokens"""						, []							)
								kwargs["additional_special_tokens"] += [
								    word for word in madeup_words if word not in kwargs["additional_special_tokens"]
								]

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

								a									=				spm.SentencePieceProcessor(**self.sp_model_kwargs							)
								self.sp_model.Load(str(__magic_name__							)							)
								a									=				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'

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

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

								a									=				len(self.sp_model							)
								a									=				{F'<madeupword{i}>': sp_size + i + self.fairseq_offset for i in range(self.num_madeup_words							)}
								self.fairseq_tokens_to_ids.update(__magic_name__							)

								a									=				{v: k for k, v in self.fairseq_tokens_to_ids.items()}
							def __getstate__(		self		:List[Any]							):







								'''simple docstring'''







								a									=				self.__dict__.copy()
								a									=				None
								a									=				self.sp_model.serialized_model_proto()
								return state
							def __setstate__(		self		:Any						, __magic_name__		:Any							):







								'''simple docstring'''







								a									=				d

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

								a									=				spm.SentencePieceProcessor(**self.sp_model_kwargs							)
								self.sp_model.LoadFromSerializedProto(self.sp_model_proto							)
							def  lowerCamelCase__				(		self		:Any						, __magic_name__		:List[int]						, __magic_name__		:Optional[List[int]] = None							):







								'''simple docstring'''







								if token_ids_a is None:
									return [self.sep_token_id] + token_ids_a
								a									=				[self.sep_token_id]
								return sep + token_ids_a + sep + sep + token_ids_a
							def  lowerCamelCase__				(		self		:Any						, __magic_name__		:List[int]						, __magic_name__		:Optional[List[int]] = None						, __magic_name__		:bool = False							):







								'''simple docstring'''







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

								if token_ids_a is None:
									return [1] + ([0] * len(__magic_name__							))
								return [1] + ([0] * len(__magic_name__							)) + [1, 1] + ([0] * len(__magic_name__							))
							def  lowerCamelCase__				(		self		:Optional[Any]						, __magic_name__		:List[int]						, __magic_name__		:Optional[List[int]] = None							):







								'''simple docstring'''







								a									=				[self.sep_token_id]

								if token_ids_a is None:
									return len(sep + token_ids_a							) * [0]
								return len(sep + token_ids_a + sep + sep + token_ids_a							) * [0]
							@property
							def  lowerCamelCase__				(		self		:str							):







								'''simple docstring'''







								return len(self.sp_model							) + self.fairseq_offset + self.num_madeup_words
							def  lowerCamelCase__				(		self		:Dict							):







								'''simple docstring'''







								a									=				{self.convert_ids_to_tokens(__magic_name__							): i for i in range(self.vocab_size							)}
								vocab.update(self.added_tokens_encoder							)
								return vocab
							def  lowerCamelCase__				(		self		:Any						, __magic_name__		:str							):







								'''simple docstring'''







								return self.sp_model.encode(__magic_name__						, out_type=__magic_name__							)
							def  lowerCamelCase__				(		self		:List[str]						, __magic_name__		:Any							):







								'''simple docstring'''







								if token in self.fairseq_tokens_to_ids:
									return self.fairseq_tokens_to_ids[token]
								a									=				self.sp_model.PieceToId(__magic_name__							)

								# 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  lowerCamelCase__				(		self		:List[str]						, __magic_name__		:Dict							):







								'''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  lowerCamelCase__				(		self		:List[str]						, __magic_name__		:List[str]							):







								'''simple docstring'''







								a									=				"""""".join(__magic_name__							).replace(__magic_name__						, """ """							).strip()
								return out_string





							def  lowerCamelCase__				(		self		:List[Any]						, __magic_name__		:str						, __magic_name__		:Optional[str] = None							):







								'''simple docstring'''







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

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

								return (out_vocab_file,)

