



import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class lowerCAmelCase_       (   UpperCamelCase__ ,UpperCamelCase__ ,unittest.TestCase			):







	'''simple docstring'''




	A_       :     Tuple									=		StableDiffusionParadigmsPipeline
	A_       :     Optional[int]									=		TEXT_TO_IMAGE_PARAMS
	A_       :     Dict									=		TEXT_TO_IMAGE_BATCH_PARAMS
	A_       :     List[str]									=		TEXT_TO_IMAGE_IMAGE_PARAMS
	A_       :     Tuple									=		TEXT_TO_IMAGE_IMAGE_PARAMS

	def 				_A			(	self       ):



				'''simple docstring'''


				torch.manual_seed(0       )
				a__					=  UNetaDConditionModel(
				    block_out_channels=(32, 64)   ,       layers_per_block=2   ,       sample_size=32   ,       in_channels=4   ,       out_channels=4   ,       down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")   ,       up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")   ,       cross_attention_dim=32   ,       attention_head_dim=(2, 4)   ,       use_linear_projection=lowerCamelCase   ,       )
				a__					=  DDIMParallelScheduler(
				    beta_start=0.0_0_0_8_5   ,       beta_end=0.0_1_2   ,       beta_schedule="""scaled_linear"""   ,       clip_sample=lowerCamelCase   ,       set_alpha_to_one=lowerCamelCase   ,       )
				torch.manual_seed(0       )
				a__					=  AutoencoderKL(
				    block_out_channels=[32, 64]   ,       in_channels=3   ,       out_channels=3   ,       down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]   ,       up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]   ,       latent_channels=4   ,       sample_size=128   ,       )
				torch.manual_seed(0       )
				a__					=  CLIPTextConfig(
				    bos_token_id=0   ,       eos_token_id=2   ,       hidden_size=32   ,       intermediate_size=37   ,       layer_norm_eps=1e-05   ,       num_attention_heads=4   ,       num_hidden_layers=5   ,       pad_token_id=1   ,       vocab_size=1000   ,       hidden_act="""gelu"""   ,       projection_dim=512   ,       )
				a__					=  CLIPTextModel(lowerCamelCase       )
				a__					=  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""       )

				a__					=  {
				    """unet""": unet,
				    """scheduler""": scheduler,
				    """vae""": vae,
				    """text_encoder""": text_encoder,
				    """tokenizer""": tokenizer,
				    """safety_checker""": None,
				    """feature_extractor""": None,
				}
				return components

	def 				_A			(	self   ,       lowerCamelCase   ,       lowerCamelCase=0       ):



				'''simple docstring'''


				if str(lowerCamelCase       ).startswith("""mps"""       ):
							a__					=  torch.manual_seed(lowerCamelCase       )
				else:
							a__					=  torch.Generator(device=lowerCamelCase       ).manual_seed(lowerCamelCase       )
				a__					=  {
				    """prompt""": """a photograph of an astronaut riding a horse""",
				    """generator""": generator,
				    """num_inference_steps""": 10,
				    """guidance_scale""": 6.0,
				    """output_type""": """numpy""",
				    """parallel""": 3,
				    """debug""": True,
				}
				return inputs

	def 				_A			(	self       ):



				'''simple docstring'''


				a__					=  """cpu"""  # ensure determinism for the device-dependent torch.Generator
				a__					=  self.get_dummy_components()
				a__					=  StableDiffusionParadigmsPipeline(**lowerCamelCase       )
				a__					=  sd_pipe.to(lowerCamelCase       )
				sd_pipe.set_progress_bar_config(disable=lowerCamelCase       )

				a__					=  self.get_dummy_inputs(lowerCamelCase       )
				a__					=  sd_pipe(**lowerCamelCase       ).images
				a__					=  image[0, -3:, -3:, -1]
				assert image.shape == (1, 64, 64, 3)

				a__					=  np.array([0.4_7_7_3, 0.5_4_1_7, 0.4_7_2_3, 0.4_9_2_5, 0.5_6_3_1, 0.4_7_5_2, 0.5_2_4_0, 0.4_9_3_5, 0.5_0_2_3]       )

				assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2

	def 				_A			(	self       ):



				'''simple docstring'''


				a__					=  """cpu"""  # ensure determinism for the device-dependent torch.Generator
				a__					=  self.get_dummy_components()
				torch.manual_seed(0       )
				a__					=  DDPMParallelScheduler()
				torch.manual_seed(0       )
				a__					=  StableDiffusionParadigmsPipeline(**lowerCamelCase       )
				a__					=  sd_pipe.to(lowerCamelCase       )
				sd_pipe.set_progress_bar_config(disable=lowerCamelCase       )

				a__					=  self.get_dummy_inputs(lowerCamelCase       )
				a__					=  sd_pipe(**lowerCamelCase       ).images
				a__					=  image[0, -3:, -3:, -1]
				assert image.shape == (1, 64, 64, 3)

				a__					=  np.array([0.3_5_7_3, 0.4_4_2_0, 0.4_9_6_0, 0.4_7_9_9, 0.3_7_9_6, 0.3_8_7_9, 0.4_8_1_9, 0.4_3_6_5, 0.4_4_6_8]       )

				assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2

	def 				_A			(	self       ):



				'''simple docstring'''


				super().test_inference_batch_consistent(batch_sizes=[1, 2]       )

	def 				_A			(	self       ):



				'''simple docstring'''


				super().test_inference_batch_single_identical(batch_size=2   ,       expected_max_diff=3e-3       )







	def 				_A			(	self       ):



				'''simple docstring'''


				a__					=  """cpu"""  # ensure determinism for the device-dependent torch.Generator
				a__					=  self.get_dummy_components()
				a__					=  StableDiffusionParadigmsPipeline(**lowerCamelCase       )
				a__					=  sd_pipe.to(lowerCamelCase       )
				sd_pipe.set_progress_bar_config(disable=lowerCamelCase       )

				a__					=  self.get_dummy_inputs(lowerCamelCase       )
				a__					=  """french fries"""
				a__					=  sd_pipe(**lowerCamelCase   ,       negative_prompt=lowerCamelCase       )
				a__					=  output.images
				a__					=  image[0, -3:, -3:, -1]

				assert image.shape == (1, 64, 64, 3)

				a__					=  np.array([0.4_7_7_1, 0.5_4_2_0, 0.4_6_8_3, 0.4_9_1_8, 0.5_6_3_6, 0.4_7_2_5, 0.5_2_3_0, 0.4_9_2_3, 0.5_0_1_5]       )

				assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2


