



import inspect
from typing import List, Optional, Union

import torch
from torch import nn
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPImageProcessor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput


class 	snake_case_			(nn.Module	):




       """simple docstring"""






       def __init__(  self  ,lowercase  ,lowercase=1.0):

             """simple docstring"""





             super().__init__()

             UpperCAmelCase_	:						Union[str, Any]					   =		cut_size
             UpperCAmelCase_	:						str					   =		cut_power



       def 			A_ (  self  ,lowercase  ,lowercase):

             """simple docstring"""





             UpperCAmelCase_   ,   UpperCAmelCase_	:						Optional[int]					   =		pixel_values.shape[2:4]
             UpperCAmelCase_	:						Optional[int]					   =		min(__UpperCAmelCase  ,__UpperCAmelCase)
             UpperCAmelCase_	:						List[Any]					   =		min(__UpperCAmelCase  ,__UpperCAmelCase  ,self.cut_size)
             UpperCAmelCase_	:						str					   =		[]
             for _ in range(__UpperCAmelCase):
                   UpperCAmelCase_	:						int					   =		int(torch.rand([]) ** self.cut_power * (max_size - min_size) + min_size)
                   UpperCAmelCase_	:						Any					   =		torch.randint(0  ,sideX - size + 1  ,())
                   UpperCAmelCase_	:						Dict					   =		torch.randint(0  ,sideY - size + 1  ,())
                   UpperCAmelCase_	:						Dict					   =		pixel_values[:, :, offsety : offsety + size, offsetx : offsetx + size]
                   cutouts.append(F.adaptive_avg_poolad(__UpperCAmelCase  ,self.cut_size))
             return torch.cat(__UpperCAmelCase)







def  _snake_case			(							__snake_case       ,	__snake_case     )  ->  Tuple:






      '''simple docstring'''







      UpperCAmelCase_	:						Any					   =		F.normalize(__snake_case       ,	dim=-1     )
      UpperCAmelCase_	:						List[str]					   =		F.normalize(__snake_case       ,	dim=-1     )
      return (x - y).norm(dim=-1     ).div(2     ).arcsin().pow(2     ).mul(2     )







def  _snake_case			(							__snake_case       ,	__snake_case     )  ->  Tuple:






      '''simple docstring'''







      for param in model.parameters():
            UpperCAmelCase_	:						List[str]					   =		value





