





import inspect
import os
from argparse import ArgumentParser, Namespace
from importlib import import_module

import huggingface_hub
import numpy as np
from packaging import version

from .. import (
    FEATURE_EXTRACTOR_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoImageProcessor,
    AutoProcessor,
    AutoTokenizer,
    is_datasets_available,
    is_tf_available,
    is_torch_available,
)
from ..utils import TF2_WEIGHTS_INDEX_NAME, TF2_WEIGHTS_NAME, logging
from . import BaseTransformersCLICommand


if is_tf_available():
 import tensorflow as tf

 tf.config.experimental.enable_tensor_float_aa_execution(False)

if is_torch_available():
 import torch

if is_datasets_available():
 from datasets import load_dataset


_lowerCAmelCase							:      Dict						=	5e-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors

def      __snake_case						(					_lowerCAmelCase : Dict	) ->						str:

   '''simple docstring'''





   return PTtoTFCommand(
       args.model_name				,				args.local_dir				,				args.max_error				,				args.new_weights				,				args.no_pr				,				args.push				,				args.extra_commit_description				,				args.override_model_class				,				)





class 					__magic_name__  (							_SCREAMING_SNAKE_CASE							):





     """simple docstring"""



     @staticmethod
     def 						SCREAMING_SNAKE_CASE		(							snake_case       :Optional[Any]  ):





        '''simple docstring'''




        A_					:				Optional[Any]          =			parser.add_parser(
            "pt-to-tf"			,      help=(
                "CLI tool to run convert a transformers model from a PyTorch checkpoint to a TensorFlow checkpoint."
                " Can also be used to validate existing weights without opening PRs, with --no-pr."
            )			,      )
        train_parser.add_argument(
            "--model-name"			,      type=__A			,      required=__A			,      help="The model name, including owner/organization, as seen on the hub."			,      )
        train_parser.add_argument(
            "--local-dir"			,      type=__A			,      default=""			,      help="Optional local directory of the model repository. Defaults to /tmp/{model_name}"			,      )
        train_parser.add_argument(
            "--max-error"			,      type=__A			,      default=__A			,      help=(
                f"Maximum error tolerance. Defaults to {MAX_ERROR}. This flag should be avoided, use at your own risk."
            )			,      )
        train_parser.add_argument(
            "--new-weights"			,      action="store_true"			,      help="Optional flag to create new TensorFlow weights, even if they already exist."			,      )
        train_parser.add_argument(
            "--no-pr"			,      action="store_true"			,      help="Optional flag to NOT open a PR with converted weights."  )
        train_parser.add_argument(
            "--push"			,      action="store_true"			,      help="Optional flag to push the weights directly to `main` (requires permissions)"			,      )
        train_parser.add_argument(
            "--extra-commit-description"			,      type=__A			,      default=""			,      help="Optional additional commit description to use when opening a PR (e.g. to tag the owner)."			,      )
        train_parser.add_argument(
            "--override-model-class"			,      type=__A			,      default=__A			,      help="If you think you know better than the auto-detector, you can specify the model class here. "
            "Can be either an AutoModel class or a specific model class like BertForSequenceClassification."			,      )
        train_parser.set_defaults(func=__A  )



     @staticmethod
     def 						SCREAMING_SNAKE_CASE		(							snake_case       :List[str]			,      snake_case       :Tuple  ):





        '''simple docstring'''




        A_					:				Union[str, Any]          =			set(pt_outputs.keys()  )
        A_					:				List[str]          =			set(tf_outputs.keys()  )
        if pt_out_attrs != tf_out_attrs:
           raise ValueError(
               f"The model outputs have different attributes, aborting. (Pytorch: {pt_out_attrs}, TensorFlow:"
               f" {tf_out_attrs})"  )

        # 2. For each output attribute, computes the difference
        def _find_pt_tf_differences(snake_case       :Optional[Any]			,      snake_case       :List[Any]			,      snake_case       :List[Any]			,      snake_case       :Optional[Any]=""  ):
           # If the current attribute is a tensor, it is a leaf and we make the comparison. Otherwise, we will dig in
           # recursivelly, keeping the name of the attribute.
           if isinstance(__A			,      torch.Tensor  ):
              A_					:				Any          =			np.max(np.abs(pt_out.numpy() - tf_out.numpy()  )  )
              A_					:				Optional[int]          =			tensor_difference
           else:
              A_					:				Dict          =			attr_name
              for i, pt_item in enumerate(__A  ):
                 # If it is a named attribute, we keep the name. Otherwise, just its index.
                 if isinstance(__A			,      __A  ):
                    A_					:				Any          =			root_name + pt_item
                    A_					:				Dict          =			tf_out[pt_item]
                    A_					:				Optional[Any]          =			pt_out[pt_item]
                 else:
                    A_					:				List[str]          =			root_name + f"[{i}]"
                    A_					:				Tuple          =			tf_out[i]
                 A_					:				Optional[int]          =			_find_pt_tf_differences(__A			,      __A			,      __A			,      __A  )

           return differences

        return _find_pt_tf_differences(__A			,      __A			,      {}  )



     def __init__(							self       :Tuple			,      snake_case       :str			,      snake_case       :Optional[Any]			,      snake_case       :Tuple			,      snake_case       :List[Any]			,      snake_case       :int			,      snake_case       :int			,      snake_case       :List[Any]			,      snake_case       :Optional[int]			,      *snake_case       :int			,      ):





        '''simple docstring'''




        A_					:				str          =			logging.get_logger("transformers-cli/pt_to_tf"  )
        A_					:				int          =			model_name
        A_					:				List[str]          =			local_dir if local_dir else os.path.join("/tmp"			,      __A  )
        A_					:				int          =			max_error
        A_					:				Dict          =			new_weights
        A_					:				str          =			no_pr
        A_					:				Optional[Any]          =			push
        A_					:				Any          =			extra_commit_description
        A_					:				str          =			override_model_class



     def 						SCREAMING_SNAKE_CASE		(							self       :Any			,      snake_case       :Optional[Any]			,      snake_case       :List[str]			,      snake_case       :int  ):





        '''simple docstring'''




        def _get_audio_input():
           A_					:				Optional[int]          =			load_dataset("hf-internal-testing/librispeech_asr_dummy"			,      "clean"			,      split="validation"  )
           A_					:				Any          =			ds.sort("id"  ).select(range(2  )  )[:2]["audio"]
           A_					:				str          =			[x["array"] for x in speech_samples]
           return raw_samples

        A_					:				List[str]          =			type(pt_model.config  )
        if model_config_class in PROCESSOR_MAPPING:
           A_					:				Any          =			AutoProcessor.from_pretrained(self._local_dir  )
           if model_config_class in TOKENIZER_MAPPING and processor.tokenizer.pad_token is None:
              A_					:				Optional[Any]          =			processor.tokenizer.eos_token
        elif model_config_class in IMAGE_PROCESSOR_MAPPING:
           A_					:				int          =			AutoImageProcessor.from_pretrained(self._local_dir  )
        elif model_config_class in FEATURE_EXTRACTOR_MAPPING:
           A_					:				Optional[Any]          =			AutoFeatureExtractor.from_pretrained(self._local_dir  )
        elif model_config_class in TOKENIZER_MAPPING:
           A_					:				Dict          =			AutoTokenizer.from_pretrained(self._local_dir  )
           if processor.pad_token is None:
              A_					:				Dict          =			processor.eos_token
        else:
           raise ValueError(f"Unknown data processing type (model config type: {model_config_class})"  )

        A_					:				str          =			set(inspect.signature(pt_model.forward  ).parameters.keys()  )
        A_					:				int          =			{}
        if "input_ids" in model_forward_signature:
           processor_inputs.update(
               {
                   "text": ["Hi there!", "I am a batch with more than one row and different input lengths."],
                   "padding": True,
                   "truncation": True,
               }  )
        if "pixel_values" in model_forward_signature:
           A_					:				int          =			load_dataset("cifar10"			,      "plain_text"			,      split="test"  )[:2]["img"]
           processor_inputs.update({"images": sample_images}  )
        if "input_features" in model_forward_signature:
           A_					:				str          =			inspect.signature(processor.feature_extractor  ).parameters
           # Pad to the largest input length by default but take feature extractor default
           # padding value if it exists e.g. "max_length" and is not False or None
           if "padding" in feature_extractor_signature:
              A_					:				str          =			feature_extractor_signature["padding"].default
              if default_strategy is not False and default_strategy is not None:
                 A_					:				Optional[int]          =			default_strategy
              else:
                 A_					:				int          =			True
           else:
              A_					:				int          =			True
           processor_inputs.update({"audio": _get_audio_input(), "padding": padding_strategy}  )
        if "input_values" in model_forward_signature:  # Wav2Vec2 audio input
           processor_inputs.update({"audio": _get_audio_input(), "padding": True}  )
        A_					:				Optional[int]          =			processor(**__A			,      return_tensors="pt"  )
        A_					:				List[str]          =			processor(**__A			,      return_tensors="tf"  )

        # Extra input requirements, in addition to the input modality
        if (
            config.is_encoder_decoder
            or (hasattr(__A			,      "encoder"  ) and hasattr(__A			,      "decoder"  ))
            or "decoder_input_ids" in tf_dummy_inputs
        ):
           A_					:				Optional[int]          =			np.asarray([[1], [1]]			,      dtype=__A  ) * (pt_model.config.decoder_start_token_id or 0)
           pt_input.update({"decoder_input_ids": torch.tensor(__A  )}  )
           tf_input.update({"decoder_input_ids": tf.convert_to_tensor(__A  )}  )

        return pt_input, tf_input





     def 						SCREAMING_SNAKE_CASE		(							self       :Dict  ):





        '''simple docstring'''




        if version.parse(huggingface_hub.__version__  ) < version.parse("0.9.0"  ):
           raise ImportError(
               "The huggingface_hub version must be >= 0.9.0 to use this command. Please update your huggingface_hub"
               " installation."  )
        else:
           from huggingface_hub import Repository, create_commit
           from huggingface_hub._commit_api import CommitOperationAdd

        # Fetch remote data
        A_					:				List[str]          =			Repository(local_dir=self._local_dir			,      clone_from=self._model_name  )

        # Load config and get the appropriate architecture -- the latter is needed to convert the head's weights
        A_					:				Optional[int]          =			AutoConfig.from_pretrained(self._local_dir  )
        A_					:				int          =			config.architectures
        if self._override_model_class is not None:
           if self._override_model_class.startswith("TF"  ):
              A_					:				str          =			[self._override_model_class[2:]]
           else:
              A_					:				Tuple          =			[self._override_model_class]
           try:
              A_					:				str          =			getattr(import_module("transformers"  )			,      architectures[0]  )
           except AttributeError:
              raise ValueError(f"Model class {self._override_model_class} not found in transformers."  )
           try:
              A_					:				List[str]          =			getattr(import_module("transformers"  )			,      "TF" + architectures[0]  )
           except AttributeError:
              raise ValueError(f"TF model class TF{self._override_model_class} not found in transformers."  )
        elif architectures is None:  # No architecture defined -- use auto classes
           A_					:				Optional[int]          =			getattr(import_module("transformers"  )			,      "AutoModel"  )
           A_					:				Tuple          =			getattr(import_module("transformers"  )			,      "TFAutoModel"  )
           self._logger.warning("No detected architecture, using AutoModel/TFAutoModel"  )
        else:  # Architecture defined -- use it
           if len(__A  ) > 1:
              raise ValueError(f"More than one architecture was found, aborting. (architectures = {architectures})"  )
           self._logger.warning(f"Detected architecture: {architectures[0]}"  )
           A_					:				str          =			getattr(import_module("transformers"  )			,      architectures[0]  )
           try:
              A_					:				Tuple          =			getattr(import_module("transformers"  )			,      "TF" + architectures[0]  )
           except AttributeError:
              raise AttributeError(f"The TensorFlow equivalent of {architectures[0]} doesn\'t exist in transformers."  )

        # Check the TF dummy inputs to see what keys we need in the forward pass
        A_					:				List[str]          =			tf_class.from_config(__A  )
        A_					:				Any          =			tf_from_pt_model.dummy_inputs

        del tf_from_pt_model  # Try to keep only one model in memory at a time

        # Load the model and get some basic inputs
        A_					:				List[str]          =			pt_class.from_pretrained(self._local_dir  )
        pt_model.eval()

        A_      ,						A_					:				int          =			self.get_inputs(__A			,      __A			,      __A  )

        with torch.no_grad():
           A_					:				Any          =			pt_model(**__A			,      output_hidden_states=__A  )
        del pt_model  # will no longer be used, and may have a large memory footprint

        A_					:				Union[str, Any]          =			tf_class.from_pretrained(self._local_dir			,      from_pt=__A  )
        A_					:				Optional[Any]          =			tf_from_pt_model(**__A			,      output_hidden_states=__A			,      training=__A  )

        # Confirms that cross loading PT weights into TF worked.
        A_					:				Union[str, Any]          =			self.find_pt_tf_differences(__A			,      __A  )
        A_					:				Optional[Any]          =			{k: v for k, v in crossload_differences.items() if "hidden" not in k}
        A_					:				Union[str, Any]          =			{k: v for k, v in crossload_differences.items() if "hidden" in k}
        if len(__A  ) == 0 and architectures is not None:
           raise ValueError(
               f"Something went wrong -- the config file has architectures ({architectures}), but no model head"
               " output was found. All outputs start with \'hidden\'"  )
        A_					:				List[Any]          =			max(output_differences.values()  ) if output_differences else 0.0
        A_					:				Optional[int]          =			max(hidden_differences.values()  )
        if max_crossload_output_diff > self._max_error or max_crossload_hidden_diff > self._max_error:
           raise ValueError(
               "The cross-loaded TensorFlow model has different outputs, something went wrong!\n"
               + f"\nList of maximum output differences above the threshold ({self._max_error}):\n"
               + "\n".join([f"{k}: {v:.3e}" for k, v in output_differences.items() if v > self._max_error]  )
               + f"\n\nList of maximum hidden layer differences above the threshold ({self._max_error}):\n"
               + "\n".join([f"{k}: {v:.3e}" for k, v in hidden_differences.items() if v > self._max_error]  )  )

        # Save the weights in a TF format (if needed) and confirms that the results are still good
        A_					:				Tuple          =			os.path.join(self._local_dir			,      __A  )
        A_					:				Tuple          =			os.path.join(self._local_dir			,      __A  )
        if (not os.path.exists(__A  ) and not os.path.exists(__A  )) or self._new_weights:
           tf_from_pt_model.save_pretrained(self._local_dir  )
        del tf_from_pt_model  # will no longer be used, and may have a large memory footprint

        A_					:				Union[str, Any]          =			tf_class.from_pretrained(self._local_dir  )
        A_					:				Optional[Any]          =			tf_model(**__A			,      output_hidden_states=__A  )

        A_					:				int          =			self.find_pt_tf_differences(__A			,      __A  )
        A_					:				List[Any]          =			{k: v for k, v in conversion_differences.items() if "hidden" not in k}
        A_					:				List[Any]          =			{k: v for k, v in conversion_differences.items() if "hidden" in k}
        if len(__A  ) == 0 and architectures is not None:
           raise ValueError(
               f"Something went wrong -- the config file has architectures ({architectures}), but no model head"
               " output was found. All outputs start with \'hidden\'"  )
        A_					:				Dict          =			max(output_differences.values()  ) if output_differences else 0.0
        A_					:				Optional[int]          =			max(hidden_differences.values()  )
        if max_conversion_output_diff > self._max_error or max_conversion_hidden_diff > self._max_error:
           raise ValueError(
               "The converted TensorFlow model has different outputs, something went wrong!\n"
               + f"\nList of maximum output differences above the threshold ({self._max_error}):\n"
               + "\n".join([f"{k}: {v:.3e}" for k, v in output_differences.items() if v > self._max_error]  )
               + f"\n\nList of maximum hidden layer differences above the threshold ({self._max_error}):\n"
               + "\n".join([f"{k}: {v:.3e}" for k, v in hidden_differences.items() if v > self._max_error]  )  )

        A_					:				List[str]          =			"Update TF weights" if self._new_weights else "Add TF weights"
        if self._push:
           repo.git_add(auto_lfs_track=__A  )
           repo.git_commit(__A  )
           repo.git_push(blocking=__A  )  # this prints a progress bar with the upload
           self._logger.warning(f"TF weights pushed into {self._model_name}"  )
        elif not self._no_pr:
           self._logger.warning("Uploading the weights into a new PR..."  )
           A_					:				Tuple          =			(
               "Model converted by the [`transformers`\' `pt_to_tf`"
               " CLI](https://github.com/huggingface/transformers/blob/main/src/transformers/commands/pt_to_tf.py). "
               "All converted model outputs and hidden layers were validated against its PyTorch counterpart.\n\n"
               f"Maximum crossload output difference={max_crossload_output_diff:.3e}; "
               f"Maximum crossload hidden layer difference={max_crossload_hidden_diff:.3e};\n"
               f"Maximum conversion output difference={max_conversion_output_diff:.3e}; "
               f"Maximum conversion hidden layer difference={max_conversion_hidden_diff:.3e};\n"
           )
           if self._max_error > MAX_ERROR:
              commit_descrition += (
                  f"\n\nCAUTION: The maximum admissible error was manually increased to {self._max_error}!"
              )
           if self._extra_commit_description:
              commit_descrition += "\n\n" + self._extra_commit_description

           # sharded model -> adds all related files (index and .h5 shards)
           if os.path.exists(__A  ):
              A_					:				Optional[int]          =			[
                  CommitOperationAdd(path_in_repo=__A			,      path_or_fileobj=__A  )
              ]
              for shard_path in tf.io.gfile.glob(self._local_dir + "/tf_model-*.h5"  ):
                 operations += [
                     CommitOperationAdd(path_in_repo=os.path.basename(__A  )			,      path_or_fileobj=__A  )
                 ]
           else:
              A_					:				Tuple          =			[CommitOperationAdd(path_in_repo=__A			,      path_or_fileobj=__A  )]

           A_					:				int          =			create_commit(
               repo_id=self._model_name			,      operations=__A			,      commit_message=__A			,      commit_description=__A			,      repo_type="model"			,      create_pr=__A			,      ).pr_url
           self._logger.warning(f"PR open in {hub_pr_url}"  )







