


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


__lowerCamelCase	:		int										=   logging.get_logger(__name__)







class  _lowercase   (   __lowerCAmelCase						):
      _a							:						Union[str, Any]										=							['''input_features''', '''is_longer''']
      def __init__(     self				,			a=6_4				,			a=4_8_0_0_0				,			a=4_8_0				,			a=1_0				,			a=1_0_2_4				,			a=0.0				,			a=False				,			a = 0				,			a = 1_4_0_0_0				,			a = None				,			a = "fusion"				,			a = "repeatpad"				,			**a				,			):
       super().__init__(
           feature_size=lowerCAmelCase_				,			sampling_rate=lowerCAmelCase_				,			padding_value=lowerCAmelCase_				,			return_attention_mask=lowerCAmelCase_				,			**lowerCAmelCase_				,			)
       snake_case__  :    Optional[int]							       =top_db
       snake_case__  :    str							       =truncation
       snake_case__  :    Optional[Any]							       =padding
       snake_case__  :    Tuple							       =fft_window_size
       snake_case__  :    Dict							       =(fft_window_size >> 1) + 1
       snake_case__  :    List[Any]							       =hop_length
       snake_case__  :    List[str]							       =max_length_s
       snake_case__  :    Dict							       =max_length_s * sampling_rate
       snake_case__  :    int							       =sampling_rate
       snake_case__  :    Union[str, Any]							       =frequency_min
       snake_case__  :    Dict							       =frequency_max
       snake_case__  :    Union[str, Any]							       =mel_filter_bank(
           num_frequency_bins=self.nb_frequency_bins				,			num_mel_filters=lowerCAmelCase_				,			min_frequency=lowerCAmelCase_				,			max_frequency=lowerCAmelCase_				,			sampling_rate=lowerCAmelCase_				,			norm=lowerCAmelCase_				,			mel_scale="""htk"""				,			)
       snake_case__  :    Tuple							       =mel_filter_bank(
           num_frequency_bins=self.nb_frequency_bins				,			num_mel_filters=lowerCAmelCase_				,			min_frequency=lowerCAmelCase_				,			max_frequency=lowerCAmelCase_				,			sampling_rate=lowerCAmelCase_				,			norm="""slaney"""				,			mel_scale="""slaney"""				,			)
      def 						lowercase__					(     self   ):
       snake_case__  :    Dict							       =copy.deepcopy(self.__dict__   )
       snake_case__  :    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 						lowercase__					(     self				,			a				,			a = None   ):
       snake_case__  :    Tuple							       =spectrogram(
           lowerCAmelCase_				,			window_function(self.fft_window_size				,			"""hann"""   )				,			frame_length=self.fft_window_size				,			hop_length=self.hop_length				,			power=2.0				,			mel_filters=lowerCAmelCase_				,			log_mel="""dB"""				,			)
       return log_mel_spectrogram.T
      def 						lowercase__					(     self				,			a				,			a				,			a   ):
       snake_case__  :    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
        snake_case__  :    Any							       =[0]
       if len(ranges[2]   ) == 0:
        # if the audio is too short, we just use the first chunk
        snake_case__  :    Union[str, Any]							       =[0]
       # randomly choose index for each part
       snake_case__  :    Any							       =np.random.choice(ranges[0]   )
       snake_case__  :    Optional[Any]							       =np.random.choice(ranges[1]   )
       snake_case__  :    str							       =np.random.choice(ranges[2]   )

       snake_case__  :    Optional[Any]							       =mel[idx_front : idx_front + chunk_frames, :]
       snake_case__  :    Dict							       =mel[idx_middle : idx_middle + chunk_frames, :]
       snake_case__  :    List[str]							       =mel[idx_back : idx_back + chunk_frames, :]

       snake_case__  :    List[Any]							       =torch.tensor(mel[None, None, :]   )
       snake_case__  :    Optional[int]							       =torch.nn.functional.interpolate(
           lowerCAmelCase_				,			size=[chunk_frames, 6_4]				,			mode="""bilinear"""				,			align_corners=lowerCAmelCase_   )
       snake_case__  :    List[str]							       =mel_shrink[0][0].numpy()
       snake_case__  :    Dict							       =np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]				,			axis=0   )
       return mel_fusion
      def 						lowercase__					(     self				,			a				,			a				,			a				,			a   ):
       if waveform.shape[0] > max_length:
        if truncation == "rand_trunc":
         snake_case__  :    Optional[Any]							       =True
         # random crop to max_length (for compatibility) -> this should be handled by self.pad
         snake_case__  :    str							       =len(lowerCAmelCase_   ) - max_length
         snake_case__  :    Optional[int]							       =np.random.randint(0				,			overflow + 1   )
         snake_case__  :    Any							       =waveform[idx : idx + max_length]
         snake_case__  :    Any							       =self._np_extract_fbank_features(lowerCAmelCase_				,			self.mel_filters_slaney   )[None, :]
        elif truncation == "fusion":
         snake_case__  :    List[Any]							       =self._np_extract_fbank_features(lowerCAmelCase_				,			self.mel_filters   )
         snake_case__  :    str							       =max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
         snake_case__  :    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.
          snake_case__  :    List[str]							       =np.stack([mel, mel, mel, mel]				,			axis=0   )
          snake_case__  :    str							       =False
         else:
          snake_case__  :    str							       =self._random_mel_fusion(lowerCAmelCase_				,			lowerCAmelCase_				,			lowerCAmelCase_   )
          snake_case__  :    int							       =True
        else:
         raise NotImplementedError(F"data_truncating {truncation} not implemented"   )

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

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

       return input_mel, longer



      def __call__(     self				,			a				,			a = None				,			a = None				,			a = None				,			a = None				,			a = None				,			**a				,			):
       snake_case__  :    Dict							       =truncation if truncation is not None else self.truncation
       snake_case__  :    Tuple							       =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[Any]							       =isinstance(lowerCAmelCase_				,			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__  :    Optional[Any]							       =is_batched_numpy or (
           isinstance(lowerCAmelCase_				,			(list, tuple)   ) and (isinstance(raw_speech[0]				,			(np.ndarray, tuple, list)   ))
       )

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

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

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

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

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

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

       # is_longer is a list of bool
       snake_case__  :    List[Any]							       =[[longer] for longer in is_longer]

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

       if return_tensors is not None:
        snake_case__  :    Any							       =input_features.convert_to_tensors(lowerCAmelCase_   )

       return input_features













from collections import OrderedDict
from typing import Mapping

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


__lowerCamelCase	:		Dict										=   logging.get_logger(__name__)

__lowerCamelCase	:		Union[str, Any]										=   {
    """xlm-mlm-en-2048""": """https://huggingface.co/xlm-mlm-en-2048/resolve/main/config.json""",
    """xlm-mlm-ende-1024""": """https://huggingface.co/xlm-mlm-ende-1024/resolve/main/config.json""",
    """xlm-mlm-enfr-1024""": """https://huggingface.co/xlm-mlm-enfr-1024/resolve/main/config.json""",
    """xlm-mlm-enro-1024""": """https://huggingface.co/xlm-mlm-enro-1024/resolve/main/config.json""",
    """xlm-mlm-tlm-xnli15-1024""": """https://huggingface.co/xlm-mlm-tlm-xnli15-1024/resolve/main/config.json""",
    """xlm-mlm-xnli15-1024""": """https://huggingface.co/xlm-mlm-xnli15-1024/resolve/main/config.json""",
    """xlm-clm-enfr-1024""": """https://huggingface.co/xlm-clm-enfr-1024/resolve/main/config.json""",
    """xlm-clm-ende-1024""": """https://huggingface.co/xlm-clm-ende-1024/resolve/main/config.json""",
    """xlm-mlm-17-1280""": """https://huggingface.co/xlm-mlm-17-1280/resolve/main/config.json""",
    """xlm-mlm-100-1280""": """https://huggingface.co/xlm-mlm-100-1280/resolve/main/config.json""",
}







class  _lowercase   (   _A						):
      _a							:						int										=							'xlm'
      _a							:						List[Any]										=							{
          'hidden_size': 'emb_dim',
          'num_attention_heads': 'n_heads',
          'num_hidden_layers': 'n_layers',
          'n_words': 'vocab_size',  # For backward compatibility
      }
      def __init__(     self				,			a=3_0_1_4_5				,			a=2_0_4_8				,			a=1_2				,			a=1_6				,			a=0.1				,			a=0.1				,			a=True				,			a=False				,			a=False				,			a=False				,			a=1				,			a=True				,			a=5_1_2				,			a=2_0_4_8**-0.5				,			a=1e-12				,			a=0.02				,			a=0				,			a=1				,			a=2				,			a=3				,			a=5				,			a=True				,			a="first"				,			a=True				,			a=None				,			a=True				,			a=0.1				,			a=5				,			a=5				,			a=0				,			a=0				,			a=2				,			a=0				,			**a				,			):
       snake_case__  :    str							       =vocab_size
       snake_case__  :    List[Any]							       =emb_dim
       snake_case__  :    List[str]							       =n_layers
       snake_case__  :    Any							       =n_heads
       snake_case__  :    Any							       =dropout
       snake_case__  :    Optional[int]							       =attention_dropout
       snake_case__  :    Dict							       =gelu_activation
       snake_case__  :    Tuple							       =sinusoidal_embeddings
       snake_case__  :    Optional[int]							       =causal
       snake_case__  :    Optional[int]							       =asm
       snake_case__  :    Optional[int]							       =n_langs
       snake_case__  :    Union[str, Any]							       =use_lang_emb
       snake_case__  :    Optional[Any]							       =layer_norm_eps
       snake_case__  :    Union[str, Any]							       =bos_index
       snake_case__  :    List[Any]							       =eos_index
       snake_case__  :    Optional[Any]							       =pad_index
       snake_case__  :    Union[str, Any]							       =unk_index
       snake_case__  :    Dict							       =mask_index
       snake_case__  :    str							       =is_encoder
       snake_case__  :    Optional[Any]							       =max_position_embeddings
       snake_case__  :    Any							       =embed_init_std
       snake_case__  :    Optional[Any]							       =init_std
       snake_case__  :    str							       =summary_type
       snake_case__  :    List[Any]							       =summary_use_proj
       snake_case__  :    str							       =summary_activation
       snake_case__  :    Optional[Any]							       =summary_proj_to_labels
       snake_case__  :    Optional[int]							       =summary_first_dropout
       snake_case__  :    Optional[Any]							       =start_n_top
       snake_case__  :    List[Any]							       =end_n_top
       snake_case__  :    Union[str, Any]							       =mask_token_id
       snake_case__  :    Any							       =lang_id

       if "n_words" in kwargs:
        snake_case__  :    int							       =kwargs["""n_words"""]

       super().__init__(pad_token_id=a				,			bos_token_id=a				,			**a   )







class  _lowercase   (   _A						):
      @property
      def 						lowercase__					(     self   ):
       if self.task == "multiple-choice":
        snake_case__  :    Any							       ={0: """batch""", 1: """choice""", 2: """sequence"""}
       else:
        snake_case__  :    Union[str, Any]							       ={0: """batch""", 1: """sequence"""}
       return OrderedDict(
           [
               ("""input_ids""", dynamic_axis),
               ("""attention_mask""", dynamic_axis),
               ("""token_type_ids""", dynamic_axis),
           ]   )





