

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


__UpperCAmelCase	=			5E-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors
def 					lowerCAmelCase  (   __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    __lowercase     (      __lowerCamelCase   ):

  @staticmethod
  def 				__lowercase    (					A						:     ArgumentParser      ):






   '''simple docstring'''
   UpperCAmelCase__   :					int   				=   parser.add_parser(
       """pt-to-tf"""   ,help=(
           """CLI tool to run convert a transformers model from a PyTorch checkpoint to a TensorFlow checkpoint."""
           """ Can also be used to validate existing weights without opening PRs, with --no-pr."""
       )   ,)
   train_parser.add_argument(
       """--model-name"""   ,type=A   ,required=A   ,help="""The model name, including owner/organization, as seen on the hub."""   ,)
   train_parser.add_argument(
       """--local-dir"""   ,type=A   ,default=""""""   ,help="""Optional local directory of the model repository. Defaults to /tmp/{model_name}"""   ,)
   train_parser.add_argument(
       """--max-error"""   ,type=A   ,default=A   ,help=(
           f"Maximum error tolerance. Defaults to {MAX_ERROR}. This flag should be avoided, use at your own risk."
       )   ,)
   train_parser.add_argument(
       """--new-weights"""   ,action="""store_true"""   ,help="""Optional flag to create new TensorFlow weights, even if they already exist."""   ,)
   train_parser.add_argument(
       """--no-pr"""   ,action="""store_true"""   ,help="""Optional flag to NOT open a PR with converted weights."""      )
   train_parser.add_argument(
       """--push"""   ,action="""store_true"""   ,help="""Optional flag to push the weights directly to `main` (requires permissions)"""   ,)
   train_parser.add_argument(
       """--extra-commit-description"""   ,type=A   ,default=""""""   ,help="""Optional additional commit description to use when opening a PR (e.g. to tag the owner)."""   ,)
   train_parser.add_argument(
       """--override-model-class"""   ,type=A   ,default=A   ,help="""If you think you know better than the auto-detector, you can specify the model class here. """
       """Can be either an AutoModel class or a specific model class like BertForSequenceClassification."""   ,)
   train_parser.set_defaults(func=A      )

  @staticmethod
  def 				__lowercase    (					A						:     Union[str, Any]   ,A						:     str      ):






   '''simple docstring'''
   UpperCAmelCase__   :					Dict   				=   set(pt_outputs.keys()      )
   UpperCAmelCase__   :					int   				=   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(A						:     List[Any]   ,A						:     str   ,A						:     Optional[Any]   ,A						:     List[str]=""      ):
    # If the current attribute is a tensor, it is a leaf and we make the comparison. Otherwise, we will dig in
    # recursivelly, keeping the name of the attribute.
    if isinstance(A   ,torch.Tensor      ):
     UpperCAmelCase__   :					Optional[Any]   				=   np.max(np.abs(pt_out.numpy() - tf_out.numpy()      )      )
     UpperCAmelCase__   :					Union[str, Any]   				=   tensor_difference
    else:
     UpperCAmelCase__   :					int   				=   attr_name
     for i, pt_item in enumerate(A      ):
      # If it is a named attribute, we keep the name. Otherwise, just its index.
      if isinstance(A   ,A      ):
       UpperCAmelCase__   :					Optional[int]   				=   root_name + pt_item
       UpperCAmelCase__   :					Optional[Any]   				=   tf_out[pt_item]
       UpperCAmelCase__   :					Tuple   				=   pt_out[pt_item]
      else:
       UpperCAmelCase__   :					Tuple   				=   root_name + f"[{i}]"
       UpperCAmelCase__   :					Optional[Any]   				=   tf_out[i]
      UpperCAmelCase__   :					int   				=   _find_pt_tf_differences(A   ,A   ,A   ,A      )

    return differences

   return _find_pt_tf_differences(A   ,A   ,{}      )

  def __init__(					self						:     Optional[int]   ,A						:     str   ,A						:     str   ,A						:     float   ,A						:     bool   ,A						:     bool   ,A						:     bool   ,A						:     str   ,A						:     str   ,*A						:     Tuple   ,):






   '''simple docstring'''
   UpperCAmelCase__   :					List[Any]   				=   logging.get_logger("""transformers-cli/pt_to_tf"""      )
   UpperCAmelCase__   :					Optional[int]   				=   model_name
   UpperCAmelCase__   :					str   				=   local_dir if local_dir else os.path.join("""/tmp"""   ,A      )
   UpperCAmelCase__   :					Optional[int]   				=   max_error
   UpperCAmelCase__   :					Tuple   				=   new_weights
   UpperCAmelCase__   :					int   				=   no_pr
   UpperCAmelCase__   :					Dict   				=   push
   UpperCAmelCase__   :					Dict   				=   extra_commit_description
   UpperCAmelCase__   :					List[Any]   				=   override_model_class

  def 				__lowercase    (					self						:     str   ,A						:     Tuple   ,A						:     Dict   ,A						:     Dict      ):






   '''simple docstring'''

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

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

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

   # Extra input requirements, in addition to the input modality
   if (
       config.is_encoder_decoder
       or (hasattr(A   ,"""encoder"""      ) and hasattr(A   ,"""decoder"""      ))
       or "decoder_input_ids" in tf_dummy_inputs
   ):
    UpperCAmelCase__   :					List[Any]   				=   np.asarray([[1], [1]]   ,dtype=A      ) * (pt_model.config.decoder_start_token_id or 0)
    pt_input.update({"""decoder_input_ids""": torch.tensor(A      )}      )
    tf_input.update({"""decoder_input_ids""": tf.convert_to_tensor(A      )}      )

   return pt_input, tf_input




  def 				__lowercase    (					self						:     Tuple      ):






   '''simple docstring'''
   # 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
   UpperCAmelCase__   :					List[Any]   				=   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
   UpperCAmelCase__   :					str   				=   AutoConfig.from_pretrained(self._local_dir      )
   UpperCAmelCase__   :					List[str]   				=   config.architectures
   if self._override_model_class is not None:
    if self._override_model_class.startswith("""TF"""      ):
     UpperCAmelCase__   :					Tuple   				=   [self._override_model_class[2:]]
    else:
     UpperCAmelCase__   :					List[str]   				=   [self._override_model_class]
    try:
     UpperCAmelCase__   :					Tuple   				=   getattr(import_module("""transformers"""      )   ,architectures[0]      )
    except AttributeError:
     raise ValueError(f"Model class {self._override_model_class} not found in transformers."      )
    try:
     UpperCAmelCase__   :					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
    UpperCAmelCase__   :					Union[str, Any]   				=   getattr(import_module("""transformers"""      )   ,"""AutoModel"""      )
    UpperCAmelCase__   :					Dict   				=   getattr(import_module("""transformers"""      )   ,"""TFAutoModel"""      )
    self._logger.warning("""No detected architecture, using AutoModel/TFAutoModel"""      )
   else:  # Architecture defined -- use it
    if len(A      ) > 1:
     raise ValueError(f"More than one architecture was found, aborting. (architectures = {architectures})"      )
    self._logger.warning(f"Detected architecture: {architectures[0]}"      )
    UpperCAmelCase__   :					Optional[int]   				=   getattr(import_module("""transformers"""      )   ,architectures[0]      )
    try:
     UpperCAmelCase__   :					Any   				=   getattr(import_module("""transformers"""      )   ,"""TF""" + architectures[0]      )
    except AttributeError:
     raise AttributeError(f"The TensorFlow equivalent of {architectures[0]} doesn't exist in transformers."      )

        # Check the TF dummy inputs to see what keys we need in the forward pass
   UpperCAmelCase__   :					List[Any]   				=   tf_class.from_config(A      )
   UpperCAmelCase__   :					Optional[int]   				=   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
   UpperCAmelCase__   :					Dict   				=   pt_class.from_pretrained(self._local_dir      )
   pt_model.eval()

   UpperCAmelCase__   :					Union[str, Any]   				=   self.get_inputs(A   ,A   ,A      )

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

   UpperCAmelCase__   :					Union[str, Any]   				=   tf_class.from_pretrained(self._local_dir   ,from_pt=A      )
   UpperCAmelCase__   :					Dict   				=   tf_from_pt_model(**A   ,output_hidden_states=A   ,training=A      )

   # Confirms that cross loading PT weights into TF worked.
   UpperCAmelCase__   :					Tuple   				=   self.find_pt_tf_differences(A   ,A      )
   UpperCAmelCase__   :					List[Any]   				=   {k: v for k, v in crossload_differences.items() if """hidden""" not in k}
   UpperCAmelCase__   :					Dict   				=   {k: v for k, v in crossload_differences.items() if """hidden""" in k}
   if len(A      ) == 0 and architectures is not None:
    raise ValueError(
        f"Something went wrong -- the config file has architectures ({architectures}), but no model head"
        """ output was found. All outputs start with 'hidden'"""      )
   UpperCAmelCase__   :					str   				=   max(output_differences.values()      ) if output_differences else 0.0
   UpperCAmelCase__   :					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
   UpperCAmelCase__   :					int   				=   os.path.join(self._local_dir   ,A      )
   UpperCAmelCase__   :					str   				=   os.path.join(self._local_dir   ,A      )
   if (not os.path.exists(A      ) and not os.path.exists(A      )) or self._new_weights:
    tf_from_pt_model.save_pretrained(self._local_dir      )
   del tf_from_pt_model  # will no longer be used, and may have a large memory footprint

   UpperCAmelCase__   :					Dict   				=   tf_class.from_pretrained(self._local_dir      )
   UpperCAmelCase__   :					str   				=   tf_model(**A   ,output_hidden_states=A      )

   UpperCAmelCase__   :					int   				=   self.find_pt_tf_differences(A   ,A      )
   UpperCAmelCase__   :					int   				=   {k: v for k, v in conversion_differences.items() if """hidden""" not in k}
   UpperCAmelCase__   :					str   				=   {k: v for k, v in conversion_differences.items() if """hidden""" in k}
   if len(A      ) == 0 and architectures is not None:
    raise ValueError(
        f"Something went wrong -- the config file has architectures ({architectures}), but no model head"
        """ output was found. All outputs start with 'hidden'"""      )
   UpperCAmelCase__   :					Dict   				=   max(output_differences.values()      ) if output_differences else 0.0
   UpperCAmelCase__   :					List[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]      )      )

   UpperCAmelCase__   :					Optional[int]   				=   """Update TF weights""" if self._new_weights else """Add TF weights"""
   if self._push:
    repo.git_add(auto_lfs_track=A      )
    repo.git_commit(A      )
    repo.git_push(blocking=A      )  # this prints a progress bar with the upload
    self._logger.warning(f"TF weights pushed into {self._model_name}"      )
   elif not self._no_pr:
    self._logger.warning("""Uploading the weights into a new PR..."""      )
    UpperCAmelCase__   :					List[Any]   				=   (
        """Model converted by the [`transformers`' `pt_to_tf`"""
        """ CLI](https://github.com/huggingface/transformers/blob/main/src/transformers/commands/pt_to_tf.py). """
        """All converted model outputs and hidden layers were validated against its PyTorch counterpart.\n\n"""
        f"Maximum crossload output difference={max_crossload_output_diff:.3e}; "
        f"Maximum crossload hidden layer difference={max_crossload_hidden_diff:.3e};\n"
        f"Maximum conversion output difference={max_conversion_output_diff:.3e}; "
        f"Maximum conversion hidden layer difference={max_conversion_hidden_diff:.3e};\n"
    )
    if self._max_error > MAX_ERROR:
     commit_descrition += (
         f"\n\nCAUTION: The maximum admissible error was manually increased to {self._max_error}!"
     )
    if self._extra_commit_description:
     commit_descrition += "\n\n" + self._extra_commit_description

    # sharded model -> adds all related files (index and .h5 shards)
    if os.path.exists(A      ):
     UpperCAmelCase__   :					Any   				=   [
         CommitOperationAdd(path_in_repo=A   ,path_or_fileobj=A      )
     ]
     for shard_path in tf.io.gfile.glob(self._local_dir + """/tf_model-*.h5"""      ):
      operations += [
          CommitOperationAdd(path_in_repo=os.path.basename(A      )   ,path_or_fileobj=A      )
      ]
    else:
     UpperCAmelCase__   :					int   				=   [CommitOperationAdd(path_in_repo=A   ,path_or_fileobj=A      )]

    UpperCAmelCase__   :					List[str]   				=   create_commit(
        repo_id=self._model_name   ,operations=A   ,commit_message=A   ,commit_description=A   ,repo_type="""model"""   ,create_pr=A   ,).pr_url
    self._logger.warning(f"PR open in {hub_pr_url}"      )








