
'''simple docstring'''

import argparse
import collections
import os
import re
import tempfile

import pandas as pd
from datasets import Dataset
from huggingface_hub import hf_hub_download, upload_folder

from transformers.utils import direct_transformers_import


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/update_metadata.py
__snake_case:  Tuple								=			"src/transformers"


# This is to make sure the transformers module imported is the one in the repo.
__snake_case:  str								=			direct_transformers_import(TRANSFORMERS_PATH)


# Regexes that match TF/Flax/PT model names.
__snake_case:  Tuple								=			re.compile(R"TF(.*)(?:Model|Encoder|Decoder|ForConditionalGeneration)")
__snake_case:  Dict								=			re.compile(R"Flax(.*)(?:Model|Encoder|Decoder|ForConditionalGeneration)")
# Will match any TF or Flax model too so need to be in an else branch afterthe two previous regexes.
__snake_case:  Dict								=			re.compile(R"(.*)(?:Model|Encoder|Decoder|ForConditionalGeneration)")


# Fill this with tuples (pipeline_tag, model_mapping, auto_model)
__snake_case:  Union[str, Any]								=			[
    ("pretraining", "MODEL_FOR_PRETRAINING_MAPPING_NAMES", "AutoModelForPreTraining"),
    ("feature-extraction", "MODEL_MAPPING_NAMES", "AutoModel"),
    ("audio-classification", "MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING_NAMES", "AutoModelForAudioClassification"),
    ("text-generation", "MODEL_FOR_CAUSAL_LM_MAPPING_NAMES", "AutoModelForCausalLM"),
    ("automatic-speech-recognition", "MODEL_FOR_CTC_MAPPING_NAMES", "AutoModelForCTC"),
    ("image-classification", "MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING_NAMES", "AutoModelForImageClassification"),
    ("image-segmentation", "MODEL_FOR_IMAGE_SEGMENTATION_MAPPING_NAMES", "AutoModelForImageSegmentation"),
    ("fill-mask", "MODEL_FOR_MASKED_LM_MAPPING_NAMES", "AutoModelForMaskedLM"),
    ("object-detection", "MODEL_FOR_OBJECT_DETECTION_MAPPING_NAMES", "AutoModelForObjectDetection"),
    (
        "zero-shot-object-detection",
        "MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING_NAMES",
        "AutoModelForZeroShotObjectDetection",
    ),
    ("question-answering", "MODEL_FOR_QUESTION_ANSWERING_MAPPING_NAMES", "AutoModelForQuestionAnswering"),
    ("text2text-generation", "MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING_NAMES", "AutoModelForSeq2SeqLM"),
    ("text-classification", "MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING_NAMES", "AutoModelForSequenceClassification"),
    ("automatic-speech-recognition", "MODEL_FOR_SPEECH_SEQ_2_SEQ_MAPPING_NAMES", "AutoModelForSpeechSeq2Seq"),
    (
        "table-question-answering",
        "MODEL_FOR_TABLE_QUESTION_ANSWERING_MAPPING_NAMES",
        "AutoModelForTableQuestionAnswering",
    ),
    ("token-classification", "MODEL_FOR_TOKEN_CLASSIFICATION_MAPPING_NAMES", "AutoModelForTokenClassification"),
    ("multiple-choice", "MODEL_FOR_MULTIPLE_CHOICE_MAPPING_NAMES", "AutoModelForMultipleChoice"),
    (
        "next-sentence-prediction",
        "MODEL_FOR_NEXT_SENTENCE_PREDICTION_MAPPING_NAMES",
        "AutoModelForNextSentencePrediction",
    ),
    (
        "audio-frame-classification",
        "MODEL_FOR_AUDIO_FRAME_CLASSIFICATION_MAPPING_NAMES",
        "AutoModelForAudioFrameClassification",
    ),
    ("audio-xvector", "MODEL_FOR_AUDIO_XVECTOR_MAPPING_NAMES", "AutoModelForAudioXVector"),
    (
        "document-question-answering",
        "MODEL_FOR_DOCUMENT_QUESTION_ANSWERING_MAPPING_NAMES",
        "AutoModelForDocumentQuestionAnswering",
    ),
    (
        "visual-question-answering",
        "MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING_NAMES",
        "AutoModelForVisualQuestionAnswering",
    ),
    ("image-to-text", "MODEL_FOR_FOR_VISION_2_SEQ_MAPPING_NAMES", "AutoModelForVision2Seq"),
    (
        "zero-shot-image-classification",
        "MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING_NAMES",
        "AutoModelForZeroShotImageClassification",
    ),
    ("depth-estimation", "MODEL_FOR_DEPTH_ESTIMATION_MAPPING_NAMES", "AutoModelForDepthEstimation"),
    ("video-classification", "MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING_NAMES", "AutoModelForVideoClassification"),
    ("mask-generation", "MODEL_FOR_MASK_GENERATION_MAPPING_NAMES", "AutoModelForMaskGeneration"),
]






