





"""simple docstring"""




import argparse
import json

import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image

from transformers import ViTImageProcessor, ViTMSNConfig, ViTMSNModel
from transformers.image_utils import IMAGENET_DEFAULT_MEAN, IMAGENET_DEFAULT_STD


torch.set_grad_enabled(False)






def       _SCREAMING_SNAKE_CASE		(	_lowercase						:    List[Any]    ,       _lowercase						:    Tuple=False					) ->List[str]:






	'''simple docstring'''






	a      :		Dict	=				[]
	for i in range(config.num_hidden_layers					):
		# encoder layers: output projection, 2 feedforward neural networks and 2 layernorms
		rename_keys.append((F"""module.blocks.{i}.norm1.weight""", F"""vit.encoder.layer.{i}.layernorm_before.weight""")					)
		rename_keys.append((F"""module.blocks.{i}.norm1.bias""", F"""vit.encoder.layer.{i}.layernorm_before.bias""")					)
		rename_keys.append(
		    (F"""module.blocks.{i}.attn.proj.weight""", F"""vit.encoder.layer.{i}.attention.output.dense.weight""")					)
		rename_keys.append((F"""module.blocks.{i}.attn.proj.bias""", F"""vit.encoder.layer.{i}.attention.output.dense.bias""")					)
		rename_keys.append((F"""module.blocks.{i}.norm2.weight""", F"""vit.encoder.layer.{i}.layernorm_after.weight""")					)
		rename_keys.append((F"""module.blocks.{i}.norm2.bias""", F"""vit.encoder.layer.{i}.layernorm_after.bias""")					)
		rename_keys.append((F"""module.blocks.{i}.mlp.fc1.weight""", F"""vit.encoder.layer.{i}.intermediate.dense.weight""")					)
		rename_keys.append((F"""module.blocks.{i}.mlp.fc1.bias""", F"""vit.encoder.layer.{i}.intermediate.dense.bias""")					)
		rename_keys.append((F"""module.blocks.{i}.mlp.fc2.weight""", F"""vit.encoder.layer.{i}.output.dense.weight""")					)
		rename_keys.append((F"""module.blocks.{i}.mlp.fc2.bias""", F"""vit.encoder.layer.{i}.output.dense.bias""")					)

	# projection layer + position embeddings
	rename_keys.extend(
	    [
	        ("module.cls_token", "vit.embeddings.cls_token"),
	        ("module.patch_embed.proj.weight", "vit.embeddings.patch_embeddings.projection.weight"),
	        ("module.patch_embed.proj.bias", "vit.embeddings.patch_embeddings.projection.bias"),
	        ("module.pos_embed", "vit.embeddings.position_embeddings"),
	    ]					)

	if base_model:
		# layernorm + pooler
		rename_keys.extend(
		    [
		        ("module.norm.weight", "layernorm.weight"),
		        ("module.norm.bias", "layernorm.bias"),
		    ]					)

		# if just the base model, we should remove "vit" from all keys that start with "vit"
		a      :		int	=				[(pair[0], pair[1][4:]) if pair[1].startswith("vit"					) else pair for pair in rename_keys]
	else:
		# layernorm + classification head
		rename_keys.extend(
		    [
		        ("norm.weight", "vit.layernorm.weight"),
		        ("norm.bias", "vit.layernorm.bias"),
		        ("head.weight", "classifier.weight"),
		        ("head.bias", "classifier.bias"),
		    ]					)

	return rename_keys






def       _SCREAMING_SNAKE_CASE		(	_lowercase						:    Optional[int]    ,       _lowercase						:    int    ,       _lowercase						:    List[Any]=False					) ->int:






	'''simple docstring'''






	for i in range(config.num_hidden_layers					):
		if base_model:
			a      :		List[Any]	=				""
		else:
			a      :		int	=				"vit."
		# read in weights + bias of input projection layer (in timm, this is a single matrix + bias)
		a      :		Optional[int]	=				state_dict.pop(F"""module.blocks.{i}.attn.qkv.weight"""					)
		a      :		Tuple	=				state_dict.pop(F"""module.blocks.{i}.attn.qkv.bias"""					)
		# next, add query, keys and values (in that order) to the state dict
		a      :		int	=				in_proj_weight[
		    : config.hidden_size, :
		]
		a      :		Optional[Any]	=				in_proj_bias[: config.hidden_size]
		a      :		str	=				in_proj_weight[
		    config.hidden_size : config.hidden_size * 2, :
		]
		a      :		Optional[int]	=				in_proj_bias[
		    config.hidden_size : config.hidden_size * 2
		]
		a      :		Dict	=				in_proj_weight[
		    -config.hidden_size :, :
		]
		a      :		Optional[int]	=				in_proj_bias[-config.hidden_size :]






