



"""simple docstring"""

import argparse

import requests
import torch
from PIL import Image

from transformers import ViTMAEConfig, ViTMAEForPreTraining, ViTMAEImageProcessor




def __lowercase    (							_a					):
    if "cls_token" in name:
        snake_case_				:  Union[str, Any]       =			name.replace('''cls_token'''	,    '''vit.embeddings.cls_token'''					)
    if "mask_token" in name:
        snake_case_				:  List[str]       =			name.replace('''mask_token'''	,    '''decoder.mask_token'''					)
    if "decoder_pos_embed" in name:
        snake_case_				:  Optional[Any]       =			name.replace('''decoder_pos_embed'''	,    '''decoder.decoder_pos_embed'''					)
    if "pos_embed" in name and "decoder" not in name:
        snake_case_				:  int       =			name.replace('''pos_embed'''	,    '''vit.embeddings.position_embeddings'''					)
    if "patch_embed.proj" in name:
        snake_case_				:  Optional[int]       =			name.replace('''patch_embed.proj'''	,    '''vit.embeddings.patch_embeddings.projection'''					)
    if "patch_embed.norm" in name:
        snake_case_				:  Any       =			name.replace('''patch_embed.norm'''	,    '''vit.embeddings.norm'''					)
    if "decoder_blocks" in name:
        snake_case_				:  Tuple       =			name.replace('''decoder_blocks'''	,    '''decoder.decoder_layers'''					)
    if "blocks" in name:
        snake_case_				:  List[Any]       =			name.replace('''blocks'''	,    '''vit.encoder.layer'''					)
    if "attn.proj" in name:
        snake_case_				:  Dict       =			name.replace('''attn.proj'''	,    '''attention.output.dense'''					)
    if "attn" in name:
        snake_case_				:  Union[str, Any]       =			name.replace('''attn'''	,    '''attention.self'''					)
    if "norm1" in name:
        snake_case_				:  Tuple       =			name.replace('''norm1'''	,    '''layernorm_before'''					)
    if "norm2" in name:
        snake_case_				:  str       =			name.replace('''norm2'''	,    '''layernorm_after'''					)
    if "mlp.fc1" in name:
        snake_case_				:  Union[str, Any]       =			name.replace('''mlp.fc1'''	,    '''intermediate.dense'''					)
    if "mlp.fc2" in name:
        snake_case_				:  Any       =			name.replace('''mlp.fc2'''	,    '''output.dense'''					)
    if "decoder_embed" in name:
        snake_case_				:  List[str]       =			name.replace('''decoder_embed'''	,    '''decoder.decoder_embed'''					)
    if "decoder_norm" in name:
        snake_case_				:  int       =			name.replace('''decoder_norm'''	,    '''decoder.decoder_norm'''					)
    if "decoder_pred" in name:
        snake_case_				:  str       =			name.replace('''decoder_pred'''	,    '''decoder.decoder_pred'''					)
    if "norm.weight" in name and "decoder" not in name:
        snake_case_				:  int       =			name.replace('''norm.weight'''	,    '''vit.layernorm.weight'''					)
    if "norm.bias" in name and "decoder" not in name:
        snake_case_				:  Any       =			name.replace('''norm.bias'''	,    '''vit.layernorm.bias'''					)

    return name




def __lowercase    (							_a	,    _a					):
    for key in orig_state_dict.copy().keys():
        snake_case_				:  Dict       =			orig_state_dict.pop(_a					)

        if "qkv" in key:
            snake_case_				:  Union[str, Any]       =			key.split('''.'''					)
            snake_case_				:  Any       =			int(key_split[1]					)
            if "decoder_blocks" in key:
                snake_case_				:  List[Any]       =			config.decoder_hidden_size
                snake_case_				:  Optional[Any]       =			'''decoder.decoder_layers.'''
                if "weight" in key:
                    snake_case_				:  Tuple       =			val[:dim, :]
                    snake_case_				:  Any       =			val[dim : dim * 2, :]
                    snake_case_				:  List[str]       =			val[-dim:, :]
                elif "bias" in key:
                    snake_case_				:  Union[str, Any]       =			val[:dim]
                    snake_case_				:  Tuple       =			val[dim : dim * 2]
                    snake_case_				:  List[Any]       =			val[-dim:]
            else:
                snake_case_				:  Dict       =			config.hidden_size
                snake_case_				:  List[Any]       =			'''vit.encoder.layer.'''
                if "weight" in key:
                    snake_case_				:  Tuple       =			val[:dim, :]
                    snake_case_				:  Dict       =			val[dim : dim * 2, :]
                    snake_case_				:  Dict       =			val[-dim:, :]
                elif "bias" in key:
                    snake_case_				:  Any       =			val[:dim]
                    snake_case_				:  Tuple       =			val[dim : dim * 2]
                    snake_case_				:  str       =			val[-dim:]

        else:
            snake_case_				:  List[str]       =			val

    return orig_state_dict




