





import timeit
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_auto import MODEL_MAPPING, MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_torch_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_torch_available():
  import torch

  from .benchmark_args import PyTorchBenchmarkArguments


if is_pyanvml_available():
  import pyanvml.pyanvml as nvml


_A							= logging.get_logger(__name__)



class 				__snake_case  (     __SCREAMING_SNAKE_CASE						):


    '''simple docstring'''







    lowerCamelCase__						:		PyTorchBenchmarkArguments
    lowerCamelCase__						:		PretrainedConfig
    lowerCamelCase__						:		str							=  "PyTorch"



    @property
    def    lowercase_				(					self    ):





        '''simple docstring'''

        return torch.__version__



    def    lowercase_				(					self      ,							A_      ,							A_      ,							A_    ):





        '''simple docstring'''

        SCREAMING_SNAKE_CASE__       							= self._prepare_inference_func(A_      ,							A_      ,							A_    )
        return self._measure_speed(_inference    )



    def    lowercase_				(					self      ,							A_      ,							A_      ,							A_    ):





        '''simple docstring'''

        SCREAMING_SNAKE_CASE__       							= self._prepare_inference_func(A_      ,							A_      ,							A_    )
        return self._measure_memory(_inference    )



    def    lowercase_				(					self      ,							A_      ,							A_      ,							A_    ):





        '''simple docstring'''

        SCREAMING_SNAKE_CASE__       							= self._prepare_train_func(A_      ,							A_      ,							A_    )
        return self._measure_speed(_train    )



    def    lowercase_				(					self      ,							A_      ,							A_      ,							A_    ):





        '''simple docstring'''

        SCREAMING_SNAKE_CASE__       							= self._prepare_train_func(A_      ,							A_      ,							A_    )
        return self._measure_memory(_train    )



    def    lowercase_				(					self      ,							A_      ,							A_      ,							A_    ):





        '''simple docstring'''

        SCREAMING_SNAKE_CASE__       							= self.config_dict[model_name]

        if self.args.torchscript:
            SCREAMING_SNAKE_CASE__       							= True

        SCREAMING_SNAKE_CASE__       							= (
            hasattr(A_      ,							'''architectures'''    )
            and isinstance(config.architectures      ,							A_    )
            and len(config.architectures    ) > 0
        )
        if not self.args.only_pretrain_model and has_model_class_in_config:
            try:
                SCREAMING_SNAKE_CASE__       							= config.architectures[0]
                SCREAMING_SNAKE_CASE__       							= __import__('''transformers'''      ,							fromlist=[model_class]    )
                SCREAMING_SNAKE_CASE__       							= getattr(A_      ,							A_    )
                SCREAMING_SNAKE_CASE__       							= model_cls(A_    )
            except ImportError:
                raise ImportError(
                    f'''{model_class} does not exist. If you just want to test the pretrained model, you might want to'''
                    ''' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'''    )
        else:
            SCREAMING_SNAKE_CASE__       							= MODEL_MAPPING[config.__class__](A_    )

        model.eval()
        model.to(self.args.device    )

        # encoder-decoder has vocab size saved differently
        SCREAMING_SNAKE_CASE__       							= config.vocab_size if hasattr(A_      ,							'''vocab_size'''    ) else config.encoder.vocab_size
        SCREAMING_SNAKE_CASE__       							= torch.randint(A_      ,							(batch_size, sequence_length)      ,							dtype=torch.long      ,							device=self.args.device    )

        if self.args.fpaa:
            logger.info('''Running training in Mixed Precision...'''    )
            if not self.args.is_gpu:
                raise ValueError('''Mixed precision is possible only for GPU.'''    )
            # amp seems to have memory leaks so that memory usage
            # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
            model.half()

        if self.args.torchscript:
            with torch.no_grad():
                SCREAMING_SNAKE_CASE__       							= torch.jit.trace(A_      ,							A_    )
        else:
            SCREAMING_SNAKE_CASE__       							= model

        def encoder_decoder_forward():
            with torch.no_grad():
                SCREAMING_SNAKE_CASE__       							= inference_model(A_      ,							decoder_input_ids=A_    )
            return outputs

        def encoder_forward():
            with torch.no_grad():
                SCREAMING_SNAKE_CASE__       							= inference_model(A_    )
            return outputs

        SCREAMING_SNAKE_CASE__       							= encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
        return _forward



    def    lowercase_				(					self      ,							A_      ,							A_      ,							A_    ):





        '''simple docstring'''

        SCREAMING_SNAKE_CASE__       							= self.config_dict[model_name]

        SCREAMING_SNAKE_CASE__       							= (
            hasattr(A_      ,							'''architectures'''    )
            and isinstance(config.architectures      ,							A_    )
            and len(config.architectures    ) > 0
        )
        if not self.args.only_pretrain_model and has_model_class_in_config:
            try:
                SCREAMING_SNAKE_CASE__       							= config.architectures[0]
                SCREAMING_SNAKE_CASE__       							= __import__('''transformers'''      ,							fromlist=[model_class]    )
                SCREAMING_SNAKE_CASE__       							= getattr(A_      ,							A_    )
                SCREAMING_SNAKE_CASE__       							= model_cls(A_    )
            except ImportError:
                raise ImportError(
                    f'''{model_class} does not exist. If you just want to test the pretrained model, you might want to'''
                    ''' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'''    )
        else:
            SCREAMING_SNAKE_CASE__       							= MODEL_WITH_LM_HEAD_MAPPING[config.__class__](A_    )

        if self.args.torchscript:
            raise NotImplementedError('''Training for torchscript is currently not implemented'''    )
        else:
            SCREAMING_SNAKE_CASE__       							= model

        model.train()
        model.to(self.args.device    )

        # encoder-decoder has vocab size saved differently
        SCREAMING_SNAKE_CASE__       							= config.vocab_size if hasattr(A_      ,							'''vocab_size'''    ) else config.encoder.vocab_size
        SCREAMING_SNAKE_CASE__       							= torch.randint(A_      ,							(batch_size, sequence_length)      ,							dtype=torch.long      ,							device=self.args.device    )

        if self.args.fpaa:
            logger.info('''Running training in Mixed Precision...'''    )
            if not self.args.is_gpu:
                raise ValueError('''Mixed precision is possible only for GPU.'''    )

            # amp seems to have memory leaks so that memory usage
            # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
            model.half()

        def compute_loss_and_backprob_encoder():
            SCREAMING_SNAKE_CASE__       							= train_model(A_      ,							labels=A_    )[0]
            loss.backward()
            return loss

        def compute_loss_and_backprob_encoder_decoder():
            SCREAMING_SNAKE_CASE__       							= train_model(A_      ,							decoder_input_ids=A_      ,							labels=A_    )[0]
            loss.backward()
            return loss

        SCREAMING_SNAKE_CASE__       							= (
            compute_loss_and_backprob_encoder_decoder
            if config.is_encoder_decoder
            else compute_loss_and_backprob_encoder
        )
        return _train



    def    lowercase_				(					self      ,							A_    ):





        '''simple docstring'''

        try:
            if self.args.is_tpu or self.args.torchscript:
                # run additional 10 times to stabilize compilation for tpu and torchscript
                logger.info('''Do inference on TPU or torchscript. Running model 5 times to stabilize compilation'''    )
                timeit.repeat(
                    A_      ,							repeat=1      ,							number=5      ,							)

            # as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
            SCREAMING_SNAKE_CASE__       							= timeit.repeat(
                A_      ,							repeat=self.args.repeat      ,							number=10      ,							)

            if self.args.is_tpu and self.args.torch_xla_tpu_print_metrics:
                import torch_xla.debug.metrics as met

                self.print_fn(met.metrics_report()    )

            return min(A_    ) / 10.0
        except RuntimeError as e:
            self.print_fn(f'''Doesn\'t fit on GPU. {e}'''    )
            return "N/A"




    def    lowercase_				(					self      ,							A_    ):





        '''simple docstring'''

        try:
            if self.args.trace_memory_line_by_line:
                SCREAMING_SNAKE_CASE__       							= start_memory_tracing('''transformers'''    )

            if self.args.is_tpu:
                # tpu
                raise NotImplementedError(
                    '''Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking with'''
                    ''' `--no-memory` or `args.memory=False`'''    )
            elif self.args.is_gpu:
                if not is_pyanvml_available():
                    logger.warning(
                        '''py3nvml not installed, we won\'t log GPU memory usage. '''
                        '''Install py3nvml (pip install py3nvml) to log information about GPU.'''    )
                    SCREAMING_SNAKE_CASE__       							= '''N/A'''
                else:
                    logger.info(
                        '''Measuring total GPU usage on GPU device. Make sure to not have additional processes running'''
                        ''' on the same GPU.'''    )
                    # init nvml
                    nvml.nvmlInit()
                    func()
                    SCREAMING_SNAKE_CASE__       							= nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx    )
                    SCREAMING_SNAKE_CASE__       							= nvml.nvmlDeviceGetMemoryInfo(A_    )
                    SCREAMING_SNAKE_CASE__       							= meminfo.used
                    SCREAMING_SNAKE_CASE__       							= Memory(A_    )
                    # shutdown nvml
                    nvml.nvmlShutdown()
            else:
                # cpu
                SCREAMING_SNAKE_CASE__       							= measure_peak_memory_cpu(A_    )
                SCREAMING_SNAKE_CASE__       							= Memory(A_    ) if isinstance(A_      ,							A_    ) else memory_bytes

            if self.args.trace_memory_line_by_line:
                SCREAMING_SNAKE_CASE__       							= stop_memory_tracing(A_    )
            else:
                SCREAMING_SNAKE_CASE__       							= None

            return memory, summary
        except RuntimeError as e:
            self.print_fn(f'''Doesn\'t fit on GPU. {e}'''    )
            return "N/A", None