@slow
@require_torch_gpu
class lowerCAmelCase_       (   unittest.TestCase			):







	'''simple docstring'''




	def 				_A			(	self       ):



				'''simple docstring'''


				super().tearDown()
				gc.collect()
				torch.cuda.empty_cache()

	def 				_A			(	self   ,       lowerCamelCase=0       ):



				'''simple docstring'''


				a__					=  torch.Generator(device=lowerCamelCase       ).manual_seed(lowerCamelCase       )
				a__					=  {
				    """prompt""": """a photograph of an astronaut riding a horse""",
				    """generator""": generator,
				    """num_inference_steps""": 10,
				    """guidance_scale""": 7.5,
				    """output_type""": """numpy""",
				    """parallel""": 3,
				    """debug""": True,
				}
				return inputs







	def 				_A			(	self       ):



				'''simple docstring'''


				a__					=  """stabilityai/stable-diffusion-2-base"""
				a__					=  DDIMParallelScheduler.from_pretrained(lowerCamelCase   ,       subfolder="""scheduler"""       )
				a__					=  StableDiffusionParadigmsPipeline.from_pretrained(lowerCamelCase   ,       scheduler=lowerCamelCase   ,       safety_checker=lowerCamelCase       )
				pipe.to(lowerCamelCase       )
				pipe.set_progress_bar_config(disable=lowerCamelCase       )
				pipe.enable_attention_slicing()

				a__					=  self.get_inputs()
				a__					=  pipe(**lowerCamelCase       ).images
				a__					=  image[0, -3:, -3:, -1].flatten()

				assert image.shape == (1, 512, 512, 3)

				a__					=  np.array([0.9_6_2_2, 0.9_6_0_2, 0.9_7_4_8, 0.9_5_9_1, 0.9_6_3_0, 0.9_6_9_1, 0.9_6_6_1, 0.9_6_3_1, 0.9_7_4_1]       )

				assert np.abs(expected_slice - image_slice       ).max() < 1e-2






'''simple docstring'''



import argparse
import json

