
'''simple docstring'''




import argparse
import json

import gdown
import numpy as np
import torch
from huggingface_hub import hf_hub_download

from transformers import (
    VideoMAEConfig,
    VideoMAEForPreTraining,
    VideoMAEForVideoClassification,
    VideoMAEImageProcessor,
)







def 		lowerCAmelCase_				(   snake_case__						):




    '''simple docstring'''



    A       :   Dict					=						VideoMAEConfig()

    set_architecture_configs(snake_case__		,				snake_case__						)

    if "finetuned" not in model_name:
        A       :   int					=						False

    if "finetuned" in model_name:
        A       :   List[Any]					=						'''huggingface/label-files'''
        if "kinetics" in model_name:
            A       :   int					=						400
            A       :   str					=						'''kinetics400-id2label.json'''
        elif "ssv2" in model_name:
            A       :   Dict					=						174
            A       :   Dict					=						'''something-something-v2-id2label.json'''
        else:
            raise ValueError('''Model name should either contain \'kinetics\' or \'ssv2\' in case it\'s fine-tuned.'''						)
        A       :   Optional[int]					=						json.load(open(hf_hub_download(snake_case__		,				snake_case__		,				repo_type='''dataset'''						)		,				'''r'''						)						)
        A       :   int					=						{int(snake_case__						): v for k, v in idalabel.items()}
        A       :   Union[str, Any]					=						idalabel
        A       :   Tuple					=						{v: k for k, v in idalabel.items()}

    return config







def 		lowerCAmelCase_				(   snake_case__		,				snake_case__						):




    '''simple docstring'''



    if "small" in model_name:
        A       :   int					=						384
        A       :   Tuple					=						1536
        A       :   Optional[Any]					=						12
        A       :   List[str]					=						16
        A       :   Optional[Any]					=						12
        A       :   Optional[Any]					=						3
        A       :   List[Any]					=						192
        A       :   List[str]					=						768
    elif "large" in model_name:
        A       :   Optional[int]					=						1024
        A       :   Optional[Any]					=						4096
        A       :   Tuple					=						24
        A       :   List[Any]					=						16
        A       :   Tuple					=						12
        A       :   Union[str, Any]					=						8
        A       :   Optional[int]					=						512
        A       :   str					=						2048
    elif "huge" in model_name:
        A       :   List[Any]					=						1280
        A       :   Tuple					=						5120
        A       :   Tuple					=						32
        A       :   Optional[Any]					=						16
        A       :   Dict					=						12
        A       :   List[str]					=						8
        A       :   str					=						640
        A       :   Optional[int]					=						2560
    elif "base" not in model_name:
        raise ValueError('''Model name should include either "small", "base", "large", or "huge"'''						)







def 		lowerCAmelCase_				(   snake_case__						):




    '''simple docstring'''



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

    return name







def 		lowerCAmelCase_				(   snake_case__		,				snake_case__						):




    '''simple docstring'''



    for key in orig_state_dict.copy().keys():
        A       :   List[Any]					=						orig_state_dict.pop(snake_case__						)

        if key.startswith('''encoder.'''						):
            A       :   str					=						key.replace('''encoder.'''		,				''''''						)

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

    return orig_state_dict







def 		lowerCAmelCase_				(   ):




    '''simple docstring'''



    A       :   Dict					=						hf_hub_download(
        repo_id='''hf-internal-testing/spaghetti-video'''		,				filename='''eating_spaghetti.npy'''		,				repo_type='''dataset'''						)
    A       :   Any					=						np.load(snake_case__						)
    return list(snake_case__						)







