




'''simple docstring'''
import importlib
import inspect
import json
import os
from collections import OrderedDict

# Build the list of all feature extractors
from ...configuration_utils import PretrainedConfig
from ...dynamic_module_utils import get_class_from_dynamic_module, resolve_trust_remote_code
from ...feature_extraction_utils import FeatureExtractionMixin
from ...image_processing_utils import ImageProcessingMixin
from ...tokenization_utils import TOKENIZER_CONFIG_FILE
from ...utils import FEATURE_EXTRACTOR_NAME, get_file_from_repo, logging
from .auto_factory import _LazyAutoMapping
from .configuration_auto import (
    CONFIG_MAPPING_NAMES,
    AutoConfig,
    model_type_to_module_name,
    replace_list_option_in_docstrings,
)
from .feature_extraction_auto import AutoFeatureExtractor
from .image_processing_auto import AutoImageProcessor
from .tokenization_auto import AutoTokenizer


A							:				Optional[int]			     =	logging.get_logger(__name__)

A							:				Union[str, Any]			     =	OrderedDict(
    [
        ("""align""", """AlignProcessor"""),
        ("""altclip""", """AltCLIPProcessor"""),
        ("""bark""", """BarkProcessor"""),
        ("""blip""", """BlipProcessor"""),
        ("""blip-2""", """Blip2Processor"""),
        ("""bridgetower""", """BridgeTowerProcessor"""),
        ("""chinese_clip""", """ChineseCLIPProcessor"""),
        ("""clap""", """ClapProcessor"""),
        ("""clip""", """CLIPProcessor"""),
        ("""clipseg""", """CLIPSegProcessor"""),
        ("""flava""", """FlavaProcessor"""),
        ("""git""", """GitProcessor"""),
        ("""groupvit""", """CLIPProcessor"""),
        ("""hubert""", """Wav2Vec2Processor"""),
        ("""instructblip""", """InstructBlipProcessor"""),
        ("""layoutlmv2""", """LayoutLMv2Processor"""),
        ("""layoutlmv3""", """LayoutLMv3Processor"""),
        ("""markuplm""", """MarkupLMProcessor"""),
        ("""mctct""", """MCTCTProcessor"""),
        ("""mgp-str""", """MgpstrProcessor"""),
        ("""oneformer""", """OneFormerProcessor"""),
        ("""owlvit""", """OwlViTProcessor"""),
        ("""pix2struct""", """Pix2StructProcessor"""),
        ("""sam""", """SamProcessor"""),
        ("""sew""", """Wav2Vec2Processor"""),
        ("""sew-d""", """Wav2Vec2Processor"""),
        ("""speech_to_text""", """Speech2TextProcessor"""),
        ("""speech_to_text_2""", """Speech2Text2Processor"""),
        ("""speecht5""", """SpeechT5Processor"""),
        ("""trocr""", """TrOCRProcessor"""),
        ("""tvlt""", """TvltProcessor"""),
        ("""unispeech""", """Wav2Vec2Processor"""),
        ("""unispeech-sat""", """Wav2Vec2Processor"""),
        ("""vilt""", """ViltProcessor"""),
        ("""vision-text-dual-encoder""", """VisionTextDualEncoderProcessor"""),
        ("""wav2vec2""", """Wav2Vec2Processor"""),
        ("""wav2vec2-conformer""", """Wav2Vec2Processor"""),
        ("""wavlm""", """Wav2Vec2Processor"""),
        ("""whisper""", """WhisperProcessor"""),
        ("""xclip""", """XCLIPProcessor"""),
    ]
)

