




'''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
lowercase      :   Union[str, Any]      =       """src/transformers"""


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


# Regexes that match TF/Flax/PT model names.
lowercase      :   Optional[Any]      =       re.compile(R"""TF(.*)(?:Model|Encoder|Decoder|ForConditionalGeneration)""")
lowercase      :   List[Any]      =       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.
lowercase      :   List[str]      =       re.compile(R"""(.*)(?:Model|Encoder|Decoder|ForConditionalGeneration)""")


# Fill this with tuples (pipeline_tag, model_mapping, auto_model)
lowercase      :   Optional[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        lowerCamelCase__				(      __lowercase				):
     snake_case			:      Dict        =			re.finditer(""".+?(?:(?<=[a-z])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])|$)"""     ,   __lowercase				)
     return [m.group(0				) for m in matches]



def        lowerCamelCase__				(      ):
     snake_case			:      str        =			transformers_module.models.auto.configuration_auto.CONFIG_MAPPING_NAMES
     snake_case			:      List[Any]        =			{
         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.
     snake_case			:      Dict        =			collections.defaultdict(__lowercase				)
     snake_case			:      str        =			collections.defaultdict(__lowercase				)
     snake_case			:      Union[str, Any]        =			collections.defaultdict(__lowercase				)

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

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

     snake_case			:      int        =			set(list(pt_models.keys()				) + list(tf_models.keys()				) + list(flax_models.keys()				)				)
     snake_case			:      str        =			list(__lowercase				)
     all_models.sort()

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

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

     snake_case			:      Any        =			[processors[t] for t in all_models]

     return pd.DataFrame(__lowercase				)



def        lowerCamelCase__				(      __lowercase				):
     snake_case			:      Dict        =			[
         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:
          snake_case			:      List[Any]        =			[model_mapping, F'''TF_{model_mapping}''', F'''FLAX_{model_mapping}''']
          snake_case			:      Optional[Any]        =			[auto_class, F'''TF_{auto_class}''', F'''Flax_{auto_class}''']
          # Loop through all three frameworks
          for module, cls, mapping in zip(__lowercase     ,   __lowercase     ,   __lowercase				):
               # The type of pipeline may not exist in this framework
               if not hasattr(__lowercase     ,   __lowercase				):
                    continue
               # First extract all model_names
               snake_case			:      int        =			[]
               for name in getattr(__lowercase     ,   __lowercase				).values():
                    if isinstance(__lowercase     ,   __lowercase				):
                         model_names.append(__lowercase				)
                    else:
                         model_names.extend(list(__lowercase				)				)

            # 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        lowerCamelCase__				(      __lowercase     ,   __lowercase				):
     snake_case			:      str        =			get_frameworks_table()
     snake_case			:      Any        =			Dataset.from_pandas(__lowercase				)

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

     # Sort the model classes to avoid some nondeterministic updates to create false update commits.
     snake_case			:      List[Any]        =			sorted(table.keys()				)
     snake_case			:      Any        =			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],
         }				)
     snake_case			:      List[Any]        =			Dataset.from_pandas(__lowercase				)

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

          if commit_sha is not None:
               snake_case			:      Optional[Any]        =			(
                   F'''Update with commit {commit_sha}\n\nSee: '''
                   F'''https://github.com/huggingface/transformers/commit/{commit_sha}'''
               )
          else:
               snake_case			:      List[str]        =			"""Update"""

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



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

     if len(__lowercase				) > 0:
          snake_case			:      Union[str, Any]        =			""", """.join(__lowercase				)
          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__":
      lowercase      :   str      =       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.""")
      lowercase      :   List[Any]      =       parser.parse_args()

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









'''simple docstring'''







# We ignore warnings about stepping the scheduler since we step it ourselves during gradient accumulation

import warnings

from .state import AcceleratorState, GradientState


warnings.filterwarnings("""ignore""", category=UserWarning, module="""torch.optim.lr_scheduler""")







class 					_a :





     '''simple docstring'''



     def __init__(				self							,__a							,__a							,__a = True							,__a = False							)      ->				Dict:
          snake_case			:      List[Any]        =			scheduler
          snake_case			:      List[Any]        =			optimizers if isinstance(__a							,(list, tuple)							) else [optimizers]
          snake_case			:      List[str]        =			split_batches
          snake_case			:      List[str]        =			step_with_optimizer
          snake_case			:      int        =			GradientState()







     def 			snake_case_      (				self							,*__a							,**__a							)      ->				Optional[Any]:
          if not self.step_with_optimizer:
               # No link between scheduler and optimizer -> just step
               self.scheduler.step(*__a							,**__a							)
               return

          # Otherwise, first make sure the optimizer was stepped.
          if not self.gradient_state.sync_gradients:
               if self.gradient_state.adjust_scheduler:
                    self.scheduler._step_count += 1
               return

          for opt in self.optimizers:
               if opt.step_was_skipped:
                    return
          if self.split_batches:
               # Split batches -> the training dataloader batch size is not changed so one step per training step
               self.scheduler.step(*__a							,**__a							)
          else:
               # Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
               # num_processes steps per training step
               snake_case			:      List[str]        =			AcceleratorState().num_processes
               for _ in range(__a							):
                    # Special case when using OneCycle and `drop_last` was not used
                    if hasattr(self.scheduler							,"""total_steps"""							):
                         if self.scheduler._step_count <= self.scheduler.total_steps:
                              self.scheduler.step(*__a							,**__a							)
                    else:
                         self.scheduler.step(*__a							,**__a							)







     def 			snake_case_      (				self							)      ->				Dict:
          return self.scheduler.get_last_lr()







     def 			snake_case_      (				self							)      ->				int:
          return self.scheduler.state_dict()







     def 			snake_case_      (				self							,__a							)      ->				Any:
          self.scheduler.load_state_dict(__a							)







     def 			snake_case_      (				self							)      ->				Any:
          return self.scheduler.get_lr()







     def 			snake_case_      (				self							,*__a							,**__a							)      ->				Optional[Any]:
          return self.scheduler.print_lr(*__a							,**__a							)



