



import argparse

# utilities to import the model weights and config file
import os
import pickle as pkl

# PyTorch + new model classes
import torch
from torch import nn

from transformers import AutoTokenizer, MegaConfig, MegaForMaskedLM


# import the EncoderLayer class used to pretrain
# !! NOTE !! this requires the version of fairseq that is built when you install the Mega source
try:
						from fairseq.modules.mega_layer import MegaEncoderLayer
except ImportError:
						raise ImportError('''You need to install the version of fairseq from the Mega repo!''')



class     __snake_case						(					nn.Module						):




							def __init__(	self					:	Optional[Any]    ,    A_					:	Union[str, Any]    ,    A_					:	Optional[int]    ,    A_					:	int):
									super().__init__()
									lowerCAmelCase_      :					Union[str, Any]     =			mega_args
									lowerCAmelCase_      :					Optional[Any]     =			nn.Embedding(A_    ,    self.mega_args.encoder_embed_dim)
									lowerCAmelCase_      :					Optional[int]     =			nn.ModuleList([MegaEncoderLayer(self.mega_args) for _ in range(A_)])
									lowerCAmelCase_      :					Optional[int]     =			depth







							def        UpperCAmelCase__      (	self					:	Union[str, Any]    ,    A_					:	List[str]    ,    A_					:	List[str]    ,    A_					:	List[str]=True    ,    A_					:	Optional[int]=0):

									# Mega expects embeddings to be (time, batch, embedding size), but
									# Hugging Face returns tokens as (batch, time)
									if batch_first:
											lowerCAmelCase_      :					Optional[Any]     =			input_ids.T

									# to make things more confusing, Mega expects the attention mask to
									# be (batch, time), but with values of 0 (normal token) and 1 (ignore token)
									# which is the opposite of what HF returns
									if ignore_mask_value == 0:
											lowerCAmelCase_      :					Optional[Any]     =			1 - attention_mask

									# get token embeddings from IDs
									lowerCAmelCase_      :					Any     =			self.embedding_layer(A_)

									# pass through the Mega layers
									# input is (time, batch, encoder dim) and output is the same
									for encoder in self.encoders:
											lowerCAmelCase_      :					str     =			encoder(A_    ,    A_)

									# return according to the shape specified
									if batch_first:
											# (T, B, H) --> (B, T, H)
											return torch.transpose(A_    ,    0    ,    1)
									else:
											return embeds





class     __snake_case						(					nn.Module						):




							def __init__(	self					:	str    ,    A_					:	Tuple    ,    A_					:	Any    ,    A_					:	List[str]):
									super().__init__()
									lowerCAmelCase_      :					Optional[Any]     =			MegaLM(A_    ,    A_    ,    A_)
									lowerCAmelCase_      :					Any     =			nn.Linear(mega_args.encoder_embed_dim    ,    A_)
									lowerCAmelCase_      :					List[Any]     =			nn.Dropout(p=0.1)







							def        UpperCAmelCase__      (	self					:	Optional[Any]    ,    A_					:	Any    ,    A_					:	List[str]    ,    A_					:	Dict=True    ,    A_					:	str=0):
									lowerCAmelCase_      :					int     =			self.mega(A_    ,    A_    ,    A_    ,    A_)
									return self.mlm_head(self.dropout(A_))





