


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    snake_case_     (  snake_case     )  ->				Optional[Any]:
       # 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    snake_case_     (  snake_case			,							snake_case     )  ->				Any:
       lowercase__:   List[str]  			=   {}

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

              lowercase__:   Dict  			=   key.replace('heads.cmd.mim_head.cls.predictions'			,							'mmm_image_head'     )
              lowercase__:   List[Any]  			=   key.replace('heads.cmd.mlm_head.cls.predictions'			,							'mmm_text_head'     )
              lowercase__:   Any  			=   key.replace('heads.cmd.itm_head.cls'			,							'itm_head'     )
              lowercase__:   str  			=   key.replace('heads.cmd.itm_head.pooler'			,							'itm_head.pooler'     )
              lowercase__:   Dict  			=   key.replace('heads.cmd.clip_head.logit_scale'			,							'flava.logit_scale'     )
              lowercase__:   Dict  			=   key.replace('heads.fairseq_mlm.cls.predictions'			,							'mlm_head'     )
              lowercase__:   str  			=   key.replace('heads.imagenet.mim_head.cls.predictions'			,							'mim_head'     )
              lowercase__:   Union[str, Any]  			=   key.replace('mm_text_projection'			,							'flava.text_to_mm_projection'     )
              lowercase__:   List[str]  			=   key.replace('mm_image_projection'			,							'flava.image_to_mm_projection'     )
              lowercase__:   Union[str, Any]  			=   key.replace('image_encoder.module'			,							'flava.image_model'     )
              lowercase__:   int  			=   key.replace('text_encoder.module'			,							'flava.text_model'     )
              lowercase__:   str  			=   key.replace('mm_encoder.module.encoder.cls_token'			,							'flava.multimodal_model.cls_token'     )
              lowercase__:   str  			=   key.replace('mm_encoder.module'			,							'flava.multimodal_model'     )
              lowercase__:   List[Any]  			=   key.replace('text_projection'			,							'flava.text_projection'     )
              lowercase__:   Tuple  			=   key.replace('image_projection'			,							'flava.image_projection'     )

              lowercase__:   Optional[Any]  			=   value.float()

       for key, value in codebook_state_dict.items():
              lowercase__:   Optional[Any]  			=   value

       return upgrade




@torch.no_grad()
def    snake_case_     (  snake_case			,							snake_case			,							snake_case			,							snake_case=None     )  ->				str:
       if config_path is not None:
              lowercase__:   Union[str, Any]  			=   FlavaConfig.from_pretrained(snake_case     )
       else:
              lowercase__:   int  			=   FlavaConfig()

       lowercase__:   Optional[int]  			=   FlavaForPreTraining(snake_case     ).eval()

       lowercase__:   Dict  			=   convert_dalle_checkpoint(snake_case			,							snake_case			,							save_checkpoint=snake_case     )

       if os.path.exists(snake_case     ):
              lowercase__:   Optional[int]  			=   torch.load(snake_case			,							map_location='cpu'     )
       else:
              lowercase__:   Any  			=   torch.hub.load_state_dict_from_url(snake_case			,							map_location='cpu'     )

       lowercase__:   Any  			=   upgrade_state_dict(snake_case			,							snake_case     )
       hf_model.load_state_dict(snake_case     )
       lowercase__:   int  			=   hf_model.state_dict()
       lowercase__:   List[str]  			=   count_parameters(snake_case     )
       lowercase__:   Any  			=   count_parameters(snake_case     ) + count_parameters(snake_case     )

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

       hf_model.save_pretrained(snake_case     )


if __name__ == "__main__":
 __lowerCAmelCase		       =							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''')
 __lowerCAmelCase		       =							parser.parse_args()

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






from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
 from PIL import Image

 from ..image_utils import load_image

if is_torch_available():
 from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
 from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
 from ..tf_utils import stable_softmax

__lowerCAmelCase		       =							logging.get_logger(__name__)





@add_end_docstrings(__UpperCamelCase      )
class   __a			(    __UpperCamelCase      ):
 def __init__(   self				, **lowerCAmelCase__ )					->				List[str]:




        '''simple docstring'''
        super().__init__(**lowerCAmelCase__ )

        requires_backends(self				, 'vision' )
        self.check_model_type(
            TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
            if self.framework == 'tf'
            else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING )
 def __call__(   self				, lowerCAmelCase__				, **lowerCAmelCase__ )					->				Optional[Any]:




        '''simple docstring'''
        return super().__call__(lowerCAmelCase__				, **lowerCAmelCase__ )
 def 		SCREAMING_SNAKE_CASE__   (   self				, **lowerCAmelCase__ )					->				Dict:




        '''simple docstring'''
        lowercase__:   int  			=   {}
        if "candidate_labels" in kwargs:
               lowercase__:   Dict  			=   kwargs['candidate_labels']
        if "hypothesis_template" in kwargs:
               lowercase__:   List[Any]  			=   kwargs['hypothesis_template']

        return preprocess_params, {}, {}
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__				, lowerCAmelCase__=None				, lowerCAmelCase__="This is a photo of {}." )					->				int:




        '''simple docstring'''
        lowercase__:   Optional[int]  			=   load_image(lowerCAmelCase__ )
        lowercase__:   Dict  			=   self.image_processor(images=[image]				, return_tensors=self.framework )
        lowercase__:   Tuple  			=   candidate_labels
        lowercase__:   List[str]  			=   [hypothesis_template.format(lowerCAmelCase__ ) for x in candidate_labels]
        lowercase__:   Optional[Any]  			=   self.tokenizer(lowerCAmelCase__				, return_tensors=self.framework				, padding=lowerCAmelCase__ )
        lowercase__:   str  			=   [text_inputs]
        return inputs
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__ )					->				int:




        '''simple docstring'''
        lowercase__:   str  			=   model_inputs.pop('candidate_labels' )
        lowercase__:   List[str]  			=   model_inputs.pop('text_inputs' )
        if isinstance(text_inputs[0]				, lowerCAmelCase__ ):
               lowercase__:   Any  			=   text_inputs[0]
        else:
               # Batching case.
               lowercase__:   Optional[int]  			=   text_inputs[0][0]

        lowercase__:   Tuple  			=   self.model(**lowerCAmelCase__				, **lowerCAmelCase__ )

        lowercase__:   Any  			=   {
            'candidate_labels': candidate_labels,
            'logits': outputs.logits_per_image,
        }
        return model_outputs
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__ )					->				Union[str, Any]:




        '''simple docstring'''
        lowercase__:   Tuple  			=   model_outputs.pop('candidate_labels' )
        lowercase__:   Dict  			=   model_outputs['logits'][0]
        if self.framework == "pt":
               lowercase__:   Optional[int]  			=   logits.softmax(dim=-1 ).squeeze(-1 )
               lowercase__:   Dict  			=   probs.tolist()
               if not isinstance(lowerCAmelCase__				, lowerCAmelCase__ ):
                      lowercase__:   Dict  			=   [scores]
        elif self.framework == "tf":
               lowercase__:   Optional[int]  			=   stable_softmax(lowerCAmelCase__				, axis=-1 )
               lowercase__:   Union[str, Any]  			=   probs.numpy().tolist()
        else:
               raise ValueError(F'Unsupported framework: {self.framework}' )

        lowercase__:   List[Any]  			=   [
            {'score': score, 'label': candidate_label}
            for score, candidate_label in sorted(zip(lowerCAmelCase__				, lowerCAmelCase__ )				, key=lambda lowerCAmelCase__ : -x[0] )
        ]
        return result


