



'''simple docstring'''






import inspect
import unittest

import torch
import torch.nn as nn

from accelerate.hooks import (
    AlignDevicesHook,
    ModelHook,
    SequentialHook,
    add_hook_to_module,
    attach_align_device_hook,
    remove_hook_from_module,
    remove_hook_from_submodules,
)
from accelerate.test_utils import require_multi_gpu

class 							a__(      nn.Module			):







    def __init__(   self				:						Dict			):
      super().__init__()
      a		:      int       =  nn.Linear(3		,    4			)
      a		:      Tuple       =  nn.BatchNormad(4			)
      a		:      Union[str, Any]       =  nn.Linear(4		,    5			)







    def     lowercase_ (   self				:						List[Any]		,    __snake_case				:						Optional[int]			):
      return self.lineara(self.batchnorm(self.lineara(UpperCamelCase__			)			)			)

class 							a__(      __A			):







    def     lowercase_ (   self				:						Any		,    __snake_case				:						int		,    *__snake_case				:						Optional[int]		,    **__snake_case				:						int			):
      return (args[0] + 1,) + args[1:], kwargs

class 							a__(      __A			):







    def     lowercase_ (   self				:						List[Any]		,    __snake_case				:						Optional[Any]		,    __snake_case				:						Optional[Any]			):
      return output + 1

