code: 
  
  
  
  
  
  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
  
  
  _lowercase		:     int 					=5e-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors
  
  
  def 		_SCREAMING_SNAKE_CASE    (   lowerCAmelCase__	):
    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_     (     a_       ):
  
  
  
       @staticmethod
       def      __a    (		lowerCamelCase    :    List[Any]	):
         lowerCamelCase_  :    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=_snake_case					,			required=_snake_case					,			help='The model name, including owner/organization, as seen on the hub.'					,			)
         train_parser.add_argument(
             '--local-dir'					,			type=_snake_case					,			default=''					,			help='Optional local directory of the model repository. Defaults to /tmp/{model_name}'					,			)
         train_parser.add_argument(
             '--max-error'					,			type=_snake_case					,			default=_snake_case					,			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=_snake_case					,			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=_snake_case					,			default=_snake_case					,			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=_snake_case	)
  
  
  
       @staticmethod
       def      __a    (		lowerCamelCase    :    List[str]					,			lowerCamelCase    :    Dict	):
         lowerCamelCase_  :    Any   	=			set(pt_outputs.keys()	)
         lowerCamelCase_  :    Dict   	=			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    :    List[str]					,			lowerCamelCase    :    List[str]					,			lowerCamelCase    :    Optional[int]					,			lowerCamelCase    :    int=""	):
           # 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(_snake_case					,			torch.Tensor	):
             lowerCamelCase_  :    Optional[int]   	=			np.max(np.abs(pt_out.numpy() - tf_out.numpy()	)	)
             lowerCamelCase_  :    str   	=			tensor_difference
           else:
             lowerCamelCase_  :    List[Any]   	=			attr_name
             for i, pt_item in enumerate(_snake_case	):
               # If it is a named attribute, we keep the name. Otherwise, just its index.
               if isinstance(_snake_case					,			_snake_case	):
                 lowerCamelCase_  :    Dict   	=			root_name + pt_item
                 lowerCamelCase_  :    str   	=			tf_out[pt_item]
                 lowerCamelCase_  :    List[str]   	=			pt_out[pt_item]
               else:
                 lowerCamelCase_  :    List[str]   	=			root_name + F"[{i}]"
                 lowerCamelCase_  :    List[str]   	=			tf_out[i]
               lowerCamelCase_  :    List[Any]   	=			_find_pt_tf_differences(_snake_case					,			_snake_case					,			_snake_case					,			_snake_case	)
  
           return differences
  
         return _find_pt_tf_differences(_snake_case					,			_snake_case					,			{}	)
  
  
  
       def __init__(		self    :    Union[str, Any]					,			lowerCamelCase    :    Dict					,			lowerCamelCase    :    Any					,			lowerCamelCase    :    List[Any]					,			lowerCamelCase    :    List[Any]					,			lowerCamelCase    :    List[str]					,			lowerCamelCase    :    List[str]					,			lowerCamelCase    :    Optional[Any]					,			lowerCamelCase    :    str					,			*lowerCamelCase    :    Optional[Any]					,			):
         lowerCamelCase_  :    int   	=			logging.get_logger('transformers-cli/pt_to_tf'	)
         lowerCamelCase_  :    List[Any]   	=			model_name
         lowerCamelCase_  :    Any   	=			local_dir if local_dir else os.path.join('/tmp'					,			_snake_case	)
         lowerCamelCase_  :    Optional[int]   	=			max_error
         lowerCamelCase_  :    Optional[int]   	=			new_weights
         lowerCamelCase_  :    Dict   	=			no_pr
         lowerCamelCase_  :    Tuple   	=			push
         lowerCamelCase_  :    Tuple   	=			extra_commit_description
         lowerCamelCase_  :    List[Any]   	=			override_model_class
  
  
  
       def      __a    (		self    :    Optional[Any]					,			lowerCamelCase    :    Union[str, Any]					,			lowerCamelCase    :    Optional[int]					,			lowerCamelCase    :    Union[str, Any]	):
  
         def _get_audio_input():
           lowerCamelCase_  :    Dict   	=			load_dataset('hf-internal-testing/librispeech_asr_dummy'					,			'clean'					,			split='validation'	)
           lowerCamelCase_  :    Union[str, Any]   	=			ds.sort('id'	).select(range(2	)	)[:2]['''audio''']
           lowerCamelCase_  :    Dict   	=			[x['''array'''] for x in speech_samples]
           return raw_samples
  
         lowerCamelCase_  :    Optional[int]   	=			type(pt_model.config	)
         if model_config_class in PROCESSOR_MAPPING:
           lowerCamelCase_  :    Optional[Any]   	=			AutoProcessor.from_pretrained(self._local_dir	)
           if model_config_class in TOKENIZER_MAPPING and processor.tokenizer.pad_token is None:
             lowerCamelCase_  :    List[Any]   	=			processor.tokenizer.eos_token
         elif model_config_class in IMAGE_PROCESSOR_MAPPING:
           lowerCamelCase_  :    Optional[Any]   	=			AutoImageProcessor.from_pretrained(self._local_dir	)
         elif model_config_class in FEATURE_EXTRACTOR_MAPPING:
           lowerCamelCase_  :    Any   	=			AutoFeatureExtractor.from_pretrained(self._local_dir	)
         elif model_config_class in TOKENIZER_MAPPING:
           lowerCamelCase_  :    Any   	=			AutoTokenizer.from_pretrained(self._local_dir	)
           if processor.pad_token is None:
             lowerCamelCase_  :    Optional[Any]   	=			processor.eos_token
         else:
           raise ValueError(F"Unknown data processing type (model config type: {model_config_class})"	)
  
         lowerCamelCase_  :    Optional[int]   	=			set(inspect.signature(pt_model.forward	).parameters.keys()	)
         lowerCamelCase_  :    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:
           lowerCamelCase_  :    Optional[int]   	=			load_dataset('cifar10'					,			'plain_text'					,			split='test'	)[:2]['''img''']
           processor_inputs.update({'images': sample_images}	)
         if "input_features" in model_forward_signature:
           lowerCamelCase_  :    Optional[Any]   	=			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:
             lowerCamelCase_  :    Any   	=			feature_extractor_signature['''padding'''].default
             if default_strategy is not False and default_strategy is not None:
               lowerCamelCase_  :    int   	=			default_strategy
             else:
               lowerCamelCase_  :    Any   	=			True
           else:
             lowerCamelCase_  :    int   	=			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}	)
         lowerCamelCase_  :    Optional[int]   	=			processor(**_snake_case					,			return_tensors='pt'	)
         lowerCamelCase_  :    List[str]   	=			processor(**_snake_case					,			return_tensors='tf'	)
  
         # Extra input requirements, in addition to the input modality
         if (
             config.is_encoder_decoder
             or (hasattr(_snake_case					,			'encoder'	) and hasattr(_snake_case					,			'decoder'	))
             or "decoder_input_ids" in tf_dummy_inputs
         ):
           lowerCamelCase_  :    Dict   	=			np.asarray([[1], [1]]					,			dtype=_snake_case	) * (pt_model.config.decoder_start_token_id or 0)
           pt_input.update({'decoder_input_ids': torch.tensor(_snake_case	)}	)
           tf_input.update({'decoder_input_ids': tf.convert_to_tensor(_snake_case	)}	)
  
         return pt_input, tf_input
  
  
  
  
  
       def      __a    (		self    :    List[str]	):
         # 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
         lowerCamelCase_  :    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
         lowerCamelCase_  :    int   	=			AutoConfig.from_pretrained(self._local_dir	)
         lowerCamelCase_  :    int   	=			config.architectures
         if self._override_model_class is not None:
           if self._override_model_class.startswith('TF'	):
             lowerCamelCase_  :    str   	=			[self._override_model_class[2:]]
           else:
             lowerCamelCase_  :    int   	=			[self._override_model_class]
           try:
             lowerCamelCase_  :    Any   	=			getattr(import_module('transformers'	)					,			architectures[0]	)
           except AttributeError:
             raise ValueError(F"Model class {self._override_model_class} not found in transformers."	)
           try:
             lowerCamelCase_  :    Union[str, 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
           lowerCamelCase_  :    List[str]   	=			getattr(import_module('transformers'	)					,			'AutoModel'	)
           lowerCamelCase_  :    Tuple   	=			getattr(import_module('transformers'	)					,			'TFAutoModel'	)
           self._logger.warning('No detected architecture, using AutoModel/TFAutoModel'	)
         else:  # Architecture defined -- use it
           if len(_snake_case	) > 1:
             raise ValueError(F"More than one architecture was found, aborting. (architectures = {architectures})"	)
           self._logger.warning(F"Detected architecture: {architectures[0]}"	)
           lowerCamelCase_  :    Tuple   	=			getattr(import_module('transformers'	)					,			architectures[0]	)
           try:
             lowerCamelCase_  :    List[str]   	=			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
         lowerCamelCase_  :    Tuple   	=			tf_class.from_config(_snake_case	)
         lowerCamelCase_  :    List[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
         lowerCamelCase_  :    Any   	=			pt_class.from_pretrained(self._local_dir	)
         pt_model.eval()
  
         lowerCamelCase_  :    List[Any]   	=			self.get_inputs(_snake_case					,			_snake_case					,			_snake_case	)
  
         with torch.no_grad():
           lowerCamelCase_  :    List[Any]   	=			pt_model(**_snake_case					,			output_hidden_states=_snake_case	)
         del pt_model  # will no longer be used, and may have a large memory footprint
  
         lowerCamelCase_  :    Union[str, Any]   	=			tf_class.from_pretrained(self._local_dir					,			from_pt=_snake_case	)
         lowerCamelCase_  :    Tuple   	=			tf_from_pt_model(**_snake_case					,			output_hidden_states=_snake_case					,			training=_snake_case	)
  
         # Confirms that cross loading PT weights into TF worked.
         lowerCamelCase_  :    Any   	=			self.find_pt_tf_differences(_snake_case					,			_snake_case	)
         lowerCamelCase_  :    List[str]   	=			{k: v for k, v in crossload_differences.items() if '''hidden''' not in k}
         lowerCamelCase_  :    Optional[int]   	=			{k: v for k, v in crossload_differences.items() if '''hidden''' in k}
         if len(_snake_case	) == 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\''	)
         lowerCamelCase_  :    Union[str, Any]   	=			max(output_differences.values()	) if output_differences else 0.0
         lowerCamelCase_  :    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
         lowerCamelCase_  :    Dict   	=			os.path.join(self._local_dir					,			_snake_case	)
         lowerCamelCase_  :    Union[str, Any]   	=			os.path.join(self._local_dir					,			_snake_case	)
         if (not os.path.exists(_snake_case	) and not os.path.exists(_snake_case	)) 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
  
         lowerCamelCase_  :    Optional[int]   	=			tf_class.from_pretrained(self._local_dir	)
         lowerCamelCase_  :    Any   	=			tf_model(**_snake_case					,			output_hidden_states=_snake_case	)
  
         lowerCamelCase_  :    int   	=			self.find_pt_tf_differences(_snake_case					,			_snake_case	)
         lowerCamelCase_  :    str   	=			{k: v for k, v in conversion_differences.items() if '''hidden''' not in k}
         lowerCamelCase_  :    List[str]   	=			{k: v for k, v in conversion_differences.items() if '''hidden''' in k}
         if len(_snake_case	) == 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\''	)
         lowerCamelCase_  :    Any   	=			max(output_differences.values()	) if output_differences else 0.0
         lowerCamelCase_  :    Union[str, 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]	)	)
  
         lowerCamelCase_  :    Dict   	=			'''Update TF weights''' if self._new_weights else '''Add TF weights'''
         if self._push:
           repo.git_add(auto_lfs_track=_snake_case	)
           repo.git_commit(_snake_case	)
           repo.git_push(blocking=_snake_case	)  # 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...'	)
           lowerCamelCase_  :    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(_snake_case	):
             lowerCamelCase_  :    int   	=			[
                 CommitOperationAdd(path_in_repo=_snake_case					,			path_or_fileobj=_snake_case	)
             ]
             for shard_path in tf.io.gfile.glob(self._local_dir + '/tf_model-*.h5'	):
               operations += [
                   CommitOperationAdd(path_in_repo=os.path.basename(_snake_case	)					,			path_or_fileobj=_snake_case	)
               ]
           else:
             lowerCamelCase_  :    str   	=			[CommitOperationAdd(path_in_repo=_snake_case					,			path_or_fileobj=_snake_case	)]
  
           lowerCamelCase_  :    List[Any]   	=			create_commit(
               repo_id=self._model_name					,			operations=_snake_case					,			commit_message=_snake_case					,			commit_description=_snake_case					,			repo_type='model'					,			create_pr=_snake_case					,			).pr_url
           self._logger.warning(F"PR open in {hub_pr_url}"	)
  