A							:				List[Any]			     =	_LazyAutoMapping(CONFIG_MAPPING_NAMES, PROCESSOR_MAPPING_NAMES)
def     snake_case_  (       a__		:     str      ):



    """simple docstring"""




    for module_name, processors in PROCESSOR_MAPPING_NAMES.items():
        if class_name in processors:
            __lowercase          =						model_type_to_module_name(a__      )

            __lowercase          =						importlib.import_module(f'.{module_name}'   ,"""transformers.models"""      )
            try:
                return getattr(a__   ,a__      )
            except AttributeError:
                continue

    for processor in PROCESSOR_MAPPING._extra_content.values():
        if getattr(a__   ,"""__name__"""   ,a__      ) == class_name:
            return processor

    # We did not fine the class, but maybe it's because a dep is missing. In that case, the class will be in the main
    # init and we return the proper dummy to get an appropriate error message.
    __lowercase          =						importlib.import_module("""transformers"""      )
    if hasattr(a__   ,a__      ):
        return getattr(a__   ,a__      )

    return None






class 	SCREAMING_SNAKE_CASE:

       def __init__(   self					)   -> str:

           """simple docstring"""



           raise EnvironmentError(
               """AutoProcessor is designed to be instantiated """
               """using the `AutoProcessor.from_pretrained(pretrained_model_name_or_path)` method."""					)

       @classmethod
       @replace_list_option_in_docstrings(lowerCamelCase__					)
       def    snake_case__      (   cls ,		lowerCamelCase__ ,		**lowerCamelCase__					)   -> str:

           """simple docstring"""



           __lowercase          =						kwargs.pop("""config""" ,		lowerCamelCase__					)
           __lowercase          =						kwargs.pop("""trust_remote_code""" ,		lowerCamelCase__					)
           __lowercase          =						True

           __lowercase          =						None
           __lowercase          =						None

           # First, let's see if we have a preprocessor config.
           # Filter the kwargs for `get_file_from_repo`.
           __lowercase          =						{
               key: kwargs[key] for key in inspect.signature(lowerCamelCase__					).parameters.keys() if key in kwargs
           }
           # Let's start by checking whether the processor class is saved in an image processor
           __lowercase          =						get_file_from_repo(
               lowerCamelCase__ ,		lowerCamelCase__ ,		**lowerCamelCase__					)
           if preprocessor_config_file is not None:
               __lowercase    ,__lowercase          =						ImageProcessingMixin.get_image_processor_dict(lowerCamelCase__ ,		**lowerCamelCase__					)
               __lowercase          =						config_dict.get("""processor_class""" ,		lowerCamelCase__					)
               if "AutoProcessor" in config_dict.get("""auto_map""" ,		{}					):
                   __lowercase          =						config_dict["""auto_map"""]["""AutoProcessor"""]

        # If not found, let's check whether the processor class is saved in a feature extractor config
           if preprocessor_config_file is not None and processor_class is None:
               __lowercase    ,__lowercase          =						FeatureExtractionMixin.get_feature_extractor_dict(lowerCamelCase__ ,		**lowerCamelCase__					)
               __lowercase          =						config_dict.get("""processor_class""" ,		lowerCamelCase__					)
               if "AutoProcessor" in config_dict.get("""auto_map""" ,		{}					):
                   __lowercase          =						config_dict["""auto_map"""]["""AutoProcessor"""]

           if processor_class is None:
               # Next, let's check whether the processor class is saved in a tokenizer
               __lowercase          =						get_file_from_repo(
                   lowerCamelCase__ ,		lowerCamelCase__ ,		**lowerCamelCase__					)
               if tokenizer_config_file is not None:
                   with open(lowerCamelCase__ ,		encoding="""utf-8"""					) as reader:
                       __lowercase          =						json.load(lowerCamelCase__					)

                   __lowercase          =						config_dict.get("""processor_class""" ,		lowerCamelCase__					)
                   if "AutoProcessor" in config_dict.get("""auto_map""" ,		{}					):
                       __lowercase          =						config_dict["""auto_map"""]["""AutoProcessor"""]

           if processor_class is None:
               # Otherwise, load config, if it can be loaded.
               if not isinstance(lowerCamelCase__ ,		lowerCamelCase__					):
                   __lowercase          =						AutoConfig.from_pretrained(
                       lowerCamelCase__ ,		trust_remote_code=lowerCamelCase__ ,		**lowerCamelCase__					)

               # And check if the config contains the processor class.
               __lowercase          =						getattr(lowerCamelCase__ ,		"""processor_class""" ,		lowerCamelCase__					)
               if hasattr(lowerCamelCase__ ,		"""auto_map"""					) and "AutoProcessor" in config.auto_map:
                   __lowercase          =						config.auto_map["""AutoProcessor"""]

           if processor_class is not None:
               __lowercase          =						processor_class_from_name(lowerCamelCase__					)

           __lowercase          =						processor_auto_map is not None
           __lowercase          =						processor_class is not None or type(lowerCamelCase__					) in PROCESSOR_MAPPING
           __lowercase          =						resolve_trust_remote_code(
               lowerCamelCase__ ,		lowerCamelCase__ ,		lowerCamelCase__ ,		lowerCamelCase__					)

           if has_remote_code and trust_remote_code:
               __lowercase          =						get_class_from_dynamic_module(
                   lowerCamelCase__ ,		lowerCamelCase__ ,		**lowerCamelCase__					)
               __lowercase          =						kwargs.pop("""code_revision""" ,		lowerCamelCase__					)
               if os.path.isdir(lowerCamelCase__					):
                   processor_class.register_for_auto_class()
               return processor_class.from_pretrained(
                   lowerCamelCase__ ,		trust_remote_code=lowerCamelCase__ ,		**lowerCamelCase__					)
           elif processor_class is not None:
               return processor_class.from_pretrained(
                   lowerCamelCase__ ,		trust_remote_code=lowerCamelCase__ ,		**lowerCamelCase__					)
           # Last try: we use the PROCESSOR_MAPPING.
           elif type(lowerCamelCase__					) in PROCESSOR_MAPPING:
               return PROCESSOR_MAPPING[type(lowerCamelCase__					)].from_pretrained(lowerCamelCase__ ,		**lowerCamelCase__					)

           # At this stage, there doesn't seem to be a `Processor` class available for this model, so let's try a
           # tokenizer.
           try:
               return AutoTokenizer.from_pretrained(
                   lowerCamelCase__ ,		trust_remote_code=lowerCamelCase__ ,		**lowerCamelCase__					)
           except Exception:
               try:
                   return AutoImageProcessor.from_pretrained(
                       lowerCamelCase__ ,		trust_remote_code=lowerCamelCase__ ,		**lowerCamelCase__					)
               except Exception:
                   pass

               try:
                   return AutoFeatureExtractor.from_pretrained(
                       lowerCamelCase__ ,		trust_remote_code=lowerCamelCase__ ,		**lowerCamelCase__					)
               except Exception:
                   pass

           raise ValueError(
               F'Unrecognized processing class in {pretrained_model_name_or_path}. Can\'t instantiate a processor, a '
               """tokenizer, an image processor or a feature extractor for this model. Make sure the repository contains"""
               """the files of at least one of those processing classes."""					)




       @staticmethod
       def    snake_case__      (   lowerCamelCase__ ,		lowerCamelCase__					)   -> Optional[int]:

           """simple docstring"""



           PROCESSOR_MAPPING.register(lowerCamelCase__ ,		lowerCamelCase__					)










