

'''simple docstring'''
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  :		int  =		logging.get_logger(__name__)

class 						__magic_name__   (							_UpperCamelCase ):
 UpperCamelCase__     =       ["""input_features""", """is_longer"""]







 def __init__(       self  , snake_case_=64  , snake_case_=4_80_00  , snake_case_=4_80  , snake_case_=10  , snake_case_=10_24  , snake_case_=0.0  , snake_case_=False  , snake_case_ = 0  , snake_case_ = 1_40_00  , snake_case_ = None  , snake_case_ = "fusion"  , snake_case_ = "repeatpad"  , **snake_case_  , ):
   super().__init__(
       feature_size=snake_case_  , sampling_rate=snake_case_  , padding_value=snake_case_  , return_attention_mask=snake_case_  , **snake_case_  , )
   lowercase									=top_db
   lowercase									=truncation
   lowercase									=padding
   lowercase									=fft_window_size
   lowercase									=(fft_window_size >> 1) + 1
   lowercase									=hop_length
   lowercase									=max_length_s
   lowercase									=max_length_s * sampling_rate
   lowercase									=sampling_rate
   lowercase									=frequency_min
   lowercase									=frequency_max
   lowercase									=mel_filter_bank(
       num_frequency_bins=self.nb_frequency_bins  , num_mel_filters=snake_case_  , min_frequency=snake_case_  , max_frequency=snake_case_  , sampling_rate=snake_case_  , norm=snake_case_  , mel_scale='''htk'''  , )
   lowercase									=mel_filter_bank(
       num_frequency_bins=self.nb_frequency_bins  , num_mel_filters=snake_case_  , min_frequency=snake_case_  , max_frequency=snake_case_  , sampling_rate=snake_case_  , norm='''slaney'''  , mel_scale='''slaney'''  , )







 def 						_A(       self   ):
   lowercase									=copy.deepcopy(self.__dict__   )
   lowercase									=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 						_A(       self  , snake_case_  , snake_case_ = None   ):
   lowercase									=spectrogram(
       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=snake_case_  , log_mel='''dB'''  , )
   return log_mel_spectrogram.T







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

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

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







 def 						_A(       self  , snake_case_  , snake_case_  , snake_case_  , snake_case_   ):





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

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

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

   return input_mel, longer







 def __call__(       self  , snake_case_  , snake_case_ = None  , snake_case_ = None  , snake_case_ = None  , snake_case_ = None  , snake_case_ = None  , **snake_case_  , ):
   lowercase									=truncation if truncation is not None else self.truncation
   lowercase									=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.'''   )

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

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

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

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

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

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

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

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

   lowercase									={"input_features": input_mel, "is_longer": is_longer}
   lowercase									=BatchFeature(snake_case_   )

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

   return input_features







import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


SCREAMING_SNAKE_CASE__							:      Tuple         =    logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__							:      int         =    {
    """google/pix2struct-textcaps-base""": (
        """https://huggingface.co/google/pix2struct-textcaps-base/resolve/main/config.json"""
    ),
}







class 	__lowerCAmelCase      (       _UpperCamelCase	):
   _UpperCamelCase			:    Optional[int]  		=					"""pix2struct_text_model"""
   _UpperCamelCase			:    Dict  		=					["""past_key_values"""]
   _UpperCamelCase			:    Union[str, Any]  		=					{
       """hidden_size""": """hidden_size""",
       """num_attention_heads""": """num_heads""",
       """num_hidden_layers""": """num_layers""",
   }





   def __init__(			self       ,      snake_case=50_244       ,      snake_case=768       ,      snake_case=64       ,      snake_case=2_048       ,      snake_case=12       ,      snake_case=12       ,      snake_case=32       ,      snake_case=128       ,      snake_case=0.1       ,      snake_case=1E-6       ,      snake_case=1.0       ,      snake_case="gelu_new"       ,      snake_case=0       ,      snake_case=False       ,      snake_case=0       ,      snake_case=1       ,      snake_case=False       ,      snake_case=True       ,      **snake_case       ,      )					->	Dict:
      """simple docstring"""





      a__						:    Tuple   		=	vocab_size
      a__						:    List[str]   		=	hidden_size
      a__						:    Optional[Any]   		=	d_kv
      a__						:    List[str]   		=	d_ff
      a__						:    Optional[int]   		=	num_layers
      a__						:    Tuple   		=	num_heads
      a__						:    List[Any]   		=	relative_attention_num_buckets
      a__						:    Optional[int]   		=	relative_attention_max_distance
      a__						:    Optional[int]   		=	dropout_rate
      a__						:    List[Any]   		=	layer_norm_epsilon
      a__						:    List[str]   		=	initializer_factor
      a__						:    Any   		=	use_cache

      a__						:    Any   		=	eos_token_id
      a__						:    List[Any]   		=	decoder_start_token_id

      # for backwards compatibility
      a__						:    Tuple   		=	dense_act_fn

      super().__init__(
          pad_token_id=snake_case       ,      eos_token_id=snake_case       ,      decoder_start_token_id=snake_case       ,      tie_word_embeddings=snake_case       ,      is_decoder=snake_case       ,      **snake_case       ,      )





   @classmethod
   def 						_snake_case			(			cls       ,      snake_case       ,      **snake_case   )					->	"PretrainedConfig":
      """simple docstring"""





      cls._set_token_in_kwargs(snake_case   )

      a__		,  a__						:    Optional[Any]   		=	cls.get_config_dict(snake_case       ,      **snake_case   )

      # get the text config dict if we are loading from Pix2StructConfig
      if config_dict.get("model_type"   ) == "pix2struct":
         a__						:    Optional[int]   		=	config_dict["text_config"]

      if "model_type" in config_dict and hasattr(cls       ,      "model_type"   ) and config_dict["model_type"] != cls.model_type:
         logger.warning(
             F"""You are using a model of type {config_dict["model_type"]} to instantiate a model of type """
             F"""{cls.model_type}. This is not supported for all configurations of models and can yield errors."""   )

      return cls.from_dict(snake_case       ,      **snake_case   )







class 	__lowerCAmelCase      (       _UpperCamelCase	):
   _UpperCamelCase			:    int  		=					"""pix2struct_vision_model"""





   def __init__(			self       ,      snake_case=768       ,      snake_case=768       ,      snake_case=2_048       ,      snake_case=64       ,      snake_case=12       ,      snake_case=12       ,      snake_case="gelu_new"       ,      snake_case=1E-6       ,      snake_case=0.0       ,      snake_case=0.0       ,      snake_case=1E-10       ,      snake_case=1.0       ,      snake_case=4_096       ,      snake_case=32       ,      snake_case=128       ,      **snake_case       ,      )					->	Optional[Any]:
      """simple docstring"""





      super().__init__(**snake_case   )

      a__						:    Optional[Any]   		=	hidden_size
      a__						:    Tuple   		=	patch_embed_hidden_size
      a__						:    Tuple   		=	d_ff
      a__						:    List[Any]   		=	dropout_rate
      a__						:    Optional[Any]   		=	num_hidden_layers
      a__						:    Optional[Any]   		=	num_attention_heads
      a__						:    List[str]   		=	initializer_range
      a__						:    List[str]   		=	initializer_factor
      a__						:    List[Any]   		=	attention_dropout
      a__						:    Optional[int]   		=	layer_norm_eps
      a__						:    int   		=	dense_act_fn
      a__						:    Optional[Any]   		=	seq_len
      a__						:    List[Any]   		=	relative_attention_num_buckets
      a__						:    int   		=	relative_attention_max_distance
      a__						:    int   		=	d_kv





   @classmethod
   def 						_snake_case			(			cls       ,      snake_case       ,      **snake_case   )					->	"PretrainedConfig":
      """simple docstring"""





      cls._set_token_in_kwargs(snake_case   )

      a__		,  a__						:    int   		=	cls.get_config_dict(snake_case       ,      **snake_case   )

      # get the vision config dict if we are loading from Pix2StructConfig
      if config_dict.get("model_type"   ) == "pix2struct":
         a__						:    int   		=	config_dict["vision_config"]

      if "model_type" in config_dict and hasattr(cls       ,      "model_type"   ) and config_dict["model_type"] != cls.model_type:
         logger.warning(
             F"""You are using a model of type {config_dict["model_type"]} to instantiate a model of type """
             F"""{cls.model_type}. This is not supported for all configurations of models and can yield errors."""   )

      return cls.from_dict(snake_case       ,      **snake_case   )







class 	__lowerCAmelCase      (       _UpperCamelCase	):
   _UpperCamelCase			:    List[Any]  		=					"""pix2struct"""
   _UpperCamelCase			:    Dict  		=					True





   def __init__(			self       ,      snake_case=None       ,      snake_case=None       ,      snake_case=1.0       ,      snake_case=0.02       ,      snake_case=False       ,      snake_case=False       ,      snake_case=True       ,      **snake_case       ,      )					->	Optional[int]:
      """simple docstring"""





      super().__init__(tie_word_embeddings=snake_case       ,      is_encoder_decoder=snake_case       ,      **snake_case   )

      if text_config is None:
         a__						:    Optional[Any]   		=	{}
         logger.info("text_config is None. Initializing the Pix2StructTextConfig with default values."   )

      if vision_config is None:
         a__						:    List[Any]   		=	{}
         logger.info("vision_config is None. Initializing the Pix2StructVisionConfig with default values."   )

      a__						:    List[str]   		=	PixaStructTextConfig(**snake_case   )
      a__						:    List[str]   		=	PixaStructVisionConfig(**snake_case   )

      a__						:    Tuple   		=	self.text_config.decoder_start_token_id
      a__						:    int   		=	self.text_config.pad_token_id
      a__						:    List[Any]   		=	self.text_config.eos_token_id

      a__						:    str   		=	initializer_factor
      a__						:    Dict   		=	initializer_range

      a__						:    Optional[int]   		=	self.initializer_range
      a__						:    Union[str, Any]   		=	self.initializer_range

      a__						:    Union[str, Any]   		=	is_vqa





   @classmethod
   def 						_snake_case			(			cls       ,      snake_case       ,      snake_case       ,      **snake_case   )					->	Dict:
      """simple docstring"""





      return cls(text_config=text_config.to_dict()       ,      vision_config=vision_config.to_dict()       ,      **snake_case   )





   def 						_snake_case			(			self   )					->	int:
      """simple docstring"""





      a__						:    Tuple   		=	copy.deepcopy(self.__dict__   )
      a__						:    List[str]   		=	self.text_config.to_dict()
      a__						:    Union[str, Any]   		=	self.vision_config.to_dict()
      a__						:    List[Any]   		=	self.__class__.model_type
      return output

