





import argparse
import os

import torch

from transformers import FlavaConfig, FlavaForPreTraining
from transformers.models.flava.convert_dalle_to_flava_codebook import convert_dalle_checkpoint


def       UpperCamelCase__(			UpperCamelCase__     :			List[str]    )->str:
       # encoder.embeddings are double copied in original FLAVA
       return sum(param.float().sum() if '''encoder.embeddings''' not in key else 0 for key, param in state_dict.items()    )


def       UpperCamelCase__(			UpperCamelCase__     :			str					,	UpperCamelCase__     :			Dict    )->Union[str, Any]:
       A__		   =		{}

       for key, value in state_dict.items():
              if "text_encoder.embeddings" in key or "image_encoder.embeddings" in key:
                     continue

              A__		   =		key.replace('''heads.cmd.mim_head.cls.predictions'''					,	'''mmm_image_head'''    )
              A__		   =		key.replace('''heads.cmd.mlm_head.cls.predictions'''					,	'''mmm_text_head'''    )
              A__		   =		key.replace('''heads.cmd.itm_head.cls'''					,	'''itm_head'''    )
              A__		   =		key.replace('''heads.cmd.itm_head.pooler'''					,	'''itm_head.pooler'''    )
              A__		   =		key.replace('''heads.cmd.clip_head.logit_scale'''					,	'''flava.logit_scale'''    )
              A__		   =		key.replace('''heads.fairseq_mlm.cls.predictions'''					,	'''mlm_head'''    )
              A__		   =		key.replace('''heads.imagenet.mim_head.cls.predictions'''					,	'''mim_head'''    )
              A__		   =		key.replace('''mm_text_projection'''					,	'''flava.text_to_mm_projection'''    )
              A__		   =		key.replace('''mm_image_projection'''					,	'''flava.image_to_mm_projection'''    )
              A__		   =		key.replace('''image_encoder.module'''					,	'''flava.image_model'''    )
              A__		   =		key.replace('''text_encoder.module'''					,	'''flava.text_model'''    )
              A__		   =		key.replace('''mm_encoder.module.encoder.cls_token'''					,	'''flava.multimodal_model.cls_token'''    )
              A__		   =		key.replace('''mm_encoder.module'''					,	'''flava.multimodal_model'''    )
              A__		   =		key.replace('''text_projection'''					,	'''flava.text_projection'''    )
              A__		   =		key.replace('''image_projection'''					,	'''flava.image_projection'''    )

              A__		   =		value.float()

       for key, value in codebook_state_dict.items():
              A__		   =		value

       return upgrade


@torch.no_grad()
def       UpperCamelCase__(			UpperCamelCase__     :			List[str]					,	UpperCamelCase__     :			Any					,	UpperCamelCase__     :			Any					,	UpperCamelCase__     :			str=None    )->Optional[int]:
       if config_path is not None:
              A__		   =		FlavaConfig.from_pretrained(UpperCamelCase__    )
       else:
              A__		   =		FlavaConfig()

       A__		   =		FlavaForPreTraining(UpperCamelCase__    ).eval()

       A__		   =		convert_dalle_checkpoint(UpperCamelCase__					,	UpperCamelCase__					,	save_checkpoint=UpperCamelCase__    )

       if os.path.exists(UpperCamelCase__    ):
              A__		   =		torch.load(UpperCamelCase__					,	map_location='''cpu'''    )
       else:
              A__		   =		torch.hub.load_state_dict_from_url(UpperCamelCase__					,	map_location='''cpu'''    )

       A__		   =		upgrade_state_dict(UpperCamelCase__					,	UpperCamelCase__    )
       hf_model.load_state_dict(UpperCamelCase__    )
       A__		   =		hf_model.state_dict()
       A__		   =		count_parameters(UpperCamelCase__    )
       A__		   =		count_parameters(UpperCamelCase__    ) + count_parameters(UpperCamelCase__    )

       assert torch.allclose(UpperCamelCase__					,	UpperCamelCase__					,	atol=1e-3    )

       hf_model.save_pretrained(UpperCamelCase__    )


if __name__ == "__main__":
       a__: 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 flava checkpoint')
       parser.add_argument('--codebook_path', default=None, type=str, help='Path to flava codebook checkpoint')
       parser.add_argument('--config_path', default=None, type=str, help='Path to hf config.json of model to convert')
       a__: Dict							   =						parser.parse_args()

       convert_flava_checkpoint(args.checkpoint_path, args.codebook_path, args.pytorch_dump_folder_path, args.config_path)












import unittest

from huggingface_hub import hf_hub_download

from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VideoMAEFeatureExtractor
from transformers.pipelines import VideoClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_decord,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
)

from .test_pipelines_common import ANY




@is_pipeline_test
@require_torch_or_tf
@require_vision
@require_decord
class        SCREAMING_SNAKE_CASE__					(		unittest.TestCase ):
 __SCREAMING_SNAKE_CASE							=      MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING



 def 							UpperCamelCase     (				self,__lowerCamelCase,__lowerCamelCase,__lowerCamelCase					):
        A__		   =		hf_hub_download(
            repo_id='''nateraw/video-demo''',filename='''archery.mp4''',repo_type='''dataset'''					)
        A__		   =		VideoClassificationPipeline(model=__lowerCamelCase,image_processor=__lowerCamelCase,top_k=2					)
        A__		   =		[
            example_video_filepath,
            '''https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4''',
        ]
        return video_classifier, examples



 def 							UpperCamelCase     (				self,__lowerCamelCase,__lowerCamelCase					):
        for example in examples:
               A__		   =		video_classifier(__lowerCamelCase					)

               self.assertEqual(
                   __lowerCamelCase,[
                       {'''score''': ANY(__lowerCamelCase					), '''label''': ANY(__lowerCamelCase					)},
                       {'''score''': ANY(__lowerCamelCase					), '''label''': ANY(__lowerCamelCase					)},
                   ],)



 @require_torch
 def 							UpperCamelCase     (				self					):
        A__		   =		'''hf-internal-testing/tiny-random-VideoMAEForVideoClassification'''
        A__		   =		VideoMAEFeatureExtractor(
            size={'''shortest_edge''': 10},crop_size={'''height''': 10, '''width''': 10}					)
        A__		   =		pipeline(
            '''video-classification''',model=__lowerCamelCase,feature_extractor=__lowerCamelCase,frame_sampling_rate=4					)

        A__		   =		hf_hub_download(repo_id='''nateraw/video-demo''',filename='''archery.mp4''',repo_type='''dataset'''					)
        A__		   =		video_classifier(__lowerCamelCase,top_k=2					)
        self.assertEqual(
            nested_simplify(__lowerCamelCase,decimals=4					),[{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],)

        A__		   =		video_classifier(
            [
                video_file_path,
                video_file_path,
            ],top_k=2,)
        self.assertEqual(
            nested_simplify(__lowerCamelCase,decimals=4					),[
                [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],
                [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],
            ],)




 @require_tf
 def 							UpperCamelCase     (				self					):
        pass





