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 :     List[Any]          =  5E-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors




def 				_snake_case     (       lowerCAmelCase							:       Namespace     ):
		"""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 		a__						( A__ ):

					@staticmethod
					def     __UpperCamelCase       (   _A				:	ArgumentParser   ):







							"""simple docstring"""






							SCREAMING_SNAKE_CASE_  :		Dict    				=   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     __UpperCamelCase       (   _A				:	List[str],_A				:	Union[str, Any]   ):







							"""simple docstring"""






							SCREAMING_SNAKE_CASE_  :		Union[str, Any]    				=   set(pt_outputs.keys()   )
							SCREAMING_SNAKE_CASE_  :		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(_A				:	int,_A				:	Any,_A				:	List[str],_A				:	Union[str, 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   ):
											SCREAMING_SNAKE_CASE_  :		Dict    				=   np.max(np.abs(pt_out.numpy() - tf_out.numpy()   )   )
											SCREAMING_SNAKE_CASE_  :		Any    				=   tensor_difference
									else:
											SCREAMING_SNAKE_CASE_  :		Tuple    				=   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   ):
															SCREAMING_SNAKE_CASE_  :		Union[str, Any]    				=   root_name + pt_item
															SCREAMING_SNAKE_CASE_  :		List[str]    				=   tf_out[pt_item]
															SCREAMING_SNAKE_CASE_  :		List[Any]    				=   pt_out[pt_item]
													else:
															SCREAMING_SNAKE_CASE_  :		str    				=   root_name + F'[{i}]'
															SCREAMING_SNAKE_CASE_  :		List[str]    				=   tf_out[i]
													SCREAMING_SNAKE_CASE_  :		Tuple    				=   _find_pt_tf_differences(_A,_A,_A,_A   )

									return differences

							return _find_pt_tf_differences(_A,_A,{}   )

					def __init__(   self				:	Dict,_A				:	str,_A				:	str,_A				:	float,_A				:	bool,_A				:	bool,_A				:	bool,_A				:	str,_A				:	str,*_A				:	Any,):







							"""simple docstring"""






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

					def     __UpperCamelCase       (   self				:	Optional[Any],_A				:	Dict,_A				:	Union[str, Any],_A				:	Dict   ):







							"""simple docstring"""






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

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

							SCREAMING_SNAKE_CASE_  :		Optional[Any]    				=   set(inspect.signature(pt_model.forward   ).parameters.keys()   )
							SCREAMING_SNAKE_CASE_  :		Tuple    				=   {}
							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:
									SCREAMING_SNAKE_CASE_  :		Union[str, Any]    				=   load_dataset("cifar10","plain_text",split="test"   )[:2]["img"]
									processor_inputs.update({"images": sample_images}   )
							if "input_features" in model_forward_signature:
									SCREAMING_SNAKE_CASE_  :		Dict    				=   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:
											SCREAMING_SNAKE_CASE_  :		Optional[Any]    				=   feature_extractor_signature["padding"].default
											if default_strategy is not False and default_strategy is not None:
													SCREAMING_SNAKE_CASE_  :		List[Any]    				=   default_strategy
											else:
													SCREAMING_SNAKE_CASE_  :		Union[str, Any]    				=   True
									else:
											SCREAMING_SNAKE_CASE_  :		Optional[Any]    				=   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}   )
							SCREAMING_SNAKE_CASE_  :		List[Any]    				=   processor(**_A,return_tensors="pt"   )
							SCREAMING_SNAKE_CASE_  :		List[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
							):
									SCREAMING_SNAKE_CASE_  :		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     __UpperCamelCase       (   self				:	Union[str, Any]   ):







							"""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
							SCREAMING_SNAKE_CASE_  :		Optional[Any]    				=   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
							SCREAMING_SNAKE_CASE_  :		str    				=   AutoConfig.from_pretrained(self._local_dir   )
							SCREAMING_SNAKE_CASE_  :		str    				=   config.architectures
							if self._override_model_class is not None:
									if self._override_model_class.startswith("TF"   ):
											SCREAMING_SNAKE_CASE_  :		Tuple    				=   [self._override_model_class[2:]]
									else:
											SCREAMING_SNAKE_CASE_  :		int    				=   [self._override_model_class]
									try:
											SCREAMING_SNAKE_CASE_  :		Optional[int]    				=   getattr(import_module("transformers"   ),architectures[0]   )
									except AttributeError:
											raise ValueError(F'Model class {self._override_model_class} not found in transformers.'   )
									try:
											SCREAMING_SNAKE_CASE_  :		Optional[int]    				=   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
									SCREAMING_SNAKE_CASE_  :		Union[str, Any]    				=   getattr(import_module("transformers"   ),"AutoModel"   )
									SCREAMING_SNAKE_CASE_  :		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]}'   )
									SCREAMING_SNAKE_CASE_  :		Tuple    				=   getattr(import_module("transformers"   ),architectures[0]   )
									try:
											SCREAMING_SNAKE_CASE_  :		Optional[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
							SCREAMING_SNAKE_CASE_  :		str    				=   tf_class.from_config(_A   )
							SCREAMING_SNAKE_CASE_  :		str    				=   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
							SCREAMING_SNAKE_CASE_  :		Any    				=   pt_class.from_pretrained(self._local_dir   )
							pt_model.eval()

							SCREAMING_SNAKE_CASE_				,   SCREAMING_SNAKE_CASE_  :		List[str]    				=   self.get_inputs(_A,_A,_A   )

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

							SCREAMING_SNAKE_CASE_  :		Tuple    				=   tf_class.from_pretrained(self._local_dir,from_pt=_A   )
							SCREAMING_SNAKE_CASE_  :		List[Any]    				=   tf_from_pt_model(**_A,output_hidden_states=_A,training=_A   )

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

							SCREAMING_SNAKE_CASE_  :		int    				=   tf_class.from_pretrained(self._local_dir   )
							SCREAMING_SNAKE_CASE_  :		List[str]    				=   tf_model(**_A,output_hidden_states=_A   )

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

							SCREAMING_SNAKE_CASE_  :		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..."   )
									SCREAMING_SNAKE_CASE_  :		int    				=   (
									    "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   ):
											SCREAMING_SNAKE_CASE_  :		str    				=   [
											    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:
											SCREAMING_SNAKE_CASE_  :		Any    				=   [CommitOperationAdd(path_in_repo=_A,path_or_fileobj=_A   )]

									SCREAMING_SNAKE_CASE_  :		List[Any]    				=   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

from transformers import ConvBertConfig, ConvBertModel, TFConvBertModel, load_tf_weights_in_convbert
from transformers.utils import logging


logging.set_verbosity_info()




def 				_snake_case     (       lowerCAmelCase							:       Dict       ,			lowerCAmelCase							:       int       ,			lowerCAmelCase							:       Union[str, Any]     ):
		"""simple docstring"""







		SCREAMING_SNAKE_CASE_  :		List[Any]    				=   ConvBertConfig.from_json_file(lowerCAmelCase     )
		SCREAMING_SNAKE_CASE_  :		Tuple    				=   ConvBertModel(lowerCAmelCase     )

		SCREAMING_SNAKE_CASE_  :		Dict    				=   load_tf_weights_in_convbert(lowerCAmelCase       ,			lowerCAmelCase       ,			lowerCAmelCase     )
		model.save_pretrained(lowerCAmelCase     )

		SCREAMING_SNAKE_CASE_  :		Union[str, Any]    				=   TFConvBertModel.from_pretrained(lowerCAmelCase       ,			from_pt=lowerCAmelCase     )
		tf_model.save_pretrained(lowerCAmelCase     )


if __name__ == "__main__":
		__lowerCamelCase :     Optional[Any]          =  argparse.ArgumentParser()
		# Required parameters
		parser.add_argument(
		    '''--tf_checkpoint_path''', default=None, type=str, required=True, help='''Path to the TensorFlow checkpoint path.'''
		)
		parser.add_argument(
		    '''--convbert_config_file''',
		    default=None,
		    type=str,
		    required=True,
		    help=(
		        '''The config json file corresponding to the pre-trained ConvBERT model. \n'''
		        '''This specifies the model architecture.'''
		    ),
		)
		parser.add_argument(
		    '''--pytorch_dump_path''', default=None, type=str, required=True, help='''Path to the output PyTorch model.'''
		)
		__lowerCamelCase :     Optional[int]          =  parser.parse_args()
		convert_orig_tfa_checkpoint_to_pytorch(args.tf_checkpoint_path, args.convbert_config_file, args.pytorch_dump_path)