class 	snake_case_			(__SCREAMING_SNAKE_CASE	):




       """simple docstring"""






       def __init__(  self  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,):

             """simple docstring"""





             super().__init__()
             self.register_modules(
                 vae=__UpperCAmelCase  ,text_encoder=__UpperCAmelCase  ,clip_model=__UpperCAmelCase  ,tokenizer=__UpperCAmelCase  ,unet=__UpperCAmelCase  ,scheduler=__UpperCAmelCase  ,feature_extractor=__UpperCAmelCase  ,)

             UpperCAmelCase_	:						int					   =		transforms.Normalize(mean=feature_extractor.image_mean  ,std=feature_extractor.image_std)
             UpperCAmelCase_	:						Optional[Any]					   =		(
                 feature_extractor.size
                 if isinstance(feature_extractor.size  ,__UpperCAmelCase)
                 else feature_extractor.size["shortest_edge"]
             )
             UpperCAmelCase_	:						List[Any]					   =		MakeCutouts(self.cut_out_size)

             set_requires_grad(self.text_encoder  ,__UpperCAmelCase)
             set_requires_grad(self.clip_model  ,__UpperCAmelCase)



       def 			A_ (  self  ,lowercase = "auto"):

             """simple docstring"""





             if slice_size == "auto":
                   # half the attention head size is usually a good trade-off between
                   # speed and memory
                   UpperCAmelCase_	:						List[Any]					   =		self.unet.config.attention_head_dim // 2
             self.unet.set_attention_slice(__UpperCAmelCase)



       def 			A_ (  self):

             """simple docstring"""





             self.enable_attention_slicing(__UpperCAmelCase)



       def 			A_ (  self):

             """simple docstring"""





             set_requires_grad(self.vae  ,__UpperCAmelCase)



       def 			A_ (  self):

             """simple docstring"""





             set_requires_grad(self.vae  ,__UpperCAmelCase)



       def 			A_ (  self):

             """simple docstring"""





             set_requires_grad(self.unet  ,__UpperCAmelCase)



       def 			A_ (  self):

             """simple docstring"""





             set_requires_grad(self.unet  ,__UpperCAmelCase)



       @torch.enable_grad()
       def 			A_ (  self  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase=True  ,):

             """simple docstring"""





             UpperCAmelCase_	:						Dict					   =		latents.detach().requires_grad_()

             UpperCAmelCase_	:						Optional[int]					   =		self.scheduler.scale_model_input(__UpperCAmelCase  ,__UpperCAmelCase)

             # predict the noise residual
             UpperCAmelCase_	:						int					   =		self.unet(__UpperCAmelCase  ,__UpperCAmelCase  ,encoder_hidden_states=__UpperCAmelCase).sample

             if isinstance(self.scheduler  ,(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)):
                   UpperCAmelCase_	:						Tuple					   =		self.scheduler.alphas_cumprod[timestep]
                   UpperCAmelCase_	:						Optional[Any]					   =		1 - alpha_prod_t
                   # compute predicted original sample from predicted noise also called
                   # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
                   UpperCAmelCase_	:						Optional[int]					   =		(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

                   UpperCAmelCase_	:						Optional[int]					   =		torch.sqrt(__UpperCAmelCase)
                   UpperCAmelCase_	:						List[Any]					   =		pred_original_sample * (fac) + latents * (1 - fac)
             elif isinstance(self.scheduler  ,__UpperCAmelCase):
                   UpperCAmelCase_	:						Optional[int]					   =		self.scheduler.sigmas[index]
                   UpperCAmelCase_	:						Any					   =		latents - sigma * noise_pred
             else:
                   raise ValueError(F"""scheduler type {type(self.scheduler)} not supported""")

             UpperCAmelCase_	:						Optional[int]					   =		1 / self.vae.config.scaling_factor * sample
             UpperCAmelCase_	:						str					   =		self.vae.decode(__UpperCAmelCase).sample
             UpperCAmelCase_	:						List[str]					   =		(image / 2 + 0.5).clamp(0  ,1)

             if use_cutouts:
                   UpperCAmelCase_	:						str					   =		self.make_cutouts(__UpperCAmelCase  ,__UpperCAmelCase)
             else:
                   UpperCAmelCase_	:						str					   =		transforms.Resize(self.cut_out_size)(__UpperCAmelCase)
             UpperCAmelCase_	:						int					   =		self.normalize(__UpperCAmelCase).to(latents.dtype)

             UpperCAmelCase_	:						str					   =		self.clip_model.get_image_features(__UpperCAmelCase)
             UpperCAmelCase_	:						int					   =		image_embeddings_clip / image_embeddings_clip.norm(p=2  ,dim=-1  ,keepdim=__UpperCAmelCase)

             if use_cutouts:
                   UpperCAmelCase_	:						str					   =		spherical_dist_loss(__UpperCAmelCase  ,__UpperCAmelCase)
                   UpperCAmelCase_	:						Any					   =		dists.view([num_cutouts, sample.shape[0], -1])
                   UpperCAmelCase_	:						Optional[int]					   =		dists.sum(2).mean(0).sum() * clip_guidance_scale
             else:
                   UpperCAmelCase_	:						str					   =		spherical_dist_loss(__UpperCAmelCase  ,__UpperCAmelCase).mean() * clip_guidance_scale

             UpperCAmelCase_	:						Dict					   =		-torch.autograd.grad(__UpperCAmelCase  ,__UpperCAmelCase)[0]

             if isinstance(self.scheduler  ,__UpperCAmelCase):
                   UpperCAmelCase_	:						Optional[int]					   =		latents.detach() + grads * (sigma**2)
                   UpperCAmelCase_	:						List[Any]					   =		noise_pred_original
             else:
                   UpperCAmelCase_	:						int					   =		noise_pred_original - torch.sqrt(__UpperCAmelCase) * grads
             return noise_pred, latents



       @torch.no_grad()
       def __call__(  self  ,lowercase  ,lowercase = 512  ,lowercase = 512  ,lowercase = 50  ,lowercase = 7.5  ,lowercase = 1  ,lowercase = 0.0  ,lowercase = 100  ,lowercase = None  ,lowercase = 4  ,lowercase = True  ,lowercase = None  ,lowercase = None  ,lowercase = "pil"  ,lowercase = True  ,):

             """simple docstring"""





             if isinstance(__UpperCAmelCase  ,__UpperCAmelCase):
                   UpperCAmelCase_	:						int					   =		1
             elif isinstance(__UpperCAmelCase  ,__UpperCAmelCase):
                   UpperCAmelCase_	:						int					   =		len(__UpperCAmelCase)
             else:
                   raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(__UpperCAmelCase)}""")

             if height % 8 != 0 or width % 8 != 0:
                   raise ValueError(F"""`height` and `width` have to be divisible by 8 but are {height} and {width}.""")

             # get prompt text embeddings
             UpperCAmelCase_	:						Optional[Any]					   =		self.tokenizer(
                 __UpperCAmelCase  ,padding="max_length"  ,max_length=self.tokenizer.model_max_length  ,truncation=__UpperCAmelCase  ,return_tensors="pt"  ,)
             UpperCAmelCase_	:						List[Any]					   =		self.text_encoder(text_input.input_ids.to(self.device))[0]
             # duplicate text embeddings for each generation per prompt
             UpperCAmelCase_	:						List[Any]					   =		text_embeddings.repeat_interleave(__UpperCAmelCase  ,dim=0)

             if clip_guidance_scale > 0:
                   if clip_prompt is not None:
                         UpperCAmelCase_	:						Optional[Any]					   =		self.tokenizer(
                             __UpperCAmelCase  ,padding="max_length"  ,max_length=self.tokenizer.model_max_length  ,truncation=__UpperCAmelCase  ,return_tensors="pt"  ,).input_ids.to(self.device)
                   else:
                         UpperCAmelCase_	:						Any					   =		text_input.input_ids.to(self.device)
                   UpperCAmelCase_	:						List[Any]					   =		self.clip_model.get_text_features(__UpperCAmelCase)
                   UpperCAmelCase_	:						Optional[Any]					   =		text_embeddings_clip / text_embeddings_clip.norm(p=2  ,dim=-1  ,keepdim=__UpperCAmelCase)
                   # duplicate text embeddings clip for each generation per prompt
                   UpperCAmelCase_	:						str					   =		text_embeddings_clip.repeat_interleave(__UpperCAmelCase  ,dim=0)

             # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
             # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
             # corresponds to doing no classifier free guidance.
             UpperCAmelCase_	:						Dict					   =		guidance_scale > 1.0
             # get unconditional embeddings for classifier free guidance
             if do_classifier_free_guidance:
                   UpperCAmelCase_	:						Tuple					   =		text_input.input_ids.shape[-1]
                   UpperCAmelCase_	:						str					   =		self.tokenizer([""]  ,padding="max_length"  ,max_length=__UpperCAmelCase  ,return_tensors="pt")
                   UpperCAmelCase_	:						List[str]					   =		self.text_encoder(uncond_input.input_ids.to(self.device))[0]
                   # duplicate unconditional embeddings for each generation per prompt
                   UpperCAmelCase_	:						Any					   =		uncond_embeddings.repeat_interleave(__UpperCAmelCase  ,dim=0)

                   # For classifier free guidance, we need to do two forward passes.
                   # Here we concatenate the unconditional and text embeddings into a single batch
                   # to avoid doing two forward passes
                   UpperCAmelCase_	:						Dict					   =		torch.cat([uncond_embeddings, text_embeddings])

             # get the initial random noise unless the user supplied it

             # Unlike in other pipelines, latents need to be generated in the target device
             # for 1-to-1 results reproducibility with the CompVis implementation.
             # However this currently doesn't work in `mps`.
             UpperCAmelCase_	:						List[str]					   =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
             UpperCAmelCase_	:						int					   =		text_embeddings.dtype
             if latents is None:
                   if self.device.type == "mps":
                         # randn does not work reproducibly on mps
                         UpperCAmelCase_	:						str					   =		torch.randn(__UpperCAmelCase  ,generator=__UpperCAmelCase  ,device="cpu"  ,dtype=__UpperCAmelCase).to(
                             self.device)
                   else:
                         UpperCAmelCase_	:						Union[str, Any]					   =		torch.randn(__UpperCAmelCase  ,generator=__UpperCAmelCase  ,device=self.device  ,dtype=__UpperCAmelCase)
             else:
                   if latents.shape != latents_shape:
                         raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}""")
                   UpperCAmelCase_	:						Tuple					   =		latents.to(self.device)

             # set timesteps
             UpperCAmelCase_	:						Optional[Any]					   =		"offset" in set(inspect.signature(self.scheduler.set_timesteps).parameters.keys())
             UpperCAmelCase_	:						str					   =		{}
             if accepts_offset:
                   UpperCAmelCase_	:						Tuple					   =		1

             self.scheduler.set_timesteps(__UpperCAmelCase  ,**__UpperCAmelCase)

             # Some schedulers like PNDM have timesteps as arrays
             # It's more optimized to move all timesteps to correct device beforehand
             UpperCAmelCase_	:						List[Any]					   =		self.scheduler.timesteps.to(self.device)

             # scale the initial noise by the standard deviation required by the scheduler
             UpperCAmelCase_	:						Optional[Any]					   =		latents * self.scheduler.init_noise_sigma

             # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
             # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
             # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
             # and should be between [0, 1]
             UpperCAmelCase_	:						Any					   =		"eta" in set(inspect.signature(self.scheduler.step).parameters.keys())
             UpperCAmelCase_	:						Optional[Any]					   =		{}
             if accepts_eta:
                   UpperCAmelCase_	:						str					   =		eta

             # check if the scheduler accepts generator
             UpperCAmelCase_	:						str					   =		"generator" in set(inspect.signature(self.scheduler.step).parameters.keys())
             if accepts_generator:
                   UpperCAmelCase_	:						int					   =		generator

             for i, t in enumerate(self.progress_bar(__UpperCAmelCase)):
                   # expand the latents if we are doing classifier free guidance
                   UpperCAmelCase_	:						str					   =		torch.cat([latents] * 2) if do_classifier_free_guidance else latents
                   UpperCAmelCase_	:						str					   =		self.scheduler.scale_model_input(__UpperCAmelCase  ,__UpperCAmelCase)

                   # predict the noise residual
                   UpperCAmelCase_	:						int					   =		self.unet(__UpperCAmelCase  ,__UpperCAmelCase  ,encoder_hidden_states=__UpperCAmelCase).sample

                   # perform classifier free guidance
                   if do_classifier_free_guidance:
                         UpperCAmelCase_   ,   UpperCAmelCase_	:						Optional[int]					   =		noise_pred.chunk(2)
                         UpperCAmelCase_	:						Optional[Any]					   =		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                   # perform clip guidance
                   if clip_guidance_scale > 0:
                         UpperCAmelCase_	:						str					   =		(
                             text_embeddings.chunk(2)[1] if do_classifier_free_guidance else text_embeddings
                         )
                         UpperCAmelCase_   ,   UpperCAmelCase_	:						Optional[int]					   =		self.cond_fn(
                             __UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,)

                   # compute the previous noisy sample x_t -> x_t-1
                   UpperCAmelCase_	:						Any					   =		self.scheduler.step(__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase).prev_sample

             # scale and decode the image latents with vae
             UpperCAmelCase_	:						Union[str, Any]					   =		1 / self.vae.config.scaling_factor * latents
             UpperCAmelCase_	:						Optional[Any]					   =		self.vae.decode(__UpperCAmelCase).sample

             UpperCAmelCase_	:						Dict					   =		(image / 2 + 0.5).clamp(0  ,1)
             UpperCAmelCase_	:						List[str]					   =		image.cpu().permute(0  ,2  ,3  ,1).numpy()

             if output_type == "pil":
                   UpperCAmelCase_	:						Optional[int]					   =		self.numpy_to_pil(__UpperCAmelCase)

             if not return_dict:
                   return (image, None)

             return StableDiffusionPipelineOutput(images=__UpperCAmelCase  ,nsfw_content_detected=__UpperCAmelCase)


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


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