def       UpperCamelCase(  __UpperCamelCase							:		int					,__UpperCamelCase							:		int					,__UpperCamelCase							:		List[Any]			):
		with open(os.path.join(__UpperCamelCase					,'''model_args.pkl'''			)					,'''rb'''			) as f:
				lowerCAmelCase_      :					Dict     =			pkl.load(__UpperCamelCase			)

		# load the original encoder
		lowerCAmelCase_      :					int     =			OriginalMegaForMaskedLM(**__UpperCamelCase			).eval()

		# load its weights
		print(
		    '''Original Mega encoder:'''					,original_mlm.mega.load_state_dict(
		        torch.load(os.path.join(__UpperCamelCase					,'''encoder_weights.pt'''			)					,map_location='''cpu'''			)			)					,)
		print(
		    '''Original Mega MLM layer:'''					,original_mlm.mlm_head.load_state_dict(
		        torch.load(os.path.join(__UpperCamelCase					,'''mlm_head_weights.pt'''			)					,map_location='''cpu'''			)			)					,)

		# create a new config from the old one
		lowerCAmelCase_      :					Any     =			MegaConfig(
		    num_hidden_layers=mega_original_args['''depth''']					,vocab_size=mega_original_args['''vocab_size''']					,hidden_size=mega_original_args['''mega_args'''].encoder_embed_dim					,shared_representation_size=mega_original_args['''mega_args'''].encoder_z_dim					,intermediate_size=mega_original_args['''mega_args'''].encoder_hidden_dim					,ema_projection_size=mega_original_args['''mega_args'''].encoder_n_dim					,dropout_prob=mega_original_args['''mega_args'''].dropout					,attention_probs_dropout_prob=mega_original_args['''mega_args'''].attention_dropout					,hidden_dropout_prob=mega_original_args['''mega_args'''].hidden_dropout					,activation=mega_original_args['''mega_args'''].activation_fn					,attention_activation=mega_original_args['''mega_args'''].attention_activation_fn					,bidirectional=mega_original_args['''mega_args'''].bidirectional					,use_chunking=mega_original_args['''mega_args'''].encoder_chunk_size > 0					,chunk_size=mega_original_args['''mega_args'''].encoder_chunk_size					,truncation=mega_original_args['''mega_args'''].truncation_length					,normalization_type=mega_original_args['''mega_args'''].normalization_type					,normalize_before_mega=__UpperCamelCase					,norm_affine=__UpperCamelCase					,use_feature_dropout=mega_original_args['''mega_args'''].feature_dropout					,relative_positional_bias=mega_original_args['''mega_args'''].rel_pos_bias					,max_positions=mega_original_args['''mega_args'''].max_source_positions					,nffn_hidden_size=mega_original_args['''mega_args'''].encoder_ffn_embed_dim					,normalize_before_ffn=mega_original_args['''mega_args'''].normalize_before					,nffn_activation_dropout_prob=0.0					,add_token_type_embeddings=__UpperCamelCase					,add_lm_hidden_dense_layer=__UpperCamelCase					,)

		lowerCAmelCase_      :					Tuple     =			MegaForMaskedLM(__UpperCamelCase			).eval()

		# the originl checkpoint just uses nn.Embedding for the word embeddings
		# we use a wrapper module for embeddings to add support for positional embeddings
		lowerCAmelCase_      :					List[Any]     =			original_mlm.mega.embedding_layer.weight

		# modify the state dictionary of the original checkpoint to account for naming issues in the Hugging Face
		# ecosystem -- any names containing "beta" or "gamma" aren't safe to use and are renamed upon _load_pretrained,
		# also renaming previously confusing parameter names
		lowerCAmelCase_      :					List[Any]     =			original_mlm.mega.encoders.state_dict()
		lowerCAmelCase_      :					Dict     =			{}
		for module_name in original_state_dict.keys():
				lowerCAmelCase_      :					Optional[Any]     =			None
				# have to handle gamma, beta, and alpha differently due to their use
				# in multiple modules within the original repository;
				# beta is used in EMA, MovingAverageGatedAttention, and RotaryRelativePositionalBias, and must be renamed due to flax/tf weights
				# the EMA sublayer was renamed from "move" to "ema_gate" for readability, so that is also done here
				if "beta" in module_name:
						# EMA sub-layers were always called "move" in the original repo
						if "move.beta" in module_name:
								lowerCAmelCase_      :					Optional[int]     =			module_name.replace('''move.beta'''					,'''ema_gate.ema_expansion_matrix'''			)
						elif "mega_layer.beta" in module_name:
								lowerCAmelCase_      :					Dict     =			module_name.replace('''beta'''					,'''qk_bias'''			)
						else:
								lowerCAmelCase_      :					Union[str, Any]     =			module_name.replace('''beta'''					,'''b_param'''			)
        # beta is used in EMA and MovingAverageGatedAttention, and must be renamed due to flax/tf weights
				elif "gamma" in module_name:
						if "move.gamma" in module_name:
								lowerCAmelCase_      :					Optional[int]     =			module_name.replace('''move.gamma'''					,'''ema_gate.kernel_projection_matrix'''			)
						elif "mega_layer.gamma" in module_name:
								lowerCAmelCase_      :					Tuple     =			module_name.replace('''gamma'''					,'''qk_weight'''			)
						else:
								lowerCAmelCase_      :					Tuple     =			module_name.replace('''gamma'''					,'''g_param'''			)
        # alpha is used in EMA and positional bias; renaming to improve readability
				elif "move.alpha" in module_name:
						lowerCAmelCase_      :					str     =			module_name.replace('''move.alpha'''					,'''ema_gate.decay_factor'''			)
				# delta is only used in EMA; renaming to improve readability
				elif "move.delta" in module_name:
						lowerCAmelCase_      :					List[Any]     =			module_name.replace('''move.delta'''					,'''ema_gate.damping_factor'''			)
				# omega is only used in EMA; renaming to improve readability
				elif "omega" in module_name:
						lowerCAmelCase_      :					Any     =			module_name.replace('''move.omega'''					,'''ema_gate.residual_weight'''			)

				if new_module_name:
						lowerCAmelCase_      :					List[str]     =			new_module_name

		if len(__UpperCamelCase			) != 0:
				print(f"""Renaming these keys: {updated_keys.keys()}"""			)
		else:
				print('''No need to rename state dict entries'''			)
		for old, new in updated_keys.items():
				lowerCAmelCase_      :					Any     =			original_state_dict.pop(__UpperCamelCase			)

		# now attempt to load the state dictionary with updated names
		# note that we now call it `mega.layers` instead of `mega.encoders` due to hugging face style
		print('''HF Mega encoder:'''					,hf_mlm.mega.layers.load_state_dict(__UpperCamelCase			)			)

		# load the MLM head weights directly
		print(
		    '''HF Mega MLM layer:'''					,hf_mlm.mlm_head.load_state_dict(
		        torch.load(os.path.join(__UpperCamelCase					,'''mlm_head_weights.pt'''			)					,map_location='''cpu'''			)			)					,)

		# test on a randomly generated input sequence
		lowerCAmelCase_      :					Dict     =			torch.randint(0					,hf_config.vocab_size					,size=(4, 256)			)
		lowerCAmelCase_      :					Optional[Any]     =			torch.ones_like(__UpperCamelCase			)
		# mask a few tokens to make sure masking is applied appropriately :)
		lowerCAmelCase_      :					Optional[Any]     =			0

		# run forward passes
		lowerCAmelCase_      :					str     =			original_mlm(__UpperCamelCase					,__UpperCamelCase					,batch_first=__UpperCamelCase					,ignore_mask_value=0			)
		lowerCAmelCase_      :					Optional[int]     =			hf_mlm(__UpperCamelCase					,__UpperCamelCase			)[0]

		# print shapes and diff
		print(f"""original output {original_output.shape}"""			)
		print(f"""hf output {hf_output.shape}"""			)
		print(f"""max diff: {(original_output - hf_output).max()}"""			)  # 0.0
		lowerCAmelCase_      :					Any     =			torch.allclose(__UpperCamelCase					,__UpperCamelCase					,atol=1e-3			)

		if success:
				print('''Yay!'''			)
				hf_mlm.save_pretrained(__UpperCamelCase			)
		else:
				raise RuntimeError(f"""Something's broken :(\nOriginal:\n{original_output}\n\nHF\n{hf_output}\n{hf_mlm}"""			)

		if includes_tokenizer:
				print('''Transferring tokenizer'''			)
				lowerCAmelCase_      :					Optional[int]     =			AutoTokenizer.from_pretrained(__UpperCamelCase			)
				tokenizer.save_pretrained(__UpperCamelCase			)


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

						parser.add_argument(
						    '''--pretrained_checkpoint_path''',
						    default=None,
						    type=str,
						    required=True,
						    help='''Point to the directory containing your model weights using the official Mega repo''',
						)

						parser.add_argument(
						    '''--output_path''', default=None, type=str, required=True, help='''Location to save the Hugging Face version'''
						)

						parser.add_argument(
						    '''--includes_tokenizer''',
						    action='''store_true''',
						    help='''Use this flag if there is a Hugging Face tokenizer in the original checkpoint repo''',
						)

						A__  :  Any       = parser.parse_args()

						convert_checkpoint_to_huggingface(args.pretrained_checkpoint_path, args.output_path, args.includes_tokenizer)