def       _SCREAMING_SNAKE_CASE		(	_lowercase						:    Any					) ->Any:






	'''simple docstring'''






	a      :		Union[str, Any]	=				["head.weight", "head.bias"]
	for k in ignore_keys:
		state_dict.pop(_lowercase    ,       _lowercase					)






def       _SCREAMING_SNAKE_CASE		(	_lowercase						:    Dict					) ->Tuple:






	'''simple docstring'''






	a      :		Optional[int]	=				[
	    "module.fc.fc1.weight",
	    "module.fc.fc1.bias",
	    "module.fc.bn1.weight",
	    "module.fc.bn1.bias",
	    "module.fc.bn1.running_mean",
	    "module.fc.bn1.running_var",
	    "module.fc.bn1.num_batches_tracked",
	    "module.fc.fc2.weight",
	    "module.fc.fc2.bias",
	    "module.fc.bn2.weight",
	    "module.fc.bn2.bias",
	    "module.fc.bn2.running_mean",
	    "module.fc.bn2.running_var",
	    "module.fc.bn2.num_batches_tracked",
	    "module.fc.fc3.weight",
	    "module.fc.fc3.bias",
	]
	for k in ignore_keys:
		state_dict.pop(_lowercase    ,       _lowercase					)






def       _SCREAMING_SNAKE_CASE		(	_lowercase						:    List[Any]    ,       _lowercase						:    str    ,       _lowercase						:    Any					) ->List[Any]:






	'''simple docstring'''






	a      :		Union[str, Any]	=				dct.pop(_lowercase					)
	a      :		Optional[Any]	=				val






def       _SCREAMING_SNAKE_CASE		(	_lowercase						:    str    ,       _lowercase						:    Dict					) ->List[str]:






	'''simple docstring'''






	a      :		Tuple	=				ViTMSNConfig()
	a      :		List[Any]	=				1000

	a      :		str	=				"datasets/huggingface/label-files"
	a      :		Optional[Any]	=				"imagenet-1k-id2label.json"
	a      :		str	=				json.load(open(hf_hub_download(_lowercase    ,       _lowercase					)    ,       "r"					)					)
	a      :		Optional[Any]	=				{int(_lowercase					): v for k, v in idalabel.items()}
	a      :		Dict	=				idalabel
	a      :		Union[str, Any]	=				{v: k for k, v in idalabel.items()}

	if "s16" in checkpoint_url:
		a      :		Any	=				384
		a      :		Any	=				1536
		a      :		Optional[int]	=				6
	elif "l16" in checkpoint_url:
		a      :		str	=				1024
		a      :		Any	=				4096
		a      :		str	=				24
		a      :		List[str]	=				16
		a      :		Any	=				0.1
	elif "b4" in checkpoint_url:
		a      :		Optional[Any]	=				4
	elif "l7" in checkpoint_url:
		a      :		Dict	=				7
		a      :		int	=				1024
		a      :		List[str]	=				4096
		a      :		Optional[int]	=				24
		a      :		int	=				16
		a      :		List[Any]	=				0.1

	a      :		List[str]	=				ViTMSNModel(_lowercase					)

	a      :		Union[str, Any]	=				torch.hub.load_state_dict_from_url(_lowercase    ,       map_location="cpu"					)["target_encoder"]

	a      :		List[str]	=				ViTImageProcessor(size=config.image_size					)

	remove_projection_head(_lowercase					)
	a      :		int	=				create_rename_keys(_lowercase    ,       base_model=_lowercase					)

	for src, dest in rename_keys:
		rename_key(_lowercase    ,       _lowercase    ,       _lowercase					)
	read_in_q_k_v(_lowercase    ,       _lowercase    ,       base_model=_lowercase					)

	model.load_state_dict(_lowercase					)
	model.eval()

	a      :		str	=				"http://images.cocodataset.org/val2017/000000039769.jpg"

	a      :		Union[str, Any]	=				Image.open(requests.get(_lowercase    ,       stream=_lowercase					).raw					)
	a      :		List[Any]	=				ViTImageProcessor(
	    size=config.image_size    ,       image_mean=_lowercase    ,       image_std=_lowercase					)
	a      :		List[Any]	=				image_processor(images=_lowercase    ,       return_tensors="pt"					)

	# forward pass
	torch.manual_seed(2					)
	a      :		List[Any]	=				model(**_lowercase					)
	a      :		Optional[Any]	=				outputs.last_hidden_state

	# The following Colab Notebook was used to generate these outputs:
	# https://colab.research.google.com/gist/sayakpaul/3672419a04f5997827503fd84079bdd1/scratchpad.ipynb
	if "s16" in checkpoint_url:
		a      :		Dict	=				torch.tensor([[-1.0915, -1.4876, -1.1809]]					)
	elif "b16" in checkpoint_url:
		a      :		Union[str, Any]	=				torch.tensor([[14.2889, -18.9045, 11.7281]]					)
	elif "l16" in checkpoint_url:
		a      :		Any	=				torch.tensor([[41.5028, -22.8681, 45.6475]]					)
	elif "b4" in checkpoint_url:
		a      :		Optional[Any]	=				torch.tensor([[-4.3868, 5.2932, -0.4137]]					)
	else:
		a      :		Tuple	=				torch.tensor([[-0.1792, -0.6465, 2.4263]]					)

	# verify logits
	assert torch.allclose(last_hidden_state[:, 0, :3]    ,       _lowercase    ,       atol=1E-4					)

	print(F"""Saving model to {pytorch_dump_folder_path}"""					)
	model.save_pretrained(_lowercase					)

	print(F"""Saving image processor to {pytorch_dump_folder_path}"""					)
	image_processor.save_pretrained(_lowercase					)


