
import copy

from transformers.configuration_utils import PretrainedConfig
from transformers.utils import logging


_A							=       logging.get_logger(__name__)

class 		lowercase_							(							_lowerCamelCase					):
 A__    :					Optional[Any]    =     """hybrid-clip"""
 A__    :					Dict    =     True

 def __init__(   self  ,						__UpperCamelCase=5_1_2  ,						**__UpperCamelCase  ):




        """simple docstring"""


        super().__init__(**a_  )

        if "text_config" not in kwargs:
               raise ValueError("""`text_config` can not be `None`."""  )

        if "vision_config" not in kwargs:
               raise ValueError("""`vision_config` can not be `None`."""  )

        UpperCamelCase_											=							kwargs.pop("""text_config"""  )
        UpperCamelCase_											=							kwargs.pop("""vision_config"""  )

        UpperCamelCase_											=							text_config.pop("""model_type"""  )
        UpperCamelCase_											=							vision_config.pop("""model_type"""  )

        from transformers import AutoConfig

        UpperCamelCase_											=							AutoConfig.for_model(a_  ,						**a_  )

        if vision_model_type == "clip":
               UpperCamelCase_											=							AutoConfig.for_model(a_  ,						**a_  ).vision_config
        elif vision_model_type == "clip_vision_model":
               from transformers import CLIPVisionConfig

               UpperCamelCase_											=							CLIPVisionConfig(**a_  )
        else:
               UpperCamelCase_											=							AutoConfig.for_model(a_  ,						**a_  )

        UpperCamelCase_											=							projection_dim
        UpperCamelCase_											=							1.0

 @classmethod
 def lowerCamelCase_					(   cls  ,						__UpperCamelCase  ,						__UpperCamelCase  ,						**__UpperCamelCase  ):




        """simple docstring"""


        return cls(text_config=text_config.to_dict()  ,						vision_config=vision_config.to_dict()  ,						**a_  )

 def lowerCamelCase_					(   self  ):




        """simple docstring"""


        UpperCamelCase_											=							copy.deepcopy(self.__dict__  )
        UpperCamelCase_											=							self.text_config.to_dict()
        UpperCamelCase_											=							self.vision_config.to_dict()
        UpperCamelCase_											=							self.__class__.model_type
        return output








'''simple docstring'''






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


A_				:					Optional[Any]							=      5e-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors




def        snake_case_   (	lowerCAmelCase_						)->						Dict:





       '''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 				lowercase			(    _lowerCamelCase							):


 """simple docstring"""


 @staticmethod
 def _snake_case					(				a_				)						->  List[str]:
        _UpperCAmelCase    :		Union[str, 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 _snake_case					(				a_    ,a_				)						->  Union[str, Any]:
        _UpperCAmelCase    :		Optional[int]     	=					set(pt_outputs.keys()				)
        _UpperCAmelCase    :		Any     	=					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(a_    ,a_    ,a_    ,a_=""				):
               # 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				):
                      _UpperCAmelCase    :		List[str]     	=					np.max(np.abs(pt_out.numpy() - tf_out.numpy()				)				)
                      _UpperCAmelCase    :		Dict     	=					tensor_difference
               else:
                      _UpperCAmelCase    :		int     	=					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_				):
                                    _UpperCAmelCase    :		Any     	=					root_name + pt_item
                                    _UpperCAmelCase    :		Optional[int]     	=					tf_out[pt_item]
                                    _UpperCAmelCase    :		Optional[int]     	=					pt_out[pt_item]
                             else:
                                    _UpperCAmelCase    :		List[Any]     	=					root_name + f'''[{i}]'''
                                    _UpperCAmelCase    :		str     	=					tf_out[i]
                             _UpperCAmelCase    :		Tuple     	=					_find_pt_tf_differences(a_    ,a_    ,a_    ,a_				)

               return differences

        return _find_pt_tf_differences(a_    ,a_    ,{}				)







 def __init__(				self    ,a_    ,a_    ,a_    ,a_    ,a_    ,a_    ,a_    ,a_    ,*a_    ,)						->  Optional[int]:
        _UpperCAmelCase    :		Optional[int]     	=					logging.get_logger("""transformers-cli/pt_to_tf"""				)
        _UpperCAmelCase    :		Optional[Any]     	=					model_name
        _UpperCAmelCase    :		str     	=					local_dir if local_dir else os.path.join("""/tmp"""    ,a_				)
        _UpperCAmelCase    :		str     	=					max_error
        _UpperCAmelCase    :		str     	=					new_weights
        _UpperCAmelCase    :		List[Any]     	=					no_pr
        _UpperCAmelCase    :		List[Any]     	=					push
        _UpperCAmelCase    :		List[Any]     	=					extra_commit_description
        _UpperCAmelCase    :		str     	=					override_model_class







 def _snake_case					(				self    ,a_    ,a_    ,a_				)						->  Tuple:

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

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

        _UpperCAmelCase    :		Optional[Any]     	=					set(inspect.signature(pt_model.forward				).parameters.keys()				)
        _UpperCAmelCase    :		str     	=					{}
        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:
               _UpperCAmelCase    :		Optional[Any]     	=					load_dataset("""cifar10"""    ,"""plain_text"""    ,split="""test"""				)[:2]["""img"""]
               processor_inputs.update({"""images""": sample_images}				)
        if "input_features" in model_forward_signature:
               _UpperCAmelCase    :		Optional[int]     	=					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:
                      _UpperCAmelCase    :		Optional[Any]     	=					feature_extractor_signature["""padding"""].default
                      if default_strategy is not False and default_strategy is not None:
                             _UpperCAmelCase    :		Optional[Any]     	=					default_strategy
                      else:
                             _UpperCAmelCase    :		int     	=					True
               else:
                      _UpperCAmelCase    :		str     	=					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}				)
        _UpperCAmelCase    :		Tuple     	=					processor(**a_    ,return_tensors="""pt"""				)
        _UpperCAmelCase    :		Union[str, Any]     	=					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
        ):
               _UpperCAmelCase    :		Dict     	=					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 _snake_case					(				self				)						->  Optional[Any]:
        # hub version 0.9.0 introduced the possibility of programmatically opening PRs with normal write tokens.
        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
        _UpperCAmelCase    :		Tuple     	=					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
        _UpperCAmelCase    :		Optional[int]     	=					AutoConfig.from_pretrained(self._local_dir				)
        _UpperCAmelCase    :		Any     	=					config.architectures
        if self._override_model_class is not None:
               if self._override_model_class.startswith("""TF"""				):
                      _UpperCAmelCase    :		int     	=					[self._override_model_class[2:]]
               else:
                      _UpperCAmelCase    :		int     	=					[self._override_model_class]
               try:
                      _UpperCAmelCase    :		str     	=					getattr(import_module("""transformers"""				)    ,architectures[0]				)
               except AttributeError:
                      raise ValueError(f'''Model class {self._override_model_class} not found in transformers.'''				)
               try:
                      _UpperCAmelCase    :		Any     	=					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
               _UpperCAmelCase    :		Tuple     	=					getattr(import_module("""transformers"""				)    ,"""AutoModel"""				)
               _UpperCAmelCase    :		Any     	=					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]}'''				)
               _UpperCAmelCase    :		int     	=					getattr(import_module("""transformers"""				)    ,architectures[0]				)
               try:
                      _UpperCAmelCase    :		List[Any]     	=					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
        _UpperCAmelCase    :		Union[str, Any]     	=					tf_class.from_config(a_				)
        _UpperCAmelCase    :		Dict     	=					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
        _UpperCAmelCase    :		Union[str, Any]     	=					pt_class.from_pretrained(self._local_dir				)
        pt_model.eval()

        _UpperCAmelCase	,_UpperCAmelCase    :		Dict     	=					self.get_inputs(a_    ,a_    ,a_				)

        with torch.no_grad():
               _UpperCAmelCase    :		List[str]     	=					pt_model(**a_    ,output_hidden_states=a_				)
        del pt_model  # will no longer be used, and may have a large memory footprint

        _UpperCAmelCase    :		Optional[int]     	=					tf_class.from_pretrained(self._local_dir    ,from_pt=a_				)
        _UpperCAmelCase    :		Dict     	=					tf_from_pt_model(**a_    ,output_hidden_states=a_    ,training=a_				)

        # Confirms that cross loading PT weights into TF worked.
        _UpperCAmelCase    :		int     	=					self.find_pt_tf_differences(a_    ,a_				)
        _UpperCAmelCase    :		Any     	=					{k: v for k, v in crossload_differences.items() if """hidden""" not in k}
        _UpperCAmelCase    :		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'"""				)
        _UpperCAmelCase    :		str     	=					max(output_differences.values()				) if output_differences else 0.0
        _UpperCAmelCase    :		Any     	=					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
        _UpperCAmelCase    :		Union[str, Any]     	=					os.path.join(self._local_dir    ,a_				)
        _UpperCAmelCase    :		List[str]     	=					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

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

        _UpperCAmelCase    :		Dict     	=					self.find_pt_tf_differences(a_    ,a_				)
        _UpperCAmelCase    :		Optional[Any]     	=					{k: v for k, v in conversion_differences.items() if """hidden""" not in k}
        _UpperCAmelCase    :		Tuple     	=					{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'"""				)
        _UpperCAmelCase    :		int     	=					max(output_differences.values()				) if output_differences else 0.0
        _UpperCAmelCase    :		List[str]     	=					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]				)				)

        _UpperCAmelCase    :		Dict     	=					"""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..."""				)
               _UpperCAmelCase    :		List[Any]     	=					(
                   """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_				):
                      _UpperCAmelCase    :		List[Any]     	=					[
                          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:
                      _UpperCAmelCase    :		Dict     	=					[CommitOperationAdd(path_in_repo=a_    ,path_or_fileobj=a_				)]

               _UpperCAmelCase    :		Dict     	=					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}'''				)