# DO **NOT** add item to this set (unless the reason is approved)
UpperCamelCase      :							int          =					{
    "tests/models/esm/test_modeling_esmfold.py",  # The pipeline test mapping is added to `test_modeling_esm.py`
}






def 				A		(							snake_case :List[str]						) ->      Optional[Any]:

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

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

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

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

 # cache the results
 __UpperCamelCase							    =			mapping
 return mapping






def 				A		(							snake_case :int							,   snake_case :Tuple						) ->      List[Any]:
 __UpperCamelCase							    =			get_framework(snake_case						)
 if framework is None:
  return None
 __UpperCamelCase							    =			get_mapping_for_task(snake_case							,   snake_case						)
 if mapping is None:
  return None

 __UpperCamelCase							    =			list({model_class.config_class for model_class in test_class.all_model_classes}						)
 if len(snake_case						) != 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.
 __UpperCamelCase							    =			mapping.get(config_classes[0]							,   snake_case						)
 if isinstance(snake_case							,   (tuple, list)						):
  __UpperCamelCase							    =			sorted(snake_case							,   key=lambda snake_case						: x.__name__						)

 return model_class






def 				A		(							snake_case :List[Any]						) ->      Any:
 __UpperCamelCase							    =			[(task, get_model_for_pipeline_test(snake_case							,   snake_case						)) for task in pipeline_test_mapping]
 __UpperCamelCase							    =			sorted([(task, model) for task, model in mapping if model is not None]							,   key=lambda snake_case						: x[0]						)

 return dict(snake_case						)






