



import math
from collections import defaultdict
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin, SchedulerOutput



def     _lowerCAmelCase		(  _lowerCAmelCase  ,_lowerCAmelCase=0.999  ,_lowerCAmelCase="cosine"  ,):
   '''simple docstring'''


   if alpha_transform_type == "cosine":

      def alpha_bar_fn(_lowerCAmelCase     ):
         return math.cos((t + 0.008) / 1.008 * math.pi / 2     ) ** 2

   elif alpha_transform_type == "exp":

      def alpha_bar_fn(_lowerCAmelCase     ):
         return math.exp(t * -12.0     )

   else:
      raise ValueError(f"""Unsupported alpha_tranform_type: {alpha_transform_type}"""     )

   A_     :				int			      =	[]
   for i in range(_lowerCAmelCase     ):
      A_     :				List[str]			      =	i / num_diffusion_timesteps
      A_     :				List[Any]			      =	(i + 1) / num_diffusion_timesteps
      betas.append(min(1 - alpha_bar_fn(_lowerCAmelCase     ) / alpha_bar_fn(_lowerCAmelCase     )  ,_lowerCAmelCase     )     )
   return torch.tensor(_lowerCAmelCase  ,dtype=torch.floataa     )







class    _UpperCAmelCase							(		_lowerCAmelCase			,       _lowerCAmelCase     ):
      a						       =							[e.name for e in KarrasDiffusionSchedulers]
      a						       =							2
      @register_to_config
      def __init__(		self      ,       a__ = 1000      ,       a__ = 0.0_0085      ,       a__ = 0.012      ,       a__ = "linear"      ,       a__ = None      ,       a__ = "epsilon"      ,       a__ = False      ,       a__ = False      ,       a__ = 1.0      ,       a__ = "linspace"      ,       a__ = 0      ,       ):







         if trained_betas is not None:
            A_     :				List[str]			      =	torch.tensor(_lowerCAmelCase      ,       dtype=torch.floataa       )
         elif beta_schedule == "linear":
            A_     :				Tuple			      =	torch.linspace(_lowerCAmelCase      ,       _lowerCAmelCase      ,       _lowerCAmelCase      ,       dtype=torch.floataa       )
         elif beta_schedule == "scaled_linear":
            # this schedule is very specific to the latent diffusion model.
            A_     :				Optional[int]			      =	(
                torch.linspace(beta_start**0.5      ,       beta_end**0.5      ,       _lowerCAmelCase      ,       dtype=torch.floataa       ) ** 2
            )
         elif beta_schedule == "squaredcos_cap_v2":
            # Glide cosine schedule
            A_     :				Any			      =	betas_for_alpha_bar(_lowerCAmelCase      ,       alpha_transform_type="""cosine"""       )
         elif beta_schedule == "exp":
            A_     :				Optional[int]			      =	betas_for_alpha_bar(_lowerCAmelCase      ,       alpha_transform_type="""exp"""       )
         else:
            raise NotImplementedError(F"""{beta_schedule} does is not implemented for {self.__class__}"""       )

         A_     :				Any			      =	1.0 - self.betas
         A_     :				str			      =	torch.cumprod(self.alphas      ,       dim=0       )

         #  set all values
         self.set_timesteps(_lowerCAmelCase      ,       _lowerCAmelCase      ,       _lowerCAmelCase       )
         A_     :				Optional[Any]			      =	use_karras_sigmas
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__=None       ):







         if schedule_timesteps is None:
            A_     :				Optional[int]			      =	self.timesteps

         A_     :				List[Any]			      =	(schedule_timesteps == timestep).nonzero()

         # The sigma index that is taken for the **very** first `step`
         # is always the second index (or the last index if there is only 1)
         # This way we can ensure we don't accidentally skip a sigma in
         # case we start in the middle of the denoising schedule (e.g. for image-to-image)
         if len(self._index_counter       ) == 0:
            A_     :				int			      =	1 if len(_lowerCAmelCase       ) > 1 else 0
         else:
            A_     :				List[Any]			      =	timestep.cpu().item() if torch.is_tensor(_lowerCAmelCase       ) else timestep
            A_     :				List[Any]			      =	self._index_counter[timestep_int]

         return indices[pos].item()
      @property
      def 		_lowerCamelCase      (		self       ):







         if self.config.timestep_spacing in ["linspace", "trailing"]:
            return self.sigmas.max()

         return (self.sigmas.max() ** 2 + 1) ** 0.5
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       ):
         A_     :				int			      =	self.index_for_timestep(_lowerCAmelCase       )

         A_     :				Tuple			      =	self.sigmas[step_index]
         A_     :				Optional[Any]			      =	sample / ((sigma**2 + 1) ** 0.5)
         return sample
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__ = None      ,       a__ = None      ,       ):
         A_     :				int			      =	num_inference_steps

         A_     :				int			      =	num_train_timesteps or self.config.num_train_timesteps

         # "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
         if self.config.timestep_spacing == "linspace":
            A_     :				Union[str, Any]			      =	np.linspace(0      ,       num_train_timesteps - 1      ,       _lowerCAmelCase      ,       dtype=_lowerCAmelCase       )[::-1].copy()
         elif self.config.timestep_spacing == "leading":
            A_     :				List[Any]			      =	num_train_timesteps // self.num_inference_steps
            # creates integer timesteps by multiplying by ratio
            # casting to int to avoid issues when num_inference_step is power of 3
            A_     :				int			      =	(np.arange(0      ,       _lowerCAmelCase       ) * step_ratio).round()[::-1].copy().astype(_lowerCAmelCase       )
            timesteps += self.config.steps_offset
         elif self.config.timestep_spacing == "trailing":
            A_     :				Any			      =	num_train_timesteps / self.num_inference_steps
            # creates integer timesteps by multiplying by ratio
            # casting to int to avoid issues when num_inference_step is power of 3
            A_     :				int			      =	(np.arange(_lowerCAmelCase      ,       0      ,       -step_ratio       )).round().copy().astype(_lowerCAmelCase       )
            timesteps -= 1
         else:
            raise ValueError(
                F"""{self.config.timestep_spacing} is not supported. Please make sure to choose one of \'linspace\', \'leading\' or \'trailing\'."""       )

         A_     :				Tuple			      =	np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5       )
         A_     :				str			      =	np.log(_lowerCAmelCase       )
         A_     :				Tuple			      =	np.interp(_lowerCAmelCase      ,       np.arange(0      ,       len(_lowerCAmelCase       )       )      ,       _lowerCAmelCase       )

         if self.config.use_karras_sigmas:
            A_     :				int			      =	self._convert_to_karras(in_sigmas=_lowerCAmelCase      ,       num_inference_steps=self.num_inference_steps       )
            A_     :				List[Any]			      =	np.array([self._sigma_to_t(_lowerCAmelCase      ,       _lowerCAmelCase       ) for sigma in sigmas]       )

         A_     :				str			      =	np.concatenate([sigmas, [0.0]]       ).astype(np.floataa       )
         A_     :				Union[str, Any]			      =	torch.from_numpy(_lowerCAmelCase       ).to(device=_lowerCAmelCase       )
         A_     :				Optional[Any]			      =	torch.cat([sigmas[:1], sigmas[1:-1].repeat_interleave(2       ), sigmas[-1:]]       )

         A_     :				Any			      =	torch.from_numpy(_lowerCAmelCase       )
         A_     :				Any			      =	torch.cat([timesteps[:1], timesteps[1:].repeat_interleave(2       )]       )

         if str(_lowerCAmelCase       ).startswith("""mps"""       ):
            # mps does not support float64
            A_     :				List[str]			      =	timesteps.to(_lowerCAmelCase      ,       dtype=torch.floataa       )
         else:
            A_     :				List[Any]			      =	timesteps.to(device=_lowerCAmelCase       )

         # empty dt and derivative
         A_     :				Optional[int]			      =	None
         A_     :				List[Any]			      =	None

         # for exp beta schedules, such as the one for `pipeline_shap_e.py`
         # we need an index counter
         A_     :				List[str]			      =	defaultdict(_lowerCAmelCase       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__       ):
         A_     :				Dict			      =	np.log(_lowerCAmelCase       )

         # get distribution
         A_     :				Optional[int]			      =	log_sigma - log_sigmas[:, np.newaxis]

         # get sigmas range
         A_     :				int			      =	np.cumsum((dists >= 0)      ,       axis=0       ).argmax(axis=0       ).clip(max=log_sigmas.shape[0] - 2       )
         A_     :				List[str]			      =	low_idx + 1

         A_     :				Any			      =	log_sigmas[low_idx]
         A_     :				Optional[Any]			      =	log_sigmas[high_idx]

         # interpolate sigmas
         A_     :				int			      =	(low - log_sigma) / (low - high)
         A_     :				int			      =	np.clip(_lowerCAmelCase      ,       0      ,       1       )

         # transform interpolation to time range
         A_     :				Dict			      =	(1 - w) * low_idx + w * high_idx
         A_     :				Union[str, Any]			      =	t.reshape(sigma.shape       )
         return t
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__       ):
         A_     :				str			      =	in_sigmas[-1].item()
         A_     :				str			      =	in_sigmas[0].item()

         A_     :				Tuple			      =	7.0  # 7.0 is the value used in the paper
         A_     :				Union[str, Any]			      =	np.linspace(0      ,       1      ,       _lowerCAmelCase       )
         A_     :				Optional[Any]			      =	sigma_min ** (1 / rho)
         A_     :				Optional[Any]			      =	sigma_max ** (1 / rho)
         A_     :				Dict			      =	(max_inv_rho + ramp * (min_inv_rho - max_inv_rho)) ** rho
         return sigmas
      @property
      def 		_lowerCamelCase      (		self       ):
         return self.dt is None
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__      ,       a__ = True      ,       ):
         A_     :				Dict			      =	self.index_for_timestep(_lowerCAmelCase       )

         # advance index counter by 1
         A_     :				str			      =	timestep.cpu().item() if torch.is_tensor(_lowerCAmelCase       ) else timestep
         self._index_counter[timestep_int] += 1

         if self.state_in_first_order:
            A_     :				int			      =	self.sigmas[step_index]
            A_     :				List[str]			      =	self.sigmas[step_index + 1]
         else:
            # 2nd order / Heun's method
            A_     :				Tuple			      =	self.sigmas[step_index - 1]
            A_     :				Optional[Any]			      =	self.sigmas[step_index]

         # currently only gamma=0 is supported. This usually works best anyways.
         # We can support gamma in the future but then need to scale the timestep before
         # passing it to the model which requires a change in API
         A_     :				Optional[Any]			      =	0
         A_     :				Dict			      =	sigma * (gamma + 1)  # Note: sigma_hat == sigma for now

         # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
         if self.config.prediction_type == "epsilon":
            A_     :				List[Any]			      =	sigma_hat if self.state_in_first_order else sigma_next
            A_     :				Optional[int]			      =	sample - sigma_input * model_output
         elif self.config.prediction_type == "v_prediction":
            A_     :				int			      =	sigma_hat if self.state_in_first_order else sigma_next
            A_     :				List[Any]			      =	model_output * (-sigma_input / (sigma_input**2 + 1) ** 0.5) + (
                sample / (sigma_input**2 + 1)
            )
         elif self.config.prediction_type == "sample":
            A_     :				str			      =	model_output
         else:
            raise ValueError(
                F"""prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`"""       )

         if self.config.clip_sample:
            A_     :				Optional[Any]			      =	pred_original_sample.clamp(
                -self.config.clip_sample_range      ,       self.config.clip_sample_range       )

         if self.state_in_first_order:
            # 2. Convert to an ODE derivative for 1st order
            A_     :				Any			      =	(sample - pred_original_sample) / sigma_hat
            # 3. delta timestep
            A_     :				Optional[int]			      =	sigma_next - sigma_hat

            # store for 2nd order step
            A_     :				List[Any]			      =	derivative
            A_     :				int			      =	dt
            A_     :				Optional[int]			      =	sample
         else:
            # 2. 2nd order / Heun's method
            A_     :				Optional[Any]			      =	(sample - pred_original_sample) / sigma_next
            A_     :				Tuple			      =	(self.prev_derivative + derivative) / 2

            # 3. take prev timestep & sample
            A_     :				Optional[Any]			      =	self.dt
            A_     :				str			      =	self.sample

            # free dt and derivative
            # Note, this puts the scheduler in "first order mode"
            A_     :				str			      =	None
            A_     :				Tuple			      =	None
            A_     :				Optional[int]			      =	None

         A_     :				Dict			      =	sample + derivative * dt

         if not return_dict:
            return (prev_sample,)

         return SchedulerOutput(prev_sample=_lowerCAmelCase       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__      ,       ):
         A_     :				int			      =	self.sigmas.to(device=original_samples.device      ,       dtype=original_samples.dtype       )
         if original_samples.device.type == "mps" and torch.is_floating_point(_lowerCAmelCase       ):
            # mps does not support float64
            A_     :				Dict			      =	self.timesteps.to(original_samples.device      ,       dtype=torch.floataa       )
            A_     :				Tuple			      =	timesteps.to(original_samples.device      ,       dtype=torch.floataa       )
         else:
            A_     :				Dict			      =	self.timesteps.to(original_samples.device       )
            A_     :				str			      =	timesteps.to(original_samples.device       )

         A_     :				Optional[int]			      =	[self.index_for_timestep(_lowerCAmelCase      ,       _lowerCAmelCase       ) for t in timesteps]

         A_     :				Any			      =	sigmas[step_indices].flatten()
         while len(sigma.shape       ) < len(original_samples.shape       ):
            A_     :				Dict			      =	sigma.unsqueeze(-1       )

         A_     :				List[str]			      =	original_samples + noise * sigma
         return noisy_samples






      def __len__(		self       ):
         return self.config.num_train_timesteps