import gdown
import numpy as np
import torch
from huggingface_hub import hf_hub_download

from transformers import TimesformerConfig, TimesformerForVideoClassification, VideoMAEImageProcessor



def 	__lowerCAmelCase       (				lowerCamelCase :    str       ):






	'''simple docstring'''




	__lowerCAmelCase       				=	TimesformerConfig()

	if "large" in model_name:
		__lowerCAmelCase       				=	96

	if "hr" in model_name:
		__lowerCAmelCase       				=	16
		__lowerCAmelCase       				=	4_48

	__lowerCAmelCase       				=	"huggingface/label-files"
	if "k400" in model_name:
		__lowerCAmelCase       				=	4_00
		__lowerCAmelCase       				=	"kinetics400-id2label.json"
	elif "k600" in model_name:
		__lowerCAmelCase       				=	6_00
		__lowerCAmelCase       				=	"kinetics600-id2label.json"
	elif "ssv2" in model_name:
		__lowerCAmelCase       				=	1_74
		__lowerCAmelCase       				=	"something-something-v2-id2label.json"
	else:
		raise ValueError("Model name should either contain 'k400', 'k600' or 'ssv2'."       )
	__lowerCAmelCase       				=	json.load(open(hf_hub_download(lowerCamelCase					,		lowerCamelCase					,		repo_type="dataset"       )					,		"r"       )       )
	__lowerCAmelCase       				=	{int(lowerCamelCase       ): v for k, v in idalabel.items()}
	__lowerCAmelCase       				=	idalabel
	__lowerCAmelCase       				=	{v: k for k, v in idalabel.items()}

	return config



def 	__lowerCAmelCase       (				lowerCamelCase :    int       ):






	'''simple docstring'''




	if "encoder." in name:
		__lowerCAmelCase       				=	name.replace("encoder."					,		""       )
	if "cls_token" in name:
		__lowerCAmelCase       				=	name.replace("cls_token"					,		"timesformer.embeddings.cls_token"       )
	if "pos_embed" in name:
		__lowerCAmelCase       				=	name.replace("pos_embed"					,		"timesformer.embeddings.position_embeddings"       )
	if "time_embed" in name:
		__lowerCAmelCase       				=	name.replace("time_embed"					,		"timesformer.embeddings.time_embeddings"       )
	if "patch_embed.proj" in name:
		__lowerCAmelCase       				=	name.replace("patch_embed.proj"					,		"timesformer.embeddings.patch_embeddings.projection"       )
	if "patch_embed.norm" in name:
		__lowerCAmelCase       				=	name.replace("patch_embed.norm"					,		"timesformer.embeddings.norm"       )
	if "blocks" in name:
		__lowerCAmelCase       				=	name.replace("blocks"					,		"timesformer.encoder.layer"       )
	if "attn.proj" in name:
		__lowerCAmelCase       				=	name.replace("attn.proj"					,		"attention.output.dense"       )
	if "attn" in name and "bias" not in name and "temporal" not in name:
		__lowerCAmelCase       				=	name.replace("attn"					,		"attention.self"       )
	if "attn" in name and "temporal" not in name:
		__lowerCAmelCase       				=	name.replace("attn"					,		"attention.attention"       )
	if "temporal_norm1" in name:
		__lowerCAmelCase       				=	name.replace("temporal_norm1"					,		"temporal_layernorm"       )
	if "temporal_attn.proj" in name:
		__lowerCAmelCase       				=	name.replace("temporal_attn"					,		"temporal_attention.output.dense"       )
	if "temporal_fc" in name:
		__lowerCAmelCase       				=	name.replace("temporal_fc"					,		"temporal_dense"       )
	if "norm1" in name and "temporal" not in name:
		__lowerCAmelCase       				=	name.replace("norm1"					,		"layernorm_before"       )
	if "norm2" in name:
		__lowerCAmelCase       				=	name.replace("norm2"					,		"layernorm_after"       )
	if "mlp.fc1" in name:
		__lowerCAmelCase       				=	name.replace("mlp.fc1"					,		"intermediate.dense"       )
	if "mlp.fc2" in name:
		__lowerCAmelCase       				=	name.replace("mlp.fc2"					,		"output.dense"       )
	if "norm.weight" in name and "fc" not in name and "temporal" not in name:
		__lowerCAmelCase       				=	name.replace("norm.weight"					,		"timesformer.layernorm.weight"       )
	if "norm.bias" in name and "fc" not in name and "temporal" not in name:
		__lowerCAmelCase       				=	name.replace("norm.bias"					,		"timesformer.layernorm.bias"       )
	if "head" in name:
		__lowerCAmelCase       				=	name.replace("head"					,		"classifier"       )

	return name