if __name__ == "__main__":
		a						: Optional[Any]     	=			argparse.ArgumentParser()
		# Required parameters
		parser.add_argument(
		    '''--checkpoint_url''',
		    default='''https://dl.fbaipublicfiles.com/msn/vits16_800ep.pth.tar''',
		    type=str,
		    help='''URL of the checkpoint you\'d like to convert.''',
		)
		parser.add_argument(
		    '''--pytorch_dump_folder_path''', default=None, type=str, help='''Path to the output PyTorch model directory.'''
		)

		a						: Optional[Any]     	=			parser.parse_args()
		convert_vit_msn_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path)




'''simple docstring'''







import tensorflow as tf

from ...tf_utils import shape_list


class 						__UpperCAmelCase      (					tf.keras.layers.Layer						):


							def __init__(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_=1    ,   lowerCAmelCase_=False    ,   **lowerCAmelCase_  ):





								"""simple docstring"""






								super().__init__(**lowerCAmelCase_  )

								_snake_case   						=		vocab_size
								_snake_case   						=		d_embed
								_snake_case   						=		d_proj

								_snake_case   						=		cutoffs + [vocab_size]
								_snake_case   						=		[0] + self.cutoffs
								_snake_case   						=		div_val

								_snake_case   						=		self.cutoffs[0]
								_snake_case   						=		len(self.cutoffs  ) - 1
								_snake_case   						=		self.shortlist_size + self.n_clusters
								_snake_case   						=		keep_order

								_snake_case   						=		[]
								_snake_case   						=		[]


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								if self.n_clusters > 0:
									_snake_case   						=		self.add_weight(
									    shape=(self.n_clusters, self.d_embed)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name='cluster_weight'  )
									_snake_case   						=		self.add_weight(
									    shape=(self.n_clusters,)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name='cluster_bias'  )

								if self.div_val == 1:
									for i in range(len(self.cutoffs  )  ):
										if self.d_proj != self.d_embed:
											_snake_case   						=		self.add_weight(
											    shape=(self.d_embed, self.d_proj)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_projs_._{i}'    ,   )
											self.out_projs.append(lowerCAmelCase_  )
										else:
											self.out_projs.append(lowerCAmelCase_  )
										_snake_case   						=		self.add_weight(
										    shape=(self.vocab_size, self.d_embed)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_layers_._{i}_._weight'    ,   )
										_snake_case   						=		self.add_weight(
										    shape=(self.vocab_size,)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_layers_._{i}_._bias'    ,   )
										self.out_layers.append((weight, bias)  )
								else:
									for i in range(len(self.cutoffs  )  ):
										_snake_case						, _snake_case   						=		self.cutoff_ends[i], self.cutoff_ends[i + 1]
										_snake_case   						=		self.d_embed // (self.div_val**i)

										_snake_case   						=		self.add_weight(
										    shape=(d_emb_i, self.d_proj)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_projs_._{i}'  )
										self.out_projs.append(lowerCAmelCase_  )
										_snake_case   						=		self.add_weight(
										    shape=(r_idx - l_idx, d_emb_i)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_layers_._{i}_._weight'    ,   )
										_snake_case   						=		self.add_weight(
										    shape=(r_idx - l_idx,)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_layers_._{i}_._bias'    ,   )
										self.out_layers.append((weight, bias)  )
								super().build(lowerCAmelCase_  )


							@staticmethod
							def  lowerCamelCase				(       lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_=None  ):





								"""simple docstring"""






								_snake_case   						=		x
								if proj is not None:
									_snake_case   						=		tf.einsum('ibd,ed->ibe'    ,   lowerCAmelCase_    ,   lowerCAmelCase_  )
								return tf.einsum('ibd,nd->ibn'    ,   lowerCAmelCase_    ,   lowerCAmelCase_  ) + b


							@staticmethod
							def  lowerCamelCase				(       lowerCAmelCase_    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								_snake_case   						=		shape_list(lowerCAmelCase_  )
								_snake_case   						=		tf.range(lp_size[0]    ,   dtype=target.dtype  )
								_snake_case   						=		tf.stack([r, target]    ,   1  )
								return tf.gather_nd(lowerCAmelCase_    ,   lowerCAmelCase_  )




							def  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_=True    ,   lowerCAmelCase_=False  ):





								"""simple docstring"""






								_snake_case   						=		0
								if self.n_clusters == 0:
									_snake_case   						=		self._logit(lowerCAmelCase_    ,   self.out_layers[0][0]    ,   self.out_layers[0][1]    ,   self.out_projs[0]  )
									if target is not None:
										_snake_case   						=		tf.nn.sparse_softmax_cross_entropy_with_logits(labels=lowerCAmelCase_    ,   logits=lowerCAmelCase_  )
									_snake_case   						=		tf.nn.log_softmax(lowerCAmelCase_    ,   axis=-1  )
								else:
									_snake_case   						=		shape_list(lowerCAmelCase_  )
									_snake_case   						=		[]
									_snake_case   						=		tf.zeros(hidden_sizes[:2]  )
									for i in range(len(self.cutoffs  )  ):
										_snake_case						, _snake_case   						=		self.cutoff_ends[i], self.cutoff_ends[i + 1]
										if target is not None:
											_snake_case   						=		(target >= l_idx) & (target < r_idx)
											_snake_case   						=		tf.where(lowerCAmelCase_  )
											_snake_case   						=		tf.boolean_mask(lowerCAmelCase_    ,   lowerCAmelCase_  ) - l_idx

										if self.div_val == 1:
											_snake_case   						=		self.out_layers[0][0][l_idx:r_idx]
											_snake_case   						=		self.out_layers[0][1][l_idx:r_idx]
										else:
											_snake_case   						=		self.out_layers[i][0]
											_snake_case   						=		self.out_layers[i][1]

										if i == 0:
											_snake_case   						=		tf.concat([cur_W, self.cluster_weight]    ,   0  )
											_snake_case   						=		tf.concat([cur_b, self.cluster_bias]    ,   0  )

											_snake_case   						=		self._logit(lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   self.out_projs[0]  )
											_snake_case   						=		tf.nn.log_softmax(lowerCAmelCase_  )
											out.append(head_logprob[..., : self.cutoffs[0]]  )
											if target is not None:
												_snake_case   						=		tf.boolean_mask(lowerCAmelCase_    ,   lowerCAmelCase_  )
												_snake_case   						=		self._gather_logprob(lowerCAmelCase_    ,   lowerCAmelCase_  )
										else:
											_snake_case   						=		self._logit(lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   self.out_projs[i]  )
											_snake_case   						=		tf.nn.log_softmax(lowerCAmelCase_  )
											_snake_case   						=		self.cutoffs[0] + i - 1  # No probability for the head cluster
											_snake_case   						=		head_logprob[..., cluster_prob_idx, None] + tail_logprob
											out.append(lowerCAmelCase_  )
											if target is not None:
												_snake_case   						=		tf.boolean_mask(lowerCAmelCase_    ,   lowerCAmelCase_  )
												_snake_case   						=		tf.boolean_mask(lowerCAmelCase_    ,   lowerCAmelCase_  )
												_snake_case   						=		self._gather_logprob(lowerCAmelCase_    ,   lowerCAmelCase_  )
												cur_logprob += cur_head_logprob[:, self.cutoff_ends[1] + i - 1]
										if target is not None:
											loss += tf.scatter_nd(lowerCAmelCase_    ,   -cur_logprob    ,   shape_list(lowerCAmelCase_  )  )
									_snake_case   						=		tf.concat(lowerCAmelCase_    ,   axis=-1  )

								if target is not None:
									if return_mean:
										_snake_case   						=		tf.reduce_mean(lowerCAmelCase_  )
									# Add the training-time loss value to the layer using `self.add_loss()`.
									self.add_loss(lowerCAmelCase_  )

									# Log the loss as a metric (we could log arbitrary metrics,
									# including different metrics for training and inference.
									self.add_metric(lowerCAmelCase_    ,   name=self.name    ,   aggregation='mean' if return_mean else ''  )

								return out