import argparse

import torch

from transformers import ChineseCLIPConfig, ChineseCLIPModel

def      __snake_case						(					_lowerCAmelCase : str				,				_lowerCAmelCase : Optional[int]				,				_lowerCAmelCase : Optional[Any]	) ->						List[str]:
   A_      ,						A_      ,						A_					:				Dict          =			pt_weights[f"{prefix}.in_proj_weight"].chunk(3				,				dim=0	)
   A_      ,						A_      ,						A_					:				Union[str, Any]          =			pt_weights[f"{prefix}.in_proj_bias"].chunk(3				,				dim=0	)

   A_					:				int          =			pt_weights[f"{prefix}.out_proj.weight"]
   A_					:				Dict          =			pt_weights[f"{prefix}.out_proj.bias"]

   A_					:				Optional[int]          =			q_proj
   A_					:				Optional[int]          =			q_proj_bias

   A_					:				Tuple          =			k_proj
   A_					:				str          =			k_proj_bias

   A_					:				Optional[int]          =			v_proj
   A_					:				int          =			v_proj_bias

   A_					:				int          =			out_proj_weights
   A_					:				List[Any]          =			out_proj_bias

def      __snake_case						(					_lowerCAmelCase : int				,				_lowerCAmelCase : Dict				,				_lowerCAmelCase : int	) ->						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 : Tuple				,				_lowerCAmelCase : Optional[Any]				,				_lowerCAmelCase : Optional[int]	) ->						List[str]:
   A_					:				List[str]          =			pt_weights[f"{prefix}.weight"].data
   A_					:				int          =			pt_weights[f"{prefix}.bias"].data