style_context: 
  
  
  
  
  
  import os
  import unittest
  
  import pytest
  from parameterized import parameterized
  
  from transformers.testing_utils import require_torch, require_torch_gpu, slow, torch_device
  
  
  # skipping in unittest tests
  
  _lowercase		:     Optional[Any] 					=[(1,)]
  
  
  def 		_SCREAMING_SNAKE_CASE    (   ):
  		lowerCamelCase_  :    Union[str, Any]   	=			bool(os.getenv('RUN_SLOW'       ,0	)	)
  		if run_slow:
  				assert True
  		else:
  				assert False, "should have been skipped"
  
  
  def 		_SCREAMING_SNAKE_CASE    (   ):
  		lowerCamelCase_  :    List[str]   	=			bool(os.getenv('RUN_SLOW'       ,0	)	)
  		if run_slow and torch_device == "cuda":
  				assert True
  		else:
  				assert False, "should have been skipped"
  
  
  
  
  
  
  @require_torch
  class 	UpperCamelCase_     (     unittest.TestCase       ):
  
  
  
  					@slow
  					@require_torch_gpu
  					def      __a    (		self    :    Tuple	):
  							check_slow_torch_cuda()
  
  
  
  					@require_torch_gpu
  					@slow
  					def      __a    (		self    :    Optional[int]	):
  							check_slow_torch_cuda()
  
  
  
  
  
  					@parameterized.expand(lowerCamelCase	)
  					@slow
  					def      __a    (		self    :    List[str]					,			lowerCamelCase    :    Optional[Any]=None	):
  							check_slow()
  
  
  @slow
  @require_torch_gpu
  def 		_SCREAMING_SNAKE_CASE    (   ):
  		check_slow_torch_cuda()
  
  
  @require_torch_gpu
  @slow
  def 		_SCREAMING_SNAKE_CASE    (   ):
  		check_slow_torch_cuda()
  
  
  @slow
  @pytest.mark.parametrize('param'       ,[1]	)
  def 		_SCREAMING_SNAKE_CASE    (   lowerCAmelCase__	):
  		check_slow()
  
  
  
  
  
  @pytest.mark.parametrize('param'       ,[1]	)
  @slow
  def 		_SCREAMING_SNAKE_CASE    (   lowerCAmelCase__	):
  		check_slow()
  