



"""simple docstring"""





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       lowercase__      :
							'''simple docstring'''







							def __init__(	self						,							snake_case						,							snake_case=13						,							snake_case=10						,							snake_case=3						,							snake_case=2						,							snake_case=2						,							snake_case=2						,							snake_case=True						,							snake_case=True						,							snake_case=32						,							snake_case=5						,							snake_case=4						,							snake_case=37						,							snake_case="gelu"						,							snake_case=0.1						,							snake_case=0.1						,							snake_case=10						,							snake_case=0.02						,							snake_case=0.9						,							snake_case=None						,							)      ->				Union[str, Any]:
									_UpperCAmelCase							=		parent
									_UpperCAmelCase							=		batch_size
									_UpperCAmelCase							=		image_size
									_UpperCAmelCase							=		num_channels
									_UpperCAmelCase							=		patch_size
									_UpperCAmelCase							=		tubelet_size
									_UpperCAmelCase							=		num_frames
									_UpperCAmelCase							=		is_training
									_UpperCAmelCase							=		use_labels
									_UpperCAmelCase							=		hidden_size
									_UpperCAmelCase							=		num_hidden_layers
									_UpperCAmelCase							=		num_attention_heads
									_UpperCAmelCase							=		intermediate_size
									_UpperCAmelCase							=		hidden_act
									_UpperCAmelCase							=		hidden_dropout_prob
									_UpperCAmelCase							=		attention_probs_dropout_prob
									_UpperCAmelCase							=		type_sequence_label_size
									_UpperCAmelCase							=		initializer_range
									_UpperCAmelCase							=		mask_ratio
									_UpperCAmelCase							=		scope

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

									# use this variable to define bool_masked_pos
									_UpperCAmelCase							=		int(mask_ratio * self.seq_length    )






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

									_UpperCAmelCase							=		None
									if self.use_labels:
											_UpperCAmelCase							=		ids_tensor([self.batch_size]						,							self.type_sequence_label_size    )

									_UpperCAmelCase							=		self.get_config()

									return config, pixel_values, labels






							def 							lowerCamelCase_ (	self    )      ->				Union[str, Any]:
									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=lowerCAmelCase__						,							initializer_range=self.initializer_range						,							)






							def 							lowerCamelCase_ (	self						,							snake_case						,							snake_case						,							snake_case    )      ->				List[str]:
									_UpperCAmelCase							=		VideoMAEModel(config=lowerCAmelCase__    )
									model.to(lowerCAmelCase__    )
									model.eval()
									_UpperCAmelCase							=		model(lowerCAmelCase__    )
									self.parent.assertEqual(result.last_hidden_state.shape						,							(self.batch_size, self.seq_length, self.hidden_size)    )






							def 							lowerCamelCase_ (	self						,							snake_case						,							snake_case						,							snake_case    )      ->				str:
									_UpperCAmelCase							=		VideoMAEForPreTraining(lowerCAmelCase__    )
									model.to(lowerCAmelCase__    )
									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
									_UpperCAmelCase							=		torch.ones((self.num_masks,)    )
									_UpperCAmelCase							=		torch.cat([mask, torch.zeros(self.seq_length - mask.size(0    )    )]    )
									_UpperCAmelCase							=		mask.expand(self.batch_size						,							-1    ).bool()

									_UpperCAmelCase							=		model(lowerCAmelCase__						,							lowerCAmelCase__    )
									# model only returns predictions for masked patches
									_UpperCAmelCase							=		mask.sum().item()
									_UpperCAmelCase							=		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    )      ->				Union[str, Any]:
									_UpperCAmelCase							=		self.prepare_config_and_inputs()
									_UpperCAmelCase			,       _UpperCAmelCase			,       _UpperCAmelCase							=		config_and_inputs
									_UpperCAmelCase							=		{'pixel_values': pixel_values}
									return config, inputs_dict







