





from math import ceil
from typing import List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import BatchFeature, SequenceFeatureExtractor
from ...utils import TensorType, logging


_lowerCAmelCase 				=     logging.get_logger(__name__)

class    _UpperCAmelCase							(		_lowerCamelCase     ):
      a						       =							['''audio_values''', '''audio_mask''']
      def __init__(		self      ,       a__=2048      ,       a__=1      ,       a__=[16, 16]      ,       a__=128      ,       a__=44100      ,       a__=86      ,       a__=2048      ,       a__=0.0      ,       **a__      ,       ):
         super().__init__(
             feature_size=a__      ,       sampling_rate=a__      ,       padding_value=a__      ,       **a__      ,       )

         A_     :				Dict			      =	spectrogram_length
         A_     :				Tuple			      =	num_channels
         A_     :				Any			      =	patch_size
         A_     :				str			      =	feature_size // self.patch_size[1]
         A_     :				List[Any]			      =	n_fft
         A_     :				Optional[Any]			      =	sampling_rate // hop_length_to_sampling_rate
         A_     :				List[Any]			      =	sampling_rate
         A_     :				Optional[Any]			      =	padding_value
         A_     :				Any			      =	mel_filter_bank(
             num_frequency_bins=1 + n_fft // 2      ,       num_mel_filters=a__      ,       min_frequency=0.0      ,       max_frequency=2_2050.0      ,       sampling_rate=a__      ,       norm="""slaney"""      ,       mel_scale="""slaney"""      ,       ).T
      def 		_lowerCamelCase      (		self      ,       a__       ):
         A_     :				List[Any]			      =	spectrogram(
             a__      ,       window_function(self.n_fft      ,       """hann"""       )      ,       frame_length=self.n_fft      ,       hop_length=self.hop_length      ,       power=2.0      ,       mel_filters=self.mel_filters.T      ,       log_mel="""dB"""      ,       db_range=80.0      ,       )
         A_     :				Dict			      =	log_spec[:, :-1]
         A_     :				str			      =	log_spec - 20.0
         A_     :				List[Any]			      =	np.clip(log_spec / 40.0      ,       -2.0      ,       0.0       ) + 1.0
         return log_spec






      def __call__(		self      ,       a__      ,       a__ = None      ,       a__ = True      ,       a__ = None      ,       a__ = False      ,       a__ = False      ,       **a__      ,       ):

         if sampling_rate is not None:
            if sampling_rate != self.sampling_rate:
               raise ValueError(
                   """This feature extractor is set to support sampling rate"""
                   F""" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"""
                   F""" 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_     :				str			      =	isinstance(a__      ,       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_     :				List[str]			      =	is_batched_numpy or (
             isinstance(a__      ,       (list, tuple)       ) and (isinstance(raw_speech[0]      ,       (np.ndarray, tuple, list)       ))
         )
         if is_batched:
            A_     :				Any			      =	[np.asarray([speech]      ,       dtype=np.floataa       ).T for speech in raw_speech]
         elif not is_batched and not isinstance(a__      ,       np.ndarray       ):
            A_     :				Dict			      =	np.asarray(a__      ,       dtype=np.floataa       )
         elif isinstance(a__      ,       np.ndarray       ) and raw_speech.dtype is np.dtype(np.floataa       ):
            A_     :				int			      =	raw_speech.astype(np.floataa       )
         # always return batch
         if not is_batched:
            A_     :				Optional[int]			      =	[np.asarray([raw_speech]       ).T]

         # Convert audio signals to log mel spectrograms, truncate by time axis
         A_     :				List[str]			      =	[
             self._np_extract_fbank_features(waveform.squeeze()       ).T[: self.spectrogram_length] for waveform in raw_speech
         ]
         if isinstance(audio_features[0]      ,       a__       ):
            A_     :				Dict			      =	[np.asarray(a__      ,       dtype=np.floataa       ) for feature in audio_features]

         # Create audio attention mask
         A_     :				Dict			      =	max(
             [ceil(feature.shape[0] / self.patch_size[0]       ) * self.freq_len for feature in audio_features]       )  # The maximum number of audio patches in a batch
         if return_attention_mask:
            A_     :				List[str]			      =	[
                (ceil(feature.shape[0] / self.patch_size[0]       ) * self.freq_len) * [1]
                + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]       ) * self.freq_len) * [0]
                for feature in audio_features
            ]
            A_     :				List[Any]			      =	np.array(a__       ).astype(np.floataa       )

         # convert into correct format for padding
         A_     :				int			      =	max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
         A_     :				Optional[int]			      =	np.ones([len(a__       ), 1, max_time_len, self.feature_size]       ).astype(np.floataa       )
         A_     :				str			      =	padded_audio_features * self.padding_value
         for i in range(len(a__       )       ):
            A_     :				Union[str, Any]			      =	audio_features[i]
            A_     :				str			      =	feature

         # return as BatchFeature
         if return_attention_mask:
            A_     :				Any			      =	{"""audio_values""": padded_audio_features, """audio_mask""": audio_mask}
         else:
            A_     :				Optional[Any]			      =	{"""audio_values""": padded_audio_features}

         A_     :				List[Any]			      =	BatchFeature(data=a__      ,       tensor_type=a__       )
         return encoded_inputs













import inspect
import math
import tempfile
import unittest

import numpy as np

from transformers import ViTMAEConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
      import torch
      from torch import nn

      from transformers import ViTMAEForPreTraining, ViTMAEModel
      from transformers.models.vit.modeling_vit import VIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
      from PIL import Image

      from transformers import ViTImageProcessor

class    _UpperCAmelCase							:
      def __init__(		self      ,       a__      ,       a__=13      ,       a__=30      ,       a__=2      ,       a__=3      ,       a__=True      ,       a__=True      ,       a__=32      ,       a__=5      ,       a__=4      ,       a__=37      ,       a__="gelu"      ,       a__=0.1      ,       a__=0.1      ,       a__=10      ,       a__=0.02      ,       a__=3      ,       a__=0.6      ,       a__=None      ,       ):
         A_     :				int			      =	parent
         A_     :				Optional[int]			      =	batch_size
         A_     :				Any			      =	image_size
         A_     :				Optional[int]			      =	patch_size
         A_     :				int			      =	num_channels
         A_     :				str			      =	is_training
         A_     :				str			      =	use_labels
         A_     :				str			      =	hidden_size
         A_     :				Union[str, Any]			      =	num_hidden_layers
         A_     :				Tuple			      =	num_attention_heads
         A_     :				Any			      =	intermediate_size
         A_     :				List[Any]			      =	hidden_act
         A_     :				List[Any]			      =	hidden_dropout_prob
         A_     :				Optional[int]			      =	attention_probs_dropout_prob
         A_     :				str			      =	type_sequence_label_size
         A_     :				int			      =	initializer_range
         A_     :				List[Any]			      =	mask_ratio
         A_     :				str			      =	scope

         # in ViTMAE, the expected sequence length = (num_patches + 1) * (1 - config.mask_ratio), rounded above
         # (we add 1 for the [CLS] token)
         A_     :				Optional[Any]			      =	(image_size // patch_size) ** 2
         A_     :				List[Any]			      =	int(math.ceil((1 - mask_ratio) * (num_patches + 1)       )       )
      def 		_lowerCamelCase      (		self       ):
         A_     :				List[str]			      =	floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]       )

         A_     :				int			      =	None
         if self.use_labels:
            A_     :				List[Any]			      =	ids_tensor([self.batch_size]      ,       self.type_sequence_label_size       )

         A_     :				Any			      =	self.get_config()

         return config, pixel_values, labels
      def 		_lowerCamelCase      (		self       ):
         return ViTMAEConfig(
             image_size=self.image_size      ,       patch_size=self.patch_size      ,       num_channels=self.num_channels      ,       hidden_size=self.hidden_size      ,       num_hidden_layers=self.num_hidden_layers      ,       num_attention_heads=self.num_attention_heads      ,       intermediate_size=self.intermediate_size      ,       hidden_act=self.hidden_act      ,       hidden_dropout_prob=self.hidden_dropout_prob      ,       attention_probs_dropout_prob=self.attention_probs_dropout_prob      ,       is_decoder=a__      ,       initializer_range=self.initializer_range      ,       mask_ratio=self.mask_ratio      ,       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__       ):
         A_     :				Optional[int]			      =	ViTMAEModel(config=a__       )
         model.to(a__       )
         model.eval()
         A_     :				int			      =	model(a__       )
         self.parent.assertEqual(result.last_hidden_state.shape      ,       (self.batch_size, self.seq_length, self.hidden_size)       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__       ):
         A_     :				int			      =	ViTMAEForPreTraining(a__       )
         model.to(a__       )
         model.eval()
         A_     :				Optional[Any]			      =	model(a__       )
         A_     :				Dict			      =	(self.image_size // self.patch_size) ** 2
         A_     :				Optional[int]			      =	self.patch_size**2 * self.num_channels
         self.parent.assertEqual(result.logits.shape      ,       (self.batch_size, num_patches, expected_num_channels)       )

         # test greyscale images
         A_     :				Optional[int]			      =	1
         A_     :				Any			      =	ViTMAEForPreTraining(a__       )
         model.to(a__       )
         model.eval()
         A_     :				Dict			      =	floats_tensor([self.batch_size, 1, self.image_size, self.image_size]       )
         A_     :				List[Any]			      =	model(a__       )
         A_     :				Optional[int]			      =	self.patch_size**2
         self.parent.assertEqual(result.logits.shape      ,       (self.batch_size, num_patches, expected_num_channels)       )






      def 		_lowerCamelCase      (		self       ):
         A_     :				Optional[Any]			      =	self.prepare_config_and_inputs()
         A_      ,				A_      ,				A_     :				Any			      =	config_and_inputs
         A_     :				Dict			      =	{"""pixel_values""": pixel_values}
         return config, inputs_dict






@require_torch
class    _UpperCAmelCase							(		_lowerCamelCase			,       _lowerCamelCase			,       unittest.TestCase     ):
      a						       =							(ViTMAEModel, ViTMAEForPreTraining) if is_torch_available() else ()
      a						       =							{'''feature-extraction''': ViTMAEModel} if is_torch_available() else {}

      a						       =							False
      a						       =							False
      a						       =							False
      a						       =							False
      def 		_lowerCamelCase      (		self       ):
         A_     :				int			      =	ViTMAEModelTester(self       )
         A_     :				List[Any]			      =	ConfigTester(self      ,       config_class=a__      ,       has_text_modality=a__      ,       hidden_size=37       )
      def 		_lowerCamelCase      (		self       ):
         self.config_tester.run_common_tests()
      @unittest.skip(reason="""ViTMAE does not use inputs_embeds"""       )
      def 		_lowerCamelCase      (		self       ):
         pass
      def 		_lowerCamelCase      (		self       ):
         A_      ,				A_     :				Union[str, Any]			      =	self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
            A_     :				Dict			      =	model_class(a__       )
            self.assertIsInstance(model.get_input_embeddings()      ,       (nn.Module)       )
            A_     :				int			      =	model.get_output_embeddings()
            self.assertTrue(x is None or isinstance(a__      ,       nn.Linear       )       )
      def 		_lowerCamelCase      (		self       ):
         A_      ,				A_     :				int			      =	self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
            A_     :				Any			      =	model_class(a__       )
            A_     :				Union[str, Any]			      =	inspect.signature(model.forward       )
            # signature.parameters is an OrderedDict => so arg_names order is deterministic
            A_     :				Dict			      =	[*signature.parameters.keys()]

            A_     :				List[str]			      =	["""pixel_values"""]
            self.assertListEqual(arg_names[:1]      ,       a__       )
      def 		_lowerCamelCase      (		self       ):
         A_     :				Dict			      =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*a__       )
      def 		_lowerCamelCase      (		self       ):
         A_     :				Any			      =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_pretraining(*a__       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__       ):
         # make masks reproducible
         np.random.seed(2       )

         A_     :				Optional[Any]			      =	int((pt_model.config.image_size // pt_model.config.patch_size) ** 2       )
         A_     :				List[str]			      =	np.random.uniform(size=(self.model_tester.batch_size, num_patches)       )
         A_     :				Optional[Any]			      =	torch.from_numpy(a__       )

         # Add `noise` argument.
         # PT inputs will be prepared in `super().check_pt_tf_models()` with this added `noise` argument
         A_     :				Any			      =	pt_noise

         super().check_pt_tf_models(a__      ,       a__      ,       a__       )
      def 		_lowerCamelCase      (		self       ):
         A_      ,				A_     :				int			      =	self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
            A_     :				List[Any]			      =	model_class(a__       )
            model.to(a__       )
            model.eval()
            # make random mask reproducible
            torch.manual_seed(2       )
            with torch.no_grad():
               A_     :				Union[str, Any]			      =	model(**self._prepare_for_class(a__      ,       a__       )       )

            A_     :				int			      =	outputs[0].cpu().numpy()
            A_     :				Union[str, Any]			      =	0

            with tempfile.TemporaryDirectory() as tmpdirname:
               model.save_pretrained(a__       )
               A_     :				Union[str, Any]			      =	model_class.from_pretrained(a__       )
               model.to(a__       )
               # make random mask reproducible
               torch.manual_seed(2       )
               with torch.no_grad():
                  A_     :				Optional[int]			      =	model(**self._prepare_for_class(a__      ,       a__       )       )

               # Make sure we don't have nans
               A_     :				Optional[int]			      =	after_outputs[0].cpu().numpy()
               A_     :				str			      =	0
               A_     :				Optional[int]			      =	np.amax(np.abs(out_a - out_a       )       )
               self.assertLessEqual(a__      ,       1E-5       )
      @unittest.skip(
          reason="""ViTMAE returns a random mask + ids_restore in each forward pass. See test_save_load
    to get deterministic results."""       )
      def 		_lowerCamelCase      (		self       ):
         pass
      @unittest.skip(
          reason="""ViTMAE returns a random mask + ids_restore in each forward pass. See test_save_load
    to get deterministic results."""       )
      def 		_lowerCamelCase      (		self       ):
         pass
      @unittest.skip(
          reason="""ViTMAE returns a random mask + ids_restore in each forward pass. See test_save_load
    to get deterministic results."""       )
      def 		_lowerCamelCase      (		self       ):
         pass
      @unittest.skip(reason="""ViTMAE returns a random mask + ids_restore in each forward pass. See test_save_load"""       )
      def 		_lowerCamelCase      (		self       ):
         pass
      @unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""       )
      def 		_lowerCamelCase      (		self       ):
         pass






      @slow
      def 		_lowerCamelCase      (		self       ):
         for model_name in VIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
            A_     :				Union[str, Any]			      =	ViTMAEModel.from_pretrained(a__       )
            self.assertIsNotNone(a__       )






def     _lowerCAmelCase		(  ):
   '''simple docstring'''


   A_     :				List[Any]			      =	Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""     )
   return image







@require_torch
@require_vision
class    _UpperCAmelCase							(		unittest.TestCase     ):
      @cached_property
      def 		_lowerCamelCase      (		self       ):
         return ViTImageProcessor.from_pretrained("""facebook/vit-mae-base"""       ) if is_vision_available() else None






      @slow
      def 		_lowerCamelCase      (		self       ):
         # make random mask reproducible across the PT and TF model
         np.random.seed(2       )

         A_     :				Union[str, Any]			      =	ViTMAEForPreTraining.from_pretrained("""facebook/vit-mae-base"""       ).to(a__       )

         A_     :				Optional[Any]			      =	self.default_image_processor
         A_     :				Union[str, Any]			      =	prepare_img()
         A_     :				Union[str, Any]			      =	image_processor(images=a__      ,       return_tensors="""pt"""       ).to(a__       )

         # prepare a noise vector that will be also used for testing the TF model
         # (this way we can ensure that the PT and TF models operate on the same inputs)
         A_     :				Optional[int]			      =	ViTMAEConfig()
         A_     :				Tuple			      =	int((vit_mae_config.image_size // vit_mae_config.patch_size) ** 2       )
         A_     :				Tuple			      =	np.random.uniform(size=(1, num_patches)       )

         # forward pass
         with torch.no_grad():
            A_     :				Dict			      =	model(**a__      ,       noise=torch.from_numpy(a__       ).to(device=a__       )       )

         # verify the logits
         A_     :				Tuple			      =	torch.Size((1, 196, 768)       )
         self.assertEqual(outputs.logits.shape      ,       a__       )

         A_     :				List[str]			      =	torch.tensor(
             [[-0.0548, -1.7023, -0.9325], [0.3721, -0.5670, -0.2233], [0.8235, -1.3878, -0.3524]]       )

         self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]      ,       expected_slice.to(a__       )      ,       atol=1E-4       )       )






