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


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





def        snake_case							(		UpperCAmelCase	)->			str:







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






 @staticmethod
 def 	lowercase_						(			_A  :ArgumentParser			)				->			List[str]:







       '''simple docstring'''



       __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=_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  :Dict	,				_A  :Optional[int]			)				->			List[str]:







       '''simple docstring'''



       __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(_A  :Any	,				_A  :List[str]	,				_A  :Optional[int]	,				_A  :Optional[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(_A	,				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(_A			):
                         # If it is a named attribute, we keep the name. Otherwise, just its index.
                         if isinstance(_A	,				_A			):
                               __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(_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  :int	,				)				->			List[Any]:







       '''simple docstring'''



       __A       	=			logging.get_logger('transformers-cli/pt_to_tf'			)
       __A       	=			model_name
       __A       	=			local_dir if local_dir else os.path.join('/tmp'	,				_A			)
       __A       	=			max_error
       __A       	=			new_weights
       __A       	=			no_pr
       __A       	=			push
       __A       	=			extra_commit_description
       __A       	=			override_model_class






 def 	lowercase_						(			self  :int	,				_A  :Optional[Any]	,				_A  :Optional[int]	,				_A  :Dict			)				->			Dict:







       '''simple docstring'''



       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(**_A	,				return_tensors='pt'			)
       __A       	=			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
       ):
             __A       	=			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  :int			)				->			Union[str, Any]:







       '''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       	=			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(_A			) > 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(_A			)
       __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	,      __A       	=			self.get_inputs(_A	,				_A	,				_A			)

       with torch.no_grad():
             __A       	=			pt_model(**_A	,				output_hidden_states=_A			)
       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=_A			)
       __A       	=			tf_from_pt_model(**_A	,				output_hidden_states=_A	,				training=_A			)

       # Confirms that cross loading PT weights into TF worked.
       __A       	=			self.find_pt_tf_differences(_A	,				_A			)
       __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(_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\''			)
       __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	,				_A			)
       __A       	=			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

       __A       	=			tf_class.from_pretrained(self._local_dir			)
       __A       	=			tf_model(**_A	,				output_hidden_states=_A			)

       __A       	=			self.find_pt_tf_differences(_A	,				_A			)
       __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(_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\''			)
       __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=_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...'			)
             __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(_A			):
                   __A       	=			[
                       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:
                   __A       	=			[CommitOperationAdd(path_in_repo=_A	,				path_or_fileobj=_A			)]

             __A       	=			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

a__		:   List[Any]            =       pathlib.Path.cwd().joinpath("project_euler")
a__		:   Union[str, Any]            =       pathlib.Path.cwd().joinpath(
    "scripts", "project_euler_answers.json"
)

with open(PROJECT_EULER_ANSWERS_PATH) as file_handle:
 a__		:   dict[str, str]            =       json.load(file_handle)





def        snake_case							(		UpperCAmelCase	)->			ModuleType:







      """simple docstring"""
      __A       	=			importlib.util.spec_from_file_location(file_path.name      ,  str(UpperCAmelCase	)	)
      __A       	=			importlib.util.module_from_spec(UpperCAmelCase	)  # type: ignore
      spec.loader.exec_module(UpperCAmelCase	)  # type: ignore
      return module





def        snake_case							(		)->			list[pathlib.Path]:







      """simple docstring"""
      __A       	=			[]
      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        snake_case							(		)->			str:







      """simple docstring"""
      with open(os.environ['GITHUB_EVENT_PATH']	) as file:
            __A       	=			json.load(UpperCAmelCase	)
      return event["pull_request"]["url"] + "/files"





def        snake_case							(		)->			list[pathlib.Path]:







      """simple docstring"""
      __A       	=			[]
      __A       	=			{
          'Accept': 'application/vnd.github.v3+json',
          'Authorization': 'token ' + os.environ['GITHUB_TOKEN'],
      }
      __A       	=			requests.get(get_files_url()      ,  headers=UpperCAmelCase	).json()
      for file in files:
            __A       	=			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        snake_case							(		)->			list[pathlib.Path]:







      """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        snake_case							(		UpperCAmelCase	)->			None:







      """simple docstring"""
      __A       	=			solution_path.parent.name[8:].zfill(3	)
      __A       	=			PROBLEM_ANSWERS[problem_number]
      __A       	=			convert_path_to_module(UpperCAmelCase	)
      __A       	=			str(solution_module.solution()	)  # type: ignore
      __A       	=			hashlib.shaaaa(answer.encode()	).hexdigest()
      assert (
          answer == expected
      ), f'Expected solution to {problem_number} to have hash {expected}, got {answer}'






