





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[str] 							=						logging.get_logger(__name__)





class      lowercase__    (							__A  ):
  __UpperCamelCase	     =  ["""input_features""", """is_longer"""]







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







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       str    =		copy.deepcopy(self.__dict__						)
       lowerCAmelCase_					:       Optional[Any]    =		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 	UpperCAmelCase__		(	self    ,						_lowercase    ,						_lowercase = None						):
       lowerCAmelCase_					:       int    =		spectrogram(
           _lowercase    ,						window_function(self.fft_window_size    ,						"""hann"""						)    ,						frame_length=self.fft_window_size    ,						hop_length=self.hop_length    ,						power=2.0    ,						mel_filters=_lowercase    ,						log_mel="""dB"""    ,						)
       return log_mel_spectrogram.T







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

       lowerCAmelCase_					:       Union[str, Any]    =		mel[idx_front : idx_front + chunk_frames, :]
       lowerCAmelCase_					:       List[Any]    =		mel[idx_middle : idx_middle + chunk_frames, :]
       lowerCAmelCase_					:       Dict    =		mel[idx_back : idx_back + chunk_frames, :]

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







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

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

            if truncation == "fusion":
                 lowerCAmelCase_					:       int    =		self._np_extract_fbank_features(_lowercase    ,						self.mel_filters						)
                 lowerCAmelCase_					:       Tuple    =		np.stack([input_mel, input_mel, input_mel, input_mel]    ,						axis=0						)
            else:
                 lowerCAmelCase_					:       Optional[Any]    =		self._np_extract_fbank_features(_lowercase    ,						self.mel_filters_slaney						)[None, :]

       return input_mel, longer







  def __call__(	self    ,						_lowercase    ,						_lowercase = None    ,						_lowercase = None    ,						_lowercase = None    ,						_lowercase = None    ,						_lowercase = None    ,						**_lowercase    ,						):
       lowerCAmelCase_					:       int    =		truncation if truncation is not None else self.truncation
       lowerCAmelCase_					:       Dict    =		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."""						)

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

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

       # always return batch
       if not is_batched:
            lowerCAmelCase_					:       str    =		[np.asarray(_lowercase						)]

       # convert to mel spectrogram, truncate and pad if needed.
       lowerCAmelCase_					:       Union[str, Any]    =		[
           self._get_input_mel(_lowercase    ,						max_length if max_length else self.nb_max_samples    ,						_lowercase    ,						_lowercase						)
           for waveform in raw_speech
       ]

       lowerCAmelCase_					:       List[str]    =		[]
       lowerCAmelCase_					:       Optional[int]    =		[]
       for mel, longer in padded_inputs:
            input_mel.append(_lowercase						)
            is_longer.append(_lowercase						)

       if truncation == "fusion" and sum(_lowercase						) == 0:
            # if no audio is longer than 10s, then randomly select one audio to be longer
            lowerCAmelCase_					:       List[Any]    =		np.random.randint(0    ,						len(_lowercase						)						)
            lowerCAmelCase_					:       Optional[Any]    =		True

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

       # is_longer is a list of bool
       lowerCAmelCase_					:       Optional[int]    =		[[longer] for longer in is_longer]

       lowerCAmelCase_					:       List[str]    =		{"""input_features""": input_mel, """is_longer""": is_longer}
       lowerCAmelCase_					:       List[Any]    =		BatchFeature(_lowercase						)

       if return_tensors is not None:
            lowerCAmelCase_					:       int    =		input_features.convert_to_tensors(_lowercase						)

       return input_features







from collections import OrderedDict
from typing import Any, Mapping, Optional, Union

from ...configuration_utils import PretrainedConfig
from ...feature_extraction_utils import FeatureExtractionMixin
from ...onnx import OnnxConfig
from ...onnx.utils import compute_effective_axis_dimension
from ...tokenization_utils_base import PreTrainedTokenizerBase
from ...utils import TensorType, logging


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

UpperCAmelCase_					:   Any 							=						{
    """deepmind/language-perceiver""": """https://huggingface.co/deepmind/language-perceiver/resolve/main/config.json""",
    # See all Perceiver models at https://huggingface.co/models?filter=perceiver
}