import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_video_inputs


if is_torch_available():
						import torch

if is_vision_available():
						from PIL import Image

						from transformers import VideoMAEImageProcessor



class     __snake_case						(					unittest.TestCase						):




							def __init__(	self					:	str    ,    A_					:	str    ,    A_					:	int=7    ,    A_					:	List[Any]=3    ,    A_					:	Dict=1_0    ,    A_					:	List[str]=1_8    ,    A_					:	List[Any]=3_0    ,    A_					:	str=4_0_0    ,    A_					:	str=True    ,    A_					:	Tuple=None    ,    A_					:	Optional[Any]=True    ,    A_					:	Tuple=[0.5, 0.5, 0.5]    ,    A_					:	Any=[0.5, 0.5, 0.5]    ,    A_					:	List[str]=None    ,    ):
									lowerCAmelCase_      :					Optional[Any]     =			size if size is not None else {'''shortest_edge''': 1_8}
									lowerCAmelCase_      :					Optional[int]     =			crop_size if crop_size is not None else {'''height''': 1_8, '''width''': 1_8}

									lowerCAmelCase_      :					Any     =			parent
									lowerCAmelCase_      :					Optional[Any]     =			batch_size
									lowerCAmelCase_      :					List[Any]     =			num_channels
									lowerCAmelCase_      :					Optional[int]     =			num_frames
									lowerCAmelCase_      :					Optional[int]     =			image_size
									lowerCAmelCase_      :					List[Any]     =			min_resolution
									lowerCAmelCase_      :					str     =			max_resolution
									lowerCAmelCase_      :					List[str]     =			do_resize
									lowerCAmelCase_      :					int     =			size
									lowerCAmelCase_      :					Union[str, Any]     =			do_normalize
									lowerCAmelCase_      :					Union[str, Any]     =			image_mean
									lowerCAmelCase_      :					List[str]     =			image_std
									lowerCAmelCase_      :					Any     =			crop_size







							def        UpperCAmelCase__      (	self					:	Any):
									return {
									    "image_mean": self.image_mean,
									    "image_std": self.image_std,
									    "do_normalize": self.do_normalize,
									    "do_resize": self.do_resize,
									    "size": self.size,
									    "crop_size": self.crop_size,
									}





