




'''simple docstring'''

import unittest
import warnings

from datasets.utils import experimental




@experimental
def    snake_case__     (  )			->       Union[str, Any]:





			'''simple docstring'''






			return "success"





class __magic_name__(						unittest.TestCase      ):






						def       __lowerCAmelCase(   self						:						List[Any]       ):



									'''simple docstring'''






									with warnings.catch_warnings(record=__UpperCamelCase       ) as w:
												warnings.simplefilter("""always"""       )
												self.assertEqual(dummy_function()					,   """success"""       )
									self.assertEqual(len(__UpperCamelCase       )					,   1       )





"""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


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




def    snake_case__						(							lowerCAmelCase :		Namespace		):
					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__   (   lowercase_  ):




	@staticmethod
	def 				a_		( lowercase	:						ArgumentParser						)       ->			List[Any]:
						__snake_case        =						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=lowercase   ,     required=lowercase   ,     help="""The model name, including owner/organization, as seen on the hub."""   ,     )
						train_parser.add_argument(
						    """--local-dir"""   ,     type=lowercase   ,     default=""""""   ,     help="""Optional local directory of the model repository. Defaults to /tmp/{model_name}"""   ,     )
						train_parser.add_argument(
						    """--max-error"""   ,     type=lowercase   ,     default=lowercase   ,     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=lowercase   ,     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=lowercase   ,     default=lowercase   ,     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=lowercase						)




	@staticmethod
	def 				a_		( lowercase	:						Optional[Any]   ,     lowercase	:						Any						)       ->			Optional[Any]:
						__snake_case        =						set(pt_outputs.keys()						)
						__snake_case        =						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(lowercase	:						Optional[int]   ,     lowercase	:						Any   ,     lowercase	:						List[str]   ,     lowercase	:						Tuple=""						):
											# 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(lowercase   ,     torch.Tensor						):
																__snake_case        =						np.max(np.abs(pt_out.numpy() - tf_out.numpy()						)						)
																__snake_case        =						tensor_difference
											else:
																__snake_case        =						attr_name
																for i, pt_item in enumerate(lowercase						):
																					# If it is a named attribute, we keep the name. Otherwise, just its index.
																					if isinstance(lowercase   ,     lowercase						):
																										__snake_case        =						root_name + pt_item
																										__snake_case        =						tf_out[pt_item]
																										__snake_case        =						pt_out[pt_item]
																					else:
																										__snake_case        =						root_name + f'''[{i}]'''
																										__snake_case        =						tf_out[i]
																					__snake_case        =						_find_pt_tf_differences(lowercase   ,     lowercase   ,     lowercase   ,     lowercase						)

											return differences

						return _find_pt_tf_differences(lowercase   ,     lowercase   ,     {}						)




	def __init__( self	:						Tuple   ,     lowercase	:						str   ,     lowercase	:						str   ,     lowercase	:						float   ,     lowercase	:						bool   ,     lowercase	:						bool   ,     lowercase	:						bool   ,     lowercase	:						str   ,     lowercase	:						str   ,     *lowercase	:						Dict   ,     )       ->			Dict:
						__snake_case        =						logging.get_logger("""transformers-cli/pt_to_tf"""						)
						__snake_case        =						model_name
						__snake_case        =						local_dir if local_dir else os.path.join("""/tmp"""   ,     lowercase						)
						__snake_case        =						max_error
						__snake_case        =						new_weights
						__snake_case        =						no_pr
						__snake_case        =						push
						__snake_case        =						extra_commit_description
						__snake_case        =						override_model_class




	def 				a_		( self	:						List[str]   ,     lowercase	:						str   ,     lowercase	:						Dict   ,     lowercase	:						List[str]						)       ->			Optional[Any]:

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

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

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

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

						return pt_input, tf_input




	def 				a_		( self	:						Optional[Any]						)       ->			Dict:
						# 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
						__snake_case        =						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
						__snake_case        =						AutoConfig.from_pretrained(self._local_dir						)
						__snake_case        =						config.architectures
						if self._override_model_class is not None:
											if self._override_model_class.startswith("""TF"""						):
																__snake_case        =						[self._override_model_class[2:]]
											else:
																__snake_case        =						[self._override_model_class]
											try:
																__snake_case        =						getattr(import_module("""transformers"""						)   ,     architectures[0]						)
											except AttributeError:
																raise ValueError(f'''Model class {self._override_model_class} not found in transformers.'''						)
											try:
																__snake_case        =						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
											__snake_case        =						getattr(import_module("""transformers"""						)   ,     """AutoModel"""						)
											__snake_case        =						getattr(import_module("""transformers"""						)   ,     """TFAutoModel"""						)
											self._logger.warning("""No detected architecture, using AutoModel/TFAutoModel"""						)
						else:  # Architecture defined -- use it
											if len(lowercase						) > 1:
																raise ValueError(f'''More than one architecture was found, aborting. (architectures = {architectures})'''						)
											self._logger.warning(f'''Detected architecture: {architectures[0]}'''						)
											__snake_case        =						getattr(import_module("""transformers"""						)   ,     architectures[0]						)
											try:
																__snake_case        =						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
						__snake_case        =						tf_class.from_config(lowercase						)
						__snake_case        =						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
						__snake_case        =						pt_class.from_pretrained(self._local_dir						)
						pt_model.eval()

						__snake_case   ,   __snake_case        =						self.get_inputs(lowercase   ,     lowercase   ,     lowercase						)

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

						__snake_case        =						tf_class.from_pretrained(self._local_dir   ,     from_pt=lowercase						)
						__snake_case        =						tf_from_pt_model(**lowercase   ,     output_hidden_states=lowercase   ,     training=lowercase						)

						# Confirms that cross loading PT weights into TF worked.
						__snake_case        =						self.find_pt_tf_differences(lowercase   ,     lowercase						)
						__snake_case        =						{k: v for k, v in crossload_differences.items() if """hidden""" not in k}
						__snake_case        =						{k: v for k, v in crossload_differences.items() if """hidden""" in k}
						if len(lowercase						) == 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'"""						)
						__snake_case        =						max(output_differences.values()						) if output_differences else 0.0
						__snake_case        =						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
						__snake_case        =						os.path.join(self._local_dir   ,     lowercase						)
						__snake_case        =						os.path.join(self._local_dir   ,     lowercase						)
						if (not os.path.exists(lowercase						) and not os.path.exists(lowercase						)) 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

						__snake_case        =						tf_class.from_pretrained(self._local_dir						)
						__snake_case        =						tf_model(**lowercase   ,     output_hidden_states=lowercase						)

						__snake_case        =						self.find_pt_tf_differences(lowercase   ,     lowercase						)
						__snake_case        =						{k: v for k, v in conversion_differences.items() if """hidden""" not in k}
						__snake_case        =						{k: v for k, v in conversion_differences.items() if """hidden""" in k}
						if len(lowercase						) == 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'"""						)
						__snake_case        =						max(output_differences.values()						) if output_differences else 0.0
						__snake_case        =						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]						)						)

						__snake_case        =						"""Update TF weights""" if self._new_weights else """Add TF weights"""
						if self._push:
											repo.git_add(auto_lfs_track=lowercase						)
											repo.git_commit(lowercase						)
											repo.git_push(blocking=lowercase						)  # 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..."""						)
											__snake_case        =						(
											    """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(lowercase						):
																__snake_case        =						[
																    CommitOperationAdd(path_in_repo=lowercase   ,     path_or_fileobj=lowercase						)
																]
																for shard_path in tf.io.gfile.glob(self._local_dir + """/tf_model-*.h5"""						):
																					operations += [
																					    CommitOperationAdd(path_in_repo=os.path.basename(lowercase						)   ,     path_or_fileobj=lowercase						)
																					]
											else:
																__snake_case        =						[CommitOperationAdd(path_in_repo=lowercase   ,     path_or_fileobj=lowercase						)]

											__snake_case        =						create_commit(
											    repo_id=self._model_name   ,     operations=lowercase   ,     commit_message=lowercase   ,     commit_description=lowercase   ,     repo_type="""model"""   ,     create_pr=lowercase   ,     ).pr_url
											self._logger.warning(f'''PR open in {hub_pr_url}'''						)