class      lowercase__    (							__A  ):
  __UpperCamelCase	     =  """perceiver"""







  def __init__(	self    ,						_lowercase=256    ,						_lowercase=1_280    ,						_lowercase=768    ,						_lowercase=1    ,						_lowercase=26    ,						_lowercase=8    ,						_lowercase=8    ,						_lowercase=None    ,						_lowercase=None    ,						_lowercase="kv"    ,						_lowercase=1    ,						_lowercase=1    ,						_lowercase="gelu"    ,						_lowercase=0.1    ,						_lowercase=0.02    ,						_lowercase=1e-12    ,						_lowercase=True    ,						_lowercase=262    ,						_lowercase=2_048    ,						_lowercase=56    ,						_lowercase=[368, 496]    ,						_lowercase=16    ,						_lowercase=1_920    ,						_lowercase=16    ,						_lowercase=[1, 16, 224, 224]    ,						**_lowercase    ,						):
       super().__init__(**_lowercase						)

       lowerCAmelCase_					:       Optional[int]    =		num_latents
       lowerCAmelCase_					:       List[str]    =		d_latents
       lowerCAmelCase_					:       int    =		d_model
       lowerCAmelCase_					:       Dict    =		num_blocks
       lowerCAmelCase_					:       Union[str, Any]    =		num_self_attends_per_block
       lowerCAmelCase_					:       List[str]    =		num_self_attention_heads
       lowerCAmelCase_					:       List[str]    =		num_cross_attention_heads
       lowerCAmelCase_					:       List[Any]    =		qk_channels
       lowerCAmelCase_					:       Optional[Any]    =		v_channels
       lowerCAmelCase_					:       Optional[Any]    =		cross_attention_shape_for_attention
       lowerCAmelCase_					:       Optional[int]    =		self_attention_widening_factor
       lowerCAmelCase_					:       int    =		cross_attention_widening_factor
       lowerCAmelCase_					:       Union[str, Any]    =		hidden_act
       lowerCAmelCase_					:       int    =		attention_probs_dropout_prob
       lowerCAmelCase_					:       int    =		initializer_range
       lowerCAmelCase_					:       Optional[Any]    =		layer_norm_eps
       lowerCAmelCase_					:       Optional[Any]    =		use_query_residual
       # masked language modeling attributes
       lowerCAmelCase_					:       Tuple    =		vocab_size
       lowerCAmelCase_					:       List[Any]    =		max_position_embeddings
       # image classification attributes
       lowerCAmelCase_					:       List[Any]    =		image_size
       # flow attributes
       lowerCAmelCase_					:       Dict    =		train_size
       # multimodal autoencoding attributes
       lowerCAmelCase_					:       Optional[Any]    =		num_frames
       lowerCAmelCase_					:       int    =		audio_samples_per_frame
       lowerCAmelCase_					:       Any    =		samples_per_patch
       lowerCAmelCase_					:       Any    =		output_shape





class      lowercase__    (							__A  ):







  @property
  def 	UpperCAmelCase__		(	self						):
       if self.task == "multiple-choice":
            lowerCAmelCase_					:       Optional[Any]    =		{0: """batch""", 1: """choice""", 2: """sequence"""}
       else:
            lowerCAmelCase_					:       Optional[int]    =		{0: """batch""", 1: """sequence"""}
       return OrderedDict(
           [
               ("""inputs""", dynamic_axis),
               ("""attention_mask""", dynamic_axis),
           ]						)







  @property
  def 	UpperCAmelCase__		(	self						):
       return 1e-4







  def 	UpperCAmelCase__		(	self    ,						_lowercase    ,						_lowercase = -1    ,						_lowercase = -1    ,						_lowercase = -1    ,						_lowercase = False    ,						_lowercase = None    ,						_lowercase = 3    ,						_lowercase = 40    ,						_lowercase = 40    ,						):
       # copied from `transformers.onnx.config.OnnxConfig` and slightly altered/simplified

       if isinstance(_lowercase    ,						_lowercase						):
            # If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
            lowerCAmelCase_					:       Union[str, Any]    =		compute_effective_axis_dimension(
                _lowercase    ,						fixed_dimension=OnnxConfig.default_fixed_batch    ,						num_token_to_add=0						)
            # If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
            lowerCAmelCase_					:       int    =		preprocessor.num_special_tokens_to_add(_lowercase						)
            lowerCAmelCase_					:       int    =		compute_effective_axis_dimension(
                _lowercase    ,						fixed_dimension=OnnxConfig.default_fixed_sequence    ,						num_token_to_add=_lowercase						)
            # Generate dummy inputs according to compute batch and sequence
            lowerCAmelCase_					:       Optional[int]    =		[""" """.join(["""a"""]						) * seq_length] * batch_size
            lowerCAmelCase_					:       Optional[Any]    =		dict(preprocessor(_lowercase    ,						return_tensors=_lowercase						)						)
            lowerCAmelCase_					:       Optional[Any]    =		inputs.pop("""input_ids"""						)
            return inputs
       elif isinstance(_lowercase    ,						_lowercase						) and preprocessor.model_input_names[0] == "pixel_values":
            # If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
            lowerCAmelCase_					:       Any    =		compute_effective_axis_dimension(_lowercase    ,						fixed_dimension=OnnxConfig.default_fixed_batch						)
            lowerCAmelCase_					:       str    =		self._generate_dummy_images(_lowercase    ,						_lowercase    ,						_lowercase    ,						_lowercase						)
            lowerCAmelCase_					:       List[str]    =		dict(preprocessor(images=_lowercase    ,						return_tensors=_lowercase						)						)
            lowerCAmelCase_					:       Tuple    =		inputs.pop("""pixel_values"""						)
            return inputs
       else:
            raise ValueError(
                """Unable to generate dummy inputs for the model. Please provide a tokenizer or a preprocessor."""						)
