



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


__snake_case	:				int  					=			5E-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors



def   a_	(			__a      ):
     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_				):





     """simple docstring"""






     @staticmethod
     def 						A__  (					_lowerCamelCase :							str ):
          A__				=   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 :							Tuple	,      _lowerCamelCase :							Dict ):
          A__				=   set(pt_outputs.keys() )
          A__				=   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 :							int	,      _lowerCamelCase :							List[str]	,      _lowerCamelCase :							int	,      _lowerCamelCase :							List[Any]="" ):
               # 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 ):
                    A__				=   np.max(np.abs(pt_out.numpy() - tf_out.numpy() ) )
                    A__				=   tensor_difference
               else:
                    A__				=   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 ):
                              A__				=   root_name + pt_item
                              A__				=   tf_out[pt_item]
                              A__				=   pt_out[pt_item]
                         else:
                              A__				=   root_name + F'''[{i}]'''
                              A__				=   tf_out[i]
                         A__				=   _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 :							str	,      _lowerCamelCase :							List[Any]	,      _lowerCamelCase :							str	,      _lowerCamelCase :							int	,      _lowerCamelCase :							Tuple	,      _lowerCamelCase :							List[str]	,      _lowerCamelCase :							Optional[int]	,      _lowerCamelCase :							Optional[Any]	,      _lowerCamelCase :							Optional[Any]	,      *_lowerCamelCase :							Tuple	,      ):
          A__				=   logging.get_logger('''transformers-cli/pt_to_tf''' )
          A__				=   model_name
          A__				=   local_dir if local_dir else os.path.join('''/tmp'''	,      _snake_case )
          A__				=   max_error
          A__				=   new_weights
          A__				=   no_pr
          A__				=   push
          A__				=   extra_commit_description
          A__				=   override_model_class





     def 						A__  (					self :							int	,      _lowerCamelCase :							Dict	,      _lowerCamelCase :							Union[str, Any]	,      _lowerCamelCase :							Any ):

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

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

          A__				=   set(inspect.signature(pt_model.forward ).parameters.keys() )
          A__				=   {}
          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__				=   load_dataset('''cifar10'''	,      '''plain_text'''	,      split='''test''' )[:2]['''img''']
               processor_inputs.update({'''images''': sample_images} )
          if "input_features" in model_forward_signature:
               A__				=   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__				=   feature_extractor_signature['''padding'''].default
                    if default_strategy is not False and default_strategy is not None:
                         A__				=   default_strategy
                    else:
                         A__				=   True
               else:
                    A__				=   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__				=   processor(**_snake_case	,      return_tensors='''pt''' )
          A__				=   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
          ):
               A__				=   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 :							int ):
          # 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
          A__				=   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__				=   AutoConfig.from_pretrained(self._local_dir )
          A__				=   config.architectures
          if self._override_model_class is not None:
               if self._override_model_class.startswith('''TF''' ):
                    A__				=   [self._override_model_class[2:]]
               else:
                    A__				=   [self._override_model_class]
               try:
                    A__				=   getattr(import_module('''transformers''' )	,      architectures[0] )
               except AttributeError:
                    raise ValueError(F'''Model class {self._override_model_class} not found in transformers.''' )
               try:
                    A__				=   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__				=   getattr(import_module('''transformers''' )	,      '''AutoModel''' )
               A__				=   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]}''' )
               A__				=   getattr(import_module('''transformers''' )	,      architectures[0] )
               try:
                    A__				=   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__				=   tf_class.from_config(_snake_case )
          A__				=   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__				=   pt_class.from_pretrained(self._local_dir )
          pt_model.eval()

          A__				=   self.get_inputs(_snake_case	,      _snake_case	,      _snake_case )

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

          A__				=   tf_class.from_pretrained(self._local_dir	,      from_pt=_snake_case )
          A__				=   tf_from_pt_model(**_snake_case	,      output_hidden_states=_snake_case	,      training=_snake_case )

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

          A__				=   tf_class.from_pretrained(self._local_dir )
          A__				=   tf_model(**_snake_case	,      output_hidden_states=_snake_case )

          A__				=   self.find_pt_tf_differences(_snake_case	,      _snake_case )
          A__				=   {k: v for k, v in conversion_differences.items() if '''hidden''' not in k}
          A__				=   {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\'''' )
          A__				=   max(output_differences.values() ) if output_differences else 0.0
          A__				=   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__				=   '''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...''' )
               A__				=   (
                   '''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 ):
                    A__				=   [
                        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:
                    A__				=   [CommitOperationAdd(path_in_repo=_snake_case	,      path_or_fileobj=_snake_case )]

               A__				=   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}''' )








"""simple docstring"""







def   a_	(			__a      ):
     if not isinstance(__a		,							__a      ):
          raise TypeError('''recursive_lucas_number accepts only integer arguments.'''      )
     if n_th_number == 0:
          return 2
     if n_th_number == 1:
          return 1

     return recursive_lucas_number(n_th_number - 1      ) + recursive_lucas_number(
         n_th_number - 2      )



def   a_	(			__a      ):
     if not isinstance(__a		,							__a      ):
          raise TypeError('''dynamic_lucas_number accepts only integer arguments.'''      )
     A__	,     A__				=   2, 1
     for _ in range(__a      ):
          A__	,     A__				=   b, a + b
     return a


if __name__ == "__main__":
 from doctest import testmod

 testmod()
 __snake_case	:				Dict  					=			int(input('Enter the number of terms in lucas series:\n').strip())
 print('Using recursive function to calculate lucas series:')
 print(' '.join(str(recursive_lucas_number(i)) for i in range(n)))
 print('\nUsing dynamic function to calculate lucas series:')
 print(' '.join(str(dynamic_lucas_number(i)) for i in range(n)))