'''simple docstring'''
import argparse
import glob
import inspect
import os
import re
import unittest

from get_test_info import get_test_classes

from tests.test_pipeline_mixin import pipeline_test_mapping


A							:				Any			     =	{}
for task, _ in pipeline_test_mapping.items():
  A							:				Optional[Any]			     =	{"""pt""": None, """tf""": None}


# DO **NOT** add item to this set (unless the reason is approved)
A							:				Union[str, Any]			     =	{
    """tests/models/esm/test_modeling_esmfold.py""",  # The pipeline test mapping is added to `test_modeling_esm.py`
}
def     snake_case_  (       a__		:     Optional[Any]      ):



    """simple docstring"""




    if "ModelTesterMixin" in [x.__name__ for x in test_class.__bases__]:
        return "pt"
    elif "TFModelTesterMixin" in [x.__name__ for x in test_class.__bases__]:
        return "tf"
    elif "FlaxModelTesterMixin" in [x.__name__ for x in test_class.__bases__]:
        return "flax"
    else:
        return None
def     snake_case_  (       a__		:     Union[str, Any]   ,a__		:     str      ):



    """simple docstring"""




    if PIPELINE_TEST_MAPPING[task].get(a__   ,a__      ) is not None:
        return PIPELINE_TEST_MAPPING[task][framework]

    __lowercase          =						pipeline_test_mapping[task]["""test"""]
    __lowercase          =						None

    if framework == "pt":
        __lowercase          =						getattr(a__   ,"""model_mapping"""   ,a__      )
    elif framework == "tf":
        __lowercase          =						getattr(a__   ,"""tf_model_mapping"""   ,a__      )

    if mapping is not None:
        __lowercase          =						dict(mapping.items()      )

    # cache the results
    __lowercase          =						mapping
    return mapping