def 				A		(							snake_case :Optional[int]						) ->      List[str]:
 __UpperCamelCase							    =			get_framework(snake_case						)
 if framework == "pt":
  __UpperCamelCase							    =			'torch'
 __UpperCamelCase							    =			'{}'

 __UpperCamelCase							    =			get_pipeline_model_mapping(snake_case						)
 if len(snake_case						) == 0:
  return ""

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

 return text






def 				A		(							snake_case :Optional[int]						) ->      Optional[int]:
 __UpperCamelCase							    =			{'ModelTesterMixin', 'TFModelTesterMixin', 'FlaxModelTesterMixin'}
 if not issubclass(snake_case							,   unittest.TestCase						):
  return False
 return len(base_class_names.intersection([x.__name__ for x in test_class.__bases__]						)						) > 0






def 				A		(							snake_case :Dict						) ->      Tuple:
 __UpperCamelCase							    =			[x for x in get_test_classes(snake_case						) if is_valid_test_class(snake_case						)]

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

 return target_test_class






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

 return end_idx






def 				A		(							snake_case :List[str]							,   snake_case :str=False						) ->      List[str]:
 if getattr(snake_case							,   'pipeline_model_mapping'							,   snake_case						) is not None:
  if not overwrite:
   return "", -1

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

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

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

 # 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
 __UpperCamelCase							    =			re.compile(r'\s(is_\S+?_available\(\))\s'						)
 for line in class_lines[start_idx : end_idx + 1]:
  __UpperCamelCase							    =			r.search(snake_case						)
  if backend_condition is not None:
   # replace the leading and trailing whitespaces to the space character " ".
   __UpperCamelCase							    =			' ' + backend_condition[0][1:-1] + ' '
   __UpperCamelCase							    =			r.sub(snake_case							,   snake_case						)
   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`.
  __UpperCamelCase							    =			end_idx
 else:
  # `pipeline_model_mapping` is defined. The target index is set to be one **BEFORE** its start index.
  __UpperCamelCase							    =			start_idx - 1
  # mark the lines of the currently existing `pipeline_model_mapping` to be removed.
  for idx in range(snake_case							,   end_idx + 1						):
   # These lines are going to be removed before writing to the test file.
   __UpperCamelCase							    =			None  # noqa

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

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

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

 return line_to_add






def 				A		(							snake_case :int							,   snake_case :str=False						) ->      Dict:
 __UpperCamelCase							    =			find_test_class(snake_case						)
 if test_class:
  add_pipeline_model_mapping(snake_case							,   overwrite=snake_case						)


if __name__ == "__main__":
      UpperCamelCase      :							Optional[int]          =					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      :							int          =					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      :							List[Any]          =					[]
      if args.test_file:
            UpperCamelCase      :							Union[str, Any]          =					[args.test_file]
      else:
            UpperCamelCase      :							List[Any]          =					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)