def 	_snake_case						(      A_       :    List[str]   ):
  """simple docstring"""







  a_      :	List[Any]      =							re.finditer(""".+?(?:(?<=[a-z])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])|$)"""     ,		A_   )
  return [m.group(0   ) for m in matches]






def 	_snake_case						(      ):
  """simple docstring"""







  a_      :	Tuple      =							transformers_module.models.auto.configuration_auto.CONFIG_MAPPING_NAMES
  a_      :	str      =							{
      config.replace("""Config"""     ,		""""""   ): model_type for model_type, config in config_maping_names.items()
  }

  # Dictionaries flagging if each model prefix has a backend in PT/TF/Flax.
  a_      :	Any      =							collections.defaultdict(A_   )
  a_      :	Optional[Any]      =							collections.defaultdict(A_   )
  a_      :	List[str]      =							collections.defaultdict(A_   )

  # Let's lookup through all transformers object (once) and find if models are supported by a given backend.
  for attr_name in dir(A_   ):
    a_      :	int      =							None
    if _re_tf_models.match(A_   ) is not None:
      a_      :	List[Any]      =							tf_models
      a_      :	str      =							_re_tf_models.match(A_   ).groups()[0]
    elif _re_flax_models.match(A_   ) is not None:
      a_      :	int      =							flax_models
      a_      :	List[Any]      =							_re_flax_models.match(A_   ).groups()[0]
    elif _re_pt_models.match(A_   ) is not None:
      a_      :	str      =							pt_models
      a_      :	Optional[int]      =							_re_pt_models.match(A_   ).groups()[0]

    if lookup_dict is not None:
      while len(A_   ) > 0:
        if attr_name in model_prefix_to_model_type:
          a_      :	Any      =							True
          break
        # Try again after removing the last word in the name
        a_      :	Tuple      =							"""""".join(camel_case_split(A_   )[:-1]   )

  a_      :	Tuple      =							set(list(pt_models.keys()   ) + list(tf_models.keys()   ) + list(flax_models.keys()   )   )
  a_      :	Dict      =							list(A_   )
  all_models.sort()

  a_      :	str      =							{"""model_type""": all_models}
  a_      :	Union[str, Any]      =							[pt_models[t] for t in all_models]
  a_      :	Union[str, Any]      =							[tf_models[t] for t in all_models]
  a_      :	List[str]      =							[flax_models[t] for t in all_models]

  # Now let's use the auto-mapping names to make sure
  a_      :	int      =							{}
  for t in all_models:
    if t in transformers_module.models.auto.processing_auto.PROCESSOR_MAPPING_NAMES:
      a_      :	int      =							"""AutoProcessor"""
    elif t in transformers_module.models.auto.tokenization_auto.TOKENIZER_MAPPING_NAMES:
      a_      :	Union[str, Any]      =							"""AutoTokenizer"""
    elif t in transformers_module.models.auto.feature_extraction_auto.FEATURE_EXTRACTOR_MAPPING_NAMES:
      a_      :	int      =							"""AutoFeatureExtractor"""
    else:
      # Default to AutoTokenizer if a model has nothing, for backward compatibility.
      a_      :	Dict      =							"""AutoTokenizer"""

  a_      :	List[str]      =							[processors[t] for t in all_models]

  return pd.DataFrame(A_   )