import argparse

import torch

from transformers import ChineseCLIPConfig, ChineseCLIPModel


def        __snake_case       (						lowerCAmelCase_   ,       lowerCAmelCase_   ,       lowerCAmelCase_  )							->	str:
    SCREAMING_SNAKE_CASE__							,		SCREAMING_SNAKE_CASE__							,		SCREAMING_SNAKE_CASE__       							= pt_weights[f'''{prefix}.in_proj_weight'''].chunk(3   ,       dim=0  )
    SCREAMING_SNAKE_CASE__							,		SCREAMING_SNAKE_CASE__							,		SCREAMING_SNAKE_CASE__       							= pt_weights[f'''{prefix}.in_proj_bias'''].chunk(3   ,       dim=0  )

    SCREAMING_SNAKE_CASE__       							= pt_weights[f'''{prefix}.out_proj.weight''']
    SCREAMING_SNAKE_CASE__       							= pt_weights[f'''{prefix}.out_proj.bias''']

    SCREAMING_SNAKE_CASE__       							= q_proj
    SCREAMING_SNAKE_CASE__       							= q_proj_bias

    SCREAMING_SNAKE_CASE__       							= k_proj
    SCREAMING_SNAKE_CASE__       							= k_proj_bias

    SCREAMING_SNAKE_CASE__       							= v_proj
    SCREAMING_SNAKE_CASE__       							= v_proj_bias

    SCREAMING_SNAKE_CASE__       							= out_proj_weights
    SCREAMING_SNAKE_CASE__       							= out_proj_bias