def __lowercase    (							_a	,    _a					):
    snake_case_				:  int       =			ViTMAEConfig()
    if "large" in checkpoint_url:
        snake_case_				:  int       =			1_024
        snake_case_				:  Optional[int]       =			4_096
        snake_case_				:  List[Any]       =			24
        snake_case_				:  Optional[int]       =			16
    elif "huge" in checkpoint_url:
        snake_case_				:  List[str]       =			14
        snake_case_				:  Any       =			1_280
        snake_case_				:  Optional[int]       =			5_120
        snake_case_				:  str       =			32
        snake_case_				:  Optional[Any]       =			16

    snake_case_				:  Any       =			ViTMAEForPreTraining(_a					)

    snake_case_				:  Optional[int]       =			torch.hub.load_state_dict_from_url(_a	,    map_location='''cpu'''					)['''model''']

    snake_case_				:  str       =			ViTMAEImageProcessor(size=config.image_size					)

    snake_case_				:  Union[str, Any]       =			convert_state_dict(_a	,    _a					)

    model.load_state_dict(_a					)
    model.eval()

    snake_case_				:  str       =			'''https://user-images.githubusercontent.com/11435359/147738734-196fd92f-9260-48d5-ba7e-bf103d29364d.jpg'''

    snake_case_				:  Optional[Any]       =			Image.open(requests.get(_a	,    stream=_a					).raw					)
    snake_case_				:  Tuple       =			ViTMAEImageProcessor(size=config.image_size					)
    snake_case_				:  str       =			image_processor(images=_a	,    return_tensors='''pt'''					)

    # forward pass
    torch.manual_seed(2					)
    snake_case_				:  Dict       =			model(**_a					)
    snake_case_				:  Union[str, Any]       =			outputs.logits

    if "large" in checkpoint_url:
        snake_case_				:  Dict       =			torch.tensor(
            [[-0.7309, -0.7128, -1.0169], [-1.0161, -0.9058, -1.1878], [-1.0478, -0.9411, -1.1911]]					)
    elif "huge" in checkpoint_url:
        snake_case_				:  Optional[Any]       =			torch.tensor(
            [[-1.1599, -0.9199, -1.2221], [-1.1952, -0.9269, -1.2307], [-1.2143, -0.9337, -1.2262]]					)
    else:
        snake_case_				:  Dict       =			torch.tensor(
            [[-0.9192, -0.8481, -1.1259], [-1.1349, -1.0034, -1.2599], [-1.1757, -1.0429, -1.2726]]					)

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

    print(f"Saving model 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__":
     lowercase__					:						int												=			argparse.ArgumentParser()
     # Required parameters
     parser.add_argument(
         '''--checkpoint_url''',
         default='''https://dl.fbaipublicfiles.com/mae/visualize/mae_visualize_vit_base.pth''',
         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.'''
     )

     lowercase__					:						Optional[int]												=			parser.parse_args()
     convert_vit_mae_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path)










"""simple docstring"""

import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
     import torch_xla.core.xla_model as xm
     import torch_xla.debug.metrics as met




class  _UpperCAmelCase (    lowerCAmelCase__):

 def __init__(  self	: List[Any]						,  *lowercase_	: List[Any]						,  lowercase_	: Optional[Any]=None						,  lowercase_	: Tuple=None						,  **lowercase_	: Union[str, Any]       ):
     super().__init__(*lowercase_						,  **lowercase_       )
     snake_case_				:  str       =			eval_examples
     snake_case_				:  Optional[Any]       =			post_process_function

 def 		_snake_case      (  self	: Union[str, Any]						,  lowercase_	: Optional[Dataset] = None						,  lowercase_	: str=None						,  lowercase_	: Optional[List[str]] = None						,  lowercase_	: str = "eval"						,  **lowercase_	: Dict						,  ):
     snake_case_				:  int       =			gen_kwargs.copy()
     snake_case_				:  Dict       =			(
         gen_kwargs['''max_length'''] if gen_kwargs.get('''max_length'''       ) is not None else self.args.generation_max_length
     )
     snake_case_				:  Tuple       =			(
         gen_kwargs['''num_beams'''] if gen_kwargs.get('''num_beams'''       ) is not None else self.args.generation_num_beams
     )
     snake_case_				:  Optional[Any]       =			gen_kwargs

     snake_case_				:  Optional[int]       =			self.eval_dataset if eval_dataset is None else eval_dataset
     snake_case_				:  Optional[Any]       =			self.get_eval_dataloader(lowercase_       )
     snake_case_				:  Dict       =			self.eval_examples if eval_examples is None else eval_examples

     # Temporarily disable metric computation, we will do it in the loop here.
     snake_case_				:  Optional[int]       =			self.compute_metrics
     snake_case_				:  str       =			None
     snake_case_				:  int       =			time.time()
     snake_case_				:  Any       =			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
     try:
         snake_case_				:  str       =			eval_loop(
             lowercase_						,  description='''Evaluation'''						,  prediction_loss_only=True if compute_metrics is None else None						,  ignore_keys=lowercase_						,  metric_key_prefix=lowercase_						,  )
     finally:
         snake_case_				:  Optional[int]       =			compute_metrics
     snake_case_				:  Dict       =			self.args.eval_batch_size * self.args.world_size
     if f"{metric_key_prefix}_jit_compilation_time" in output.metrics:
         start_time += output.metrics[f"{metric_key_prefix}_jit_compilation_time"]
     output.metrics.update(
         speed_metrics(
             lowercase_						,  lowercase_						,  num_samples=output.num_samples						,  num_steps=math.ceil(output.num_samples / total_batch_size       )						,  )       )

     if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
         # Only the main node write the results by default
         snake_case_				:  Tuple       =			self.post_process_function(lowercase_						,  lowercase_						,  lowercase_       )
         snake_case_				:  Optional[int]       =			self.compute_metrics(lowercase_       )

         # Prefix all keys with metric_key_prefix + '_'
         for key in list(metrics.keys()       ):
             if not key.startswith(f"{metric_key_prefix}_"       ):
                 snake_case_				:  int       =			metrics.pop(lowercase_       )

         metrics.update(output.metrics       )
     else:
         snake_case_				:  Tuple       =			output.metrics

     if self.args.should_log:
         # Only the main node log the results by default
         self.log(lowercase_       )

     if self.args.tpu_metrics_debug or self.args.debug:
         # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
         xm.master_print(met.metrics_report()       )

     snake_case_				:  str       =			self.callback_handler.on_evaluate(self.args						,  self.state						,  self.control						,  lowercase_       )
     return metrics






 def 		_snake_case      (  self	: Any						,  lowercase_	: int						,  lowercase_	: List[str]						,  lowercase_	: str=None						,  lowercase_	: str = "test"						,  **lowercase_	: Dict       ):
     snake_case_				:  Dict       =			gen_kwargs.copy()

     snake_case_				:  Dict       =			self.get_test_dataloader(lowercase_       )

     # Temporarily disable metric computation, we will do it in the loop here.
     snake_case_				:  Tuple       =			self.compute_metrics
     snake_case_				:  int       =			None
     snake_case_				:  List[Any]       =			time.time()
     snake_case_				:  int       =			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
     try:
         snake_case_				:  List[str]       =			eval_loop(
             lowercase_						,  description='''Prediction'''						,  prediction_loss_only=True if compute_metrics is None else None						,  ignore_keys=lowercase_						,  metric_key_prefix=lowercase_						,  )
     finally:
         snake_case_				:  str       =			compute_metrics

     snake_case_				:  int       =			self.args.eval_batch_size * self.args.world_size
     if f"{metric_key_prefix}_jit_compilation_time" in output.metrics:
         start_time += output.metrics[f"{metric_key_prefix}_jit_compilation_time"]
     output.metrics.update(
         speed_metrics(
             lowercase_						,  lowercase_						,  num_samples=output.num_samples						,  num_steps=math.ceil(output.num_samples / total_batch_size       )						,  )       )
     if self.post_process_function is None or self.compute_metrics is None:
         return output

     snake_case_				:  Tuple       =			self.post_process_function(lowercase_						,  lowercase_						,  lowercase_						,  '''predict'''       )
     snake_case_				:  Dict       =			self.compute_metrics(lowercase_       )

     # Prefix all keys with metric_key_prefix + '_'
     for key in list(metrics.keys()       ):
         if not key.startswith(f"{metric_key_prefix}_"       ):
             snake_case_				:  Union[str, Any]       =			metrics.pop(lowercase_       )
     metrics.update(output.metrics       )
     return PredictionOutput(predictions=predictions.predictions						,  label_ids=predictions.label_ids						,  metrics=lowercase_       )





