



'''simple docstring'''





# limitations under the License.


from typing import Optional, Tuple, Union

import torch

from diffusers import DiffusionPipeline, ImagePipelineOutput






class __A    (						A		):



 '''simple docstring'''







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


      """simple docstring"""



      super().__init__()
      self.register_modules(unet=A				,     scheduler=A )



 @torch.no_grad()
 def __call__(self				,     A = 1				,     A = None				,     A = 50				,     A = "pil"				,     A = True				,     **A				,     )					->  Union[ImagePipelineOutput, Tuple]:


      """simple docstring"""



      _a											=						torch.randn(
          (batch_size, self.unet.config.in_channels, self.unet.config.sample_size, self.unet.config.sample_size)				,     generator=A				,     )
      _a											=						image.to(self.device )

      # set step values
      self.scheduler.set_timesteps(A )

      for t in self.progress_bar(self.scheduler.timesteps ):
           # 1. predict noise model_output
           _a											=						self.unet(A				,     A ).sample

           # 2. predict previous mean of image x_t-1 and add variance depending on eta
           # eta corresponds to η in paper and should be between [0, 1]
           # do x_t -> x_t-1
           _a											=						self.scheduler.step(A				,     A				,     A ).prev_sample

      _a											=						(image / 2 + 0.5).clamp(0				,     1 )
      _a											=						image.cpu().permute(0				,     2				,     3				,     1 ).numpy()
      if output_type == "pil":
           _a											=						self.numpy_to_pil(A )

      if not return_dict:
           return (image,), "This is a local test"

      return ImagePipelineOutput(images=A ), "This is a local test"








'''simple docstring'''





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 __A    (						A     ,     unittest.TestCase		):



 '''simple docstring'''







 __lowerCamelCase					: Dict    = 'hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'



 def       a__						(self				,     A=0 )					->  str:


      """simple docstring"""



      _a											=						floats_tensor((1, 3, 128, 128)				,     rng=random.Random(A ) )
      _a											=						np.random.RandomState(A )
      _a											=						{
          '''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 )					->  str:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      pipe.set_progress_bar_config(disable=A )

      _a											=						self.get_dummy_inputs()
      _a											=						pipe(**A ).images
      _a											=						image[0, -3:, -3:, -1].flatten()

      assert image.shape == (1, 128, 128, 3)
      _a											=						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 )					->  Tuple:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						PNDMScheduler.from_config(pipe.scheduler.config				,     skip_prk_steps=A )
      pipe.set_progress_bar_config(disable=A )

      _a											=						self.get_dummy_inputs()
      _a											=						pipe(**A ).images
      _a											=						image[0, -3:, -3:, -1]

      assert image.shape == (1, 128, 128, 3)
      _a											=						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 )					->  int:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						LMSDiscreteScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

      # warmup pass to apply optimizations
      _a											=						pipe(**self.get_dummy_inputs() )

      _a											=						self.get_dummy_inputs()
      _a											=						pipe(**A ).images
      _a											=						image[0, -3:, -3:, -1]

      assert image.shape == (1, 128, 128, 3)
      _a											=						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 )					->  Dict:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						EulerDiscreteScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

      _a											=						self.get_dummy_inputs()
      _a											=						pipe(**A ).images
      _a											=						image[0, -3:, -3:, -1]

      assert image.shape == (1, 128, 128, 3)
      _a											=						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 )					->  Tuple:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

      _a											=						self.get_dummy_inputs()
      _a											=						pipe(**A ).images
      _a											=						image[0, -3:, -3:, -1]

      assert image.shape == (1, 128, 128, 3)
      _a											=						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]:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

      _a											=						self.get_dummy_inputs()
      _a											=						pipe(**A ).images
      _a											=						image[0, -3:, -3:, -1]

      assert image.shape == (1, 128, 128, 3)
      _a											=						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 __A    (						unittest.TestCase		):



 '''simple docstring'''







 @property
 def       a__						(self )					->  Tuple:


      """simple docstring"""



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



 @property
 def       a__						(self )					->  Any:


      """simple docstring"""



      _a											=						ort.SessionOptions()
      _a											=						False
      return options



 def       a__						(self )					->  str:


      """simple docstring"""



      _a											=						load_image(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
          '''/img2img/sketch-mountains-input.jpg''' )
      _a											=						init_image.resize((768, 512) )
      # using the PNDM scheduler by default
      _a											=						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											=						'''A fantasy landscape, trending on artstation'''

      _a											=						np.random.RandomState(0 )
      _a											=						pipe(
          prompt=A				,     image=A				,     strength=0.75				,     guidance_scale=7.5				,     num_inference_steps=10				,     generator=A				,     output_type='''np'''				,     )
      _a											=						output.images
      _a											=						images[0, 255:258, 383:386, -1]

      assert images.shape == (1, 512, 768, 3)
      _a											=						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[int]:


      """simple docstring"""



      _a											=						load_image(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
          '''/img2img/sketch-mountains-input.jpg''' )
      _a											=						init_image.resize((768, 512) )
      _a											=						LMSDiscreteScheduler.from_pretrained(
          '''runwayml/stable-diffusion-v1-5'''				,     subfolder='''scheduler'''				,     revision='''onnx''' )
      _a											=						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											=						'''A fantasy landscape, trending on artstation'''

      _a											=						np.random.RandomState(0 )
      _a											=						pipe(
          prompt=A				,     image=A				,     strength=0.75				,     guidance_scale=7.5				,     num_inference_steps=20				,     generator=A				,     output_type='''np'''				,     )
      _a											=						output.images
      _a											=						images[0, 255:258, 383:386, -1]

      assert images.shape == (1, 512, 768, 3)
      _a											=						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



