


"""simple docstring"""




import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import TimesformerConfig
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,
	    TimesformerForVideoClassification,
	    TimesformerModel,
	)
	from transformers.models.timesformer.modeling_timesformer import TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
	from transformers import VideoMAEImageProcessor
class 					lowerCAmelCase	:




		'''simple docstring'''







		def __init__(	self						,       lowerCAmelCase__						,       lowerCAmelCase__=13						,       lowerCAmelCase__=10						,       lowerCAmelCase__=3						,       lowerCAmelCase__=2						,       lowerCAmelCase__=2						,       lowerCAmelCase__=True						,       lowerCAmelCase__=True						,       lowerCAmelCase__=32						,       lowerCAmelCase__=5						,       lowerCAmelCase__=4						,       lowerCAmelCase__=37						,       lowerCAmelCase__="gelu"						,       lowerCAmelCase__=0.1						,       lowerCAmelCase__=0.1						,       lowerCAmelCase__=10						,       lowerCAmelCase__=0.02						,       lowerCAmelCase__="divided_space_time"						,       lowerCAmelCase__=None						,       )     ->							Tuple:
								SCREAMING_SNAKE_CASE													=					parent
								SCREAMING_SNAKE_CASE													=					batch_size
								SCREAMING_SNAKE_CASE													=					image_size
								SCREAMING_SNAKE_CASE													=					num_channels
								SCREAMING_SNAKE_CASE													=					patch_size
								SCREAMING_SNAKE_CASE													=					num_frames
								SCREAMING_SNAKE_CASE													=					is_training
								SCREAMING_SNAKE_CASE													=					use_labels
								SCREAMING_SNAKE_CASE													=					hidden_size
								SCREAMING_SNAKE_CASE													=					num_hidden_layers
								SCREAMING_SNAKE_CASE													=					num_attention_heads
								SCREAMING_SNAKE_CASE													=					intermediate_size
								SCREAMING_SNAKE_CASE													=					hidden_act
								SCREAMING_SNAKE_CASE													=					hidden_dropout_prob
								SCREAMING_SNAKE_CASE													=					attention_probs_dropout_prob
								SCREAMING_SNAKE_CASE													=					attention_type
								SCREAMING_SNAKE_CASE													=					initializer_range
								SCREAMING_SNAKE_CASE													=					scope
								SCREAMING_SNAKE_CASE													=					num_labels

								# in TimeSformer, the number of spatial tokens equals num_frames * num_patches per frame + 1 CLS token
								SCREAMING_SNAKE_CASE													=					(image_size // patch_size) ** 2
								SCREAMING_SNAKE_CASE													=					(num_frames) * self.num_patches_per_frame + 1







		def 							__A						(	self				)     ->							Optional[int]:
								SCREAMING_SNAKE_CASE													=					floats_tensor(
								    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]				)

								SCREAMING_SNAKE_CASE													=					None
								if self.use_labels:
														SCREAMING_SNAKE_CASE													=					ids_tensor([self.batch_size]						,       self.num_labels				)

								SCREAMING_SNAKE_CASE													=					self.get_config()

								return config, pixel_values, labels







		def 							__A						(	self				)     ->							int:
								SCREAMING_SNAKE_CASE													=					TimesformerConfig(
								    image_size=self.image_size						,       patch_size=self.patch_size						,       num_channels=self.num_channels						,       num_frames=self.num_frames						,       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						,       attention_type=self.attention_type						,       )
								SCREAMING_SNAKE_CASE													=					self.num_labels
								return config







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							Tuple:
								SCREAMING_SNAKE_CASE													=					TimesformerModel(config=lowerCAmelCase__				)
								model.to(lowerCAmelCase__				)
								model.eval()
								SCREAMING_SNAKE_CASE													=					model(lowerCAmelCase__				)
								self.parent.assertEqual(result.last_hidden_state.shape						,       (self.batch_size, self.seq_length, self.hidden_size)				)







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							int:
								SCREAMING_SNAKE_CASE													=					TimesformerForVideoClassification(lowerCAmelCase__				)
								model.to(lowerCAmelCase__				)
								model.eval()

								SCREAMING_SNAKE_CASE													=					model(lowerCAmelCase__				)

								# verify the logits shape
								SCREAMING_SNAKE_CASE													=					torch.Size((self.batch_size, self.num_labels)				)
								self.parent.assertEqual(result.logits.shape						,       lowerCAmelCase__				)







		def 							__A						(	self				)     ->							Optional[int]:
								SCREAMING_SNAKE_CASE													=					self.prepare_config_and_inputs()
								SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE													=					config_and_inputs
								SCREAMING_SNAKE_CASE													=					{'pixel_values': pixel_values}
								return config, inputs_dict