@require_torch
class       lowercase__      (   _UpperCAmelCase,       _UpperCAmelCase,       unittest.TestCase     ):
							'''simple docstring'''







							_UpperCAmelCase   					=       (
							    (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
							)
							_UpperCAmelCase   					=       (
							    {'''feature-extraction''': VideoMAEModel, '''video-classification''': VideoMAEForVideoClassification}
							    if is_torch_available()
							    else {}
							)

							_UpperCAmelCase   					=       False
							_UpperCAmelCase   					=       False
							_UpperCAmelCase   					=       False
							_UpperCAmelCase   					=       False






							def 							lowerCamelCase_ (	self    )      ->				List[Any]:
									_UpperCAmelCase							=		VideoMAEModelTester(self    )
									_UpperCAmelCase							=		ConfigTester(self						,							config_class=lowerCAmelCase__						,							has_text_modality=lowerCAmelCase__						,							hidden_size=37    )






							def 							lowerCamelCase_ (	self						,							snake_case						,							snake_case						,							snake_case=False    )      ->				Optional[Any]:
									_UpperCAmelCase							=		copy.deepcopy(lowerCAmelCase__    )

									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
											_UpperCAmelCase							=		torch.ones((self.model_tester.num_masks,)    )
											_UpperCAmelCase							=		torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0    )    )]    )
											_UpperCAmelCase							=		mask.expand(self.model_tester.batch_size						,							-1    ).bool()
											_UpperCAmelCase							=		bool_masked_pos.to(lowerCAmelCase__    )

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

									return inputs_dict






							def 							lowerCamelCase_ (	self    )      ->				str:
									self.config_tester.run_common_tests()






							@unittest.skip(reason='VideoMAE does not use inputs_embeds'    )
							def 							lowerCamelCase_ (	self    )      ->				Optional[int]:
									pass






							def 							lowerCamelCase_ (	self    )      ->				Tuple:
									_UpperCAmelCase			,       _UpperCAmelCase							=		self.model_tester.prepare_config_and_inputs_for_common()

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






							def 							lowerCamelCase_ (	self    )      ->				Dict:
									_UpperCAmelCase			,       _UpperCAmelCase							=		self.model_tester.prepare_config_and_inputs_for_common()

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

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






							def 							lowerCamelCase_ (	self    )      ->				Dict:
									_UpperCAmelCase							=		self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_model(*lowerCAmelCase__    )






							def 							lowerCamelCase_ (	self    )      ->				Dict:
									_UpperCAmelCase							=		self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_pretraining(*lowerCAmelCase__    )






							@slow
							def 							lowerCamelCase_ (	self    )      ->				Dict:


									for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											_UpperCAmelCase							=		VideoMAEModel.from_pretrained(lowerCAmelCase__    )
											self.assertIsNotNone(lowerCAmelCase__    )






							def 							lowerCamelCase_ (	self    )      ->				Tuple:


									if not self.has_attentions:
											pass

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

											for model_class in self.all_model_classes:
													_UpperCAmelCase							=		self.model_tester.seq_length - self.model_tester.num_masks
													_UpperCAmelCase							=		(
													    num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
													)

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

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

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

													_UpperCAmelCase							=		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 							lowerCamelCase_ (	self    )      ->				Dict:


									def check_hidden_states_output(snake_case						,							snake_case						,							snake_case    ):
											_UpperCAmelCase							=		model_class(lowerCAmelCase__    )
											model.to(lowerCAmelCase__    )
											model.eval()

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

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

											_UpperCAmelCase							=		self.model_tester.seq_length - self.model_tester.num_masks
											_UpperCAmelCase							=		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]						,							)

									_UpperCAmelCase			,       _UpperCAmelCase							=		self.model_tester.prepare_config_and_inputs_for_common()

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

											# check that output_hidden_states also work using config
											del inputs_dict["output_hidden_states"]
											_UpperCAmelCase							=		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 							lowerCamelCase_ (	self    )      ->				str:
									pass







def 						UpperCAmelCase  (		):

		'''simple docstring'''


		_UpperCAmelCase							=		hf_hub_download(
		    repo_id='hf-internal-testing/spaghetti-video'  ,					filename='eating_spaghetti.npy'  ,					repo_type='dataset'							)
		_UpperCAmelCase							=		np.load(A							)
		return list(A							)