def 	_snake_case						(      A_       :    Optional[int]   ):
  """simple docstring"""







  a_      :	Any      =							[
      transformers_module.models.auto.modeling_auto,
      transformers_module.models.auto.modeling_tf_auto,
      transformers_module.models.auto.modeling_flax_auto,
  ]
  for pipeline_tag, model_mapping, auto_class in PIPELINE_TAGS_AND_AUTO_MODELS:
    a_      :	List[Any]      =							[model_mapping, f'''TF_{model_mapping}''', f'''FLAX_{model_mapping}''']
    a_      :	Dict      =							[auto_class, f'''TF_{auto_class}''', f'''Flax_{auto_class}''']
    # Loop through all three frameworks
    for module, cls, mapping in zip(A_     ,		A_     ,		A_   ):
      # The type of pipeline may not exist in this framework
      if not hasattr(A_     ,		A_   ):
        continue
      # First extract all model_names
      a_      :	Optional[Any]      =							[]
      for name in getattr(A_     ,		A_   ).values():
        if isinstance(A_     ,		A_   ):
          model_names.append(A_   )
        else:
          model_names.extend(list(A_   )   )

            # Add pipeline tag and auto model class for those models
      table.update({model_name: (pipeline_tag, cls) for model_name in model_names}   )

  return table






def 	_snake_case						(      A_       :    Optional[Any]     ,		A_       :    int   ):
  """simple docstring"""







  a_      :	Tuple      =							get_frameworks_table()
  a_      :	Optional[int]      =							Dataset.from_pandas(A_   )

  a_      :	Optional[Any]      =							hf_hub_download(
      """huggingface/transformers-metadata"""     ,		"""pipeline_tags.json"""     ,		repo_type="""dataset"""     ,		token=A_   )
  a_      :	List[Any]      =							Dataset.from_json(A_   )
  a_      :	List[Any]      =							{
      tags_dataset[i]["""model_class"""]: (tags_dataset[i]["""pipeline_tag"""], tags_dataset[i]["""auto_class"""])
      for i in range(len(A_   )   )
  }
  a_      :	int      =							update_pipeline_and_auto_class_table(A_   )

  # Sort the model classes to avoid some nondeterministic updates to create false update commits.
  a_      :	Optional[int]      =							sorted(table.keys()   )
  a_      :	str      =							pd.DataFrame(
      {
          """model_class""": model_classes,
          """pipeline_tag""": [table[m][0] for m in model_classes],
          """auto_class""": [table[m][1] for m in model_classes],
      }   )
  a_      :	Optional[int]      =							Dataset.from_pandas(A_   )

  with tempfile.TemporaryDirectory() as tmp_dir:
    frameworks_dataset.to_json(os.path.join(A_     ,		"""frameworks.json"""   )   )
    tags_dataset.to_json(os.path.join(A_     ,		"""pipeline_tags.json"""   )   )

    if commit_sha is not None:
      a_      :	Tuple      =							(
          f'''Update with commit {commit_sha}\n\nSee: '''
          f'''https://github.com/huggingface/transformers/commit/{commit_sha}'''
      )
    else:
      a_      :	Tuple      =							"""Update"""

    upload_folder(
        repo_id="""huggingface/transformers-metadata"""     ,		folder_path=A_     ,		repo_type="""dataset"""     ,		token=A_     ,		commit_message=A_     ,		)






