






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


UpperCAmelCase__            =	{}
for task, _ in pipeline_test_mapping.items():
     UpperCAmelCase__            =	{"pt": None, "tf": None}


# DO **NOT** add item to this set (unless the reason is approved)
UpperCAmelCase__            =	{
    "tests/models/esm/test_modeling_esmfold.py",  # The pipeline test mapping is added to `test_modeling_esm.py`
}
def 		_a					(   a				:Optional[int]				)							->    str:

 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 		_a					(   a				:Union[str, Any]			,							a				:List[str]				)							->    Tuple:
 # Use the cached results
 if PIPELINE_TEST_MAPPING[task].get(a			,							a				) is not None:
  return PIPELINE_TEST_MAPPING[task][framework]

 a							=   pipeline_test_mapping[task]['''test''']
 a							=   None

 if framework == "pt":
  a							=   getattr(a			,							'''model_mapping'''			,							a				)
 elif framework == "tf":
  a							=   getattr(a			,							'''tf_model_mapping'''			,							a				)

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

 # cache the results
 a							=   mapping
 return mapping
def 		_a					(   a				:int			,							a				:str				)							->    str:
 a							=   get_framework(a				)
 if framework is None:
  return None
 a							=   get_mapping_for_task(a			,							a				)
 if mapping is None:
  return None

 a							=   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.
 a							=   mapping.get(config_classes[0]			,							a				)
 if isinstance(a			,							(tuple, list)				):
  a							=   sorted(a			,							key=lambda a				: x.__name__				)

 return model_class
def 		_a					(   a				:Optional[int]				)							->    Union[str, Any]:
 a							=   [(task, get_model_for_pipeline_test(a			,							a				)) for task in pipeline_test_mapping]
 a							=   sorted([(task, model) for task, model in mapping if model is not None]			,							key=lambda a				: x[0]				)

 return dict(a				)
def 		_a					(   a				:Union[str, Any]				)							->    Any:
 a							=   get_framework(a				)
 if framework == "pt":
  a							=   '''torch'''
 a							=   '''{}'''

 a							=   get_pipeline_model_mapping(a				)
 if len(a				) == 0:
  return ""

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

 return text