import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import VideoMAEConfig
from transformers.models.auto import get_values
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 (
          MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING,
          VideoMAEForPreTraining,
          VideoMAEForVideoClassification,
          VideoMAEModel,
      )
      from transformers.models.videomae.modeling_videomae import VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
      from transformers import VideoMAEImageProcessor

class    _UpperCAmelCase							:
      def __init__(		self      ,       a__      ,       a__=13      ,       a__=10      ,       a__=3      ,       a__=2      ,       a__=2      ,       a__=2      ,       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__=0.9      ,       a__=None      ,       ):
         A_     :				Tuple			      =	parent
         A_     :				Union[str, Any]			      =	batch_size
         A_     :				str			      =	image_size
         A_     :				Union[str, Any]			      =	num_channels
         A_     :				List[str]			      =	patch_size
         A_     :				Optional[Any]			      =	tubelet_size
         A_     :				List[Any]			      =	num_frames
         A_     :				str			      =	is_training
         A_     :				List[Any]			      =	use_labels
         A_     :				List[str]			      =	hidden_size
         A_     :				Optional[Any]			      =	num_hidden_layers
         A_     :				str			      =	num_attention_heads
         A_     :				Union[str, Any]			      =	intermediate_size
         A_     :				Dict			      =	hidden_act
         A_     :				int			      =	hidden_dropout_prob
         A_     :				Union[str, Any]			      =	attention_probs_dropout_prob
         A_     :				Union[str, Any]			      =	type_sequence_label_size
         A_     :				int			      =	initializer_range
         A_     :				Dict			      =	mask_ratio
         A_     :				Optional[Any]			      =	scope

         # in VideoMAE, the number of tokens equals num_frames/tubelet_size * num_patches per frame
         A_     :				int			      =	(image_size // patch_size) ** 2
         A_     :				Dict			      =	(num_frames // tubelet_size) * self.num_patches_per_frame

         # use this variable to define bool_masked_pos
         A_     :				Dict			      =	int(mask_ratio * self.seq_length       )
      def 		_lowerCamelCase      (		self       ):
         A_     :				Any			      =	floats_tensor(
             [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]       )

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

         A_     :				List[Any]			      =	self.get_config()

         return config, pixel_values, labels
      def 		_lowerCamelCase      (		self       ):
         return VideoMAEConfig(
             image_size=self.image_size      ,       patch_size=self.patch_size      ,       num_channels=self.num_channels      ,       num_frames=self.num_frames      ,       tubelet_size=self.tubelet_size      ,       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      ,       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__       ):
         A_     :				Dict			      =	VideoMAEModel(config=a__       )
         model.to(a__       )
         model.eval()
         A_     :				Dict			      =	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_     :				Optional[Any]			      =	VideoMAEForPreTraining(a__       )
         model.to(a__       )
         model.eval()
         # important: each video needs to have the same number of masked patches
         # hence we define a single mask, which we then repeat for each example in the batch
         A_     :				List[Any]			      =	torch.ones((self.num_masks,)       )
         A_     :				Dict			      =	torch.cat([mask, torch.zeros(self.seq_length - mask.size(0       )       )]       )
         A_     :				int			      =	mask.expand(self.batch_size      ,       -1       ).bool()

         A_     :				List[Any]			      =	model(a__      ,       a__       )
         # model only returns predictions for masked patches
         A_     :				Union[str, Any]			      =	mask.sum().item()
         A_     :				Tuple			      =	3 * self.tubelet_size * self.patch_size**2
         self.parent.assertEqual(result.logits.shape      ,       (self.batch_size, num_masked_patches, decoder_num_labels)       )






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






@require_torch
class    _UpperCAmelCase							(		_lowerCamelCase			,       _lowerCamelCase			,       unittest.TestCase     ):
      a						       =							(
          (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
      )
      a						       =							(
          {'''feature-extraction''': VideoMAEModel, '''video-classification''': VideoMAEForVideoClassification}
          if is_torch_available()
          else {}
      )

      a						       =							False
      a						       =							False
      a						       =							False
      a						       =							False
      def 		_lowerCamelCase      (		self       ):
         A_     :				int			      =	VideoMAEModelTester(self       )
         A_     :				Any			      =	ConfigTester(self      ,       config_class=a__      ,       has_text_modality=a__      ,       hidden_size=37       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__=False       ):
         A_     :				Optional[Any]			      =	copy.deepcopy(a__       )

         if model_class == VideoMAEForPreTraining:
            # important: each video needs to have the same number of masked patches
            # hence we define a single mask, which we then repeat for each example in the batch
            A_     :				List[Any]			      =	torch.ones((self.model_tester.num_masks,)       )
            A_     :				List[str]			      =	torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0       )       )]       )
            A_     :				Union[str, Any]			      =	mask.expand(self.model_tester.batch_size      ,       -1       ).bool()
            A_     :				int			      =	bool_masked_pos.to(a__       )

         if return_labels:
            if model_class in [
                *get_values(a__       ),
            ]:
               A_     :				List[str]			      =	torch.zeros(
                   self.model_tester.batch_size      ,       dtype=torch.long      ,       device=a__       )

         return inputs_dict
      def 		_lowerCamelCase      (		self       ):
         self.config_tester.run_common_tests()
      @unittest.skip(reason="""VideoMAE does not use inputs_embeds"""       )
      def 		_lowerCamelCase      (		self       ):
         pass
      def 		_lowerCamelCase      (		self       ):
         A_      ,				A_     :				Optional[Any]			      =	self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
            A_     :				Tuple			      =	model_class(a__       )
            self.assertIsInstance(model.get_input_embeddings()      ,       (nn.Module)       )
            A_     :				Union[str, Any]			      =	model.get_output_embeddings()
            self.assertTrue(x is None or isinstance(a__      ,       nn.Linear       )       )
      def 		_lowerCamelCase      (		self       ):
         A_      ,				A_     :				Optional[int]			      =	self.model_tester.prepare_config_and_inputs_for_common()

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

            A_     :				str			      =	["""pixel_values"""]
            self.assertListEqual(arg_names[:1]      ,       a__       )
      def 		_lowerCamelCase      (		self       ):
         A_     :				List[Any]			      =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*a__       )
      def 		_lowerCamelCase      (		self       ):
         A_     :				int			      =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_pretraining(*a__       )
      @slow
      def 		_lowerCamelCase      (		self       ):
         for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
            A_     :				Any			      =	VideoMAEModel.from_pretrained(a__       )
            self.assertIsNotNone(a__       )
      def 		_lowerCamelCase      (		self       ):
         if not self.has_attentions:
            pass

         else:
            A_      ,				A_     :				List[Any]			      =	self.model_tester.prepare_config_and_inputs_for_common()
            A_     :				Optional[int]			      =	True

            for model_class in self.all_model_classes:
               A_     :				str			      =	self.model_tester.seq_length - self.model_tester.num_masks
               A_     :				List[Any]			      =	(
                   num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
               )

               A_     :				Optional[int]			      =	True
               A_     :				Optional[Any]			      =	False
               A_     :				List[Any]			      =	True
               A_     :				List[str]			      =	model_class(a__       )
               model.to(a__       )
               model.eval()
               with torch.no_grad():
                  A_     :				List[Any]			      =	model(**self._prepare_for_class(a__      ,       a__       )       )
               A_     :				Dict			      =	outputs.attentions
               self.assertEqual(len(a__       )      ,       self.model_tester.num_hidden_layers       )

               # check that output_attentions also work using config
               del inputs_dict["output_attentions"]
               A_     :				List[Any]			      =	True
               A_     :				List[str]			      =	model_class(a__       )
               model.to(a__       )
               model.eval()
               with torch.no_grad():
                  A_     :				int			      =	model(**self._prepare_for_class(a__      ,       a__       )       )
               A_     :				Any			      =	outputs.attentions
               self.assertEqual(len(a__       )      ,       self.model_tester.num_hidden_layers       )

               self.assertListEqual(
                   list(attentions[0].shape[-3:]       )      ,       [self.model_tester.num_attention_heads, seq_len, seq_len]      ,       )
               A_     :				Union[str, Any]			      =	len(a__       )

               # Check attention is always last and order is fine
               A_     :				Optional[Any]			      =	True
               A_     :				int			      =	True
               A_     :				List[str]			      =	model_class(a__       )
               model.to(a__       )
               model.eval()
               with torch.no_grad():
                  A_     :				Tuple			      =	model(**self._prepare_for_class(a__      ,       a__       )       )

               self.assertEqual(out_len + 1      ,       len(a__       )       )

               A_     :				Optional[Any]			      =	outputs.attentions

               self.assertEqual(len(a__       )      ,       self.model_tester.num_hidden_layers       )
               self.assertListEqual(
                   list(self_attentions[0].shape[-3:]       )      ,       [self.model_tester.num_attention_heads, seq_len, seq_len]      ,       )
      def 		_lowerCamelCase      (		self       ):
         def check_hidden_states_output(a__      ,       a__      ,       a__       ):
            A_     :				str			      =	model_class(a__       )
            model.to(a__       )
            model.eval()

            with torch.no_grad():
               A_     :				Union[str, Any]			      =	model(**self._prepare_for_class(a__      ,       a__       )       )

            A_     :				Optional[Any]			      =	outputs.hidden_states
            A_     :				Union[str, Any]			      =	self.model_tester.num_hidden_layers + 1
            self.assertEqual(len(a__       )      ,       a__       )

            A_     :				str			      =	self.model_tester.seq_length - self.model_tester.num_masks
            A_     :				str			      =	num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length

            self.assertListEqual(
                list(hidden_states[0].shape[-2:]       )      ,       [seq_length, self.model_tester.hidden_size]      ,       )

         A_      ,				A_     :				Any			      =	self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
            A_     :				Union[str, Any]			      =	True
            check_hidden_states_output(a__      ,       a__      ,       a__       )

            # check that output_hidden_states also work using config
            del inputs_dict["output_hidden_states"]
            A_     :				Tuple			      =	True

            check_hidden_states_output(a__      ,       a__      ,       a__       )






      @unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""       )
      def 		_lowerCamelCase      (		self       ):
         pass






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


   A_     :				int			      =	hf_hub_download(
       repo_id="""hf-internal-testing/spaghetti-video"""  ,filename="""eating_spaghetti.npy"""  ,repo_type="""dataset"""     )
   A_     :				Optional[int]			      =	np.load(_lowerCAmelCase     )
   return list(_lowerCAmelCase     )