"""simple docstring"""




import hashlib
import importlib.util
import json
import os
import pathlib
from types import ModuleType

import pytest
import requests

__UpperCAmelCase	=			pathlib.Path.cwd().joinpath('project_euler')
__UpperCAmelCase	=			pathlib.Path.cwd().joinpath(
    'scripts', 'project_euler_answers.json'
)

with open(PROJECT_EULER_ANSWERS_PATH) as file_handle:
      __UpperCAmelCase	=			json.load(file_handle)
def 					lowerCAmelCase  (   __UpperCamelCase ):
 '''simple docstring'''





 UpperCAmelCase__   :					Dict   				=   importlib.util.spec_from_file_location(file_path.name							,      str(__UpperCamelCase ) )
 UpperCAmelCase__   :					Optional[Any]   				=   importlib.util.module_from_spec(__UpperCamelCase )  # type: ignore
 spec.loader.exec_module(__UpperCamelCase )  # type: ignore
 return module
def 					lowerCAmelCase  (   ):
 '''simple docstring'''





 UpperCAmelCase__   :					Dict   				=   []
 for problem_dir_path in PROJECT_EULER_DIR_PATH.iterdir():
  if problem_dir_path.is_file() or problem_dir_path.name.startswith("""_""" ):
   continue
  for file_path in problem_dir_path.iterdir():
   if file_path.suffix != ".py" or file_path.name.startswith(("""_""", """test""") ):
    continue
   solution_file_paths.append(__UpperCamelCase )
 return solution_file_paths