def 		_a					(   a				:List[str]				)							->    str:
 a							=   {'''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 		_a					(   a				:Dict				)							->    List[str]:
 a							=   [x for x in get_test_classes(a				) if is_valid_test_class(a				)]

 a							=   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:
   a							=   test_class
   break
    # Take the test class with the shortest name (just a heuristic)
 if target_test_class is None and len(a				) > 0:
  a							=   sorted(a			,							key=lambda a				: (len(x.__name__				), x.__name__)				)[0]

 return target_test_class
def 		_a					(   a				:Optional[int]			,							a				:Tuple			,							a				:int				)							->    Optional[Any]:
 a							=   start_idx
 for idx, line in enumerate(lines[start_idx:]				):
  a							=   len(a				) - len(line.lstrip()				)
  if idx == 0 or indent > indent_level or (indent == indent_level and line.strip() == ")"):
   a							=   start_idx + idx
  elif idx > 0 and indent <= indent_level:
   # Outside the definition block of `pipeline_model_mapping`
   break

 return end_idx
def 		_a					(   a				:str			,							a				:Any=False				)							->    Any:
 if getattr(a			,							'''pipeline_model_mapping'''			,							a				) is not None:
  if not overwrite:
   return "", -1

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

 # The code defined the class `test_class`
 a						,  a							=   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 '''				):
   a							=   class_lines[idx:]
   class_start_line_no += idx
   break
 a							=   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`.
 a							=   None
 # The indent level of the line at `class_lines[start_idx]` (if defined)
 a							=   0
 # To record if `pipeline_model_mapping` is found in `test_class`.
 a							=   None
 for idx, line in enumerate(a				):
  if line.strip().startswith('''all_model_classes = '''				):
   a							=   len(a				) - len(line.lstrip()				)
   a							=   idx
  elif line.strip().startswith('''all_generative_model_classes = '''				):
   a							=   len(a				) - len(line.lstrip()				)
   a							=   idx
  elif line.strip().startswith('''pipeline_model_mapping = '''				):
   a							=   len(a				) - len(line.lstrip()				)
   a							=   idx
   a							=   line
   break

 if start_idx is None:
  return "", -1
 # Find the ending index (inclusive) of the above found block.
 a							=   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
 a							=   re.compile(r'''\s(is_\S+?_available\(\))\s'''				)
 for line in class_lines[start_idx : end_idx + 1]:
  a							=   r.search(a				)
  if backend_condition is not None:
   # replace the leading and trailing whitespaces to the space character " ".
   a							=   ''' ''' + backend_condition[0][1:-1] + ''' '''
   a							=   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`.
  a							=   end_idx
 else:
  # `pipeline_model_mapping` is defined. The target index is set to be one **BEFORE** its start index.
  a							=   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.
   a							=   None  # noqa

    # Make sure the test class is a subclass of `PipelineTesterMixin`.
 a							=   [x.__name__ for x in test_class.__bases__]
 if "PipelineTesterMixin" not in parent_classes:
  # Put `PipelineTesterMixin` just before `unittest.TestCase`
  a							=   [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'''				)
  a							=   ''', '''.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				):
     a							=   None  # noqa
    break
        # Add the new, one-line, class declaration for `test_class`
  a							=   F"""class {test_class.__name__}({parent_classes}):\n"""

 # Add indentation
 a							=   ''' ''' * indent_level + line_to_add
 # Insert `pipeline_model_mapping` to `class_lines`.
 # (The line at `target_idx` should be kept by definition!)
 a							=   class_lines[: target_idx + 1] + [line_to_add] + class_lines[target_idx + 1 :]
 # Remove the lines that are marked to be removed
 a							=   [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)
 a							=   inspect.getsourcelines(inspect.getmodule(a				)				)[0]
 # Be careful with the 1-off between line numbers and array indices
 a							=   module_lines[: class_start_line_no - 1] + class_lines + module_lines[class_end_line_no:]
 a							=   ''''''.join(a				)

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

 return line_to_add
def 		_a					(   a				:Optional[int]			,							a				:List[str]=False				)							->    Optional[Any]:
 a							=   find_test_class(a				)
 if test_class:
  add_pipeline_model_mapping(a			,							overwrite=a				)


if __name__ == "__main__":
     UpperCAmelCase__            =	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`.",
     )
     UpperCAmelCase__            =	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.")

     UpperCAmelCase__            =	[]
     if args.test_file:
          UpperCAmelCase__            =	[args.test_file]
     else:
          UpperCAmelCase__            =	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)










import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding






class 	lowercase_      (			lowercase     ):






      '''simple docstring'''




      __snake_case  						=						['''image_processor''', '''tokenizer''']
      __snake_case  						=						'''AutoImageProcessor'''
      __snake_case  						=						'''AutoTokenizer'''







      def __init__(   self   :		Any				,      __UpperCAmelCase   :		Union[str, Any]=None				,      __UpperCAmelCase   :		int=None				,      **__UpperCAmelCase   :		Union[str, Any]						)					->Dict:




       """simple docstring"""


       a							=   None
       if "feature_extractor" in kwargs:
        warnings.warn(
            '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
            ''' instead.'''				,      __UpperCAmelCase				,      )
        a							=   kwargs.pop('''feature_extractor'''						)

       a							=   image_processor if image_processor is not None else feature_extractor
       if image_processor is None:
        raise ValueError('''You have to specify an image_processor.'''						)
       if tokenizer is None:
        raise ValueError('''You have to specify a tokenizer.'''						)

       super().__init__(__UpperCAmelCase				,      __UpperCAmelCase						)
       a							=   self.image_processor







      def __call__(   self   :		List[Any]				,      __UpperCAmelCase   :		Optional[Any]=None				,      __UpperCAmelCase   :		Optional[int]=None				,      __UpperCAmelCase   :		str=None				,      **__UpperCAmelCase   :		List[Any]						)					->Dict:




       """simple docstring"""


       if text is None and images is None:
        raise ValueError('''You have to specify either text or images. Both cannot be none.'''						)

       if text is not None:
        a							=   self.tokenizer(__UpperCAmelCase				,      return_tensors=__UpperCAmelCase				,      **__UpperCAmelCase						)

       if images is not None:
        a							=   self.image_processor(__UpperCAmelCase				,      return_tensors=__UpperCAmelCase				,      **__UpperCAmelCase						)

       if text is not None and images is not None:
        a							=   image_features.pixel_values
        return encoding
       elif text is not None:
        return encoding
       else:
        return BatchEncoding(data=dict(**__UpperCAmelCase						)				,      tensor_type=__UpperCAmelCase						)







      def    __lowerCAmelCase      (   self   :		List[Any]				,      *__UpperCAmelCase   :		Optional[Any]				,      **__UpperCAmelCase   :		List[str]						)					->Optional[Any]:




       """simple docstring"""


       return self.tokenizer.batch_decode(*__UpperCAmelCase				,      **__UpperCAmelCase						)







      def    __lowerCAmelCase      (   self   :		str				,      *__UpperCAmelCase   :		List[Any]				,      **__UpperCAmelCase   :		Optional[Any]						)					->Union[str, Any]:




       """simple docstring"""


       return self.tokenizer.decode(*__UpperCAmelCase				,      **__UpperCAmelCase						)







      @property
      def    __lowerCAmelCase      (   self   :		str						)					->str:




       """simple docstring"""


       a							=   self.tokenizer.model_input_names
       a							=   self.image_processor.model_input_names
       return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names						)						)







      @property
      def    __lowerCAmelCase      (   self   :		str						)					->Optional[Any]:




       """simple docstring"""


       warnings.warn(
           '''`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'''				,      __UpperCAmelCase				,      )
       return self.image_processor_class







      @property
      def    __lowerCAmelCase      (   self   :		int						)					->Optional[Any]:




       """simple docstring"""


       warnings.warn(
           '''`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'''				,      __UpperCAmelCase				,      )
       return self.image_processor


