"""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     __UpperCAmelCase							(       UpperCAmelCase_       :							Namespace    )				->     str:

	'''simple docstring'''





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

class 						UpperCamelCase     (       lowercase		):



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

						return differences

					return _find_pt_tf_differences(_A      ,    _A      ,    {})



				def __init__(self							: str      ,    _A							: str      ,    _A							: str      ,    _A							: float      ,    _A							: bool      ,    _A							: bool      ,    _A							: bool      ,    _A							: str      ,    _A							: str      ,    *_A							: Any      ,    )       ->      int:
					__snake_case     :				Optional[int] 						=					logging.get_logger('transformers-cli/pt_to_tf')
					__snake_case     :				Dict 						=					model_name
					__snake_case     :				str 						=					local_dir if local_dir else os.path.join('/tmp'      ,    _A)
					__snake_case     :				Tuple 						=					max_error
					__snake_case     :				Dict 						=					new_weights
					__snake_case     :				Tuple 						=					no_pr
					__snake_case     :				List[Any] 						=					push
					__snake_case     :				int 						=					extra_commit_description
					__snake_case     :				List[Any] 						=					override_model_class



				def    _lowercase	(self							: Union[str, Any]      ,    _A							: Optional[int]      ,    _A							: str      ,    _A							: Optional[Any])       ->      Any:

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

					__snake_case     :				Optional[int] 						=					type(pt_model.config)
					if model_config_class in PROCESSOR_MAPPING:
						__snake_case     :				Optional[Any] 						=					AutoProcessor.from_pretrained(self._local_dir)
						if model_config_class in TOKENIZER_MAPPING and processor.tokenizer.pad_token is None:
							__snake_case     :				int 						=					processor.tokenizer.eos_token
					elif model_config_class in IMAGE_PROCESSOR_MAPPING:
						__snake_case     :				Dict 						=					AutoImageProcessor.from_pretrained(self._local_dir)
					elif model_config_class in FEATURE_EXTRACTOR_MAPPING:
						__snake_case     :				Optional[int] 						=					AutoFeatureExtractor.from_pretrained(self._local_dir)
					elif model_config_class in TOKENIZER_MAPPING:
						__snake_case     :				str 						=					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     :				Optional[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     :				Any 						=					load_dataset('cifar10'      ,    'plain_text'      ,    split='test')[:2]['img']
						processor_inputs.update({'images': sample_images})
					if "input_features" in model_forward_signature:
						__snake_case     :				str 						=					inspect.signature(processor.feature_extractor).parameters
						# Pad to the largest input length by default but take feature extractor default
						# padding value if it exists e.g. "max_length" and is not False or None
						if "padding" in feature_extractor_signature:
							__snake_case     :				Optional[int] 						=					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     :				str 						=					True
						else:
							__snake_case     :				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})
					__snake_case     :				int 						=					processor(**_A      ,    return_tensors='pt')
					__snake_case     :				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
					):
						__snake_case     :				List[str] 						=					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    _lowercase	(self							: List[str])       ->      List[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
					__snake_case     :				Dict 						=					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     :				Optional[Any] 						=					AutoConfig.from_pretrained(self._local_dir)
					__snake_case     :				List[str] 						=					config.architectures
					if self._override_model_class is not None:
						if self._override_model_class.startswith('TF'):
							__snake_case     :				int 						=					[self._override_model_class[2:]]
						else:
							__snake_case     :				Union[str, Any] 						=					[self._override_model_class]
						try:
							__snake_case     :				List[str] 						=					getattr(import_module('transformers')      ,    architectures[0])
						except AttributeError:
							raise ValueError(f"Model class {self._override_model_class} not found in transformers.")
						try:
							__snake_case     :				List[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
						__snake_case     :				Optional[int] 						=					getattr(import_module('transformers')      ,    'AutoModel')
						__snake_case     :				Union[str, 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]}")
						__snake_case     :				Any 						=					getattr(import_module('transformers')      ,    architectures[0])
						try:
							__snake_case     :				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
					__snake_case     :				Any 						=					tf_class.from_config(_A)
					__snake_case     :				Any 						=					tf_from_pt_model.dummy_inputs

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

					# Load the model and get some basic inputs
					__snake_case     :				Optional[int] 						=					pt_class.from_pretrained(self._local_dir)
					pt_model.eval()

					__snake_case						,							__snake_case     :				Any 						=					self.get_inputs(_A      ,    _A      ,    _A)

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

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

					# Confirms that cross loading PT weights into TF worked.
					__snake_case     :				List[str] 						=					self.find_pt_tf_differences(_A      ,    _A)
					__snake_case     :				List[str] 						=					{k: v for k, v in crossload_differences.items() if 'hidden' not in k}
					__snake_case     :				Union[str, Any] 						=					{k: v for k, v in crossload_differences.items() if 'hidden' in k}
					if len(_A) == 0 and architectures is not None:
						raise ValueError(
						    f"Something went wrong -- the config file has architectures ({architectures}), but no model head"
						    ' output was found. All outputs start with \'hidden\'')
					__snake_case     :				Optional[Any] 						=					max(output_differences.values()) if output_differences else 0.0
					__snake_case     :				List[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     :				Union[str, Any] 						=					os.path.join(self._local_dir      ,    _A)
					__snake_case     :				Union[str, Any] 						=					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

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

					__snake_case     :				Any 						=					self.find_pt_tf_differences(_A      ,    _A)
					__snake_case     :				Union[str, Any] 						=					{k: v for k, v in conversion_differences.items() if 'hidden' not in k}
					__snake_case     :				List[str] 						=					{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\'')
					__snake_case     :				List[str] 						=					max(output_differences.values()) if output_differences else 0.0
					__snake_case     :				Optional[int] 						=					max(hidden_differences.values())
					if max_conversion_output_diff > self._max_error or max_conversion_hidden_diff > self._max_error:
						raise ValueError(
						    'The converted TensorFlow model has different outputs, something went wrong!\n'
						    + f"\nList of maximum output differences above the threshold ({self._max_error}):\n"
						    + '\n'.join([f"{k}: {v:.3e}" for k, v in output_differences.items() if v > self._max_error])
						    + f"\n\nList of maximum hidden layer differences above the threshold ({self._max_error}):\n"
						    + '\n'.join([f"{k}: {v:.3e}" for k, v in hidden_differences.items() if v > self._max_error]))

					__snake_case     :				Optional[Any] 						=					'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...')
						__snake_case     :				Optional[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):
							__snake_case     :				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:
							__snake_case     :				Dict 						=					[CommitOperationAdd(path_in_repo=_A      ,    path_or_fileobj=_A)]

						__snake_case     :				Optional[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}")






"""simple docstring"""







_a	:	Optional[Any]=					{
    "+": lambda x, y: x + y,
    "-": lambda x, y: x - y,
    "*": lambda x, y: x * y,
    "/": lambda x, y: x / y,
}
def     __UpperCAmelCase							(       UpperCAmelCase_       :							List[Any]    )				->     List[Any]:

	'''simple docstring'''





	return c.isdigit()
def     __UpperCAmelCase							(       UpperCAmelCase_       :							int    )				->     List[Any]:

	'''simple docstring'''





	__snake_case     :				Dict 						=					[]

	# iterate over the string in reverse order
	for c in expression.split()[::-1]:
		# push operand to stack
		if is_operand(UpperCAmelCase_    ):
			stack.append(int(UpperCAmelCase_    )    )

		else:
			# pop values from stack can calculate the result
			# push the result onto the stack again
			__snake_case     :				Optional[Any] 						=					stack.pop()
			__snake_case     :				int 						=					stack.pop()
			stack.append(calc[c](UpperCAmelCase_     ,				UpperCAmelCase_    )    )

	return stack.pop()


# Driver code
if __name__ == "__main__":
	_a	:	Union[str, Any]=					"+ 9 * 2 6"
	print(evaluate(test_expression))

	_a	:	List[Any]=					"/ * 10 2 + 4 1 "
	print(evaluate(test_expression))