def 		lowerCAmelCase_				(   snake_case__		,				snake_case__		,				snake_case__		,				snake_case__						):




    '''simple docstring'''



    A       :   List[Any]					=						get_videomae_config(snake_case__						)

    if "finetuned" in model_name:
        A       :   int					=						VideoMAEForVideoClassification(snake_case__						)
    else:
        A       :   List[str]					=						VideoMAEForPreTraining(snake_case__						)

    # download original checkpoint, hosted on Google Drive
    A       :   str					=						'''pytorch_model.bin'''
    gdown.cached_download(snake_case__		,				snake_case__		,				quiet=snake_case__						)
    A       :   Tuple					=						torch.load(snake_case__		,				map_location='''cpu'''						)
    if "model" in files:
        A       :   Optional[Any]					=						files['''model''']
    else:
        A       :   Tuple					=						files['''module''']
    A       :   Dict					=						convert_state_dict(snake_case__		,				snake_case__						)

    model.load_state_dict(snake_case__						)
    model.eval()

    # verify model on basic input
    A       :   Union[str, Any]					=						VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]		,				image_std=[0.5, 0.5, 0.5]						)
    A       :   Any					=						prepare_video()
    A       :   Union[str, Any]					=						image_processor(snake_case__		,				return_tensors='''pt'''						)

    if "finetuned" not in model_name:
        A       :   str					=						hf_hub_download(repo_id='''hf-internal-testing/bool-masked-pos'''		,				filename='''bool_masked_pos.pt'''						)
        A       :   Optional[Any]					=						torch.load(snake_case__						)

    A       :   List[str]					=						model(**snake_case__						)
    A       :   Optional[int]					=						outputs.logits

    A       :   List[str]					=						[
        '''videomae-small-finetuned-kinetics''',
        '''videomae-small-finetuned-ssv2''',
        # Kinetics-400 checkpoints (short = pretrained only for 800 epochs instead of 1600)
        '''videomae-base-short''',
        '''videomae-base-short-finetuned-kinetics''',
        '''videomae-base''',
        '''videomae-base-finetuned-kinetics''',
        '''videomae-large''',
        '''videomae-large-finetuned-kinetics''',
        '''videomae-huge-finetuned-kinetics''',
        # Something-Something-v2 checkpoints (short = pretrained only for 800 epochs instead of 2400)
        '''videomae-base-short-ssv2''',
        '''videomae-base-short-finetuned-ssv2''',
        '''videomae-base-ssv2''',
        '''videomae-base-finetuned-ssv2''',
    ]

    # NOTE: logits were tested with image_mean and image_std equal to [0.5, 0.5, 0.5] and [0.5, 0.5, 0.5]
    if model_name == "videomae-small-finetuned-kinetics":
        A       :   int					=						torch.Size([1, 400]						)
        A       :   List[Any]					=						torch.tensor([-0.92_91, -0.40_61, -0.93_07]						)
    elif model_name == "videomae-small-finetuned-ssv2":
        A       :   Any					=						torch.Size([1, 174]						)
        A       :   Union[str, Any]					=						torch.tensor([0.26_71, -0.46_89, -0.82_35]						)
    elif model_name == "videomae-base":
        A       :   List[str]					=						torch.Size([1, 1408, 1536]						)
        A       :   Union[str, Any]					=						torch.tensor([[0.77_39, 0.79_68, 0.70_89], [0.67_01, 0.74_87, 0.62_09], [0.42_87, 0.51_58, 0.47_73]]						)
    elif model_name == "videomae-base-short":
        A       :   List[str]					=						torch.Size([1, 1408, 1536]						)
        A       :   Optional[Any]					=						torch.tensor([[0.79_94, 0.96_12, 0.85_08], [0.74_01, 0.89_58, 0.83_02], [0.58_62, 0.74_68, 0.73_25]]						)
        # we verified the loss both for normalized and unnormalized targets for this one
        A       :   List[str]					=						torch.tensor([0.51_42]						) if config.norm_pix_loss else torch.tensor([0.64_69]						)
    elif model_name == "videomae-large":
        A       :   List[str]					=						torch.Size([1, 1408, 1536]						)
        A       :   Any					=						torch.tensor([[0.71_49, 0.79_97, 0.69_66], [0.67_68, 0.78_69, 0.69_48], [0.51_39, 0.62_21, 0.56_05]]						)
    elif model_name == "videomae-large-finetuned-kinetics":
        A       :   Any					=						torch.Size([1, 400]						)
        A       :   str					=						torch.tensor([0.07_71, 0.00_11, -0.36_25]						)
    elif model_name == "videomae-huge-finetuned-kinetics":
        A       :   Optional[int]					=						torch.Size([1, 400]						)
        A       :   str					=						torch.tensor([0.24_33, 0.16_32, -0.48_94]						)
    elif model_name == "videomae-base-short-finetuned-kinetics":
        A       :   List[Any]					=						torch.Size([1, 400]						)
        A       :   List[Any]					=						torch.tensor([0.65_88, 0.09_90, -0.24_93]						)
    elif model_name == "videomae-base-finetuned-kinetics":
        A       :   Union[str, Any]					=						torch.Size([1, 400]						)
        A       :   List[Any]					=						torch.tensor([0.36_69, -0.06_88, -0.24_21]						)
    elif model_name == "videomae-base-short-ssv2":
        A       :   Any					=						torch.Size([1, 1408, 1536]						)
        A       :   List[str]					=						torch.tensor([[0.47_12, 0.52_96, 0.57_86], [0.22_78, 0.27_29, 0.40_26], [0.03_52, 0.07_30, 0.25_06]]						)
    elif model_name == "videomae-base-short-finetuned-ssv2":
        A       :   Optional[Any]					=						torch.Size([1, 174]						)
        A       :   Optional[Any]					=						torch.tensor([-0.05_37, -0.15_39, -0.32_66]						)
    elif model_name == "videomae-base-ssv2":
        A       :   str					=						torch.Size([1, 1408, 1536]						)
        A       :   Union[str, Any]					=						torch.tensor([[0.81_31, 0.87_27, 0.85_46], [0.73_66, 0.93_77, 0.88_70], [0.59_35, 0.88_74, 0.85_64]]						)
    elif model_name == "videomae-base-finetuned-ssv2":
        A       :   Optional[int]					=						torch.Size([1, 174]						)
        A       :   str					=						torch.tensor([0.19_61, -0.83_37, -0.63_89]						)
    else:
        raise ValueError(F'Model name not supported. Should be one of {model_names}'						)

    # verify logits
    assert logits.shape == expected_shape
    if "finetuned" in model_name:
        assert torch.allclose(logits[0, :3]		,				snake_case__		,				atol=1E-4						)
    else:
        print('''Logits:'''		,				logits[0, :3, :3]						)
        assert torch.allclose(logits[0, :3, :3]		,				snake_case__		,				atol=1E-4						)
    print('''Logits ok!'''						)

    # verify loss, if applicable
    if model_name == "videomae-base-short":
        A       :   List[str]					=						outputs.loss
        assert torch.allclose(snake_case__		,				snake_case__		,				atol=1E-4						)
        print('''Loss ok!'''						)

    if pytorch_dump_folder_path is not None:
        print(F'Saving model and image processor to {pytorch_dump_folder_path}'						)
        image_processor.save_pretrained(snake_case__						)
        model.save_pretrained(snake_case__						)

    if push_to_hub:
        print('''Pushing to the hub...'''						)
        model.push_to_hub(snake_case__		,				organization='''nielsr'''						)


