






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


lowercase  :							Optional[Any]			  =    logging.get_logger(__name__)




class 		UpperCAmelCase_						(	SCREAMING_SNAKE_CASE__					):


						'''simple docstring'''




						A     :   List[str]						=					['input_features', 'is_longer']






						def __init__(	self		,				_SCREAMING_SNAKE_CASE=64		,				_SCREAMING_SNAKE_CASE=4_8000		,				_SCREAMING_SNAKE_CASE=480		,				_SCREAMING_SNAKE_CASE=10		,				_SCREAMING_SNAKE_CASE=1024		,				_SCREAMING_SNAKE_CASE=0.0		,				_SCREAMING_SNAKE_CASE=False		,				_SCREAMING_SNAKE_CASE = 0		,				_SCREAMING_SNAKE_CASE = 1_4000		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = "fusion"		,				_SCREAMING_SNAKE_CASE = "repeatpad"		,				**_SCREAMING_SNAKE_CASE		,				) ->				Tuple:
								super().__init__(
								    feature_size=_SCREAMING_SNAKE_CASE		,				sampling_rate=_SCREAMING_SNAKE_CASE		,				padding_value=_SCREAMING_SNAKE_CASE		,				return_attention_mask=_SCREAMING_SNAKE_CASE		,				**_SCREAMING_SNAKE_CASE		,				)
								snake_case_				:							List[str]						 =       top_db
								snake_case_				:							List[str]						 =       truncation
								snake_case_				:							Optional[int]						 =       padding
								snake_case_				:							List[Any]						 =       fft_window_size
								snake_case_				:							Dict						 =       (fft_window_size >> 1) + 1
								snake_case_				:							Union[str, Any]						 =       hop_length
								snake_case_				:							int						 =       max_length_s
								snake_case_				:							Any						 =       max_length_s * sampling_rate
								snake_case_				:							int						 =       sampling_rate
								snake_case_				:							List[Any]						 =       frequency_min
								snake_case_				:							str						 =       frequency_max
								snake_case_				:							List[Any]						 =       mel_filter_bank(
								    num_frequency_bins=self.nb_frequency_bins		,				num_mel_filters=_SCREAMING_SNAKE_CASE		,				min_frequency=_SCREAMING_SNAKE_CASE		,				max_frequency=_SCREAMING_SNAKE_CASE		,				sampling_rate=_SCREAMING_SNAKE_CASE		,				norm=_SCREAMING_SNAKE_CASE		,				mel_scale="htk"		,				)
								snake_case_				:							List[str]						 =       mel_filter_bank(
								    num_frequency_bins=self.nb_frequency_bins		,				num_mel_filters=_SCREAMING_SNAKE_CASE		,				min_frequency=_SCREAMING_SNAKE_CASE		,				max_frequency=_SCREAMING_SNAKE_CASE		,				sampling_rate=_SCREAMING_SNAKE_CASE		,				norm="slaney"		,				mel_scale="slaney"		,				)






						def       _lowerCAmelCase       (	self				) ->				Dict[str, Any]:
								snake_case_				:							str						 =       copy.deepcopy(self.__dict__				)
								snake_case_				:							List[str]						 =       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		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE = None				) ->				np.ndarray:
								snake_case_				:							str						 =       spectrogram(
								    _SCREAMING_SNAKE_CASE		,				window_function(self.fft_window_size		,				"hann"				)		,				frame_length=self.fft_window_size		,				hop_length=self.hop_length		,				power=2.0		,				mel_filters=_SCREAMING_SNAKE_CASE		,				log_mel="dB"		,				)
								return log_mel_spectrogram.T






						def       _lowerCAmelCase       (	self		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE				) ->				Optional[Any]:
								snake_case_				:							Dict						 =       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
										snake_case_				:							int						 =       [0]
								if len(ranges[2]				) == 0:
										# if the audio is too short, we just use the first chunk
										snake_case_				:							Optional[int]						 =       [0]
								# randomly choose index for each part
								snake_case_				:							str						 =       np.random.choice(ranges[0]				)
								snake_case_				:							Tuple						 =       np.random.choice(ranges[1]				)
								snake_case_				:							List[Any]						 =       np.random.choice(ranges[2]				)

								snake_case_				:							Optional[int]						 =       mel[idx_front : idx_front + chunk_frames, :]
								snake_case_				:							str						 =       mel[idx_middle : idx_middle + chunk_frames, :]
								snake_case_				:							int						 =       mel[idx_back : idx_back + chunk_frames, :]

								snake_case_				:							Tuple						 =       torch.tensor(mel[None, None, :]				)
								snake_case_				:							Optional[Any]						 =       torch.nn.functional.interpolate(
								    _SCREAMING_SNAKE_CASE		,				size=[chunk_frames, 64]		,				mode="bilinear"		,				align_corners=_SCREAMING_SNAKE_CASE				)
								snake_case_				:							Any						 =       mel_shrink[0][0].numpy()
								snake_case_				:							List[Any]						 =       np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]		,				axis=0				)
								return mel_fusion






						def       _lowerCAmelCase       (	self		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE				) ->				np.array:
								if waveform.shape[0] > max_length:
										if truncation == "rand_trunc":
												snake_case_				:							Any						 =       True
												# random crop to max_length (for compatibility) -> this should be handled by self.pad
												snake_case_				:							Dict						 =       len(_SCREAMING_SNAKE_CASE				) - max_length
												snake_case_				:							List[str]						 =       np.random.randint(0		,				overflow + 1				)
												snake_case_				:							str						 =       waveform[idx : idx + max_length]
												snake_case_				:							Any						 =       self._np_extract_fbank_features(_SCREAMING_SNAKE_CASE		,				self.mel_filters_slaney				)[None, :]
										elif truncation == "fusion":
												snake_case_				:							Union[str, Any]						 =       self._np_extract_fbank_features(_SCREAMING_SNAKE_CASE		,				self.mel_filters				)
												snake_case_				:							List[str]						 =       max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
												snake_case_				:							Any						 =       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.
														snake_case_				:							Union[str, Any]						 =       np.stack([mel, mel, mel, mel]		,				axis=0				)
														snake_case_				:							Optional[int]						 =       False
												else:
														snake_case_				:							Union[str, Any]						 =       self._random_mel_fusion(_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE				)
														snake_case_				:							List[Any]						 =       True
										else:
												raise NotImplementedError(f'''data_truncating {truncation} not implemented'''				)

								else:
										snake_case_				:							str						 =       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":
														snake_case_				:							str						 =       int(max_length / len(_SCREAMING_SNAKE_CASE				)				)
														snake_case_				:							Tuple						 =       np.stack(np.tile(_SCREAMING_SNAKE_CASE		,				n_repeat + 1				)				)[:max_length]
												if padding == "repeatpad":
														snake_case_				:							List[str]						 =       int(max_length / len(_SCREAMING_SNAKE_CASE				)				)
														snake_case_				:							List[str]						 =       np.stack(np.tile(_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE				)				)
												snake_case_				:							int						 =       np.pad(_SCREAMING_SNAKE_CASE		,				(0, max_length - waveform.shape[0])		,				mode="constant"		,				constant_values=0				)

										if truncation == "fusion":
												snake_case_				:							int						 =       self._np_extract_fbank_features(_SCREAMING_SNAKE_CASE		,				self.mel_filters				)
												snake_case_				:							List[str]						 =       np.stack([input_mel, input_mel, input_mel, input_mel]		,				axis=0				)
										else:
												snake_case_				:							Dict						 =       self._np_extract_fbank_features(_SCREAMING_SNAKE_CASE		,				self.mel_filters_slaney				)[None, :]

								return input_mel, longer






						def __call__(	self		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = None		,				**_SCREAMING_SNAKE_CASE		,				) ->				BatchFeature:
								snake_case_				:							List[str]						 =       truncation if truncation is not None else self.truncation
								snake_case_				:							Union[str, Any]						 =       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."				)

								snake_case_				:							List[str]						 =       isinstance(_SCREAMING_SNAKE_CASE		,				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_				:							List[str]						 =       is_batched_numpy or (
								    isinstance(_SCREAMING_SNAKE_CASE		,				(list, tuple)				) and (isinstance(raw_speech[0]		,				(np.ndarray, tuple, list)				))
								)

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

								# always return batch
								if not is_batched:
										snake_case_				:							Tuple						 =       [np.asarray(_SCREAMING_SNAKE_CASE				)]

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

								snake_case_				:							Union[str, Any]						 =       []
								snake_case_				:							List[Any]						 =       []
								for mel, longer in padded_inputs:
										input_mel.append(_SCREAMING_SNAKE_CASE				)
										is_longer.append(_SCREAMING_SNAKE_CASE				)

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

								if isinstance(input_mel[0]		,				_SCREAMING_SNAKE_CASE				):
										snake_case_				:							Optional[Any]						 =       [np.asarray(_SCREAMING_SNAKE_CASE		,				dtype=np.floataa				) for feature in input_mel]

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

								snake_case_				:							Optional[Any]						 =       {"input_features": input_mel, "is_longer": is_longer}
								snake_case_				:							Tuple						 =       BatchFeature(_SCREAMING_SNAKE_CASE				)

								if return_tensors is not None:
										snake_case_				:							Tuple						 =       input_features.convert_to_tensors(_SCREAMING_SNAKE_CASE				)

								return input_features








from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowercase  :							Optional[int]			  =    {
    '''configuration_nllb_moe''': [
        '''NLLB_MOE_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''NllbMoeConfig''',
    ]
}

try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							lowercase  :							Optional[Any]			  =    [
							    '''NLLB_MOE_PRETRAINED_MODEL_ARCHIVE_LIST''',
							    '''NllbMoeForConditionalGeneration''',
							    '''NllbMoeModel''',
							    '''NllbMoePreTrainedModel''',
							    '''NllbMoeTop2Router''',
							    '''NllbMoeSparseMLP''',
							]


if TYPE_CHECKING:
							from .configuration_nllb_moe import (
							    NLLB_MOE_PRETRAINED_CONFIG_ARCHIVE_MAP,
							    NllbMoeConfig,
							)

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_nllb_moe import (
														    NLLB_MOE_PRETRAINED_MODEL_ARCHIVE_LIST,
														    NllbMoeForConditionalGeneration,
														    NllbMoeModel,
														    NllbMoePreTrainedModel,
														    NllbMoeSparseMLP,
														    NllbMoeTopaRouter,
														)


else:
							import sys

							lowercase  :							Tuple			  =    _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)
