





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



'''simple docstring'''

import inspect
import json
import os
import tempfile
import unittest

from transformers.models.speech_to_text_a import SpeechaTextaTokenizer
from transformers.models.speech_to_text_a.tokenization_speech_to_text_a import VOCAB_FILES_NAMES

from ...test_tokenization_common import TokenizerTesterMixin


class    a__					(			__A				,    unittest.TestCase   ):







				"""simple docstring"""



				__UpperCamelCase     :		Any						=						SpeechaTextaTokenizer
				__UpperCamelCase     :		str						=						False




				def        _snake_case	(self		):
										super().setUp()

										__lowerCAmelCase											=					'''<s> <pad> </s> <unk> here@@ a couple of@@ words for the he@@ re@@ vocab'''.split(''' '''		)
										__lowerCAmelCase											=					['''he re</w> 123''', '''here a 1456''']
										__lowerCAmelCase											=					dict(zip(__lowercase							,      range(len(__lowercase		)		)		)		)

										__lowerCAmelCase											=					{'''pad_token''': '''<pad>''', '''unk_token''': '''<unk>''', '''bos_token''': '''<s>''', '''eos_token''': '''</s>'''}

										__lowerCAmelCase											=					tempfile.mkdtemp()
										__lowerCAmelCase											=					os.path.join(self.tmpdirname							,      VOCAB_FILES_NAMES['''vocab_file''']		)
										__lowerCAmelCase											=					os.path.join(self.tmpdirname							,      VOCAB_FILES_NAMES['''merges_file''']		)

										with open(self.vocab_file							,      '''w'''							,      encoding='''utf-8'''		) as fp:
																fp.write(json.dumps(__lowercase		) + '''\n'''		)

										with open(self.merges_file							,      '''w'''		) as fp:
																fp.write('''\n'''.join(__lowercase		)		)




				def        _snake_case	(self		):
										__lowerCAmelCase											=					list(self.get_tokenizer().get_vocab().keys()		)

										self.assertEqual(vocab_keys[0]							,      '''<s>'''		)
										self.assertEqual(vocab_keys[1]							,      '''<pad>'''		)
										self.assertEqual(vocab_keys[-1]							,      '''vocab'''		)
										self.assertEqual(len(__lowercase		)							,      14		)




				def        _snake_case	(self		):
										self.assertEqual(self.get_tokenizer().vocab_size							,      14		)




				def        _snake_case	(self		):
										__lowerCAmelCase											=					SpeechaTextaTokenizer.from_pretrained(self.tmpdirname		)

										# make sure @@ is correctly concatenated
										__lowerCAmelCase											=					[4, 6, 8, 7, 10]  # ["here@@", "couple", "words", "of@@", "the"]
										__lowerCAmelCase											=					tokenizer.decode(__lowercase		)

										self.assertTrue(output_string == '''herecouple words ofthe'''		)




				def        _snake_case	(self		):
										__lowerCAmelCase											=					SpeechaTextaTokenizer.from_pretrained(self.tmpdirname		)

										with tempfile.TemporaryDirectory() as tmp_dirname:
																tokenizer.save_pretrained(__lowercase		)
																os.remove(os.path.join(__lowercase							,      '''merges.txt'''		)		)

																# load tokenizer without merges file should not throw an error
																__lowerCAmelCase											=					SpeechaTextaTokenizer.from_pretrained(__lowercase		)

										with tempfile.TemporaryDirectory() as tmp_dirname:
																# save tokenizer and load again
																tokenizer.save_pretrained(__lowercase		)
																__lowerCAmelCase											=					SpeechaTextaTokenizer.from_pretrained(__lowercase		)

										self.assertIsNotNone(__lowercase		)




				def        _snake_case	(self		):
										if not self.test_slow_tokenizer:
																return

										__lowerCAmelCase											=					inspect.signature(self.tokenizer_class.__init__		)
										__lowerCAmelCase											=					self.get_tokenizer()

										for parameter_name, parameter in signature.parameters.items():
																if parameter.default != inspect.Parameter.empty and parameter_name != "merges_file":
																						self.assertIn(__lowercase							,      tokenizer.init_kwargs		)