def 	__lowerCAmelCase       (				lowerCamelCase :    int					,		lowerCamelCase :    List[str]       ):






	'''simple docstring'''




	for key in orig_state_dict.copy().keys():
		__lowerCAmelCase       				=	orig_state_dict.pop(lowerCamelCase       )

		if key.startswith("model."       ):
			__lowerCAmelCase       				=	key.replace("model."					,		""       )

		if "qkv" in key:
			__lowerCAmelCase       				=	key.split("."       )
			__lowerCAmelCase       				=	int(key_split[1]       )
			__lowerCAmelCase       				=	"timesformer.encoder.layer."
			if "temporal" in key:
				__lowerCAmelCase       				=	".temporal_attention.attention.qkv."
			else:
				__lowerCAmelCase       				=	".attention.attention.qkv."
			if "weight" in key:
				__lowerCAmelCase       				=	val
			else:
				__lowerCAmelCase       				=	val
		else:
			__lowerCAmelCase       				=	val

	return orig_state_dict



def 	__lowerCAmelCase       (				):






	'''simple docstring'''




	__lowerCAmelCase       				=	hf_hub_download(
	    repo_id="hf-internal-testing/spaghetti-video"					,		filename="eating_spaghetti.npy"					,		repo_type="dataset"       )
	__lowerCAmelCase       				=	np.load(lowerCamelCase       )
	return list(lowerCamelCase       )



