





"""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[int]         =   5E-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors





def 		UpperCAmelCase__						(	_UpperCAmelCase   ):






  """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    (  UpperCAmelCase__						):



      '''simple docstring'''

      @staticmethod
      def    lowerCamelCase_     (	snake_case_    ):

        """simple docstring"""

        A_    :			List[str]									=    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    lowerCamelCase_     (	snake_case_   , snake_case_    ):

        """simple docstring"""

        A_    :			Optional[int]									=    set(pt_outputs.keys()    )
        A_    :			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(snake_case_   , snake_case_   , snake_case_   , snake_case_=""    ):
          # 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_    :			Union[str, Any]									=    np.max(np.abs(pt_out.numpy() - tf_out.numpy()    )    )
            A_    :			Any									=    tensor_difference
          else:
            A_    :			List[str]									=    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_    :			Any									=    root_name + pt_item
                A_    :			Union[str, Any]									=    tf_out[pt_item]
                A_    :			str									=    pt_out[pt_item]
              else:
                A_    :			Union[str, Any]									=    root_name + F"""[{i}]"""
                A_    :			int									=    tf_out[i]
              A_    :			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   , snake_case_   , snake_case_   , snake_case_   , snake_case_   , snake_case_   , snake_case_   , snake_case_   , snake_case_   , *snake_case_   , ):

        """simple docstring"""

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

      def    lowerCamelCase_     (	self   , snake_case_   , snake_case_   , snake_case_    ):

        """simple docstring"""

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

        A_    :			Dict									=    type(pt_model.config    )
        if model_config_class in PROCESSOR_MAPPING:
          A_    :			int									=    AutoProcessor.from_pretrained(self._local_dir    )
          if model_config_class in TOKENIZER_MAPPING and processor.tokenizer.pad_token is None:
            A_    :			List[str]									=    processor.tokenizer.eos_token
        elif model_config_class in IMAGE_PROCESSOR_MAPPING:
          A_    :			List[str]									=    AutoImageProcessor.from_pretrained(self._local_dir    )
        elif model_config_class in FEATURE_EXTRACTOR_MAPPING:
          A_    :			List[str]									=    AutoFeatureExtractor.from_pretrained(self._local_dir    )
        elif model_config_class in TOKENIZER_MAPPING:
          A_    :			Tuple									=    AutoTokenizer.from_pretrained(self._local_dir    )
          if processor.pad_token is None:
            A_    :			Optional[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_    :			Tuple									=    {}
        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[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:
            A_    :			Any									=    feature_extractor_signature['padding'].default
            if default_strategy is not False and default_strategy is not None:
              A_    :			Optional[int]									=    default_strategy
            else:
              A_    :			int									=    True
          else:
            A_    :			Union[str, 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}    )
        A_    :			Dict									=    processor(**snake_case_   , return_tensors='pt'    )
        A_    :			List[Any]									=    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_    :			int									=    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    lowerCamelCase_     (	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_    :			List[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
        A_    :			Tuple									=    AutoConfig.from_pretrained(self._local_dir    )
        A_    :			List[Any]									=    config.architectures
        if self._override_model_class is not None:
          if self._override_model_class.startswith('TF'    ):
            A_    :			List[Any]									=    [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_    :			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
          A_    :			Dict									=    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(snake_case_    ) > 1:
            raise ValueError(F"""More than one architecture was found, aborting. (architectures = {architectures})"""    )
          self._logger.warning(F"""Detected architecture: {architectures[0]}"""    )
          A_    :			Dict									=    getattr(import_module('transformers'    )   , architectures[0]    )
          try:
            A_    :			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
        A_    :			Union[str, Any]									=    tf_class.from_config(snake_case_    )
        A_    :			Tuple									=    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_    :			List[Any]									=    pt_class.from_pretrained(self._local_dir    )
        pt_model.eval()

        A_		,			A_    :			Optional[int]									=    self.get_inputs(snake_case_   , snake_case_   , snake_case_    )

        with torch.no_grad():
          A_    :			Optional[int]									=    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_    :			Optional[Any]									=    tf_class.from_pretrained(self._local_dir   , from_pt=snake_case_    )
        A_    :			Dict									=    tf_from_pt_model(**snake_case_   , output_hidden_states=snake_case_   , training=snake_case_    )

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

        A_    :			Union[str, Any]									=    self.find_pt_tf_differences(snake_case_   , snake_case_    )
        A_    :			List[Any]									=    {k: v for k, v in conversion_differences.items() if 'hidden' not in k}
        A_    :			Any									=    {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_    :			Optional[int]									=    max(output_differences.values()    ) if output_differences else 0.0
        A_    :			Optional[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]    )    )

        A_    :			Any									=    '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_    :			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(snake_case_    ):
            A_    :			List[Any]									=    [
                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_    :			int									=    [CommitOperationAdd(path_in_repo=snake_case_   , path_or_fileobj=snake_case_    )]

          A_    :			Union[str, 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}"""    )



from __future__ import annotations

__SCREAMING_SNAKE_CASE	:							Union[str, Any]           =  [2, 1, 4, 5, 6, 0, 8, 9, 1, 2, 0, 6, 4, 2, 0, 6, 5, 3, 2, 7]


class        __lowerCamelCase      :


				"""simple docstring"""







				def __init__(				self     :    Dict	,							lowerCamelCase_     :    int					):
								_lowerCAmelCase  						=-1
								_lowerCAmelCase  						=-1
								_lowerCAmelCase  						=[-1] * length
								_lowerCAmelCase  						=None
								_lowerCAmelCase  						=None






				def __repr__(				self     :    Tuple					):
								return F"Node(min_value={self.minn} max_value={self.maxx})"

def 		snake_case_   (				lowercase__     :    list[int]     ):






				'''simple docstring'''






				_lowerCAmelCase  						=Node(len(lowercase__     )     )
				_lowerCAmelCase						,							_lowerCAmelCase  						=min(lowercase__     ), max(lowercase__     )
				# Leaf node case where the node contains only one unique value
				if root.minn == root.maxx:
								return root
				_lowerCAmelCase  						=(root.minn + root.maxx) // 2

				_lowerCAmelCase  						=[]
				_lowerCAmelCase  						=[]

				for index, num in enumerate(lowercase__     ):
								if num <= pivot:
												left_arr.append(lowercase__     )
								else:
												right_arr.append(lowercase__     )
								_lowerCAmelCase  						=len(lowercase__     )
				_lowerCAmelCase  						=build_tree(lowercase__     )
				_lowerCAmelCase  						=build_tree(lowercase__     )
				return root

def 		snake_case_   (				lowercase__     :    Node | None       ,     lowercase__     :    int       ,     lowercase__     :    int     ):






				'''simple docstring'''






				if index < 0 or node is None:
								return 0
				# Leaf node cases
				if node.minn == node.maxx:
								return index + 1 if node.minn == num else 0
				_lowerCAmelCase  						=(node.minn + node.maxx) // 2
				if num <= pivot:
								# go the left subtree and map index to the left subtree
								return rank_till_index(node.left       ,     lowercase__       ,     node.map_left[index] - 1     )
				else:
								# go to the right subtree and map index to the right subtree
								return rank_till_index(node.right       ,     lowercase__       ,     index - node.map_left[index]     )

def 		snake_case_   (				lowercase__     :    Node | None       ,     lowercase__     :    int       ,     lowercase__     :    int       ,     lowercase__     :    int     ):






				'''simple docstring'''






				if start > end:
								return 0
				_lowerCAmelCase  						=rank_till_index(lowercase__       ,     lowercase__       ,     lowercase__     )
				_lowerCAmelCase  						=rank_till_index(lowercase__       ,     lowercase__       ,     start - 1     )
				return rank_till_end - rank_before_start

def 		snake_case_   (				lowercase__     :    Node | None       ,     lowercase__     :    int       ,     lowercase__     :    int       ,     lowercase__     :    int     ):






				'''simple docstring'''






				if index > (end - start) or start > end or node is None:
								return -1
				# Leaf node case
				if node.minn == node.maxx:
								return node.minn
				# Number of elements in the left subtree in interval [start, end]
				_lowerCAmelCase  						=node.map_left[end] - (
				    node.map_left[start - 1] if start else 0
				)
				if num_elements_in_left_tree > index:
								return quantile(
								    node.left       ,     lowercase__       ,     (node.map_left[start - 1] if start else 0)       ,     node.map_left[end] - 1       ,     )
				else:
								return quantile(
								    node.right       ,     index - num_elements_in_left_tree       ,     start - (node.map_left[start - 1] if start else 0)       ,     end - node.map_left[end]       ,     )

def 		snake_case_   (				lowercase__     :    Node | None       ,     lowercase__     :    int       ,     lowercase__     :    int       ,     lowercase__     :    int       ,     lowercase__     :    int     ):






				'''simple docstring'''






				if (
				    start > end
				    or node is None
				    or start_num > end_num
				    or node.minn > end_num
				    or node.maxx < start_num
				):
								return 0
				if start_num <= node.minn and node.maxx <= end_num:
								return end - start + 1
				_lowerCAmelCase  						=range_counting(
				    node.left       ,     (node.map_left[start - 1] if start else 0)       ,     node.map_left[end] - 1       ,     lowercase__       ,     lowercase__       ,     )
				_lowerCAmelCase  						=range_counting(
				    node.right       ,     start - (node.map_left[start - 1] if start else 0)       ,     end - node.map_left[end]       ,     lowercase__       ,     lowercase__       ,     )
				return left + right


if __name__ == "__main__":
		import doctest

		doctest.testmod()