@require_torch
@require_vision
class     __snake_case						(					UpperCamelCase_						,unittest.TestCase						):
							_a      				=			VideoMAEImageProcessor if is_vision_available() else None




							def        UpperCAmelCase__      (	self					:	Tuple):
									lowerCAmelCase_      :					int     =			VideoMAEImageProcessingTester(self)




							@property
							def        UpperCAmelCase__      (	self					:	Tuple):
									return self.image_processor_tester.prepare_image_processor_dict()




							def        UpperCAmelCase__      (	self					:	Optional[int]):
									lowerCAmelCase_      :					int     =			self.image_processing_class(**self.image_processor_dict)
									self.assertTrue(hasattr(A_    ,    '''image_mean'''))
									self.assertTrue(hasattr(A_    ,    '''image_std'''))
									self.assertTrue(hasattr(A_    ,    '''do_normalize'''))
									self.assertTrue(hasattr(A_    ,    '''do_resize'''))
									self.assertTrue(hasattr(A_    ,    '''do_center_crop'''))
									self.assertTrue(hasattr(A_    ,    '''size'''))




							def        UpperCAmelCase__      (	self					:	Dict):
									lowerCAmelCase_      :					int     =			self.image_processing_class.from_dict(self.image_processor_dict)
									self.assertEqual(image_processor.size    ,    {'''shortest_edge''': 1_8})
									self.assertEqual(image_processor.crop_size    ,    {'''height''': 1_8, '''width''': 1_8})

									lowerCAmelCase_      :					str     =			self.image_processing_class.from_dict(self.image_processor_dict    ,    size=4_2    ,    crop_size=8_4)
									self.assertEqual(image_processor.size    ,    {'''shortest_edge''': 4_2})
									self.assertEqual(image_processor.crop_size    ,    {'''height''': 8_4, '''width''': 8_4})




							def        UpperCAmelCase__      (	self					:	Dict):
									pass




							def        UpperCAmelCase__      (	self					:	str):
									# Initialize image_processing
									lowerCAmelCase_      :					int     =			self.image_processing_class(**self.image_processor_dict)
									# create random PIL videos
									lowerCAmelCase_      :					Optional[int]     =			prepare_video_inputs(self.image_processor_tester    ,    equal_resolution=A_)
									for video in video_inputs:
											self.assertIsInstance(A_    ,    A_)
											self.assertIsInstance(video[0]    ,    Image.Image)

									# Test not batched input
									lowerCAmelCase_      :					int     =			image_processing(video_inputs[0]    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        1,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )

									# Test batched
									lowerCAmelCase_      :					Tuple     =			image_processing(A_    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )




							def        UpperCAmelCase__      (	self					:	int):
									# Initialize image_processing
									lowerCAmelCase_      :					Dict     =			self.image_processing_class(**self.image_processor_dict)
									# create random numpy tensors
									lowerCAmelCase_      :					str     =			prepare_video_inputs(self.image_processor_tester    ,    equal_resolution=A_    ,    numpify=A_)
									for video in video_inputs:
											self.assertIsInstance(A_    ,    A_)
											self.assertIsInstance(video[0]    ,    np.ndarray)

									# Test not batched input
									lowerCAmelCase_      :					Tuple     =			image_processing(video_inputs[0]    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        1,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )

									# Test batched
									lowerCAmelCase_      :					Union[str, Any]     =			image_processing(A_    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )







							def        UpperCAmelCase__      (	self					:	List[str]):
									# Initialize image_processing
									lowerCAmelCase_      :					Tuple     =			self.image_processing_class(**self.image_processor_dict)
									# create random PyTorch tensors
									lowerCAmelCase_      :					Tuple     =			prepare_video_inputs(self.image_processor_tester    ,    equal_resolution=A_    ,    torchify=A_)
									for video in video_inputs:
											self.assertIsInstance(A_    ,    A_)
											self.assertIsInstance(video[0]    ,    torch.Tensor)

									# Test not batched input
									lowerCAmelCase_      :					List[str]     =			image_processing(video_inputs[0]    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        1,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )

									# Test batched
									lowerCAmelCase_      :					List[Any]     =			image_processing(A_    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )





