
"""simple docstring"""
import random
import unittest

import numpy as np
import torch
from PIL import Image
from torch import nn
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import KandinskyVaaPriorEmbaEmbPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import floats_tensor, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()






class   _UpperCAmelCase					(       _snake_case		,       unittest.TestCase):
    __lowercase				:			Union[str, Any]	=       KandinskyVaaPriorEmbaEmbPipeline
    __lowercase				:			Optional[Any]	=       ["""prompt""", """image"""]
    __lowercase				:			List[str]	=       ["""prompt""", """image"""]
    __lowercase				:			int	=       [
        """num_images_per_prompt""",
        """strength""",
        """generator""",
        """num_inference_steps""",
        """latents""",
        """negative_prompt""",
        """guidance_scale""",
        """output_type""",
        """return_dict""",
    ]
    __lowercase				:			Optional[Any]	=       False
    @property
    def        lowerCamelCase__						(     self  ):
      return 32
    @property
    def        lowerCamelCase__						(     self  ):
      return 32
    @property
    def        lowerCamelCase__						(     self  ):
      return self.time_input_dim
    @property
    def        lowerCamelCase__						(     self  ):
      return self.time_input_dim * 4
    @property
    def        lowerCamelCase__						(     self  ):
      return 1_00
    @property
    def        lowerCamelCase__						(     self  ):
      _snake_case  :    Dict            = CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"  )
      return tokenizer
    @property
    def        lowerCamelCase__						(     self  ):
      torch.manual_seed(0  )
      _snake_case  :    Tuple            = CLIPTextConfig(
          bos_token_id=0    ,     eos_token_id=2    ,     hidden_size=self.text_embedder_hidden_size    ,     projection_dim=self.text_embedder_hidden_size    ,     intermediate_size=37    ,     layer_norm_eps=1E-05    ,     num_attention_heads=4    ,     num_hidden_layers=5    ,     pad_token_id=1    ,     vocab_size=10_00    ,     )
      return CLIPTextModelWithProjection(snake_case_  )
    @property
    def        lowerCamelCase__						(     self  ):
      torch.manual_seed(0  )

      _snake_case  :    Tuple            = {
          "num_attention_heads": 2,
          "attention_head_dim": 12,
          "embedding_dim": self.text_embedder_hidden_size,
          "num_layers": 1,
      }

      _snake_case  :    Optional[int]            = PriorTransformer(**snake_case_  )
      # clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
      _snake_case  :    Optional[int]            = nn.Parameter(torch.ones(model.clip_std.shape  )  )
      return model
    @property
    def        lowerCamelCase__						(     self  ):
      torch.manual_seed(0  )
      _snake_case  :    int            = CLIPVisionConfig(
          hidden_size=self.text_embedder_hidden_size    ,     image_size=2_24    ,     projection_dim=self.text_embedder_hidden_size    ,     intermediate_size=37    ,     num_attention_heads=4    ,     num_channels=3    ,     num_hidden_layers=5    ,     patch_size=14    ,     )

      _snake_case  :    Optional[Any]            = CLIPVisionModelWithProjection(snake_case_  )
      return model
    @property
    def        lowerCamelCase__						(     self  ):
      _snake_case  :    Optional[Any]            = CLIPImageProcessor(
          crop_size=2_24    ,     do_center_crop=snake_case_    ,     do_normalize=snake_case_    ,     do_resize=snake_case_    ,     image_mean=[0.48145466, 0.4578275, 0.40821073]    ,     image_std=[0.26862954, 0.26130258, 0.27577711]    ,     resample=3    ,     size=2_24    ,     )

      return image_processor
    def        lowerCamelCase__						(     self  ):
      _snake_case  :    Optional[int]            = self.dummy_prior
      _snake_case  :    str            = self.dummy_image_encoder
      _snake_case  :    Any            = self.dummy_text_encoder
      _snake_case  :    List[str]            = self.dummy_tokenizer
      _snake_case  :    List[str]            = self.dummy_image_processor

      _snake_case  :    Tuple            = UnCLIPScheduler(
          variance_type="fixed_small_log"    ,     prediction_type="sample"    ,     num_train_timesteps=10_00    ,     clip_sample=snake_case_    ,     clip_sample_range=10.0    ,     )

      _snake_case  :    List[Any]            = {
          "prior": prior,
          "image_encoder": image_encoder,
          "text_encoder": text_encoder,
          "tokenizer": tokenizer,
          "scheduler": scheduler,
          "image_processor": image_processor,
      }

      return components
    def        lowerCamelCase__						(     self    ,     snake_case_    ,     snake_case_=0  ):
      if str(snake_case_  ).startswith("mps"  ):
        _snake_case  :    str            = torch.manual_seed(snake_case_  )
      else:
        _snake_case  :    List[str]            = torch.Generator(device=snake_case_  ).manual_seed(snake_case_  )

      _snake_case  :    Union[str, Any]            = floats_tensor((1, 3, 64, 64)    ,     rng=random.Random(snake_case_  )  ).to(snake_case_  )
      _snake_case  :    str            = image.cpu().permute(0    ,     2    ,     3    ,     1  )[0]
      _snake_case  :    Union[str, Any]            = Image.fromarray(np.uinta(snake_case_  )  ).convert("RGB"  ).resize((2_56, 2_56)  )

      _snake_case  :    Optional[Any]            = {
          "prompt": "horse",
          "image": init_image,
          "strength": 0.5,
          "generator": generator,
          "guidance_scale": 4.0,
          "num_inference_steps": 2,
          "output_type": "np",
      }
      return inputs
    def        lowerCamelCase__						(     self  ):
      _snake_case  :    Tuple            = "cpu"

      _snake_case  :    str            = self.get_dummy_components()

      _snake_case  :    List[str]            = self.pipeline_class(**snake_case_  )
      _snake_case  :    List[Any]            = pipe.to(snake_case_  )

      pipe.set_progress_bar_config(disable=snake_case_  )

      _snake_case  :    Dict            = pipe(**self.get_dummy_inputs(snake_case_  )  )
      _snake_case  :    Union[str, Any]            = output.image_embeds

      _snake_case  :    Optional[int]            = pipe(
          **self.get_dummy_inputs(snake_case_  )    ,     return_dict=snake_case_    ,     )[0]

      _snake_case  :    str            = image[0, -10:]
      _snake_case  :    Optional[Any]            = image_from_tuple[0, -10:]

      assert image.shape == (1, 32)

      _snake_case  :    List[Any]            = np.array(
          [
              0.1071284,
              1.3330271,
              0.61260223,
              -0.6691065,
              -0.3846852,
              -1.0303661,
              0.22716111,
              0.03348901,
              0.30040675,
              -0.24805029,
          ]  )

      assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1E-2
      assert np.abs(image_from_tuple_slice.flatten() - expected_slice  ).max() < 1E-2
    @skip_mps
    def        lowerCamelCase__						(     self  ):
      _snake_case  :    List[Any]            = torch_device == "cpu"
      _snake_case  :    Union[str, Any]            = True
      _snake_case  :    Any            = False

      self._test_inference_batch_single_identical(
          test_max_difference=snake_case_    ,     relax_max_difference=snake_case_    ,     test_mean_pixel_difference=snake_case_    ,     )


    @skip_mps
    def        lowerCamelCase__						(     self  ):
      _snake_case  :    Union[str, Any]            = torch_device == "cpu"
      _snake_case  :    Optional[int]            = False

      self._test_attention_slicing_forward_pass(
          test_max_difference=snake_case_    ,     test_mean_pixel_difference=snake_case_    ,     )



