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__	:  Optional[Any]						=  5E-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors
def 					UpperCamelCase__		(							A__       )  ->   List[str]:
				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 					__snake_case    (     _lowercase      ):




					@staticmethod
					def      __a     (  __UpperCamelCase						)	->    Optional[Any]:



									'''simple docstring'''







									snake_case__   :      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=__lowerCAmelCase       ,			required=__lowerCAmelCase       ,			help='The model name, including owner/organization, as seen on the hub.'       ,			)
									train_parser.add_argument(
									    '--local-dir'       ,			type=__lowerCAmelCase       ,			default=''       ,			help='Optional local directory of the model repository. Defaults to /tmp/{model_name}'       ,			)
									train_parser.add_argument(
									    '--max-error'       ,			type=__lowerCAmelCase       ,			default=__lowerCAmelCase       ,			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=__lowerCAmelCase       ,			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=__lowerCAmelCase       ,			default=__lowerCAmelCase       ,			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=__lowerCAmelCase						)




					@staticmethod
					def      __a     (  __UpperCamelCase       ,			__UpperCamelCase						)	->    List[Any]:



									'''simple docstring'''







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

													return differences

									return _find_pt_tf_differences(__lowerCAmelCase       ,			__lowerCAmelCase       ,			{}						)




					def __init__(  self       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			*__UpperCamelCase       ,			)	->    int:



									'''simple docstring'''







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




					def      __a     (  self       ,			__UpperCamelCase       ,			__UpperCamelCase       ,			__UpperCamelCase						)	->    Dict:



									'''simple docstring'''







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

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

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

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

									return pt_input, tf_input




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

									snake_case__				,    snake_case__   :      Optional[Any]   				= self.get_inputs(__lowerCAmelCase       ,			__lowerCAmelCase       ,			__lowerCAmelCase						)

									with torch.no_grad():
													snake_case__   :      Optional[Any]   				= pt_model(**__lowerCAmelCase       ,			output_hidden_states=__lowerCAmelCase						)
									del pt_model  # will no longer be used, and may have a large memory footprint

									snake_case__   :      List[Any]   				= tf_class.from_pretrained(self._local_dir       ,			from_pt=__lowerCAmelCase						)
									snake_case__   :      int   				= tf_from_pt_model(**__lowerCAmelCase       ,			output_hidden_states=__lowerCAmelCase       ,			training=__lowerCAmelCase						)

									# Confirms that cross loading PT weights into TF worked.
									snake_case__   :      Dict   				= self.find_pt_tf_differences(__lowerCAmelCase       ,			__lowerCAmelCase						)
									snake_case__   :      Tuple   				= {k: v for k, v in crossload_differences.items() if 'hidden' not in k}
									snake_case__   :      Tuple   				= {k: v for k, v in crossload_differences.items() if 'hidden' in k}
									if len(__lowerCAmelCase						) == 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__   :      str   				= max(output_differences.values()						) if output_differences else 0.0
									snake_case__   :      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
									snake_case__   :      str   				= os.path.join(self._local_dir       ,			__lowerCAmelCase						)
									snake_case__   :      Dict   				= os.path.join(self._local_dir       ,			__lowerCAmelCase						)
									if (not os.path.exists(__lowerCAmelCase						) and not os.path.exists(__lowerCAmelCase						)) 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__   :      Optional[Any]   				= tf_class.from_pretrained(self._local_dir						)
									snake_case__   :      Dict   				= tf_model(**__lowerCAmelCase       ,			output_hidden_states=__lowerCAmelCase						)

									snake_case__   :      Union[str, Any]   				= self.find_pt_tf_differences(__lowerCAmelCase       ,			__lowerCAmelCase						)
									snake_case__   :      List[Any]   				= {k: v for k, v in conversion_differences.items() if 'hidden' not in k}
									snake_case__   :      Dict   				= {k: v for k, v in conversion_differences.items() if 'hidden' in k}
									if len(__lowerCAmelCase						) == 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__   :      Union[str, Any]   				= max(output_differences.values()						) if output_differences else 0.0
									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]						)						)

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

													snake_case__   :      Union[str, Any]   				= create_commit(
													    repo_id=self._model_name       ,			operations=__lowerCAmelCase       ,			commit_message=__lowerCAmelCase       ,			commit_description=__lowerCAmelCase       ,			repo_type='model'       ,			create_pr=__lowerCAmelCase       ,			).pr_url
													self._logger.warning(F"""PR open in {hub_pr_url}"""						)









def      a    (   _lowercase :  Dict			,    _lowercase :  List[str]							)  ->List[Any]:
		if len(str(_lowercase							)							) == 1 or len(str(_lowercase							)							) == 1:
				return a * b

		UpperCamelCase__								=						max(len(str(_lowercase							)							)			,    len(str(_lowercase							)							)							)
		UpperCamelCase__								=						ma // 2

		UpperCamelCase__	,      UpperCamelCase__								=						divmod(_lowercase			,    1_0**ma							)
		UpperCamelCase__	,      UpperCamelCase__								=						divmod(_lowercase			,    1_0**ma							)

		UpperCamelCase__								=						karatsuba(_lowercase			,    _lowercase							)
		UpperCamelCase__								=						karatsuba((aa + aa)			,    (ba + ba)							)
		UpperCamelCase__								=						karatsuba(_lowercase			,    _lowercase							)

		return (z * 1_0 ** (2 * ma)) + ((y - z - x) * 1_0 ** (ma)) + (x)

def      a    (   )  ->List[Any]:
		print(karatsuba(1_5_4_6_3			,    2_3_4_8_9							)							)


if __name__ == "__main__":
				main()





