


"""simple docstring"""


import argparse
from pathlib import Path

import fairseq
import torch
from fairseq.models.xmod import XMODModel as FairseqXmodModel
from packaging import version

from transformers import XmodConfig, XmodForMaskedLM, XmodForSequenceClassification
from transformers.utils import logging


if version.parse(fairseq.__version__) < version.parse("0.12.2"):
	raise Exception("requires fairseq >= 0.12.2")
if version.parse(fairseq.__version__) > version.parse("2"):
	raise Exception("requires fairseq < v2")

logging.set_verbosity_info()
A  :     int    					=    logging.get_logger(__name__)

A  :     Tuple    					=    "Hello, World!"
A  :     List[str]    					=    "en_XX"


def _lowerCamelCase			(	_UpperCamelCase					,				_UpperCamelCase					,				_UpperCamelCase   ):

			'''simple docstring'''




			__lowerCAmelCase       					=    Path("data_bin"   )
			__lowerCAmelCase       					=    FairseqXmodModel.from_pretrained(
			    model_name_or_path=str(Path(_UpperCamelCase   ).parent   )					,				checkpoint_file=Path(_UpperCamelCase   ).name					,				_name="xmod_base"					,				arch="xmod_base"					,				task="multilingual_masked_lm"					,				data_name_or_path=str(_UpperCamelCase   )					,				bpe="sentencepiece"					,				sentencepiece_model=str(Path(_UpperCamelCase   ).parent / "sentencepiece.bpe.model"   )					,				src_dict=str(data_dir / "dict.txt"   )					,				)
			xmod.eval()  # disable dropout
			print(_UpperCamelCase   )

			__lowerCAmelCase       					=    xmod.model.encoder.sentence_encoder
			__lowerCAmelCase       					=    XmodConfig(
			    vocab_size=xmod_sent_encoder.embed_tokens.num_embeddings					,				hidden_size=xmod.cfg.model.encoder_embed_dim					,				num_hidden_layers=xmod.cfg.model.encoder_layers					,				num_attention_heads=xmod.cfg.model.encoder_attention_heads					,				intermediate_size=xmod.cfg.model.encoder_ffn_embed_dim					,				max_position_embeddings=514					,				type_vocab_size=1					,				layer_norm_eps=1e-5					,				pre_norm=xmod.cfg.model.encoder_normalize_before					,				adapter_reduction_factor=getattr(xmod.cfg.model					,				"bottleneck"					,				2   )					,				adapter_layer_norm=xmod.cfg.model.adapter_layer_norm					,				adapter_reuse_layer_norm=xmod.cfg.model.adapter_reuse_layer_norm					,				ln_before_adapter=xmod.cfg.model.ln_before_adapter					,				languages=xmod.cfg.model.languages					,				)
			if classification_head:
						__lowerCAmelCase       					=    xmod.model.classification_heads["mnli"].out_proj.weight.shape[0]

			print("Our X-MOD config:"					,				_UpperCamelCase   )

			__lowerCAmelCase       					=    XmodForSequenceClassification(_UpperCamelCase   ) if classification_head else XmodForMaskedLM(_UpperCamelCase   )
			model.eval()

			# Now let's copy all the weights.
			# Embeddings
			__lowerCAmelCase       					=    xmod_sent_encoder.embed_tokens.weight
			__lowerCAmelCase       					=    xmod_sent_encoder.embed_positions.weight
			__lowerCAmelCase       					=    torch.zeros_like(
			    model.roberta.embeddings.token_type_embeddings.weight   )  # just zero them out b/c xmod doesn't use them.

			__lowerCAmelCase       					=    xmod_sent_encoder.layernorm_embedding.weight
			__lowerCAmelCase       					=    xmod_sent_encoder.layernorm_embedding.bias

			for i in range(config.num_hidden_layers   ):
						# Encoder: start of layer
						__lowerCAmelCase       					=    model.roberta.encoder.layer[i]
						__lowerCAmelCase       					=    xmod_sent_encoder.layers[i]

						# self attention
						__lowerCAmelCase       					=    layer.attention.self
						if not (
						    xmod_layer.self_attn.k_proj.weight.data.shape
						    == xmod_layer.self_attn.q_proj.weight.data.shape
						    == xmod_layer.self_attn.v_proj.weight.data.shape
						    == torch.Size((config.hidden_size, config.hidden_size)   )
						):
									raise AssertionError("Dimensions of self-attention weights do not match."   )

						__lowerCAmelCase       					=    xmod_layer.self_attn.q_proj.weight
						__lowerCAmelCase       					=    xmod_layer.self_attn.q_proj.bias
						__lowerCAmelCase       					=    xmod_layer.self_attn.k_proj.weight
						__lowerCAmelCase       					=    xmod_layer.self_attn.k_proj.bias
						__lowerCAmelCase       					=    xmod_layer.self_attn.v_proj.weight
						__lowerCAmelCase       					=    xmod_layer.self_attn.v_proj.bias

						# self-attention output
						__lowerCAmelCase       					=    layer.attention.output
						if self_output.dense.weight.shape != xmod_layer.self_attn.out_proj.weight.shape:
									raise AssertionError("Dimensions of self-attention output weights do not match."   )
						__lowerCAmelCase       					=    xmod_layer.self_attn.out_proj.weight
						__lowerCAmelCase       					=    xmod_layer.self_attn.out_proj.bias
						__lowerCAmelCase       					=    xmod_layer.self_attn_layer_norm.weight
						__lowerCAmelCase       					=    xmod_layer.self_attn_layer_norm.bias

						# intermediate
						__lowerCAmelCase       					=    layer.intermediate
						if intermediate.dense.weight.shape != xmod_layer.fca.weight.shape:
									raise AssertionError("Dimensions of intermediate weights do not match."   )
						__lowerCAmelCase       					=    xmod_layer.fca.weight
						__lowerCAmelCase       					=    xmod_layer.fca.bias

						# output
						__lowerCAmelCase       					=    layer.output
						if bert_output.dense.weight.shape != xmod_layer.fca.weight.shape:
									raise AssertionError("Dimensions of feed-forward weights do not match."   )
						__lowerCAmelCase       					=    xmod_layer.fca.weight
						__lowerCAmelCase       					=    xmod_layer.fca.bias
						__lowerCAmelCase       					=    xmod_layer.final_layer_norm.weight
						__lowerCAmelCase       					=    xmod_layer.final_layer_norm.bias
						if bert_output.adapter_layer_norm is not None:
									__lowerCAmelCase       					=    xmod_layer.adapter_layer_norm.weight
									__lowerCAmelCase       					=    xmod_layer.adapter_layer_norm.bias

						if sorted(bert_output.adapter_modules.keys()   ) != sorted(xmod_layer.adapter_modules.keys()   ):
									raise AssertionError("Lists of language adapters do not match."   )
						for lang_code, adapter in xmod_layer.adapter_modules.items():
									__lowerCAmelCase       					=    bert_output.adapter_modules[lang_code]
									__lowerCAmelCase       					=    xmod_layer.adapter_modules[lang_code]
									__lowerCAmelCase       					=    from_adapter.fca.weight
									__lowerCAmelCase       					=    from_adapter.fca.bias
									__lowerCAmelCase       					=    from_adapter.fca.weight
									__lowerCAmelCase       					=    from_adapter.fca.bias

						# end of layer

			if xmod_sent_encoder.layer_norm is not None:
						__lowerCAmelCase       					=    xmod_sent_encoder.layer_norm.weight
						__lowerCAmelCase       					=    xmod_sent_encoder.layer_norm.bias

			if classification_head:
						__lowerCAmelCase       					=    xmod.model.classification_heads["mnli"].dense.weight
						__lowerCAmelCase       					=    xmod.model.classification_heads["mnli"].dense.bias
						__lowerCAmelCase       					=    xmod.model.classification_heads["mnli"].out_proj.weight
						__lowerCAmelCase       					=    xmod.model.classification_heads["mnli"].out_proj.bias
			else:
						# LM Head
						__lowerCAmelCase       					=    xmod.model.encoder.lm_head.dense.weight
						__lowerCAmelCase       					=    xmod.model.encoder.lm_head.dense.bias
						__lowerCAmelCase       					=    xmod.model.encoder.lm_head.layer_norm.weight
						__lowerCAmelCase       					=    xmod.model.encoder.lm_head.layer_norm.bias
						__lowerCAmelCase       					=    xmod.model.encoder.lm_head.weight
						__lowerCAmelCase       					=    xmod.model.encoder.lm_head.bias

			# Let's check that we get the same results.
			__lowerCAmelCase       					=    xmod.encode(_UpperCamelCase   ).unsqueeze(0   )  # batch of size 1
			model.roberta.set_default_language(_UpperCamelCase   )

			__lowerCAmelCase       					=    model(_UpperCamelCase   )[0]
			if classification_head:
						__lowerCAmelCase       					=    xmod.model.classification_heads["mnli"](xmod.extract_features(_UpperCamelCase   )   )
			else:
						__lowerCAmelCase       					=    xmod.model(_UpperCamelCase					,				lang_id=[SAMPLE_LANGUAGE]   )[0]
			print(our_output.shape					,				their_output.shape   )
			__lowerCAmelCase       					=    torch.max(torch.abs(our_output - their_output   )   ).item()
			print(f"max_absolute_diff = {max_absolute_diff}"   )  # ~ 1e-7
			__lowerCAmelCase       					=    torch.allclose(_UpperCamelCase					,				_UpperCamelCase					,				atol=1e-3   )
			print("Do both models output the same tensors?"					,				"🔥" if success else "💩"   )
			if not success:
						raise Exception("Something went wRoNg"   )

			Path(_UpperCamelCase   ).mkdir(parents=_UpperCamelCase					,				exist_ok=_UpperCamelCase   )
			print(f"Saving model to {pytorch_dump_folder_path}"   )
			model.save_pretrained(_UpperCamelCase   )