class 							a__(      unittest.TestCase			):







    def     lowercase_ (   self				:						List[Any]			):
      a		:      Dict       =  ModelForTest()
      a		:      Union[str, Any]       =  ModelHook()

      add_hook_to_module(UpperCamelCase__		,    UpperCamelCase__			)
      self.assertEqual(test_model._hf_hook		,    UpperCamelCase__			)
      self.assertTrue(hasattr(UpperCamelCase__		,    '_old_forward'			)			)

      # Check adding the hook did not change the name or the signature
      self.assertEqual(test_model.forward.__name__		,    'forward'			)
      self.assertListEqual(list(inspect.signature(test_model.forward			).parameters			)		,    ['x']			)

      remove_hook_from_module(UpperCamelCase__			)
      self.assertFalse(hasattr(UpperCamelCase__		,    '_hf_hook'			)			)
      self.assertFalse(hasattr(UpperCamelCase__		,    '_old_forward'			)			)







    def     lowercase_ (   self				:						int			):
      a		:      Optional[Any]       =  ModelForTest()
      a		:      Optional[Any]       =  ModelHook()

      add_hook_to_module(UpperCamelCase__		,    UpperCamelCase__			)
      add_hook_to_module(UpperCamelCase__		,    UpperCamelCase__		,    append=UpperCamelCase__			)

      self.assertEqual(isinstance(test_model._hf_hook		,    UpperCamelCase__			)		,    UpperCamelCase__			)
      self.assertEqual(len(test_model._hf_hook.hooks			)		,    2			)
      self.assertTrue(hasattr(UpperCamelCase__		,    '_old_forward'			)			)

      # Check adding the hook did not change the name or the signature
      self.assertEqual(test_model.forward.__name__		,    'forward'			)
      self.assertListEqual(list(inspect.signature(test_model.forward			).parameters			)		,    ['x']			)

      remove_hook_from_module(UpperCamelCase__			)
      self.assertFalse(hasattr(UpperCamelCase__		,    '_hf_hook'			)			)
      self.assertFalse(hasattr(UpperCamelCase__		,    '_old_forward'			)			)







    def     lowercase_ (   self				:						Union[str, Any]			):
      a		:      int       =  ModelForTest()
      a		:      Dict       =  torch.randn(2		,    3			)
      a		:      Optional[int]       =  test_model(x + 1			)
      a		:      Dict       =  test_model(x + 2			)

      a		:      List[Any]       =  PreForwardHook()
      add_hook_to_module(UpperCamelCase__		,    UpperCamelCase__			)
      a		:      Tuple       =  test_model(UpperCamelCase__			)
      self.assertTrue(torch.allclose(UpperCamelCase__		,    UpperCamelCase__		,    atol=1e-5			)			)

      # Attaching a hook to a model when it already has one replaces, does not chain
      a		:      Optional[int]       =  PreForwardHook()
      add_hook_to_module(UpperCamelCase__		,    UpperCamelCase__			)
      a		:      List[str]       =  test_model(UpperCamelCase__			)
      self.assertTrue(torch.allclose(UpperCamelCase__		,    UpperCamelCase__		,    atol=1e-5			)			)

      # You need to use the sequential hook to chain two or more hooks
      a		:      Any       =  SequentialHook(PreForwardHook()		,    PreForwardHook()			)
      add_hook_to_module(UpperCamelCase__		,    UpperCamelCase__			)

      a		:      Tuple       =  test_model(UpperCamelCase__			)
      assert torch.allclose(UpperCamelCase__		,    UpperCamelCase__		,    atol=1e-5			)







    def     lowercase_ (   self				:						Optional[Any]			):
      a		:      Tuple       =  ModelForTest()
      a		:      List[Any]       =  torch.randn(2		,    3			)
      a		:      Tuple       =  test_model(UpperCamelCase__			)

      a		:      Union[str, Any]       =  PostForwardHook()
      add_hook_to_module(UpperCamelCase__		,    UpperCamelCase__			)
      a		:      Any       =  test_model(UpperCamelCase__			)
      self.assertTrue(torch.allclose(UpperCamelCase__		,    output + 1		,    atol=1e-5			)			)

      # Attaching a hook to a model when it already has one replaces, does not chain
      a		:      Optional[int]       =  PostForwardHook()
      add_hook_to_module(UpperCamelCase__		,    UpperCamelCase__			)
      a		:      int       =  test_model(UpperCamelCase__			)
      self.assertTrue(torch.allclose(UpperCamelCase__		,    output + 1		,    atol=1e-5			)			)

      # You need to use the sequential hook to chain two or more hooks
      a		:      Tuple       =  SequentialHook(PostForwardHook()		,    PostForwardHook()			)
      add_hook_to_module(UpperCamelCase__		,    UpperCamelCase__			)

      a		:      str       =  test_model(UpperCamelCase__			)
      assert torch.allclose(UpperCamelCase__		,    output + 2		,    atol=1e-5			)







    def     lowercase_ (   self				:						List[Any]			):
      a		:      Optional[Any]       =  ModelForTest()
      a		:      Optional[Any]       =  torch.randn(2		,    3			)
      a		:      List[Any]       =  test_model(UpperCamelCase__			)

      a		:      Any       =  PostForwardHook()
      add_hook_to_module(UpperCamelCase__		,    UpperCamelCase__			)
      a		:      List[Any]       =  test_model(UpperCamelCase__			)
      self.assertTrue(torch.allclose(UpperCamelCase__		,    output + 1			)			)
      self.assertTrue(outputa.requires_grad			)

      a		:      Optional[int]       =  True
      a		:      Optional[Any]       =  test_model(UpperCamelCase__			)
      self.assertFalse(outputa.requires_grad			)







    @require_multi_gpu
    def     lowercase_ (   self				:						List[Any]			):
      a		:      Tuple       =  ModelForTest()
      # Everything is on CPU
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)

      # This will move each submodule on different devices
      add_hook_to_module(model.lineara		,    AlignDevicesHook(execution_device=0			)			)
      add_hook_to_module(model.batchnorm		,    AlignDevicesHook(execution_device=0			)			)
      add_hook_to_module(model.lineara		,    AlignDevicesHook(execution_device=1			)			)

      self.assertEqual(model.lineara.weight.device		,    torch.device(0			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device(0			)			)
      self.assertEqual(model.batchnorm.running_mean.device		,    torch.device(0			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device(1			)			)

      # We can still make a forward pass. The input does not need to be on any particular device
      a		:      int       =  torch.randn(2		,    3			)
      a		:      Dict       =  model(UpperCamelCase__			)
      self.assertEqual(output.device		,    torch.device(1			)			)

      # We can add a general hook to put back output on same device as input.
      add_hook_to_module(UpperCamelCase__		,    AlignDevicesHook(io_same_device=UpperCamelCase__			)			)
      a		:      Union[str, Any]       =  torch.randn(2		,    3			).to(0			)
      a		:      int       =  model(UpperCamelCase__			)
      self.assertEqual(output.device		,    torch.device(0			)			)







    def     lowercase_ (   self				:						int			):
      a		:      str       =  ModelForTest()

      # Everything is on CPU
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)

      # This will move each submodule on different devices
      a		:      str       =  {'execution_device': 0 if torch.cuda.is_available() else 'cpu', 'offload': True}

      add_hook_to_module(model.lineara		,    AlignDevicesHook(**UpperCamelCase__			)			)
      add_hook_to_module(model.batchnorm		,    AlignDevicesHook(**UpperCamelCase__			)			)
      add_hook_to_module(model.lineara		,    AlignDevicesHook(**UpperCamelCase__			)			)

      # Parameters have been offloaded, so on the meta device
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      # Buffers are not included in the offload by default, so are on the execution device
      a		:      Dict       =  torch.device(hook_kwargs['execution_device']			)
      self.assertEqual(model.batchnorm.running_mean.device		,    UpperCamelCase__			)

      a		:      str       =  torch.randn(2		,    3			)
      a		:      Any       =  model(UpperCamelCase__			)
      self.assertEqual(output.device		,    UpperCamelCase__			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_module(model.lineara			)
      remove_hook_from_module(model.batchnorm			)
      remove_hook_from_module(model.lineara			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)

      # Now test with buffers included in the offload
      a		:      str       =  {
          'execution_device': 0 if torch.cuda.is_available() else 'cpu',
          'offload': True,
          'offload_buffers': True,
      }

      add_hook_to_module(model.lineara		,    AlignDevicesHook(**UpperCamelCase__			)			)
      add_hook_to_module(model.batchnorm		,    AlignDevicesHook(**UpperCamelCase__			)			)
      add_hook_to_module(model.lineara		,    AlignDevicesHook(**UpperCamelCase__			)			)

      # Parameters have been offloaded, so on the meta device, buffers included
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.batchnorm.running_mean.device		,    torch.device('meta'			)			)

      a		:      Union[str, Any]       =  torch.randn(2		,    3			)
      a		:      Union[str, Any]       =  model(UpperCamelCase__			)
      self.assertEqual(output.device		,    UpperCamelCase__			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_module(model.lineara			)
      remove_hook_from_module(model.batchnorm			)
      remove_hook_from_module(model.lineara			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)







    def     lowercase_ (   self				:						List[str]			):
      a		:      Union[str, Any]       =  ModelForTest()

      # Everything is on CPU
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)

      # This will move each submodule on different devices
      a		:      Union[str, Any]       =  0 if torch.cuda.is_available() else 'cpu'
      attach_align_device_hook(UpperCamelCase__		,    execution_device=UpperCamelCase__		,    offload=UpperCamelCase__			)

      # Parameters have been offloaded, so on the meta device
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      # Buffers are not included in the offload by default, so are on the execution device
      a		:      Any       =  torch.device(UpperCamelCase__			)
      self.assertEqual(model.batchnorm.running_mean.device		,    UpperCamelCase__			)

      a		:      Optional[Any]       =  torch.randn(2		,    3			)
      a		:      Optional[int]       =  model(UpperCamelCase__			)
      self.assertEqual(output.device		,    UpperCamelCase__			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(UpperCamelCase__			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)

      # Now test with buffers included in the offload
      attach_align_device_hook(UpperCamelCase__		,    execution_device=UpperCamelCase__		,    offload=UpperCamelCase__		,    offload_buffers=UpperCamelCase__			)

      # Parameters have been offloaded, so on the meta device, buffers included
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.batchnorm.running_mean.device		,    torch.device('meta'			)			)

      a		:      int       =  torch.randn(2		,    3			)
      a		:      List[str]       =  model(UpperCamelCase__			)
      self.assertEqual(output.device		,    UpperCamelCase__			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(UpperCamelCase__			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)







    def     lowercase_ (   self				:						Optional[Any]			):
      a		:      Any       =  ModelForTest()

      # Everything is on CPU
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)

      # This will move each submodule on different devices
      a		:      Union[str, Any]       =  0 if torch.cuda.is_available() else 'cpu'
      attach_align_device_hook(
          UpperCamelCase__		,    execution_device=UpperCamelCase__		,    offload=UpperCamelCase__		,    weights_map=model.state_dict()			)

      # Parameters have been offloaded, so on the meta device
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      # Buffers are not included in the offload by default, so are on the execution device
      a		:      List[str]       =  torch.device(UpperCamelCase__			)
      self.assertEqual(model.batchnorm.running_mean.device		,    UpperCamelCase__			)

      a		:      Any       =  torch.randn(2		,    3			)
      a		:      List[str]       =  model(UpperCamelCase__			)
      self.assertEqual(output.device		,    UpperCamelCase__			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(UpperCamelCase__			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)

      # Now test with buffers included in the offload
      attach_align_device_hook(
          UpperCamelCase__		,    execution_device=UpperCamelCase__		,    offload=UpperCamelCase__		,    weights_map=model.state_dict()		,    offload_buffers=UpperCamelCase__		,    )

      # Parameters have been offloaded, so on the meta device, buffers included
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('meta'			)			)
      self.assertEqual(model.batchnorm.running_mean.device		,    torch.device('meta'			)			)

      a		:      Tuple       =  torch.randn(2		,    3			)
      a		:      Dict       =  model(UpperCamelCase__			)
      self.assertEqual(output.device		,    UpperCamelCase__			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(UpperCamelCase__			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.batchnorm.weight.device		,    torch.device('cpu'			)			)
      self.assertEqual(model.lineara.weight.device		,    torch.device('cpu'			)			)




'''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__(      lowerCamelCase__	,				unittest.TestCase			):
    lowercase__				=					"""hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"""







    def     lowercase_ (   self				:						Dict		,    __snake_case				:						Union[str, Any]=0			):
      a		:      List[str]       =  floats_tensor((1, 3, 1_28, 1_28)		,    rng=random.Random(__snake_case			)			)
      a		:      Optional[Any]       =  np.random.RandomState(__snake_case			)
      a		:      List[str]       =  {
          '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     lowercase_ (   self				:						Union[str, Any]			):
      a		:      Dict       =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,    provider='CPUExecutionProvider'			)
      pipe.set_progress_bar_config(disable=__snake_case			)

      a		:      Any       =  self.get_dummy_inputs()
      a		:      int       =  pipe(**__snake_case			).images
      a		:      List[Any]       =  image[0, -3:, -3:, -1].flatten()

      assert image.shape == (1, 1_28, 1_28, 3)
      a		:      Dict       =  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     lowercase_ (   self				:						str			):
      a		:      List[Any]       =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,    provider='CPUExecutionProvider'			)
      a		:      Dict       =  PNDMScheduler.from_config(pipe.scheduler.config		,    skip_prk_steps=__snake_case			)
      pipe.set_progress_bar_config(disable=__snake_case			)

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

      assert image.shape == (1, 1_28, 1_28, 3)
      a		:      int       =  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     lowercase_ (   self				:						List[str]			):
      a		:      Optional[int]       =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,    provider='CPUExecutionProvider'			)
      a		:      List[Any]       =  LMSDiscreteScheduler.from_config(pipe.scheduler.config			)
      pipe.set_progress_bar_config(disable=__snake_case			)

      # warmup pass to apply optimizations
      a		:      Any       =  pipe(**self.get_dummy_inputs()			)

      a		:      Optional[int]       =  self.get_dummy_inputs()
      a		:      Optional[int]       =  pipe(**__snake_case			).images
      a		:      Any       =  image[0, -3:, -3:, -1]

      assert image.shape == (1, 1_28, 1_28, 3)
      a		:      Any       =  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     lowercase_ (   self				:						List[Any]			):
      a		:      Dict       =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,    provider='CPUExecutionProvider'			)
      a		:      Any       =  EulerDiscreteScheduler.from_config(pipe.scheduler.config			)
      pipe.set_progress_bar_config(disable=__snake_case			)

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

      assert image.shape == (1, 1_28, 1_28, 3)
      a		:      Dict       =  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     lowercase_ (   self				:						Tuple			):
      a		:      Any       =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,    provider='CPUExecutionProvider'			)
      a		:      Optional[Any]       =  EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config			)
      pipe.set_progress_bar_config(disable=__snake_case			)

      a		:      Dict       =  self.get_dummy_inputs()
      a		:      Union[str, Any]       =  pipe(**__snake_case			).images
      a		:      int       =  image[0, -3:, -3:, -1]

      assert image.shape == (1, 1_28, 1_28, 3)
      a		:      int       =  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     lowercase_ (   self				:						Optional[Any]			):
      a		:      Optional[int]       =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,    provider='CPUExecutionProvider'			)
      a		:      Optional[int]       =  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config			)
      pipe.set_progress_bar_config(disable=__snake_case			)

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

      assert image.shape == (1, 1_28, 1_28, 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 							a__(      unittest.TestCase			):







    @property
    def     lowercase_ (   self				:						List[Any]			):
      return (
          "CUDAExecutionProvider",
          {
              "gpu_mem_limit": "15000000000",  # 15GB
              "arena_extend_strategy": "kSameAsRequested",
          },
      )







    @property
    def     lowercase_ (   self				:						Dict			):
      a		:      Tuple       =  ort.SessionOptions()
      a		:      Optional[Any]       =  False
      return options







    def     lowercase_ (   self				:						List[str]			):
      a		:      Dict       =  load_image(
          'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
          '/img2img/sketch-mountains-input.jpg'			)
      a		:      Optional[int]       =  init_image.resize((7_68, 5_12)			)
      # using the PNDM scheduler by default
      a		:      Any       =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(
          'CompVis/stable-diffusion-v1-4'		,    revision='onnx'		,    safety_checker=__snake_case		,    feature_extractor=__snake_case		,    provider=self.gpu_provider		,    sess_options=self.gpu_options		,    )
      pipe.set_progress_bar_config(disable=__snake_case			)

      a		:      Union[str, Any]       =  'A fantasy landscape, trending on artstation'

      a		:      Dict       =  np.random.RandomState(0			)
      a		:      Optional[int]       =  pipe(
          prompt=__snake_case		,    image=__snake_case		,    strength=0.75		,    guidance_scale=7.5		,    num_inference_steps=10		,    generator=__snake_case		,    output_type='np'		,    )
      a		:      str       =  output.images
      a		:      Optional[int]       =  images[0, 2_55:2_58, 3_83:3_86, -1]

      assert images.shape == (1, 5_12, 7_68, 3)
      a		:      List[str]       =  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     lowercase_ (   self				:						str			):
      a		:      Optional[Any]       =  load_image(
          'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
          '/img2img/sketch-mountains-input.jpg'			)
      a		:      List[str]       =  init_image.resize((7_68, 5_12)			)
      a		:      Union[str, Any]       =  LMSDiscreteScheduler.from_pretrained(
          'runwayml/stable-diffusion-v1-5'		,    subfolder='scheduler'		,    revision='onnx'			)
      a		:      Any       =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(
          'runwayml/stable-diffusion-v1-5'		,    revision='onnx'		,    scheduler=__snake_case		,    safety_checker=__snake_case		,    feature_extractor=__snake_case		,    provider=self.gpu_provider		,    sess_options=self.gpu_options		,    )
      pipe.set_progress_bar_config(disable=__snake_case			)

      a		:      Optional[int]       =  'A fantasy landscape, trending on artstation'

      a		:      str       =  np.random.RandomState(0			)
      a		:      List[str]       =  pipe(
          prompt=__snake_case		,    image=__snake_case		,    strength=0.75		,    guidance_scale=7.5		,    num_inference_steps=20		,    generator=__snake_case		,    output_type='np'		,    )
      a		:      str       =  output.images
      a		:      Tuple       =  images[0, 2_55:2_58, 3_83:3_86, -1]

      assert images.shape == (1, 5_12, 7_68, 3)
      a		:      Union[str, 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