




import random
import unittest

import numpy as np

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionImgaImgPipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
  import onnxruntime as ort







class 						lowercase							(   UpperCamelCase__,unittest.TestCase		):
     _a							 =						"hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"

     def 					a__	(  self		,     _a=0				)						->							Optional[Any]:
           _A							:     Tuple			=						floats_tensor((1, 3, 128, 128)		,     rng=random.Random(_a				)				)
           _A							:     Optional[Any]			=						np.random.RandomState(_a				)
           _A							:     Optional[Any]			=						{
               """prompt""": """A painting of a squirrel eating a burger""",
               """image""": image,
               """generator""": generator,
               """num_inference_steps""": 3,
               """strength""": 0.75,
               """guidance_scale""": 7.5,
               """output_type""": """numpy""",
           }
           return inputs

     def 					a__	(  self				)						->							int:
           _A							:     int			=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,     provider="""CPUExecutionProvider"""				)
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     Any			=						self.get_dummy_inputs()
           _A							:     List[str]			=						pipe(**_a				).images
           _A							:     Optional[Any]			=						image[0, -3:, -3:, -1].flatten()

           assert image.shape == (1, 128, 128, 3)
           _A							:     List[Any]			=						np.array([0.69643, 0.58484, 0.50314, 0.58760, 0.55368, 0.59643, 0.51529, 0.41217, 0.49087]				)
           assert np.abs(image_slice - expected_slice				).max() < 1e-1

     def 					a__	(  self				)						->							Dict:
           _A							:     List[Any]			=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,     provider="""CPUExecutionProvider"""				)
           _A							:     List[Any]			=						PNDMScheduler.from_config(pipe.scheduler.config		,     skip_prk_steps=_a				)
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     List[Any]			=						self.get_dummy_inputs()
           _A							:     Union[str, Any]			=						pipe(**_a				).images
           _A							:     Optional[int]			=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 128, 128, 3)
           _A							:     Dict			=						np.array([0.61737, 0.54642, 0.53183, 0.54465, 0.52742, 0.60525, 0.49969, 0.40655, 0.48154]				)

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

     def 					a__	(  self				)						->							List[Any]:
           _A							:     Dict			=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,     provider="""CPUExecutionProvider"""				)
           _A							:     Tuple			=						LMSDiscreteScheduler.from_config(pipe.scheduler.config				)
           pipe.set_progress_bar_config(disable=_a				)

           # warmup pass to apply optimizations
           _A							:     List[Any]			=						pipe(**self.get_dummy_inputs()				)

           _A							:     Tuple			=						self.get_dummy_inputs()
           _A							:     List[str]			=						pipe(**_a				).images
           _A							:     Optional[Any]			=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 128, 128, 3)
           _A							:     Dict			=						np.array([0.52761, 0.59977, 0.49033, 0.49619, 0.54282, 0.50311, 0.47600, 0.40918, 0.45203]				)

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

     def 					a__	(  self				)						->							Union[str, Any]:
           _A							:     str			=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,     provider="""CPUExecutionProvider"""				)
           _A							:     Any			=						EulerDiscreteScheduler.from_config(pipe.scheduler.config				)
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     int			=						self.get_dummy_inputs()
           _A							:     Tuple			=						pipe(**_a				).images
           _A							:     Any			=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 128, 128, 3)
           _A							:     Any			=						np.array([0.52911, 0.60004, 0.49229, 0.49805, 0.54502, 0.50680, 0.47777, 0.41028, 0.45304]				)

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

     def 					a__	(  self				)						->							List[Any]:
           _A							:     int			=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,     provider="""CPUExecutionProvider"""				)
           _A							:     Union[str, Any]			=						EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config				)
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     Optional[Any]			=						self.get_dummy_inputs()
           _A							:     Tuple			=						pipe(**_a				).images
           _A							:     Any			=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 128, 128, 3)
           _A							:     str			=						np.array([0.52911, 0.60004, 0.49229, 0.49805, 0.54502, 0.50680, 0.47777, 0.41028, 0.45304]				)

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



     def 					a__	(  self				)						->							List[str]:
           _A							:     Union[str, Any]			=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,     provider="""CPUExecutionProvider"""				)
           _A							:     str			=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config				)
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     Union[str, Any]			=						self.get_dummy_inputs()
           _A							:     List[str]			=						pipe(**_a				).images
           _A							:     str			=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 128, 128, 3)
           _A							:     Optional[int]			=						np.array([0.65331, 0.58277, 0.48204, 0.56059, 0.53665, 0.56235, 0.50969, 0.40009, 0.46552]				)

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