if __name__ == "__main__":
      lowercase			:				Any    		=			argparse.ArgumentParser()
      # Required parameters
      parser.add_argument(
          '--checkpoint_url',
          default='https://drive.google.com/u/1/uc?id=1tEhLyskjb755TJ65ptsrafUG2llSwQE1&amp;export=download&amp;confirm=t&amp;uuid=aa3276eb-fb7e-482a-adec-dc7171df14c4',
          type=str,
          help=(
              'URL of the original PyTorch checkpoint (on Google Drive) you\'d like to convert. Should be a direct'
              ' download link.'
          ),
      )
      parser.add_argument(
          '--pytorch_dump_folder_path',
          default='/Users/nielsrogge/Documents/VideoMAE/Test',
          type=str,
          help='Path to the output PyTorch model directory.',
      )
      parser.add_argument('--model_name', default='videomae-base', type=str, help='Name of the model.')
      parser.add_argument(
          '--push_to_hub', action='store_true', help='Whether or not to push the converted model to the 🤗 hub.'
      )

      lowercase			:				List[Any]    		=			parser.parse_args()
      convert_videomae_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path, args.model_name, args.push_to_hub)






'''simple docstring'''




import argparse
import re

import torch
from CLAP import create_model

from transformers import AutoFeatureExtractor, ClapConfig, ClapModel