def      __snake_case						(					_lowerCAmelCase : Dict				,				_lowerCAmelCase : str				,				_lowerCAmelCase : Optional[Any]	) ->						str:
   # 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 : Optional[Any]				,				_lowerCAmelCase : int				,				_lowerCAmelCase : Optional[Any]	) ->						List[Any]:
   for layer_id, hf_layer in enumerate(_lowerCAmelCase	):
      copy_layer(_lowerCAmelCase				,				_lowerCAmelCase				,				f"{prefix}.{layer_id}"	)

def      __snake_case						(					_lowerCAmelCase : Union[str, Any]				,				_lowerCAmelCase : Any	) ->						Dict:
   # copy projection
   A_					:				Optional[Any]          =			pt_weights["text_projection"].data.T

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

def      __snake_case						(					_lowerCAmelCase : int				,				_lowerCAmelCase : Any	) ->						List[str]:
   # copy projection
   A_					:				Tuple          =			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
   A_					:				List[Any]          =			pt_weights["visual.conv1.weight"].data
   A_					:				Union[str, Any]          =			pt_weights["visual.class_embedding"].data
   A_					:				Tuple          =			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 : List[Any]				,				_lowerCAmelCase : int				,				_lowerCAmelCase : Optional[int]=None	) ->						str:
   assert config_path is not None, "Please specify the ChineseCLIP model config of the corresponding model size."
   A_					:				Optional[Any]          =			ChineseCLIPConfig.from_pretrained(_lowerCAmelCase	)

   A_					:				Tuple          =			ChineseCLIPModel(_lowerCAmelCase	).eval()

   A_					:				Any          =			torch.load(_lowerCAmelCase				,				map_location="cpu"	)["state_dict"]
   A_					:				Optional[Any]          =			{(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	)
   A_					:				List[Any]          =			pt_weights["logit_scale"].data

   hf_model.save_pretrained(_lowerCAmelCase	)


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

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