"""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   :      Tuple       =				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   :      Any       =				_LazyAutoMapping(CONFIG_MAPPING_NAMES, PROCESSOR_MAPPING_NAMES)
def 					a__ ( a :					str		):

  """simple docstring"""







  for module_name, processors in PROCESSOR_MAPPING_NAMES.items():
    if class_name in processors:
      _snake_case  :    Dict            = model_type_to_module_name(a		)

      _snake_case  :    Optional[int]            = 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.
  _snake_case  :    int            = importlib.import_module("transformers"		)
  if hasattr(a							,  a		):
    return getattr(a							,  a		)

  return None






class   _UpperCAmelCase					:
    def __init__(     self  ):
      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        lowerCamelCase__						(     cls    ,     snake_case_    ,     **snake_case_  ):
      _snake_case  :    str            = kwargs.pop("config"    ,     snake_case_  )
      _snake_case  :    Dict            = kwargs.pop("trust_remote_code"    ,     snake_case_  )
      _snake_case  :    Tuple            = True

      _snake_case  :    List[str]            = None
      _snake_case  :    Optional[Any]            = None

      # First, let's see if we have a preprocessor config.
      # Filter the kwargs for `get_file_from_repo`.
      _snake_case  :    Optional[int]            = {
          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
      _snake_case  :    Dict            = get_file_from_repo(
          snake_case_    ,     snake_case_    ,     **snake_case_  )
      if preprocessor_config_file is not None:
        _snake_case      ,       _snake_case  :    List[str]            = ImageProcessingMixin.get_image_processor_dict(snake_case_    ,     **snake_case_  )
        _snake_case  :    Union[str, Any]            = config_dict.get("processor_class"    ,     snake_case_  )
        if "AutoProcessor" in config_dict.get("auto_map"    ,     {}  ):
          _snake_case  :    Optional[int]            = 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:
        _snake_case      ,       _snake_case  :    Any            = FeatureExtractionMixin.get_feature_extractor_dict(snake_case_    ,     **snake_case_  )
        _snake_case  :    List[Any]            = config_dict.get("processor_class"    ,     snake_case_  )
        if "AutoProcessor" in config_dict.get("auto_map"    ,     {}  ):
          _snake_case  :    int            = config_dict["auto_map"]["AutoProcessor"]

      if processor_class is None:
        # Next, let's check whether the processor class is saved in a tokenizer
        _snake_case  :    Union[str, Any]            = 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:
            _snake_case  :    Tuple            = json.load(snake_case_  )

          _snake_case  :    List[Any]            = config_dict.get("processor_class"    ,     snake_case_  )
          if "AutoProcessor" in config_dict.get("auto_map"    ,     {}  ):
            _snake_case  :    Tuple            = 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_  ):
          _snake_case  :    Union[str, Any]            = AutoConfig.from_pretrained(
              snake_case_    ,     trust_remote_code=snake_case_    ,     **snake_case_  )

        # And check if the config contains the processor class.
        _snake_case  :    List[Any]            = getattr(snake_case_    ,     "processor_class"    ,     snake_case_  )
        if hasattr(snake_case_    ,     "auto_map"  ) and "AutoProcessor" in config.auto_map:
          _snake_case  :    str            = config.auto_map["AutoProcessor"]

      if processor_class is not None:
        _snake_case  :    Optional[int]            = processor_class_from_name(snake_case_  )

      _snake_case  :    Optional[int]            = processor_auto_map is not None
      _snake_case  :    Any            = processor_class is not None or type(snake_case_  ) in PROCESSOR_MAPPING
      _snake_case  :    Union[str, Any]            = resolve_trust_remote_code(
          snake_case_    ,     snake_case_    ,     snake_case_    ,     snake_case_  )

      if has_remote_code and trust_remote_code:
        _snake_case  :    Dict            = get_class_from_dynamic_module(
            snake_case_    ,     snake_case_    ,     **snake_case_  )
        _snake_case  :    Union[str, Any]            = 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        lowerCamelCase__						(     snake_case_    ,     snake_case_  ):
      PROCESSOR_MAPPING.register(snake_case_    ,     snake_case_  )