def 	_snake_case						(      ):
  """simple docstring"""







  a_      :	List[str]      =							{tag: cls for tag, _, cls in PIPELINE_TAGS_AND_AUTO_MODELS}
  a_      :	Optional[int]      =							transformers_module.pipelines.SUPPORTED_TASKS
  a_      :	int      =							[]
  for key in pipeline_tasks:
    if key not in in_table:
      a_      :	Any      =							pipeline_tasks[key]["""pt"""]
      if isinstance(A_     ,		(list, tuple)   ):
        a_      :	int      =							model[0]
      a_      :	List[Any]      =							model.__name__
      if model not in in_table.values():
        missing.append(A_   )

  if len(A_   ) > 0:
    a_      :	List[Any]      =							""", """.join(A_   )
    raise ValueError(
        """The following pipeline tags are not present in the `PIPELINE_TAGS_AND_AUTO_MODELS` constant inside """
        f'''`utils/update_metadata.py`: {msg}. Please add them!'''   )


if __name__ == "__main__":
    __snake_case:  Optional[Any]								=			argparse.ArgumentParser()
    parser.add_argument("--token", type=str, help="The token to use to push to the transformers-metadata dataset.")
    parser.add_argument("--commit_sha", type=str, help="The sha of the commit going with this update.")
    parser.add_argument("--check-only", action="store_true", help="Activate to just check all pipelines are present.")
    __snake_case:  List[Any]								=			parser.parse_args()

    if args.check_only:
        check_pipeline_tags()
    else:
        update_metadata(args.token, args.commit_sha)




'''simple docstring'''

import unittest

import numpy as np

from transformers.testing_utils import is_flaky, require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
    import torch

if is_vision_available():
    from PIL import Image

    from transformers import DonutImageProcessor


class  _UpperCAmelCase ( unittest.TestCase    ):

     """simple docstring"""

     def __init__(      self							,		lowerCAmelCase_							,		lowerCAmelCase_=7							,		lowerCAmelCase_=3							,		lowerCAmelCase_=18							,		lowerCAmelCase_=30							,		lowerCAmelCase_=4_00							,		lowerCAmelCase_=True							,		lowerCAmelCase_=None							,		lowerCAmelCase_=True							,		lowerCAmelCase_=False							,		lowerCAmelCase_=True							,		lowerCAmelCase_=True							,		lowerCAmelCase_=[0.5, 0.5, 0.5]							,		lowerCAmelCase_=[0.5, 0.5, 0.5]							,		):

       '''simple docstring'''

       a_      :	Optional[int]      =							parent
       a_      :	Union[str, Any]      =							batch_size
       a_      :	Any      =							num_channels
       a_      :	Tuple      =							image_size
       a_      :	Any      =							min_resolution
       a_      :	Dict      =							max_resolution
       a_      :	Optional[Any]      =							do_resize
       a_      :	Tuple      =							size if size is not None else {"""height""": 18, """width""": 20}
       a_      :	int      =							do_thumbnail
       a_      :	Optional[int]      =							do_align_axis
       a_      :	Tuple      =							do_pad
       a_      :	int      =							do_normalize
       a_      :	Tuple      =							image_mean
       a_      :	Optional[Any]      =							image_std






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return {
           "do_resize": self.do_resize,
           "size": self.size,
           "do_thumbnail": self.do_thumbnail,
           "do_align_long_axis": self.do_align_axis,
           "do_pad": self.do_pad,
           "do_normalize": self.do_normalize,
           "image_mean": self.image_mean,
           "image_std": self.image_std,
       }