def     snake_case_  (       a__		:     Optional[int]   ,a__		:     Optional[Any]      ):



    """simple docstring"""




    __lowercase          =						get_framework(a__      )
    if framework is None:
        return None
    __lowercase          =						get_mapping_for_task(a__   ,a__      )
    if mapping is None:
        return None

    __lowercase          =						list({model_class.config_class for model_class in test_class.all_model_classes}      )
    if len(a__      ) != 1:
        raise ValueError("""There should be exactly one configuration class from `test_class.all_model_classes`."""      )

    # This could be a list/tuple of model classes, but it's rare.
    __lowercase          =						mapping.get(config_classes[0]   ,a__      )
    if isinstance(a__   ,(tuple, list)      ):
        __lowercase          =						sorted(a__   ,key=lambda a__      : x.__name__      )

    return model_class
def     snake_case_  (       a__		:     Any      ):



    """simple docstring"""




    __lowercase          =						[(task, get_model_for_pipeline_test(a__   ,a__      )) for task in pipeline_test_mapping]
    __lowercase          =						sorted([(task, model) for task, model in mapping if model is not None]   ,key=lambda a__      : x[0]      )

    return dict(a__      )
def     snake_case_  (       a__		:     Optional[int]      ):



    """simple docstring"""




    __lowercase          =						get_framework(a__      )
    if framework == "pt":
        __lowercase          =						"""torch"""
    __lowercase          =						"""{}"""

    __lowercase          =						get_pipeline_model_mapping(a__      )
    if len(a__      ) == 0:
        return ""

    __lowercase          =						[]
    for task, model_classes in mapping.items():
        if isinstance(a__   ,(tuple, list)      ):
            # A list/tuple of model classes
            __lowercase          =						"""(""" + """, """.join([x.__name__ for x in model_classes]      ) + """)"""
        else:
            # A single model class
            __lowercase          =						model_classes.__name__
        texts.append(f'"{task}": {value}'      )
    __lowercase          =						"""{""" + """, """.join(a__      ) + """}"""
    __lowercase          =						f'pipeline_model_mapping = {text} if is_{framework}_available() else {default_value}'

    return text
def     snake_case_  (       a__		:     List[Any]      ):



    """simple docstring"""




    __lowercase          =						{"""ModelTesterMixin""", """TFModelTesterMixin""", """FlaxModelTesterMixin"""}
    if not issubclass(a__   ,unittest.TestCase      ):
        return False
    return len(base_class_names.intersection([x.__name__ for x in test_class.__bases__]      )      ) > 0
def     snake_case_  (       a__		:     List[Any]      ):



    """simple docstring"""




    __lowercase          =						[x for x in get_test_classes(a__      ) if is_valid_test_class(a__      )]

    __lowercase          =						None
    for test_class in test_classes:
        # If a test class has defined `pipeline_model_mapping`, let's take it
        if getattr(a__   ,"""pipeline_model_mapping"""   ,a__      ) is not None:
            __lowercase          =						test_class
            break
    # Take the test class with the shortest name (just a heuristic)
    if target_test_class is None and len(a__      ) > 0:
        __lowercase          =						sorted(a__   ,key=lambda a__      : (len(x.__name__      ), x.__name__)      )[0]

    return target_test_class
