






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


lowerCAmelCase_		:			List[Any]				   =					logging.get_logger(__name__)

lowerCAmelCase_		:			List[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'''),
    ]
)

lowerCAmelCase_		:			Optional[int]				   =					_LazyAutoMapping(CONFIG_MAPPING_NAMES, PROCESSOR_MAPPING_NAMES)






def 					_lowerCAmelCase				(					lowerCAmelCase						):







       '''simple docstring'''
       for module_name, processors in PROCESSOR_MAPPING_NAMES.items():
              if class_name in processors:
                     UpperCAmelCase					=	model_type_to_module_name(lowerCAmelCase						)

                     UpperCAmelCase					=	importlib.import_module(F'''.{module_name}'''		,					"""transformers.models"""						)
                     try:
                            return getattr(lowerCAmelCase		,					lowerCAmelCase						)
                     except AttributeError:
                            continue

       for processor in PROCESSOR_MAPPING._extra_content.values():
              if getattr(lowerCAmelCase		,					"""__name__"""		,					lowerCAmelCase						) == 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.
       UpperCAmelCase					=	importlib.import_module("""transformers"""						)
       if hasattr(lowerCAmelCase		,					lowerCAmelCase						):
              return getattr(lowerCAmelCase		,					lowerCAmelCase						)

       return None





class      UpperCamelCase_       :



  def __init__(					self )	->	List[Any]:



         """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(snake_case__ )
  def 						UpperCamelCase_ (					cls       ,						snake_case__       ,						**snake_case__ )	->	str:



         """simple docstring"""





         UpperCAmelCase					=	kwargs.pop("""config"""       ,						snake_case__ )
         UpperCAmelCase					=	kwargs.pop("""trust_remote_code"""       ,						snake_case__ )
         UpperCAmelCase					=	True

         UpperCAmelCase					=	None
         UpperCAmelCase					=	None

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

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

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

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

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

         if processor_class is not None:
                UpperCAmelCase					=	processor_class_from_name(snake_case__ )

         UpperCAmelCase					=	processor_auto_map is not None
         UpperCAmelCase					=	processor_class is not None or type(snake_case__ ) in PROCESSOR_MAPPING
         UpperCAmelCase					=	resolve_trust_remote_code(
             snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__ )

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

         # 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(
                    snake_case__       ,						trust_remote_code=snake_case__       ,						**snake_case__ )
         except Exception:
                try:
                       return AutoImageProcessor.from_pretrained(
                           snake_case__       ,						trust_remote_code=snake_case__       ,						**snake_case__ )
                except Exception:
                       pass

                try:
                       return AutoFeatureExtractor.from_pretrained(
                           snake_case__       ,						trust_remote_code=snake_case__       ,						**snake_case__ )
                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 						UpperCamelCase_ (					snake_case__       ,						snake_case__ )	->	Any:



         """simple docstring"""





         PROCESSOR_MAPPING.register(snake_case__       ,						snake_case__ )








"""simple docstring"""







import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





class      UpperCamelCase_       (				a_      ,    a_      ,    unittest.TestCase     ):
  _A				:    str =		StableDiffusionParadigmsPipeline
  _A				:    Any =		TEXT_TO_IMAGE_PARAMS
  _A				:    List[Any] =		TEXT_TO_IMAGE_BATCH_PARAMS
  _A				:    str =		TEXT_TO_IMAGE_IMAGE_PARAMS
  _A				:    Tuple =		TEXT_TO_IMAGE_IMAGE_PARAMS



  def 						UpperCamelCase_ (					self )	->	Dict:



         """simple docstring"""





         torch.manual_seed(0 )
         UpperCAmelCase					=	UNetaDConditionModel(
             block_out_channels=(32, 64)       ,						layers_per_block=2       ,						sample_size=32       ,						in_channels=4       ,						out_channels=4       ,						down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")       ,						up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")       ,						cross_attention_dim=32       ,						attention_head_dim=(2, 4)       ,						use_linear_projection=snake_case__       ,						)
         UpperCAmelCase					=	DDIMParallelScheduler(
             beta_start=0.00_085       ,						beta_end=0.012       ,						beta_schedule="""scaled_linear"""       ,						clip_sample=snake_case__       ,						set_alpha_to_one=snake_case__       ,						)
         torch.manual_seed(0 )
         UpperCAmelCase					=	AutoencoderKL(
             block_out_channels=[32, 64]       ,						in_channels=3       ,						out_channels=3       ,						down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]       ,						up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]       ,						latent_channels=4       ,						sample_size=1_28       ,						)
         torch.manual_seed(0 )
         UpperCAmelCase					=	CLIPTextConfig(
             bos_token_id=0       ,						eos_token_id=2       ,						hidden_size=32       ,						intermediate_size=37       ,						layer_norm_eps=1e-05       ,						num_attention_heads=4       ,						num_hidden_layers=5       ,						pad_token_id=1       ,						vocab_size=10_00       ,						hidden_act="""gelu"""       ,						projection_dim=5_12       ,						)
         UpperCAmelCase					=	CLIPTextModel(snake_case__ )
         UpperCAmelCase					=	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" )

         UpperCAmelCase					=	{
             """unet""": unet,
             """scheduler""": scheduler,
             """vae""": vae,
             """text_encoder""": text_encoder,
             """tokenizer""": tokenizer,
             """safety_checker""": None,
             """feature_extractor""": None,
         }
         return components



  def 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__=0 )	->	Any:



         """simple docstring"""





         if str(snake_case__ ).startswith("""mps""" ):
                UpperCAmelCase					=	torch.manual_seed(snake_case__ )
         else:
                UpperCAmelCase					=	torch.Generator(device=snake_case__ ).manual_seed(snake_case__ )
         UpperCAmelCase					=	{
             """prompt""": """a photograph of an astronaut riding a horse""",
             """generator""": generator,
             """num_inference_steps""": 10,
             """guidance_scale""": 6.0,
             """output_type""": """numpy""",
             """parallel""": 3,
             """debug""": True,
         }
         return inputs



  def 						UpperCamelCase_ (					self )	->	Union[str, Any]:



         """simple docstring"""





         UpperCAmelCase					=	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         UpperCAmelCase					=	self.get_dummy_components()
         UpperCAmelCase					=	StableDiffusionParadigmsPipeline(**snake_case__ )
         UpperCAmelCase					=	sd_pipe.to(snake_case__ )
         sd_pipe.set_progress_bar_config(disable=snake_case__ )

         UpperCAmelCase					=	self.get_dummy_inputs(snake_case__ )
         UpperCAmelCase					=	sd_pipe(**snake_case__ ).images
         UpperCAmelCase					=	image[0, -3:, -3:, -1]
         assert image.shape == (1, 64, 64, 3)

         UpperCAmelCase					=	np.array([0.4_773, 0.5_417, 0.4_723, 0.4_925, 0.5_631, 0.4_752, 0.5_240, 0.4_935, 0.5_023] )

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1e-2



  def 						UpperCamelCase_ (					self )	->	List[Any]:



         """simple docstring"""





         UpperCAmelCase					=	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         UpperCAmelCase					=	self.get_dummy_components()
         torch.manual_seed(0 )
         UpperCAmelCase					=	DDPMParallelScheduler()
         torch.manual_seed(0 )
         UpperCAmelCase					=	StableDiffusionParadigmsPipeline(**snake_case__ )
         UpperCAmelCase					=	sd_pipe.to(snake_case__ )
         sd_pipe.set_progress_bar_config(disable=snake_case__ )

         UpperCAmelCase					=	self.get_dummy_inputs(snake_case__ )
         UpperCAmelCase					=	sd_pipe(**snake_case__ ).images
         UpperCAmelCase					=	image[0, -3:, -3:, -1]
         assert image.shape == (1, 64, 64, 3)

         UpperCAmelCase					=	np.array([0.3_573, 0.4_420, 0.4_960, 0.4_799, 0.3_796, 0.3_879, 0.4_819, 0.4_365, 0.4_468] )

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1e-2



  def 						UpperCamelCase_ (					self )	->	Tuple:



         """simple docstring"""





         super().test_inference_batch_consistent(batch_sizes=[1, 2] )



  def 						UpperCamelCase_ (					self )	->	Optional[int]:



         """simple docstring"""





         super().test_inference_batch_single_identical(batch_size=2       ,						expected_max_diff=3e-3 )



  def 						UpperCamelCase_ (					self )	->	Union[str, Any]:



         """simple docstring"""





         UpperCAmelCase					=	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         UpperCAmelCase					=	self.get_dummy_components()
         UpperCAmelCase					=	StableDiffusionParadigmsPipeline(**snake_case__ )
         UpperCAmelCase					=	sd_pipe.to(snake_case__ )
         sd_pipe.set_progress_bar_config(disable=snake_case__ )

         UpperCAmelCase					=	self.get_dummy_inputs(snake_case__ )
         UpperCAmelCase					=	"""french fries"""
         UpperCAmelCase					=	sd_pipe(**snake_case__       ,						negative_prompt=snake_case__ )
         UpperCAmelCase					=	output.images
         UpperCAmelCase					=	image[0, -3:, -3:, -1]

         assert image.shape == (1, 64, 64, 3)

         UpperCAmelCase					=	np.array([0.4_771, 0.5_420, 0.4_683, 0.4_918, 0.5_636, 0.4_725, 0.5_230, 0.4_923, 0.5_015] )

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1e-2





@slow
@require_torch_gpu
class      UpperCamelCase_       (				unittest.TestCase     ):



  def 						UpperCamelCase_ (					self )	->	Union[str, Any]:



         """simple docstring"""





         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()



  def 						UpperCamelCase_ (					self       ,						snake_case__=0 )	->	int:



         """simple docstring"""





         UpperCAmelCase					=	torch.Generator(device=snake_case__ ).manual_seed(snake_case__ )
         UpperCAmelCase					=	{
             """prompt""": """a photograph of an astronaut riding a horse""",
             """generator""": generator,
             """num_inference_steps""": 10,
             """guidance_scale""": 7.5,
             """output_type""": """numpy""",
             """parallel""": 3,
             """debug""": True,
         }
         return inputs



  def 						UpperCamelCase_ (					self )	->	Tuple:



         """simple docstring"""





         UpperCAmelCase					=	"""stabilityai/stable-diffusion-2-base"""
         UpperCAmelCase					=	DDIMParallelScheduler.from_pretrained(snake_case__       ,						subfolder="""scheduler""" )
         UpperCAmelCase					=	StableDiffusionParadigmsPipeline.from_pretrained(snake_case__       ,						scheduler=snake_case__       ,						safety_checker=snake_case__ )
         pipe.to(snake_case__ )
         pipe.set_progress_bar_config(disable=snake_case__ )
         pipe.enable_attention_slicing()

         UpperCAmelCase					=	self.get_inputs()
         UpperCAmelCase					=	pipe(**snake_case__ ).images
         UpperCAmelCase					=	image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 5_12, 5_12, 3)

         UpperCAmelCase					=	np.array([0.9_622, 0.9_602, 0.9_748, 0.9_591, 0.9_630, 0.9_691, 0.9_661, 0.9_631, 0.9_741] )

         assert np.abs(expected_slice - image_slice ).max() < 1e-2