@nightly
@require_onnxruntime
@require_torch_gpu
class 						lowercase							(   unittest.TestCase		):

     @property
     def 					a__	(  self				)						->							str:
           return (
               "CUDAExecutionProvider",
               {
                   "gpu_mem_limit": "15000000000",  # 15GB
                   "arena_extend_strategy": "kSameAsRequested",
               },
           )

     @property
     def 					a__	(  self				)						->							Optional[int]:
           _A							:     Union[str, Any]			=						ort.SessionOptions()
           _A							:     List[str]			=						False
           return options

     def 					a__	(  self				)						->							Tuple:
           _A							:     str			=						load_image(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
               """/img2img/sketch-mountains-input.jpg"""				)
           _A							:     Optional[Any]			=						init_image.resize((768, 512)				)
           # using the PNDM scheduler by default
           _A							:     List[str]			=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(
               """CompVis/stable-diffusion-v1-4"""		,     revision="""onnx"""		,     safety_checker=_a		,     feature_extractor=_a		,     provider=self.gpu_provider		,     sess_options=self.gpu_options		,     )
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     str			=						"""A fantasy landscape, trending on artstation"""

           _A							:     List[str]			=						np.random.RandomState(0				)
           _A							:     Dict			=						pipe(
               prompt=_a		,     image=_a		,     strength=0.75		,     guidance_scale=7.5		,     num_inference_steps=10		,     generator=_a		,     output_type="""np"""		,     )
           _A							:     Optional[Any]			=						output.images
           _A							:     Optional[Any]			=						images[0, 255:258, 383:386, -1]

           assert images.shape == (1, 512, 768, 3)
           _A							:     Optional[int]			=						np.array([0.4909, 0.5059, 0.5372, 0.4623, 0.4876, 0.5049, 0.4820, 0.4956, 0.5019]				)
           # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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



     def 					a__	(  self				)						->							Optional[Any]:
           _A							:     Any			=						load_image(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
               """/img2img/sketch-mountains-input.jpg"""				)
           _A							:     Union[str, Any]			=						init_image.resize((768, 512)				)
           _A							:     Any			=						LMSDiscreteScheduler.from_pretrained(
               """runwayml/stable-diffusion-v1-5"""		,     subfolder="""scheduler"""		,     revision="""onnx"""				)
           _A							:     int			=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(
               """runwayml/stable-diffusion-v1-5"""		,     revision="""onnx"""		,     scheduler=_a		,     safety_checker=_a		,     feature_extractor=_a		,     provider=self.gpu_provider		,     sess_options=self.gpu_options		,     )
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     str			=						"""A fantasy landscape, trending on artstation"""

           _A							:     int			=						np.random.RandomState(0				)
           _A							:     Optional[Any]			=						pipe(
               prompt=_a		,     image=_a		,     strength=0.75		,     guidance_scale=7.5		,     num_inference_steps=20		,     generator=_a		,     output_type="""np"""		,     )
           _A							:     Tuple			=						output.images
           _A							:     List[Any]			=						images[0, 255:258, 383:386, -1]

           assert images.shape == (1, 512, 768, 3)
           _A							:     List[Any]			=						np.array([0.8043, 0.926, 0.9581, 0.8119, 0.8954, 0.913, 0.7209, 0.7463, 0.7431]				)
           # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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







import importlib
import json
import os
from collections import OrderedDict
from typing import Dict, Optional, Union

# 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 ...utils import CONFIG_NAME, 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,
)


_snake_case										=       logging.get_logger(__name__)