def     snake_case_  (       a__		:     Any   ,a__		:     Optional[int]   ,a__		:     int      ):



    """simple docstring"""




    __lowercase          =						start_idx
    for idx, line in enumerate(lines[start_idx:]      ):
        __lowercase          =						len(a__      ) - len(line.lstrip()      )
        if idx == 0 or indent > indent_level or (indent == indent_level and line.strip() == ")"):
            __lowercase          =						start_idx + idx
        elif idx > 0 and indent <= indent_level:
            # Outside the definition block of `pipeline_model_mapping`
            break

    return end_idx
def     snake_case_  (       a__		:     int   ,a__		:     List[Any]=False      ):



    """simple docstring"""




    if getattr(a__   ,"""pipeline_model_mapping"""   ,a__      ) is not None:
        if not overwrite:
            return "", -1

    __lowercase          =						get_pipeline_model_mapping_string(a__      )
    if len(a__      ) == 0:
        return "", -1
    __lowercase          =						line_to_add + """\n"""

    # The code defined the class `test_class`
    __lowercase    ,__lowercase          =						inspect.getsourcelines(a__      )
    # `inspect` gives the code for an object, including decorator(s) if any.
    # We (only) need the exact line of the class definition.
    for idx, line in enumerate(a__      ):
        if line.lstrip().startswith("""class """      ):
            __lowercase          =						class_lines[idx:]
            class_start_line_no += idx
            break
    __lowercase          =						class_start_line_no + len(a__      ) - 1

    # The index in `class_lines` that starts the definition of `all_model_classes`, `all_generative_model_classes` or
    # `pipeline_model_mapping`. This assumes they are defined in such order, and we take the start index of the last
    # block that appears in a `test_class`.
    __lowercase          =						None
    # The indent level of the line at `class_lines[start_idx]` (if defined)
    __lowercase          =						0
    # To record if `pipeline_model_mapping` is found in `test_class`.
    __lowercase          =						None
    for idx, line in enumerate(a__      ):
        if line.strip().startswith("""all_model_classes = """      ):
            __lowercase          =						len(a__      ) - len(line.lstrip()      )
            __lowercase          =						idx
        elif line.strip().startswith("""all_generative_model_classes = """      ):
            __lowercase          =						len(a__      ) - len(line.lstrip()      )
            __lowercase          =						idx
        elif line.strip().startswith("""pipeline_model_mapping = """      ):
            __lowercase          =						len(a__      ) - len(line.lstrip()      )
            __lowercase          =						idx
            __lowercase          =						line
            break

    if start_idx is None:
        return "", -1
    # Find the ending index (inclusive) of the above found block.
    __lowercase          =						find_block_ending(a__   ,a__   ,a__      )

    # Extract `is_xxx_available()` from existing blocks: some models require specific libraries like `timm` and use
    # `is_timm_available()` instead of `is_torch_available()`.
    # Keep leading and trailing whitespaces
    __lowercase          =						re.compile(r"""\s(is_\S+?_available\(\))\s"""      )
    for line in class_lines[start_idx : end_idx + 1]:
        __lowercase          =						r.search(a__      )
        if backend_condition is not None:
            # replace the leading and trailing whitespaces to the space character " ".
            __lowercase          =						""" """ + backend_condition[0][1:-1] + """ """
            __lowercase          =						r.sub(a__   ,a__      )
            break

    if def_line is None:
        # `pipeline_model_mapping` is not defined. The target index is set to the ending index (inclusive) of
        # `all_model_classes` or `all_generative_model_classes`.
        __lowercase          =						end_idx
    else:
        # `pipeline_model_mapping` is defined. The target index is set to be one **BEFORE** its start index.
        __lowercase          =						start_idx - 1
        # mark the lines of the currently existing `pipeline_model_mapping` to be removed.
        for idx in range(a__   ,end_idx + 1      ):
            # These lines are going to be removed before writing to the test file.
            __lowercase          =						None  # noqa

    # Make sure the test class is a subclass of `PipelineTesterMixin`.
    __lowercase          =						[x.__name__ for x in test_class.__bases__]
    if "PipelineTesterMixin" not in parent_classes:
        # Put `PipelineTesterMixin` just before `unittest.TestCase`
        __lowercase          =						[x for x in parent_classes if x != """TestCase"""] + ["""PipelineTesterMixin"""]
        if "TestCase" in parent_classes:
            # Here we **assume** the original string is always with `unittest.TestCase`.
            _parent_classes.append("""unittest.TestCase"""      )
        __lowercase          =						""", """.join(_parent_classes      )
        for idx, line in enumerate(a__      ):
            # Find the ending of the declaration of `test_class`
            if line.strip().endswith("""):"""      ):
                # mark the lines of the declaration of `test_class` to be removed
                for _idx in range(idx + 1      ):
                    __lowercase          =						None  # noqa
                break
        # Add the new, one-line, class declaration for `test_class`
        __lowercase          =						f'class {test_class.__name__}({parent_classes}):\n'

    # Add indentation
    __lowercase          =						""" """ * indent_level + line_to_add
    # Insert `pipeline_model_mapping` to `class_lines`.
    # (The line at `target_idx` should be kept by definition!)
    __lowercase          =						class_lines[: target_idx + 1] + [line_to_add] + class_lines[target_idx + 1 :]
    # Remove the lines that are marked to be removed
    __lowercase          =						[x for x in class_lines if x is not None]

    # Move from test class to module (in order to write to the test file)
    __lowercase          =						inspect.getsourcelines(inspect.getmodule(a__      )      )[0]
    # Be careful with the 1-off between line numbers and array indices
    __lowercase          =						module_lines[: class_start_line_no - 1] + class_lines + module_lines[class_end_line_no:]
    __lowercase          =						"""""".join(a__      )

    __lowercase          =						inspect.getsourcefile(a__      )
    with open(a__   ,"""w"""   ,encoding="""UTF-8"""   ,newline="""\n"""      ) as fp:
        fp.write(a__      )

    return line_to_add