def 	__lowerCAmelCase       (				lowerCamelCase :    Union[str, Any]					,		lowerCamelCase :    Dict					,		lowerCamelCase :    List[Any]					,		lowerCamelCase :    Optional[Any]       ):






	'''simple docstring'''




	__lowerCAmelCase       				=	get_timesformer_config(lowerCamelCase       )

	__lowerCAmelCase       				=	TimesformerForVideoClassification(lowerCamelCase       )

	# download original checkpoint, hosted on Google Drive
	__lowerCAmelCase       				=	"pytorch_model.bin"
	gdown.cached_download(lowerCamelCase					,		lowerCamelCase					,		quiet=lowerCamelCase       )
	__lowerCAmelCase       				=	torch.load(lowerCamelCase					,		map_location="cpu"       )
	if "model" in files:
		__lowerCAmelCase       				=	files["model"]
	elif "module" in files:
		__lowerCAmelCase       				=	files["module"]
	else:
		__lowerCAmelCase       				=	files["model_state"]
	__lowerCAmelCase       				=	convert_state_dict(lowerCamelCase					,		lowerCamelCase       )

	model.load_state_dict(lowerCamelCase       )
	model.eval()

	# verify model on basic input
	__lowerCAmelCase       				=	VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]					,		image_std=[0.5, 0.5, 0.5]       )
	__lowerCAmelCase       				=	prepare_video()
	__lowerCAmelCase       				=	image_processor(video[:8]					,		return_tensors="pt"       )

	__lowerCAmelCase       				=	model(**lowerCamelCase       )
	__lowerCAmelCase       				=	outputs.logits

	__lowerCAmelCase       				=	[
	    # Kinetics-400 checkpoints (hr = high resolution input of 448px instead of 224px)
	    "timesformer-base-finetuned-k400",
	    "timesformer-large-finetuned-k400",
	    "timesformer-hr-finetuned-k400",
	    # Kinetics-600 checkpoints (hr = high resolution input of 448px instead of 224px)
	    "timesformer-base-finetuned-k600",
	    "timesformer-large-finetuned-k600",
	    "timesformer-hr-finetuned-k600",
	    # Something-Something-v2 checkpoints (hr = high resolution input of 448px instead of 224px)
	    "timesformer-base-finetuned-ssv2",
	    "timesformer-large-finetuned-ssv2",
	    "timesformer-hr-finetuned-ssv2",
	]

	# NOTE: logits were tested with image_mean and image_std equal to [0.5, 0.5, 0.5] and [0.5, 0.5, 0.5]
	if model_name == "timesformer-base-finetuned-k400":
		__lowerCAmelCase       				=	torch.Size([1, 4_00]       )
		__lowerCAmelCase       				=	torch.tensor([-0.3_0_1_6, -0.7_7_1_3, -0.4_2_0_5]       )
	elif model_name == "timesformer-base-finetuned-k600":
		__lowerCAmelCase       				=	torch.Size([1, 6_00]       )
		__lowerCAmelCase       				=	torch.tensor([-0.7_2_6_7, -0.7_4_6_6, 3.2_4_0_4]       )
	elif model_name == "timesformer-base-finetuned-ssv2":
		__lowerCAmelCase       				=	torch.Size([1, 1_74]       )
		__lowerCAmelCase       				=	torch.tensor([-0.9_0_5_9, 0.6_4_3_3, -3.1_4_5_7]       )
	elif model_name == "timesformer-large-finetuned-k400":
		__lowerCAmelCase       				=	torch.Size([1, 4_00]       )
		__lowerCAmelCase       				=	torch.tensor([0, 0, 0]       )
	elif model_name == "timesformer-large-finetuned-k600":
		__lowerCAmelCase       				=	torch.Size([1, 6_00]       )
		__lowerCAmelCase       				=	torch.tensor([0, 0, 0]       )
	elif model_name == "timesformer-large-finetuned-ssv2":
		__lowerCAmelCase       				=	torch.Size([1, 1_74]       )
		__lowerCAmelCase       				=	torch.tensor([0, 0, 0]       )
	elif model_name == "timesformer-hr-finetuned-k400":
		__lowerCAmelCase       				=	torch.Size([1, 4_00]       )
		__lowerCAmelCase       				=	torch.tensor([-0.9_6_1_7, -3.7_3_1_1, -3.7_7_0_8]       )
	elif model_name == "timesformer-hr-finetuned-k600":
		__lowerCAmelCase       				=	torch.Size([1, 6_00]       )
		__lowerCAmelCase       				=	torch.tensor([2.5_2_7_3, 0.7_1_2_7, 1.8_8_4_8]       )
	elif model_name == "timesformer-hr-finetuned-ssv2":
		__lowerCAmelCase       				=	torch.Size([1, 1_74]       )
		__lowerCAmelCase       				=	torch.tensor([-3.6_7_5_6, -0.7_5_1_3, 0.7_1_8_0]       )
	else:
		raise ValueError(f'''Model name not supported. Should be one of {model_names}'''       )

	# verify logits
	assert logits.shape == expected_shape
	assert torch.allclose(logits[0, :3]					,		lowerCamelCase					,		atol=1e-4       )
	print("Logits ok!"       )

	if pytorch_dump_folder_path is not None:
		print(f'''Saving model and image processor to {pytorch_dump_folder_path}'''       )
		image_processor.save_pretrained(lowerCamelCase       )
		model.save_pretrained(lowerCamelCase       )

	if push_to_hub:
		print("Pushing to the hub..."       )
		model.push_to_hub(f'''fcakyon/{model_name}'''       )


if __name__ == "__main__":
	lowerCAmelCase						:		Dict						      = argparse.ArgumentParser()
	# Required parameters
	parser.add_argument(
	    '''--checkpoint_url''',
	    default='''https://drive.google.com/u/1/uc?id=17yvuYp9L4mn-HpIcK5Zo6K3UoOy1kA5l&export=download''',
	    type=str,
	    help=(
	        '''URL of the original PyTorch checkpoint (on Google Drive) you\'d like to convert. Should be a direct'''
	        ''' download link.'''
	    ),
	)
	parser.add_argument(
	    '''--pytorch_dump_folder_path''',
	    default='''''',
	    type=str,
	    help='''Path to the output PyTorch model directory.''',
	)
	parser.add_argument('''--model_name''', default='''timesformer-base-finetuned-k400''', type=str, help='''Name of the model.''')
	parser.add_argument(
	    '''--push_to_hub''', action='''store_true''', help='''Whether or not to push the converted model to the 🤗 hub.'''
	)

	lowerCAmelCase						:		Optional[int]						      = parser.parse_args()
	convert_timesformer_checkpoint(
	    args.checkpoint_url, args.pytorch_dump_folder_path, args.model_name, args.push_to_hub
	)