_snake_case										=       OrderedDict(
    [
        ("audio-spectrogram-transformer", "ASTFeatureExtractor"),
        ("beit", "BeitFeatureExtractor"),
        ("chinese_clip", "ChineseCLIPFeatureExtractor"),
        ("clap", "ClapFeatureExtractor"),
        ("clip", "CLIPFeatureExtractor"),
        ("clipseg", "ViTFeatureExtractor"),
        ("conditional_detr", "ConditionalDetrFeatureExtractor"),
        ("convnext", "ConvNextFeatureExtractor"),
        ("cvt", "ConvNextFeatureExtractor"),
        ("data2vec-audio", "Wav2Vec2FeatureExtractor"),
        ("data2vec-vision", "BeitFeatureExtractor"),
        ("deformable_detr", "DeformableDetrFeatureExtractor"),
        ("deit", "DeiTFeatureExtractor"),
        ("detr", "DetrFeatureExtractor"),
        ("dinat", "ViTFeatureExtractor"),
        ("donut-swin", "DonutFeatureExtractor"),
        ("dpt", "DPTFeatureExtractor"),
        ("encodec", "EncodecFeatureExtractor"),
        ("flava", "FlavaFeatureExtractor"),
        ("glpn", "GLPNFeatureExtractor"),
        ("groupvit", "CLIPFeatureExtractor"),
        ("hubert", "Wav2Vec2FeatureExtractor"),
        ("imagegpt", "ImageGPTFeatureExtractor"),
        ("layoutlmv2", "LayoutLMv2FeatureExtractor"),
        ("layoutlmv3", "LayoutLMv3FeatureExtractor"),
        ("levit", "LevitFeatureExtractor"),
        ("maskformer", "MaskFormerFeatureExtractor"),
        ("mctct", "MCTCTFeatureExtractor"),
        ("mobilenet_v1", "MobileNetV1FeatureExtractor"),
        ("mobilenet_v2", "MobileNetV2FeatureExtractor"),
        ("mobilevit", "MobileViTFeatureExtractor"),
        ("nat", "ViTFeatureExtractor"),
        ("owlvit", "OwlViTFeatureExtractor"),
        ("perceiver", "PerceiverFeatureExtractor"),
        ("poolformer", "PoolFormerFeatureExtractor"),
        ("regnet", "ConvNextFeatureExtractor"),
        ("resnet", "ConvNextFeatureExtractor"),
        ("segformer", "SegformerFeatureExtractor"),
        ("sew", "Wav2Vec2FeatureExtractor"),
        ("sew-d", "Wav2Vec2FeatureExtractor"),
        ("speech_to_text", "Speech2TextFeatureExtractor"),
        ("speecht5", "SpeechT5FeatureExtractor"),
        ("swiftformer", "ViTFeatureExtractor"),
        ("swin", "ViTFeatureExtractor"),
        ("swinv2", "ViTFeatureExtractor"),
        ("table-transformer", "DetrFeatureExtractor"),
        ("timesformer", "VideoMAEFeatureExtractor"),
        ("tvlt", "TvltFeatureExtractor"),
        ("unispeech", "Wav2Vec2FeatureExtractor"),
        ("unispeech-sat", "Wav2Vec2FeatureExtractor"),
        ("van", "ConvNextFeatureExtractor"),
        ("videomae", "VideoMAEFeatureExtractor"),
        ("vilt", "ViltFeatureExtractor"),
        ("vit", "ViTFeatureExtractor"),
        ("vit_mae", "ViTFeatureExtractor"),
        ("vit_msn", "ViTFeatureExtractor"),
        ("wav2vec2", "Wav2Vec2FeatureExtractor"),
        ("wav2vec2-conformer", "Wav2Vec2FeatureExtractor"),
        ("wavlm", "Wav2Vec2FeatureExtractor"),
        ("whisper", "WhisperFeatureExtractor"),
        ("xclip", "CLIPFeatureExtractor"),
        ("yolos", "YolosFeatureExtractor"),
    ]
)

_snake_case										=       _LazyAutoMapping(CONFIG_MAPPING_NAMES, FEATURE_EXTRACTOR_MAPPING_NAMES)







def 				lowerCAmelCase_	(     snake_case_     ):
      for module_name, extractors in FEATURE_EXTRACTOR_MAPPING_NAMES.items():
            if class_name in extractors:
                  _A							:     List[str]			=						model_type_to_module_name(snake_case_     )

                  _A							:     List[Any]			=						importlib.import_module(f'''.{module_name}''',"""transformers.models"""     )
                  try:
                        return getattr(snake_case_,snake_case_     )
                  except AttributeError:
                        continue

      for _, extractor in FEATURE_EXTRACTOR_MAPPING._extra_content.items():
            if getattr(snake_case_,"""__name__""",snake_case_     ) == class_name:
                  return extractor

    # 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.
      _A							:     List[Any]			=						importlib.import_module("""transformers"""     )
      if hasattr(snake_case_,snake_case_     ):
            return getattr(snake_case_,snake_case_     )

      return None