def     snake_case_  (       a__		:     List[str]   ,a__		:     str=False      ):



    """simple docstring"""




    __lowercase          =						find_test_class(a__      )
    if test_class:
        add_pipeline_model_mapping(a__   ,overwrite=a__      )


if __name__ == "__main__":
  A							:				List[Any]			     =	argparse.ArgumentParser()
  parser.add_argument(
      """--test_file""", type=str, help="""A path to the test file, starting with the repository's `tests` directory."""
  )
  parser.add_argument(
      """--all""",
      action="""store_true""",
      help="""If to check and modify all test files.""",
  )
  parser.add_argument(
      """--overwrite""",
      action="""store_true""",
      help="""If to overwrite a test class if it has already defined `pipeline_model_mapping`.""",
  )
  A							:				Dict			     =	parser.parse_args()

  if not args.all and not args.test_file:
    raise ValueError("""Please specify either `test_file` or pass `--all` to check/modify all test files.""")
  elif args.all and args.test_file:
    raise ValueError("""Only one of `--test_file` and `--all` could be specified.""")

  A							:				Union[str, Any]			     =	[]
  if args.test_file:
    A							:				int			     =	[args.test_file]
  else:
    A							:				Optional[int]			     =	os.path.join("""tests""", """models""", """**""", """test_modeling_*.py""")
    for test_file in glob.glob(pattern):
      # `Flax` is not concerned at this moment
      if not test_file.startswith("""test_modeling_flax_"""):
        test_files.append(test_file)

  for test_file in test_files:
    if test_file in TEST_FILE_TO_IGNORE:
      print(F"""[SKIPPED] {test_file} is skipped as it is in `TEST_FILE_TO_IGNORE` in the file {__file__}.""")
      continue
    add_pipeline_model_mapping_to_test_file(test_file, overwrite=args.overwrite)




