




import argparse
import json
from pathlib import Path

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

from transformers import ViTConfig, ViTForImageClassification, ViTImageProcessor, ViTModel
from transformers.utils import logging


logging.set_verbosity_info()
lowerCAmelCase_		      = logging.get_logger(__name__)






def 					snake_case(				__magic_name__ ,		__magic_name__=False    ) ->							int:





     '''simple docstring'''


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

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

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

          # if just the base model, we should remove "vit" from all keys that start with "vit"
          lowercase					:   Optional[Any]								= [(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 					snake_case(				__magic_name__ ,		__magic_name__ ,		__magic_name__=False    ) ->							Union[str, Any]:





     '''simple docstring'''


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






def 					snake_case(				__magic_name__    ) ->							List[str]:





     '''simple docstring'''


     lowercase					:   Dict								= ['''head.weight''', '''head.bias''']
     for k in ignore_keys:
          state_dict.pop(__A ,		__A    )






def 					snake_case(				__magic_name__ ,		__magic_name__ ,		__magic_name__    ) ->							Optional[int]:





     '''simple docstring'''


     lowercase					:   List[Any]								= dct.pop(__A    )
     lowercase					:   Tuple								= val






def 					snake_case(				) ->							Any:





     '''simple docstring'''


     lowercase					:   str								= '''http://images.cocodataset.org/val2017/000000039769.jpg'''
     lowercase					:   List[str]								= Image.open(requests.get(__A ,		stream=__A    ).raw    )
     return im






@torch.no_grad()
def 					snake_case(				__magic_name__ ,		__magic_name__ ,		__magic_name__=True    ) ->							Optional[int]:





     '''simple docstring'''


     lowercase					:   Optional[int]								= ViTConfig()
     # patch_size
     if model_name[-1] == "8":
          lowercase					:   List[str]								= 8
     # set labels if required
     if not base_model:
          lowercase					:   Tuple								= 10_00
          lowercase					:   List[str]								= '''huggingface/label-files'''
          lowercase					:   Optional[int]								= '''imagenet-1k-id2label.json'''
          lowercase					:   Optional[Any]								= json.load(open(hf_hub_download(__A ,		__A ,		repo_type='''dataset'''    ) ,		'''r'''    )    )
          lowercase					:   Any								= {int(__A    ): v for k, v in idalabel.items()}
          lowercase					:   Optional[Any]								= idalabel
          lowercase					:   Any								= {v: k for k, v in idalabel.items()}
     # size of the architecture
     if model_name in ["dino_vits8", "dino_vits16"]:
          lowercase					:   List[str]								= 3_84
          lowercase					:   List[Any]								= 15_36
          lowercase					:   Dict								= 12
          lowercase					:   Union[str, Any]								= 6

     # load original model from torch hub
     lowercase					:   Any								= torch.hub.load('''facebookresearch/dino:main''' ,		__A    )
     original_model.eval()

     # load state_dict of original model, remove and rename some keys
     lowercase					:   Any								= original_model.state_dict()
     if base_model:
          remove_classification_head_(__A    )
     lowercase					:   List[Any]								= create_rename_keys(__A ,		base_model=__A    )
     for src, dest in rename_keys:
          rename_key(__A ,		__A ,		__A    )
     read_in_q_k_v(__A ,		__A ,		__A    )

     # load HuggingFace model
     if base_model:
          lowercase					:   List[str]								= ViTModel(__A ,		add_pooling_layer=__A    ).eval()
     else:
          lowercase					:   str								= ViTForImageClassification(__A    ).eval()
     model.load_state_dict(__A    )

     # Check outputs on an image, prepared by ViTImageProcessor
     lowercase					:   Union[str, Any]								= ViTImageProcessor()
     lowercase					:   Any								= image_processor(images=prepare_img() ,		return_tensors='''pt'''    )
     lowercase					:   Union[str, Any]								= encoding['''pixel_values''']
     lowercase					:   int								= model(__A    )

     if base_model:
          lowercase					:   str								= original_model(__A    )
          assert torch.allclose(__A ,		outputs.last_hidden_state[:, 0, :] ,		atol=1e-1    )
     else:
          lowercase					:   Tuple								= original_model(__A    )
          assert logits.shape == outputs.logits.shape
          assert torch.allclose(__A ,		outputs.logits ,		atol=1e-3    )

     Path(__A    ).mkdir(exist_ok=__A    )
     print(F"""Saving model {model_name} to {pytorch_dump_folder_path}"""    )
     model.save_pretrained(__A    )
     print(F"""Saving image processor to {pytorch_dump_folder_path}"""    )
     image_processor.save_pretrained(__A    )


if __name__ == "__main__":
 lowerCAmelCase_		      = argparse.ArgumentParser()
 # Required parameters
 parser.add_argument(
     '--model_name',
     default='dino_vitb16',
     type=str,
     help='Name of the model trained with DINO you\'d like to convert.',
 )
 parser.add_argument(
     '--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model directory.'
 )
 parser.add_argument(
     '--base_model',
     action='store_true',
     help='Whether to only convert the base model (no projection head weights).',
 )

 parser.set_defaults(base_model=True)
 lowerCAmelCase_		      = parser.parse_args()
 convert_vit_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.base_model)

'''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