@require_torch
@require_vision
class  _UpperCAmelCase ( lowerCAmelCase__	,unittest.TestCase    ):

     """simple docstring"""

     a_    	=					DonutImageProcessor if is_vision_available() else None


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Optional[int]      =							DonutImageProcessingTester(self )


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return self.image_processor_tester.prepare_image_processor_dict()


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Any      =							self.image_processing_class(**self.image_processor_dict )
       self.assertTrue(hasattr(lowerCAmelCase_							,		"""do_resize""" ) )
       self.assertTrue(hasattr(lowerCAmelCase_							,		"""size""" ) )
       self.assertTrue(hasattr(lowerCAmelCase_							,		"""do_thumbnail""" ) )
       self.assertTrue(hasattr(lowerCAmelCase_							,		"""do_align_long_axis""" ) )
       self.assertTrue(hasattr(lowerCAmelCase_							,		"""do_pad""" ) )
       self.assertTrue(hasattr(lowerCAmelCase_							,		"""do_normalize""" ) )
       self.assertTrue(hasattr(lowerCAmelCase_							,		"""image_mean""" ) )
       self.assertTrue(hasattr(lowerCAmelCase_							,		"""image_std""" ) )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	List[str]      =							self.image_processing_class.from_dict(self.image_processor_dict )
       self.assertEqual(image_processor.size							,		{"""height""": 18, """width""": 20} )

       a_      :	str      =							self.image_processing_class.from_dict(self.image_processor_dict							,		size=42 )
       self.assertEqual(image_processor.size							,		{"""height""": 42, """width""": 42} )

       # Previous config had dimensions in (width, height) order
       a_      :	Any      =							self.image_processing_class.from_dict(self.image_processor_dict							,		size=(42, 84) )
       self.assertEqual(image_processor.size							,		{"""height""": 84, """width""": 42} )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       pass


     @is_flaky()
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Dict      =							self.image_processing_class(**self.image_processor_dict )
       # create random PIL images
       a_      :	str      =							prepare_image_inputs(self.image_processor_tester							,		equal_resolution=lowerCAmelCase_ )
       for image in image_inputs:
         self.assertIsInstance(lowerCAmelCase_							,		Image.Image )

       # Test not batched input
       a_      :	Dict      =							image_processing(image_inputs[0]							,		return_tensors="""pt""" ).pixel_values
       self.assertEqual(
           encoded_images.shape							,		(
               1,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size["""height"""],
               self.image_processor_tester.size["""width"""],
           )							,		)

       # Test batched
       a_      :	int      =							image_processing(lowerCAmelCase_							,		return_tensors="""pt""" ).pixel_values
       self.assertEqual(
           encoded_images.shape							,		(
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size["""height"""],
               self.image_processor_tester.size["""width"""],
           )							,		)


     @is_flaky()
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	List[str]      =							self.image_processing_class(**self.image_processor_dict )
       # create random numpy tensors
       a_      :	int      =							prepare_image_inputs(self.image_processor_tester							,		equal_resolution=lowerCAmelCase_							,		numpify=lowerCAmelCase_ )
       for image in image_inputs:
         self.assertIsInstance(lowerCAmelCase_							,		np.ndarray )

       # Test not batched input
       a_      :	Optional[Any]      =							image_processing(image_inputs[0]							,		return_tensors="""pt""" ).pixel_values
       self.assertEqual(
           encoded_images.shape							,		(
               1,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size["""height"""],
               self.image_processor_tester.size["""width"""],
           )							,		)

       # Test batched
       a_      :	Dict      =							image_processing(lowerCAmelCase_							,		return_tensors="""pt""" ).pixel_values
       self.assertEqual(
           encoded_images.shape							,		(
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size["""height"""],
               self.image_processor_tester.size["""width"""],
           )							,		)






     @is_flaky()
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							self.image_processing_class(**self.image_processor_dict )
       # create random PyTorch tensors
       a_      :	Optional[Any]      =							prepare_image_inputs(self.image_processor_tester							,		equal_resolution=lowerCAmelCase_							,		torchify=lowerCAmelCase_ )
       for image in image_inputs:
         self.assertIsInstance(lowerCAmelCase_							,		torch.Tensor )

       # Test not batched input
       a_      :	Optional[int]      =							image_processing(image_inputs[0]							,		return_tensors="""pt""" ).pixel_values
       self.assertEqual(
           encoded_images.shape							,		(
               1,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size["""height"""],
               self.image_processor_tester.size["""width"""],
           )							,		)

       # Test batched
       a_      :	Any      =							image_processing(lowerCAmelCase_							,		return_tensors="""pt""" ).pixel_values
       self.assertEqual(
           encoded_images.shape							,		(
               self.image_processor_tester.batch_size,
               self.image_processor_tester.num_channels,
               self.image_processor_tester.size["""height"""],
               self.image_processor_tester.size["""width"""],
           )							,		)


