

import unittest

import numpy as np

from diffusers import LMSDiscreteScheduler, OnnxStableDiffusionInpaintPipeline
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 A__    ( A  ,		unittest.TestCase ):



 """simple docstring"""







 pass





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



 """simple docstring"""







 @property
 def    __magic_name__       (		self :	Optional[Any]       ):






      '''simple docstring'''




      return (
          "CUDAExecutionProvider",
          {
              "gpu_mem_limit": "15000000000",  # 15GB
              "arena_extend_strategy": "kSameAsRequested",
          },
      )

 @property
 def    __magic_name__       (		self :	Optional[int]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Union[str, Any]        =    ort.SessionOptions()
      _lowerCAmelCase   :					str        =    False
      return options

 def    __magic_name__       (		self :	int       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Tuple        =    load_image(
          "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
          "/in_paint/overture-creations-5sI6fQgYIuo.png"       )
      _lowerCAmelCase   :					Dict        =    load_image(
          "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
          "/in_paint/overture-creations-5sI6fQgYIuo_mask.png"       )
      _lowerCAmelCase   :					List[str]        =    OnnxStableDiffusionInpaintPipeline.from_pretrained(
          "runwayml/stable-diffusion-inpainting"     ,						revision="onnx"     ,						safety_checker=A_     ,						feature_extractor=A_     ,						provider=self.gpu_provider     ,						sess_options=self.gpu_options     ,						)
      pipe.set_progress_bar_config(disable=A_       )

      _lowerCAmelCase   :					Optional[int]        =    "A red cat sitting on a park bench"

      _lowerCAmelCase   :					Optional[int]        =    np.random.RandomState(0       )
      _lowerCAmelCase   :					Optional[int]        =    pipe(
          prompt=A_     ,						image=A_     ,						mask_image=A_     ,						guidance_scale=7.5     ,						num_inference_steps=1_0     ,						generator=A_     ,						output_type="np"     ,						)
      _lowerCAmelCase   :					int        =    output.images
      _lowerCAmelCase   :					List[Any]        =    images[0, 2_5_5:2_5_8, 2_5_5:2_5_8, -1]

      assert images.shape == (1, 5_1_2, 5_1_2, 3)
      _lowerCAmelCase   :					List[Any]        =    np.array([0.2514, 0.3007, 0.3517, 0.1790, 0.2382, 0.3167, 0.1944, 0.2273, 0.2464]       )

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


 def    __magic_name__       (		self :	int       ):






      '''simple docstring'''




      _lowerCAmelCase   :					int        =    load_image(
          "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
          "/in_paint/overture-creations-5sI6fQgYIuo.png"       )
      _lowerCAmelCase   :					Optional[int]        =    load_image(
          "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
          "/in_paint/overture-creations-5sI6fQgYIuo_mask.png"       )
      _lowerCAmelCase   :					List[str]        =    LMSDiscreteScheduler.from_pretrained(
          "runwayml/stable-diffusion-inpainting"     ,						subfolder="scheduler"     ,						revision="onnx"       )
      _lowerCAmelCase   :					Optional[Any]        =    OnnxStableDiffusionInpaintPipeline.from_pretrained(
          "runwayml/stable-diffusion-inpainting"     ,						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_       )

      _lowerCAmelCase   :					Optional[int]        =    "A red cat sitting on a park bench"

      _lowerCAmelCase   :					List[Any]        =    np.random.RandomState(0       )
      _lowerCAmelCase   :					Union[str, Any]        =    pipe(
          prompt=A_     ,						image=A_     ,						mask_image=A_     ,						guidance_scale=7.5     ,						num_inference_steps=2_0     ,						generator=A_     ,						output_type="np"     ,						)
      _lowerCAmelCase   :					List[Any]        =    output.images
      _lowerCAmelCase   :					Union[str, Any]        =    images[0, 2_5_5:2_5_8, 2_5_5:2_5_8, -1]

      assert images.shape == (1, 5_1_2, 5_1_2, 3)
      _lowerCAmelCase   :					Optional[Any]        =    np.array([0.0086, 0.0077, 0.0083, 0.0093, 0.0107, 0.0139, 0.0094, 0.0097, 0.0125]       )

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









from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_tf_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_tf_available():
  import tensorflow as tf


__UpperCAmelCase    				=     logging.get_logger(__name__)





@dataclass
class A__    ( A ):



 """simple docstring"""







 _lowercase     :					List[Any]      =						[
     '''no_inference''',
     '''no_cuda''',
     '''no_tpu''',
     '''no_speed''',
     '''no_memory''',
     '''no_env_print''',
     '''no_multi_process''',
 ]

 def __init__(		self :	List[Any]     ,						**A_ :	str       ):






      '''simple docstring'''




      for deprecated_arg in self.deprecated_args:
           if deprecated_arg in kwargs:
                _lowerCAmelCase   :					str        =    deprecated_arg[3:]
                _lowerCAmelCase   :					List[Any]        =    not kwargs.pop(A_       )
                logger.warning(
                    F'''{deprecated_arg} is depreciated. Please use --no-{positive_arg} or'''
                    F''' {positive_arg}={kwargs[positive_arg]}'''       )
      _lowerCAmelCase   :					Any        =    kwargs.pop("tpu_name"     ,						self.tpu_name       )
      _lowerCAmelCase   :					str        =    kwargs.pop("device_idx"     ,						self.device_idx       )
      _lowerCAmelCase   :					str        =    kwargs.pop("eager_mode"     ,						self.eager_mode       )
      _lowerCAmelCase   :					Tuple        =    kwargs.pop("use_xla"     ,						self.use_xla       )
      super().__init__(**A_       )

 _lowercase     :					str      =						field(
     default=A  ,		metadata={'''help''': '''Name of TPU'''}  ,		)
 _lowercase     :					int      =						field(
     default=0  ,		metadata={'''help''': '''CPU / GPU device index. Defaults to 0.'''}  ,		)
 _lowercase     :					bool      =						field(default=A  ,		metadata={'''help''': '''Benchmark models in eager model.'''} )
 _lowercase     :					bool      =						field(
     default=A  ,		metadata={
         '''help''': '''Benchmark models using XLA JIT compilation. Note that `eager_model` has to be set to `False`.'''
     }  ,		)

 @cached_property
 def    __magic_name__       (		self :	Tuple       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      _lowerCAmelCase   :					Any        =    None
      if self.tpu:
           try:
                if self.tpu_name:
                     _lowerCAmelCase   :					Union[str, Any]        =    tf.distribute.cluster_resolver.TPUClusterResolver(self.tpu_name       )
                else:
                     _lowerCAmelCase   :					List[str]        =    tf.distribute.cluster_resolver.TPUClusterResolver()
           except ValueError:
                _lowerCAmelCase   :					Optional[int]        =    None
      return tpu

 @cached_property
 def    __magic_name__       (		self :	Optional[int]       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      if self.is_tpu:
           tf.config.experimental_connect_to_cluster(self._setup_tpu       )
           tf.tpu.experimental.initialize_tpu_system(self._setup_tpu       )

           _lowerCAmelCase   :					List[Any]        =    tf.distribute.TPUStrategy(self._setup_tpu       )
      else:
           # currently no multi gpu is allowed
           if self.is_gpu:
                # TODO: Currently only single GPU is supported
                tf.config.set_visible_devices(self.gpu_list[self.device_idx]     ,						"GPU"       )
                _lowerCAmelCase   :					Dict        =    tf.distribute.OneDeviceStrategy(device=F'''/gpu:{self.device_idx}'''       )
           else:
                tf.config.set_visible_devices([]     ,						"GPU"       )  # disable GPU
                _lowerCAmelCase   :					Optional[Any]        =    tf.distribute.OneDeviceStrategy(device=F'''/cpu:{self.device_idx}'''       )

      return strategy

 @property
 def    __magic_name__       (		self :	int       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      return self._setup_tpu is not None

 @property
 def    __magic_name__       (		self :	List[Any]       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      return self._setup_strategy

 @property
 def    __magic_name__       (		self :	List[str]       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      return tf.config.list_physical_devices("GPU"       )

 @property
 def    __magic_name__       (		self :	Optional[Any]       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      if self.cuda:
           return len(self.gpu_list       )
      return 0


 @property
 def    __magic_name__       (		self :	Optional[Any]       ):






      '''simple docstring'''




      return self.n_gpu > 0






