

"""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      _UpperCAmelCase      :
			def __init__(		self   ,		_A   ,		_A=13   ,		_A=10   ,		_A=3   ,		_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="divided_space_time"   ,		_A=None   ,		)     ->     Optional[int]:






				'''simple docstring'''







				_UpperCAmelCase     :				str          =       parent
				_UpperCAmelCase     :				Tuple          =       batch_size
				_UpperCAmelCase     :				Optional[int]          =       image_size
				_UpperCAmelCase     :				Optional[Any]          =       num_channels
				_UpperCAmelCase     :				Dict          =       patch_size
				_UpperCAmelCase     :				Union[str, Any]          =       num_frames
				_UpperCAmelCase     :				int          =       is_training
				_UpperCAmelCase     :				int          =       use_labels
				_UpperCAmelCase     :				Union[str, Any]          =       hidden_size
				_UpperCAmelCase     :				str          =       num_hidden_layers
				_UpperCAmelCase     :				Optional[int]          =       num_attention_heads
				_UpperCAmelCase     :				Dict          =       intermediate_size
				_UpperCAmelCase     :				Optional[int]          =       hidden_act
				_UpperCAmelCase     :				Union[str, Any]          =       hidden_dropout_prob
				_UpperCAmelCase     :				Dict          =       attention_probs_dropout_prob
				_UpperCAmelCase     :				List[Any]          =       attention_type
				_UpperCAmelCase     :				Union[str, Any]          =       initializer_range
				_UpperCAmelCase     :				Dict          =       scope
				_UpperCAmelCase     :				List[Any]          =       num_labels

				# in TimeSformer, the number of spatial tokens equals num_frames * num_patches per frame + 1 CLS token
				_UpperCAmelCase     :				Optional[Any]          =       (image_size // patch_size) ** 2
				_UpperCAmelCase     :				int          =       (num_frames) * self.num_patches_per_frame + 1
			def 				__snake_case			(		self  )     ->     List[str]:






				'''simple docstring'''







				_UpperCAmelCase     :				int          =       floats_tensor(
				    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]  )

				_UpperCAmelCase     :				Dict          =       None
				if self.use_labels:
					_UpperCAmelCase     :				List[Any]          =       ids_tensor([self.batch_size]   ,		self.num_labels  )

				_UpperCAmelCase     :				int          =       self.get_config()

				return config, pixel_values, labels
			def 				__snake_case			(		self  )     ->     str:






				'''simple docstring'''







				_UpperCAmelCase     :				str          =       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   ,		)
				_UpperCAmelCase     :				Union[str, Any]          =       self.num_labels
				return config
			def 				__snake_case			(		self   ,		_A   ,		_A   ,		_A  )     ->     Tuple:






				'''simple docstring'''







				_UpperCAmelCase     :				Optional[int]          =       TimesformerModel(config=__lowerCAmelCase  )
				model.to(__lowerCAmelCase  )
				model.eval()
				_UpperCAmelCase     :				Any          =       model(__lowerCAmelCase  )
				self.parent.assertEqual(result.last_hidden_state.shape   ,		(self.batch_size, self.seq_length, self.hidden_size)  )
			def 				__snake_case			(		self   ,		_A   ,		_A   ,		_A  )     ->     Tuple:






				'''simple docstring'''







				_UpperCAmelCase     :				Any          =       TimesformerForVideoClassification(__lowerCAmelCase  )
				model.to(__lowerCAmelCase  )
				model.eval()

				_UpperCAmelCase     :				Tuple          =       model(__lowerCAmelCase  )

				# verify the logits shape
				_UpperCAmelCase     :				str          =       torch.Size((self.batch_size, self.num_labels)  )
				self.parent.assertEqual(result.logits.shape   ,		__lowerCAmelCase  )
			def 				__snake_case			(		self  )     ->     Optional[int]:






				'''simple docstring'''







				_UpperCAmelCase     :				Optional[int]          =       self.prepare_config_and_inputs()
				_UpperCAmelCase    , _UpperCAmelCase    , _UpperCAmelCase     :				Any          =       config_and_inputs
				_UpperCAmelCase     :				Optional[Any]          =       {"""pixel_values""": pixel_values}
				return config, inputs_dict






@require_torch
class      _UpperCAmelCase      (		__a     ,  __a     ,  unittest.TestCase):
			__a			:							Dict					       =				(TimesformerModel, TimesformerForVideoClassification) if is_torch_available() else ()
			__a			:							List[str]					       =				(
			    {'''feature-extraction''': TimesformerModel, '''video-classification''': TimesformerForVideoClassification}
			    if is_torch_available()
			    else {}
			)

			__a			:							Union[str, Any]					       =				False
			__a			:							Tuple					       =				False
			__a			:							str					       =				False
			__a			:							Any					       =				False
			def 				__snake_case			(		self  )     ->     List[Any]:






				'''simple docstring'''







				_UpperCAmelCase     :				Any          =       TimesformerModelTester(self  )
				_UpperCAmelCase     :				Tuple          =       ConfigTester(
				    self   ,		config_class=__lowerCAmelCase   ,		has_text_modality=__lowerCAmelCase   ,		hidden_size=37  )
			def 				__snake_case			(		self   ,		_A   ,		_A   ,		_A=False  )     ->     str:






				'''simple docstring'''







				_UpperCAmelCase     :				Tuple          =       copy.deepcopy(__lowerCAmelCase  )

				if return_labels:
					if model_class in get_values(__lowerCAmelCase  ):
						_UpperCAmelCase     :				Union[str, Any]          =       torch.zeros(
						    self.model_tester.batch_size   ,		dtype=torch.long   ,		device=__lowerCAmelCase  )

				return inputs_dict
			def 				__snake_case			(		self  )     ->     Optional[Any]:






				'''simple docstring'''







				self.config_tester.run_common_tests()
			@unittest.skip(reason="""TimeSformer does not use inputs_embeds"""  )
			def 				__snake_case			(		self  )     ->     Optional[Any]:






				'''simple docstring'''







				pass
			def 				__snake_case			(		self  )     ->     Any:






				'''simple docstring'''







				_UpperCAmelCase    , _UpperCAmelCase     :				Union[str, Any]          =       self.model_tester.prepare_config_and_inputs_for_common()

				for model_class in self.all_model_classes:
					_UpperCAmelCase     :				List[Any]          =       model_class(__lowerCAmelCase  )
					self.assertIsInstance(model.get_input_embeddings()   ,		(nn.Module)  )
					_UpperCAmelCase     :				int          =       model.get_output_embeddings()
					self.assertTrue(x is None or isinstance(__lowerCAmelCase   ,		nn.Linear  )  )
			def 				__snake_case			(		self  )     ->     int:






				'''simple docstring'''







				_UpperCAmelCase    , _UpperCAmelCase     :				Optional[Any]          =       self.model_tester.prepare_config_and_inputs_for_common()

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

					_UpperCAmelCase     :				Tuple          =       ["""pixel_values"""]
					self.assertListEqual(arg_names[:1]   ,		__lowerCAmelCase  )
			def 				__snake_case			(		self  )     ->     Union[str, Any]:






				'''simple docstring'''







				_UpperCAmelCase     :				Any          =       self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_model(*__lowerCAmelCase  )
			def 				__snake_case			(		self  )     ->     Any:






				'''simple docstring'''







				_UpperCAmelCase     :				Dict          =       self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_for_video_classification(*__lowerCAmelCase  )
			@slow
			def 				__snake_case			(		self  )     ->     Dict:






				'''simple docstring'''







				for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
					_UpperCAmelCase     :				Optional[int]          =       TimesformerModel.from_pretrained(__lowerCAmelCase  )
					self.assertIsNotNone(__lowerCAmelCase  )
			def 				__snake_case			(		self  )     ->     Optional[Any]:






				'''simple docstring'''







				if not self.has_attentions:
					pass

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

					for model_class in self.all_model_classes:
						_UpperCAmelCase     :				List[Any]          =       self.model_tester.seq_length
						_UpperCAmelCase     :				Optional[int]          =       self.model_tester.num_frames

						_UpperCAmelCase     :				List[str]          =       True
						_UpperCAmelCase     :				Dict          =       False
						_UpperCAmelCase     :				Optional[int]          =       True
						_UpperCAmelCase     :				str          =       model_class(__lowerCAmelCase  )
						model.to(__lowerCAmelCase  )
						model.eval()
						with torch.no_grad():
							_UpperCAmelCase     :				str          =       model(**self._prepare_for_class(__lowerCAmelCase   ,		__lowerCAmelCase  )  )
						_UpperCAmelCase     :				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"]
						_UpperCAmelCase     :				Union[str, Any]          =       True
						_UpperCAmelCase     :				List[Any]          =       model_class(__lowerCAmelCase  )
						model.to(__lowerCAmelCase  )
						model.eval()
						with torch.no_grad():
							_UpperCAmelCase     :				List[str]          =       model(**self._prepare_for_class(__lowerCAmelCase   ,		__lowerCAmelCase  )  )
						_UpperCAmelCase     :				List[Any]          =       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]   ,		)
						_UpperCAmelCase     :				Union[str, Any]          =       len(__lowerCAmelCase  )

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

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

						_UpperCAmelCase     :				Optional[Any]          =       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 				__snake_case			(		self  )     ->     Tuple:






				'''simple docstring'''







				def check_hidden_states_output(_A   ,		_A   ,		_A  ):
					_UpperCAmelCase     :				Optional[Any]          =       model_class(__lowerCAmelCase  )
					model.to(__lowerCAmelCase  )
					model.eval()

					with torch.no_grad():
						_UpperCAmelCase     :				Optional[Any]          =       model(**self._prepare_for_class(__lowerCAmelCase   ,		__lowerCAmelCase  )  )

					_UpperCAmelCase     :				Any          =       outputs.hidden_states
					_UpperCAmelCase     :				int          =       self.model_tester.num_hidden_layers + 1
					self.assertEqual(len(__lowerCAmelCase  )   ,		__lowerCAmelCase  )

					_UpperCAmelCase     :				Tuple          =       self.model_tester.seq_length

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

				_UpperCAmelCase    , _UpperCAmelCase     :				Union[str, Any]          =       self.model_tester.prepare_config_and_inputs_for_common()

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

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

					check_hidden_states_output(__lowerCAmelCase   ,		__lowerCAmelCase   ,		__lowerCAmelCase  )
def    UpperCamelCase       (      )					->      Any:
	_UpperCAmelCase     :				str          =       hf_hub_download(
	    repo_id="""hf-internal-testing/spaghetti-video""",	filename="""eating_spaghetti.npy""",	repo_type="""dataset"""			)
	_UpperCAmelCase     :				Optional[int]          =       np.load(__a			)
	return list(__a			)






@require_torch
@require_vision
class      _UpperCAmelCase      (		unittest.TestCase):
			@cached_property
			def 				__snake_case			(		self  )     ->     Any:






				'''simple docstring'''







				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 				__snake_case			(		self  )     ->     Tuple:






				'''simple docstring'''







				_UpperCAmelCase     :				Union[str, Any]          =       TimesformerForVideoClassification.from_pretrained("""facebook/timesformer-base-finetuned-k400"""  ).to(
				    __lowerCAmelCase  )

				_UpperCAmelCase     :				Any          =       self.default_image_processor
				_UpperCAmelCase     :				int          =       prepare_video()
				_UpperCAmelCase     :				Any          =       image_processor(video[:8]   ,		return_tensors="""pt"""  ).to(__lowerCAmelCase  )

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

				# verify the logits
				_UpperCAmelCase     :				List[str]          =       torch.Size((1, 4_00)  )
				self.assertEqual(outputs.logits.shape   ,		__lowerCAmelCase  )

				_UpperCAmelCase     :				str          =       torch.tensor([-0.3016, -0.7713, -0.4205]  ).to(__lowerCAmelCase  )

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












import argparse

from transformers import CLIPImageProcessor, CLIPVisionModelWithProjection

from diffusers import UnCLIPImageVariationPipeline, UnCLIPPipeline


if __name__ == "__main__":
					A       :   List[Any]					= argparse.ArgumentParser()

					parser.add_argument('''--dump_path''', default=None, type=str, required=True, help='''Path to the output model.''')

					parser.add_argument(
					    '''--txt2img_unclip''',
					    default='''kakaobrain/karlo-v1-alpha''',
					    type=str,
					    required=False,
					    help='''The pretrained txt2img unclip.''',
					)

					A       :   List[Any]					= parser.parse_args()

					A       :   int					= UnCLIPPipeline.from_pretrained(args.txtaimg_unclip)

					A       :   Tuple					= CLIPImageProcessor()
					A       :   int					= CLIPVisionModelWithProjection.from_pretrained('''openai/clip-vit-large-patch14''')

					A       :   Tuple					= UnCLIPImageVariationPipeline(
					    decoder=txtaimg.decoder,
					    text_encoder=txtaimg.text_encoder,
					    tokenizer=txtaimg.tokenizer,
					    text_proj=txtaimg.text_proj,
					    feature_extractor=feature_extractor,
					    image_encoder=image_encoder,
					    super_res_first=txtaimg.super_res_first,
					    super_res_last=txtaimg.super_res_last,
					    decoder_scheduler=txtaimg.decoder_scheduler,
					    super_res_scheduler=txtaimg.super_res_scheduler,
					)

					imgaimg.save_pretrained(args.dump_path)