@require_torch
class 					lowerCAmelCase	(					lowerCamelCase_	,    lowerCamelCase_	,    unittest.TestCase    ):




		'''simple docstring'''







		SCREAMING_SNAKE_CASE_						:     Optional[Any]								=     (TimesformerModel, TimesformerForVideoClassification) if is_torch_available() else ()
		SCREAMING_SNAKE_CASE_						:     Dict								=     (
		    {"""feature-extraction""": TimesformerModel, """video-classification""": TimesformerForVideoClassification}
		    if is_torch_available()
		    else {}
		)

		SCREAMING_SNAKE_CASE_						:     Any								=     False
		SCREAMING_SNAKE_CASE_						:     int								=     False
		SCREAMING_SNAKE_CASE_						:     List[Any]								=     False
		SCREAMING_SNAKE_CASE_						:     str								=     False







		def 							__A						(	self				)     ->							str:
								SCREAMING_SNAKE_CASE													=					TimesformerModelTester(self				)
								SCREAMING_SNAKE_CASE													=					ConfigTester(
								    self						,       config_class=lowerCAmelCase__						,       has_text_modality=lowerCAmelCase__						,       hidden_size=37				)







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__=False				)     ->							Optional[Any]:
								SCREAMING_SNAKE_CASE													=					copy.deepcopy(lowerCAmelCase__				)

								if return_labels:
														if model_class in get_values(lowerCAmelCase__				):
																				SCREAMING_SNAKE_CASE													=					torch.zeros(
																				    self.model_tester.batch_size						,       dtype=torch.long						,       device=lowerCAmelCase__				)

								return inputs_dict







		def 							__A						(	self				)     ->							Union[str, Any]:
								self.config_tester.run_common_tests()







		@unittest.skip(reason='TimeSformer does not use inputs_embeds'				)
		def 							__A						(	self				)     ->							Tuple:
								pass







		def 							__A						(	self				)     ->							List[str]:
								SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE													=					self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
														SCREAMING_SNAKE_CASE													=					model_class(lowerCAmelCase__				)
														self.assertIsInstance(model.get_input_embeddings()						,       (nn.Module)				)
														SCREAMING_SNAKE_CASE													=					model.get_output_embeddings()
														self.assertTrue(x is None or isinstance(lowerCAmelCase__						,       nn.Linear				)				)







		def 							__A						(	self				)     ->							int:
								SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE													=					self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
														SCREAMING_SNAKE_CASE													=					model_class(lowerCAmelCase__				)
														SCREAMING_SNAKE_CASE													=					inspect.signature(model.forward				)
														# signature.parameters is an OrderedDict => so arg_names order is deterministic
														SCREAMING_SNAKE_CASE													=					[*signature.parameters.keys()]

														SCREAMING_SNAKE_CASE													=					['pixel_values']
														self.assertListEqual(arg_names[:1]						,       lowerCAmelCase__				)







		def 							__A						(	self				)     ->							Optional[Any]:
								SCREAMING_SNAKE_CASE													=					self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*lowerCAmelCase__				)







		def 							__A						(	self				)     ->							str:
								SCREAMING_SNAKE_CASE													=					self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_video_classification(*lowerCAmelCase__				)







		@slow
		def 							__A						(	self				)     ->							List[str]:
								for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														SCREAMING_SNAKE_CASE													=					TimesformerModel.from_pretrained(lowerCAmelCase__				)
														self.assertIsNotNone(lowerCAmelCase__				)







		def 							__A						(	self				)     ->							Tuple:
								if not self.has_attentions:
														pass

								else:
														SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE													=					self.model_tester.prepare_config_and_inputs_for_common()
														SCREAMING_SNAKE_CASE													=					True

														for model_class in self.all_model_classes:
																				SCREAMING_SNAKE_CASE													=					self.model_tester.seq_length
																				SCREAMING_SNAKE_CASE													=					self.model_tester.num_frames

																				SCREAMING_SNAKE_CASE													=					True
																				SCREAMING_SNAKE_CASE													=					False
																				SCREAMING_SNAKE_CASE													=					True
																				SCREAMING_SNAKE_CASE													=					model_class(lowerCAmelCase__				)
																				model.to(lowerCAmelCase__				)
																				model.eval()
																				with torch.no_grad():
																										SCREAMING_SNAKE_CASE													=					model(**self._prepare_for_class(lowerCAmelCase__						,       lowerCAmelCase__				)				)
																				SCREAMING_SNAKE_CASE													=					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"]
																				SCREAMING_SNAKE_CASE													=					True
																				SCREAMING_SNAKE_CASE													=					model_class(lowerCAmelCase__				)
																				model.to(lowerCAmelCase__				)
																				model.eval()
																				with torch.no_grad():
																										SCREAMING_SNAKE_CASE													=					model(**self._prepare_for_class(lowerCAmelCase__						,       lowerCAmelCase__				)				)
																				SCREAMING_SNAKE_CASE													=					outputs.attentions
																				self.assertEqual(len(lowerCAmelCase__				)						,       self.model_tester.num_hidden_layers				)

																				# attentions has shape (batch_size x num_frames) x num_heads x (num_patches per frame + 1) x (num_patches per frame + 1)
																				self.assertListEqual(
																				    list(attentions[0].shape[-3:]				)						,       [self.model_tester.num_attention_heads, seq_len // num_frames + 1, seq_len // num_frames + 1]						,       )
																				SCREAMING_SNAKE_CASE													=					len(lowerCAmelCase__				)

																				# Check attention is always last and order is fine
																				SCREAMING_SNAKE_CASE													=					True
																				SCREAMING_SNAKE_CASE													=					True
																				SCREAMING_SNAKE_CASE													=					model_class(lowerCAmelCase__				)
																				model.to(lowerCAmelCase__				)
																				model.eval()
																				with torch.no_grad():
																										SCREAMING_SNAKE_CASE													=					model(**self._prepare_for_class(lowerCAmelCase__						,       lowerCAmelCase__				)				)

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

																				SCREAMING_SNAKE_CASE													=					outputs.attentions

																				self.assertEqual(len(lowerCAmelCase__				)						,       self.model_tester.num_hidden_layers				)

																				# attentions has shape (batch_size x num_frames) x num_heads x (num_patches per frame + 1) x (num_patches per frame + 1)
																				self.assertListEqual(
																				    list(self_attentions[0].shape[-3:]				)						,       [self.model_tester.num_attention_heads, seq_len // num_frames + 1, seq_len // num_frames + 1]						,       )







		def 							__A						(	self				)     ->							str:
								def check_hidden_states_output(lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__				):
														SCREAMING_SNAKE_CASE													=					model_class(lowerCAmelCase__				)
														model.to(lowerCAmelCase__				)
														model.eval()

														with torch.no_grad():
																				SCREAMING_SNAKE_CASE													=					model(**self._prepare_for_class(lowerCAmelCase__						,       lowerCAmelCase__				)				)

														SCREAMING_SNAKE_CASE													=					outputs.hidden_states
														SCREAMING_SNAKE_CASE													=					self.model_tester.num_hidden_layers + 1
														self.assertEqual(len(lowerCAmelCase__				)						,       lowerCAmelCase__				)

														SCREAMING_SNAKE_CASE													=					self.model_tester.seq_length

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

								SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE													=					self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
														SCREAMING_SNAKE_CASE													=					True
														check_hidden_states_output(lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__				)

														# check that output_hidden_states also work using config
														del inputs_dict["output_hidden_states"]
														SCREAMING_SNAKE_CASE													=					True

														check_hidden_states_output(lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__				)





def     lowercase			()   ->						Optional[Any]:
						SCREAMING_SNAKE_CASE													=					hf_hub_download(
						    repo_id='hf-internal-testing/spaghetti-video' ,			filename='eating_spaghetti.npy' ,			repo_type='dataset'  )
						SCREAMING_SNAKE_CASE													=					np.load(SCREAMING_SNAKE_CASE_  )
						return list(SCREAMING_SNAKE_CASE_  )


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




		'''simple docstring'''







		@cached_property
		def 							__A						(	self				)     ->							Union[str, Any]:
								# 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 							__A						(	self				)     ->							Any:
								SCREAMING_SNAKE_CASE													=					TimesformerForVideoClassification.from_pretrained('facebook/timesformer-base-finetuned-k400'				).to(
								    lowerCAmelCase__				)

								SCREAMING_SNAKE_CASE													=					self.default_image_processor
								SCREAMING_SNAKE_CASE													=					prepare_video()
								SCREAMING_SNAKE_CASE													=					image_processor(video[:8]						,       return_tensors='pt'				).to(lowerCAmelCase__				)

								# forward pass
								with torch.no_grad():
														SCREAMING_SNAKE_CASE													=					model(**lowerCAmelCase__				)

								# verify the logits
								SCREAMING_SNAKE_CASE													=					torch.Size((1, 400)				)
								self.assertEqual(outputs.logits.shape						,       lowerCAmelCase__				)

								SCREAMING_SNAKE_CASE													=					torch.tensor([-0.30_16, -0.77_13, -0.42_05]				).to(lowerCAmelCase__				)

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






"""simple docstring"""




import importlib

import torch
import yaml
from omegaconf import OmegaConf
from taming.models.vqgan import VQModel


def     lowercase			(SCREAMING_SNAKE_CASE_		:				int ,			SCREAMING_SNAKE_CASE_		:				Any=False  )   ->						str:
						SCREAMING_SNAKE_CASE													=					OmegaConf.load(SCREAMING_SNAKE_CASE_  )
						if display:
												print(yaml.dump(OmegaConf.to_container(SCREAMING_SNAKE_CASE_  )  )  )
						return config


def     lowercase			(SCREAMING_SNAKE_CASE_		:				Tuple ,			SCREAMING_SNAKE_CASE_		:				Tuple=None ,			SCREAMING_SNAKE_CASE_		:				Tuple=None  )   ->						Any:
						if conf_path is None:
												SCREAMING_SNAKE_CASE													=					'./model_checkpoints/vqgan_only.yaml'
						SCREAMING_SNAKE_CASE													=					load_config(SCREAMING_SNAKE_CASE_ ,			display=SCREAMING_SNAKE_CASE_  )
						SCREAMING_SNAKE_CASE													=					VQModel(**config.model.params  )
						if ckpt_path is None:
												SCREAMING_SNAKE_CASE													=					'./model_checkpoints/vqgan_only.pt'
						SCREAMING_SNAKE_CASE													=					torch.load(SCREAMING_SNAKE_CASE_ ,			map_location=SCREAMING_SNAKE_CASE_  )
						if ".ckpt" in ckpt_path:
												SCREAMING_SNAKE_CASE													=					sd['state_dict']
						model.load_state_dict(SCREAMING_SNAKE_CASE_ ,			strict=SCREAMING_SNAKE_CASE_  )
						model.to(SCREAMING_SNAKE_CASE_  )
						del sd
						return model


def     lowercase			(SCREAMING_SNAKE_CASE_		:				List[str] ,			SCREAMING_SNAKE_CASE_		:				int  )   ->						List[Any]:
						SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE													=					model.encode(SCREAMING_SNAKE_CASE_  )
						print(F'VQGAN --- {model.__class__.__name__}: latent shape: {z.shape[2:]}'  )
						SCREAMING_SNAKE_CASE													=					model.decode(SCREAMING_SNAKE_CASE_  )
						return xrec


def     lowercase			(SCREAMING_SNAKE_CASE_		:				int ,			SCREAMING_SNAKE_CASE_		:				Any=False  )   ->						Dict:
						SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE													=					string.rsplit('.' ,			1  )
						if reload:
												SCREAMING_SNAKE_CASE													=					importlib.import_module(SCREAMING_SNAKE_CASE_  )
												importlib.reload(SCREAMING_SNAKE_CASE_  )
						return getattr(importlib.import_module(SCREAMING_SNAKE_CASE_ ,			package=SCREAMING_SNAKE_CASE_  ) ,			cls  )


def     lowercase			(SCREAMING_SNAKE_CASE_		:				Any  )   ->						Dict:
						if "target" not in config:
												raise KeyError('Expected key `target` to instantiate.'  )
						return get_obj_from_str(config['target']  )(**config.get('params' ,			{}  )  )


def     lowercase			(SCREAMING_SNAKE_CASE_		:				Tuple ,			SCREAMING_SNAKE_CASE_		:				Any ,			SCREAMING_SNAKE_CASE_		:				List[Any]=True ,			SCREAMING_SNAKE_CASE_		:				int=True  )   ->						Any:
						SCREAMING_SNAKE_CASE													=					instantiate_from_config(SCREAMING_SNAKE_CASE_  )
						if sd is not None:
												model.load_state_dict(SCREAMING_SNAKE_CASE_  )
						if gpu:
												model.cuda()
						if eval_mode:
												model.eval()
						return {"model": model}


def     lowercase			(SCREAMING_SNAKE_CASE_		:				str ,			SCREAMING_SNAKE_CASE_		:				List[Any] ,			SCREAMING_SNAKE_CASE_		:				Optional[Any] ,			SCREAMING_SNAKE_CASE_		:				List[str]  )   ->						Any:
						# load the specified checkpoint
						if ckpt:
												SCREAMING_SNAKE_CASE													=					torch.load(SCREAMING_SNAKE_CASE_ ,			map_location='cpu'  )
												SCREAMING_SNAKE_CASE													=					pl_sd['global_step']
												print(F'loaded model from global step {global_step}.'  )
						else:
												SCREAMING_SNAKE_CASE													=					{'state_dict': None}
												SCREAMING_SNAKE_CASE													=					None
						SCREAMING_SNAKE_CASE													=					load_model_from_config(config.model ,			pl_sd['state_dict'] ,			gpu=SCREAMING_SNAKE_CASE_ ,			eval_mode=SCREAMING_SNAKE_CASE_  )['model']
						return model, global_step