def        __snake_case       (						lowerCAmelCase_   ,       lowerCAmelCase_   ,       lowerCAmelCase_  )							->	Dict:
    copy_linear(hf_mlp.fca   ,       lowerCAmelCase_   ,       f'''{prefix}.c_fc'''  )
    copy_linear(hf_mlp.fca   ,       lowerCAmelCase_   ,       f'''{prefix}.c_proj'''  )


def        __snake_case       (						lowerCAmelCase_   ,       lowerCAmelCase_   ,       lowerCAmelCase_  )							->	str:
    SCREAMING_SNAKE_CASE__       							= pt_weights[f'''{prefix}.weight'''].data
    SCREAMING_SNAKE_CASE__       							= pt_weights[f'''{prefix}.bias'''].data


def        __snake_case       (						lowerCAmelCase_   ,       lowerCAmelCase_   ,       lowerCAmelCase_  )							->	Dict:
    # copy layer norms
    copy_linear(hf_layer.layer_norma   ,       lowerCAmelCase_   ,       f'''{prefix}.ln_1'''  )
    copy_linear(hf_layer.layer_norma   ,       lowerCAmelCase_   ,       f'''{prefix}.ln_2'''  )

    # copy MLP
    copy_mlp(hf_layer.mlp   ,       lowerCAmelCase_   ,       f'''{prefix}.mlp'''  )

    # copy attn
    copy_attn_layer(hf_layer.self_attn   ,       lowerCAmelCase_   ,       f'''{prefix}.attn'''  )