lowercase			:				Union[str, Any]    		=			{
    'text_branch': 'text_model',
    'audio_branch': 'audio_model.audio_encoder',
    'attn': 'attention.self',
    'self.proj': 'output.dense',
    'attention.self_mask': 'attn_mask',
    'mlp.fc1': 'intermediate.dense',
    'mlp.fc2': 'output.dense',
    'norm1': 'layernorm_before',
    'norm2': 'layernorm_after',
    'bn0': 'batch_norm',
}

lowercase			:				Tuple    		=			AutoFeatureExtractor.from_pretrained('laion/clap-htsat-unfused', truncation='rand_trunc')







def 		lowerCAmelCase_				(   snake_case__		,				snake_case__=False						):




    '''simple docstring'''



    A,  A       :   Tuple					=						create_model(
        '''HTSAT-tiny'''		,				'''roberta'''		,				snake_case__		,				precision='''fp32'''		,				device='''cuda:0''' if torch.cuda.is_available() else '''cpu'''		,				enable_fusion=snake_case__		,				fusion_type='''aff_2d''' if enable_fusion else None		,				)
    return model, model_cfg







def 		lowerCAmelCase_				(   snake_case__						):




    '''simple docstring'''



    A       :   Dict					=						{}

    A       :   str					=						R'''.*sequential.(\d+).*'''
    A       :   Union[str, Any]					=						R'''.*_projection.(\d+).*'''

    for key, value in state_dict.items():
        # check if any key needs to be modified
        for key_to_modify, new_key in KEYS_TO_MODIFY_MAPPING.items():
            if key_to_modify in key:
                A       :   Any					=						key.replace(snake_case__		,				snake_case__						)

        if re.match(snake_case__		,				snake_case__						):
            # replace sequential layers with list
            A       :   Any					=						re.match(snake_case__		,				snake_case__						).group(1						)

            A       :   List[str]					=						key.replace(F'sequential.{sequential_layer}.'		,				F'layers.{int(snake_case__						)//3}.linear.'						)
        elif re.match(snake_case__		,				snake_case__						):
            A       :   Union[str, Any]					=						int(re.match(snake_case__		,				snake_case__						).group(1						)						)

            # Because in CLAP they use `nn.Sequential`...
            A       :   str					=						1 if projecton_layer == 0 else 2

            A       :   Optional[Any]					=						key.replace(F'_projection.{projecton_layer}.'		,				F'_projection.linear{transformers_projection_layer}.'						)

        if "audio" and "qkv" in key:
            # split qkv into query key and value
            A       :   int					=						value
            A       :   List[Any]					=						mixed_qkv.size(0						) // 3

            A       :   Union[str, Any]					=						mixed_qkv[:qkv_dim]
            A       :   Optional[int]					=						mixed_qkv[qkv_dim : qkv_dim * 2]
            A       :   Optional[int]					=						mixed_qkv[qkv_dim * 2 :]

            A       :   Tuple					=						query_layer
            A       :   Union[str, Any]					=						key_layer
            A       :   Optional[int]					=						value_layer
        else:
            A       :   Dict					=						value

    return model_state_dict







def 		lowerCAmelCase_				(   snake_case__		,				snake_case__		,				snake_case__		,				snake_case__=False						):




    '''simple docstring'''



    A,  A       :   int					=						init_clap(snake_case__		,				enable_fusion=snake_case__						)

    clap_model.eval()
    A       :   str					=						clap_model.state_dict()
    A       :   Union[str, Any]					=						rename_state_dict(snake_case__						)

    A       :   Tuple					=						ClapConfig()
    A       :   str					=						enable_fusion
    A       :   str					=						ClapModel(snake_case__						)

    # ignore the spectrogram embedding layer
    model.load_state_dict(snake_case__		,				strict=snake_case__						)

    model.save_pretrained(snake_case__						)
    transformers_config.save_pretrained(snake_case__						)


if __name__ == "__main__":
      lowercase			:				List[str]    		=			argparse.ArgumentParser()
      parser.add_argument('--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model.')
      parser.add_argument('--checkpoint_path', default=None, type=str, help='Path to fairseq checkpoint')
      parser.add_argument('--config_path', default=None, type=str, help='Path to hf config.json of model to convert')
      parser.add_argument('--enable_fusion', action='store_true', help='Whether to enable fusion or not')
      lowercase			:				Tuple    		=			parser.parse_args()

      convert_clap_checkpoint(args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.enable_fusion)