@require_torch
@require_vision
class       lowercase__      (   unittest.TestCase     ):
							'''simple docstring'''







							@cached_property
							def 							lowerCamelCase_ (	self    )      ->				Union[str, Any]:
									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    )      ->				Union[str, Any]:
									_UpperCAmelCase							=		VideoMAEForVideoClassification.from_pretrained('MCG-NJU/videomae-base-finetuned-kinetics'    ).to(
									    lowerCAmelCase__    )

									_UpperCAmelCase							=		self.default_image_processor
									_UpperCAmelCase							=		prepare_video()
									_UpperCAmelCase							=		image_processor(lowerCAmelCase__						,							return_tensors='pt'    ).to(lowerCAmelCase__    )

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

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

									_UpperCAmelCase							=		torch.tensor([0.3669, -0.0688, -0.2421]    ).to(lowerCAmelCase__    )

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






							@slow
							def 							lowerCamelCase_ (	self    )      ->				List[str]:
									_UpperCAmelCase							=		VideoMAEForPreTraining.from_pretrained('MCG-NJU/videomae-base-short'    ).to(lowerCAmelCase__    )

									_UpperCAmelCase							=		self.default_image_processor
									_UpperCAmelCase							=		prepare_video()
									_UpperCAmelCase							=		image_processor(lowerCAmelCase__						,							return_tensors='pt'    ).to(lowerCAmelCase__    )

									# add boolean mask, indicating which patches to mask
									_UpperCAmelCase							=		hf_hub_download(repo_id='hf-internal-testing/bool-masked-pos'						,							filename='bool_masked_pos.pt'    )
									_UpperCAmelCase							=		torch.load(lowerCAmelCase__    )

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

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

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

									# verify the loss (`config.norm_pix_loss` = `False`)
									_UpperCAmelCase							=		VideoMAEForPreTraining.from_pretrained('MCG-NJU/videomae-base-short'						,							norm_pix_loss=lowerCAmelCase__    ).to(
									    lowerCAmelCase__    )

									with torch.no_grad():
											_UpperCAmelCase							=		model(**lowerCAmelCase__    )

									_UpperCAmelCase							=		torch.tensor(torch.tensor([0.6469]    )						,							device=lowerCAmelCase__    )
									self.assertTrue(torch.allclose(outputs.loss						,							lowerCAmelCase__						,							atol=1E-4    )    )



def 							UpperCAmelCase		(       lowercase				,     lowercase				):







				"""simple docstring"""



				__lowercase					     =				word.split()

				def justify(lowercase				,     lowercase				,     lowercase				) -> str:
								__lowercase					     =				max_width - width
								__lowercase					     =				len(lowercase				)
								if len(lowercase				) == 1:
												# if there is only word in line
												# just insert overall_spaces_count for the remainder of line
												return line[0] + " " * overall_spaces_count
								else:
												__lowercase					     =				words_count - 1
												# num_spaces_between_words_list[i] : tells you to insert
												# num_spaces_between_words_list[i] spaces
												# after word on line[i]
												__lowercase					     =				spaces_to_insert_between_words * [
												    overall_spaces_count // spaces_to_insert_between_words
												]
												__lowercase					     =				(
												    overall_spaces_count % spaces_to_insert_between_words
												)
												# distribute spaces via round robin to the left words
												for i in range(lowercase				):
																num_spaces_between_words_list[i] += 1
												__lowercase					     =				[]
												for i in range(lowercase				):
																# add the word
																aligned_words_list.append(line[i]				)
																# add the spaces to insert
																aligned_words_list.append(num_spaces_between_words_list[i] * ''' '''				)
												# just add the last word to the sentence
												aligned_words_list.append(line[-1]				)
												# join the aligned words list to form a justified line
												return "".join(lowercase				)

				__lowercase					     =				[]
				__lowercase					     =				[]
				__lowercase					     =				0
				for word in words:
								if width + len(lowercase				) + len(lowercase				) <= max_width:
												# keep adding words until we can fill out max_width
												# width = sum of length of all words (without overall_spaces_count)
												# len(word) = length of current word
												# len(line) = number of overall_spaces_count to insert between words
												line.append(lowercase				)
												width += len(lowercase				)
								else:
												# justify the line and add it to result
												answer.append(justify(lowercase				,     lowercase				,     lowercase				)				)
												# reset new line and new width
												__lowercase     ,	__lowercase					     =				[word], len(lowercase				)
				__lowercase					     =				max_width - width - len(lowercase				)
				answer.append(''' '''.join(lowercase				) + (remaining_spaces + 1) * ''' '''				)
				return answer


if __name__ == "__main__":
		from doctest import testmod

		testmod()