if __name__ == "__main__":
	A  :     str    					=    argparse.ArgumentParser()
	# Required parameters
	parser.add_argument(
	    "--xmod_checkpoint_path", default=None, type=str, required=True, help="Path the official PyTorch dump."
	)
	parser.add_argument(
	    "--pytorch_dump_folder_path", default=None, type=str, required=True, help="Path to the output PyTorch model."
	)
	parser.add_argument(
	    "--classification_head", action="store_true", help="Whether to convert a final classification head."
	)
	A  :     Optional[Any]    					=    parser.parse_args()
	convert_xmod_checkpoint_to_pytorch(
	    args.xmod_checkpoint_path, args.pytorch_dump_folder_path, args.classification_head
	)








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







							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.0_2					,     __a="divided_space_time"					,     __a=None					,     ):
										__lowerCAmelCase       					=    parent
										__lowerCAmelCase       					=    batch_size
										__lowerCAmelCase       					=    image_size
										__lowerCAmelCase       					=    num_channels
										__lowerCAmelCase       					=    patch_size
										__lowerCAmelCase       					=    num_frames
										__lowerCAmelCase       					=    is_training
										__lowerCAmelCase       					=    use_labels
										__lowerCAmelCase       					=    hidden_size
										__lowerCAmelCase       					=    num_hidden_layers
										__lowerCAmelCase       					=    num_attention_heads
										__lowerCAmelCase       					=    intermediate_size
										__lowerCAmelCase       					=    hidden_act
										__lowerCAmelCase       					=    hidden_dropout_prob
										__lowerCAmelCase       					=    attention_probs_dropout_prob
										__lowerCAmelCase       					=    attention_type
										__lowerCAmelCase       					=    initializer_range
										__lowerCAmelCase       					=    scope
										__lowerCAmelCase       					=    num_labels

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

							def     snake_case       (  self			):
										__lowerCAmelCase       					=    floats_tensor(
										    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]			)

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

										__lowerCAmelCase       					=    self.get_config()

										return config, pixel_values, labels

							def     snake_case       (  self			):
										__lowerCAmelCase       					=    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					,     )
										__lowerCAmelCase       					=    self.num_labels
										return config

							def     snake_case       (  self					,     __a					,     __a					,     __a			):
										__lowerCAmelCase       					=    TimesformerModel(config=__a			)
										model.to(__a			)
										model.eval()
										__lowerCAmelCase       					=    model(__a			)
										self.parent.assertEqual(result.last_hidden_state.shape					,     (self.batch_size, self.seq_length, self.hidden_size)			)

							def     snake_case       (  self					,     __a					,     __a					,     __a			):
										__lowerCAmelCase       					=    TimesformerForVideoClassification(__a			)
										model.to(__a			)
										model.eval()

										__lowerCAmelCase       					=    model(__a			)

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



							def     snake_case       (  self			):
										__lowerCAmelCase       					=    self.prepare_config_and_inputs()
										__lowerCAmelCase					,			__lowerCAmelCase					,			__lowerCAmelCase       					=    config_and_inputs
										__lowerCAmelCase       					=    {"pixel_values": pixel_values}
										return config, inputs_dict