def 				lowerCAmelCase_	(     snake_case_,snake_case_ = None,snake_case_ = False,snake_case_ = False,snake_case_ = None,snake_case_ = None,snake_case_ = None,snake_case_ = False,**snake_case_,):
      _A							:     Optional[int]			=						get_file_from_repo(
          snake_case_,snake_case_,cache_dir=snake_case_,force_download=snake_case_,resume_download=snake_case_,proxies=snake_case_,use_auth_token=snake_case_,revision=snake_case_,local_files_only=snake_case_,)
      if resolved_config_file is None:
            logger.info(
                """Could not locate the feature extractor configuration file, will try to use the model config instead."""     )
            return {}

      with open(snake_case_,encoding="""utf-8"""     ) as reader:
            return json.load(snake_case_     )







class 						lowercase							:

     def __init__(  self				)						->							List[Any]:
           raise EnvironmentError(
               """AutoFeatureExtractor is designed to be instantiated """
               """using the `AutoFeatureExtractor.from_pretrained(pretrained_model_name_or_path)` method."""				)

     @classmethod
     @replace_list_option_in_docstrings(_a				)
     def 					a__	(  cls		,     _a		,     **_a				)						->							Any:
           _A							:     Tuple			=						kwargs.pop("""config"""		,     _a				)
           _A							:     Tuple			=						kwargs.pop("""trust_remote_code"""		,     _a				)
           _A							:     List[Any]			=						True

           _A			,   _A							:     Tuple			=						FeatureExtractionMixin.get_feature_extractor_dict(_a		,     **_a				)
           _A							:     Tuple			=						config_dict.get("""feature_extractor_type"""		,     _a				)
           _A							:     int			=						None
           if "AutoFeatureExtractor" in config_dict.get("""auto_map"""		,     {}				):
                 _A							:     Optional[int]			=						config_dict["""auto_map"""]["""AutoFeatureExtractor"""]

           # If we don't find the feature extractor class in the feature extractor config, let's try the model config.
           if feature_extractor_class is None and feature_extractor_auto_map is None:
                 if not isinstance(_a		,     _a				):
                       _A							:     int			=						AutoConfig.from_pretrained(_a		,     **_a				)
                 # It could be in `config.feature_extractor_type``
                 _A							:     Optional[int]			=						getattr(_a		,     """feature_extractor_type"""		,     _a				)
                 if hasattr(_a		,     """auto_map"""				) and "AutoFeatureExtractor" in config.auto_map:
                       _A							:     Tuple			=						config.auto_map["""AutoFeatureExtractor"""]

           if feature_extractor_class is not None:
                 _A							:     Optional[Any]			=						feature_extractor_class_from_name(_a				)

           _A							:     List[Any]			=						feature_extractor_auto_map is not None
           _A							:     Union[str, Any]			=						feature_extractor_class is not None or type(_a				) in FEATURE_EXTRACTOR_MAPPING
           _A							:     Optional[int]			=						resolve_trust_remote_code(
               _a		,     _a		,     _a		,     _a				)

           if has_remote_code and trust_remote_code:
                 _A							:     Dict			=						get_class_from_dynamic_module(
                     _a		,     _a		,     **_a				)
                 _A							:     str			=						kwargs.pop("""code_revision"""		,     _a				)
                 if os.path.isdir(_a				):
                       feature_extractor_class.register_for_auto_class()
                 return feature_extractor_class.from_dict(_a		,     **_a				)
           elif feature_extractor_class is not None:
                 return feature_extractor_class.from_dict(_a		,     **_a				)
           # Last try: we use the FEATURE_EXTRACTOR_MAPPING.
           elif type(_a				) in FEATURE_EXTRACTOR_MAPPING:
                 _A							:     Dict			=						FEATURE_EXTRACTOR_MAPPING[type(_a				)]
                 return feature_extractor_class.from_dict(_a		,     **_a				)

           raise ValueError(
               F'''Unrecognized feature extractor in {pretrained_model_name_or_path}. Should have a '''
               F'''`feature_extractor_type` key in its {FEATURE_EXTRACTOR_NAME} of {CONFIG_NAME}, or one of the following '''
               F'''`model_type` keys in its {CONFIG_NAME}: {", ".join(c for c in FEATURE_EXTRACTOR_MAPPING_NAMES.keys()				)}'''				)



     @staticmethod
     def 					a__	(  _a		,     _a				)						->							Optional[int]:
           FEATURE_EXTRACTOR_MAPPING.register(_a		,     _a				)

