
import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import VivitConfig
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


if is_torch_available():
		import torch
		from torch import nn

		from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VivitForVideoClassification, VivitModel
		from transformers.models.vivit.modeling_vivit import VIVIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
		from transformers import VivitImageProcessor


class 						__magic_name__				:





		def __init__(							self      :		Dict  ,					lowerCamelCase__      :		Any  ,					lowerCamelCase__      :		List[str]=2  ,					lowerCamelCase__      :		Optional[int]=True  ,					lowerCamelCase__      :		List[Any]=True  ,					lowerCamelCase__      :		Optional[Any]=1_0  ,					lowerCamelCase__      :		Optional[int]=1_0  ,					lowerCamelCase__      :		int=8  ,					lowerCamelCase__      :		str=[2, 4, 4]  ,					lowerCamelCase__      :		List[str]=3  ,					lowerCamelCase__      :		Any=7_6_8  ,					lowerCamelCase__      :		Optional[int]=5  ,					lowerCamelCase__      :		Dict=4  ,					lowerCamelCase__      :		List[Any]=3_7  ,					lowerCamelCase__      :		Optional[Any]="gelu_fast"  ,					lowerCamelCase__      :		List[str]=0.0  ,					lowerCamelCase__      :		Optional[int]=0.0  ,					lowerCamelCase__      :		int=0.0_2  ,					lowerCamelCase__      :		Dict=1E-06  ,					lowerCamelCase__      :		Tuple=True  ,					lowerCamelCase__      :		Optional[Any]=None  ,					):
							lowerCAmelCase						:   List[str]							=       parent
							lowerCAmelCase						:   Union[str, Any]							=       batch_size
							lowerCAmelCase						:   str							=       is_training
							lowerCAmelCase						:   Union[str, Any]							=       use_labels
							lowerCAmelCase						:   List[Any]							=       num_labels
							lowerCAmelCase						:   Optional[int]							=       image_size
							lowerCAmelCase						:   Tuple							=       num_frames
							lowerCAmelCase						:   Union[str, Any]							=       tubelet_size
							lowerCAmelCase						:   Optional[int]							=       num_channels
							lowerCAmelCase						:   int							=       hidden_size
							lowerCAmelCase						:   Union[str, Any]							=       num_hidden_layers
							lowerCAmelCase						:   List[Any]							=       num_attention_heads
							lowerCAmelCase						:   str							=       intermediate_size
							lowerCAmelCase						:   str							=       hidden_act
							lowerCAmelCase						:   Tuple							=       hidden_dropout_prob
							lowerCAmelCase						:   Tuple							=       attention_probs_dropout_prob
							lowerCAmelCase						:   Optional[int]							=       initializer_range
							lowerCAmelCase						:   List[str]							=       layer_norm_eps
							lowerCAmelCase						:   Optional[Any]							=       qkv_bias
							lowerCAmelCase						:   str							=       scope

							lowerCAmelCase						:   Any							=       (
							    (self.image_size // self.tubelet_size[2])
							    * (self.image_size // self.tubelet_size[1])
							    * (self.num_frames // self.tubelet_size[0])
							) + 1  # CLS token





		def  _A							(							self      :		int			):
							lowerCAmelCase						:   Tuple							=       floats_tensor(
							    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]			)

							lowerCAmelCase						:   List[str]							=       None
							if self.use_labels:
												lowerCAmelCase						:   Optional[Any]							=       ids_tensor([self.batch_size]  ,					self.num_labels			)

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

							return config, pixel_values, labels





		def  _A							(							self      :		Dict			):
							lowerCAmelCase						:   Optional[int]							=       VivitConfig(
							    num_frames=self.num_frames  ,					image_size=self.image_size  ,					tubelet_size=self.tubelet_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  ,					initializer_range=self.initializer_range  ,					layer_norm_eps=self.layer_norm_eps  ,					qkv_bias=self.qkv_bias  ,					)
							lowerCAmelCase						:   str							=       self.num_labels
							return config





		def  _A							(							self      :		Dict  ,					lowerCamelCase__      :		int  ,					lowerCamelCase__      :		Any  ,					lowerCamelCase__      :		Optional[Any]			):
							lowerCAmelCase						:   List[str]							=       VivitModel(config=lowerCamelCase__			)
							model.to(lowerCamelCase__			)
							model.eval()
							lowerCAmelCase						:   Union[str, Any]							=       model(lowerCamelCase__			)
							self.parent.assertEqual(result.last_hidden_state.shape  ,					(self.batch_size, self.seq_length, self.hidden_size)			)





		def  _A							(							self      :		Dict  ,					lowerCamelCase__      :		Union[str, Any]  ,					lowerCamelCase__      :		str  ,					lowerCamelCase__      :		Tuple			):
							lowerCAmelCase						:   Union[str, Any]							=       VivitForVideoClassification(lowerCamelCase__			)
							model.to(lowerCamelCase__			)
							model.eval()

							lowerCAmelCase						:   Tuple							=       model(lowerCamelCase__			)

							# verify the logits shape
							lowerCAmelCase						:   Union[str, Any]							=       torch.Size((self.batch_size, self.num_labels)			)
							self.parent.assertEqual(result.logits.shape  ,					lowerCamelCase__			)





		def  _A							(							self      :		str			):
							lowerCAmelCase						:   Any							=       self.prepare_config_and_inputs()
							lowerCAmelCase		,							lowerCAmelCase		,							lowerCAmelCase						:   Union[str, Any]							=       config_and_inputs
							lowerCAmelCase						:   Any							=       {'''pixel_values''': pixel_values}
							return config, inputs_dict



@require_torch
class 						__magic_name__				(  snake_case,			unittest.TestCase      ):
		_lowerCAmelCase						=		(VivitModel, VivitForVideoClassification) if is_torch_available() else ()

		_lowerCAmelCase						=		False
		_lowerCAmelCase						=		False
		_lowerCAmelCase						=		False
		_lowerCAmelCase						=		False





		def  _A							(							self      :		List[Any]			):
							lowerCAmelCase						:   List[Any]							=       VivitModelTester(self			)
							lowerCAmelCase						:   List[Any]							=       ConfigTester(self  ,					config_class=lowerCamelCase__  ,					has_text_modality=lowerCamelCase__  ,					hidden_size=3_7			)





		def  _A							(							self      :		Dict  ,					lowerCamelCase__      :		Union[str, Any]  ,					lowerCamelCase__      :		Tuple  ,					lowerCamelCase__      :		List[str]=False			):
							lowerCAmelCase						:   Union[str, Any]							=       copy.deepcopy(lowerCamelCase__			)

							if return_labels:
												if model_class in get_values(lowerCamelCase__			):
																	lowerCAmelCase						:   Optional[Any]							=       torch.zeros(
																	    self.model_tester.batch_size  ,					dtype=torch.long  ,					device=lowerCamelCase__			)

							return inputs_dict





		def  _A							(							self      :		Dict			):
							self.config_tester.run_common_tests()





		@unittest.skip(reason='''Vivit does not use inputs_embeds'''			)
		def  _A							(							self      :		Optional[int]			):
							pass





		def  _A							(							self      :		Optional[int]			):
							lowerCAmelCase		,							lowerCAmelCase						:   Any							=       self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
												lowerCAmelCase						:   int							=       model_class(lowerCamelCase__			)
												self.assertIsInstance(model.get_input_embeddings()  ,					(nn.Module)			)
												lowerCAmelCase						:   Optional[Any]							=       model.get_output_embeddings()
												self.assertTrue(x is None or isinstance(lowerCamelCase__  ,					nn.Linear			)			)





		def  _A							(							self      :		List[str]			):
							lowerCAmelCase		,							lowerCAmelCase						:   List[str]							=       self.model_tester.prepare_config_and_inputs_for_common()

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

												lowerCAmelCase						:   List[str]							=       ['''pixel_values''', '''head_mask''']
												self.assertListEqual(arg_names[:2]  ,					lowerCamelCase__			)





		def  _A							(							self      :		List[Any]			):
							lowerCAmelCase						:   Optional[Any]							=       self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*lowerCamelCase__			)





		def  _A							(							self      :		Union[str, Any]			):
							lowerCAmelCase						:   int							=       self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_video_classification(*lowerCamelCase__			)





		@slow
		def  _A							(							self      :		Union[str, Any]			):
							for model_name in VIVIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
												lowerCAmelCase						:   Optional[Any]							=       VivitModel.from_pretrained(lowerCamelCase__			)
												self.assertIsNotNone(lowerCamelCase__			)





		def  _A							(							self      :		int			):
							lowerCAmelCase		,							lowerCAmelCase						:   Optional[Any]							=       self.model_tester.prepare_config_and_inputs_for_common()
							lowerCAmelCase						:   str							=       True

							for model_class in self.all_model_classes:
												lowerCAmelCase						:   Optional[Any]							=       self.model_tester.seq_length

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

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

												self.assertListEqual(
												    list(attentions[0].shape[-3:]			)  ,					[self.model_tester.num_attention_heads, seq_len, seq_len]  ,					)
												lowerCAmelCase						:   Optional[int]							=       len(lowerCamelCase__			)

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

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

												lowerCAmelCase						:   Optional[Any]							=       outputs.attentions

												self.assertEqual(len(lowerCamelCase__			)  ,					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  _A							(							self      :		Any			):
							def check_hidden_states_output(lowerCamelCase__      :		Union[str, Any]  ,					lowerCamelCase__      :		int  ,					lowerCamelCase__      :		Optional[Any]			):
												lowerCAmelCase						:   Optional[Any]							=       model_class(lowerCamelCase__			)
												model.to(lowerCamelCase__			)
												model.eval()

												with torch.no_grad():
																	lowerCAmelCase						:   Optional[int]							=       model(**self._prepare_for_class(lowerCamelCase__  ,					lowerCamelCase__			)			)

												lowerCAmelCase						:   List[Any]							=       outputs.hidden_states
												lowerCAmelCase						:   List[Any]							=       self.model_tester.num_hidden_layers + 1
												self.assertEqual(len(lowerCamelCase__			)  ,					lowerCamelCase__			)

												lowerCAmelCase						:   Union[str, Any]							=       self.model_tester.seq_length

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

							lowerCAmelCase		,							lowerCAmelCase						:   Optional[int]							=       self.model_tester.prepare_config_and_inputs_for_common()

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

												# check that output_hidden_states also work using config
												del inputs_dict["output_hidden_states"]
												lowerCAmelCase						:   List[Any]							=       True

												check_hidden_states_output(lowerCamelCase__  ,					lowerCamelCase__  ,					lowerCamelCase__			)





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



def       UpperCAmelCase__							(    ):


					'''simple docstring'''

					lowerCAmelCase						:   Union[str, Any]							=       hf_hub_download(
					    repo_id='''hf-internal-testing/spaghetti-video'''							,					filename='''eating_spaghetti_32_frames.npy'''							,					repo_type='''dataset'''					)
					lowerCAmelCase						:   List[Any]							=       np.load(__magic_name__					)
					return list(__magic_name__					)


@require_torch
@require_vision
class 						__magic_name__				(  unittest.TestCase      ):





		@cached_property
		def  _A							(							self      :		int			):
							return VivitImageProcessor() if is_vision_available() else None





		@slow
		def  _A							(							self      :		List[Any]			):
							lowerCAmelCase						:   Union[str, Any]							=       VivitForVideoClassification.from_pretrained('''google/vivit-b-16x2-kinetics400'''			).to(lowerCamelCase__			)

							lowerCAmelCase						:   List[str]							=       self.default_image_processor
							lowerCAmelCase						:   Union[str, Any]							=       prepare_video()
							lowerCAmelCase						:   str							=       image_processor(lowerCamelCase__  ,					return_tensors='''pt'''			).to(lowerCamelCase__			)

							# forward pass
							with torch.no_grad():
												lowerCAmelCase						:   int							=       model(**lowerCamelCase__			)

							# verify the logits
							lowerCAmelCase						:   List[Any]							=       torch.Size((1, 4_0_0)			)
							self.assertEqual(outputs.logits.shape  ,					lowerCamelCase__			)

							# taken from original model
							lowerCAmelCase						:   Tuple							=       torch.tensor([-1.0_5_4_3, 2.0_7_6_4, -0.2_1_0_4, 0.4_4_3_9, -0.9_6_5_8]			).to(lowerCamelCase__			)

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


import gc
import unittest

import numpy as np
import torch

from diffusers import RePaintPipeline, RePaintScheduler, UNetaDModel
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    load_image,
    load_numpy,
    nightly,
    require_torch_gpu,
    skip_mps,
    torch_device,
)

from ..pipeline_params import IMAGE_INPAINTING_BATCH_PARAMS, IMAGE_INPAINTING_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


class 						__magic_name__				(  snake_case,			unittest.TestCase      ):
		_lowerCAmelCase						=		RePaintPipeline
		_lowerCAmelCase						=		IMAGE_INPAINTING_PARAMS - {"width", "height", "guidance_scale"}
		_lowerCAmelCase						=		PipelineTesterMixin.required_optional_params - {
		    "latents",
		    "num_images_per_prompt",
		    "callback",
		    "callback_steps",
		}
		_lowerCAmelCase						=		IMAGE_INPAINTING_BATCH_PARAMS
		_lowerCAmelCase						=		False





		def  _A							(							self      :		Optional[int]			):
							torch.manual_seed(0			)
							torch.manual_seed(0			)
							lowerCAmelCase						:   Dict							=       UNetaDModel(
							    block_out_channels=(3_2, 6_4)  ,					layers_per_block=2  ,					sample_size=3_2  ,					in_channels=3  ,					out_channels=3  ,					down_block_types=('''DownBlock2D''', '''AttnDownBlock2D''')  ,					up_block_types=('''AttnUpBlock2D''', '''UpBlock2D''')  ,					)
							lowerCAmelCase						:   List[Any]							=       RePaintScheduler()
							lowerCAmelCase						:   Tuple							=       {'''unet''': unet, '''scheduler''': scheduler}
							return components





		def  _A							(							self      :		Any  ,					lowerCamelCase__      :		Any  ,					lowerCamelCase__      :		int=0			):
							if str(lowerCamelCase__			).startswith('''mps'''			):
												lowerCAmelCase						:   Union[str, Any]							=       torch.manual_seed(lowerCamelCase__			)
							else:
												lowerCAmelCase						:   Optional[Any]							=       torch.Generator(device=lowerCamelCase__			).manual_seed(lowerCamelCase__			)
							lowerCAmelCase						:   List[str]							=       np.random.RandomState(lowerCamelCase__			).standard_normal((1, 3, 3_2, 3_2)			)
							lowerCAmelCase						:   Tuple							=       torch.from_numpy(lowerCamelCase__			).to(device=lowerCamelCase__  ,					dtype=torch.floataa			)
							lowerCAmelCase						:   Tuple							=       (image > 0).to(device=lowerCamelCase__  ,					dtype=torch.floataa			)
							lowerCAmelCase						:   List[Any]							=       {
							    '''image''': image,
							    '''mask_image''': mask,
							    '''generator''': generator,
							    '''num_inference_steps''': 5,
							    '''eta''': 0.0,
							    '''jump_length''': 2,
							    '''jump_n_sample''': 2,
							    '''output_type''': '''numpy''',
							}
							return inputs





		def  _A							(							self      :		Union[str, Any]			):
							lowerCAmelCase						:   List[str]							=       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
							lowerCAmelCase						:   Dict							=       self.get_dummy_components()
							lowerCAmelCase						:   str							=       RePaintPipeline(**lowerCamelCase__			)
							lowerCAmelCase						:   Union[str, Any]							=       sd_pipe.to(lowerCamelCase__			)
							sd_pipe.set_progress_bar_config(disable=lowerCamelCase__			)

							lowerCAmelCase						:   List[Any]							=       self.get_dummy_inputs(lowerCamelCase__			)
							lowerCAmelCase						:   int							=       sd_pipe(**lowerCamelCase__			).images
							lowerCAmelCase						:   Dict							=       image[0, -3:, -3:, -1]

							assert image.shape == (1, 3_2, 3_2, 3)
							lowerCAmelCase						:   List[str]							=       np.array([1.0_0_0_0, 0.5_4_2_6, 0.5_4_9_7, 0.2_2_0_0, 1.0_0_0_0, 1.0_0_0_0, 0.5_6_2_3, 1.0_0_0_0, 0.6_2_7_4]			)

							assert np.abs(image_slice.flatten() - expected_slice			).max() < 1E-3





		@skip_mps
		def  _A							(							self      :		Optional[Any]			):
							return super().test_save_load_local()





		@unittest.skip('''non-deterministic pipeline'''			)
		def  _A							(							self      :		Any			):
							return super().test_inference_batch_single_identical()





		@skip_mps
		def  _A							(							self      :		List[Any]			):
							return super().test_dict_tuple_outputs_equivalent()





		@skip_mps
		def  _A							(							self      :		Dict			):
							return super().test_save_load_optional_components()





		@skip_mps
		def  _A							(							self      :		Tuple			):
							return super().test_attention_slicing_forward_pass()



@nightly
@require_torch_gpu
class 						__magic_name__				(  unittest.TestCase      ):





		def  _A							(							self      :		Optional[Any]			):
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()





		def  _A							(							self      :		Union[str, Any]			):
							lowerCAmelCase						:   str							=       load_image(
							    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/'''
							    '''repaint/celeba_hq_256.png'''			)
							lowerCAmelCase						:   Tuple							=       load_image(
							    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/repaint/mask_256.png'''			)
							lowerCAmelCase						:   int							=       load_numpy(
							    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/'''
							    '''repaint/celeba_hq_256_result.npy'''			)

							lowerCAmelCase						:   List[str]							=       '''google/ddpm-ema-celebahq-256'''
							lowerCAmelCase						:   List[str]							=       UNetaDModel.from_pretrained(lowerCamelCase__			)
							lowerCAmelCase						:   str							=       RePaintScheduler.from_pretrained(lowerCamelCase__			)

							lowerCAmelCase						:   Optional[int]							=       RePaintPipeline(unet=lowerCamelCase__  ,					scheduler=lowerCamelCase__			).to(lowerCamelCase__			)
							repaint.set_progress_bar_config(disable=lowerCamelCase__			)
							repaint.enable_attention_slicing()

							lowerCAmelCase						:   str							=       torch.manual_seed(0			)
							lowerCAmelCase						:   Dict							=       repaint(
							    lowerCamelCase__  ,					lowerCamelCase__  ,					num_inference_steps=2_5_0  ,					eta=0.0  ,					jump_length=1_0  ,					jump_n_sample=1_0  ,					generator=lowerCamelCase__  ,					output_type='''np'''  ,					)
							lowerCAmelCase						:   str							=       output.images[0]

							assert image.shape == (2_5_6, 2_5_6, 3)
							assert np.abs(expected_image - image			).mean() < 1E-2