@require_torch
class 						_UpperCamelCase						(     lowerCAmelCase__ ,lowerCAmelCase__ ,unittest.TestCase	):
							'''simple docstring'''







							__UpperCAmelCase		:					Optional[int]          =(TimesformerModel, TimesformerForVideoClassification) if is_torch_available() else ()
							__UpperCAmelCase		:					Tuple          =(
							    {"""feature-extraction""": TimesformerModel, """video-classification""": TimesformerForVideoClassification}
							    if is_torch_available()
							    else {}
							)

							__UpperCAmelCase		:					Any          =False
							__UpperCAmelCase		:					Optional[int]          =False
							__UpperCAmelCase		:					Union[str, Any]          =False
							__UpperCAmelCase		:					Any          =False

							def     snake_case       (  self			):
										__lowerCAmelCase       					=    TimesformerModelTester(self			)
										__lowerCAmelCase       					=    ConfigTester(
										    self					,     config_class=__a					,     has_text_modality=__a					,     hidden_size=37			)

							def     snake_case       (  self					,     __a					,     __a					,     __a=False			):
										__lowerCAmelCase       					=    copy.deepcopy(__a			)

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

										return inputs_dict

							def     snake_case       (  self			):
										self.config_tester.run_common_tests()

							@unittest.skip(reason="TimeSformer does not use inputs_embeds"			)
							def     snake_case       (  self			):
										pass

							def     snake_case       (  self			):
										__lowerCAmelCase					,			__lowerCAmelCase       					=    self.model_tester.prepare_config_and_inputs_for_common()

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

							def     snake_case       (  self			):
										__lowerCAmelCase					,			__lowerCAmelCase       					=    self.model_tester.prepare_config_and_inputs_for_common()

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

													__lowerCAmelCase       					=    ["pixel_values"]
													self.assertListEqual(arg_names[:1]					,     __a			)

							def     snake_case       (  self			):
										__lowerCAmelCase       					=    self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_model(*__a			)

							def     snake_case       (  self			):
										__lowerCAmelCase       					=    self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_for_video_classification(*__a			)

							@slow
							def     snake_case       (  self			):
										for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
													__lowerCAmelCase       					=    TimesformerModel.from_pretrained(__a			)
													self.assertIsNotNone(__a			)

							def     snake_case       (  self			):
										if not self.has_attentions:
													pass

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

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

																__lowerCAmelCase       					=    True
																__lowerCAmelCase       					=    False
																__lowerCAmelCase       					=    True
																__lowerCAmelCase       					=    model_class(__a			)
																model.to(__a			)
																model.eval()
																with torch.no_grad():
																			__lowerCAmelCase       					=    model(**self._prepare_for_class(__a					,     __a			)			)
																__lowerCAmelCase       					=    outputs.attentions
																self.assertEqual(len(__a			)					,     self.model_tester.num_hidden_layers			)

																# check that output_attentions also work using config
																del inputs_dict["output_attentions"]
																__lowerCAmelCase       					=    True
																__lowerCAmelCase       					=    model_class(__a			)
																model.to(__a			)
																model.eval()
																with torch.no_grad():
																			__lowerCAmelCase       					=    model(**self._prepare_for_class(__a					,     __a			)			)
																__lowerCAmelCase       					=    outputs.attentions
																self.assertEqual(len(__a			)					,     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]					,     )
																__lowerCAmelCase       					=    len(__a			)

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

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

																__lowerCAmelCase       					=    outputs.attentions

																self.assertEqual(len(__a			)					,     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			):
										def check_hidden_states_output(__a					,     __a					,     __a			):
													__lowerCAmelCase       					=    model_class(__a			)
													model.to(__a			)
													model.eval()

													with torch.no_grad():
																__lowerCAmelCase       					=    model(**self._prepare_for_class(__a					,     __a			)			)

													__lowerCAmelCase       					=    outputs.hidden_states
													__lowerCAmelCase       					=    self.model_tester.num_hidden_layers + 1
													self.assertEqual(len(__a			)					,     __a			)

													__lowerCAmelCase       					=    self.model_tester.seq_length

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

										__lowerCAmelCase					,			__lowerCAmelCase       					=    self.model_tester.prepare_config_and_inputs_for_common()

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

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

													check_hidden_states_output(__a					,     __a					,     __a			)


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(_UpperCamelCase   )
			return list(_UpperCamelCase   )


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







							@cached_property
							def     snake_case       (  self			):
										# 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     snake_case       (  self			):
										__lowerCAmelCase       					=    TimesformerForVideoClassification.from_pretrained("facebook/timesformer-base-finetuned-k400"			).to(
										    __a			)

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

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

										# verify the logits
										__lowerCAmelCase       					=    torch.Size((1, 4_00)			)
										self.assertEqual(outputs.logits.shape					,     __a			)

										__lowerCAmelCase       					=    torch.tensor([-0.3_0_1_6, -0.7_7_1_3, -0.4_2_0_5]			).to(__a			)

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




