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


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




def 						_lowercase				(  _SCREAMING_SNAKE_CASE				:				List[str]						)    ->     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 							__snake_case(					_lowerCamelCase	):


    '''simple docstring'''





    @staticmethod
    def 						_a							(  __lowerCamelCase	):




         '''simple docstring'''



         __A					:					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 						_a							(  __lowerCamelCase					,       __lowerCamelCase	):




         '''simple docstring'''



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

              return differences

         return _find_pt_tf_differences(a_					,       a_					,       {}	)
    def __init__(  self					,       __lowerCamelCase					,       __lowerCamelCase					,       __lowerCamelCase					,       __lowerCamelCase					,       __lowerCamelCase					,       __lowerCamelCase					,       __lowerCamelCase					,       __lowerCamelCase					,       *__lowerCamelCase					,       ):




         '''simple docstring'''



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




         '''simple docstring'''



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

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

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




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

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

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

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

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

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

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

         __A					:					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...'	)
              __A					:					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_	):
                   __A					:					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:
                   __A					:					Dict								=		[CommitOperationAdd(path_in_repo=a_					,       path_or_fileobj=a_	)]

              __A					:					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}'	)








'''simple docstring'''






from unittest.mock import patch

import datasets
from datasets import Dataset




def        snake_case_   (	)->						str:





       '''simple docstring'''


       _UpperCAmelCase    :		Tuple     	=					Dataset.from_dict({"""col_1""": [3, 2, 0, 1]}						)

       with patch("""tqdm.auto.tqdm"""						) as mock_tqdm:
              datasets.disable_progress_bar()
              dset.map(lambda lowerCAmelCase_						: {"col_2": x["col_1"] + 1}						)
              mock_tqdm.assert_not_called()

              mock_tqdm.reset_mock()

              datasets.enable_progress_bar()
              dset.map(lambda lowerCAmelCase_						: {"col_2": x["col_1"] + 1}						)
              mock_tqdm.assert_called()