@require_torch
@require_vision
class    _UpperCAmelCase							(		unittest.TestCase     ):
      @cached_property
      def 		_lowerCamelCase      (		self       ):
         # logits were tested with a different mean and std, so we use the same here
         return (
             VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]      ,       image_std=[0.5, 0.5, 0.5]       )
             if is_vision_available()
             else None
         )
      @slow
      def 		_lowerCamelCase      (		self       ):
         A_     :				List[str]			      =	VideoMAEForVideoClassification.from_pretrained("""MCG-NJU/videomae-base-finetuned-kinetics"""       ).to(
             a__       )

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

         # forward pass
         with torch.no_grad():
            A_     :				Any			      =	model(**a__       )

         # verify the logits
         A_     :				List[str]			      =	torch.Size((1, 400)       )
         self.assertEqual(outputs.logits.shape      ,       a__       )

         A_     :				Optional[Any]			      =	torch.tensor([0.3669, -0.0688, -0.2421]       ).to(a__       )

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






      @slow
      def 		_lowerCamelCase      (		self       ):
         A_     :				int			      =	VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""       ).to(a__       )

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

         # add boolean mask, indicating which patches to mask
         A_     :				Optional[Any]			      =	hf_hub_download(repo_id="""hf-internal-testing/bool-masked-pos"""      ,       filename="""bool_masked_pos.pt"""       )
         A_     :				Dict			      =	torch.load(a__       )

         # forward pass
         with torch.no_grad():
            A_     :				Optional[int]			      =	model(**a__       )

         # verify the logits
         A_     :				int			      =	torch.Size([1, 1408, 1536]       )
         A_     :				Union[str, Any]			      =	torch.tensor(
             [[0.7994, 0.9612, 0.8508], [0.7401, 0.8958, 0.8302], [0.5862, 0.7468, 0.7325]]      ,       device=a__       )
         self.assertEqual(outputs.logits.shape      ,       a__       )
         self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]      ,       a__      ,       atol=1E-4       )       )

         # verify the loss (`config.norm_pix_loss` = `True`)
         A_     :				Optional[Any]			      =	torch.tensor([0.5142]      ,       device=a__       )
         self.assertTrue(torch.allclose(outputs.loss      ,       a__      ,       atol=1E-4       )       )

         # verify the loss (`config.norm_pix_loss` = `False`)
         A_     :				Optional[int]			      =	VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""      ,       norm_pix_loss=a__       ).to(
             a__       )

         with torch.no_grad():
            A_     :				Optional[Any]			      =	model(**a__       )

         A_     :				List[Any]			      =	torch.tensor(torch.tensor([0.6469]       )      ,       device=a__       )
         self.assertTrue(torch.allclose(outputs.loss      ,       a__      ,       atol=1E-4       )       )