def 					lowerCAmelCase  (   ):
 '''simple docstring'''





 with open(os.environ["""GITHUB_EVENT_PATH"""] ) as file:
  UpperCAmelCase__   :					List[Any]   				=   json.load(__UpperCamelCase )
 return event["pull_request"]["url"] + "/files"
def 					lowerCAmelCase  (   ):
 '''simple docstring'''





 UpperCAmelCase__   :					Tuple   				=   []
 UpperCAmelCase__   :					List[str]   				=   {
     """Accept""": """application/vnd.github.v3+json""",
     """Authorization""": """token """ + os.environ["""GITHUB_TOKEN"""],
 }
 UpperCAmelCase__   :					Tuple   				=   requests.get(get_files_url()							,      headers=__UpperCamelCase ).json()
 for file in files:
  UpperCAmelCase__   :					List[str]   				=   pathlib.Path.cwd().joinpath(file["""filename"""] )
  if (
      filepath.suffix != ".py"
      or filepath.name.startswith(("""_""", """test""") )
      or not filepath.name.startswith("""sol""" )
  ):
   continue
  solution_file_paths.append(__UpperCamelCase )
 return solution_file_paths
def 					lowerCAmelCase  (   ):
 '''simple docstring'''





 if os.environ.get("""CI""" ) and os.environ.get("""GITHUB_EVENT_NAME""" ) == "pull_request":
  # Return only if there are any, otherwise default to all solutions
  if filepaths := added_solution_file_path():
   return filepaths
 return all_solution_file_paths()




@pytest.mark.parametrize(
    """solution_path"""							,      collect_solution_file_paths()							,      ids=lambda __UpperCamelCase : F"{path.parent.name}/{path.name}"							,      )
def 					lowerCAmelCase  (   __UpperCamelCase ):
 '''simple docstring'''





 UpperCAmelCase__   :					str   				=   solution_path.parent.name[8:].zfill(3 )
 UpperCAmelCase__   :					str   				=   PROBLEM_ANSWERS[problem_number]
 UpperCAmelCase__   :					int   				=   convert_path_to_module(__UpperCamelCase )
 UpperCAmelCase__   :					str   				=   str(solution_module.solution() )  # type: ignore
 UpperCAmelCase__   :					Optional[Any]   				=   hashlib.shaaaa(answer.encode() ).hexdigest()
 assert (
     answer == expected
 ), F"Expected solution to {problem_number} to have hash {expected}, got {answer}"