def        __snake_case       (						lowerCAmelCase_   ,       lowerCAmelCase_   ,       lowerCAmelCase_  )							->	Optional[int]:
    for layer_id, hf_layer in enumerate(lowerCAmelCase_  ):
        copy_layer(lowerCAmelCase_   ,       lowerCAmelCase_   ,       f'''{prefix}.{layer_id}'''  )


def        __snake_case       (						lowerCAmelCase_   ,       lowerCAmelCase_  )							->	Any:
    # copy projection
    SCREAMING_SNAKE_CASE__       							= pt_weights['''text_projection'''].data.T

    # copy text encoder
    for name, param in hf_model.text_model.named_parameters():
        SCREAMING_SNAKE_CASE__       							= pt_weights[f'''bert.{name}'''].data


def        __snake_case       (						lowerCAmelCase_   ,       lowerCAmelCase_  )							->	List[str]:
    # copy projection
    SCREAMING_SNAKE_CASE__       							= pt_weights['''visual.proj'''].data.T

    # copy layer norms
    copy_linear(hf_model.vision_model.pre_layrnorm   ,       lowerCAmelCase_   ,       '''visual.ln_pre'''  )
    copy_linear(hf_model.vision_model.post_layernorm   ,       lowerCAmelCase_   ,       '''visual.ln_post'''  )

    # copy embeddings
    SCREAMING_SNAKE_CASE__       							= pt_weights['''visual.conv1.weight'''].data
    SCREAMING_SNAKE_CASE__       							= pt_weights['''visual.class_embedding'''].data
    SCREAMING_SNAKE_CASE__       							= pt_weights['''visual.positional_embedding'''].data

    # copy encoder
    copy_layers(hf_model.vision_model.encoder.layers   ,       lowerCAmelCase_   ,       '''visual.transformer.resblocks'''  )


@torch.no_grad()
def        __snake_case       (						lowerCAmelCase_   ,       lowerCAmelCase_   ,       lowerCAmelCase_=None  )							->	str:
    assert config_path is not None, "Please specify the ChineseCLIP model config of the corresponding model size."
    SCREAMING_SNAKE_CASE__       							= ChineseCLIPConfig.from_pretrained(lowerCAmelCase_  )

    SCREAMING_SNAKE_CASE__       							= ChineseCLIPModel(lowerCAmelCase_  ).eval()

    SCREAMING_SNAKE_CASE__       							= torch.load(lowerCAmelCase_   ,       map_location='''cpu'''  )['''state_dict''']
    SCREAMING_SNAKE_CASE__       							= {(name[7:] if name.startswith('''module.'''  ) else name): value for name, value in pt_weights.items()}

    copy_text_model_and_projection(lowerCAmelCase_   ,       lowerCAmelCase_  )
    copy_vision_model_and_projection(lowerCAmelCase_   ,       lowerCAmelCase_  )
    SCREAMING_SNAKE_CASE__       							= pt_weights['''logit_scale'''].data

    hf_model.save_pretrained(lowerCAmelCase_  )


if __name__ == "__main__":
  _A				:							Union[str, Any]							= argparse.ArgumentParser()
  parser.add_argument(
      """--pytorch_dump_folder_path""",
      default=None,
      type=str,
      help="""Path to the output folder storing converted hf PyTorch model.""",
  )
  parser.add_argument(
      """--checkpoint_path""", default=None, type=str, help="""Path to original github format ChineseCLIP checkpoint."""
  )
  parser.add_argument(
      """--config_path""", default=None, required=True, type=str, help="""Path to hf config.json of model to convert."""
  )
  _A				:							Union[str, Any]							= parser.parse_args()

  convert_chinese_clip_checkpoint(args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path)
  print("""The conversion is finished!""")
